rosdiff / api /keys.py
Chandra Kiran
Add Stage 1 MuJoCo scene generation API
f850954 unverified
Raw History Blame Contribute Delete
3.2 kB
"""Manage tester API keys in testers.yaml.
python -m api.keys add NAME [--note TEXT] create a key (printed once)
python -m api.keys list show testers (never keys)
python -m api.keys revoke NAME deactivate a tester
"""
from __future__ import annotations
import argparse
import datetime as dt
import secrets
import sys
from pathlib import Path
import yaml
from .auth import hash_key
from .config import get_settings
def _read(path: Path) -> dict:
if path.is_file():
return yaml.safe_load(path.read_text()) or {}
return {}
def _write(path: Path, data: dict) -> None:
header = "# Stage 1 tester allowlist. Manage with `python -m api.keys`; keys are stored as SHA-256 only.\n"
path.write_text(header + yaml.safe_dump(data, sort_keys=False))
path.chmod(0o600)
def add(path: Path, name: str, note: str = "") -> str:
data = _read(path)
testers = data.setdefault("testers", []) or []
data["testers"] = testers
if any(t["name"] == name and t.get("active", True) for t in testers):
raise ValueError(f"an active tester named {name!r} already exists (revoke it first)")
key = "rd_" + secrets.token_urlsafe(24)
testers.append(
{
"name": name,
"key_sha256": hash_key(key),
"active": True,
"added": dt.date.today().isoformat(),
"note": note,
}
)
_write(path, data)
return key
def revoke(path: Path, name: str) -> int:
data = _read(path)
n = 0
for t in data.get("testers") or []:
if t["name"] == name and t.get("active", True):
t["active"] = False
t["revoked"] = dt.date.today().isoformat()
n += 1
if n:
_write(path, data)
return n
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--file", type=Path, help="default: $ROSDIFF_TESTERS_FILE or ./testers.yaml")
sub = parser.add_subparsers(dest="cmd", required=True)
p = sub.add_parser("add")
p.add_argument("name")
p.add_argument("--note", default="")
sub.add_parser("list")
p = sub.add_parser("revoke")
p.add_argument("name")
args = parser.parse_args(argv)
path = args.file or get_settings().testers_file
if args.cmd == "add":
try:
key = add(path, args.name, args.note)
except ValueError as e:
print(f"error: {e}", file=sys.stderr)
return 1
print(f"Added tester {args.name!r} to {path}.")
print(f"API key (shown once, send it to the tester privately):\n\n {key}\n")
return 0
if args.cmd == "list":
for t in _read(path).get("testers") or []:
state = "active " if t.get("active", True) else "revoked"
print(f"{state} {t['name']:<20} added {t.get('added', '?')} {t.get('note', '')}")
return 0
n = revoke(path, args.name)
print(f"revoked {n} key(s) for {args.name!r}" if n else f"no active tester named {args.name!r}")
return 0 if n else 1
if __name__ == "__main__":
sys.exit(main())