113 lines
4.5 KiB
Python
Executable File
113 lines
4.5 KiB
Python
Executable File
from flask import render_template, request, redirect, url_for, flash, session
|
|
import urllib.parse
|
|
from flask_login import login_user, login_required, logout_user, current_user
|
|
from scripts.models import db, User
|
|
from utils.casdoor_auth import casdoor_auth
|
|
|
|
def init_auth_routes(app):
|
|
# 初始化 Casdoor 认证
|
|
casdoor_auth.init_app(app)
|
|
|
|
@app.route('/login', methods=['GET', 'POST'])
|
|
def login():
|
|
# 已登录直接进入仪表盘，否则跳转 Casdoor 登录
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('dashboard'))
|
|
return redirect(url_for('casdoor_login'))
|
|
|
|
@app.route('/casdoor/login')
|
|
def casdoor_login():
|
|
"""Casdoor 登录入口，重定向到 Casdoor 授权页面"""
|
|
if not app.config.get('CASDOOR_CLIENT_ID'):
|
|
flash('Casdoor 未配置，请使用手机号登录')
|
|
return redirect(url_for('login'))
|
|
|
|
auth_url = casdoor_auth.get_authorization_url()
|
|
app.logger.info('重定向到 Casdoor 授权页面')
|
|
return redirect(auth_url)
|
|
|
|
@app.route('/callback')
|
|
def casdoor_callback():
|
|
"""Casdoor OAuth 回调处理"""
|
|
code = request.args.get('code')
|
|
state = request.args.get('state')
|
|
error = request.args.get('error')
|
|
|
|
if error:
|
|
app.logger.error(f'Casdoor 认证失败: {error}')
|
|
flash(f'登录失败: {error}')
|
|
return redirect(url_for('login'))
|
|
|
|
if not code:
|
|
app.logger.warning('Casdoor 回调缺少授权码')
|
|
flash('登录失败：未收到授权码')
|
|
return redirect(url_for('login'))
|
|
|
|
# 验证 state 参数
|
|
if not casdoor_auth.verify_state(state):
|
|
app.logger.warning('Casdoor state 验证失败，可能存在 CSRF 攻击')
|
|
flash('登录失败：安全验证失败')
|
|
return redirect(url_for('login'))
|
|
|
|
# 获取访问令牌
|
|
token_data = casdoor_auth.get_token(code)
|
|
if not token_data or 'access_token' not in token_data:
|
|
app.logger.error('获取 Casdoor token 失败')
|
|
flash('登录失败：无法获取访问令牌')
|
|
return redirect(url_for('login'))
|
|
|
|
access_token = token_data['access_token']
|
|
|
|
# 获取用户信息
|
|
user_info = casdoor_auth.get_user_info(access_token)
|
|
if not user_info:
|
|
app.logger.error('获取 Casdoor 用户信息失败')
|
|
flash('登录失败：无法获取用户信息')
|
|
return redirect(url_for('login'))
|
|
|
|
# 从 Casdoor 用户信息中提取数据
|
|
# Casdoor 返回的用户信息可能包含：name, email, phone, id 等字段
|
|
casdoor_id = user_info.get('sub') or user_info.get('id') or user_info.get('name')
|
|
phone = user_info.get('phone') or user_info.get('phoneNumber') or casdoor_id
|
|
name = user_info.get('name') or user_info.get('displayName') or '未知用户'
|
|
|
|
app.logger.info(f'Casdoor 用户信息: {user_info}')
|
|
|
|
# 查找或创建本地用户
|
|
user = User.query.filter_by(phone=phone).first()
|
|
if not user:
|
|
# 如果用户不存在，创建新用户
|
|
user = User(
|
|
phone=phone,
|
|
name=name,
|
|
branch=user_info.get('affiliation', ''),
|
|
role=user_info.get('type', '')
|
|
)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
app.logger.info(f'创建新用户: {name}({phone})')
|
|
else:
|
|
# 更新用户信息
|
|
user.name = name
|
|
if user_info.get('affiliation'):
|
|
user.branch = user_info.get('affiliation')
|
|
if user_info.get('type'):
|
|
user.role = user_info.get('type')
|
|
db.session.commit()
|
|
app.logger.info(f'更新用户信息: {name}({phone})')
|
|
|
|
# 登录用户
|
|
login_user(user)
|
|
app.logger.info(f'用户 {user.name}({user.phone}) 通过 Casdoor 登录成功')
|
|
|
|
# 清除可能残留的未登录提示
|
|
session.pop('_flashes', None)
|
|
return redirect(url_for('dashboard'))
|
|
|
|
@app.route('/logout')
|
|
@login_required
|
|
def logout():
|
|
app.logger.info(f'用户 {current_user.name}({current_user.phone}) 登出')
|
|
logout_user()
|
|
# 仅退出本地会话，保留 Casdoor 登录态
|
|
return redirect(url_for('login')) |