From 1839347459759fb3b23cc747ec4368b724fee008 Mon Sep 17 00:00:00 2001
From: Nsable-sh <nsable-sh@noreply.git.bitflash.network>
Date: Tue, 11 Aug 2026 14:13:26 -0300
Subject: tools: harden chain audit scripts

---
 scripts/verify-utxo-set.py | 375 +++------------------------------------------
 1 file changed, 21 insertions(+), 354 deletions(-)

(limited to 'scripts/verify-utxo-set.py')

diff --git a/scripts/verify-utxo-set.py b/scripts/verify-utxo-set.py
index 12b024f..b5bf3d5 100644
--- a/scripts/verify-utxo-set.py
+++ b/scripts/verify-utxo-set.py
@@ -1,363 +1,24 @@
 #!/usr/bin/env python3
-"""Build a deterministic Bitflash UTXO-set commitment from local blk*.dat.
-
-This is deliberately outside consensus. It lets auditors reconstruct the chain
-state from disk blocks and compare the same canonical UTXO root independently.
-"""
+"""Build a deterministic Bitflash UTXO-set commitment from local blk*.dat."""
 
 import argparse
-import datetime as _dt
 import hashlib
-import json
-import os
-import struct
 import sys
 from pathlib import Path
 
-
-MAGIC = bytes([0xBF, 0x20, 0x5C, 0xFD])
-COIN = 100_000_000
-GENESIS_HASH = "5bd7cb255d814e48cebcdfb72da4dc87b34bd774227f8ceb546c5640f4bdc169"
-GENESIS_MERKLE = "1a45b4482532abb29b10e234d3f13132230525a339ecea91658ffa675a5b1325"
-
-
-class ParseError(Exception):
-    pass
-
-
-class Reader:
-    def __init__(self, data):
-        self.data = data
-        self.pos = 0
-
-    def take(self, n):
-        if self.pos + n > len(self.data):
-            raise ParseError("unexpected end of data")
-        out = self.data[self.pos:self.pos + n]
-        self.pos += n
-        return out
-
-    def u8(self):
-        return self.take(1)[0]
-
-    def u32(self):
-        return struct.unpack("<I", self.take(4))[0]
-
-    def i32(self):
-        return struct.unpack("<i", self.take(4))[0]
-
-    def i64(self):
-        return struct.unpack("<q", self.take(8))[0]
-
-
-def sha256(data):
-    return hashlib.sha256(data).digest()
-
-
-def sha256d(data):
-    return hashlib.sha256(hashlib.sha256(data).digest()).digest()
-
-
-def hash_hex(serialized):
-    return sha256d(serialized)[::-1].hex()
-
-
-def compact_size(r):
-    first = r.u8()
-    if first < 253:
-        return first
-    if first == 253:
-        return struct.unpack("<H", r.take(2))[0]
-    if first == 254:
-        return r.u32()
-    return struct.unpack("<Q", r.take(8))[0]
-
-
-def read_varbytes(r):
-    return r.take(compact_size(r))
-
-
-def ser_compact_size(n):
-    if n < 253:
-        return bytes([n])
-    if n <= 0xFFFF:
-        return b"\xfd" + struct.pack("<H", n)
-    if n <= 0xFFFFFFFF:
-        return b"\xfe" + struct.pack("<I", n)
-    return b"\xff" + struct.pack("<Q", n)
-
-
-def ser_varbytes(b):
-    return ser_compact_size(len(b)) + b
-
-
-def ser_tx(tx):
-    out = bytearray()
-    out += struct.pack("<i", tx["version"])
-    out += ser_compact_size(len(tx["vin"]))
-    for txin in tx["vin"]:
-        out += bytes.fromhex(txin["prevout_hash"])[::-1]
-        out += struct.pack("<I", txin["prevout_n"])
-        out += ser_varbytes(bytes.fromhex(txin["script_sig_hex"]))
-        out += struct.pack("<I", txin["sequence"])
-    out += ser_compact_size(len(tx["vout"]))
-    for txout in tx["vout"]:
-        out += struct.pack("<q", txout["value_satoshis"])
-        out += ser_varbytes(bytes.fromhex(txout["script_pub_key_hex"]))
-    out += struct.pack("<I", tx["lock_time"])
-    return bytes(out)
-
-
-def parse_tx(r):
-    tx = {"version": r.i32(), "vin": [], "vout": [], "lock_time": 0}
-    for _ in range(compact_size(r)):
-        tx["vin"].append({
-            "prevout_hash": r.take(32)[::-1].hex(),
-            "prevout_n": r.u32(),
-            "script_sig_hex": read_varbytes(r).hex(),
-            "sequence": r.u32(),
-        })
-    for _ in range(compact_size(r)):
-        value = r.i64()
-        tx["vout"].append({
-            "value_satoshis": value,
-            "value_btf": format_money(value),
-            "script_pub_key_hex": read_varbytes(r).hex(),
-        })
-    tx["lock_time"] = r.u32()
-    tx["txid"] = hash_hex(ser_tx(tx))
-    return tx
-
-
-def merkle_root(txids):
-    if not txids:
-        return "0" * 64
-    layer = [bytes.fromhex(x)[::-1] for x in txids]
-    while len(layer) > 1:
-        nxt = []
-        for i in range(0, len(layer), 2):
-            left = layer[i]
-            right = layer[i + 1] if i + 1 < len(layer) else left
-            nxt.append(sha256d(left + right))
-        layer = nxt
-    return layer[0][::-1].hex()
-
-
-def parse_block(payload, file_name="", offset=0):
-    r = Reader(payload)
-    header = r.take(80)
-    version, = struct.unpack("<i", header[0:4])
-    prev = header[4:36][::-1].hex()
-    merkle = header[36:68][::-1].hex()
-    ntime, bits, nonce = struct.unpack("<III", header[68:80])
-    txs = [parse_tx(r) for _ in range(compact_size(r))]
-    if r.pos != len(payload):
-        raise ParseError("trailing bytes inside block")
-    txids = [tx["txid"] for tx in txs]
-    return {
-        "hash": hash_hex(header),
-        "version": version,
-        "previous_hash": prev,
-        "merkle_root": merkle,
-        "computed_merkle_root": merkle_root(txids),
-        "time": ntime,
-        "time_utc": utc(ntime),
-        "bits": "0x%08x" % bits,
-        "nonce": nonce,
-        "tx_count": len(txs),
-        "transactions": txs,
-        "source_file": file_name,
-        "source_offset": offset,
-    }
-
-
-def read_blocks(datadir, max_blocks):
-    blocks = []
-    for path in sorted(Path(datadir).glob("blk*.dat")):
-        data = path.read_bytes()
-        pos = 0
-        while True:
-            idx = data.find(MAGIC, pos)
-            if idx < 0:
-                break
-            if idx + 8 > len(data):
-                break
-            size = struct.unpack("<I", data[idx + 4:idx + 8])[0]
-            start = idx + 8
-            end = start + size
-            if end > len(data):
-                break
-            blocks.append(parse_block(data[start:end], path.name, start))
-            if max_blocks and len(blocks) >= max_blocks:
-                return blocks
-            pos = end
-    return blocks
-
-
-def select_main_chain(blocks):
-    by_hash = {}
-    for block in blocks:
-        by_hash.setdefault(block["hash"], block)
-    if GENESIS_HASH not in by_hash:
-        raise ParseError("genesis block not found in block files")
-
-    children = {}
-    for block in by_hash.values():
-        children.setdefault(block["previous_hash"], []).append(block)
-    for v in children.values():
-        v.sort(key=lambda b: (b["hash"], b["source_file"], b["source_offset"]))
-
-    best = []
-    stack = [(by_hash[GENESIS_HASH], [by_hash[GENESIS_HASH]])]
-    while stack:
-        block, path = stack.pop()
-        if len(path) > len(best) or (len(path) == len(best) and block["hash"] < best[-1]["hash"]):
-            best = path
-        for child in children.get(block["hash"], []):
-            stack.append((child, path + [child]))
-    return best
-
-
-def is_coinbase(tx):
-    return (
-        len(tx["vin"]) == 1
-        and tx["vin"][0]["prevout_hash"] == "0" * 64
-        and tx["vin"][0]["prevout_n"] == 0xFFFFFFFF
-    )
-
-
-def utxo_leaf_hash(outpoint, utxo):
-    # Text fields are length-delimited by separators and all variable data is
-    # hex or decimal ASCII. That keeps the commitment language-neutral.
-    payload = "|".join([
-        "BTFUTXO1",
-        outpoint,
-        str(utxo["value_satoshis"]),
-        utxo["script_pub_key_hex"],
-        str(utxo["created_height"]),
-        utxo["created_txid"],
-    ]).encode("ascii")
-    return sha256(payload)
-
-
-def utxo_root(utxos):
-    leaves = [utxo_leaf_hash(k, utxos[k]) for k in sorted(utxos)]
-    if not leaves:
-        return sha256(b"BTFUTXO1|empty").hex()
-    layer = leaves
-    while len(layer) > 1:
-        nxt = []
-        for i in range(0, len(layer), 2):
-            left = layer[i]
-            right = layer[i + 1] if i + 1 < len(layer) else left
-            nxt.append(sha256(b"BTFNODE1|" + left + right))
-        layer = nxt
-    return layer[0].hex()
-
-
-def apply_blocks(blocks):
-    utxos = {}
-    spent = 0
-    missing_spends = []
-    duplicate_outputs = []
-    total_coinbase = 0
-    total_fees = 0
-    normal_txs = 0
-
-    for height, block in enumerate(blocks):
-        if block["computed_merkle_root"] != block["merkle_root"]:
-            raise ParseError("bad merkle root at height %d" % height)
-        for tx_index, tx in enumerate(block["transactions"]):
-            coinbase = is_coinbase(tx)
-            input_value = 0
-            output_value = sum(out["value_satoshis"] for out in tx["vout"])
-            if coinbase:
-                total_coinbase += output_value
-            else:
-                normal_txs += 1
-                for txin in tx["vin"]:
-                    prev = "%s:%d" % (txin["prevout_hash"], txin["prevout_n"])
-                    prev_utxo = utxos.pop(prev, None)
-                    if prev_utxo is None:
-                        missing_spends.append({
-                            "height": height,
-                            "txid": tx["txid"],
-                            "missing_outpoint": prev,
-                        })
-                    else:
-                        spent += prev_utxo["value_satoshis"]
-                        input_value += prev_utxo["value_satoshis"]
-                if input_value >= output_value:
-                    total_fees += input_value - output_value
-            for n, out in enumerate(tx["vout"]):
-                key = "%s:%d" % (tx["txid"], n)
-                if key in utxos:
-                    duplicate_outputs.append({"height": height, "outpoint": key})
-                utxos[key] = {
-                    "txid": tx["txid"],
-                    "vout": n,
-                    "value_satoshis": out["value_satoshis"],
-                    "value_btf": out["value_btf"],
-                    "script_pub_key_hex": out["script_pub_key_hex"],
-                    "created_height": height,
-                    "created_block_hash": block["hash"],
-                    "created_txid": tx["txid"],
-                    "coinbase": coinbase,
-                }
-
-    total_unspent = sum(u["value_satoshis"] for u in utxos.values())
-    return {
-        "utxos": utxos,
-        "stats": {
-            "utxo_count": len(utxos),
-            "spent_outputs": spent,
-            "total_unspent_satoshis": total_unspent,
-            "total_unspent_btf": format_money(total_unspent),
-            "total_coinbase_satoshis": total_coinbase,
-            "total_coinbase_btf": format_money(total_coinbase),
-            "coinbase_minus_unspent_satoshis": total_coinbase - total_unspent,
-            "coinbase_minus_unspent_btf": format_money(total_coinbase - total_unspent),
-            "total_fees_satoshis": total_fees,
-            "total_fees_btf": format_money(total_fees),
-            "normal_transactions": normal_txs,
-            "missing_spends": missing_spends,
-            "duplicate_outputs": duplicate_outputs,
-        },
-    }
-
-
-def format_money(value):
-    sign = "-" if value < 0 else ""
-    value = abs(value)
-    return "%s%d.%08d" % (sign, value // COIN, value % COIN)
-
-
-def utc(ts):
-    return _dt.datetime.fromtimestamp(ts, _dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
-
-
-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 canonical_bytes(report):
-    return (json.dumps(report, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + "\n").encode("ascii")
+import bitflash_chain as chain
 
 
-def build_report(raw_blocks, include_utxos=False):
-    blocks = select_main_chain(raw_blocks)
-    state = apply_blocks(blocks)
+def build_report(raw_blocks, include_utxos=False, strict_duplicates=False):
+    blocks = chain.select_main_chain(raw_blocks)
+    state = chain.apply_blocks(blocks, strict_duplicates=strict_duplicates)
     utxos = state["utxos"]
     stats = state["stats"]
     tip = blocks[-1]
-    root = utxo_root(utxos)
+    root = chain.utxo_root(utxos)
     checks = {
-        "genesis_hash_matches_source": blocks[0]["hash"] == GENESIS_HASH,
-        "genesis_merkle_matches_source": blocks[0]["merkle_root"] == GENESIS_MERKLE,
+        "genesis_hash_matches_source": blocks[0]["hash"] == chain.GENESIS_HASH,
+        "genesis_merkle_matches_source": blocks[0]["merkle_root"] == chain.GENESIS_MERKLE,
         "all_scanned_merkle_roots_match": all(b["computed_merkle_root"] == b["merkle_root"] for b in blocks),
         "no_missing_spends": len(stats["missing_spends"]) == 0,
     }
@@ -370,8 +31,8 @@ def build_report(raw_blocks, include_utxos=False):
         "chain": {
             "name": "Bitflash",
             "ticker": "BTF",
-            "message_start_hex": MAGIC.hex(),
-            "coin": COIN,
+            "message_start_hex": chain.MAGIC.hex(),
+            "coin": chain.COIN,
         },
         "commitment": {
             "algorithm": "sorted-outpoint-merkle-sha256",
@@ -400,7 +61,7 @@ def build_report(raw_blocks, include_utxos=False):
 
 
 def print_text(report):
-    data_hash = hashlib.sha256(canonical_bytes(report)).hexdigest()
+    data_hash = hashlib.sha256(chain.canonical_bytes(report)).hexdigest()
     print("Bitflash UTXO-set commitment verifier")
     print("schema: %s" % report["schema"])
     print("canonical report sha256: %s" % data_hash)
@@ -437,20 +98,26 @@ def print_text(report):
 
 def main(argv):
     ap = argparse.ArgumentParser(description=__doc__)
-    ap.add_argument("--datadir", default=default_datadir(), help="Bitflash data directory")
+    ap.add_argument("--datadir", default=chain.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("--include-utxos", action="store_true",
                     help="include the full sorted UTXO map in JSON output")
+    ap.add_argument("--strict-duplicates", action="store_true",
+                    help="fail instead of using legacy last-write-wins duplicate output handling")
     ap.add_argument("--json", action="store_true", help="print canonical JSON")
     ap.add_argument("--out", help="write canonical JSON to this path")
     args = ap.parse_args(argv)
 
-    blocks = read_blocks(args.datadir, args.max_blocks)
+    blocks = chain.read_blocks(args.datadir, args.max_blocks)
     if not blocks:
         raise SystemExit("no Bitflash blocks found in %s" % args.datadir)
-    report = build_report(blocks, include_utxos=args.include_utxos)
-    data = canonical_bytes(report)
+    try:
+        report = build_report(blocks, include_utxos=args.include_utxos,
+                              strict_duplicates=args.strict_duplicates)
+    except chain.ParseError as e:
+        raise SystemExit(str(e))
+    data = chain.canonical_bytes(report)
     if args.out:
         Path(args.out).write_bytes(data)
     if args.json:
-- 
cgit v1.2.3