#!/usr/bin/env python3 """fas_registry_client.py — join and publish to the Frederick & Sons Trust Registry. The registry (https://frederickandsons.xyz/registry/) is a public, non-custodial log of self-signed did:key attestations. This is the ISSUER-side companion to the offline verifier (verify_registry_record.py): generate a keypair, register your identity, publish attestations, rotate or revoke your key — from one file, with nothing to install beyond `cryptography`. pip install cryptography # the only dependency; everything else is stdlib Quickstart: python3 fas_registry_client.py keygen python3 fas_registry_client.py identity --label "My Service" --endpoint https://example.com python3 fas_registry_client.py attest --statement "example.com is our production domain" python3 fas_registry_client.py verify record.json # offline, no network Your key NEVER leaves this machine: records are signed locally and only the signed JSON is sent. The registry holds no keys and issues nothing — your self-signature is the only write credential. Anyone can verify what you publish, offline, with verify_registry_record.py. Registration mines a small proof-of-work stamp (~2^20 sha256 attempts, sub-second on any modern machine) — an anti-flood tax, not gatekeeping. Writes are permanent (append-only log): publish only what you are happy to stand behind forever. Use --dry-run to build + validate a record against the live registry's /verify endpoint WITHOUT writing anything. Canonicalization note (do not "fix"): the JSON serialization below (sort_keys, tight separators, ensure_ascii=False, JCS-safe subset enforced) must stay byte-identical to the registry server and the standalone verifier — any deviation makes your signatures invalid. """ from __future__ import annotations import argparse import hashlib import json import os import sys import urllib.error import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any try: from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey except ImportError: # pragma: no cover sys.exit("this tool needs the 'cryptography' package: pip install cryptography") DEFAULT_REGISTRY = "https://frederickandsons.xyz" DEFAULT_KEY_PATH = Path.home() / ".fas-registry-key" SIGNED_OVER = "fas-registry-record/v1" POW_BITS = 20 # server-enforced minimum difficulty for identity registration MAX_STATEMENT_CHARS = 512 # server bounds — checked client-side so you fail fast, locally MAX_LABEL_CHARS = 128 MAX_REFS = 16 MAX_ENDPOINTS = 4 # --- canonicalization (byte-identical to the registry server + verify_registry_record.py) ----- _JCS_MAX_INT = (2**53) - 1 _JCS_MIN_INT = -((2**53) - 1) _ASTRAL_PLANE_START = 0x10000 def assert_jcs_safe(obj: Any) -> None: if isinstance(obj, bool): return if isinstance(obj, float): raise ValueError(f"JCS-unsafe: float value {obj!r} (float/NaN/Infinity have no JCS representation)") if isinstance(obj, int): if not (_JCS_MIN_INT <= obj <= _JCS_MAX_INT): raise ValueError(f"JCS-unsafe: integer {obj!r} exceeds +/-(2**53-1)") return if obj is None or isinstance(obj, str): return if isinstance(obj, list): for item in obj: assert_jcs_safe(item) return if isinstance(obj, dict): for key, value in obj.items(): if not isinstance(key, str): raise ValueError(f"JCS-unsafe: dict key {key!r} is not a string") if any(ord(ch) >= _ASTRAL_PLANE_START for ch in key): raise ValueError(f"JCS-unsafe: dict key {key!r} contains an astral-plane (>= U+10000) code point") assert_jcs_safe(value) return raise ValueError(f"JCS-unsafe: value of type {type(obj).__name__} is not JSON-safe: {obj!r}") def canonical_bytes(obj: Any) -> bytes: assert_jcs_safe(obj) return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def canonical_now() -> str: dt = datetime.now(timezone.utc) return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond:06d}Z" def record_id(record: dict) -> str: body_for_hash = {k: v for k, v in record.items() if k not in ("id", "proof")} return hashlib.sha256(canonical_bytes(body_for_hash)).hexdigest() # --- did:key (ed25519, multicodec 0xed01, base58btc, 'z' prefix) ------------------------------ _B58_ALPHABET = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" def b58encode(data: bytes) -> str: n_pad = len(data) - len(data.lstrip(b"\x00")) num = int.from_bytes(data, "big") out = [] while num: num, rem = divmod(num, 58) out.append(_B58_ALPHABET[rem]) out.reverse() return ("1" * n_pad) + "".join(out) def did_key_from_pubkey_hex(pubkey_hex: str) -> str: raw = bytes.fromhex(pubkey_hex) if len(raw) != 32: raise ValueError(f"ed25519 public key must be 32 raw bytes, got {len(raw)}") return "did:key:z" + b58encode(bytes((0xED, 0x01)) + raw) # --- proof of work (identity registration only) ------------------------------------------------ def mine_pow(did: str, bits: int = POW_BITS) -> str: """Find a nonce with >= `bits` leading zero bits in sha256(did+nonce). ~2^20 attempts at the default — sub-second in CPython's C-accelerated hashlib.""" attempt = 0 while True: nonce = format(attempt, "x") digest = hashlib.sha256((did + nonce).encode("utf-8")).digest() count = 0 for byte in digest: if byte == 0: count += 8 continue count += 8 - byte.bit_length() break if count >= bits: return nonce attempt += 1 # --- key handling (local only — the private key never leaves this machine) --------------------- def keygen(key_path: Path) -> str: if key_path.exists(): sys.exit(f"refusing to overwrite existing key at {key_path} — move it first if you really " "mean to (a lost key cannot be recovered; a replaced one must be rotated on-registry)") sk = Ed25519PrivateKey.generate() raw = sk.private_bytes(encoding=serialization.Encoding.Raw, format=serialization.PrivateFormat.Raw, encryption_algorithm=serialization.NoEncryption()) key_path.write_text(raw.hex() + "\n", encoding="utf-8") os.chmod(key_path, 0o600) return did_of(key_path) def _load_key(key_path: Path) -> Ed25519PrivateKey: if not key_path.exists(): sys.exit(f"no key at {key_path} — run `keygen` first (or pass --key)") return Ed25519PrivateKey.from_private_bytes(bytes.fromhex(key_path.read_text().strip())) def _pubkey_hex(sk: Ed25519PrivateKey) -> str: return sk.public_key().public_bytes(encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw).hex() def did_of(key_path: Path) -> str: return did_key_from_pubkey_hex(_pubkey_hex(_load_key(key_path))) # --- record building (pure — no network; every record is self-verifying by construction) ------- def build_record(sk: Ed25519PrivateKey, rtype: str, subject: str | None, body: dict) -> dict: pub_hex = _pubkey_hex(sk) did = did_key_from_pubkey_hex(pub_hex) body_for_hash = { "fas_registry": "v1", "type": rtype, "issuer": did, "subject": subject or did, "issued_at": canonical_now(), "body": body, } cb = canonical_bytes(body_for_hash) return { **body_for_hash, "id": hashlib.sha256(cb).hexdigest(), "proof": { "algo": "ed25519", "signed_over": SIGNED_OVER, "public_key": pub_hex, "signature": sk.sign(cb).hex(), }, } def build_identity(sk: Ed25519PrivateKey, label: str, endpoints: list[dict]) -> dict: if len(label) > MAX_LABEL_CHARS: sys.exit(f"label exceeds {MAX_LABEL_CHARS} chars (server would reject it)") if len(endpoints) > MAX_ENDPOINTS: sys.exit(f"more than {MAX_ENDPOINTS} endpoints (server would reject it)") did = did_key_from_pubkey_hex(_pubkey_hex(sk)) print(f"mining proof-of-work ({POW_BITS} bits — typically < 1s) ...", file=sys.stderr) nonce = mine_pow(did) body = {"label": label, "endpoints": endpoints, "pow": {"algo": "sha256-leading-zeros", "nonce": nonce, "bits": POW_BITS}} return build_record(sk, "identity", None, body) def build_attestation(sk: Ed25519PrivateKey, statement: str, subject: str | None, refs: list[str], claims: dict) -> dict: if len(statement) > MAX_STATEMENT_CHARS: sys.exit(f"statement exceeds {MAX_STATEMENT_CHARS} chars (server would reject it)") if len(refs) > MAX_REFS: sys.exit(f"more than {MAX_REFS} refs (server would reject it)") return build_record(sk, "attestation", subject, {"statement": statement, "claims": claims, "refs": refs}) _RECEIPT_OUTCOMES = ("settled", "final", "contested", "failed", "pending") def build_receipt(sk: Ed25519PrivateKey, settlement_id: str, outcome: str, amount: str | None, asset_id: str | None, ledger: str | None, parties: dict | None, tx_refs: list[str], audit_head: str | None) -> dict: """A settlement receipt: subject == settlement_id. `amount`, if given, is integer BASE UNITS as a decimal string — never a float (a float would be rejected by the registry's JCS guard and is lossy for money). The receipt makes an external settlement verifiable without the settler storing per-settlement state.""" if outcome not in _RECEIPT_OUTCOMES: sys.exit(f"outcome must be one of {_RECEIPT_OUTCOMES}") if amount is not None and not (amount.isascii() and amount.isdigit() and len(amount) <= 32 and (amount == "0" or amount[0] != "0")): sys.exit("amount must be integer base units as a decimal string <= 32 digits (e.g. '1000000'), not a float") if len(tx_refs) > MAX_REFS: sys.exit(f"more than {MAX_REFS} tx_refs (server would reject it)") body: dict[str, Any] = {"settlement_id": settlement_id, "outcome": outcome, "tx_refs": tx_refs} if amount is not None: body["amount"] = amount for k, v in (("asset_id", asset_id), ("ledger", ledger), ("audit_head", audit_head)): if v is not None: body[k] = v if parties is not None: body["parties"] = parties return build_record(sk, "receipt", settlement_id, body) # --- offline verify (same algorithm as verify_registry_record.py) ------------------------------ def verify_record(record: Any) -> tuple[bool, list[str]]: if not isinstance(record, dict): return False, ["record is not a JSON object"] required = {"fas_registry", "type", "id", "issuer", "subject", "issued_at", "body", "proof"} missing, extra = required - record.keys(), set(record.keys()) - required if missing or extra: return False, [f"missing: {sorted(missing)}, unexpected: {sorted(extra)}"] proof = record.get("proof") if not isinstance(proof, dict): return False, ["proof must be an object"] preq = {"algo", "signed_over", "public_key", "signature"} if preq - proof.keys() or proof.keys() - preq: return False, ["proof keys must be exactly algo/signed_over/public_key/signature"] if proof.get("algo") != "ed25519" or proof.get("signed_over") != SIGNED_OVER: return False, [f"unsupported proof.algo/signed_over: {proof.get('algo')!r}/{proof.get('signed_over')!r}"] body_for_hash = {k: v for k, v in record.items() if k not in ("id", "proof")} try: cb = canonical_bytes(body_for_hash) except (TypeError, ValueError) as exc: return False, [f"record is not JSON-canonicalizable: {exc}"] computed = hashlib.sha256(cb).hexdigest() if record.get("id") != computed: return False, [f"id does not match content (computed {computed})"] try: pub_raw = bytes.fromhex(proof["public_key"]) sig_raw = bytes.fromhex(proof["signature"]) Ed25519PublicKey.from_public_bytes(pub_raw).verify(sig_raw, cb) except InvalidSignature: return False, ["signature does not verify"] except Exception as exc: return False, [f"signature verification error: {exc}"] if record.get("issuer") != did_key_from_pubkey_hex(proof["public_key"]): return False, ["record.issuer does not match did:key(proof.public_key)"] return True, [] # --- network ------------------------------------------------------------------------------------ def _post(registry: str, path: str, record: dict) -> tuple[int, dict]: # An explicit User-Agent is REQUIRED, not cosmetic: the registry sits behind Cloudflare, whose # Browser Integrity Check hard-bans UA-less requests (urllib's default) with `error code: 1010`. # Identify honestly as this client. req = urllib.request.Request(registry.rstrip("/") + path, data=json.dumps(record).encode("utf-8"), headers={"Content-Type": "application/json", "Accept": "application/json", "User-Agent": "fas-registry-client/1.0"}, method="POST") try: with urllib.request.urlopen(req, timeout=20) as resp: return resp.status, json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: try: return exc.code, json.loads(exc.read().decode("utf-8")) except Exception: return exc.code, {"error": "unreadable error body"} except urllib.error.URLError as exc: sys.exit(f"cannot reach {registry}: {exc.reason}") def publish(registry: str, path: str, record: dict, dry_run: bool) -> None: ok, reasons = verify_record(record) # never send a record we can't verify ourselves if not ok: sys.exit(f"BUG: locally-built record failed self-verification: {reasons}") if dry_run: status, resp = _post(registry, "/registry/v1/verify", record) print(json.dumps(record, indent=2, ensure_ascii=False)) print(f"\n--dry-run: registry /verify says: {resp} (HTTP {status}) — nothing was written", file=sys.stderr) return status, resp = _post(registry, path, record) if status in (200, 201): tag = "published" if status == 201 else "already present (idempotent)" print(f"{tag}: id={resp.get('id', record['id'])} seq={resp.get('seq')}") print(f"view: {registry.rstrip('/')}/registry/v1/records/{record['id']}") else: sys.exit(f"registry rejected the record (HTTP {status}): {resp}") # --- CLI ---------------------------------------------------------------------------------------- def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__.splitlines()[0], formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--key", type=Path, default=DEFAULT_KEY_PATH, help=f"private-key file (default {DEFAULT_KEY_PATH})") ap.add_argument("--registry", default=DEFAULT_REGISTRY, help=f"registry base URL (default {DEFAULT_REGISTRY})") sub = ap.add_subparsers(dest="cmd", required=True) sub.add_parser("keygen", help="generate an ed25519 keypair and print your did:key") sub.add_parser("whoami", help="print the did:key for your key file") p_id = sub.add_parser("identity", help="register your self-signed identity (mines proof-of-work)") p_id.add_argument("--label", required=True, help=f"display label (<= {MAX_LABEL_CHARS} chars)") p_id.add_argument("--endpoint", action="append", default=[], help=f"service URL (repeatable, <= {MAX_ENDPOINTS})") p_id.add_argument("--dry-run", action="store_true", help="validate against the registry without writing") p_at = sub.add_parser("attest", help="publish a signed attestation") p_at.add_argument("--statement", required=True, help=f"the claim you stand behind (<= {MAX_STATEMENT_CHARS} chars)") p_at.add_argument("--subject", default=None, help="DID the attestation is about (default: yourself)") p_at.add_argument("--ref", action="append", default=[], help=f"supporting URL/hash (repeatable, <= {MAX_REFS})") p_at.add_argument("--claims", default="{}", help='JSON object of structured claims (default {})') p_at.add_argument("--dry-run", action="store_true", help="validate against the registry without writing") p_rc = sub.add_parser("receipt", help="publish a settlement receipt (subject = settlement_id)") p_rc.add_argument("--settlement-id", required=True, help="the settlement this receipt attests") p_rc.add_argument("--outcome", required=True, choices=_RECEIPT_OUTCOMES) p_rc.add_argument("--amount", default=None, help="integer BASE UNITS as a decimal string (never a float)") p_rc.add_argument("--asset-id", default=None) p_rc.add_argument("--ledger", default=None, help="e.g. xrpl") p_rc.add_argument("--from", dest="party_from", default=None, help="paying party (r-address or DID)") p_rc.add_argument("--to", dest="party_to", default=None, help="receiving party (r-address or DID)") p_rc.add_argument("--tx-ref", action="append", default=[], help=f"settlement tx hash (repeatable, <= {MAX_REFS})") p_rc.add_argument("--audit-head", default=None, help="settler audit-chain head at settlement time") p_rc.add_argument("--dry-run", action="store_true", help="validate against the registry without writing") p_ro = sub.add_parser("rotate", help="rotate your identity to a successor key (permanent)") p_ro.add_argument("--new-controller", required=True, help="successor did:key") p_rv = sub.add_parser("revoke", help="revoke your identity outright (permanent, terminal)") p_rv.add_argument("--reason", default="", help="optional public reason") p_vf = sub.add_parser("verify", help="verify a record JSON file offline (no network)") p_vf.add_argument("file", type=Path) args = ap.parse_args(argv) if args.cmd == "keygen": did = keygen(args.key) print(f"key written to {args.key} (0600) — BACK IT UP; it cannot be recovered") print(f"your identity: {did}") return 0 if args.cmd == "whoami": print(did_of(args.key)) return 0 if args.cmd == "verify": rec = json.loads(args.file.read_text(encoding="utf-8")) ok, reasons = verify_record(rec) print("VALID" if ok else "INVALID") for r in reasons: print(f" - {r}") return 0 if ok else 1 sk = _load_key(args.key) if args.cmd == "identity": endpoints = [{"type": "service", "url": u} for u in args.endpoint] rec = build_identity(sk, args.label, endpoints) publish(args.registry, "/registry/v1/identities", rec, args.dry_run) elif args.cmd == "attest": claims = json.loads(args.claims) if not isinstance(claims, dict): sys.exit("--claims must be a JSON object") rec = build_attestation(sk, args.statement, args.subject, args.ref, claims) publish(args.registry, "/registry/v1/attestations", rec, args.dry_run) elif args.cmd == "receipt": parties = None if args.party_from or args.party_to: parties = {} if args.party_from: parties["from"] = args.party_from if args.party_to: parties["to"] = args.party_to rec = build_receipt(sk, args.settlement_id, args.outcome, args.amount, args.asset_id, args.ledger, parties, args.tx_ref, args.audit_head) publish(args.registry, "/registry/v1/receipts", rec, args.dry_run) elif args.cmd == "rotate": did = did_key_from_pubkey_hex(_pubkey_hex(sk)) rec = build_record(sk, "rotation", did, {"new_controller": args.new_controller}) publish(args.registry, f"/registry/v1/identities/{did}/rotate", rec, False) elif args.cmd == "revoke": did = did_key_from_pubkey_hex(_pubkey_hex(sk)) print("REVOCATION IS TERMINAL — your DID can never be reactivated. Ctrl-C now to abort.", file=sys.stderr) rec = build_record(sk, "revocation", did, {"reason": args.reason}) publish(args.registry, f"/registry/v1/identities/{did}/revoke", rec, False) return 0 if __name__ == "__main__": raise SystemExit(main())