alexveley
143 lines
4.2 KiB
Python
143 lines
4.2 KiB
Python
"""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}")
|