Ubuntu
109 lines
3.2 KiB
Python
109 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
认证相关路由和装饰器 —— 双模式认证（本地 Session + SSO Cookie）
|
||
"""
|
||
from functools import wraps
|
||
from datetime import timedelta
|
||
from flask import Blueprint, request, jsonify, session
|
||
from flask_login import login_user, logout_user, login_required, current_user
|
||
from models import User, db
|
||
from sso_client import get_user_from_sso_cookie
|
||
|
||
auth_bp = Blueprint('auth', __name__)
|
||
|
||
|
||
def api_login_required(f):
|
||
"""API 登录检查装饰器 - 双模式认证
|
||
1. 优先检查本地 Flask-Login Session
|
||
2. 无 Session 则检查 SSO Cookie，本地验签后查本地 User 表
|
||
"""
|
||
@wraps(f)
|
||
def decorated_function(*args, **kwargs):
|
||
# 模式一：本地 Session
|
||
if current_user.is_authenticated:
|
||
return f(*args, **kwargs)
|
||
|
||
# 模式二：SSO Cookie
|
||
sso_user = get_user_from_sso_cookie(request)
|
||
if sso_user:
|
||
username = sso_user.get("username")
|
||
user = User.query.filter_by(username=username).first()
|
||
if user:
|
||
# 自动登录到本地 Session，后续请求走模式一
|
||
login_user(user, remember=True, duration=timedelta(days=30))
|
||
return f(*args, **kwargs)
|
||
|
||
return jsonify({
|
||
'error': 'Unauthorized',
|
||
'message': '请先登录',
|
||
'code': 401
|
||
}), 401
|
||
return decorated_function
|
||
|
||
|
||
@auth_bp.route('/api/auth/login', methods=['POST'])
|
||
def login():
|
||
"""用户登录 API"""
|
||
data = request.get_json()
|
||
|
||
if not data:
|
||
return jsonify({'error': '请求数据为空'}), 400
|
||
|
||
username = data.get('username', '').strip()
|
||
password = data.get('password', '').strip()
|
||
|
||
if not username or not password:
|
||
return jsonify({'error': '用户名和密码不能为空'}), 400
|
||
|
||
user = User.query.filter_by(username=username).first()
|
||
|
||
if not user or not user.check_password(password):
|
||
return jsonify({'error': '用户名或密码错误'}), 401
|
||
|
||
# 登录用户（记住我 30 天）
|
||
login_user(user, remember=True, duration=timedelta(days=30))
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': '登录成功',
|
||
'user': user.to_dict()
|
||
})
|
||
|
||
|
||
@auth_bp.route('/api/auth/logout', methods=['POST'])
|
||
def logout():
|
||
"""用户登出 API —— 不要求登录态，确保未登录也能退出"""
|
||
if current_user.is_authenticated:
|
||
logout_user()
|
||
return jsonify({
|
||
'success': True,
|
||
'message': '已登出'
|
||
})
|
||
|
||
|
||
@auth_bp.route('/api/auth/status')
|
||
def auth_status():
|
||
"""检查登录状态 API —— 双模式"""
|
||
# 模式一：本地 Session
|
||
if current_user.is_authenticated:
|
||
return jsonify({
|
||
'authenticated': True,
|
||
'user': current_user.to_dict()
|
||
})
|
||
|
||
# 模式二：SSO Cookie
|
||
sso_user = get_user_from_sso_cookie(request)
|
||
if sso_user:
|
||
username = sso_user.get("username")
|
||
user = User.query.filter_by(username=username).first()
|
||
if user:
|
||
return jsonify({
|
||
'authenticated': True,
|
||
'user': user.to_dict()
|
||
})
|
||
|
||
return jsonify({
|
||
'authenticated': False,
|
||
'user': None
|
||
})
|