#!/usr/bin/env python3 """verify.py — independent verifier for the AILedger operational record. Python 3.8+ standard library only. Anyone (a founder, a customer, counsel, an auditor) can run this against a copy of the record and check, without trusting the writer or this repository's host: 1. the canonical form every row's hash_chain_self = SHA-256 of its AILedger canonical row (`ailedger-v1`, below) 2. the chain row k's hash_chain_prev = row k-1's hash_chain_self; the first row's = 64 zeros 3. the signatures `sig` = Ed25519 by a PINNED node key over the ASCII hex of hash_chain_self; `cosig` = the same by a pinned principal (desk) key; `attest` = the SUBMITTING node's own signature (source attestation, `ailedger-attest-1`, below) by a pinned attester key. An unpinned key, a wrong fingerprint or a bad signature FAILS. 4. the checkpoints each signed tree head (activity/checkpoints/*.json) = RFC 6962 Merkle root over the first tree_size rows' hash_chain_self, signed by the pinned node key; consecutive heads are proven consistent (append-only) by the RFC 6962 consistency proof they carry 5. the published head activity/HEAD.json names a prefix of the chain (row_count rows ending in chain_head_hash) python3 verify.py verify ./activity against ./trust.json (exit 0 = everything holds) python3 verify.py --dir PATH --trust FILE another tier / another pin set (e.g. a private tier Jake shares) python3 verify.py --json machine-readable result python3 verify.py prove EVENT_ID [--dir …] > proof.json inclusion proof of ONE row under the latest checkpoint python3 verify.py check-proof proof.json [--trust FILE] verify it with nothing else — no other row is disclosed python3 verify.py --selfcheck run the golden vectors in testdata/vectors.json (RFC 8032, RFC 6962, rows) Canonical form `ailedger-v1` (AILedger param_canonicalization_v1 as run by activity.py; CANON_V "1"). The canonical row is the 24 cells below joined by '|', UTF-8, SHA-256, lowercase hex: event_id|timestamp|tenant_id|system_id|actor|source|decision_type|summary|refs|outcome|evidence|cost_usd|latency_s| model_version|inputs_hash|human_in_loop|flags_raised|anchor_event_id|tier|topic|caused_by|dedup_key|canon_v|hash_chain_prev cell rules: missing/null scalar -> ''; missing/null array (refs, flags_raised, caused_by) -> '[]'; true/false -> 'true'/'false'; array/object -> JSON with sorted keys, separators (',', ':'), non-ASCII kept as UTF-8; any other value -> its text as stored (the writer stores numbers as fixed-point strings, e.g. cost_usd "0.0100"). Fields outside the list (hash_chain_self, signer, sig, cosig, attest) are NOT hashed: stripping a signature loses attribution; it cannot forge one. Source attestation `ailedger-attest-1` (since 2026-09-30): a node that submits rows to the writer holds its own key and signs ailedger-attest-1|event_id|at|actor|source|decision_type|summary|topic (cells as above; `at` = attest.at, the rest = the row's own values). Pinned under `attesters` in trust.json — a namespace apart from node keys, so an attester key never passes as a row or tree-head signature. A row then carries two signatures: the submitting terminal's (attest) and the writer's (sig). Known limit: array/object cells sort keys by code point (Python), not RFC 8785's UTF-16 order; identical for ASCII keys, which is all the writer emits. Full RFC 8785 is a listed follow-up. Signed tree head `ailedger-sth-1`: the node key signs the ASCII bytes of ailedger-sth-1|tenant_id|tree_size|root_hash|chain_head_hash|last_event_id|timestamp Merkle tree: RFC 6962 §2.1 — leaf = SHA-256(0x00 || 32 raw bytes of hash_chain_self), node = SHA-256(0x01 || left || right). Ed25519: RFC 8032 §5.1.7 verification, pure Python (the RFC's §6 reference arithmetic), pinned to the RFC 8032 test vectors by --selfcheck. Not constant-time — irrelevant for verification, which handles no secrets. ~5 ms per signature. """ import argparse, base64, glob, hashlib, json, os, re, sys CANON = "ailedger-v1" STH_V = "ailedger-sth-1" GENESIS = "0" * 64 FIELDS = ("event_id", "timestamp", "tenant_id", "system_id", "actor", "source", "decision_type", "summary", "refs", "outcome", "evidence", "cost_usd", "latency_s", "model_version", "inputs_hash", "human_in_loop", "flags_raised", "anchor_event_id", "tier", "topic", "caused_by", "dedup_key", "canon_v", "hash_chain_prev") ARRAYS = ("refs", "flags_raised", "caused_by") HERE = os.path.dirname(os.path.abspath(__file__)) # ---- canonical form ------------------------------------------------------------------------------------------------------ def cell(k, v): if v is None: return "[]" if k in ARRAYS else "" if isinstance(v, bool): return "true" if v else "false" if isinstance(v, (list, dict)): return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False) return str(v) def canonical_row(row): return "|".join(cell(k, row.get(k)) for k in FIELDS) def row_hash(row): return hashlib.sha256(canonical_row(row).encode("utf-8")).hexdigest() def fingerprint(pubkey_hex): return hashlib.sha256(bytes.fromhex(pubkey_hex)).hexdigest()[:16] # ---- Ed25519 verification (RFC 8032 §6 reference arithmetic, extended coordinates) --------------------------------------- _p = 2 ** 255 - 19 _q = 2 ** 252 + 27742317777372353535851937790883648493 _d = -121665 * pow(121666, _p - 2, _p) % _p _SQRT_M1 = pow(2, (_p - 1) // 4, _p) def _add(P, Q): A, B = (P[1] - P[0]) * (Q[1] - Q[0]) % _p, (P[1] + P[0]) * (Q[1] + Q[0]) % _p C, D = 2 * P[3] * Q[3] * _d % _p, 2 * P[2] * Q[2] % _p E, F, G, H = B - A, D - C, D + C, B + A return (E * F, G * H, F * G, E * H) def _mul(s, P): Q = (0, 1, 1, 0) while s > 0: if s & 1: Q = _add(Q, P) P = _add(P, P) s >>= 1 return Q def _eq(P, Q): return (P[0] * Q[2] - Q[0] * P[2]) % _p == 0 and (P[1] * Q[2] - Q[1] * P[2]) % _p == 0 def _recover_x(y, sign): if y >= _p: return None x2 = (y * y - 1) * pow(_d * y * y + 1, _p - 2, _p) if x2 == 0: return None if sign else 0 x = pow(x2, (_p + 3) // 8, _p) if (x * x - x2) % _p: x = x * _SQRT_M1 % _p if (x * x - x2) % _p: return None return _p - x if (x & 1) != sign else x _GY = 4 * pow(5, _p - 2, _p) % _p _GX = _recover_x(_GY, 0) G = (_GX, _GY, 1, _GX * _GY % _p) def _decompress(s): if len(s) != 32: return None y = int.from_bytes(s, "little"); sign = y >> 255; y &= (1 << 255) - 1 x = _recover_x(y, sign) return None if x is None else (x, y, 1, x * y % _p) def compress(P): zi = pow(P[2], _p - 2, _p); x, y = P[0] * zi % _p, P[1] * zi % _p return (y | ((x & 1) << 255)).to_bytes(32, "little") def ed25519_verify(public, msg, sig): """True iff sig (64 bytes) is a valid Ed25519 signature of msg under public (32 bytes). Never raises.""" try: if len(public) != 32 or len(sig) != 64: return False A, R = _decompress(public), _decompress(sig[:32]) if A is None or R is None: return False s = int.from_bytes(sig[32:], "little") if s >= _q: return False h = int.from_bytes(hashlib.sha512(sig[:32] + public + msg).digest(), "little") % _q return _eq(_mul(s, G), _add(R, _mul(h, A))) except (TypeError, ValueError): return False # ---- RFC 6962 Merkle --------------------------------------------------------------------------------------------------- def leaf_hash(chain_hash_hex): return hashlib.sha256(b"\x00" + bytes.fromhex(chain_hash_hex)).digest() def _node(l, r): return hashlib.sha256(b"\x01" + l + r).digest() def _k(n): k = 1 while (k << 1) < n: k <<= 1 return k def merkle_root(leaves): n = len(leaves) if n == 0: return hashlib.sha256(b"").digest() if n == 1: return leaves[0] k = _k(n) return _node(merkle_root(leaves[:k]), merkle_root(leaves[k:])) def inclusion_path(leaves, i): n = len(leaves) if n <= 1: return [] k = _k(n) return inclusion_path(leaves[:k], i) + [merkle_root(leaves[k:])] if i < k else \ inclusion_path(leaves[k:], i - k) + [merkle_root(leaves[:k])] def root_from_inclusion(i, n, leaf, path): """RFC 9162 §2.1.3.2. Returns the root, or None if the path does not fit the position.""" if not 0 <= i < n: return None fn, sn, r = i, n - 1, leaf for p in path: if sn == 0: return None if fn & 1 or fn == sn: r = _node(p, r) while not fn & 1 and fn != 0: fn >>= 1; sn >>= 1 else: r = _node(r, p) fn >>= 1; sn >>= 1 return r if sn == 0 else None def _subproof(m, leaves, complete): n = len(leaves) if m == n: return [] if complete else [merkle_root(leaves)] k = _k(n) return _subproof(m, leaves[:k], complete) + [merkle_root(leaves[k:])] if m <= k else \ _subproof(m - k, leaves[k:], False) + [merkle_root(leaves[:k])] def consistency_path(m, leaves): return [] if m == len(leaves) or m == 0 else _subproof(m, leaves, True) def verify_consistency(m, n, old_root, new_root, path): """RFC 9162 §2.1.4.2: the tree of size n extends the tree of size m (append-only).""" if m < 0 or n < m: return False if m == 0: return True if m == n: return not path and old_root == new_root proof = list(path) if m & (m - 1) == 0: proof = [old_root] + proof if not proof: return False fn, sn = m - 1, n - 1 while fn & 1: fn >>= 1; sn >>= 1 fr = sr = proof[0] for c in proof[1:]: if sn == 0: return False if fn & 1 or fn == sn: fr, sr = _node(c, fr), _node(c, sr) while not fn & 1 and fn != 0: fn >>= 1; sn >>= 1 else: sr = _node(sr, c) fn >>= 1; sn >>= 1 return sn == 0 and fr == old_root and sr == new_root # ---- signatures against the pinned trust set ----------------------------------------------------------------------------- def load_trust(path): with open(path, encoding="utf-8") as f: return json.load(f) def _pinned(trust, node_id=None, principal=None): e = ((trust.get("principals") or {}).get(principal) if principal else trust.get(node_id)) or {} return e.get("signer_pubkey") if isinstance(e, dict) else None def check_sig(signer, sig_b64, message, trust, canon, principal=None): """-> 'ok' | 'untrusted' | 'bad'. signer = {signer_pubkey, key_fingerprint, node_id|principal, canon}.""" signer = signer or {} pin = _pinned(trust, node_id=signer.get("node_id"), principal=principal) if not pin or pin != signer.get("signer_pubkey") or signer.get("key_fingerprint") != fingerprint(pin): return "untrusted" if signer.get("canon", canon) != canon: return "bad" try: raw = base64.b64decode(sig_b64, validate=True) except (ValueError, TypeError): return "bad" return "ok" if ed25519_verify(bytes.fromhex(pin), message, raw) else "bad" ATTEST_V = "ailedger-attest-1" ATTEST_FIELDS = ("event_id", "at", "actor", "source", "decision_type", "summary", "topic") def attest_message(row, at): return "|".join([ATTEST_V] + ["" if (at if k == "at" else row.get(k)) is None else str(at if k == "at" else row.get(k)) for k in ATTEST_FIELDS]).encode("utf-8") def check_attest(row, trust): """-> ok|untrusted|bad|none for the row's `attest` (source attestation) against pinned `attesters`.""" a = row.get("attest") if not a: return "none" e = ((trust.get("attesters") or {}).get(a.get("node_id")) or {}) pin = e.get("signer_pubkey") if isinstance(e, dict) else None if not pin or pin != a.get("signer_pubkey") or a.get("key_fingerprint") != fingerprint(pin): return "untrusted" if a.get("canon") != ATTEST_V: return "bad" try: raw = base64.b64decode(a.get("sig", ""), validate=True) except (ValueError, TypeError): return "bad" return "ok" if ed25519_verify(bytes.fromhex(pin), attest_message(row, a.get("at")), raw) else "bad" def check_row_sigs(row, trust): """-> (sig_status, cosig_status) each in ok|untrusted|bad|none.""" msg = str(row.get("hash_chain_self", "")).encode() s = check_sig(row.get("signer"), row["sig"], msg, trust, CANON) if row.get("sig") else "none" c = row.get("cosig") cs = check_sig(c, c.get("sig", ""), msg, trust, CANON, principal=c.get("principal")) if c else "none" return s, cs def sth_message(h): return "|".join(str(h.get(k, "")) for k in ("v", "tenant_id", "tree_size", "root_hash", "chain_head_hash", "last_event_id", "timestamp")).encode() # ---- the log --------------------------------------------------------------------------------------------------------- def day_files(d): return sorted(p for p in glob.glob(os.path.join(d, "*.jsonl")) if re.fullmatch(r"\d{4}-\d{2}-\d{2}\.jsonl", os.path.basename(p))) def load_rows(d): rows = [] for p in day_files(d): with open(p, encoding="utf-8") as f: for n, l in enumerate(f, 1): if l.strip(): try: rows.append(json.loads(l)) except ValueError: raise SystemExit(f"{os.path.basename(p)}:{n}: not JSON") return rows def checkpoints(d): out = [] for p in glob.glob(os.path.join(d, "checkpoints", "*.json")): with open(p, encoding="utf-8") as f: out.append((p, json.load(f))) return sorted(out, key=lambda x: (x[1].get("tree_size", 0), x[1].get("timestamp", ""))) def verify_rows(rows, trust): r = {"ok": True, "broken_at_id": None, "reason": None, "row_count": 0, "chain_head_hash": None, "signed": 0, "covered_rows": 0, "unsigned_after": 0, "cosigned": {}, "attested": {}} prev = GENESIS for i, row in enumerate(rows): want = row_hash(row) why = None if row.get("hash_chain_prev") != prev: why = f"hash_chain_prev {str(row.get('hash_chain_prev'))[:16]}… != previous row {prev[:16]}…" elif row.get("hash_chain_self") != want: why = f"hash_chain_self {str(row.get('hash_chain_self'))[:16]}… != canonical {want[:16]}…" if not why: s, c = check_row_sigs(row, trust) if s in ("bad", "untrusted"): why = f"node signature {s}" elif c in ("bad", "untrusted"): why = f"cosignature {c} ({(row.get('cosig') or {}).get('principal')})" else: a = check_attest(row, trust) if a in ("bad", "untrusted"): why = f"source attestation {a} ({(row.get('attest') or {}).get('node_id')})" if why: r.update(ok=False, broken_at_id=row.get("event_id"), reason=why, broken_at_index=i); return r prev = want r["row_count"] = i + 1; r["chain_head_hash"] = want if s == "ok": r.update(signed=r["signed"] + 1, covered_rows=i + 1, unsigned_after=0) else: r["unsigned_after"] += 1 if c == "ok": p = row["cosig"].get("principal"); r["cosigned"][p] = r["cosigned"].get(p, 0) + 1 if row.get("attest"): p = row["attest"].get("node_id"); r["attested"][p] = r["attested"].get(p, 0) + 1 return r def verify_checkpoints(rows, cps, trust): leaves = [leaf_hash(r["hash_chain_self"]) for r in rows] out, last = [], None for path, h in cps: name, n = os.path.basename(path), h.get("tree_size", -1) err = None if h.get("v") != STH_V: err = f"unknown version {h.get('v')}" elif not isinstance(n, int) or not 0 < n <= len(leaves): err = f"tree_size {n} beyond the {len(leaves)}-row log" elif merkle_root(leaves[:n]).hex() != h.get("root_hash"): err = "root_hash is not the Merkle root of the first tree_size rows" elif rows[n - 1]["hash_chain_self"] != h.get("chain_head_hash"): err = "chain_head_hash is not row tree_size's hash" elif rows[n - 1].get("event_id") != h.get("last_event_id"): err = "last_event_id mismatch" else: s = check_sig(h.get("signer"), h.get("sig", ""), sth_message(h), trust, STH_V) if s != "ok": err = f"tree-head signature {s}" elif last and (h.get("prev") or {}).get("tree_size") == last["tree_size"]: pv = h["prev"] if pv.get("root_hash") != last["root_hash"] or not verify_consistency( last["tree_size"], n, bytes.fromhex(last["root_hash"]), bytes.fromhex(h["root_hash"]), [bytes.fromhex(x) for x in pv.get("consistency", [])]): err = f"consistency proof from tree_size {last['tree_size']} fails" out.append({"file": name, "tree_size": n, "ok": err is None, "error": err}) if err is None: last = h return out def verify_head(d, rows): p = os.path.join(d, "HEAD.json") if not os.path.exists(p): return None with open(p, encoding="utf-8") as f: h = json.load(f) n = h.get("row_count", 0) ok = isinstance(n, int) and 0 < n <= len(rows) and rows[n - 1]["hash_chain_self"] == h.get("chain_head_hash") return {"ok": ok, "row_count": n, "chain_head_hash": h.get("chain_head_hash"), "note": None if ok else "published HEAD.json does not name a prefix of this chain"} def verify_dir(d, trust): rows = load_rows(d) res = {"dir": d, "chain": verify_rows(rows, trust)} good = rows[:res["chain"]["row_count"]] if not res["chain"]["ok"] else rows res["checkpoints"] = verify_checkpoints(good, checkpoints(d), trust) res["head"] = verify_head(d, good) res["ok"] = res["chain"]["ok"] and all(c["ok"] for c in res["checkpoints"]) and (res["head"] is None or res["head"]["ok"]) return res # ---- one-row inclusion proofs ------------------------------------------------------------------------------------------ def prove(d, event_id): rows = load_rows(d) cps = [h for _, h in checkpoints(d)] idx = next((i for i, r in enumerate(rows) if r.get("event_id") == event_id), None) if idx is None: raise SystemExit(f"no row {event_id}") cp = next((h for h in reversed(cps) if h.get("tree_size", 0) > idx), None) if not cp: raise SystemExit(f"row {idx} is not yet under a signed checkpoint (latest covers " f"{cps[-1]['tree_size'] if cps else 0} rows) — wait for the next one") leaves = [leaf_hash(r["hash_chain_self"]) for r in rows[:cp["tree_size"]]] return {"v": "ailedger-inclusion-1", "row": rows[idx], "index": idx, "tree_size": cp["tree_size"], "audit_path": [x.hex() for x in inclusion_path(leaves, idx)], "checkpoint": cp} def check_proof(pf, trust): row, cp = pf["row"], pf["checkpoint"] if row_hash(row) != row.get("hash_chain_self"): return False, "row does not hash to its hash_chain_self (altered)" if pf["tree_size"] != cp.get("tree_size"): return False, "proof and checkpoint disagree on tree_size" root = root_from_inclusion(pf["index"], pf["tree_size"], leaf_hash(row["hash_chain_self"]), [bytes.fromhex(x) for x in pf["audit_path"]]) if root is None or root.hex() != cp.get("root_hash"): return False, "audit path does not lead to the checkpoint root" s = check_sig(cp.get("signer"), cp.get("sig", ""), sth_message(cp), trust, STH_V) if s != "ok": return False, f"checkpoint signature {s}" rs, _ = check_row_sigs(row, trust) if rs == "bad": return False, "row signature bad" if check_attest(row, trust) == "bad": return False, "row source attestation bad" return True, f"row {pf['index']} is in the {pf['tree_size']}-row log signed by {cp['signer']['key_fingerprint']} at {cp['timestamp']}" # ---- golden vectors ----------------------------------------------------------------------------------------------------- def selfcheck(path): with open(path, encoding="utf-8") as f: v = json.load(f) n = 0 for t in v["ed25519"]: got = ed25519_verify(bytes.fromhex(t["public"]), bytes.fromhex(t["message"]), bytes.fromhex(t["signature"])) assert got is t["valid"], f"ed25519 vector {t['name']}: {got}"; n += 1 for t in v["canonical"]: assert canonical_row(t["row"]) == t["canonical"], f"canonical {t['name']}" assert row_hash(t["row"]) == t["sha256"], f"sha256 {t['name']}"; n += 1 for t in v["merkle"]["roots"]: leaves = [leaf_hash(h) for h in v["merkle"]["leaves"][:t["size"]]] assert merkle_root(leaves).hex() == t["root"], f"merkle root size {t['size']}"; n += 1 for t in v["merkle"]["inclusion"]: leaves = [leaf_hash(h) for h in v["merkle"]["leaves"][:t["size"]]] assert [x.hex() for x in inclusion_path(leaves, t["index"])] == t["path"] assert root_from_inclusion(t["index"], t["size"], leaves[t["index"]], [bytes.fromhex(x) for x in t["path"]]).hex() == t["root"] n += 1 for t in v["merkle"]["consistency"]: L = [leaf_hash(h) for h in v["merkle"]["leaves"]] assert [x.hex() for x in consistency_path(t["old"], L[:t["new"]])] == t["path"] assert verify_consistency(t["old"], t["new"], merkle_root(L[:t["old"]]), merkle_root(L[:t["new"]]), [bytes.fromhex(x) for x in t["path"]]); n += 1 for t in v["chains"]: r = verify_rows(t["rows"], t["trust"]) assert r["ok"] is t["expect_ok"] and (t["expect_ok"] or t["expect_reason"] in r["reason"]), f"chain {t['name']}: {r}" if t.get("checkpoint"): c = verify_checkpoints(t["rows"], [("vector.json", t["checkpoint"])], t["trust"]) assert c[0]["ok"] is t["checkpoint_ok"], f"checkpoint {t['name']}: {c}" n += 1 for t in v.get("attest", []): assert check_attest(t["row"], t["trust"]) == t["expect"], f"attest {t['name']}"; n += 1 for t in v["live_rows"]: assert row_hash(t["row"]) == t["row"]["hash_chain_self"], f"live {t['name']}" assert check_row_sigs(t["row"], v["live_trust"])[0] == "ok", f"live sig {t['name']}"; n += 1 print(f"verify.py selfcheck OK — {n} golden vectors (RFC 8032 Ed25519, RFC 6962 Merkle root/inclusion/consistency, " f"ailedger-v1 canonical rows, signed chains incl. tamper cases, source attestations, live rows under the pinned node key)") return 0 def main(argv=None): ap = argparse.ArgumentParser(description="Verify the AILedger operational record (stdlib only).") ap.add_argument("cmd", nargs="?", default="verify", choices=("verify", "prove", "check-proof")) ap.add_argument("arg", nargs="?") ap.add_argument("--dir", default=os.path.join(HERE, "activity")) ap.add_argument("--trust", default=os.path.join(HERE, "trust.json")) ap.add_argument("--json", action="store_true") ap.add_argument("--selfcheck", action="store_true") a = ap.parse_args(argv) if a.selfcheck: return selfcheck(os.path.join(HERE, "testdata", "vectors.json")) trust = load_trust(a.trust) if a.cmd == "prove": print(json.dumps(prove(a.dir, a.arg), indent=1, ensure_ascii=False)); return 0 if a.cmd == "check-proof": with open(a.arg, encoding="utf-8") as f: ok, why = check_proof(json.load(f), trust) print(("OK " if ok else "FAIL ") + why); return 0 if ok else 1 r = verify_dir(a.dir, trust) if a.json: print(json.dumps(r, indent=1)) else: c = r["chain"] print(f"chain {'OK' if c['ok'] else 'BROKEN at ' + str(c['broken_at_id']) + ': ' + c['reason']} " f"{c['row_count']} rows, head {str(c['chain_head_hash'])[:16]}…") print(f"signatures {c['signed']} node-signed rows; rows 1..{c['covered_rows']} covered by a pinned signature; " f"{c['unsigned_after']} unsigned at the head; cosigned {c['cosigned'] or 'none'}; " f"source-attested {c.get('attested') or 'none'}") for cp in r["checkpoints"]: print(f"checkpoint {cp['file']}: {'OK' if cp['ok'] else 'FAIL — ' + cp['error']} (tree_size {cp['tree_size']})") if not r["checkpoints"]: print("checkpoint none published yet") if r["head"]: print(f"HEAD.json {'OK' if r['head']['ok'] else 'FAIL — ' + r['head']['note']} ({r['head']['row_count']} rows)") print("RESULT " + ("VERIFIED" if r["ok"] else "FAILED")) return 0 if r["ok"] else 1 if __name__ == "__main__": sys.exit(main())