from uuid import uuid4 from fastapi import Request from app.config import config from app.models.exception import HttpException def get_task_id(request: Request): task_id = request.headers.get("x-task-id") if not task_id: task_id = uuid4() return str(task_id) def get_api_key(request: Request): # Try multiple ways to get API key api_key = request.headers.get("x-api-key") if not api_key: api_key = request.headers.get("authorization") if api_key and api_key.startswith("Bearer "): api_key = api_key[7:] # Remove "Bearer " prefix return api_key def verify_token(request: Request): """验证API密钥(可选)""" # Check if API authentication is enabled api_enabled = config.app.get("api_enabled", True) configured_api_key = config.app.get("api_key", "") # If API is disabled or no API key is configured, skip authentication if not api_enabled or not configured_api_key: return True # Get API key from request provided_api_key = get_api_key(request) # Verify API key if not provided_api_key or provided_api_key != configured_api_key: request_id = get_task_id(request) # 使用通用错误消息,不暴露系统详细信息 raise HttpException( task_id=request_id, status_code=401, message="Authentication failed. Please provide a valid API key via X-API-Key header or Authorization Bearer token.", ) return True