git / moonbite-network / spv.py · raw
"""MyCoin — Milestone 7: Simplified Payment Verification (whitepaper section 8).An SPV (Simplified Payment Verification) client is a *light* node. Rather thandownloading and validating every transaction in every block, it keeps only theblock *headers* — the most-work chain of proof-of-work commitments. To confirmthat a specific payment happened, it asks a full node for a Merkle *branch*linking that transaction's id to the `merkle_root` recorded in a block header(section 8). If the branch recomputes to the header's root, the transaction isprovably included in that block, and the number of headers chained on top of it(confirmations) measures how much work would have to be redone to reverse it.How the pieces fit together: * `merkle.verify_merkle_proof` checks inclusion of one txid under one root. * The header chain (`SPVClient`) supplies the trusted root for a block hash and the confirmation depth. * A full node uses `build_proof_bundle` to package (txid, block_hash, branch) for the client; the client never sees the other transactions.Trade-off vs. a full node: An SPV client trusts that the most-work header chain was produced by an honest majority of hashing power. It *cannot* independently detect an invalid block (e.g. one that mints coins out of thin air or contains a double-spend) because it never validates the transactions inside blocks — it only checks Merkle inclusion and proof-of-work depth. As the whitepaper notes, this verification is reliable "as long as honest nodes control the network," but can be fooled by an attacker who can sustain more proof-of-work than the honest network. To mitigate this, real SPV clients rely on the honest majority and on network *alerts* from full nodes that detect an invalid block, prompting the light client to download the full block and confirm the inconsistency. That safety net, and the inability to self-validate, is the price paid for storing only headers instead of the whole chain."""from __future__ import annotationsimport timefrom dataclasses import dataclass, fieldfrom block import BlockHeader, Block, genesis_blockfrom merkle import merkle_proof, verify_merkle_prooffrom params import MAX_FUTURE_TIME, MEDIAN_TIME_SPAN, RETARGET_INTERVALfrom pow import calculate_next_bits, hash_meets_target@dataclassclass MerkleProofBundle: """What a full node hands an SPV client to prove a single payment. Contains just enough to check one transaction's inclusion in one block: the transaction id, the hash of the block it claims membership in, and the Merkle branch (list of (sibling_hash_hex, side) pairs) linking the txid to that block's Merkle root. """ txid: str block_hash: str proof: list[tuple[str, str]] = field(default_factory=list) def to_dict(self) -> dict: return { "txid": self.txid, "block_hash": self.block_hash, # tuples are serialized as 2-element lists for JSON transport "proof": [[sibling, side] for sibling, side in self.proof], } @staticmethod def from_dict(d: dict) -> "MerkleProofBundle": return MerkleProofBundle( txid=d["txid"], block_hash=d["block_hash"], proof=[(sibling, side) for sibling, side in d["proof"]], )class SPVClient: """A light client that stores only block headers (whitepaper section 8). The chain is seeded with the genesis header and grown by `add_header`, which accepts a header only if it links to the current tip. Payments are checked against the header's Merkle root; confirmation depth is read off the chain. """ def __init__(self) -> None: # Headers only — never full blocks or transactions. self._headers: list[BlockHeader] = [genesis_block().header] # hash -> index, for O(1) lookup by block hash. self._index_by_hash: dict[str, int] = {self._headers[0].hash(): 0} def _expected_bits(self, height: int) -> int: """Difficulty `bits` a header at `height` must carry, from stored headers. Mirrors the full node (node._expected_bits): difficulty is unchanged except on a retarget boundary, where it is recomputed from the actual timespan of the preceding interval. The client holds every header, so it can derive this independently rather than trusting the header's claim. """ prev = self._headers[height - 1] if height % RETARGET_INTERVAL != 0: return prev.bits first = self._headers[height - RETARGET_INTERVAL] actual = prev.timestamp - first.timestamp return calculate_next_bits(prev.bits, actual) def _median_time_past(self, index: int) -> int: """Median timestamp of the last MEDIAN_TIME_SPAN headers up to `index`.""" lo = max(0, index - MEDIAN_TIME_SPAN + 1) stamps = sorted(h.timestamp for h in self._headers[lo : index + 1]) return stamps[len(stamps) // 2] def add_header(self, header: BlockHeader) -> bool: """Append `header` iff it links to the tip AND carries real work. Security: an SPV client's whole guarantee is that the header chain it follows embodies proof-of-work no attacker can cheaply forge. So beyond linking to the current tip, the header must independently satisfy the same work rules the full node enforces (node._validate_block): * its `bits` equals the difficulty the retarget schedule demands at this height (a forger cannot pick a trivial target), and * its hash actually meets that target (the work was really done), and * its timestamp is neither in the future nor at/below the median of the preceding window (no timewarp to cheapen the next retarget). Without these checks a malicious node could fabricate an unlimited chain of unmined headers for free and forge any confirmation depth. A header failing any rule is rejected and the stored chain is left unchanged. """ if header.prev_hash != self.tip_hash: return False height = len(self._headers) # height this header would occupy if header.bits != self._expected_bits(height): return False if not hash_meets_target(header.hash(), header.bits): return False if header.timestamp > int(time.time()) + MAX_FUTURE_TIME: return False if header.timestamp <= self._median_time_past(height - 1): return False self._headers.append(header) self._index_by_hash[header.hash()] = len(self._headers) - 1 return True @property def height(self) -> int: """Tip height; the genesis header is height 0.""" return len(self._headers) - 1 @property def tip_hash(self) -> str: return self._headers[-1].hash() def header_by_hash(self, h: str) -> BlockHeader | None: idx = self._index_by_hash.get(h) return None if idx is None else self._headers[idx] def confirmations(self, block_hash: str) -> int: """Headers from `block_hash` to the tip inclusive (tip itself = 1). Returns 0 if the block hash is not on the stored chain. """ idx = self._index_by_hash.get(block_hash) if idx is None: return 0 return len(self._headers) - idx def verify_payment(self, bundle: MerkleProofBundle) -> bool: """Confirm the bundle's txid is included in its claimed block. Looks up the trusted header for `bundle.block_hash`; if the client has never seen that block, verification fails. Otherwise the Merkle branch must recompute to the header's committed Merkle root. """ header = self.header_by_hash(bundle.block_hash) if header is None: return False return verify_merkle_proof(bundle.txid, bundle.proof, header.merkle_root) def verify_payment_with_confirmations( self, bundle: MerkleProofBundle, min_confirmations: int ) -> bool: """Inclusion proof AND at least `min_confirmations` headers deep.""" if not self.verify_payment(bundle): return False return self.confirmations(bundle.block_hash) >= min_confirmationsdef build_proof_bundle(block: Block, txid: str) -> MerkleProofBundle: """Full-node helper: package the Merkle branch proving `txid` is in `block`. Raises ValueError if `txid` is not one of the block's transactions. """ txids = [tx.txid for tx in block.transactions] try: index = txids.index(txid) except ValueError: raise ValueError(f"txid {txid} is not in block {block.hash}") proof = merkle_proof(txids, index) return MerkleProofBundle(txid=txid, block_hash=block.hash, proof=proof)