""" Platform Auth Verify Endpoint 用于 nginx auth_request 子请求验证平台 JWT 返回 200 + X-User-Id/X-User-Name headers 或 401 """ import logging from fastapi import APIRouter, Request from fastapi.responses import Response from app.auth import decode_access_token logger = logging.getLogger("aiflow.auth_verify") router = APIRouter() @router.get("/verify") @router.head("/verify") async def verify_auth(request: Request): """ nginx auth_request 验证端点 从请求中提取 JWT (Authorization header 或 cookie)并验证 ��回 200 + 用户信息 headers,或 401 """ token = None # 1. 尝试 Authorization: Bearer auth_header = request.headers.get("Authorization", "") if auth_header.startswith("Bearer "): token = auth_header[7:] # 2. 尝试 cookie 中的 jwt token if not token: token = request.cookies.get("jwt") or request.cookies.get("token") or request.cookies.get("access_token") # 3. 尝试 X-Platform-Token header if not token: token = request.headers.get("X-Platform-Token") if not token: logger.debug("No token found in request") return Response(status_code=401, headers={"X-Auth-Error": "no_token"}) payload = decode_access_token(token) if not payload: logger.debug("Invalid or expired token") return Response(status_code=401, headers={"X-Auth-Error": "invalid_token"}) user_id = payload.get("sub", "") user_name = payload.get("username", "") user_role = payload.get("role", "user") tenant_id = payload.get("tenant_id", "") if not user_id: return Response(status_code=401, headers={"X-Auth-Error": "no_user_id"}) logger.info(f"Auth verify OK: user={user_id}, name={user_name}") return Response( status_code=200, headers={ "X-User-Id": str(user_id), "X-User-Name": str(user_name) if user_name else str(user_id), "X-User-Role": str(user_role), "X-Tenant-Id": str(tenant_id) if tenant_id else "", "Cache-Control": "no-cache, no-store, must-revalidate", }, )