42 lines
1.5 KiB
Python
42 lines
1.5 KiB
Python
"""FastAPI 依赖注入：提供容器获取和鉴权校验"""
|
|
import logging
|
|
import secrets
|
|
|
|
from fastapi import Depends, Request
|
|
from app.core.container import ServiceContainer
|
|
from app.core.errors import AuthenticationError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_container(request: Request) -> ServiceContainer:
|
|
"""从请求中获取服务容器"""
|
|
return request.app.state.container
|
|
|
|
|
|
def require_bearer_token(
|
|
request: Request,
|
|
container: ServiceContainer = Depends(get_container),
|
|
) -> None:
|
|
"""Bearer Token 鉴权依赖
|
|
|
|
从请求头 Authorization 中提取 Bearer Token，
|
|
与服务端配置的 token 做安全比较（使用 secrets.compare_digest 防止时序攻击）。
|
|
"""
|
|
authorization = request.headers.get("Authorization")
|
|
if not authorization:
|
|
logger.warning("鉴权失败：缺少 Authorization 请求头")
|
|
raise AuthenticationError("Missing Authorization header")
|
|
|
|
token = authorization.replace("Bearer ", "", 1).strip()
|
|
if token == authorization:
|
|
logger.warning("鉴权失败：Authorization scheme 非 Bearer")
|
|
raise AuthenticationError("Invalid Authorization scheme")
|
|
if not token:
|
|
logger.warning("鉴权失败：Bearer token 为空")
|
|
raise AuthenticationError("Invalid token")
|
|
|
|
if not secrets.compare_digest(token, container.settings.bearer_token):
|
|
logger.warning("鉴权失败：token 不匹配")
|
|
raise AuthenticationError("Invalid token")
|