"""API-key allowlist for Stage 1 testers. Keys live in ``testers.yaml`` as SHA-256 hashes (the plain key is shown once, when it is created). The file is re-read whenever it changes, so adding or revoking a tester needs no restart:: python -m api.keys add alice # prints alice's new key once python -m api.keys list python -m api.keys revoke alice For quick local runs, ``ROSDIFF_API_KEYS="name:key,name2:key2"`` adds plain keys from the environment as well. """ from __future__ import annotations import hashlib import hmac import os import threading from dataclasses import dataclass from pathlib import Path import yaml def hash_key(key: str) -> str: return hashlib.sha256(key.encode()).hexdigest() @dataclass(frozen=True) class Tester: name: str key_sha256: str active: bool = True class Allowlist: def __init__(self, path: Path): self.path = Path(path) self._lock = threading.Lock() self._mtime: float | None = None self._testers: list[Tester] = [] def _load(self) -> list[Tester]: testers: list[Tester] = [] if self.path.is_file(): data = yaml.safe_load(self.path.read_text()) or {} for t in data.get("testers") or []: testers.append(Tester(str(t["name"]), str(t["key_sha256"]).lower(), bool(t.get("active", True)))) for pair in filter(None, os.environ.get("ROSDIFF_API_KEYS", "").split(",")): name, _, key = pair.strip().partition(":") if name and key: testers.append(Tester(name, hash_key(key))) return testers def testers(self) -> list[Tester]: with self._lock: mtime = self.path.stat().st_mtime if self.path.is_file() else None if mtime != self._mtime or self._mtime is None: self._testers = self._load() self._mtime = mtime return list(self._testers) def identify(self, key: str | None) -> Tester | None: """The active tester owning ``key``, or None.""" if not key: return None digest = hash_key(key.strip()) for t in self.testers(): if t.active and hmac.compare_digest(t.key_sha256, digest): return t return None