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') 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