""" 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 get_account_info(self, access_token): """ 使用访问令牌获取完整账户信息（包含标签等字段） """ account_url = f"{self.endpoint}/api/get-account" headers = { 'Authorization': f'Bearer {access_token}' } try: response = requests.get(account_url, headers=headers) response.raise_for_status() result = response.json() # Casdoor 通常返回 {"status": "ok", "data": {...}} if isinstance(result, dict): if result.get('status') == 'ok' and isinstance(result.get('data'), dict): return result['data'] if 'data' in result and isinstance(result.get('data'), dict): return result['data'] return result if isinstance(result, dict) else None 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()