End of ze world: merging community Bitcoin forks into one verifiable ledger — notarization meta-chain, merge/verify how-to + working stdlib-only code

0-GIST.md

End of ze world: merging community Bitcoin forks into one ledger

Scenario: it's the end of the world, and the only useful crypto left is bitcoin: it runs on a phone or any computer, and it's a persistent, unfuckable-with record of comms and accounting.

Communities run their own forks: sovereign chains, fresh difficulty (CPU/phone mineable again), no permission needed. Fine. But communities amalgamate, and when they do, they want one merged history of everything that happened, without any chain having to die for it.

This is how: a notarization meta-chain. The chains stay alive; their histories get continuously woven into one merged ledger.

The key insight

A block header is already a Merkle commitment to a chain's entire history. So merging histories doesn't mean importing transactions; it means recording which block each community's chain was at, when. One 74-byte OP_RETURN per anchor interval. The whole merged ledger is cheap enough for the apocalypse: headers are 80 bytes each (all of BTC's history is about 77 MB), and an SPV proof of any event is a few KB, verifiable on a phone over a mesh link.

community A's chain --+
                      +--> notary --> anchor (OP_RETURN) --> MERGED LEDGER
community B's chain --+        |
                         every k-deep block
                         (anyone can verify with KB-scale proofs)

The anchor (74 bytes, inside the standard 80-byte nulldata rule)

OP_RETURN | "MRG1" (4B) | chain_id (2B) | height (4B LE) | block_hash (32B) | utxo_muhash (32B)

utxo_muhash is the chain's UTXO-set commitment (gettxoutsetinfo "muhash"): each anchor pins down both the history and the complete accounting state, who held what, at that height. That's the "record of comms and accounting", committed, forever.

Amalgamation, two ways

Ongoing merge (the steady state). Each community runs a notary: watch your chain, and every k-deep block, anchor it into the merged ledger. New history folds in as time passes. Cross-community ordering exists at each anchor point: a DAG of histories, growing together.

Retroactive amalgamation (a new community joins). They present their header chain from genesis (or a checkpoint); PoW is verified; an admission anchor is recorded. Their entire history, every tx since their genesis, is now part of the merged ledger, provable by anyone. No transactions imported, no balances moved, no chain retired.

Verify (phone-grade SPV)

Prove any event on any community chain is in the merged history:

1. Merkle branch: tx -> its block header (gettxoutproof)
2. Header walk: tx block -> anchored block (PoW + prevhash each step)
3. Decode the anchor OP_RETURN in the merged ledger; chain_id/height/hash must match
4. SPV-proof the anchor tx itself into the merged ledger's headers

A few KB of proofs total. No full node needed to verify. Ever.

Why this fits the end of the world

• Runs on anything: fresh chains start at trivial difficulty, so CPU/phone mining is viable again; verification is KB-scale.
• Matches social trust: a federation of community reps runs the notaries. Trust was always social; the math just makes the record tamper-evident. When comms are down, chains keep working alone; when comms return, anchors resume and the merged history heals forward.
• Censorship-resistant memory: once anchored, history can't be quietly rewritten: changing one old tx invalidates every header and anchor after it.

Honest limitations

• Shared record, not shared money: each community's coins stay sovereign. Cross-chain double-spends of shared-history UTXOs can't be prevented by anchoring; that needs a peg layer (bridge, atomic swaps, or drivechains). Amalgamate the accounting first; the money is a later, harder merge.
• The merged view lags by anchor interval + k confirmations; deeper reorgs are handled socially ("chain left the federation").
• Anchor placement trusts the federation keys; verification of what's anchored is trustless. Upgrade path when civilization reboots: native light-client verification in consensus, or ZK consensus proofs.

Working code (this gist)

• anchor.py: MRG1 payload codec
• notary.py: the daemon that folds new history into the merged ledger
• spv.py: stdlib-only SPV: PoW, header linkage, Core-faithful partial-Merkle-tree traversal
• verify.py / realverify.py: merged-history membership proofs (regtest + real mainnet)
• demo.sh: two-community end-to-end demo

Verified: regtest demo PASS (tx buried 6 deep, anchored, SPV-proven); real mainnet PASS (the 2010 pizza tx proven against the real chain; live tip headers verified at full current difficulty). Works against any Core-derived chain unchanged: BTC, BCH, BSV, XEC, DOGE, LTC, or your commune's fork.

anchor.py

"""MRG1 anchor payload codec.

Payload layout (74 bytes, fits the 80-byte standard nulldata relay limit):

    offset  size  field
    0       4     magic b"MRG1"
    4       2     chain_id   uint16 big-endian
    6       4     height     uint32 little-endian
    10      32    block_hash Bitcoin internal uint256 byte order
                            (= RPC display hex reversed)
    42      32    utxo_muhash raw bytes of the hex from gettxoutsetinfo

On-chain form: OP_RETURN <push of 74 bytes>. A 74-byte push is emitted by
Core as the direct push opcode 0x4a, but we also accept OP_PUSHDATA1.
"""
import struct

MAGIC = b"MRG1"
PAYLOAD_LEN = 74

OP_RETURN = 0x6A
OP_PUSHDATA1 = 0x4C


def encode(chain_id, height, block_hash_hex, muhash_hex):
    if not 0 <= chain_id <= 0xFFFF:
        raise ValueError("chain_id out of uint16 range")
    if not 0 <= height <= 0xFFFFFFFF:
        raise ValueError("height out of uint32 range")
    bh = bytes.fromhex(block_hash_hex)
    mh = bytes.fromhex(muhash_hex)
    if len(bh) != 32:
        raise ValueError("block_hash must be 32 bytes")
    if len(mh) != 32:
        raise ValueError("muhash must be 32 bytes")
    return MAGIC + struct.pack(">H", chain_id) + struct.pack("<I", height) \
        + bh[::-1] + mh


def decode(payload):
    """Decode a raw 74-byte payload into its fields (hashes in display hex)."""
    if len(payload) != PAYLOAD_LEN:
        raise ValueError(f"bad payload length {len(payload)} (want {PAYLOAD_LEN})")
    if payload[:4] != MAGIC:
        raise ValueError(f"bad magic {payload[:4]!r}")
    return {
        "chain_id": struct.unpack(">H", payload[4:6])[0],
        "height": struct.unpack("<I", payload[6:10])[0],
        "block_hash": payload[10:42][::-1].hex(),  # internal -> RPC display order
        "muhash": payload[42:74].hex(),
    }


def decode_scriptpubkey(script_hex):
    """Parse an OP_RETURN scriptPubKey hex and decode the MRG1 payload."""
    s = bytes.fromhex(script_hex)
    if len(s) < 2 or s[0] != OP_RETURN:
        raise ValueError("not an OP_RETURN scriptPubKey")
    op = s[1]
    if 1 <= op <= 75:            # direct push of op bytes
        n, off = op, 2
    elif op == OP_PUSHDATA1:
        if len(s) < 3:
            raise ValueError("truncated OP_PUSHDATA1")
        n, off = s[2], 3
    else:
        raise ValueError(f"unsupported push opcode 0x{op:02x}")
    if len(s) != off + n:
        raise ValueError("push length does not match script length")
    return decode(s[off:])

demo.sh

#!/usr/bin/env bash
# End-to-end demo: two regtest nodes, one anchor, one SPV verification.
set -euo pipefail
cd "$(dirname "$0")"
ROOT="$PWD"
BINDIR="$(cd "$ROOT"/core/bitcoin-*/bin && pwd)"
BITCOIND="$BINDIR/bitcoind"
CLI="$BINDIR/bitcoin-cli"
PY=python3
DEPTH=6

MD="$ROOT/data/member"
XD="$ROOT/data/meta"
M_RPC=18443; M_P2P=18444   # member chain
X_RPC=19443; X_P2P=19444   # meta chain

MCLI=("$CLI" -regtest -datadir="$MD" -rpcport=$M_RPC)
XCLI=("$CLI" -regtest -datadir="$XD" -rpcport=$X_RPC)

cleanup() {
  "${MCLI[@]}" stop >/dev/null 2>&1 || true
  "${XCLI[@]}" stop >/dev/null 2>&1 || true
}
trap cleanup EXIT

# $1... = full cli command line (without method)
wait_rpc() {
  for _ in $(seq 1 60); do
    if "$@" getblockcount >/dev/null 2>&1; then return 0; fi
    sleep 0.5
  done
  echo "node failed to start: $*" >&2
  exit 1
}

echo "== fresh datadirs"
rm -rf "$ROOT/data" "$ROOT/anchors.json"
mkdir -p "$MD" "$XD"

echo "== starting member node (rpc $M_RPC) and meta node (rpc $X_RPC)"
"$BITCOIND" -regtest -daemon -datadir="$MD" -rpcport=$M_RPC -port=$M_P2P \
  -txindex=1 -coinstatsindex=1 -fallbackfee=0.0001 -server
"$BITCOIND" -regtest -daemon -datadir="$XD" -rpcport=$X_RPC -port=$X_P2P \
  -txindex=1 -fallbackfee=0.0001 -server
wait_rpc "${MCLI[@]}"
wait_rpc "${XCLI[@]}"

echo "== creating wallets"
"${MCLI[@]}" createwallet miner >/dev/null
"${XCLI[@]}" createwallet notary >/dev/null

echo "== mining 110 blocks on each chain"
MADDR=$("${MCLI[@]}" -rpcwallet=miner getnewaddress)
XADDR=$("${XCLI[@]}" -rpcwallet=notary getnewaddress)
"${MCLI[@]}" generatetoaddress 110 "$MADDR" >/dev/null
"${XCLI[@]}" generatetoaddress 110 "$XADDR" >/dev/null

echo "== waiting for member coinstatsindex to sync"
for _ in $(seq 1 60); do
  if "${MCLI[@]}" getindexinfo 2>/dev/null | jq -e '.coinstatsindex.synced' >/dev/null 2>&1; then
    break
  fi
  sleep 0.5
done

echo "== sending 1.0 BTC on the member chain"
TO=$("${MCLI[@]}" -rpcwallet=miner getnewaddress)
TXID=$("${MCLI[@]}" -rpcwallet=miner sendtoaddress "$TO" 1.0)
echo "   txid: $TXID"
"${MCLI[@]}" generatetoaddress 1 "$MADDR" >/dev/null  # block containing the tx
"${MCLI[@]}" generatetoaddress 6 "$MADDR" >/dev/null  # bury it $DEPTH deep

echo "== waiting for member coinstatsindex to catch up"
for _ in $(seq 1 60); do
  if "${MCLI[@]}" getindexinfo 2>/dev/null | jq -e '.coinstatsindex.synced' >/dev/null 2>&1; then
    break
  fi
  sleep 0.5
done

echo "== notary: single anchor pass (target = member tip - $DEPTH)"
"$PY" notary.py --member-datadir "$MD" --member-rpcport $M_RPC \
  --meta-datadir "$XD" --meta-rpcport $X_RPC --chain-id 1 \
  --depth $DEPTH --once --index "$ROOT/anchors.json"

echo "== mining 3 blocks on the meta chain to confirm the anchor"
"${XCLI[@]}" generatetoaddress 3 "$XADDR" >/dev/null

echo "== verifying merged-history membership"
if "$PY" verify.py --member-datadir "$MD" --member-rpcport $M_RPC \
    --meta-datadir "$XD" --meta-rpcport $X_RPC --chain-id 1 \
    --txid "$TXID" --index "$ROOT/anchors.json"; then
  echo "############################################"
  echo "# DEMO PASS"
  echo "############################################"
else
  echo "############################################"
  echo "# DEMO FAIL"
  echo "############################################"
  exit 1
fi

get-core.sh

#!/usr/bin/env bash
# Download Bitcoin Core (macOS build) into ./core/ (idempotent, no system installs).
# Picks the latest stable release listed on https://bitcoincore.org/bin/ and
# verifies the tarball against the release's SHA256SUMS when available.
set -euo pipefail
cd "$(dirname "$0")"
ROOT="$PWD"
CORE="$ROOT/core"
mkdir -p "$CORE"

case "$(uname -m)" in
  arm64)  PLAT="arm64-apple-darwin" ;;
  x86_64) PLAT="x86_64-apple-darwin" ;;
  *) echo "unsupported arch: $(uname -m)" >&2; exit 1 ;;
esac

echo "== discovering latest stable release on bitcoincore.org"
LISTING=$(curl -fsSL https://bitcoincore.org/bin/)
VER=$(printf '%s' "$LISTING" \
  | grep -oE 'bitcoin-core-[0-9]+(\.[0-9]+)+' \
  | sed 's/^bitcoin-core-//' \
  | sort -u \
  | python3 -c "import sys; vs=[l.strip() for l in sys.stdin if l.strip()]; print(max(vs, key=lambda v: [int(x) for x in v.split('.')]))")
[ -n "$VER" ] || { echo "could not determine latest version" >&2; exit 1; }
echo "   latest: $VER"

TARBALL="bitcoin-${VER}-${PLAT}.tar.gz"
BASE="https://bitcoincore.org/bin/bitcoin-core-${VER}"
BIN="$CORE/bitcoin-${VER}/bin/bitcoind"

if [ -x "$BIN" ]; then
  echo "== already installed: $BIN"
  exit 0
fi

if [ ! -f "$CORE/$TARBALL" ]; then
  echo "== downloading $TARBALL"
  curl -fSL --retry 3 -o "$CORE/$TARBALL" "$BASE/$TARBALL"
fi

if curl -fsSL -o "$CORE/SHA256SUMS" "$BASE/SHA256SUMS"; then
  echo "== verifying SHA256"
  (cd "$CORE" && grep -F " $TARBALL" SHA256SUMS | shasum -a 256 -c -)
else
  echo "!! SHA256SUMS unavailable, skipping hash verification" >&2
fi

echo "== extracting"
tar -xzf "$CORE/$TARBALL" -C "$CORE"
# curl downloads normally carry no quarantine attr, but strip it just in case
xattr -dr com.apple.quarantine "$CORE/bitcoin-${VER}" 2>/dev/null || true

"$BIN" --version | head -1
echo "== installed: $BIN"

mrgrpc.py

"""Minimal stdlib JSON-RPC client for Bitcoin Core with cookie auth.

The cookie lives at <datadir>/regtest/.cookie and contains "user:password".
It is re-read on every call (it is tiny, and Core may rotate it between runs).
"""
import base64
import http.client
import json
import pathlib


class RPCError(Exception):
    def __init__(self, code, message):
        super().__init__(f"RPC error {code}: {message}")
        self.code = code
        self.message = message


class Node:
    def __init__(self, datadir, rpcport, host="127.0.0.1", wallet=None, timeout=120):
        self.datadir = pathlib.Path(datadir)
        self.host = host
        self.port = int(rpcport)
        self.wallet = wallet
        self.timeout = timeout

    def _auth_header(self):
        cookie = (self.datadir / "regtest" / ".cookie").read_text().strip()
        return "Basic " + base64.b64encode(cookie.encode()).decode()

    def with_wallet(self, name):
        return Node(self.datadir, self.port, self.host, name, self.timeout)

    def single_wallet(self):
        """Return a client bound to the node's only loaded wallet."""
        wallets = self.call("listwallets")
        if len(wallets) != 1:
            raise RPCError(None, f"expected exactly 1 loaded wallet, got {wallets}")
        return self.with_wallet(wallets[0])

    def call(self, method, *params):
        path = f"/wallet/{self.wallet}" if self.wallet else "/"
        body = json.dumps({
            "jsonrpc": "1.0", "id": "mrg", "method": method, "params": list(params),
        })
        conn = http.client.HTTPConnection(self.host, self.port, timeout=self.timeout)
        try:
            conn.request("POST", path, body=body,
                         headers={"Authorization": self._auth_header(),
                                  "Content-Type": "application/json"})
            resp = conn.getresponse()
            raw = resp.read()
            status = resp.status
        finally:
            conn.close()
        if status != 200:
            raise RPCError(status, raw[:300].decode("utf-8", "replace"))
        reply = json.loads(raw)
        if reply.get("error") is not None:
            err = reply["error"]
            raise RPCError(err.get("code"), err.get("message"))
        return reply["result"]

notary.py

#!/usr/bin/env python3
"""Notary daemon: anchors member-chain state into the meta chain.

Each anchor is an OP_RETURN transaction on the META chain embedding an MRG1
payload: (chain_id, member height, member block hash, member UTXO muhash).
Anchored heights are recorded in a JSON index so passes are idempotent.
"""
import argparse
import json
import os
import sys
import time

import anchor
from mrgrpc import Node, RPCError

POLL_SECONDS = 5


def load_index(path):
    if os.path.exists(path):
        with open(path) as f:
            return json.load(f)
    return []


def save_index(path, recs):
    tmp = path + ".tmp"
    with open(tmp, "w") as f:
        json.dump(recs, f, indent=2)
        f.write("\n")
    os.replace(tmp, path)


def utxo_commitment(member, height):
    """UTXO-set commitment, preferably scoped to `height`.

    Height-scoped queries need Core >= 28 with -coinstatsindex; otherwise we
    fall back to the commitment at the member tip. If the modern "muhash"
    hash type is rejected, fall back to legacy "hash_serialized_3".
    Returns (hex_value, description_of_what_was_committed).
    """
    attempts = [
        (["muhash", height], "muhash", f"muhash@{height}"),
        (["muhash"], "muhash", "muhash@tip"),
        (["hash_serialized_3", height], "hash_serialized_3", f"hash_serialized_3@{height}"),
        (["hash_serialized_3"], "hash_serialized_3", "hash_serialized_3@tip"),
    ]
    last = None
    for args, field, desc in attempts:
        try:
            r = member.call("gettxoutsetinfo", *args)
            return r[field], desc
        except RPCError as e:
            last = e
    raise last


def anchor_height(member, meta, metaw, chain_id, h, index_path):
    """Create one anchor on the meta chain for member height h. Returns record."""
    block_hash = member.call("getblockhash", h)
    muhash, scope = utxo_commitment(member, h)
    payload = anchor.encode(chain_id, h, block_hash, muhash)
    assert len(payload) == anchor.PAYLOAD_LEN

    raw = meta.call("createrawtransaction", [], [{"data": payload.hex()}])
    funded = metaw.call("fundrawtransaction", raw)
    signed = metaw.call("signrawtransactionwithwallet", funded["hex"])
    if not signed.get("complete"):
        raise RuntimeError("meta wallet could not sign anchor tx")
    txid = meta.call("sendrawtransaction", signed["hex"])

    rec = {
        "chain_id": chain_id,
        "height": h,
        "block_hash": block_hash,
        "muhash": muhash,
        "anchor_txid": txid,
    }
    recs = load_index(index_path)
    recs.append(rec)
    save_index(index_path, recs)
    print(f"anchored member height {h}: block {block_hash[:16]}… "
          f"{scope}={muhash[:16]}… anchor_txid={txid}", flush=True)
    return rec


def one_pass(args, member, meta, metaw):
    recs = load_index(args.index)
    indexed = {(r["chain_id"], r["height"]) for r in recs}
    tip = member.call("getblockcount")
    tmax = args.at if args.at is not None else tip - args.depth
    if tmax < 0:
        print(f"member tip {tip} below depth {args.depth}; nothing to do", flush=True)
        return
    if args.once:
        # exactly one height: the target rounded down to a multiple of --every
        h = (tmax // args.every) * args.every
        heights = [h] if (args.chain_id, h) not in indexed else []
    else:
        heights = [h for h in range(args.every, tmax + 1, args.every)
                   if (args.chain_id, h) not in indexed]
    if not heights:
        print(f"nothing new to anchor (tip={tip}, target<={tmax})", flush=True)
        return
    for h in heights:
        anchor_height(member, meta, metaw, args.chain_id, h, args.index)


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--member-datadir", required=True)
    ap.add_argument("--member-rpcport", type=int, required=True)
    ap.add_argument("--meta-datadir", required=True)
    ap.add_argument("--meta-rpcport", type=int, required=True)
    ap.add_argument("--chain-id", type=int, required=True)
    ap.add_argument("--every", type=int, default=1,
                        help="only anchor multiples of this height (default 1)")
    ap.add_argument("--depth", type=int, default=6,
                        help="anchor at member tip minus this depth (default 6)")
    ap.add_argument("--once", action="store_true",
                    help="single pass: anchor exactly one target height")
    ap.add_argument("--at", type=int, default=None,
                    help="override target height (default: member tip - depth)")
    ap.add_argument("--index", default="anchors.json")
    args = ap.parse_args()

    member = Node(args.member_datadir, args.member_rpcport)
    meta = Node(args.meta_datadir, args.meta_rpcport)
    metaw = meta.single_wallet()

    if args.once:
        one_pass(args, member, meta, metaw)
        return
    print(f"notary daemon: chain_id={args.chain_id} every={args.every} "
          f"depth={args.depth} poll={POLL_SECONDS}s", flush=True)
    while True:
        try:
            one_pass(args, member, meta, metaw)
        except (RPCError, OSError) as e:
            print(f"pass failed: {e}", file=sys.stderr, flush=True)
        time.sleep(POLL_SECONDS)


if __name__ == "__main__":
    main()

realverify.py

#!/usr/bin/env python3
"""Verify a REAL mainnet Bitcoin transaction with pure SPV math.

No local node: headers and Merkle proofs are fetched from the public
Esplora API (blockstream.info), then verified independently with spv.py:
real proof-of-work, real Merkle branches, real header linkage.

spv.py conventions: check_pow/verify_header_link RAISE on failure (return
None on success); header fields prevhash/merkleroot are internal-order bytes.

Usage: python3 realverify.py [--txid TX] [--api https://blockstream.info/api]
"""
import argparse
import json
import sys
import urllib.request

import spv  # dsha256, parse_header, check_pow, verify_header_link, header_hash_hex

PIZZA_TX = "a1075db55d416d3ca199f55b6084e2115b9345e16c5cf302fc80e9d5fbf5d48d"  # block 57043


def get(api, path, raw=False):
    with urllib.request.urlopen(api + path, timeout=30) as r:
        body = r.read()
    return body if raw else json.loads(body)


def merkle_branch_root(txid_hex, branch, pos):
    """Esplora proof format: sibling list + leaf position. Hashes are
    display (big-endian) hex; internal hashing uses reversed bytes.
    Returns the computed root in internal byte order."""
    h = bytes.fromhex(txid_hex)[::-1]
    idx = pos
    for sib_hex in branch:
        sib = bytes.fromhex(sib_hex)[::-1]
        h = spv.dsha256(h + sib) if idx % 2 == 0 else spv.dsha256(sib + h)
        idx //= 2
    return h


def header_at(api, height):
    """Fetch (block_hash_display_hex, parsed_header) for a height."""
    bhash = get(api, f"/block-height/{height}", raw=True).decode().strip()
    raw = bytes.fromhex(get(api, f"/block/{bhash}/header", raw=True).decode().strip())
    return bhash, spv.parse_header(raw)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--txid", default=PIZZA_TX)
    ap.add_argument("--api", default="https://blockstream.info/api")
    ap.add_argument("--walk", type=int, default=5, help="recent headers to PoW/link-check")
    a = ap.parse_args()
    ok = True

    tip = int(get(a.api, "/blocks/tip/height", raw=True))
    print(f"== real mainnet tip: {tip}")

    print(f"== 1. locate tx {a.txid[:16]}… on mainnet")
    proof = get(a.api, f"/tx/{a.txid}/merkle-proof")
    height = proof["block_height"]
    print(f"   included in block {height}, merkle branch of {len(proof['merkle'])} sibling(s), pos {proof['pos']}")

    print("== 2. fetch the REAL block header and check the Merkle branch")
    bhash, header = header_at(a.api, height)
    root = merkle_branch_root(a.txid, proof["merkle"], proof["pos"])
    if root == header["merkleroot"]:
        print(f"   merkle branch ok: tx -> root {root[::-1].hex()[:16]}… == header.merkleroot (block {bhash[:16]}…)")
    else:
        print("   FAIL: merkle root mismatch"); ok = False

    print("== 3. PoW of that block (real difficulty of its era)")
    try:
        spv.check_pow(header)
        print(f"   pow ok: hash {spv.header_hash_hex(header)[:24]}… <= target(bits={header['bits']:08x})")
    except ValueError as e:
        print(f"   FAIL: {e}"); ok = False

    print(f"== 4. walk the last {a.walk} REAL headers: PoW + prevhash linkage")
    child = None
    for i, h in enumerate(range(tip, tip - a.walk, -1)):
        bh, hdr = header_at(a.api, h)
        try:
            spv.check_pow(hdr)
            if child is not None:
                spv.verify_header_link(child, hdr)  # child.prevhash == hash(hdr)
        except ValueError as e:
            print(f"   FAIL at {h}: {e}"); ok = False
        child = hdr
    if ok:
        print(f"   {a.walk} headers verified: every one meets real mainnet difficulty, chain links intact")

    print("PASS (real mainnet data, independently verified)" if ok else "FAIL")
    sys.exit(0 if ok else 1)


if __name__ == "__main__":
    main()

spv.py

"""Pure-Python SPV primitives: headers, proof-of-work, partial merkle trees.

All hashes are handled in Bitcoin *internal* byte order (the natural byte
order of dsha256 digests, i.e. little-endian uint256 display convention);
display hex (= reversed) is only produced at API boundaries.
"""
import hashlib
import struct

HEADER_LEN = 80


def dsha256(b):
    return hashlib.sha256(hashlib.sha256(b).digest()).digest()


# ---------------------------------------------------------------- block header

def parse_header(b):
    if len(b) != HEADER_LEN:
        raise ValueError(f"header must be {HEADER_LEN} bytes, got {len(b)}")
    version, = struct.unpack("<i", b[0:4])
    time_, bits, nonce = struct.unpack("<III", b[68:80])
    return {
        "version": version,
        "prevhash": b[4:36],      # internal byte order
        "merkleroot": b[36:68],   # internal byte order
        "time": time_,
        "bits": bits,
        "nonce": nonce,
    }


def serialize_header(h):
    return struct.pack("<i", h["version"]) + h["prevhash"] + h["merkleroot"] \
        + struct.pack("<III", h["time"], h["bits"], h["nonce"])


def header_hash(h):
    """Block hash in internal byte order."""
    return dsha256(serialize_header(h))


def header_hash_hex(h):
    """Block hash as RPC/display hex."""
    return header_hash(h)[::-1].hex()


# ---------------------------------------------------------------- proof of work

def bits_to_target(nbits):
    """Compact nBits -> 256-bit target integer."""
    size = nbits >> 24
    word = nbits & 0x007FFFFF
    if nbits & 0x00800000:
        raise ValueError("negative target in nBits")
    if word == 0:
        raise ValueError("zero target in nBits")
    if size <= 3:
        return word >> (8 * (3 - size))
    return word << (8 * (size - 3))


def check_pow(h):
    """Raise unless uint256(dsha256(header)) <= target(nBits)."""
    target = bits_to_target(h["bits"])
    if target >= 1 << 256:
        raise ValueError("target out of range")
    val = int.from_bytes(header_hash(h), "little")
    if val > target:
        raise ValueError(
            f"insufficient proof of work: hash {val:064x} > target {target:064x}")


def verify_header_link(child, parent):
    """Raise unless child.prevhash == dsha256(parent header)."""
    if child["prevhash"] != header_hash(parent):
        raise ValueError("header does not link to its parent (prevhash mismatch)")


# ------------------------------------------------------- partial merkle trees
#
# Serialized CMerkleBlock (as returned by `gettxoutproof`):
#   80B header | uint32 nTx | varint nHashes | nHashes*32B |
#   varint nFlagBytes | flag bytes
#
# Verification re-runs Bitcoin Core's recursive traversal
# (CPartialMerkleTree::ExtractMatches):
#   * tree width at height h is (nTx + (1<<h) - 1) >> h
#   * flag bits are consumed LSB-first within each byte
#   * a node consumes one hash iff height==0 or flag==0
#   * a leaf with flag==1 is a matched txid
#   * internal nodes hash dsha256(left||right); when the right child is
#     outside the tree width, the left child is duplicated
#   * left == right at an internal node is rejected (CVE-2012-2459)

class _Reader:
    def __init__(self, b):
        self.b = b
        self.i = 0

    def read(self, n):
        r = self.b[self.i:self.i + n]
        if len(r) != n:
            raise ValueError("truncated merkle block")
        self.i += n
        return r

    def varint(self):
        ch = self.read(1)[0]
        if ch < 0xFD:
            return ch
        if ch == 0xFD:
            return struct.unpack("<H", self.read(2))[0]
        if ch == 0xFE:
            return struct.unpack("<I", self.read(4))[0]
        return struct.unpack("<Q", self.read(8))[0]


def parse_merkle_block(blob):
    r = _Reader(blob)
    header = parse_header(r.read(HEADER_LEN))
    ntx, = struct.unpack("<I", r.read(4))
    hashes = [r.read(32) for _ in range(r.varint())]
    flags = r.read(r.varint())
    return header, ntx, hashes, flags


def extract_matches(ntx, hashes, flags):
    """Core's traversal. Returns (merkle_root, matched txids as display hex)."""
    if ntx == 0:
        raise ValueError("merkle block claims zero transactions")
    def width(h):
        return (ntx + (1 << h) - 1) >> h

    height = 0
    while width(height) > 1:
        height += 1

    hash_idx = 0
    bit_idx = 0
    matches = []

    def read_bit():
        nonlocal bit_idx
        if bit_idx >= len(flags) * 8:
            raise ValueError("ran out of flag bits")
        bit = (flags[bit_idx >> 3] >> (bit_idx & 7)) & 1
        bit_idx += 1
        return bit

    def read_hash():
        nonlocal hash_idx
        if hash_idx >= len(hashes):
            raise ValueError("ran out of hashes")
        h = hashes[hash_idx]
        hash_idx += 1
        return h

    def traverse(h, pos):
        flag = read_bit()
        if h == 0 or flag == 0:
            node = read_hash()
            if h == 0 and flag == 1:
                matches.append(node)
            return node
        left = traverse(h - 1, pos * 2)
        if pos * 2 + 1 < width(h - 1):
            right = traverse(h - 1, pos * 2 + 1)
            if right == left:
                raise ValueError("invalid merkle branch: duplicated hash "
                                 "(CVE-2012-2459 mutation)")
        else:
            right = left  # odd row: duplicate the left child
        return dsha256(left + right)

    root = traverse(height, 0)
    if hash_idx != len(hashes):
        raise ValueError("proof contains unused hashes")
    if (bit_idx + 7) // 8 != len(flags):
        raise ValueError("proof contains unused flag bytes")
    return root, [m[::-1].hex() for m in matches]


def verify_merkle_proof(blob, expect_txid=None):
    """Full check of a gettxoutproof blob: PoW, merkle root, membership."""
    header, ntx, hashes, flags = parse_merkle_block(blob)
    check_pow(header)
    root, matches = extract_matches(ntx, hashes, flags)
    if root != header["merkleroot"]:
        raise ValueError("computed merkle root != header merkleroot")
    if expect_txid is not None and expect_txid.lower() not in matches:
        raise ValueError(f"txid {expect_txid} not among matched leaves")
    return {
        "header": header,
        "block_hash": header_hash_hex(header),
        "merkle_root": root,
        "matches": matches,
        "ntx": ntx,
    }

verify.py

#!/usr/bin/env python3
"""SPV verifier: prove a member-chain tx is part of the merged history.

Chain of proof (each step printed; any failure exits non-zero):
  a. locate the tx on the member chain
  b. merkle proof: tx ∈ its member block, block has valid PoW
  c. header walk: PoW + prevhash linkage from the tx block up to an
     anchored member block recorded in the local anchor index
  d. the anchor tx on the meta chain commits (via OP_RETURN payload) to
     exactly that walked-to member block and chain_id
  e. merkle proof: the anchor tx ∈ a confirmed meta-chain block
"""
import argparse
import json
import os
import sys

import anchor
import spv
from mrgrpc import Node, RPCError


def fail(msg):
    print(f"FAIL: {msg}")
    sys.exit(1)


def step(msg):
    print(f"\n== {msg}")


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--member-datadir", required=True)
    ap.add_argument("--member-rpcport", type=int, required=True)
    ap.add_argument("--meta-datadir", required=True)
    ap.add_argument("--meta-rpcport", type=int, required=True)
    ap.add_argument("--chain-id", type=int, required=True)
    ap.add_argument("--txid", required=True)
    ap.add_argument("--index", default="anchors.json")
    args = ap.parse_args()

    member = Node(args.member_datadir, args.member_rpcport)
    meta = Node(args.meta_datadir, args.meta_rpcport)
    txid = args.txid.lower()

    # a. locate the tx on the member chain
    step(f"a. locate tx {txid} on member chain")
    try:
        tx = member.call("getrawtransaction", txid, True)
    except RPCError as e:
        fail(f"tx not found on member chain: {e}")
    if "blockhash" not in tx:
        fail("tx is not confirmed in any member block")
    tx_block = tx["blockhash"]
    tx_height = member.call("getblockheader", tx_block, True)["height"]
    print(f"   confirmed in member block {tx_block} (height {tx_height})")

    # b. merkle proof of inclusion in that block (+ PoW of the block header)
    step("b. SPV: gettxoutproof on member (merkle branch + PoW)")
    try:
        proof_hex = member.call("gettxoutproof", [txid], tx_block)
    except RPCError as e:
        fail(f"gettxoutproof failed: {e}")
    try:
        res = spv.verify_merkle_proof(bytes.fromhex(proof_hex), txid)
    except ValueError as e:
        fail(f"merkle proof invalid: {e}")
    if res["block_hash"] != tx_block:
        fail(f"proof header hash {res['block_hash']} != expected block {tx_block}")
    print(f"   merkle branch ok: tx is one of {len(res['matches'])} matched leaf(s) "
          f"in a block of {res['ntx']} tx(s); PoW ok (bits={res['header']['bits']:08x})")

    # c. header walk from the tx block up to an anchored height
    step("c. member header walk: tx block -> anchored block (PoW + linkage)")
    if not os.path.exists(args.index):
        fail(f"anchor index {args.index} not found")
    with open(args.index) as f:
        index = json.load(f)
    cands = sorted((r for r in index
                    if r["chain_id"] == args.chain_id and r["height"] >= tx_height),
                   key=lambda r: r["height"])
    if not cands:
        fail(f"no anchor for chain_id {args.chain_id} at height >= {tx_height}")
    rec = cands[0]
    print(f"   using anchor at member height {rec['height']} (txid {rec['anchor_txid']})")

    cur = res["header"]  # tx block header, PoW already checked in step b
    for h in range(tx_height + 1, rec["height"] + 1):
        hh = member.call("getblockhash", h)
        hdr = spv.parse_header(bytes.fromhex(member.call("getblockheader", hh, False)))
        try:
            spv.check_pow(hdr)
            spv.verify_header_link(hdr, cur)
        except ValueError as e:
            fail(f"header chain broken at height {h}: {e}")
        cur = hdr
    walked_hash = spv.header_hash_hex(cur)
    if walked_hash != rec["block_hash"]:
        fail(f"walked-to block {walked_hash} != index block_hash {rec['block_hash']}")
    n_links = rec["height"] - tx_height
    print(f"   {n_links} header link(s) verified; reached member block {walked_hash}")

    # d. decode the anchor tx's OP_RETURN payload on the meta chain
    step("d. meta chain: decode anchor OP_RETURN payload")
    try:
        atx = meta.call("getrawtransaction", rec["anchor_txid"], True)
    except RPCError as e:
        fail(f"anchor tx not found on meta chain: {e}")
    if "blockhash" not in atx:
        fail("anchor tx is not confirmed on the meta chain")
    nulldata = [v for v in atx["vout"]
                if v["scriptPubKey"].get("type") == "nulldata"
                or v["scriptPubKey"].get("asm", "").startswith("OP_RETURN")]
    if not nulldata:
        fail("anchor tx has no OP_RETURN output")
    try:
        dec = anchor.decode_scriptpubkey(nulldata[0]["scriptPubKey"]["hex"])
    except ValueError as e:
        fail(f"anchor payload undecodable: {e}")
    if dec["chain_id"] != args.chain_id:
        fail(f"payload chain_id {dec['chain_id']} != {args.chain_id}")
    if dec["height"] != rec["height"]:
        fail(f"payload height {dec['height']} != index height {rec['height']}")
    if dec["block_hash"] != walked_hash:
        fail(f"payload block_hash {dec['block_hash']} != walked-to block {walked_hash}")
    print(f"   payload ok: chain_id={dec['chain_id']} height={dec['height']} "
          f"block={dec['block_hash'][:16]}… muhash={dec['muhash'][:16]}…")

    # e. merkle proof that the anchor tx is in a confirmed meta block
    step("e. SPV: gettxoutproof on meta for the anchor tx")
    anchor_block = atx["blockhash"]
    try:
        aproof_hex = meta.call("gettxoutproof", [rec["anchor_txid"]], anchor_block)
    except RPCError as e:
        fail(f"gettxoutproof on meta failed: {e}")
    try:
        ares = spv.verify_merkle_proof(bytes.fromhex(aproof_hex), rec["anchor_txid"])
    except ValueError as e:
        fail(f"anchor merkle proof invalid: {e}")
    if ares["block_hash"] != anchor_block:
        fail("anchor proof header != anchor block")
    confs = atx.get("confirmations", 0)
    if confs < 1:
        fail("anchor tx has no confirmations on the meta chain")
    print(f"   anchor tx included in meta block {anchor_block} "
          f"({confs} confirmation(s)); PoW ok")

    meta_height = meta.call("getblockheader", anchor_block, True)["height"]

    # f. summary
    step("f. PASS")
    print(f"   tx {txid}")
    print(f"   member height        : {tx_height} (block {tx_block})")
    print(f"   anchored at member   : {rec['height']} (block {rec['block_hash']})")
    print(f"   anchor meta-height   : {meta_height} (txid {rec['anchor_txid']})")
    print(f"   member utxo muhash   : {dec['muhash']}")
    print("PASS")
    sys.exit(0)


if __name__ == "__main__":
    main()