File size: 2,054 Bytes
30e6218 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | """FastAPI dependencies: service access and API-key enforcement."""
from __future__ import annotations
import hmac
from typing import Optional
from fastapi import Depends, HTTPException, Request
from fastapi.security import APIKeyHeader
from ..core.config import Settings
from ..rag.generator import Generator
from ..rag.service import DocumentService
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
def get_settings_dep(request: Request) -> Settings:
return request.app.state.settings
def get_service(request: Request) -> DocumentService:
service = getattr(request.app.state, "service", None)
if service is None:
raise HTTPException(status_code=503, detail="Document service is not available")
return service
def get_generator(request: Request) -> Generator:
generator = getattr(request.app.state, "generator", None)
if generator is None:
raise HTTPException(status_code=503, detail="Generator is not available")
return generator
def _check_key(settings: Settings, provided: Optional[str]) -> None:
expected = settings.api_key
if expected is None:
return # local mode: no key configured
if not provided or not hmac.compare_digest(provided.encode(), expected.encode()):
raise HTTPException(status_code=401, detail="Invalid or missing API key")
async def require_api_key(
settings: Settings = Depends(get_settings_dep),
provided: Optional[str] = Depends(api_key_header),
) -> None:
"""Enforced on every /api/v1 route when API_KEY is configured."""
_check_key(settings, provided)
async def require_configured_key(
settings: Settings = Depends(get_settings_dep),
provided: Optional[str] = Depends(api_key_header),
) -> None:
"""Destructive bulk operations require a configured *and* matching key."""
if settings.api_key is None:
raise HTTPException(
status_code=403,
detail="This operation requires API_KEY to be configured on the server",
)
_check_key(settings, provided)
|