platform_auth.py 1.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940
  1. """
  2. 平台对外 API 鉴权依赖
  3. 外部系统通过 Authorization: Bearer sk-aigc-xxx 调用对外接口时使用
  4. """
  5. from fastapi import Depends, HTTPException, status
  6. from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
  7. from sqlalchemy.orm import Session
  8. from app.database import get_db
  9. from app.services.platform_api_key_service import PlatformApiKeyService
  10. bearer_scheme = HTTPBearer(auto_error=False)
  11. def get_platform_caller(
  12. credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
  13. db: Session = Depends(get_db),
  14. ) -> str:
  15. """
  16. 验证平台 API Key,返回调用方 user_id。
  17. 请求头格式:Authorization: Bearer sk-aigc-xxxxxxxx
  18. """
  19. if not credentials or not credentials.credentials:
  20. raise HTTPException(
  21. status_code=status.HTTP_401_UNAUTHORIZED,
  22. detail="缺少 Authorization 请求头",
  23. headers={"WWW-Authenticate": "Bearer"},
  24. )
  25. result = PlatformApiKeyService(db).verify_api_key(credentials.credentials)
  26. if not result:
  27. raise HTTPException(
  28. status_code=status.HTTP_401_UNAUTHORIZED,
  29. detail="API Key 无效或已禁用",
  30. headers={"WWW-Authenticate": "Bearer"},
  31. )
  32. user_id, _ = result
  33. return user_id