T011: ledger X-API-Key auth (service/admin)

ledger/src/m2_ledger/auth.py

@@ -0,0 +1,53 @@
+"""X-API-Key auth for m2-ledger (contracts/ledger-api.md §4).
+
+Two key classes, both read from env as comma-separated lists:
+- LEDGER_SERVICE_KEYS: service-class keys (CLI/Store/indexer).
+- LEDGER_ADMIN_KEYS: admin-class keys (grants, snapshots).
+
+`require_service` accepts either class; `require_admin` accepts admin only.
+Keys are loaded lazily (first request) so tests can set env vars before import
+without needing to reload the module.
+"""
+
+import hmac
+import os
+from functools import lru_cache
+
+from fastapi import Header, HTTPException
+
+UNAUTHORIZED = HTTPException(status_code=401, detail={"error": "unauthorized"})
+
+
+def _parse_keys(env_var: str) -> frozenset[str]:
+    raw = os.environ.get(env_var, "")
+    return frozenset(key.strip() for key in raw.split(",") if key.strip())
+
+
+@lru_cache(maxsize=1)
+def _service_keys() -> frozenset[str]:
+    return _parse_keys("LEDGER_SERVICE_KEYS")
+
+
+@lru_cache(maxsize=1)
+def _admin_keys() -> frozenset[str]:
+    return _parse_keys("LEDGER_ADMIN_KEYS")
+
+
+def _matches_any(candidate: str, keys: frozenset[str]) -> bool:
+    return any(hmac.compare_digest(candidate, key) for key in keys)
+
+
+def require_service(x_api_key: str = Header(default=None)) -> str:
+    """Depends()-compatible: accepts a service OR admin key."""
+    if x_api_key and (
+        _matches_any(x_api_key, _service_keys()) or _matches_any(x_api_key, _admin_keys())
+    ):
+        return x_api_key
+    raise UNAUTHORIZED
+
+
+def require_admin(x_api_key: str = Header(default=None)) -> str:
+    """Depends()-compatible: admin key only."""
+    if x_api_key and _matches_any(x_api_key, _admin_keys()):
+        return x_api_key
+    raise UNAUTHORIZED