Spaces:
Paused
Paused
Download api/auth.py from chandrakiran06/rosdiff: direct link, hf CLI and curl.
- Browser
- Download file 2.29 kB
-
https://huggingface.co/spaces/chandrakiran06/rosdiff/resolve/main/api/auth.py
- Command line
-
hf download hf://spaces/chandrakiran06/rosdiff/api/auth.py
-
curl -L -o auth.py https://huggingface.co/spaces/chandrakiran06/rosdiff/resolve/main/api/auth.py
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() | |
| 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 | |