fastapi/backend/app/core/user_context.py

from fastapi import Header, Request, Security
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from app.core.context import UserContext
from app.core.exceptions import AppError
from app.services.external_auth_service import ExternalAuthService

bearer_scheme = HTTPBearer(auto_error=False, description="Django 用户中心 access token")


async def get_user_context(
    request: Request,
    credentials: HTTPAuthorizationCredentials | None = Security(bearer_scheme),
    x_entry_scene: str | None = Header(default=None, alias="X-Entry-Scene"),
    x_request_id: str | None = Header(default=None, alias="X-Request-Id"),
) -> UserContext:
    """用户校验：只接受宿主系统 access token，并转发 Django 用户中心 `/me` 获取真实用户。"""
    if not credentials or not credentials.credentials.strip():
        raise AppError("AUTH_CREDENTIAL_REQUIRED", "Authorization header is required", 401)

    user = await ExternalAuthService().authenticate(request)
    return UserContext(
        user_id=user.user_id,
        tenant_id=user.tenant_id,
        role=user.role,
        entry_scene=x_entry_scene,
        request_id=x_request_id,
        ip_address=request.client.host if request.client else None,
        user_agent=request.headers.get("User-Agent"),
        username=user.username,
        display_name=user.display_name,
        auth_source=user.source,
        profile=user.profile,
    )