- sms/auth.py: verify_api_key dependency (Bearer/X-API-Key, SHA-256 hash compare via secrets.compare_digest, fail-closed 503 if no hash). All routes gated via a protected router; /health stays open for probes. - config.py: new sms_api_key_hash setting (VOIPMS_SMS_API_KEY_HASH). - Dockerfile + .dockerignore + docker-compose.yml: lean python:3.13-slim image; secrets injected via env_file, never baked in; host-localhost-only port mapping (SMS_PORT override); /health healthcheck. - .env.example: committed template (placeholders only, no secrets). - sms/docs/sms-facade.dokuwiki.txt: abstraction API reference (published to the apidoc wiki at voipms:sms-facade). - README: Authentication section, Docker section, 401/503 error rows. Co-Authored-By: Claude <noreply@anthropic.com>
69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
"""Static API-key authentication for the sms façade.
|
|
|
|
Clients present a key in one of two headers:
|
|
|
|
Authorization: Bearer <raw-api-key>
|
|
X-API-Key: <raw-api-key>
|
|
|
|
The server stores only the **SHA-256 hex hash(es)** of accepted keys in the
|
|
``SMS_API_KEY_HASH`` setting (comma-separated). The raw key is never persisted;
|
|
losing it means generating a new one and updating the hash.
|
|
|
|
Fail-closed: if no hash is configured, every protected route returns 503 rather
|
|
than silently admitting traffic. ``/health`` is exempt (see ``app.py``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import secrets
|
|
|
|
from fastapi import HTTPException, Request, status
|
|
|
|
logger = logging.getLogger("voipms.sms.auth")
|
|
|
|
|
|
def _hash_key(raw: str) -> str:
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _allowed_hashes(request: Request) -> set[str]:
|
|
raw = request.app.state.settings.sms_api_key_hash or ""
|
|
return {h.strip().lower() for h in raw.split(",") if h.strip()}
|
|
|
|
|
|
def _extract_token(request: Request) -> str | None:
|
|
auth = request.headers.get("authorization")
|
|
if auth:
|
|
parts = auth.split(None, 1)
|
|
if len(parts) == 2 and parts[0].lower() == "bearer":
|
|
return parts[1].strip() or None
|
|
return None
|
|
return request.headers.get("x-api-key")
|
|
|
|
|
|
def verify_api_key(request: Request) -> None:
|
|
"""FastAPI dependency gating every protected route."""
|
|
allowed = _allowed_hashes(request)
|
|
if not allowed:
|
|
logger.error("SMS_API_KEY_HASH not configured; refusing request (fail-closed).")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="api key auth not configured",
|
|
)
|
|
token = _extract_token(request)
|
|
if not token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="missing api key",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
digest = _hash_key(token)
|
|
if not any(secrets.compare_digest(digest, h) for h in allowed):
|
|
logger.warning("rejected api key from %s", request.client.host if request.client else "?")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="invalid api key",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
) |