#!/usr/bin/env python3 """ compose-verify — verify a MULTI-AGENT decision: N per-agent receipts bound into one, each anchored, independence preserved. A single agent's receipt proves one chain. A multi-agent decision is a COMPOSITION: several agents each sign their own contribution into their OWN operator chain, and a composition receipt names the set — {kind:"touchstone.composition", decision, constituents:[{recorder, seq, entry_hash, role}]}. Binding them in ONE shared chain would be wrong: that chain has ONE operator, so the whole decision's collusion floor collapses to k=1 (one party could rewrite the framing). Linked receipts keep each agent's own provenance; the composition only names them, and its strength is the DISTINCT-CONTROL count across the contributing chains. Given the composition receipt (recorder + seq), this checks, trusting no server: 1. the composition body re-derives its digest: sha256(JCS(body)) == the receipt's committed payload_hash; 2. the composition receipt itself folds through its Merkle inclusion proof to a Bitcoin-anchored checkpoint — so the WHOLE named set is anchored, not just asserted; 3. every constituent folds to Bitcoin too, AND its live entry_hash equals the one named in the composition — so the composition names the exact entries, and none can be swapped after the fact; 4. the collusion floor: k = the number of distinct controlling identities (subject subs) across the constituent recorders. Two constituents under one operator collapse to one — so k is the minimum independent parties who must collude to forge the decision. UPPER BOUND: hidden shared control leaves no bytes, so undisclosed correlation can only LOWER k (tighten it with equivocation-check on each constituent — is that recorder really that operator's? — and with the collusion-floor model). Reuses standing-verify.py's primitives (jcs / recompute_entry_hash / fold_proof / get_json) — fetch it alongside. Dep-free. python3 compose-verify.py --recorder= --seq= [--base=https://touchstone.cv] Exit: 0 = COMPOSED (all constituents + the composition fold to Bitcoin) · 1 = NON-CONFORMANT · 2 = unanchored / malformed / missing sibling / unreachable. """ import os import sys import hashlib import importlib.util import urllib.error _HERE = os.path.dirname(os.path.abspath(__file__)) def _load_sibling(filename): path = os.path.join(_HERE, filename) if not os.path.exists(path): raise FileNotFoundError(filename) spec = importlib.util.spec_from_file_location(filename.replace("-", "_").replace(".py", ""), path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod def _entry_anchored(sv, base, recorder, seq): """Fetch the entry's inclusion proof and fold it to Bitcoin. Returns (ok, entry, checkpoint, reason).""" try: d = sv.get_json("%s/.well-known/touchstone/checkpoints/%s/entry/%d" % (base, recorder, seq)) except urllib.error.HTTPError as e: return False, None, None, ("not checkpointed / not inclusion-public (HTTP %d)" % e.code) except Exception as e: # noqa: BLE001 return False, None, None, ("unreachable (%s)" % e) entry, cp = d.get("entry") or {}, d.get("checkpoint") or {} if sv.recompute_entry_hash(entry) != entry.get("entry_hash"): return False, entry, cp, "entry_hash does not recompute from the envelope" if not (d.get("inclusion_proof") is not None and cp.get("merkle_root") and sv.fold_proof(entry.get("entry_hash"), d.get("inclusion_proof")) == cp.get("merkle_root")): return False, entry, cp, "inclusion proof does not fold to the checkpoint root" return True, entry, cp, "folds to checkpoint #%s" % cp.get("id") def _subject_of(sv, base, recorder): """The controlling identity a constituent recorder attributes itself to (its subject sub).""" try: f = sv.get_json("%s/.well-known/touchstone/checkpoints/%s" % (base, recorder)) return f.get("subject_sub") except Exception: # noqa: BLE001 return None def verify(recorder, seq, base): """Returns (exit_code, verdict, lines).""" lines = ["composition — %s seq %d" % (recorder, seq)] try: sv = _load_sibling("standing-verify.py") except FileNotFoundError: return 2, "MISSING_SIBLING", lines + ["fetch standing-verify.py alongside this file (jcs / " "recompute_entry_hash / fold_proof / get_json)."] try: doc = sv.get_json("%s/composition/%s/%d" % (base, recorder, seq)) except urllib.error.HTTPError as e: return 2, "NOT_A_COMPOSITION", lines + ["no composition receipt at %s seq %d (HTTP %d)" % (recorder, seq, e.code)] except Exception as e: # noqa: BLE001 return 2, "UNREACHABLE", lines + ["could not fetch the composition (%s)" % e] body = doc.get("composition") or {} constituents = body.get("constituents") or [] # 1. the composition body re-derives its committed digest. digest = hashlib.sha256(sv.jcs(body).encode("utf-8")).hexdigest() dig_ok = digest == doc.get("composition_sha256") lines.append(" %s composition body → digest %s… %s committed" % ("✓" if dig_ok else "✗", digest[:12], "==" if dig_ok else "!=")) # 2. the composition receipt itself folds to Bitcoin, and the anchored entry commits to this digest. c_ok, c_entry, c_cp, c_why = _entry_anchored(sv, base, recorder, seq) anchored_to_digest = bool(c_entry and c_entry.get("payload_hash") == doc.get("composition_sha256")) lines.append(" %s composition anchored — %s%s" % ( "✓" if (c_ok and anchored_to_digest) else ("·" if not c_ok else "✗"), c_why, "" if anchored_to_digest else " (but the anchored entry's payload_hash != composition digest)")) composition_bound = dig_ok and c_ok and anchored_to_digest if not constituents: return 1, "NON_CONFORMANT", lines + ["the composition names no constituents — nothing composed."] # 3. every constituent folds to Bitcoin AND its live entry_hash matches the one named here. lines.append(" constituents (%d):" % len(constituents)) all_bound = composition_bound controllers = {} # subject_sub -> count for c in constituents: cr, cs, ceh = c.get("recorder"), c.get("seq"), c.get("entry_hash") role = c.get("role") or "—" ok, entry, _cp, why = _entry_anchored(sv, base, cr, int(cs) if cs is not None else -1) named_ok = bool(entry and entry.get("entry_hash") == ceh) sub = _subject_of(sv, base, cr) if sub: controllers[sub] = controllers.get(sub, 0) + 1 good = ok and named_ok all_bound = all_bound and good lines.append(" %s [%s] %s seq %s — %s%s%s" % ( "✓" if good else "✗", role, cr, cs, why, "" if named_ok else " · NAMED HASH != live entry_hash (swapped)", (" · control=%s" % (sub[:8] + "…" if sub else "unknown")))) # 4. collusion floor = distinct controlling identities across the constituents. k = len(controllers) lines.append("") if not all_bound: lines.append("RESULT: NON_CONFORMANT ✗ — a constituent (or the composition) does not fold to Bitcoin, or a " "named entry_hash != the live one. Do not treat this as one bound decision.") return 1, "NON_CONFORMANT", lines lines.append("RESULT: COMPOSED ✓ — the composition and all %d constituents fold to Bitcoin, each named exactly. " "Collusion floor k = %d distinct controlling %s must collude to forge this decision (UPPER bound; " "hidden shared control can only lower it — tighten with equivocation-check per constituent)." % (len(constituents), k, "identity" if k == 1 else "identities")) return 0, "COMPOSED", lines def _kv(argv, key, default=None): for a in argv: if a.startswith(key + "="): return a.split("=", 1)[1] return default def main(argv): recorder = _kv(argv[1:], "--recorder") seq = _kv(argv[1:], "--seq") base = _kv(argv[1:], "--base", "https://touchstone.cv").rstrip("/") if not recorder or seq is None: print(__doc__) return 2 code, _verdict, lines = verify(recorder, int(seq), base) for ln in lines: print(ln) return code if __name__ == "__main__": sys.exit(main(sys.argv))