137 lines
5.7 KiB
Python
Executable File
137 lines
5.7 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():
|
|
# 如果已登录，直接跳转到仪表盘
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('dashboard'))
|
|
# 未登录展示登录页（登录页会自动跳转 Casdoor，如有错误提示则不跳转）
|
|
return render_template('login.html')
|
|
|
|
@app.route('/casdoor/login')
|
|
def casdoor_login():
|
|
"""重定向到 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'))
|
|
|
|
# 获取完整账户信息（包含标签）
|
|
account_info = casdoor_auth.get_account_info(access_token)
|
|
if not account_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 '未知用户'
|
|
|
|
# 基于标签的访问控制：用户标签必须等于项目名
|
|
project_name = app.config.get('PROJECT_NAME')
|
|
if not project_name:
|
|
app.logger.error('未配置 PROJECT_NAME 环境变量，拒绝登录')
|
|
flash('登录失败：未配置项目名')
|
|
return redirect(url_for('login'))
|
|
|
|
raw_tags = account_info.get('tag') or account_info.get('tags')
|
|
if isinstance(raw_tags, str):
|
|
user_tags = [t.strip() for t in raw_tags.split(',') if t.strip()]
|
|
elif isinstance(raw_tags, list):
|
|
user_tags = [str(t).strip() for t in raw_tags if str(t).strip()]
|
|
else:
|
|
user_tags = []
|
|
|
|
if project_name not in user_tags:
|
|
app.logger.warning(f'用户 {name}({phone}) 标签 {user_tags} 不匹配项目 {project_name}')
|
|
flash('您没有本系统的访问权限，请联系管理员处理')
|
|
return redirect(url_for('login'))
|
|
|
|
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')) |