From 034bc9a9497a336239aadb2e52d620661e90e536 Mon Sep 17 00:00:00 2001 From: Nsable-sh Date: Tue, 11 Aug 2026 09:31:58 -0300 Subject: tools: add UTXO inclusion proofs --- scripts/prove-utxo.py | 202 +++++++++++++++++++++++++++++++++++++++++++ scripts/verify-utxo-proof.py | 163 ++++++++++++++++++++++++++++++++++ 2 files changed, 365 insertions(+) create mode 100644 scripts/prove-utxo.py create mode 100644 scripts/verify-utxo-proof.py (limited to 'scripts') diff --git a/scripts/prove-utxo.py b/scripts/prove-utxo.py new file mode 100644 index 0000000..7a10d2a --- /dev/null +++ b/scripts/prove-utxo.py @@ -0,0 +1,202 @@ +#!/usr/bin/env python3 +"""Create a Bitflash UTXO inclusion proof from local blk*.dat files. + +The proof is deliberately outside consensus. It proves that one txid:vout is +included in the deterministic UTXO root produced by verify-utxo-set.py. +""" + +import argparse +import hashlib +import importlib.util +import json +import os +import sys +from pathlib import Path + + +SCHEMA = "bitflash-utxo-inclusion-proof-1" +NODE_DOMAIN = b"BTFNODE1|" + + +def load_utxo_tool(): + path = Path(__file__).with_name("verify-utxo-set.py") + spec = importlib.util.spec_from_file_location("bitflash_verify_utxo_set", path) + if spec is None or spec.loader is None: + raise SystemExit("could not load %s" % path) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +def canonical_bytes(obj): + return (json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + "\n").encode("ascii") + + +def split_outpoint(outpoint): + parts = outpoint.split(":") + if len(parts) != 2 or len(parts[0]) != 64: + raise SystemExit("outpoint must be txid:vout") + try: + vout = int(parts[1], 10) + except ValueError: + raise SystemExit("vout must be a decimal integer") + if vout < 0: + raise SystemExit("vout must not be negative") + return parts[0].lower(), vout + + +def build_merkle_proof(keys, leaf_hashes, index): + current_index = index + layer = list(leaf_hashes) + proof = [] + level = 0 + while len(layer) > 1: + if current_index % 2 == 0: + sibling_index = current_index + 1 + duplicated = sibling_index >= len(layer) + sibling = layer[current_index] if duplicated else layer[sibling_index] + side = "right" + else: + sibling_index = current_index - 1 + duplicated = False + sibling = layer[sibling_index] + side = "left" + proof.append({ + "level": level, + "side": side, + "hash": sibling.hex(), + "duplicated": duplicated, + }) + + next_layer = [] + for i in range(0, len(layer), 2): + left = layer[i] + right = layer[i + 1] if i + 1 < len(layer) else left + next_layer.append(hashlib.sha256(NODE_DOMAIN + left + right).digest()) + layer = next_layer + current_index //= 2 + level += 1 + return proof, layer[0].hex() + + +def build_proof(args): + tool = load_utxo_tool() + txid, vout = split_outpoint(args.outpoint) + outpoint = "%s:%d" % (txid, vout) + + raw_blocks = tool.read_blocks(args.datadir, args.max_blocks) + if not raw_blocks: + raise SystemExit("no Bitflash blocks found in %s" % args.datadir) + blocks = tool.select_main_chain(raw_blocks) + state = tool.apply_blocks(blocks) + utxos = state["utxos"] + stats = state["stats"] + if outpoint not in utxos: + raise SystemExit("outpoint is not unspent in the scanned chain: %s" % outpoint) + + keys = sorted(utxos) + leaf_hashes = [tool.utxo_leaf_hash(k, utxos[k]) for k in keys] + index = keys.index(outpoint) + proof_path, root = build_merkle_proof(keys, leaf_hashes, index) + direct_root = tool.utxo_root(utxos) + if root != direct_root: + raise SystemExit("internal proof root mismatch") + + tip = blocks[-1] + utxo = utxos[outpoint] + report = { + "schema": SCHEMA, + "deterministic": True, + "chain": { + "name": "Bitflash", + "ticker": "BTF", + "message_start_hex": tool.MAGIC.hex(), + "coin": tool.COIN, + }, + "commitment": { + "algorithm": "sorted-outpoint-merkle-sha256", + "leaf_domain": "BTFUTXO1", + "node_domain": "BTFNODE1", + "root": root, + "leaf_count": len(keys), + }, + "target": { + "outpoint": outpoint, + "txid": txid, + "vout": vout, + "leaf_index": index, + "leaf_hash": leaf_hashes[index].hex(), + "utxo": utxo, + }, + "proof": proof_path, + "scanned": { + "raw_block_count": len(raw_blocks), + "side_blocks_ignored": len(raw_blocks) - len(blocks), + "main_chain_block_count": len(blocks), + "first_height": 0, + "last_height": len(blocks) - 1, + "tip_hash": tip["hash"], + "tip_time": tip["time"], + "tip_time_utc": tip["time_utc"], + }, + "warnings": { + "legacy_duplicate_output_overwrites": len(stats["duplicate_outputs"]), + }, + "verdict": "UTXO_INCLUSION_PROOF_CREATED", + } + return report + + +def print_text(proof): + data_hash = hashlib.sha256(canonical_bytes(proof)).hexdigest() + target = proof["target"] + utxo = target["utxo"] + print("Bitflash UTXO inclusion proof") + print("schema: %s" % proof["schema"]) + print("canonical proof sha256: %s" % data_hash) + print() + print("Target") + print(" outpoint: %s" % target["outpoint"]) + print(" value: %s BTF" % utxo["value_btf"]) + print(" height: %d" % utxo["created_height"]) + print(" index: %d of %d" % (target["leaf_index"], proof["commitment"]["leaf_count"])) + print() + print("Commitment") + print(" root: %s" % proof["commitment"]["root"]) + print(" path: %d sibling hashes" % len(proof["proof"])) + print(" tip: %s" % proof["scanned"]["tip_hash"]) + print() + print("Verdict: %s" % proof["verdict"]) + + +def default_datadir(): + if os.name == "nt" and os.environ.get("APPDATA"): + return os.path.join(os.environ["APPDATA"], "Bitflash") + if os.environ.get("HOME"): + return os.path.join(os.environ["HOME"], ".bitflash") + return "." + + +def main(argv): + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("outpoint", help="unspent output to prove, as txid:vout") + ap.add_argument("--datadir", default=default_datadir(), help="Bitflash data directory") + ap.add_argument("--max-blocks", type=int, default=0, + help="number of blocks to scan; 0 means all block files") + ap.add_argument("--json", action="store_true", help="print canonical JSON") + ap.add_argument("--out", help="write canonical JSON proof to this path") + args = ap.parse_args(argv) + + proof = build_proof(args) + data = canonical_bytes(proof) + if args.out: + Path(args.out).write_bytes(data) + if args.json: + sys.stdout.buffer.write(data) + else: + print_text(proof) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) 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 @@ +#!/usr/bin/env python3 +"""Verify a Bitflash UTXO inclusion proof against a UTXO root.""" + +import argparse +import hashlib +import json +import sys +from pathlib import Path + + +SCHEMA = "bitflash-utxo-inclusion-proof-1" +LEAF_DOMAIN = "BTFUTXO1" +NODE_DOMAIN = b"BTFNODE1|" + + +def canonical_bytes(obj): + return (json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + "\n").encode("ascii") + + +def sha256(data): + return hashlib.sha256(data).digest() + + +def utxo_leaf_hash(outpoint, utxo): + payload = "|".join([ + LEAF_DOMAIN, + outpoint, + str(utxo["value_satoshis"]), + utxo["script_pub_key_hex"], + str(utxo["created_height"]), + utxo["created_txid"], + ]).encode("ascii") + return sha256(payload) + + +def load_proof(path): + try: + return json.loads(Path(path).read_text(encoding="ascii")) + except Exception as e: + raise SystemExit("could not read proof: %s" % e) + + +def verify(proof, expected_root=None): + failures = [] + if proof.get("schema") != SCHEMA: + failures.append("unexpected schema") + commitment = proof.get("commitment", {}) + if commitment.get("algorithm") != "sorted-outpoint-merkle-sha256": + failures.append("unexpected commitment algorithm") + if commitment.get("leaf_domain") != LEAF_DOMAIN: + failures.append("unexpected leaf domain") + if commitment.get("node_domain") != "BTFNODE1": + failures.append("unexpected node domain") + + target = proof.get("target", {}) + outpoint = target.get("outpoint", "") + utxo = target.get("utxo", {}) + if outpoint != "%s:%s" % (target.get("txid"), target.get("vout")): + failures.append("target outpoint does not match txid:vout") + + try: + current = utxo_leaf_hash(outpoint, utxo) + leaf_ok = target.get("leaf_hash") == current.hex() + except Exception as e: + failures.append("could not hash target leaf: %s" % e) + current = b"\x00" * 32 + leaf_ok = False + + if not leaf_ok: + failures.append("target leaf hash mismatch") + + for i, step in enumerate(proof.get("proof", [])): + try: + sibling = bytes.fromhex(step["hash"]) + except Exception: + failures.append("bad sibling hash at proof level %d" % i) + sibling = b"\x00" * 32 + if len(sibling) != 32: + failures.append("bad sibling hash length at proof level %d" % i) + side = step.get("side") + if side == "right": + current = sha256(NODE_DOMAIN + current + sibling) + elif side == "left": + current = sha256(NODE_DOMAIN + sibling + current) + else: + failures.append("bad sibling side at proof level %d" % i) + + computed_root = current.hex() + root = expected_root or commitment.get("root") + if computed_root != root: + failures.append("computed root does not match expected root") + + return { + "schema": "bitflash-utxo-proof-verification-1", + "deterministic": True, + "target": { + "outpoint": outpoint, + "value_satoshis": utxo.get("value_satoshis"), + "value_btf": utxo.get("value_btf"), + "created_height": utxo.get("created_height"), + }, + "commitment": { + "expected_root": root, + "computed_root": computed_root, + "leaf_hash": target.get("leaf_hash"), + "path_length": len(proof.get("proof", [])), + }, + "checks": { + "schema_is_supported": proof.get("schema") == SCHEMA, + "leaf_hash_matches_target": leaf_ok, + "root_matches": computed_root == root, + }, + "failures": failures, + "verdict": "UTXO_PROOF_VERIFIED" if not failures else "UTXO_PROOF_FAILED", + } + + +def print_text(result): + data_hash = hashlib.sha256(canonical_bytes(result)).hexdigest() + print("Bitflash UTXO proof verifier") + print("schema: %s" % result["schema"]) + print("canonical verification sha256: %s" % data_hash) + print() + print("Target") + print(" outpoint: %s" % result["target"]["outpoint"]) + print(" value: %s BTF" % result["target"]["value_btf"]) + print(" height: %s" % result["target"]["created_height"]) + print() + print("Commitment") + print(" expected: %s" % result["commitment"]["expected_root"]) + print(" computed: %s" % result["commitment"]["computed_root"]) + print(" path: %d sibling hashes" % result["commitment"]["path_length"]) + print() + print("Checks") + for key in sorted(result["checks"]): + print(" %s %s" % ("OK " if result["checks"][key] else "FAIL", key)) + if result["failures"]: + print() + print("Failures") + for failure in result["failures"]: + print(" - %s" % failure) + print() + print("Verdict: %s" % result["verdict"]) + + +def main(argv): + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("proof", help="canonical JSON proof from prove-utxo.py") + ap.add_argument("--root", help="expected UTXO root; defaults to root embedded in proof") + ap.add_argument("--json", action="store_true", help="print canonical JSON") + args = ap.parse_args(argv) + + result = verify(load_proof(args.proof), args.root) + data = canonical_bytes(result) + if args.json: + sys.stdout.buffer.write(data) + else: + print_text(result) + return 0 if result["verdict"] == "UTXO_PROOF_VERIFIED" else 1 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) -- cgit v1.2.3