tools: add UTXO inclusion proofs

2 files changed, 365 insertions, 0 deletions

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:]))