rosdiff / api /auth.py
Chandrakiran-08
Deploy to HF Spaces
8501103
Raw History Blame Contribute Delete
2.29 kB
"""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