|
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): |
|
|
|
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:] |
|
return api_key |
|
|
|
def verify_token(request: Request): |
|
"""验证API密钥(可选)""" |
|
|
|
api_enabled = config.app.get("api_enabled", True) |
|
configured_api_key = config.app.get("api_key", "") |
|
|
|
|
|
if not api_enabled or not configured_api_key: |
|
return True |
|
|
|
|
|
provided_api_key = get_api_key(request) |
|
|
|
|
|
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 |
|
|