""" Casdoor SSO 集成模块 ================== 实现 Casdoor OIDC 单点登录集成， 支持 JWT + Casdoor 双认证模式。 使用方式: from app.casdoor_auth import casdoor_router, get_casdoor_login_url, handle_casdoor_callback """ import os import logging import httpx from typing import Optional from datetime import datetime from urllib.parse import urlencode from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import RedirectResponse from sqlalchemy.orm import Session from pydantic import BaseModel from .database import get_db from .models import User, gen_uuid from .auth import create_access_token, hash_password logger = logging.getLogger("aiflow.casdoor") router = APIRouter() # Casdoor 配置 CASDOOR_ENDPOINT = os.environ.get("CASDOOR_ENDPOINT", "http://casdoor:8000") CASDOOR_CLIENT_ID = os.environ.get("CASDOOR_CLIENT_ID", "aiflow") CASDOOR_CLIENT_SECRET = os.environ.get("CASDOOR_CLIENT_SECRET", "") CASDOOR_ORG_NAME = os.environ.get("CASDOOR_ORG_NAME", "aiflow") CASDOOR_APP_NAME = os.environ.get("CASDOOR_APP_NAME", "aiflow-app") CASDOOR_REDIRECT_URI = os.environ.get( "CASDOOR_REDIRECT_URI", "http://210.16.161.218:12888/app/api/auth/casdoor/callback" ) APP_FRONTEND_URL = os.environ.get("APP_ACCESS_URL", "http://210.16.161.218:12888") def get_casdoor_login_url(state: str = "") -> str: """获取 Casdoor 登录跳转URL""" params = { "client_id": CASDOOR_CLIENT_ID, "response_type": "code", "redirect_uri": CASDOOR_REDIRECT_URI, "scope": "openid profile email", "state": state or "aiflow", } return f"{CASDOOR_ENDPOINT}/login/oauth/authorize?{urlencode(params)}" async def exchange_code_for_token(code: str) -> Optional[dict]: """用授权码换取 access_token 和用户信息""" try: async with httpx.AsyncClient(timeout=30) as client: # 获取 token token_resp = await client.post( f"{CASDOOR_ENDPOINT}/api/login/oauth/access_token", json={ "grant_type": "authorization_code", "client_id": CASDOOR_CLIENT_ID, "client_secret": CASDOOR_CLIENT_SECRET, "code": code, "redirect_uri": CASDOOR_REDIRECT_URI, }, ) if token_resp.status_code != 200: logger.error(f"Casdoor token exchange failed: {token_resp.status_code} {token_resp.text}") return None token_data = token_resp.json() access_token = token_data.get("access_token") if not access_token: logger.error("No access_token in Casdoor response") return None # 获取用户信息 user_resp = await client.get( f"{CASDOOR_ENDPOINT}/api/account", headers={"Authorization": f"Bearer {access_token}"}, ) if user_resp.status_code != 200: # 尝试备用端点 user_resp = await client.get( f"{CASDOOR_ENDPOINT}/api/userinfo", headers={"Authorization": f"Bearer {access_token}"}, ) if user_resp.status_code != 200: logger.error(f"Casdoor userinfo failed: {user_resp.status_code}") return {"access_token": access_token, "user": None} user_info = user_resp.json() return {"access_token": access_token, "user": user_info} except Exception as e: logger.error(f"Casdoor token exchange error: {e}") return None async def get_or_create_user(db: Session, casdoor_user: dict) -> User: """根据 Casdoor 用户信息获取或创建本地用户""" casdoor_id = casdoor_user.get("id", casdoor_user.get("name", "")) casdoor_name = casdoor_user.get("name", casdoor_user.get("preferred_username", "")) casdoor_email = casdoor_user.get("email", "") casdoor_display = casdoor_user.get("displayName", casdoor_user.get("name", "")) # 通过 Casdoor ID 查找已有用户 existing = db.query(User).filter(User.casdoor_id == casdoor_id).first() if hasattr(User, "casdoor_id") else None # 通过用户名查找 if not existing and casdoor_name: existing = db.query(User).filter(User.username == f"sso_{casdoor_name}").first() if existing and hasattr(existing, "casdoor_id"): existing.casdoor_id = casdoor_id if existing: # 更新最后登录时间 existing.last_login = datetime.now() db.commit() return existing # 创建新用户 username = f"sso_{casdoor_name}" if casdoor_name else f"sso_{casdoor_id[:8]}" # 避免用户名冲突 base_username = username counter = 1 while db.query(User).filter(User.username == username).first(): username = f"{base_username}_{counter}" counter += 1 new_user = User( id=gen_uuid(), username=username, email=casdoor_email or None, password_hash=hash_password(casdoor_id + datetime.now().isoformat()), # 随机密码 role="user", status="active", display_name=casdoor_display or username, ) # 如果 User 模型有 casdoor_id 字段则设置 if hasattr(new_user, "casdoor_id"): new_user.casdoor_id = casdoor_id db.add(new_user) db.commit() db.refresh(new_user) logger.info(f"Created SSO user: {username} (casdoor_id={casdoor_id})") return new_user # ==================== API 路由 ==================== @router.get("/casdoor/login") async def casdoor_login(state: str = ""): """跳转到 Casdoor 登录页面""" login_url = get_casdoor_login_url(state) return RedirectResponse(url=login_url) @router.get("/casdoor/callback") async def casdoor_callback( code: str, state: str = "", db: Session = Depends(get_db), ): """Casdoor OAuth 回调""" result = await exchange_code_for_token(code) if not result or not result.get("user"): # 重定向到前端登录页，带错误参数 return RedirectResponse(url=f"{APP_FRONTEND_URL}/#/login?error=casdoor_failed") casdoor_user = result["user"] user = await get_or_create_user(db, casdoor_user) # 生成平台 JWT token token_data = {"sub": user.id, "username": user.username, "role": user.role} access_token = create_access_token(token_data) # 重定向到前端，带 token redirect_url = f"{APP_FRONTEND_URL}/#/login?token={access_token}&sso=casdoor" return RedirectResponse(url=redirect_url) @router.get("/casdoor/status") async def casdoor_status(): """检查 Casdoor SSO 是否可用""" try: async with httpx.AsyncClient(timeout=5) as client: resp = await client.get(f"{CASDOOR_ENDPOINT}/api/health") if resp.status_code == 200: return {"enabled": True, "endpoint": CASDOOR_ENDPOINT} except Exception: pass return {"enabled": False, "endpoint": CASDOOR_ENDPOINT} # 配置说明： # 1. 在 Casdoor 中创建应用，获取 Client ID 和 Client Secret # 2. 设置环境变量: # - CASDOOR_ENDPOINT=http://casdoor:8000 # - CASDOOR_CLIENT_ID=your_client_id # - CASDOOR_CLIENT_SECRET=your_client_secret # - CASDOOR_REDIRECT_URI=http://your-domain/app/api/auth/casdoor/callback # 3. 在 User 模型中添加 casdoor_id 字段 (可选): # casdoor_id = Column(String(128), unique=True, nullable=True)