Confirm transactions by merkle proof and reinstate coins only on a quorum (B-83, TX-06, TX-22, TX-23, ELC-25, ELC-26, DEP-02, DEP-08, AUD-18, ADR-07)

poll_once marked a transaction confirmed on one server's `height > 0`, and
reinstate_reappeared_utxos restored a coin written off as spent outside the platform on
the active server's word alone, so a hostile active server could report a
never-relayed payout as confirmed (winner unpaid, round closed) or put a spent coin
back into a balance.

- electrum/merkle.py: pure merkle-branch verification against a block header.
- ElectrumListener.confirmation_proven(txid, height): the *other* servers are asked for
  blockchain.transaction.get_merkle; a strict majority of at least the quorum minimum
  must return a proof that recomputes to the merkle root of that block's header (block
  height and position checked too). The header is the tip the listener holds, or for an
  older block the one a majority of the quorum agrees on and that meets its own target;
  a height above our tip cannot be proven. Quorum rules are _corroborate_majority's.
- poll_once takes the listener as `network`: the active server's history only nominates
  a candidate (txid, height); the row confirms only when the proof holds, and stays
  pending otherwise. With no network nothing confirms. ConfirmationPoller wires it.
- The reconciler's late-confirmation check of failed rows (B-78) now puts each claimed
  (txid, height) to the same proof instead of counting histories.
- refresh_user splits reinstatement like a credit: find_reinstate_candidates (read-only;
  same outpoint, recorded amount, confirmed) -> corroborate_utxo_reinstatement (same
  quorum as a credit) -> reinstate_reappeared_utxos writes only what was corroborated,
  rechecking the mark under the lock. Nothing is committed in the read phase any more.
- Broadcast fan-out, the third part of the finding, shipped with B-76.

Tests changed because the spec changed: every poll_once call now passes a network that
proves (FakeNetwork) — the confirmation tests are otherwise unchanged; the failed-row
"quorum does not confirm" test now asserts the claim goes to the proof.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

app/deposits/service.py

@@ -103,12 +103,38 @@ async def credit_confirmed_utxos(session: AsyncSession, user_id: int, entries: l
 _EXTERNAL_SPEND_SENTINEL = "external-spend"
 
 
+async def find_reinstate_candidates(session: AsyncSession, user_id: int, entries: list[dict]) -> list[dict]:
+    """Outpoints flagged as spent outside the platform that this refresh's `listunspent`
+    lists again, confirmed and for the amount we recorded — the *candidates* for
+    reinstatement (B-83, DEP-02, DEP-08). Read-only: like a new credit (B-59) and a
+    candidate external spend (B-29), a reinstatement moves a balance, so the caller
+    (electrum/listener.py:refresh_user) must have the other servers corroborate each
+    candidate before reinstate_reappeared_utxos writes it. An entry whose amount differs
+    from our record is not a candidate: it is not the coin we wrote off."""
+    flagged = (
+        await session.scalars(
+            select(UtxoEvent).where(UtxoEvent.user_id == user_id, UtxoEvent.spent_txid == _EXTERNAL_SPEND_SENTINEL)
+        )
+    ).all()
+    amounts = {(row.txid, row.vout): row.amount_sats for row in flagged}
+    return [
+        entry
+        for entry in entries
+        if amounts.get((entry["tx_hash"], entry["tx_pos"])) == entry.get("value") and (entry.get("height") or 0) > 0
+    ]
+
+
 async def reinstate_reappeared_utxos(session: AsyncSession, user_id: int, entries: list[dict]) -> int:
     """The reverse of a mark applied by find_utxos_missing_from/
     mark_utxos_spent_externally (B-29): if an outpoint we'd previously flagged as
     spent outside the platform reappears as unspent in a later listunspent, undo
     the mark instead of leaving it permanent no matter what the chain says
-    afterwards. Cheap and purely DB-side — always safe to run on every refresh.
+    afterwards.
+
+    Persistence only, for entries the caller has **already corroborated** against the
+    other servers (B-83, ELC-25) — it used to run on the active server's word alone, so a
+    hostile one could put a spent coin back into a balance. Re-checks, under the write
+    lock, that each row still carries the mark before clearing it (DEP-08).
     """
     current_keys = {(entry["tx_hash"], entry["tx_pos"]) for entry in entries}
 

app/electrum/listener.py

@@ -11,12 +11,14 @@ from app.db.models import User
 from app.deposits.service import (
     credit_confirmed_utxos,
     find_new_credit_candidates,
+    find_reinstate_candidates,
     find_utxos_missing_from,
     mark_utxos_spent_externally,
     reinstate_reappeared_utxos,
 )
 from app.config import MIN_QUORUM_RESPONSES
 from app.electrum.client import ElectrumClient, ElectrumEndpoint, ElectrumRpcError
+from app.electrum.merkle import proof_matches_header
 from app.electrum.quorum import ServerAnswers
 from app.electrum.scripthash import address_to_scripthash
 from app.rounds.draw import (
@@ -646,6 +648,26 @@ class ElectrumListener:
         only credited once the quorum agrees.
         """
 
+        return await self._corroborate_unspent_confirmed(
+            scripthash, txid, vout, value, f"credit of {value} sats at {txid}:{vout}"
+        )
+
+    async def corroborate_utxo_reinstatement(self, scripthash: str, txid: str, vout: int, value: int) -> bool:
+        """B-83 (ELC-25, DEP-08): the same quorum as a credit, for an outpoint that was
+        written off as spent outside the platform and has reappeared among the unspent.
+        Balances move in both directions and both need the same agreement: reinstating
+        on the active server's word alone let a hostile one put a spent coin back into a
+        balance."""
+        return await self._corroborate_unspent_confirmed(
+            scripthash, txid, vout, value, f"reinstatement of {value} sats at {txid}:{vout}"
+        )
+
+    async def _corroborate_unspent_confirmed(
+        self, scripthash: str, txid: str, vout: int, value: int, description: str
+    ) -> bool:
+        """Do the other servers list this outpoint as unspent, for the same amount, and
+        confirmed (height > 0)? Shared by credits (B-59) and reinstatements (B-83)."""
+
         async def _ask(endpoint: ElectrumEndpoint) -> bool | None:
             client = self._client_factory(endpoint)
             try:
@@ -665,9 +687,56 @@ class ElectrumListener:
             finally:
                 await client.close()
 
-        return await self._corroborate_majority(
-            _ask, lambda agrees: agrees, f"credit of {value} sats at {txid}:{vout}"
-        )
+        return await self._corroborate_majority(_ask, lambda agrees: agrees, description)
+
+    # --- Confirmations proven by merkle branch (B-83, TX-22, ELC-26) ---------------------
+
+    async def _header_at(self, height: int) -> str | None:
+        """The header of block `height` that a merkle proof is checked against: the tip
+        we hold (validated as it arrived), or for an older block the one a strict
+        majority of at least the quorum minimum of the other servers agree on, which must
+        also satisfy its own difficulty target. None when neither is available — in
+        particular for a height above our tip, which we have no way to check."""
+        if height <= 0 or height > self.tip_height:
+            return None
+        if height == self.tip_height and self.tip_header_hex:
+            return self.tip_header_hex
+
+        async def _ask(client: ElectrumClient) -> str | None:
+            header = await client.get_block_header(height)
+            return header if isinstance(header, str) and header_meets_its_own_target(header) else None
+
+        answers = [a for a in await self._ask_other_servers(_ask) if a is not None]
+        if len(answers) < max(self._minimum_other_answers(), 1):
+            return None
+        header, votes = Counter(answers).most_common(1)[0]
+        return header if votes * 2 > len(answers) else None
+
+    async def confirmation_proven(self, txid: str, height: int) -> bool:
+        """Is `txid` in the block at `height`? Accepted only when the quorum returns a
+        valid merkle proof — `blockchain.transaction.get_merkle` from the *other* servers
+        (never the active one, which is what a hostile peer controls) that recomputes to
+        the merkle root of that block's header, which we hold or the quorum agrees on. A
+        strict majority of at least the quorum minimum must prove it; otherwise the answer
+        is False and the caller simply asks again at its next pass (a delay, never a
+        loss)."""
+        header = await self._header_at(height)
+        if header is None:
+            logger.warning("cannot check the merkle proof of %s at height %s: no trusted header for it", txid, height)
+            return False
+
+        async def _ask(endpoint: ElectrumEndpoint) -> bool | None:
+            client = self._client_factory(endpoint)
+            try:
+                await asyncio.wait_for(client.connect(), timeout=_CORROBORATION_TIMEOUT_SECONDS)
+                proof = await asyncio.wait_for(client.get_merkle(txid, height), timeout=_CORROBORATION_TIMEOUT_SECONDS)
+                return proof_matches_header(txid, height, proof, header)
+            except Exception:
+                return None
+            finally:
+                await client.close()
+
+        return await self._corroborate_majority(_ask, lambda proven: proven, f"merkle proof of {txid} at height {height}")
 
     async def _consume_headers(self, queue: asyncio.Queue) -> None:
         while True:
@@ -697,7 +766,8 @@ class ElectrumListener:
 
         async with self._session_factory() as session:
             credit_candidates = await find_new_credit_candidates(session, user_id, entries)
-            reinstated = await reinstate_reappeared_utxos(session, user_id, entries)
+            # B-83 (DEP-02): candidates only — nothing is written before the quorum agrees.
+            reinstate_candidates = await find_reinstate_candidates(session, user_id, entries)
             candidates = [
                 (row.id, row.txid, row.vout)
                 for row in await find_utxos_missing_from(session, user_id, entries)
@@ -711,6 +781,27 @@ class ElectrumListener:
             )
         ]
 
+        # B-83 (ELC-25, DEP-08): a coin written off as spent outside the platform comes back
+        # into the balance only on the same quorum as a credit.
+        corroborated_reinstatements = [
+            entry
+            for entry in reinstate_candidates
+            if await self.corroborate_utxo_reinstatement(
+                scripthash, entry["tx_hash"], entry["tx_pos"], entry["value"]
+            )
+        ]
+        reinstated = 0
+        if corroborated_reinstatements:
+            async with self._session_factory() as session:
+                reinstated = await reinstate_reappeared_utxos(session, user_id, corroborated_reinstatements)
+        if len(corroborated_reinstatements) < len(reinstate_candidates):
+            logger.warning(
+                "%s reappeared UTXO(s) for user_id=%s not corroborated by the other servers — "
+                "left out of the balance, will retry on the next refresh",
+                len(reinstate_candidates) - len(corroborated_reinstatements),
+                user_id,
+            )
+
         credited = 0
         if corroborated_credits:
             async with self._session_factory() as session:

app/electrum/merkle.py

@@ -0,0 +1,57 @@
+"""Merkle proofs (B-83, TX-22, ELC-26).
+
+`blockchain.transaction.get_merkle` answers with the branch that links a transaction to
+its block's merkle root. Checking it against a header **we hold** is what turns "a server
+says this tx is at height h" into something a hostile server cannot fake without mining.
+"""
+
+import hashlib
+
+_HEADER_LENGTH_BYTES = 80
+_MERKLE_ROOT_OFFSET = 36
+_MERKLE_ROOT_LENGTH = 32
+_MAX_BRANCH_LENGTH = 40  # 2^40 transactions in a block is already absurd
+
+
+def _sha256d(data: bytes) -> bytes:
+    return hashlib.sha256(hashlib.sha256(data).digest()).digest()
+
+
+def header_merkle_root(header_hex: str) -> bytes:
+    """The merkle root committed to by a raw 80-byte header, in internal byte order."""
+    header = bytes.fromhex(header_hex)
+    if len(header) != _HEADER_LENGTH_BYTES:
+        raise ValueError(f"a block header is {_HEADER_LENGTH_BYTES} bytes, got {len(header)}")
+    return header[_MERKLE_ROOT_OFFSET : _MERKLE_ROOT_OFFSET + _MERKLE_ROOT_LENGTH]
+
+
+def merkle_root_from_branch(txid: str, branch: list[str], position: int) -> bytes:
+    """The merkle root (internal byte order) a transaction `txid` at index `position` of
+    its block reaches through `branch` — the sibling hashes, bottom-up, in the display
+    order Electrum uses for txids."""
+    if position < 0 or len(branch) > _MAX_BRANCH_LENGTH or position >> len(branch):
+        raise ValueError("merkle position does not fit the branch")
+    current = bytes.fromhex(txid)[::-1]
+    index = position
+    for sibling_hex in branch:
+        sibling = bytes.fromhex(sibling_hex)[::-1]
+        if len(sibling) != 32:
+            raise ValueError("merkle branch element is not 32 bytes")
+        current = _sha256d(sibling + current) if index & 1 else _sha256d(current + sibling)
+        index >>= 1
+    return current
+
+
+def proof_matches_header(txid: str, height: int, proof: object, header_hex: str) -> bool:
+    """Does `proof` — a `get_merkle` reply — put `txid` in the block at `height` whose
+    header is `header_hex`? Any malformed reply is simply "no"."""
+    try:
+        if not isinstance(proof, dict) or int(proof["block_height"]) != height:
+            return False
+        branch = proof["merkle"]
+        if not isinstance(branch, list):
+            return False
+        root = merkle_root_from_branch(txid, [str(item) for item in branch], int(proof["pos"]))
+        return root == header_merkle_root(header_hex)
+    except (KeyError, TypeError, ValueError):
+        return False

app/main.py

@@ -62,7 +62,7 @@ async def lifespan(app: FastAPI):
     app.state.user_locks = UserLocks()
 
     scheduler = RoundScheduler(AsyncSessionLocal, listener)
-    poller = ConfirmationPoller(AsyncSessionLocal, lambda: listener.client)
+    poller = ConfirmationPoller(AsyncSessionLocal, lambda: listener.client, network=listener)
     bumper = RbfBumper(AsyncSessionLocal, lambda: listener.client, network=listener)
     # Resolves in-flight transactions against the chain — the piece that lets the
     # system recover on its own from a broadcast that never confirmed (B-04/B-08).

app/tx/confirmation.py

@@ -1,6 +1,7 @@
 import asyncio
 import logging
 from collections.abc import Awaitable, Callable
+from typing import Protocol
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
@@ -29,7 +30,19 @@ def register_handler(kind: str, handler: ConfirmationHandler) -> None:
     _handlers[kind] = handler
 
 
-async def poll_once(session_factory: async_sessionmaker, client: ElectrumClient) -> int:
+class ConfirmationProver(Protocol):
+    """The part of ElectrumListener a confirmation needs (B-83): the quorum's verdict on
+    a merkle proof of `txid` at `height`."""
+
+    async def confirmation_proven(self, txid: str, height: int) -> bool: ...
+
+
+async def poll_once(session_factory: async_sessionmaker, client: ElectrumClient, network: ConfirmationProver | None) -> int:
+    """One pass over the `pending` rows. The active server's history only nominates a
+    candidate: a row is confirmed when the quorum proves it with a merkle proof
+    (B-83, TX-22). With no `network` to ask nothing can be confirmed — the active
+    server's word alone is exactly what a hostile one would use to declare a payout that
+    was never broadcast confirmed."""
     async with session_factory() as session:
         # Plain columns, not entities: nothing then outlives the session, so this
         # can't break if expire_on_commit is ever turned on (B-21).
@@ -90,6 +103,18 @@ async def poll_once(session_factory: async_sessionmaker, client: ElectrumClient)
         if entry is None:
             continue
 
+        # B-83 (TX-22, ELC-26): the height is a claim of the active server. The quorum has
+        # to return a valid merkle proof of the txid against that block's header; without
+        # one the row stays pending and is looked at again on the next pass.
+        if network is None or not await network.confirmation_proven(entry["tx_hash"], entry["height"]):
+            logger.warning(
+                "pending_transaction %s: %s reported at height %s but the quorum did not prove it — still pending",
+                pending_id,
+                entry["tx_hash"],
+                entry["height"],
+            )
+            continue
+
         async with session_factory() as session:
             await begin_immediate(session)  # B-88: read the row and write its outcome under one lock
             row = await session.get(PendingTransaction, pending_id)
@@ -111,16 +136,22 @@ async def poll_once(session_factory: async_sessionmaker, client: ElectrumClient)
 
 
 class ConfirmationPoller:
-    def __init__(self, session_factory: async_sessionmaker, get_client: Callable[[], ElectrumClient | None]):
+    def __init__(
+        self,
+        session_factory: async_sessionmaker,
+        get_client: Callable[[], ElectrumClient | None],
+        network: ConfirmationProver | None = None,
+    ):
         self._session_factory = session_factory
         self._get_client = get_client
+        self._network = network  # the ElectrumListener, which holds the headers and asks the quorum
 
     async def run(self) -> None:
         while True:
             client = self._get_client()
             if client is not None:
                 try:
-                    await poll_once(self._session_factory, client)
+                    await poll_once(self._session_factory, client, self._network)
                 except asyncio.CancelledError:
                     raise
                 except Exception:

app/tx/reconcile.py

@@ -81,6 +81,9 @@ _ABANDON_AFTER_SECONDS = 6 * 60 * 60
 # for dead can still arrive in a block.
 FAILED_WATCH_DAYS = 14
 
+# How many distinct (txid, height) claims of one failed row are put to the quorum per pass.
+_MAX_RESURRECTION_CLAIMS = 3
+
 # What a node says when the transaction's inputs do not exist or are spent already.
 _MISSING_INPUTS_MARKERS = ("missing-inputs", "missingorspent", "missing inputs", "inputs-missing")
 # What it says when only the fee is the problem: not proof of anything.
@@ -236,23 +239,17 @@ def _entries_for(answers: ServerAnswers, known_txids: set[str]) -> list[tuple[st
     return found
 
 
-def _confirmed_by_the_quorum(answers: ServerAnswers, known_txids: set[str]) -> tuple[str, int] | None:
-    """A txid of the row that a strict majority of the answering *other* servers (at
-    least the quorum minimum of them) list in a block, with the height they report.
-    The active server's word is not part of it."""
-    others = answers.answered_others
-    if len(others) < max(answers.others_needed, 1):
-        return None
+def _confirmed_candidates(answers: ServerAnswers, known_txids: set[str]) -> list[tuple[str, int]]:
+    """(txid, height) of every txid of the row that any answering server lists in a block,
+    the one most servers agree on first. They are only *claims*: each is proven by the
+    quorum's merkle proofs before it is believed (B-83, TX-22)."""
     votes: dict[tuple[str, int], int] = {}
-    for history in others:
+    for history in answers.all:
         for entry in history:
             if entry.get("tx_hash") in known_txids and (entry.get("height") or 0) > 0:
                 key = (entry["tx_hash"], entry["height"])
                 votes[key] = votes.get(key, 0) + 1
-    for key, count in sorted(votes.items(), key=lambda item: -item[1]):
-        if count * 2 > len(others):
-            return key
-    return None
+    return [key for key, _ in sorted(votes.items(), key=lambda item: -item[1])]
 
 
 async def _decide(network, candidate: _Candidate, answers: ServerAnswers):
@@ -260,8 +257,12 @@ async def _decide(network, candidate: _Candidate, answers: ServerAnswers):
     ("abandon", reason), ("resurrected", (txid, height)) or None for "nothing to do or
     not enough to tell". Pure reads — nothing is written here."""
     if candidate.status == "failed":
-        confirmed = _confirmed_by_the_quorum(answers, candidate.known_txids)
-        return ("resurrected", confirmed) if confirmed is not None else None
+        if network is None:
+            return None  # nothing can be proven without the quorum
+        for txid, height in _confirmed_candidates(answers, candidate.known_txids)[:_MAX_RESURRECTION_CLAIMS]:
+            if await network.confirmation_proven(txid, height):
+                return ("resurrected", (txid, height))
+        return None
 
     found = _entries_for(answers, candidate.known_txids)
     if found: