Spaces:
Paused
Paused
File size: 2,288 Bytes
f850954 8501103 f850954 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """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
|