67 lines
2 KiB
Python
67 lines
2 KiB
Python
|
"""
|
||
|
|
API key authentication for the issue-core REST API.
|
||
|
|
|
||
|
|
A single shared key is read from the ISSUE_CORE_API_KEY environment variable.
|
||
|
|
Clients send it either as `Authorization: Bearer <key>` or as `X-API-Key: <key>`.
|
||
|
|
|
||
|
|
If ISSUE_CORE_API_KEY is unset, the server refuses to start — the workplan
|
||
|
|
explicitly forbids an unauthenticated ingestion surface.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import secrets
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
from fastapi import Header, HTTPException, status
|
||
|
|
|
||
|
|
|
||
|
|
API_KEY_ENV_VAR = "ISSUE_CORE_API_KEY"
|
||
|
|
|
||
|
|
|
||
|
|
class AuthConfigError(RuntimeError):
|
||
|
|
"""Raised at startup when no API key is configured."""
|
||
|
|
|
||
|
|
|
||
|
|
def get_configured_api_key() -> str:
|
||
|
|
key = os.environ.get(API_KEY_ENV_VAR, "").strip()
|
||
|
|
if not key:
|
||
|
|
raise AuthConfigError(
|
||
|
|
f"{API_KEY_ENV_VAR} is not set. The ingestion endpoint requires an API key. "
|
||
|
|
f"Generate one with: python -c 'import secrets; print(secrets.token_urlsafe(32))'"
|
||
|
|
)
|
||
|
|
return key
|
||
|
|
|
||
|
|
|
||
|
|
def _extract_token(
|
||
|
|
authorization: Optional[str],
|
||
|
|
x_api_key: Optional[str],
|
||
|
|
) -> Optional[str]:
|
||
|
|
if x_api_key:
|
||
|
|
return x_api_key.strip()
|
||
|
|
if authorization:
|
||
|
|
scheme, _, value = authorization.partition(" ")
|
||
|
|
if scheme.lower() == "bearer" and value:
|
||
|
|
return value.strip()
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
async def require_api_key(
|
||
|
|
authorization: Optional[str] = Header(default=None),
|
||
|
|
x_api_key: Optional[str] = Header(default=None, alias="X-API-Key"),
|
||
|
|
) -> None:
|
||
|
|
"""FastAPI dependency enforcing the shared API key."""
|
||
|
|
try:
|
||
|
|
expected = get_configured_api_key()
|
||
|
|
except AuthConfigError as exc:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||
|
|
detail=str(exc),
|
||
|
|
)
|
||
|
|
presented = _extract_token(authorization, x_api_key)
|
||
|
|
if not presented or not secrets.compare_digest(presented, expected):
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
|
|
detail="Missing or invalid API key.",
|
||
|
|
headers={"WWW-Authenticate": "Bearer"},
|
||
|
|
)
|