"""Hashed API-token clients for Firefox Agent Bridge. The host stores SHA-256(token) only. The raw secret is shown once in the extension UI (or CLI) and then lives wherever the user keeps secrets. """ from __future__ import annotations import hashlib import hmac import json import os import re import secrets from datetime import datetime, timezone from pathlib import Path from typing import Any STATE_DIR = Path(os.environ.get("LOCALAPPDATA", str(Path.home()))) / "firefox-agent-bridge" CLIENTS_PATH = STATE_DIR / "clients.json" NAME_RE = re.compile(r"^[A-Za-z0-9._-]{1,64}$") TOKEN_PREFIX = "fab_" def _now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat() def token_hash(token: str) -> str: return hashlib.sha256(token.encode("ascii")).hexdigest() def load_clients() -> list[dict[str, Any]]: if not CLIENTS_PATH.exists(): return [] data = json.loads(CLIENTS_PATH.read_text(encoding="utf-8")) return list(data.get("clients") or []) def save_clients(clients: list[dict[str, Any]]) -> None: STATE_DIR.mkdir(parents=True, exist_ok=True) CLIENTS_PATH.write_text( json.dumps({"clients": clients}, indent=2) + "\n", encoding="utf-8", ) def public_client(row: dict[str, Any]) -> dict[str, Any]: return { "id": row.get("id"), "name": row.get("name"), "created": row.get("created"), "last_used": row.get("last_used"), } def list_clients() -> list[dict[str, Any]]: return [public_client(row) for row in load_clients()] def create_client(name: str) -> dict[str, str]: name = (name or "").strip() if not NAME_RE.match(name): raise ValueError("name must be 1-64 characters: A-Za-z0-9._-") clients = load_clients() if any(row.get("name") == name for row in clients): raise ValueError(f"client already registered: {name}") token = TOKEN_PREFIX + secrets.token_hex(32) client_id = secrets.token_hex(16) clients.append( { "id": client_id, "name": name, "token_hash": token_hash(token), "created": _now(), "last_used": None, } ) save_clients(clients) return {"id": client_id, "name": name, "token": token} def revoke_client(name_or_id: str) -> str: clients = load_clients() kept = [row for row in clients if row.get("id") != name_or_id and row.get("name") != name_or_id] if len(kept) == len(clients): raise ValueError(f"no client {name_or_id!r}") save_clients(kept) return name_or_id def touch_last_used(client_id: str) -> None: clients = load_clients() changed = False stamp = _now() for row in clients: if row.get("id") == client_id: row["last_used"] = stamp changed = True break if changed: save_clients(clients) def verify_bearer(headers: dict[str, str]) -> dict[str, Any]: auth = "" for key, value in headers.items(): if key.lower() == "authorization": auth = (value or "").strip() break if not auth.lower().startswith("bearer "): raise PermissionError("Bearer token required") token = auth[7:].strip() if not token.startswith(TOKEN_PREFIX): raise PermissionError("unrecognized token") digest = token_hash(token) matched = None for row in load_clients(): stored = row.get("token_hash") or "" if stored and hmac.compare_digest(stored, digest): matched = row break if not matched: raise PermissionError("unknown token") touch_last_used(str(matched["id"])) return public_client(matched) def handle_admin(method: str, args: list[Any] | None = None) -> Any: args = args or [] if method == "clients.list": return list_clients() if method == "clients.create": name = args[0] if args else "" if isinstance(name, dict): name = name.get("name") or "" return create_client(str(name)) if method == "clients.revoke": target = args[0] if args else "" if isinstance(target, dict): target = target.get("id") or target.get("name") or "" return {"revoked": revoke_client(str(target))} raise ValueError(f"unknown admin method: {method}")