shudao-main/shudao-chat-py/utils/auth_middleware.py

from fastapi import Request, HTTPException, status
from fastapi.responses import JSONResponse
from .token import verify_local_token
from .logger import logger


async def auth_middleware(request: Request, call_next):
    """Token认证中间件"""
    # 白名单路径（不需要认证）
    whitelist_paths = [
        "/",
        "/health",
        "/docs",
        "/redoc",
        "/openapi.json",
        "/static",
        "/assets",
        "/apiv1/auth/local_login",
        "/apiv1/auth/register"
    ]

    # 检查是否在白名单中
    path = request.url.path
    for whitelist_path in whitelist_paths:
        if path.startswith(whitelist_path):
            # 白名单路径也设置一个默认user，避免后续访问出错
            request.state.user = None
            return await call_next(request)

    # 获取Token
    token = request.headers.get("token") or request.headers.get(
        "Authorization", "").replace("Bearer ", "")

    logger.info(f"认证中间件 - 路径: {path}")
    logger.info(f"认证中间件 - Token (前20字符): {token[:20] if token else 'None'}...")

    if not token:
        logger.warning("认证中间件 - 未提供Token")
        return JSONResponse(
            status_code=status.HTTP_401_UNAUTHORIZED,
            content={"code": 401, "msg": "未提供认证Token"}
        )

    # 验证Token
    logger.info("认证中间件 - 开始验证Token")
    user_info = await verify_local_token(token)

    if not user_info:
        logger.error("认证中间件 - Token验证失败，返回401")
        return JSONResponse(
            status_code=status.HTTP_401_UNAUTHORIZED,
            content={"code": 401, "msg": "Token验证失败"}
        )

    logger.info(
        f"认证中间件 - Token验证成功，用户: {user_info.username} ({user_info.account})")

    # 将用户信息存储到request.state中
    request.state.user = user_info

    response = await call_next(request)
    return response