998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
"""
|
||
简化认证中间件 - 支持 API Key 和长期 Token
|
||
用于内部服务间调用
|
||
"""
|
||
from typing import Optional
|
||
from fastapi import HTTPException, Header, status
|
||
from app.models.user import User
|
||
|
||
# 配置 API Keys（用于内部服务调用）
|
||
API_KEYS = {
|
||
"internal-service-2025-kaopeilian": {
|
||
"service": "internal",
|
||
"user_id": 1,
|
||
"username": "internal_service",
|
||
"role": "admin"
|
||
}
|
||
}
|
||
|
||
# 长期有效的 Token（用于内部服务调用）
|
||
LONG_TERM_TOKENS = {
|
||
"permanent-token-for-internal-2025": {
|
||
"service": "internal",
|
||
"user_id": 1,
|
||
"username": "internal_service",
|
||
"role": "admin"
|
||
}
|
||
}
|
||
|
||
|
||
def get_current_user_by_api_key(
|
||
x_api_key: Optional[str] = Header(None),
|
||
authorization: Optional[str] = Header(None)
|
||
) -> Optional[User]:
|
||
"""
|
||
通过 API Key 或长期 Token 获取用户
|
||
支持两种方式：
|
||
1. X-API-Key: internal-service-2025-kaopeilian
|
||
2. Authorization: Bearer permanent-token-for-internal-2025
|
||
"""
|
||
|
||
# 方式1：检查 API Key
|
||
if x_api_key and x_api_key in API_KEYS:
|
||
api_key_info = API_KEYS[x_api_key]
|
||
# 创建一个虚拟用户对象
|
||
user = User()
|
||
user.id = api_key_info["user_id"]
|
||
user.username = api_key_info["username"]
|
||
user.role = api_key_info["role"]
|
||
return user
|
||
|
||
# 方式2：检查长期 Token
|
||
if authorization and authorization.startswith("Bearer "):
|
||
token = authorization.replace("Bearer ", "")
|
||
if token in LONG_TERM_TOKENS:
|
||
token_info = LONG_TERM_TOKENS[token]
|
||
user = User()
|
||
user.id = token_info["user_id"]
|
||
user.username = token_info["username"]
|
||
user.role = token_info["role"]
|
||
return user
|
||
|
||
return None
|
||
|
||
|
||
def get_current_user_simple(
|
||
x_api_key: Optional[str] = Header(None),
|
||
authorization: Optional[str] = Header(None)
|
||
) -> User:
|
||
"""
|
||
简化的用户认证依赖项
|
||
"""
|
||
# 尝试 API Key 或长期 Token 认证
|
||
user = get_current_user_by_api_key(x_api_key, authorization)
|
||
if user:
|
||
return user
|
||
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="Invalid or missing authentication credentials",
|
||
headers={"WWW-Authenticate": "Bearer"},
|
||
)
|