46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
from dataclasses import dataclass
|
|
import hmac
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, HTTPException, Request, Security, status
|
|
from fastapi.security import APIKeyHeader
|
|
|
|
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AuthenticatedUser:
|
|
user_id: str
|
|
|
|
|
|
class ApiKeyAuthenticator:
|
|
def __init__(self, whitelist: dict[str, str]) -> None:
|
|
if not whitelist:
|
|
raise ValueError("API_KEY_WHITELIST must contain at least one user")
|
|
self._whitelist = whitelist.copy()
|
|
|
|
def authenticate(self, api_key: str | None) -> AuthenticatedUser | None:
|
|
if not api_key:
|
|
return None
|
|
for user_id, allowed_key in self._whitelist.items():
|
|
if hmac.compare_digest(api_key, allowed_key):
|
|
return AuthenticatedUser(user_id=user_id)
|
|
return None
|
|
|
|
|
|
async def require_api_key(
|
|
request: Request,
|
|
api_key: Annotated[str | None, Security(api_key_header)],
|
|
) -> AuthenticatedUser:
|
|
authenticator: ApiKeyAuthenticator = request.app.state.authenticator
|
|
user = authenticator.authenticate(api_key)
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="api_key is required",
|
|
)
|
|
return user
|
|
|
|
|
|
CurrentUser = Annotated[AuthenticatedUser, Depends(require_api_key)]
|