131 lines
3.8 KiB
Python
131 lines
3.8 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
|
|
|
|
|
|
# 创建全局实例
|
|
casdoor_auth = CasdoorAuth()
|
|
|