feat(proofs): authenticate coinbases with AuxPoW merkle branches

README.md

@@ -13,7 +13,7 @@ This dataset covers blocks that fail those rules, including failures that can be
 - [`docs/schema.md`](docs/schema.md): fields and admission rules.
 - [`docs/notes.md`](docs/notes.md): replay behaviour and incident notes.
 - `blocks/{height}-{hash}.bin`: full block, when available.
-- `proofs/{height}-{hash}.json`: for a P2SH record without a body, the failing transaction and the block's ordered transaction IDs.
+- `proofs/{height}-{hash}.json`: for a record without a body, one transaction authenticated against the header by the block's ordered transaction IDs or, for a coinbase, by its merkle branch.
 - [`data/reported-blocks.jsonl`](data/reported-blocks.jsonl): blocks reported as invalid whose failure is not established.
 
 Merge-mined recoveries generally provide a header and coinbase rather than a full Bitcoin block.
@@ -25,8 +25,10 @@ Include the 80-byte header, its decoded hash, parent hash and timestamp, height
 The header must meet the PoW target encoded in its `nBits`.
 
 Include `context` fields needed to establish the failure: BIP34 coinbase height and scriptSig, `parent_mtp` for `time_below_mtp`, or `expected_nbits` for `nbits_retarget_not_applied`.
+When the coinbase comes from an AuxPoW record, add a proof file with the coinbase and its merkle branch so CI can bind the scriptSig to the header.
 Omit unknown optional fields.
 When the pool is known, give `pool` with `pool_basis`: `tag` for a coinbase tag, `address` for a payout address listed in [mining-pools](https://github.com/bitcoin-data/mining-pools), or `reported` when only a contemporaneous report names the pool.
+A tag or address attribution needs a body or a coinbase proof; a report or mining-pools listing goes in `pool_provenance`.
 
 A block whose failure is only reported goes in [`data/reported-blocks.jsonl`](data/reported-blocks.jsonl) with its sources, until the evidence turns up.
 

ci/block_evidence.py

@@ -3,8 +3,8 @@
 This is not a consensus validator: no UTXO lookup or historical chain
 reconstruction takes place here, and the only script execution is the
 library's evaluation of one named input for the P2SH rule.
-Transaction IDs and the merkle root bind the checked non-witness data to the
-Bitcoin header. Parsing consumes the entire file so truncation cannot satisfy
+A block's transaction IDs, or a coinbase's merkle branch, and the merkle root
+bind the checked non-witness data to the Bitcoin header. Parsing consumes the entire file so truncation cannot satisfy
 a rule's requirement for a complete block body.
 
 Sigop counting mirrors Core's CScript::GetSigOpCount, GetTransactionSigOpCost
@@ -134,14 +134,36 @@ def checked_txids(txids: object, header: CBlockHeader) -> list[str]:
     return txids
 
 
+def checked_branch(branch: object) -> list[bytes]:
+    """Require a list of 64-hex sibling hashes, returned in internal byte order; empty for a single-transaction block."""
+    if not isinstance(branch, list) or any(not isinstance(node, str) or not HEX64.fullmatch(node) for node in branch):
+        raise ValueError("merkle_branch must be an array of 64-hex strings")
+    return [lx(node) for node in branch]
+
+
 def read_proof(data: bytes, header: CBlockHeader) -> CTransaction:
-    """Read a transaction and the ordered txid list that places it in the header's block."""
+    """Read one transaction placed in the header's block by a proof file.
+
+    `txids` places any transaction through the block's complete ordered txid
+    list. `merkle_branch` places a coinbase through the sibling hashes from
+    index 0 up to the merkle root, as AuxPoW records carry it; only index 0
+    has an unambiguous path without a position field.
+    """
     proof = json.loads(data)
-    if not isinstance(proof, dict) or set(proof) != {"transaction", "txids"}:
-        raise ValueError("proof must be an object with transaction and txids")
+    if not isinstance(proof, dict) or set(proof) not in ({"transaction", "txids"}, {"transaction", "merkle_branch"}):
+        raise ValueError("proof must be an object with transaction and either txids or merkle_branch")
     tx = read_transaction(bytes.fromhex(proof["transaction"]))
-    if b2lx(tx.GetTxid()) not in checked_txids(proof["txids"], header):
-        raise ValueError("proof transaction is not in the block's txid list")
+    if "txids" in proof:
+        if b2lx(tx.GetTxid()) not in checked_txids(proof["txids"], header):
+            raise ValueError("proof transaction is not in the block's txid list")
+        return tx
+    if not tx.is_coinbase():
+        raise ValueError("merkle_branch proof requires a coinbase transaction")
+    node = tx.GetTxid()
+    for sibling in checked_branch(proof["merkle_branch"]):
+        node = sha256d(node + sibling)
+    if node != header.hashMerkleRoot:
+        raise ValueError("coinbase merkle branch does not reproduce the header's merkle root")
     return tx
 
 

ci/sanity-check.py

@@ -38,7 +38,7 @@
 REPORTED_REQUIRED = {"height", "hash", "reported_failure", "sources"}
 CONTEXT_FIELDS = {
     "expected_nbits", "parent_mtp", "coinbase_height", "coinbase_scriptsig_hex",
-    "pool", "pool_basis", "parent_kind", "missing_prevout", "parent_txid", "failing_prevout",
+    "pool", "pool_basis", "pool_provenance", "parent_kind", "missing_prevout", "parent_txid", "failing_prevout",
 }
 OUTPOINT = re.compile(r"[0-9a-f]{64}:(?:0|[1-9][0-9]*)")
 OBSERVATION_REQUIRED = {"channel", "source", "provenance"}
@@ -228,8 +228,15 @@ def check_context(record: dict[str, Any]) -> None:
             raise ValueError("pool requires pool_basis")
         if details["pool_basis"] not in tuple(POOL_BASES):
             raise ValueError(f"pool_basis must be one of {sorted(POOL_BASES)}")
+        # A report or a mining-pools listing is the attribution's evidence; a tag is read from the coinbase itself.
+        if details["pool_basis"] != "tag" and "pool_provenance" not in details:
+            raise ValueError(f"pool_basis {details['pool_basis']} requires pool_provenance")
+        if "pool_provenance" in details and not http_url(details["pool_provenance"]):
+            raise ValueError("pool_provenance must be an HTTP(S) URL")
     elif "pool_basis" in details:
         raise ValueError("pool_basis requires pool")
+    elif "pool_provenance" in details:
+        raise ValueError("pool_provenance requires pool")
     if "parent_txid" in details:
         hex_value(details, "parent_txid", 32)
     for name in ("missing_prevout", "failing_prevout"):
@@ -308,7 +315,7 @@ def check_local_evidence(record: dict[str, Any], header: CBlockHeader) -> None:
     """Check the named rule against header and context already in the record.
 
     Parent MTP, expected nBits and parent_kind are not looked up on the chain.
-    Without a body, a coinbase scriptSig cannot be bound to the header.
+    Without a body or a coinbase proof, a coinbase scriptSig cannot be bound to the header.
     """
     rule = record["rule"]
     context = record.get("context", {})
@@ -366,7 +373,7 @@ def check_failure_evidence(record: dict[str, Any], block: CBlock | None, prevout
     """Require a checked failure; observations cannot substitute for bytes."""
     mode = RULES[record["rule"]][2]
     if proof is not None and mode != "p2sh":
-        raise ValueError("proof files apply only to the P2SH rule")
+        raise ValueError("spend proofs apply only to the P2SH rule")
     if mode == "local":
         return
     if block is None and proof is None:
@@ -467,23 +474,32 @@ def check_dataset(path: Path | str = DATA_PATH, blocks_dir: Path | str = BLOCKS_
             context_count += bool(record.get("context"))
             observation_count += check_observations(record)
             check_local_evidence(record, parsed_header)
-            evidence_block = None
+            evidence_block = coinbase = proof_transaction = None
             if block is not None:
                 data = block.read_bytes()
                 if data[:80] != header:
                     raise ValueError(f"{block}: first 80 bytes do not match dataset header")
                 evidence_block = read_block(data)
                 if height >= 481824:
                     verify_witness_commitment(evidence_block)
-                details = record.get("context", {})
-                if "coinbase_scriptsig_hex" in details:
-                    if evidence_block.vtx[0].vin[0].scriptSig.hex() != details["coinbase_scriptsig_hex"]:
-                        raise ValueError("coinbase scriptSig does not match context")
-            proof_transaction = None
+                coinbase = evidence_block.vtx[0]
             if proof is not None:
                 if block is not None:
                     raise ValueError("a record has either a block body or a proof file, not both")
+                # A proved coinbase binds the record's coinbase context; a proved
+                # spend is rule evidence and is checked with the failure.
                 proof_transaction = read_proof(proof.read_bytes(), parsed_header)
+                if proof_transaction.is_coinbase():
+                    coinbase = proof_transaction
+                    proof_transaction = None
+                    if "coinbase_scriptsig_hex" not in record.get("context", {}):
+                        raise ValueError("coinbase proof requires coinbase_scriptsig_hex in context")
+            details = record.get("context", {})
+            scriptsig = details.get("coinbase_scriptsig_hex")
+            if coinbase is not None and scriptsig is not None and coinbase.vin[0].scriptSig.hex() != scriptsig:
+                raise ValueError("coinbase scriptSig does not match context")
+            if details.get("pool_basis") in ("tag", "address") and coinbase is None:
+                raise ValueError(f"pool_basis {details['pool_basis']} requires a body or a coinbase proof")
             check_failure_evidence(record, evidence_block, prevouts_dir, fetch_prevouts, apis, proof_transaction)
         except (OSError, ValueError) as exc:
             problems.append(f"{where}: {exc}")

ci/test_block_evidence.py

@@ -1,13 +1,14 @@
 """Block parsing, commitments and sigop counting without mining or API access."""
 
+import json
 import unittest
 
-from bitcoin.core import CBlock, COutPoint, CTransaction, CTxIn, CTxOut, CTxWitness, CTxInWitness, b2lx
+from bitcoin.core import CBlock, CBlockHeader, COutPoint, CTransaction, CTxIn, CTxOut, CTxWitness, CTxInWitness, b2lx
 from bitcoin.core.script import CScript, CScriptWitness, OP_TRUE
 
 from block_evidence import (
     MAX_MONEY, coinbase_amounts, confirmed_at_or_after, establishes_rule, omitted_prevouts, p2sh_spend_fails, read_block,
-    read_transaction, reuses_parent_transaction, sha256d, sigop_count, witness_sigops,
+    read_proof, read_transaction, reuses_parent_transaction, sha256d, sigop_count, witness_sigops,
 )
 
 # The 123-byte spend included by 89 blocks in April to July 2012, and the transaction that funded it.
@@ -38,6 +39,16 @@ def block(*transactions):
     return bytes(36) + root + bytes(12) + bytes([len(transactions)]) + b"".join(wire for wire, _ in transactions)
 
 
+def coinbase_branch(txids):
+    """Sibling hashes from the coinbase up to the merkle root, in RPC/display order, from the library's full tree."""
+    tree = CBlock.build_merkle_tree_from_txids(txids)
+    branch, offset, width = [], 0, len(txids)
+    while width > 1:
+        branch.append(b2lx(tree[offset + 1]))
+        offset, width = offset + width, (width + 1) >> 1
+    return branch
+
+
 class BlockEvidenceChecks(unittest.TestCase):
     def test_overflow_boundary(self):
         """Only outputs above MAX_MONEY establish the output-too-large failure."""
@@ -120,6 +131,25 @@ def test_parent_transaction_reuse_excludes_coinbases_and_absent_transactions(sel
             with self.subTest(case=name):
                 self.assertEqual(reuses_parent_transaction(body, parent, witness), expected)
 
+    def test_coinbase_branch_proof_places_only_a_coinbase_at_index_zero(self):
+        """A branch of siblings from the coinbase reproduces the root; a wrong sibling or a non-coinbase transaction does not."""
+        coinbase, spend, other = transaction(), transaction(prev_hash=b"\x11" * 32, vout=0), transaction(prev_hash=b"\x22" * 32, vout=0)
+        body = block(coinbase, spend, other)
+        header = CBlockHeader.deserialize(body[:80])
+        branch = coinbase_branch([sha256d(stripped) for _, stripped in (coinbase, spend, other)])
+        proof = {"transaction": coinbase[0].hex(), "merkle_branch": branch}
+        self.assertTrue(read_proof(json.dumps(proof).encode(), header).is_coinbase())
+        with self.subTest(case="single-transaction block"):
+            single = CBlockHeader.deserialize(block(coinbase)[:80])
+            self.assertTrue(read_proof(json.dumps(dict(proof, merkle_branch=[])).encode(), single).is_coinbase())
+        cases = (
+            ("wrong sibling", dict(proof, merkle_branch=branch[::-1]), "does not reproduce"),
+            ("spend with a branch", dict(proof, transaction=spend[0].hex()), "requires a coinbase"),
+        )
+        for case, content, error in cases:
+            with self.subTest(case=case), self.assertRaisesRegex(ValueError, error):
+                read_proof(json.dumps(content).encode(), header)
+
 
 class CoinbaseAmountChecks(unittest.TestCase):
     def test_subsidy_boundaries(self):