261 lines
8.9 KiB
Python
261 lines
8.9 KiB
Python
"""
|
|
Casdoor 认证工具模块
|
|
用于处理 Casdoor OAuth/OIDC 认证流程
|
|
"""
|
|
import requests
|
|
import urllib.parse
|
|
from flask import session, current_app
|
|
import secrets
|
|
|
|
|
|
class CasdoorAuth:
|
|
"""Casdoor 认证处理类"""
|
|
|
|
def __init__(self, app=None):
|
|
self.app = app
|
|
if app:
|
|
self.init_app(app)
|
|
|
|
def init_app(self, app):
|
|
"""初始化应用配置"""
|
|
self.endpoint = app.config.get('CASDOOR_ENDPOINT')
|
|
self.client_id = app.config.get('CASDOOR_CLIENT_ID')
|
|
self.client_secret = app.config.get('CASDOOR_CLIENT_SECRET')
|
|
self.organization_name = app.config.get('CASDOOR_ORGANIZATION_NAME', 'built-in')
|
|
self.application_name = app.config.get('CASDOOR_APPLICATION_NAME', 'app-built-in')
|
|
self.redirect_uri = app.config.get('CASDOOR_REDIRECT_URI')
|
|
|
|
def get_authorization_url(self, state=None):
|
|
"""
|
|
生成 Casdoor 授权 URL
|
|
|
|
Args:
|
|
state: 状态参数，用于防止 CSRF 攻击
|
|
|
|
Returns:
|
|
str: 授权 URL
|
|
"""
|
|
if not state:
|
|
state = secrets.token_urlsafe(32)
|
|
session['casdoor_state'] = state
|
|
|
|
params = {
|
|
'client_id': self.client_id,
|
|
'response_type': 'code',
|
|
'redirect_uri': self.redirect_uri,
|
|
'scope': 'openid profile email phone',
|
|
'state': state
|
|
}
|
|
|
|
auth_url = f"{self.endpoint}/login/oauth/authorize"
|
|
return f"{auth_url}?{urllib.parse.urlencode(params)}"
|
|
|
|
def get_token(self, code):
|
|
"""
|
|
使用授权码获取访问令牌
|
|
|
|
Args:
|
|
code: 授权码
|
|
|
|
Returns:
|
|
dict: 包含 access_token 和 id_token 的字典
|
|
"""
|
|
token_url = f"{self.endpoint}/api/login/oauth/access_token"
|
|
|
|
data = {
|
|
'grant_type': 'authorization_code',
|
|
'client_id': self.client_id,
|
|
'client_secret': self.client_secret,
|
|
'code': code,
|
|
'redirect_uri': self.redirect_uri
|
|
}
|
|
|
|
try:
|
|
response = requests.post(token_url, data=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
try:
|
|
current_app.logger.error(f"获取 Casdoor token 失败: {str(e)}")
|
|
except:
|
|
pass # 如果无法记录日志，忽略错误
|
|
return None
|
|
|
|
def get_user_info(self, access_token):
|
|
"""
|
|
使用访问令牌获取用户信息
|
|
|
|
Args:
|
|
access_token: 访问令牌
|
|
|
|
Returns:
|
|
dict: 用户信息字典
|
|
"""
|
|
user_info_url = f"{self.endpoint}/api/userinfo"
|
|
|
|
headers = {
|
|
'Authorization': f'Bearer {access_token}'
|
|
}
|
|
|
|
try:
|
|
response = requests.get(user_info_url, headers=headers)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
try:
|
|
current_app.logger.error(f"获取 Casdoor 用户信息失败: {str(e)}")
|
|
except:
|
|
pass # 如果无法记录日志，忽略错误
|
|
return None
|
|
|
|
def verify_state(self, state):
|
|
"""
|
|
验证 state 参数，防止 CSRF 攻击
|
|
|
|
Args:
|
|
state: 从回调中获取的 state 参数
|
|
|
|
Returns:
|
|
bool: 验证是否通过
|
|
"""
|
|
stored_state = session.get('casdoor_state')
|
|
if stored_state and stored_state == state:
|
|
session.pop('casdoor_state', None)
|
|
return True
|
|
return False
|
|
|
|
def login_with_password(self, username, password):
|
|
"""
|
|
使用用户名和密码通过 API 登录 Casdoor
|
|
|
|
Args:
|
|
username: 用户名
|
|
password: 密码
|
|
|
|
Returns:
|
|
dict: 包含 access_token 和用户信息的字典，失败返回 None
|
|
"""
|
|
# Casdoor 使用 OAuth 2.0 password grant 方式登录
|
|
# 端点应该是 /api/login/oauth/access_token
|
|
login_url = f"{self.endpoint}/api/login/oauth/access_token"
|
|
|
|
# OAuth 2.0 password grant 需要的参数
|
|
data = {
|
|
'grant_type': 'password',
|
|
'username': username,
|
|
'password': password,
|
|
'client_id': self.client_id,
|
|
'client_secret': self.client_secret,
|
|
'scope': 'openid profile email phone'
|
|
}
|
|
|
|
try:
|
|
current_app.logger.info(f"尝试 Casdoor OAuth password grant 登录: username={username}")
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
# OAuth 2.0 password grant 通常使用 form-data 格式
|
|
response = requests.post(login_url, data=data, timeout=10)
|
|
|
|
# 如果返回 415，尝试 JSON 格式
|
|
if response.status_code == 415:
|
|
response = requests.post(login_url, json=data, timeout=10)
|
|
|
|
# 检查响应状态
|
|
if response.status_code != 200:
|
|
try:
|
|
error_msg = response.text[:200] # 限制错误消息长度
|
|
current_app.logger.warning(f"Casdoor OAuth 登录失败 (状态码 {response.status_code}): {error_msg}")
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
# 尝试解析 JSON 响应
|
|
try:
|
|
result = response.json()
|
|
except ValueError as e:
|
|
try:
|
|
current_app.logger.error(f"Casdoor API 登录响应不是有效的 JSON: {str(e)}, 响应内容: {response.text[:200]}")
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
# 检查 result 是否为 None
|
|
if result is None:
|
|
try:
|
|
current_app.logger.error("Casdoor API 登录返回 None")
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
# OAuth 2.0 返回格式通常是 {"access_token": "...", "token_type": "Bearer", ...}
|
|
# 或者 Casdoor 格式 {"status": "ok", "data": {"access_token": "..."}}
|
|
access_token = None
|
|
if isinstance(result, dict):
|
|
# 检查是否有错误
|
|
if result.get('status') == 'error':
|
|
try:
|
|
error_msg = result.get('msg', '未知错误')
|
|
current_app.logger.warning(f"Casdoor OAuth 登录错误: {error_msg}")
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
# 标准 OAuth 2.0 格式
|
|
access_token = result.get('access_token')
|
|
|
|
# Casdoor 格式 {"status": "ok", "data": {...}}
|
|
if not access_token and result.get('status') == 'ok':
|
|
token_data = result.get('data', {})
|
|
if isinstance(token_data, dict):
|
|
access_token = token_data.get('access_token') or token_data.get('token') or token_data.get('accessToken')
|
|
|
|
# 如果还是没有，尝试其他可能的字段
|
|
if not access_token:
|
|
access_token = result.get('token') or result.get('accessToken')
|
|
|
|
# 如果还是没有 token，尝试从其他字段获取
|
|
if not access_token and 'data' in result and isinstance(result.get('data'), dict):
|
|
access_token = result['data'].get('access_token') or result['data'].get('token')
|
|
|
|
if not access_token:
|
|
try:
|
|
current_app.logger.warning(f"Casdoor API 登录返回格式异常，未找到 token: {result}")
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
# 使用 token 获取用户信息
|
|
user_info = self.get_user_info(access_token)
|
|
if not user_info:
|
|
return None
|
|
|
|
return {
|
|
'access_token': access_token,
|
|
'user_info': user_info
|
|
}
|
|
except requests.exceptions.Timeout:
|
|
try:
|
|
current_app.logger.error("Casdoor API 登录超时")
|
|
except:
|
|
pass
|
|
return None
|
|
except requests.exceptions.RequestException as e:
|
|
try:
|
|
current_app.logger.error(f"Casdoor API 登录失败: {str(e)}")
|
|
except:
|
|
pass
|
|
return None
|
|
except Exception as e:
|
|
try:
|
|
current_app.logger.error(f"Casdoor API 登录发生未知错误: {str(e)}")
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
|
|
# 创建全局实例
|
|
casdoor_auth = CasdoorAuth()
|
|
|