#!/usr/bin/env python3 """Verify a XODE Sentinel receipt or window bundle — without trusting XODE. Standard library only. Give this file and a receipt to an auditor, a regulator or an end user; it talks to nothing but a public XODE RPC node of their choice. python3 sentinel_verify.py receipt.json python3 sentinel_verify.py bundle.json # completeness audit python3 sentinel_verify.py receipt.json --rpc https://your-node python3 sentinel_verify.py receipt.json --offline # hashes only What it checks, as separate claims: 1. record -> leaf the record you hold hashes to the committed leaf 2. leaf -> tenant root the leaf is one of n records the tenant committed 3. root -> commit that root sits at position #seq of the tenant's chain 4. commit -> global root the tenant's window is inside this window's root 5. global root -> block the root is in a FINALIZED block on XODE, inside an extrinsic signed by a known Sentinel anchor account Step 5's signer check is what makes the rest mean anything: anyone can put any bytes in a remark from their own account for a fraction of a cent, so a root "being on chain" proves nothing until we know who put it there. The accounts that sign Sentinel anchors are listed in KNOWN_SIGNERS below; --signer replaces that list (e.g. to check a deployment you run yourself). A bundle replaces steps 1-2 with: every leaf of the window, each exactly once, rebuilds the tenant root and the count n — nothing withheld, nothing added. Exit codes 0 verified 1 FAILED: a hash link, the format, or the chain check failed 2 hashes valid, but not on chain yet (open window, not finalized, or a simulated anchor from a development server) 3 hashes valid, chain not reachable — try --rpc 4 the verifier itself could not run (unreadable file, bug); never "tampered" 5 hashes and chain valid, but the receipt has no record attached: it proves some 32 bytes were committed, not what decision they stand for 6 hashes valid and anchored, but by an account not in the signer list (only with --any-signer; without it this is 1) Version 1 receipts (omni, 2026-10) are verified too. """ from __future__ import annotations import argparse import hashlib import json import re import sys import urllib.request SCHEMA = "xode-sentinel/1" FIELDS = ("kind", "subject", "verdict", "reason", "policy", "model", "ts", "nonce", "detail") EXTRA = ("schema", "leaf") MAX_INT = 2 ** 53 - 1 TENANT_ID_TAG = b"xode-sentinel/tenant-id/2\x00" TENANT_COMMIT_TAG = b"xode-sentinel/tenant-commit/2\x00" DEFAULT_RPC = "https://archive-rpc.xode.net" SS58_PREFIX = 280 B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" # Accounts that sign Sentinel anchors. Published here, in the verifier, on # purpose: a receipt must never be able to name its own signer. KNOWN_SIGNERS = { "XqBMcDhStCvxMaBoAPmThVKZm1unfNFebddJyzciv7Fbn1MLP": "XODE Sentinel anchor (omni, since 2026-10-04)", "XqFJ46w2rZV89S6EoSzMuudFJh3qGwawP1PdnBa7R5HbVEjFv": "XODE Sentinel anchor (sentinel.xode.net, since 2026-10-05)", } OK, FAIL, PENDING, UNREACHABLE, ERROR, LEAF_ONLY, UNKNOWN_SIGNER = range(7) class Malformed(ValueError): pass # --------------------------------------------------------- strict parsing -- def unhex(s, size: int | None = 32) -> bytes: if not isinstance(s, str) or not s.startswith("0x") \ or not re.fullmatch(r"(?:[0-9a-fA-F]{2})*", s[2:]): raise Malformed(f"not 0x-hex: {clean(repr(s))[:40]}") raw = bytes.fromhex(s[2:]) if size is not None and len(raw) != size: raise Malformed(f"expected {size} bytes, got {len(raw)}") return raw def strict_int(v, name: str, lo: int = 0) -> int: if isinstance(v, bool): raise Malformed(f"{name} must be an integer") if isinstance(v, float): if not v.is_integer(): raise Malformed(f"{name} must be an integer") v = int(v) if not isinstance(v, int) or not lo <= v <= MAX_INT: raise Malformed(f"{name} must be an integer") return v def proof_list(p) -> list: if not isinstance(p, list): raise Malformed("proof must be a list") return [unhex(x) for x in p] def obj(x, name: str) -> dict: if not isinstance(x, dict): raise Malformed(f"{name} must be an object") return x # ----------------------------------------------------------------- hashing -- def b2(b: bytes) -> bytes: return hashlib.blake2b(b, digest_size=32).digest() def pair(a: bytes, b: bytes) -> bytes: return b2(a + b if a <= b else b + a) def hx(b: bytes) -> str: return "0x" + b.hex() def canonical(record: dict) -> bytes: """Rebuild the exact bytes that were hashed when the decision was made.""" record = obj(record, "record") extra = set(record) - set(FIELDS) - set(EXTRA) if extra: raise Malformed(f"unknown record field(s): {', '.join(sorted(extra))}") if record.get("schema", SCHEMA) != SCHEMA: raise Malformed("unknown record schema") doc = {"schema": SCHEMA} for k in FIELDS: v = record.get(k, "") if k == "ts": doc[k] = strict_int(v, "ts", lo=-MAX_INT) else: v = "" if v is None else v if not isinstance(v, str): raise Malformed(f"{k} must be a string") doc[k] = v try: return json.dumps(doc, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") except UnicodeEncodeError: # a lone surrogate in a field raise Malformed("record is not valid Unicode") from None def record_leaf(record: dict) -> bytes: leaf = b2(canonical(record)) if "leaf" in record and record["leaf"] != hx(leaf): raise Malformed("record's own leaf field does not match its contents") return leaf def walk(node: bytes, proof: list[bytes]) -> bytes: for sib in proof: node = pair(node, sib) return node def merkle_root(leaves: list[bytes]) -> bytes: layer = sorted(set(leaves)) if not layer: return b"\x00" * 32 while len(layer) > 1: nxt = [pair(layer[i], layer[i + 1]) for i in range(0, len(layer) - 1, 2)] if len(layer) % 2: nxt.append(layer[-1]) layer = nxt return layer[0] def tenant_commit(tid, seq: int, prev: bytes, root: bytes, n: int) -> bytes: if not isinstance(tid, str) or not tid: raise Malformed("tenant id must be a non-empty string") return b2(TENANT_COMMIT_TAG + b2(TENANT_ID_TAG + tid.encode("utf-8")) + seq.to_bytes(8, "big") + prev + root + n.to_bytes(8, "big")) # -------------------------------------------------------------------- ss58 -- def b58encode(b: bytes) -> str: n = int.from_bytes(b, "big") out = "" while n: n, r = divmod(n, 58) out = B58[r] + out return "1" * (len(b) - len(b.lstrip(b"\x00"))) + out def ss58(pubkey: bytes, prefix: int = SS58_PREFIX) -> str: if prefix < 64: pre = bytes([prefix]) else: pre = bytes([((prefix & 0b1111_1100) >> 2) | 0b0100_0000, (prefix >> 8) | ((prefix & 0b11) << 6)]) body = pre + pubkey chk = hashlib.blake2b(b"SS58PRE" + body, digest_size=64).digest()[:2] return b58encode(body + chk) def compact(buf: bytes, i: int) -> tuple[int, int]: m = buf[i] & 3 if m == 0: return buf[i] >> 2, i + 1 if m == 1: return int.from_bytes(buf[i:i + 2], "little") >> 2, i + 2 if m == 2: return int.from_bytes(buf[i:i + 4], "little") >> 2, i + 4 ln = (buf[i] >> 2) + 4 return int.from_bytes(buf[i + 1:i + 1 + ln], "little"), i + 1 + ln def signer_of(ext_hex: str) -> str | None: """Address that signed an extrinsic (v4 signed, MultiAddress::Id), else None. Only the envelope is decoded — length, version byte, address. If XODE ever moves to v5 extrinsics this returns None and verification fails closed. """ try: b = bytes.fromhex(ext_hex[2:] if ext_hex.startswith("0x") else ext_hex) _, i = compact(b, 0) version = b[i] if not version & 0x80 or (version & 0x7F) != 4: return None if b[i + 1] != 0x00: # MultiAddress::Id return None return ss58(b[i + 2:i + 34]) except (IndexError, ValueError): return None def byte_aligned_hits(hay: str, needle: str) -> bool: """needle occurs in hay at a whole-byte boundary (even hex offset).""" start = 0 while True: i = hay.find(needle, start) if i < 0: return False if i % 2 == 0: return True start = i + 1 # --------------------------------------------------------------------- rpc -- def rpc(url: str, method: str, params: list): body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": params}).encode() req = urllib.request.Request(url, data=body, method="POST", headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=40) as r: d = json.load(r) if "error" in d: raise RuntimeError(d["error"].get("message", "rpc error")) return d["result"] # ------------------------------------------------------------------ output -- def clean(s) -> str: """Printable only: a receipt must not be able to drive the terminal.""" return "".join(ch if ch.isprintable() else "\N{REPLACEMENT CHARACTER}" for ch in str(s)) class Out: def __init__(self, quiet: bool) -> None: self.quiet = quiet def __call__(self, *a) -> None: if not self.quiet: print(*(clean(x) for x in a)) def step(self, n: str, title: str, ok: bool | None, detail: str = "") -> None: mark = {True: "PASS", False: "FAIL", None: "----"}[ok] self(f" [{mark}] {n} {title}") for line in str(detail).splitlines(): self(f" {line}") # ------------------------------------------------------------------- main --- def verify(doc, rpc_url: str, offline: bool, signers: dict | None, say: Out) -> int: """signers=None means --any-signer.""" try: doc = obj(doc, "receipt") is_bundle = doc.get("kind") == "window-bundle" v = doc.get("v", 1) if v not in (1, 2): raise Malformed("unknown receipt version") except Malformed as e: say.step("!", "receipt format", False, str(e)) return FAIL say("=" * 70) say("XODE Sentinel — independent verification" + (" (window bundle)" if is_bundle else f" (receipt v{v})")) say("=" * 70) has_record = True try: if v == 1: record = obj(doc.get("record"), "record") leaf = record_leaf(record) root = unhex(doc.get("root")) ok = walk(leaf, proof_list(doc.get("proof"))) == root say.step("1", "record → root", ok, f"{record.get('kind')}/{record.get('verdict')}/" f"{record.get('reason')} ts={record.get('ts')}") if not ok: return FAIL anchor_root, payload = root, None else: t = obj(doc.get("tenant"), "tenant") troot = unhex(t.get("root")) n = strict_int(t.get("n"), "n", lo=1) seq = strict_int(t.get("seq"), "seq", lo=1) if is_bundle: if not isinstance(doc.get("leaves"), list): raise Malformed("leaves must be a list") leaves = [unhex(x) for x in doc["leaves"]] distinct = len(set(leaves)) ok = (merkle_root(leaves) == troot and distinct == len(leaves) == n) say.step("1-2", "all leaves → tenant root, count", ok, f"{len(leaves)} leaves disclosed ({distinct} distinct)," f" n={n} committed") if not ok: say(" records were withheld, added, repeated or altered") return FAIL else: leaf = unhex(doc.get("leaf")) rec = doc.get("record") if rec is not None: ok = record_leaf(rec) == leaf say.step("1", "record → leaf", ok, f"{rec.get('kind')}/{rec.get('verdict')}/" f"{rec.get('reason')} subject={rec.get('subject')}\n" f"policy={rec.get('policy')} model={rec.get('model') or '-'}") if not ok: return FAIL else: has_record = False say.step("1", "record → leaf", None, "no record attached — this can only show that some\n" "32 bytes were committed, not which decision") tproof = proof_list(t.get("proof")) ok = walk(leaf, tproof) == troot say.step("2", "leaf → tenant root", ok, f"{len(tproof)} levels · {n} records in window") if not ok: return FAIL commit = tenant_commit(t.get("id"), seq, unhex(t.get("prev")), troot, n) ok = commit == unhex(doc.get("commit")) say.step("3", "tenant root → commit", ok, f"tenant {t.get('id')} · window #{seq} of its chain") if not ok: return FAIL g = obj(doc.get("global"), "global") groot = unhex(g.get("root")) gproof = proof_list(g.get("proof")) ok = walk(commit, gproof) == groot say.step("4", "commit → global root", ok, f"{len(gproof)} levels · {hx(groot)}") if not ok: return FAIL anchor_root = groot payload = (doc.get("anchor") or {}).get("payload") \ if isinstance(doc.get("anchor"), dict) else None except Malformed as e: say.step("!", "receipt format", False, str(e)) return FAIL anchor = doc.get("anchor") if not anchor: say.step("5", "global root → XODE block", None, "not anchored yet — the window is still open or sealing") return PENDING if not isinstance(anchor, dict): say.step("5", "anchor", False, "anchor must be an object") return FAIL if anchor.get("mode") == "simulated": say.step("5", "global root → XODE block", None, "this anchor is SIMULATED (development server) — nothing was\n" "written to XODE, so there is nothing to check on chain") return PENDING if v == 2: if not isinstance(payload, str) or not payload: say.step("5", "anchor payload", False, "missing anchor payload") return FAIL try: p = json.loads(payload) if not isinstance(p, dict) or p.get("t") != "xode-sentinel" \ or unhex(p.get("root")) != anchor_root: raise Malformed("payload does not commit to this root") except (ValueError, Malformed) as e: say.step("5", "anchor payload", False, str(e)) return FAIL needle = payload.encode("utf-8").hex() else: needle = anchor_root.hex().encode().hex() # v1: the root's hex text block_hash = anchor.get("blockHash") if not isinstance(block_hash, str) or not re.fullmatch(r"0x[0-9a-fA-F]{64}", block_hash): say.step("5", "anchor", False, "anchor has no valid blockHash") return FAIL if offline: say.step("5", "global root → XODE block", None, f"skipped (--offline). Block #{anchor.get('blockNumber')}") return PENDING if has_record else LEAF_ONLY try: block = rpc(rpc_url, "chain_getBlock", [block_hash]) if block is None: say.step("5", "global root → XODE block", False, "block not found on this node (try an archive node with --rpc)") return FAIL number = int(block["block"]["header"]["number"], 16) canon = rpc(rpc_url, "chain_getBlockHash", [number]) fin = rpc(rpc_url, "chain_getHeader", [rpc(rpc_url, "chain_getFinalizedHead", [])]) finalized = int(fin["number"], 16) except Exception as e: say.step("5", "global root → XODE block", None, f"could not reach {rpc_url}: {type(e).__name__}: {e}") return UNREACHABLE exts = block["block"]["extrinsics"] hits = [(i, signer_of(ex)) for i, ex in enumerate(exts) if byte_aligned_hits(ex.lower(), needle)] detail = f"block #{number} {block_hash}\ntx {anchor.get('txHash')}" if not hits: say.step("5", "global root → XODE block", False, detail + f"\nnot in any of the block's {len(exts)} extrinsics") return FAIL if not canon or canon.lower() != block_hash.lower(): say.step("5", "global root → XODE block", False, detail + "\nthat block is not on the canonical chain") return FAIL if number > finalized: say.step("5", "global root → XODE block", None, detail + f"\nincluded but not finalized yet (finalized: #{finalized})") return PENDING trusted = [(i, s) for i, s in hits if signers is None or s in signers] if not trusted: who = ", ".join(sorted({s or "unknown" for _, s in hits})) why = ("not a signer you passed with --signer" if signers is not KNOWN_SIGNERS else "not a known Sentinel anchor account.\nAnyone can write a remark;" " only Sentinel's signers count.\n(--signer to trust a deployment" " you run yourself)") say.step("5", "global root → XODE block", False, detail + f"\nsigned by {who} — {why}") return FAIL i, who = trusted[0] label = (signers or {}).get(who, "") say.step("5", "global root → XODE block", True, detail + f"\nextrinsic {i} of {len(exts)}, finalized," f" signed by {who}" + (f"\n({label})" if label else "")) if signers is None: return UNKNOWN_SIGNER if who not in KNOWN_SIGNERS else (OK if has_record else LEAF_ONLY) return OK if has_record else LEAF_ONLY def main() -> int: for stream in (sys.stdout, sys.stderr): # Windows consoles default to a try: # legacy code page stream.reconfigure(encoding="utf-8", errors="replace") except (AttributeError, ValueError): pass ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("file") ap.add_argument("--rpc", default=DEFAULT_RPC) ap.add_argument("--signer", action="append", default=[], help="trust this SS58 address instead of the built-in list" " (repeatable)") ap.add_argument("--any-signer", action="store_true", help="diagnostics only: accept any signer, exit 6 if unknown") ap.add_argument("--offline", action="store_true") ap.add_argument("--json", action="store_true", help="print only the result") a = ap.parse_args() with open(a.file, encoding="utf-8") as f: doc = json.load(f) signers = None if a.any_signer else ( {s: "given with --signer" for s in a.signer} if a.signer else KNOWN_SIGNERS) code = verify(doc, a.rpc, a.offline, signers, Out(a.json)) verdicts = {OK: "VERIFIED — unchanged since it was committed on XODE", FAIL: "FAILED — see the failing step above", PENDING: "HASHES VALID — not (yet) final on XODE", UNREACHABLE: "HASHES VALID — chain not reachable, try --rpc", LEAF_ONLY: "LEAF COMMITTED — but no record attached; ask for the full receipt", UNKNOWN_SIGNER: "ANCHORED BY AN UNKNOWN ACCOUNT — not a Sentinel anchor"} if a.json: print(json.dumps({"code": code, "result": verdicts[code]})) else: print("=" * 70) print("Result:", verdicts[code]) return code if __name__ == "__main__": try: sys.exit(main()) except SystemExit: raise except BaseException as e: # see exit codes above print(f"verifier error: {clean(type(e).__name__)}: {clean(e)}", file=sys.stderr) sys.exit(ERROR)