diff options
Diffstat (limited to 'scripts/verify-utxo-proof.py')
| -rw-r--r-- | scripts/verify-utxo-proof.py | 163 |
1 files changed, 163 insertions, 0 deletions
diff --git a/scripts/verify-utxo-proof.py b/scripts/verify-utxo-proof.py new file mode 100644 index 0000000..31099f7 --- /dev/null +++ b/scripts/verify-utxo-proof.py | |||
| @@ -0,0 +1,163 @@ | |||
| 1 | #!/usr/bin/env python3 | ||
| 2 | """Verify a Bitflash UTXO inclusion proof against a UTXO root.""" | ||
| 3 | |||
| 4 | import argparse | ||
| 5 | import hashlib | ||
| 6 | import json | ||
| 7 | import sys | ||
| 8 | from pathlib import Path | ||
| 9 | |||
| 10 | |||
| 11 | SCHEMA = "bitflash-utxo-inclusion-proof-1" | ||
| 12 | LEAF_DOMAIN = "BTFUTXO1" | ||
| 13 | NODE_DOMAIN = b"BTFNODE1|" | ||
| 14 | |||
| 15 | |||
| 16 | def canonical_bytes(obj): | ||
| 17 | return (json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + "\n").encode("ascii") | ||
| 18 | |||
| 19 | |||
| 20 | def sha256(data): | ||
| 21 | return hashlib.sha256(data).digest() | ||
| 22 | |||
| 23 | |||
| 24 | def utxo_leaf_hash(outpoint, utxo): | ||
| 25 | payload = "|".join([ | ||
| 26 | LEAF_DOMAIN, | ||
| 27 | outpoint, | ||
| 28 | str(utxo["value_satoshis"]), | ||
| 29 | utxo["script_pub_key_hex"], | ||
| 30 | str(utxo["created_height"]), | ||
| 31 | utxo["created_txid"], | ||
| 32 | ]).encode("ascii") | ||
| 33 | return sha256(payload) | ||
| 34 | |||
| 35 | |||
| 36 | def load_proof(path): | ||
| 37 | try: | ||
| 38 | return json.loads(Path(path).read_text(encoding="ascii")) | ||
| 39 | except Exception as e: | ||
| 40 | raise SystemExit("could not read proof: %s" % e) | ||
| 41 | |||
| 42 | |||
| 43 | def verify(proof, expected_root=None): | ||
| 44 | failures = [] | ||
| 45 | if proof.get("schema") != SCHEMA: | ||
| 46 | failures.append("unexpected schema") | ||
| 47 | commitment = proof.get("commitment", {}) | ||
| 48 | if commitment.get("algorithm") != "sorted-outpoint-merkle-sha256": | ||
| 49 | failures.append("unexpected commitment algorithm") | ||
| 50 | if commitment.get("leaf_domain") != LEAF_DOMAIN: | ||
| 51 | failures.append("unexpected leaf domain") | ||
| 52 | if commitment.get("node_domain") != "BTFNODE1": | ||
| 53 | failures.append("unexpected node domain") | ||
| 54 | |||
| 55 | target = proof.get("target", {}) | ||
| 56 | outpoint = target.get("outpoint", "") | ||
| 57 | utxo = target.get("utxo", {}) | ||
| 58 | if outpoint != "%s:%s" % (target.get("txid"), target.get("vout")): | ||
| 59 | failures.append("target outpoint does not match txid:vout") | ||
| 60 | |||
| 61 | try: | ||
| 62 | current = utxo_leaf_hash(outpoint, utxo) | ||
| 63 | leaf_ok = target.get("leaf_hash") == current.hex() | ||
| 64 | except Exception as e: | ||
| 65 | failures.append("could not hash target leaf: %s" % e) | ||
| 66 | current = b"\x00" * 32 | ||
| 67 | leaf_ok = False | ||
| 68 | |||
| 69 | if not leaf_ok: | ||
| 70 | failures.append("target leaf hash mismatch") | ||
| 71 | |||
| 72 | for i, step in enumerate(proof.get("proof", [])): | ||
| 73 | try: | ||
| 74 | sibling = bytes.fromhex(step["hash"]) | ||
| 75 | except Exception: | ||
| 76 | failures.append("bad sibling hash at proof level %d" % i) | ||
| 77 | sibling = b"\x00" * 32 | ||
| 78 | if len(sibling) != 32: | ||
| 79 | failures.append("bad sibling hash length at proof level %d" % i) | ||
| 80 | side = step.get("side") | ||
| 81 | if side == "right": | ||
| 82 | current = sha256(NODE_DOMAIN + current + sibling) | ||
| 83 | elif side == "left": | ||
| 84 | current = sha256(NODE_DOMAIN + sibling + current) | ||
| 85 | else: | ||
| 86 | failures.append("bad sibling side at proof level %d" % i) | ||
| 87 | |||
| 88 | computed_root = current.hex() | ||
| 89 | root = expected_root or commitment.get("root") | ||
| 90 | if computed_root != root: | ||
| 91 | failures.append("computed root does not match expected root") | ||
| 92 | |||
| 93 | return { | ||
| 94 | "schema": "bitflash-utxo-proof-verification-1", | ||
| 95 | "deterministic": True, | ||
| 96 | "target": { | ||
| 97 | "outpoint": outpoint, | ||
| 98 | "value_satoshis": utxo.get("value_satoshis"), | ||
| 99 | "value_btf": utxo.get("value_btf"), | ||
| 100 | "created_height": utxo.get("created_height"), | ||
| 101 | }, | ||
| 102 | "commitment": { | ||
| 103 | "expected_root": root, | ||
| 104 | "computed_root": computed_root, | ||
| 105 | "leaf_hash": target.get("leaf_hash"), | ||
| 106 | "path_length": len(proof.get("proof", [])), | ||
| 107 | }, | ||
| 108 | "checks": { | ||
| 109 | "schema_is_supported": proof.get("schema") == SCHEMA, | ||
| 110 | "leaf_hash_matches_target": leaf_ok, | ||
| 111 | "root_matches": computed_root == root, | ||
| 112 | }, | ||
| 113 | "failures": failures, | ||
| 114 | "verdict": "UTXO_PROOF_VERIFIED" if not failures else "UTXO_PROOF_FAILED", | ||
| 115 | } | ||
| 116 | |||
| 117 | |||
| 118 | def print_text(result): | ||
| 119 | data_hash = hashlib.sha256(canonical_bytes(result)).hexdigest() | ||
| 120 | print("Bitflash UTXO proof verifier") | ||
| 121 | print("schema: %s" % result["schema"]) | ||
| 122 | print("canonical verification sha256: %s" % data_hash) | ||
| 123 | print() | ||
| 124 | print("Target") | ||
| 125 | print(" outpoint: %s" % result["target"]["outpoint"]) | ||
| 126 | print(" value: %s BTF" % result["target"]["value_btf"]) | ||
| 127 | print(" height: %s" % result["target"]["created_height"]) | ||
| 128 | print() | ||
| 129 | print("Commitment") | ||
| 130 | print(" expected: %s" % result["commitment"]["expected_root"]) | ||
| 131 | print(" computed: %s" % result["commitment"]["computed_root"]) | ||
| 132 | print(" path: %d sibling hashes" % result["commitment"]["path_length"]) | ||
| 133 | print() | ||
| 134 | print("Checks") | ||
| 135 | for key in sorted(result["checks"]): | ||
| 136 | print(" %s %s" % ("OK " if result["checks"][key] else "FAIL", key)) | ||
| 137 | if result["failures"]: | ||
| 138 | print() | ||
| 139 | print("Failures") | ||
| 140 | for failure in result["failures"]: | ||
| 141 | print(" - %s" % failure) | ||
| 142 | print() | ||
| 143 | print("Verdict: %s" % result["verdict"]) | ||
| 144 | |||
| 145 | |||
| 146 | def main(argv): | ||
| 147 | ap = argparse.ArgumentParser(description=__doc__) | ||
| 148 | ap.add_argument("proof", help="canonical JSON proof from prove-utxo.py") | ||
| 149 | ap.add_argument("--root", help="expected UTXO root; defaults to root embedded in proof") | ||
| 150 | ap.add_argument("--json", action="store_true", help="print canonical JSON") | ||
| 151 | args = ap.parse_args(argv) | ||
| 152 | |||
| 153 | result = verify(load_proof(args.proof), args.root) | ||
| 154 | data = canonical_bytes(result) | ||
| 155 | if args.json: | ||
| 156 | sys.stdout.buffer.write(data) | ||
| 157 | else: | ||
| 158 | print_text(result) | ||
| 159 | return 0 if result["verdict"] == "UTXO_PROOF_VERIFIED" else 1 | ||
| 160 | |||
| 161 | |||
| 162 | if __name__ == "__main__": | ||
| 163 | raise SystemExit(main(sys.argv[1:])) | ||