File size: 1,679 Bytes
14b1bbe | 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 | """
Lightweight API-key + OAuth2 bearer authentication for the Mythos API.
Backed by an env-defined static API key (``MYTHOS_API_KEYS=key1,key2``) plus
optional JWT validation when ``MYTHOS_JWT_PUBKEY`` is set.
"""
from __future__ import annotations
import os
from typing import Any
try: # pragma: no cover
from fastapi import Header, HTTPException, status
except Exception: # noqa: BLE001
Header = HTTPException = status = None # type: ignore
try: # pragma: no cover
import jwt # type: ignore
_JWT = True
except Exception: # noqa: BLE001
_JWT = False
def _allowed_keys() -> set[str]:
return {k.strip() for k in os.getenv("MYTHOS_API_KEYS", "").split(",") if k.strip()}
def require_api_key(authorization: str | None = Header(default=None)) -> dict[str, Any]:
if HTTPException is None: # FastAPI not installed — let the caller handle it.
return {"sub": "anonymous"}
if not authorization or not authorization.lower().startswith("bearer "):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing bearer")
token = authorization.split(None, 1)[1].strip()
keys = _allowed_keys()
if keys and token in keys:
return {"sub": "api-key", "token": token[:8] + "..."}
if _JWT and (pubkey := os.getenv("MYTHOS_JWT_PUBKEY")):
try:
return jwt.decode(token, pubkey, algorithms=["RS256"])
except Exception as exc: # noqa: BLE001
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"jwt: {exc}") from exc
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid token")
|