Files

69 lines
2.1 KiB
Python

"""
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 <token>
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",
},
)