#!/usr/bin/env python3 """ verify-apa.py - independent verifier for Rubric agent payment attestation anchors. Reimplements the tier-2 aggregation from the published specification (https://rubric-protocol.com/docs/apa-v1-binding-profile). Uses no Rubric code and no Rubric network calls: the anchor is read from Hedera's public mirror node. Usage: python3 verify-apa.py proof.json where proof.json is the body of GET /v1/proof/{attestationId}. The holder of an attestation obtains that package and hands it to the verifying party; the verifying party needs nothing from Rubric to check it. Exit 0 if the chain verifies, 1 otherwise. """ import base64 import hashlib import json import sys import urllib.request MIRROR = "https://mainnet-public.mirrornode.hedera.com/api/v1/topics/{t}/messages/{s}" def canon(obj): """RFC 8785-compatible for the flat, ASCII-keyed objects used here.""" return json.dumps(obj, sort_keys=True, separators=(",", ":")) def H(algo, data): return hashlib.new(algo.replace("-", "_"), data).hexdigest() def make_leaf_v2(algo, leaf_type, data): """hashData(): algorithm-prefixed digest over the canonical typed object.""" typed = {"__leafType": leaf_type} typed.update(data) return algo + ":" + H(algo, canon(typed).encode()) def build_tree_v3(algo, leaf_hashes): """RFC 6962 domain tags. The 'algo:' prefix is stripped before hex-parsing.""" def domain_leaf(x): return H(algo, bytes([0x00]) + bytes.fromhex(x.split(":")[-1])) def domain_node(l, r): return H(algo, bytes([0x01]) + bytes.fromhex(l) + bytes.fromhex(r)) level = [domain_leaf(x) for x in leaf_hashes] while len(level) > 1: nxt = [] for i in range(0, len(level), 2): r = level[i + 1] if i + 1 < len(level) else level[i] nxt.append(domain_node(level[i], r)) level = nxt return level[0] def build_forest(algo, roots): """Roots are concatenated as hex STRINGS, then hashed.""" level = list(roots) if len(level) % 2: level.append(level[-1]) while len(level) > 1: level = [H(algo, (level[i] + level[i + 1]).encode()) for i in range(0, len(level), 2)] return level[0] def verify_commitment(record, payload_key, expected): """Bind a retained decision record to the anchored commitment.""" salt = hashlib.sha256((payload_key + ":rubric-commit-v1").encode()).hexdigest() got = hashlib.sha256((salt + canon(record)).encode()).hexdigest() return got == expected def main(path, record_path=None, payload_key=None): proof = json.load(open(path)) h0 = proof.get("hop0") or {} h1 = proof["hop1"] h2 = proof.get("hop2") or {} hcs = proof["hcs"] failures = [] def check(name, cond): print(("PASS " if cond else "FAIL ") + name) if not cond: failures.append(name) return cond # Hop 0: the signed leaf message reproduces the leaf hash. lm = h0.get("leafMessage") if lm: rc = hashlib.sha256(bytes([0x00]) + canon(lm).encode()).hexdigest() check("hop0 leafMessage -> leafHash", rc == h1["leafHash"]) print(" attestation_id: %s" % lm.get("attestation_id")) print(" issued_at : %s (issuer %s)" % (lm.get("issued_at"), lm.get("issuer_node_region"))) if record_path and payload_key: rec = json.load(open(record_path)) check("hop0 your record -> payload_commitment", verify_commitment(rec, payload_key, (lm.get("payload") or {}).get("payload_commitment"))) print(" agent : %s" % rec.get("agentId")) print(" intent : %s" % rec.get("intent")) print(" amount : %s %s" % (rec.get("amount"), rec.get("currency") or "")) print(" params : %s" % rec.get("paramsHash")) else: print(" (add record.json + payloadKey to bind the decision itself)") else: print("WARN no hop0 in package; anchor cannot be tied to a record") # Authorship: did the submitter sign this record with its own key? ca = (lm or {}).get("client_attestation") or h0.get("clientAttestation") if ca: check("authorship leaf carries a verified client key", bool(ca.get("verified"))) print(" key: %s..." % (ca.get("publicKey") or "")[:32]) print(" alg: %s" % ca.get("algorithm")) print(" (this record could not have been produced by the issuing node)") else: print("NOTE not client-signed: contents rest on the issuing node's attestation") # Federation: is the aggregate root signed by a quorum? fed = proof.get("federation") if fed: check("federation aggregate root signed by %s/%s regions" % (fed.get("obtained"), fed.get("required")), bool(fed.get("signed"))) print(" regions: %s" % ", ".join(s.get("region", "?") for s in (fed.get("signatures") or []))) else: print("NOTE no federation signature block in this proof package") # Hop 1: leaf -> batch root. SHA-256 with a 0x01 internal-node domain tag. steps = h1.get("path") if steps is None: steps = ([{"sibling": h1["sibling"], "siblingDirection": h1["siblingDirection"]}] if h1.get("sibling") else []) cur = h1["leafHash"] for st in steps: a, b = bytes.fromhex(cur), bytes.fromhex(st["sibling"]) left, right = (a, b) if st.get("siblingDirection", "R") == "R" else (b, a) cur = hashlib.sha256(bytes([0x01]) + left + right).hexdigest() check("hop1 leaf -> batchRoot (%d step(s))" % len(steps), cur == h1["batchRoot"]) # Hop 2: batch root -> aggregate root, recomputed from the flush records. algo = h2.get("algorithm", "sha3-256") tv = h2.get("treeVersion") flushes = h2.get("tier1Flushes") or [] if tv == 2: print("WARN treeVersion 2: this aggregate binds flush COUNT only, not") print(" content (RUBRIC-SEC-2026-001). Hop 2 cannot be verified.") agg = None elif not flushes: print("WARN no tier1Flushes in proof package; hop 2 not verifiable") agg = None else: leaves = [make_leaf_v2(algo, "DOCUMENT_HASH", {"forestRoot": f["forestRoot"], "itemCount": f["itemCount"]}) for f in flushes] agg = build_forest(algo, [build_tree_v3(algo, leaves)]) check("hop2 batchRoot -> aggregateRoot (recomputed)", agg == h2.get("aggregateRoot")) check("hop2 our batchRoot is among the anchored flushes", h1["batchRoot"] in [f["forestRoot"] for f in flushes]) # Hop 3: the aggregate root is on the public ledger. url = MIRROR.format(t=hcs["topicId"], s=hcs["sequenceNumber"]) with urllib.request.urlopen(url, timeout=20) as r: msg = json.load(r) onchain = json.loads(base64.b64decode(msg["message"])) check("hop3 on-chain aggregateRoot matches package", onchain.get("aggregateRoot") == h2.get("aggregateRoot")) if agg is not None: check("hop3 on-chain aggregateRoot matches OUR recomputation", onchain.get("aggregateRoot") == agg) print("") print("consensus timestamp: %s" % msg.get("consensus_timestamp")) print("topic / sequence : %s / %s" % (msg.get("topic_id"), msg.get("sequence_number"))) print("") if failures: print("VERIFICATION FAILED: " + ", ".join(failures)) return 1 if tv == 2: print("PARTIAL: hop 1 and the ledger anchor verify; hop 2 is not") print("cryptographically bound for treeVersion 2 anchors.") return 1 print("VERIFIED: leaf -> batch -> aggregate -> Hedera consensus.") return 0 if __name__ == "__main__": if len(sys.argv) == 2: sys.exit(main(sys.argv[1])) if len(sys.argv) == 4: sys.exit(main(sys.argv[1], sys.argv[2], sys.argv[3])) print(__doc__) sys.exit(2)