#!/usr/bin/env python3 """ Agent Museum catalog verifier — prove the WHOLE collection is anchored to Bitcoin, trusting no one. Each exhibit is already independently verifiable (the disclosure + verifier.js on its page). This tool verifies the other axis: that the *entire published collection* — exactly this set of objects, each with this identity — is the set the museum committed to Bitcoin, and hasn't gained, lost, or altered an object since. It: 1. recomputes the catalog fingerprint from the public catalog manifest (the same JCS-canonical hash the museum publishes), and checks it, then 2. walks that fingerprint to Bitcoin: → the Touchstone entry that carries it (payload_hash), → the checkpoint that commits it (Merkle inclusion proof), → Bitcoin (the checkpoint's .ots). Dependency-free (Python standard library), imports no museum or Touchstone code — every hash below is reimplemented — so read it, then run it. Usage: python3 catalog-verify.py [--base=https://agentmuseum.org] """ import sys import json import hashlib import urllib.request KIND = "museum.catalog" FIELDS = ("accession_no", "slug", "title", "category", "occurred_at", "fingerprint_sha256") # --- clean-room JCS canonicalization (RFC 8785 subset) --- def _sort(v): if isinstance(v, dict): return {k: _sort(v[k]) for k in sorted(v.keys())} if isinstance(v, list): return [_sort(x) for x in v] return v def jcs(value): return json.dumps(_sort(value), separators=(",", ":"), ensure_ascii=False) # --- RFC 6962 Merkle (mirrors Touchstone's MerkleService) --- def _leaf(h): return hashlib.sha256(b"\x00" + bytes.fromhex(h)).hexdigest() def _node(l, r): return hashlib.sha256(b"\x01" + bytes.fromhex(l) + bytes.fromhex(r)).hexdigest() def fold_proof(entry_hash, proof): acc = _leaf(entry_hash) for s in proof: acc = _node(s["hash"], acc) if s["side"] == "left" else _node(acc, s["hash"]) return acc def recompute_entry_hash(e): return hashlib.sha256("\n".join([ str(e["seq"]), e["prev_hash"], e["server_ts"], e["payload_hash"], e["actor_sub"], e.get("counterparty_sub") or "", e["actor_sig"], ]).encode("utf-8")).hexdigest() def get_json(url): req = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": "museum-catalog-verify/1"}) with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode("utf-8")) def catalog_digest(objects): rows = sorted( [{k: o.get(k) for k in FIELDS} for o in objects], key=lambda r: (r["accession_no"], r["slug"]), ) return hashlib.sha256(jcs({"kind": KIND, "count": len(rows), "objects": rows}).encode("utf-8")).hexdigest() def contest_body_hash(c): # The exact cleartext body Touchstone hashed for a contest (ChainService::appendContest), # rebuilt from the channel's public fields; must equal payload_hash or the reason was altered. body = { "v": 1, "type": "contest", "target_digest": c.get("target_digest"), "target_ref": c.get("target_ref"), "reason": c.get("reason"), "contestant_sub": c.get("actor_sub"), "contestant_pubkey": c.get("contestant_pubkey"), } return hashlib.sha256(jcs(body).encode("utf-8")).hexdigest() def main(argv): base = "https://agentmuseum.org" for a in argv[1:]: if a.startswith("--base="): base = a.split("=", 1)[1] m = get_json(base.rstrip("/") + "/.well-known/agent-museum/catalog-manifest.json") published = m["catalog_sha256"] print("Agent Museum catalog verifier — is the whole collection anchored to Bitcoin?\n") print("Manifest : %d objects, generated %s" % (m.get("object_count"), m.get("generated_at"))) # --- [1/3] Membership: recompute the collection fingerprint from the objects. Independent. --- recomputed = catalog_digest(m["objects"]) mem_ok = (recomputed == published) print("\n[1/3] Collection membership (fully independent — clean-room recompute)") print(" recomputed catalog fingerprint : %s" % recomputed) print(" fingerprint the museum publishes: %s" % published) if mem_ok: print(" ✓ MATCH — this exact set of %d objects is the published collection fingerprint.\n" % m.get("object_count")) else: print(" ✗ MISMATCH — the objects shown do NOT hash to the published fingerprint.\n") # --- [2/3] Anchor: fingerprint → Touchstone entry → checkpoint → Bitcoin. --- seq = m.get("touchstone_seq") print("[2/3] Anchor (fingerprint → entry → checkpoint → Bitcoin)") anchor_ok = False feed = m.get("recorder_feed") if not feed and m.get("disclosure_url"): # derive the recorder feed from the disclosure bundle (older manifests) du = m["disclosure_url"] origin = du.split("/d/", 1)[0] rec = (get_json(du).get("recorder") or {}).get("public_id") feed = origin + "/.well-known/touchstone/checkpoints/" + rec if rec else None if seq is None or not feed: print(" — no anchor reference in the manifest (fingerprint recomputable, anchor not published).\n") else: try: pf = get_json(feed + "/entry/" + str(seq)) except Exception: pf = None if not pf: print(" · entry %s not exposed as an inclusion proof (recorder not opted in, or not yet checkpointed).\n" % seq) else: e, cp = pf["entry"], pf["checkpoint"] payload_ok = (e["payload_hash"] == published) eh_ok = (recompute_entry_hash(e) == e["entry_hash"]) root_ok = (fold_proof(e["entry_hash"], pf["inclusion_proof"]) == cp["merkle_root"]) print(" %s catalog fingerprint == entry.payload_hash (this collection IS that anchored entry)" % ("✓" if payload_ok else "✗")) print(" %s entry_hash recomputed from the entry envelope matches the published leaf" % ("✓" if eh_ok else "✗")) print(" %s inclusion proof folds to checkpoint.merkle_root %s… (%d hops)" % ("✓" if root_ok else "✗", cp["merkle_root"][:12], len(pf["inclusion_proof"]))) origin = feed.split("/.well-known/", 1)[0] print(" → checkpoint → Bitcoin (standard OpenTimestamps, zero trust in Touchstone):") print(" curl -O %s%s && ots verify %s" % (origin, cp["ots"], cp["ots"].rsplit("/", 1)[-1])) if payload_ok and eh_ok and root_ok: anchor_ok = True print(" ✓ FULLY BOUND — the collection fingerprint is committed, BY MERKLE INCLUSION, inside a") print(" recorder-signed checkpoint whose root is Bitcoin-anchored. Only Bitcoin is trusted.\n") else: print(" ✗ the anchor does NOT bind this collection — do not rely on it.\n") # --- [3/3] Standing: has anyone DISPUTED this collection's membership? --- print("[3/3] Standing (is the collection uncontested?)") contested = False cc = m.get("contest_channel") if not cc or not cc.get("recorder_feed"): print(" — no contest channel named in the manifest (nothing to check).\n") else: # `published` is a bare sha-256 hex digest — already URL-safe, no escaping needed. chan_url = cc["recorder_feed"].rstrip("/") + "/contests?target=" + published try: chan = get_json(chan_url) except Exception: chan = None anchored = [] for c in (chan or {}).get("contests", []): if c.get("target_digest") != published or not c.get("anchored"): continue cp = c["checkpoint"] if (contest_body_hash(c) == c.get("payload_hash") and recompute_entry_hash(c) == c.get("entry_hash") and fold_proof(c["entry_hash"], c["inclusion_proof"]) == cp["merkle_root"]): anchored.append(c) if anchored: contested = True print(" ✗ CONTESTED — %d anchored contest(s) against this collection fingerprint:" % len(anchored)) for c in anchored: print(" seq %s by %s: %s" % (c["seq"], c["actor_sub"], (c.get("reason") or "").strip()[:160])) print() else: latest = (chan or {}).get("latest_checkpoint") asof = ("as of checkpoint #%s" % latest["id"]) if latest else "(no checkpoint yet)" print(" ✓ CLEAR — no anchored contest against this collection %s.\n" % asof) print("RESULT: COLLECTION MEMBERSHIP %s%s%s" % ("VERIFIED ✓" if mem_ok else "FAILED ✗", " | ANCHORED TO BITCOIN ✓" if anchor_ok else "", " | CONTESTED ✗" if contested else (" | UNCONTESTED ✓" if (mem_ok and cc) else ""))) return 0 if mem_ok else 1 if __name__ == "__main__": sys.exit(main(sys.argv))