#!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.11" # dependencies = ["cryptography~=45.0"] # /// """Request temporary Dropbear catalog access and make DPoP-bound API calls.""" import argparse import base64 import hashlib import json import os import secrets import time import urllib.error import urllib.parse import urllib.request from pathlib import Path from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature ORIGIN = os.environ.get("DHT_OBSERVATORY", "https://dht.adamsmith.as").rstrip("/") STATE = Path(os.environ.get( "DHT_OBSERVATORY_STATE", Path.home() / ".cache/dht-observatory/credentials.json", )) CLIENT_ID = f"{ORIGIN}/clients/ad-hoc" def b64u(data: bytes) -> str: return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") def load_state() -> dict: if not STATE.exists(): raise SystemExit(f"No credentials at {STATE}; run `request` first") return json.loads(STATE.read_text()) def save_state(state: dict) -> None: STATE.parent.mkdir(mode=0o700, parents=True, exist_ok=True) temporary = STATE.with_suffix(".tmp") temporary.write_text(json.dumps(state, indent=2) + "\n") temporary.chmod(0o600) temporary.replace(STATE) def private_key(state: dict): return serialization.load_pem_private_key(state["private_key"].encode(), password=None) def dpop(key, method: str, htu: str, token: str | None = None) -> str: numbers = key.public_key().public_numbers() jwk = { "kty": "EC", "crv": "P-256", "x": b64u(numbers.x.to_bytes(32, "big")), "y": b64u(numbers.y.to_bytes(32, "big")), } header = {"typ": "dpop+jwt", "alg": "ES256", "jwk": jwk} claims = {"htm": method, "htu": htu, "iat": int(time.time()), "jti": secrets.token_urlsafe(18)} if token is not None: claims["ath"] = b64u(hashlib.sha256(token.encode("ascii")).digest()) signing_input = ".".join( b64u(json.dumps(part, separators=(",", ":")).encode()) for part in (header, claims) ) der = key.sign(signing_input.encode("ascii"), ec.ECDSA(hashes.SHA256())) r, s = decode_dss_signature(der) return signing_input + "." + b64u(r.to_bytes(32, "big") + s.to_bytes(32, "big")) def request_json(path: str, *, form: dict | None = None, state: dict | None = None) -> dict: method = "POST" if form is not None else "GET" url = ORIGIN + path htu = ORIGIN + urllib.parse.urlsplit(path).path headers = {"Accept": "application/json", "User-Agent": "dropbear-catalog-client/1"} body = None if form is not None: body = urllib.parse.urlencode(form).encode() headers["Content-Type"] = "application/x-www-form-urlencoded" if state and state.get("access_token"): token = state["access_token"] headers["Authorization"] = f"DPoP {token}" headers["DPoP"] = dpop(private_key(state), method, htu, token) elif state: headers["DPoP"] = dpop(private_key(state), method, htu) request = urllib.request.Request(url, data=body, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=20) as response: return json.load(response) except urllib.error.HTTPError as error: try: payload = json.load(error) except Exception: payload = {"error": error.reason} if error.code == 400 and payload.get("error") in {"authorization_pending", "slow_down"}: return payload raise SystemExit(f"HTTP {error.code}: {json.dumps(payload, ensure_ascii=False)}") from error def request_torrent(infohash: str, destination: Path) -> None: state = load_state() if not state.get("access_token") or state.get("expires_at", 0) <= time.time(): raise SystemExit("Access token missing or expired; run `request`, obtain approval, then run `token`") path = f"/v1/catalog/{urllib.parse.quote(infohash)}/torrent" url = ORIGIN + path token = state["access_token"] request = urllib.request.Request(url, headers={ "Accept": "application/x-bittorrent", "Authorization": f"DPoP {token}", "DPoP": dpop(private_key(state), "GET", url, token), "User-Agent": "dropbear-catalog-client/1", }) try: with urllib.request.urlopen(request, timeout=20) as response: body = response.read(2 * 1024 * 1024 + 1) except urllib.error.HTTPError as error: raise SystemExit(f"HTTP {error.code}: {error.reason}") from error if len(body) > 2 * 1024 * 1024: raise SystemExit("Server returned an oversized torrent") destination.write_bytes(body) print(f"Wrote {len(body)} bytes to {destination}") def request_access(scopes: str) -> None: key = ec.generate_private_key(ec.SECP256R1()) pem = key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption(), ).decode() grant = request_json("/oauth/device_authorization", form={"client_id": CLIENT_ID, "scope": scopes}) save_state({ "origin": ORIGIN, "private_key": pem, "device_code": grant["device_code"], "user_code": grant["user_code"], "expires_at": int(time.time()) + grant["expires_in"], "interval": grant["interval"], }) print(f"Authorize code: {grant['user_code']}") print(f"Operator command: ssh dropbear 'docker exec mainline-dht mainline-node approve {grant['user_code']}'") print("After approval, run: catalog-client.py token") def obtain_token() -> None: state = load_state() if state.get("expires_at", 0) <= time.time(): raise SystemExit("Device code expired; run `request` again") result = request_json("/oauth/token", form={ "grant_type": "urn:ietf:params:oauth:grant-type:device_code", "device_code": state["device_code"], "client_id": CLIENT_ID, }, state=state) if result.get("error"): print(result["error"] + ": " + result.get("error_description", "")) return state.pop("device_code", None) state.pop("user_code", None) state["access_token"] = result["access_token"] state["scope"] = result["scope"] state["expires_at"] = int(time.time()) + result["expires_in"] save_state(state) print(f"Token ready for {result['expires_in']} seconds with scopes: {result['scope']}") def revoke_token() -> None: state = load_state() token = state.get("access_token") if token: request_json("/oauth/revoke", form={"token": token, "client_id": CLIENT_ID}, state={ "private_key": state["private_key"], }) STATE.unlink(missing_ok=True) print(f"Revoked server token and removed {STATE}") def api(path: str) -> None: state = load_state() if not state.get("access_token") or state.get("expires_at", 0) <= time.time(): raise SystemExit("Access token missing or expired; run `request`, obtain approval, then run `token`") print(json.dumps(request_json(path, state=state), ensure_ascii=False, indent=2)) def main() -> None: parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="command", required=True) request_parser = sub.add_parser("request") request_parser.add_argument( "--scopes", default="catalog:read catalog:search runs:read torrent:read", help="space-separated scopes", ) sub.add_parser("token") list_parser = sub.add_parser("list") list_parser.add_argument("--limit", type=int, default=20) search_parser = sub.add_parser("search") search_parser.add_argument("query") search_parser.add_argument("--limit", type=int, default=20) detail_parser = sub.add_parser("get") detail_parser.add_argument("infohash") torrent_parser = sub.add_parser("torrent") torrent_parser.add_argument("infohash") torrent_parser.add_argument("output", nargs="?", type=Path) runs_parser = sub.add_parser("runs") runs_parser.add_argument("--limit", type=int, default=20) sub.add_parser("revoke") args = parser.parse_args() if args.command == "request": request_access(args.scopes) elif args.command == "token": obtain_token() elif args.command == "list": api(f"/v1/catalog?limit={args.limit}") elif args.command == "search": api("/v1/catalog/search?" + urllib.parse.urlencode({"q": args.query, "limit": args.limit})) elif args.command == "get": api(f"/v1/catalog/{urllib.parse.quote(args.infohash)}") elif args.command == "torrent": request_torrent(args.infohash, args.output or Path(f"{args.infohash.lower()}.torrent")) elif args.command == "runs": api(f"/v1/runs?limit={args.limit}") elif args.command == "revoke": revoke_token() if __name__ == "__main__": main()