From 1839347459759fb3b23cc747ec4368b724fee008 Mon Sep 17 00:00:00 2001 From: Nsable-sh Date: Tue, 11 Aug 2026 14:13:26 -0300 Subject: tools: harden chain audit scripts --- scripts/prove-utxo.py | 43 ++++++++++++++----------------------------- 1 file changed, 14 insertions(+), 29 deletions(-) (limited to 'scripts/prove-utxo.py') diff --git a/scripts/prove-utxo.py b/scripts/prove-utxo.py index 7a10d2a..9f2704b 100644 --- a/scripts/prove-utxo.py +++ b/scripts/prove-utxo.py @@ -7,27 +7,16 @@ 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 +import bitflash_chain as chain 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") @@ -80,25 +69,27 @@ def build_merkle_proof(keys, leaf_hashes, index): 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) + raw_blocks = chain.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) + try: + blocks = chain.select_main_chain(raw_blocks) + state = chain.apply_blocks(blocks, strict_duplicates=args.strict_duplicates) + except chain.ParseError as e: + raise SystemExit(str(e)) 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] + leaf_hashes = [chain.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) + direct_root = chain.utxo_root(utxos) if root != direct_root: raise SystemExit("internal proof root mismatch") @@ -110,8 +101,8 @@ def build_proof(args): "chain": { "name": "Bitflash", "ticker": "BTF", - "message_start_hex": tool.MAGIC.hex(), - "coin": tool.COIN, + "message_start_hex": chain.MAGIC.hex(), + "coin": chain.COIN, }, "commitment": { "algorithm": "sorted-outpoint-merkle-sha256", @@ -169,20 +160,14 @@ def print_text(proof): 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("--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("--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 proof to this path") args = ap.parse_args(argv) -- cgit v1.2.3 From caa6a9668fc2b2ed1192f7ebdc0b78621f45898d Mon Sep 17 00:00:00 2001 From: Nsable-sh Date: Tue, 11 Aug 2026 14:47:35 -0300 Subject: tools: address audit follow-ups --- scripts/bitflash_chain.py | 69 +++++++++++++++++++++----------------------- scripts/build-explorer.py | 5 +++- scripts/prove-utxo.py | 7 ++--- scripts/verify-utxo-proof.py | 10 +++---- 4 files changed, 44 insertions(+), 47 deletions(-) (limited to 'scripts/prove-utxo.py') diff --git a/scripts/bitflash_chain.py b/scripts/bitflash_chain.py index f3688f1..4a3231c 100644 --- a/scripts/bitflash_chain.py +++ b/scripts/bitflash_chain.py @@ -4,6 +4,7 @@ import datetime as _dt import hashlib import json +import mmap import os import struct from pathlib import Path @@ -11,6 +12,9 @@ from pathlib import Path MAGIC = bytes([0xBF, 0x20, 0x5C, 0xFD]) MAX_BLOCK_PAYLOAD = 32 * 1024 * 1024 +UTXO_LEAF_DOMAIN = "BTFUTXO1" +UTXO_NODE_DOMAIN_TAG = "BTFNODE1" +UTXO_NODE_DOMAIN = (UTXO_NODE_DOMAIN_TAG + "|").encode("ascii") COIN = 100_000_000 GENESIS_HASH = "5bd7cb255d814e48cebcdfb72da4dc87b34bd774227f8ceb546c5640f4bdc169" GENESIS_MERKLE = "1a45b4482532abb29b10e234d3f13132230525a339ecea91658ffa675a5b1325" @@ -181,40 +185,33 @@ def iter_block_files(datadir): def iter_blocks_from_file(path): path = Path(path) with path.open("rb") as f: - window = f.read(4) - offset = 0 - while len(window) == 4: - if window != MAGIC: - nxt = f.read(1) - if not nxt: + f.seek(0, os.SEEK_END) + file_size = f.tell() + if file_size == 0: + return + f.seek(0) + mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) + try: + pos = 0 + while True: + offset = mm.find(MAGIC, pos) + if offset < 0 or offset + 8 > file_size: break - window = window[1:] + nxt - offset += 1 - continue - - size_bytes = f.read(4) - if len(size_bytes) != 4: - break - size = struct.unpack(" MAX_BLOCK_PAYLOAD: - offset += 1 - f.seek(offset) - window = f.read(4) - continue - payload_offset = offset + 8 - payload = f.read(size) - if len(payload) != size: - break - try: - yield parse_block(payload, path.name, payload_offset) - except ParseError: - offset += 1 - f.seek(offset) - window = f.read(4) - continue - - offset = f.tell() - window = f.read(4) + size = struct.unpack(" MAX_BLOCK_PAYLOAD or end > file_size: + pos = offset + 1 + continue + payload = mm[payload_offset:end] + try: + yield parse_block(payload, path.name, payload_offset) + except ParseError: + pos = offset + 1 + continue + pos = end + finally: + mm.close() def read_blocks_from_files(paths, max_blocks=0): @@ -265,7 +262,7 @@ def is_coinbase(tx): def utxo_leaf_hash(outpoint, utxo): payload = "|".join([ - "BTFUTXO1", + UTXO_LEAF_DOMAIN, outpoint, str(utxo["value_satoshis"]), utxo["script_pub_key_hex"], @@ -285,7 +282,7 @@ def utxo_root(utxos): 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)) + nxt.append(sha256(UTXO_NODE_DOMAIN + left + right)) layer = nxt return layer[0].hex() @@ -327,9 +324,9 @@ def apply_blocks(blocks, strict_duplicates=False): for n, out in enumerate(tx["vout"]): key = "%s:%d" % (tx["txid"], n) if key in utxos: - duplicate_outputs.append({"height": height, "outpoint": key}) if strict_duplicates: raise ParseError("duplicate output at height %d: %s" % (height, key)) + duplicate_outputs.append({"height": height, "outpoint": key}) utxos[key] = { "txid": tx["txid"], "vout": n, diff --git a/scripts/build-explorer.py b/scripts/build-explorer.py index 0d86b3c..97a7f4e 100755 --- a/scripts/build-explorer.py +++ b/scripts/build-explorer.py @@ -35,7 +35,10 @@ def hash160(b): def spk_info(script_hex): - spk = bytes.fromhex(script_hex) + try: + spk = bytes.fromhex(script_hex) + except ValueError: + return {"type": "malformed", "hex": script_hex[:80]} if ( len(spk) == 25 and spk[0] == 0x76 and spk[1] == 0xA9 and spk[2] == 0x14 and spk[23] == 0x88 and spk[24] == 0xAC diff --git a/scripts/prove-utxo.py b/scripts/prove-utxo.py index 9f2704b..105e677 100644 --- a/scripts/prove-utxo.py +++ b/scripts/prove-utxo.py @@ -14,7 +14,6 @@ from pathlib import Path import bitflash_chain as chain SCHEMA = "bitflash-utxo-inclusion-proof-1" -NODE_DOMAIN = b"BTFNODE1|" def canonical_bytes(obj): @@ -61,7 +60,7 @@ def build_merkle_proof(keys, leaf_hashes, index): 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()) + next_layer.append(hashlib.sha256(chain.UTXO_NODE_DOMAIN + left + right).digest()) layer = next_layer current_index //= 2 level += 1 @@ -106,8 +105,8 @@ def build_proof(args): }, "commitment": { "algorithm": "sorted-outpoint-merkle-sha256", - "leaf_domain": "BTFUTXO1", - "node_domain": "BTFNODE1", + "leaf_domain": chain.UTXO_LEAF_DOMAIN, + "node_domain": chain.UTXO_NODE_DOMAIN_TAG, "root": root, "leaf_count": len(keys), }, diff --git a/scripts/verify-utxo-proof.py b/scripts/verify-utxo-proof.py index c29caa9..50d9356 100644 --- a/scripts/verify-utxo-proof.py +++ b/scripts/verify-utxo-proof.py @@ -10,8 +10,6 @@ from pathlib import Path import bitflash_chain as chain SCHEMA = "bitflash-utxo-inclusion-proof-1" -LEAF_DOMAIN = "BTFUTXO1" -NODE_DOMAIN = b"BTFNODE1|" def canonical_bytes(obj): @@ -32,9 +30,9 @@ def verify(proof, expected_root=None): commitment = proof.get("commitment", {}) if commitment.get("algorithm") != "sorted-outpoint-merkle-sha256": failures.append("unexpected commitment algorithm") - if commitment.get("leaf_domain") != LEAF_DOMAIN: + if commitment.get("leaf_domain") != chain.UTXO_LEAF_DOMAIN: failures.append("unexpected leaf domain") - if commitment.get("node_domain") != "BTFNODE1": + if commitment.get("node_domain") != chain.UTXO_NODE_DOMAIN_TAG: failures.append("unexpected node domain") target = proof.get("target", {}) @@ -64,9 +62,9 @@ def verify(proof, expected_root=None): failures.append("bad sibling hash length at proof level %d" % i) side = step.get("side") if side == "right": - current = chain.sha256(NODE_DOMAIN + current + sibling) + current = chain.sha256(chain.UTXO_NODE_DOMAIN + current + sibling) elif side == "left": - current = chain.sha256(NODE_DOMAIN + sibling + current) + current = chain.sha256(chain.UTXO_NODE_DOMAIN + sibling + current) else: failures.append("bad sibling side at proof level %d" % i) -- cgit v1.2.3