Merge #18877: Serve cfcheckpt requests

23083856a5 [test] Add test for cfcheckpt (Jim Posen)
f9e00bb25a [net processing] Message handling for getcfcheckpt. (Jim Posen)
9ccaaba11e [init] Add -peerblockfilters option (Jim Posen)

Pull request description:

  Serve cfcheckpt messages if basic block filter index is enabled and `-peercfilters` is set.

  `NODE_COMPACT_FILTERS` is not signaled to peers, but functionality can be used for testing and serving pre-configured clients.

ACKs for top commit:
  jonatack:
    Code review re-ACK 23083856a5 the only change since my review @ 967e2b1 is an update required for #16224 that was merged yesterday.
  fjahr:
    re-ACK 23083856a5
  jkczyz:
    re-ACK 23083856a5
  ariard:
    re-Code Review ACK 2308385
  clarkmoody:
    Tested ACK 23083856a
  MarcoFalke:
    re-ACK 23083856a5 🌳
  theStack:
    ACK 23083856a5

Tree-SHA512: 8c751bbd7d1c31a413096462ae025c3d2f3163c7016cbec472a5f5ec267f8dd19a2dfc4d749876d7409c1db546e6fdd16461c6863effcfa0d3e993edcfa92a08

src/init.cpp

@@ -446,6 +446,7 @@ void SetupServerArgs(NodeContext& node)
     gArgs.AddArg("-onion=<ip:port>", "Use separate SOCKS5 proxy to reach peers via Tor hidden services, set -noonion to disable (default: -proxy)", ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     gArgs.AddArg("-onlynet=<net>", "Make outgoing connections only through network <net> (ipv4, ipv6 or onion). Incoming connections are not affected by this option. This option can be specified multiple times to allow multiple networks.", ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     gArgs.AddArg("-peerbloomfilters", strprintf("Support filtering of blocks and transaction with bloom filters (default: %u)", DEFAULT_PEERBLOOMFILTERS), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
+    gArgs.AddArg("-peerblockfilters", strprintf("Serve compact block filters to peers per BIP 157 (default: %u)", DEFAULT_PEERBLOCKFILTERS), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     gArgs.AddArg("-permitbaremultisig", strprintf("Relay non-P2SH multisig (default: %u)", DEFAULT_PERMIT_BAREMULTISIG), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     gArgs.AddArg("-port=<port>", strprintf("Listen for connections on <port> (default: %u, testnet: %u, regtest: %u)", defaultChainParams->GetDefaultPort(), testnetChainParams->GetDefaultPort(), regtestChainParams->GetDefaultPort()), ArgsManager::ALLOW_ANY | ArgsManager::NETWORK_ONLY, OptionsCategory::CONNECTION);
     gArgs.AddArg("-proxy=<ip:port>", "Connect through SOCKS5 proxy, set -noproxy to disable (default: disabled)", ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
@@ -993,6 +994,13 @@ bool AppInitParameterInteraction()
         }
     }
 
+    // Basic filters are the only supported filters. The basic filters index must be enabled
+    // to serve compact filters
+    if (gArgs.GetBoolArg("-peerblockfilters", DEFAULT_PEERBLOCKFILTERS) &&
+        g_enabled_filter_types.count(BlockFilterType::BASIC) != 1) {
+        return InitError(_("Cannot set -peerblockfilters without -blockfilterindex."));
+    }
+
     // if using block pruning, then disallow txindex
     if (gArgs.GetArg("-prune", 0)) {
         if (gArgs.GetBoolArg("-txindex", DEFAULT_TXINDEX))

src/net_processing.cpp

@@ -8,9 +8,11 @@
 #include <addrman.h>
 #include <banman.h>
 #include <blockencodings.h>
+#include <blockfilter.h>
 #include <chainparams.h>
 #include <consensus/validation.h>
 #include <hash.h>
+#include <index/blockfilterindex.h>
 #include <validation.h>
 #include <merkleblock.h>
 #include <netmessagemaker.h>
@@ -127,6 +129,8 @@ static constexpr unsigned int INVENTORY_BROADCAST_MAX = 7 * INVENTORY_BROADCAST_
 static constexpr unsigned int AVG_FEEFILTER_BROADCAST_INTERVAL = 10 * 60;
 /** Maximum feefilter broadcast delay after significant change. */
 static constexpr unsigned int MAX_FEEFILTER_CHANGE_DELAY = 5 * 60;
+/** Interval between compact filter checkpoints. See BIP 157. */
+static constexpr int CFCHECKPT_INTERVAL = 1000;
 
 struct COrphanTx {
     // When modifying, adapt the copy of this definition in tests/DoS_tests.
@@ -1969,6 +1973,107 @@ void static ProcessOrphanTx(CConnman* connman, CTxMemPool& mempool, std::set<uin
     }
 }
 
+/**
+ * Validation logic for compact filters request handling.
+ *
+ * May disconnect from the peer in the case of a bad request.
+ *
+ * @param[in]   pfrom           The peer that we received the request from
+ * @param[in]   chain_params    Chain parameters
+ * @param[in]   filter_type     The filter type the request is for. Must be basic filters.
+ * @param[in]   stop_hash       The stop_hash for the request
+ * @param[out]  stop_index      The CBlockIndex for the stop_hash block, if the request can be serviced.
+ * @param[out]  filter_index    The filter index, if the request can be serviced.
+ * @return                      True if the request can be serviced.
+ */
+static bool PrepareBlockFilterRequest(CNode* pfrom, const CChainParams& chain_params,
+                                      BlockFilterType filter_type,
+                                      const uint256& stop_hash,
+                                      const CBlockIndex*& stop_index,
+                                      const BlockFilterIndex*& filter_index)
+{
+    const bool supported_filter_type =
+        (filter_type == BlockFilterType::BASIC &&
+         gArgs.GetBoolArg("-peerblockfilters", DEFAULT_PEERBLOCKFILTERS));
+    if (!supported_filter_type) {
+        LogPrint(BCLog::NET, "peer %d requested unsupported block filter type: %d\n",
+                 pfrom->GetId(), static_cast<uint8_t>(filter_type));
+        pfrom->fDisconnect = true;
+        return false;
+    }
+
+    {
+        LOCK(cs_main);
+        stop_index = LookupBlockIndex(stop_hash);
+
+        // Check that the stop block exists and the peer would be allowed to fetch it.
+        if (!stop_index || !BlockRequestAllowed(stop_index, chain_params.GetConsensus())) {
+            LogPrint(BCLog::NET, "peer %d requested invalid block hash: %s\n",
+                     pfrom->GetId(), stop_hash.ToString());
+            pfrom->fDisconnect = true;
+            return false;
+        }
+    }
+
+    filter_index = GetBlockFilterIndex(filter_type);
+    if (!filter_index) {
+        LogPrint(BCLog::NET, "Filter index for supported type %s not found\n", BlockFilterTypeName(filter_type));
+        return false;
+    }
+
+    return true;
+}
+
+/**
+ * Handle a getcfcheckpt request.
+ *
+ * May disconnect from the peer in the case of a bad request.
+ *
+ * @param[in]   pfrom           The peer that we received the request from
+ * @param[in]   vRecv           The raw message received
+ * @param[in]   chain_params    Chain parameters
+ * @param[in]   connman         Pointer to the connection manager
+ */
+static void ProcessGetCFCheckPt(CNode* pfrom, CDataStream& vRecv, const CChainParams& chain_params,
+                                CConnman* connman)
+{
+    uint8_t filter_type_ser;
+    uint256 stop_hash;
+
+    vRecv >> filter_type_ser >> stop_hash;
+
+    const BlockFilterType filter_type = static_cast<BlockFilterType>(filter_type_ser);
+
+    const CBlockIndex* stop_index;
+    const BlockFilterIndex* filter_index;
+    if (!PrepareBlockFilterRequest(pfrom, chain_params, filter_type, stop_hash,
+                                   stop_index, filter_index)) {
+        return;
+    }
+
+    std::vector<uint256> headers(stop_index->nHeight / CFCHECKPT_INTERVAL);
+
+    // Populate headers.
+    const CBlockIndex* block_index = stop_index;
+    for (int i = headers.size() - 1; i >= 0; i--) {
+        int height = (i + 1) * CFCHECKPT_INTERVAL;
+        block_index = block_index->GetAncestor(height);
+
+        if (!filter_index->LookupFilterHeader(block_index, headers[i])) {
+            LogPrint(BCLog::NET, "Failed to find block filter header in index: filter_type=%s, block_hash=%s\n",
+                         BlockFilterTypeName(filter_type), block_index->GetBlockHash().ToString());
+            return;
+        }
+    }
+
+    CSerializedNetMsg msg = CNetMsgMaker(pfrom->GetSendVersion())
+        .Make(NetMsgType::CFCHECKPT,
+              filter_type_ser,
+              stop_index->GetBlockHash(),
+              headers);
+    connman->PushMessage(pfrom, std::move(msg));
+}
+
 bool ProcessMessage(CNode* pfrom, const std::string& msg_type, CDataStream& vRecv, int64_t nTimeReceived, const CChainParams& chainparams, CTxMemPool& mempool, CConnman* connman, BanMan* banman, const std::atomic<bool>& interruptMsgProc)
 {
     LogPrint(BCLog::NET, "received: %s (%u bytes) peer=%d\n", SanitizeString(msg_type), vRecv.size(), pfrom->GetId());
@@ -3274,6 +3379,11 @@ bool ProcessMessage(CNode* pfrom, const std::string& msg_type, CDataStream& vRec
         return true;
     }
 
+    if (msg_type == NetMsgType::GETCFCHECKPT) {
+        ProcessGetCFCheckPt(pfrom, vRecv, chainparams, connman);
+        return true;
+    }
+
     if (msg_type == NetMsgType::NOTFOUND) {
         // Remove the NOTFOUND transactions from the peer
         LOCK(cs_main);

src/net_processing.h

@@ -21,6 +21,7 @@ static const unsigned int DEFAULT_MAX_ORPHAN_TRANSACTIONS = 100;
 /** Default number of orphan+recently-replaced txn to keep around for block reconstruction */
 static const unsigned int DEFAULT_BLOCK_RECONSTRUCTION_EXTRA_TXN = 100;
 static const bool DEFAULT_PEERBLOOMFILTERS = false;
+static const bool DEFAULT_PEERBLOCKFILTERS = false;
 
 class PeerLogicValidation final : public CValidationInterface, public NetEventsInterface {
 private:

src/protocol.cpp

@@ -40,6 +40,8 @@ const char *SENDCMPCT="sendcmpct";
 const char *CMPCTBLOCK="cmpctblock";
 const char *GETBLOCKTXN="getblocktxn";
 const char *BLOCKTXN="blocktxn";
+const char *GETCFCHECKPT="getcfcheckpt";
+const char *CFCHECKPT="cfcheckpt";
 } // namespace NetMsgType
 
 /** All known message types. Keep this in the same order as the list of
@@ -71,6 +73,8 @@ const static std::string allNetMessageTypes[] = {
     NetMsgType::CMPCTBLOCK,
     NetMsgType::GETBLOCKTXN,
     NetMsgType::BLOCKTXN,
+    NetMsgType::GETCFCHECKPT,
+    NetMsgType::CFCHECKPT,
 };
 const static std::vector<std::string> allNetMessageTypesVec(allNetMessageTypes, allNetMessageTypes+ARRAYLEN(allNetMessageTypes));
 

src/protocol.h

@@ -234,6 +234,20 @@ extern const char *GETBLOCKTXN;
  * @since protocol version 70014 as described by BIP 152
  */
 extern const char *BLOCKTXN;
+/**
+ * getcfcheckpt requests evenly spaced compact filter headers, enabling
+ * parallelized download and validation of the headers between them.
+ * Only available with service bit NODE_COMPACT_FILTERS as described by
+ * BIP 157 & 158.
+ */
+extern const char *GETCFCHECKPT;
+/**
+ * cfcheckpt is a response to a getcfcheckpt request containing a vector of
+ * evenly spaced filter headers for blocks on the requested chain.
+ * Only available with service bit NODE_COMPACT_FILTERS as described by
+ * BIP 157 & 158.
+ */
+extern const char *CFCHECKPT;
 };
 
 /* Get a vector of all valid message types (see above) */

test/functional/p2p_blockfilters.py

@@ -0,0 +1,134 @@
+#!/usr/bin/env python3
+# Copyright (c) 2019 The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Tests NODE_COMPACT_FILTERS (BIP 157/158).
+
+Tests that a node configured with -blockfilterindex and -peerblockfilters can serve
+cfcheckpts.
+"""
+
+from test_framework.messages import (
+    FILTER_TYPE_BASIC,
+    msg_getcfcheckpt,
+)
+from test_framework.mininode import P2PInterface
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import (
+    assert_equal,
+    connect_nodes,
+    disconnect_nodes,
+    wait_until,
+)
+
+class CompactFiltersTest(BitcoinTestFramework):
+    def set_test_params(self):
+        self.setup_clean_chain = True
+        self.rpc_timeout = 480
+        self.num_nodes = 2
+        self.extra_args = [
+            ["-blockfilterindex", "-peerblockfilters"],
+            ["-blockfilterindex"],
+        ]
+
+    def run_test(self):
+        # Node 0 supports COMPACT_FILTERS, node 1 does not.
+        node0 = self.nodes[0].add_p2p_connection(P2PInterface())
+        node1 = self.nodes[1].add_p2p_connection(P2PInterface())
+
+        # Nodes 0 & 1 share the same first 999 blocks in the chain.
+        self.nodes[0].generate(999)
+        self.sync_blocks(timeout=600)
+
+        # Stale blocks by disconnecting nodes 0 & 1, mining, then reconnecting
+        disconnect_nodes(self.nodes[0], 1)
+
+        self.nodes[0].generate(1)
+        wait_until(lambda: self.nodes[0].getblockcount() == 1000)
+        stale_block_hash = self.nodes[0].getblockhash(1000)
+
+        self.nodes[1].generate(1001)
+        wait_until(lambda: self.nodes[1].getblockcount() == 2000)
+
+        self.log.info("get cfcheckpt on chain to be re-orged out.")
+        request = msg_getcfcheckpt(
+            filter_type=FILTER_TYPE_BASIC,
+            stop_hash=int(stale_block_hash, 16)
+        )
+        node0.send_and_ping(message=request)
+        response = node0.last_message['cfcheckpt']
+        assert_equal(response.filter_type, request.filter_type)
+        assert_equal(response.stop_hash, request.stop_hash)
+        assert_equal(len(response.headers), 1)
+
+        self.log.info("Reorg node 0 to a new chain.")
+        connect_nodes(self.nodes[0], 1)
+        self.sync_blocks(timeout=600)
+
+        main_block_hash = self.nodes[0].getblockhash(1000)
+        assert main_block_hash != stale_block_hash, "node 0 chain did not reorganize"
+
+        self.log.info("Check that peers can fetch cfcheckpt on active chain.")
+        tip_hash = self.nodes[0].getbestblockhash()
+        request = msg_getcfcheckpt(
+            filter_type=FILTER_TYPE_BASIC,
+            stop_hash=int(tip_hash, 16)
+        )
+        node0.send_and_ping(request)
+        response = node0.last_message['cfcheckpt']
+        assert_equal(response.filter_type, request.filter_type)
+        assert_equal(response.stop_hash, request.stop_hash)
+
+        main_cfcheckpt = self.nodes[0].getblockfilter(main_block_hash, 'basic')['header']
+        tip_cfcheckpt = self.nodes[0].getblockfilter(tip_hash, 'basic')['header']
+        assert_equal(
+            response.headers,
+            [int(header, 16) for header in (main_cfcheckpt, tip_cfcheckpt)]
+        )
+
+        self.log.info("Check that peers can fetch cfcheckpt on stale chain.")
+        request = msg_getcfcheckpt(
+            filter_type=FILTER_TYPE_BASIC,
+            stop_hash=int(stale_block_hash, 16)
+        )
+        node0.send_and_ping(request)
+        response = node0.last_message['cfcheckpt']
+
+        stale_cfcheckpt = self.nodes[0].getblockfilter(stale_block_hash, 'basic')['header']
+        assert_equal(
+            response.headers,
+            [int(header, 16) for header in (stale_cfcheckpt,)]
+        )
+
+        self.log.info("Requests to node 1 without NODE_COMPACT_FILTERS results in disconnection.")
+        requests = [
+            msg_getcfcheckpt(
+                filter_type=FILTER_TYPE_BASIC,
+                stop_hash=int(main_block_hash, 16)
+            ),
+        ]
+        for request in requests:
+            node1 = self.nodes[1].add_p2p_connection(P2PInterface())
+            node1.send_message(request)
+            node1.wait_for_disconnect()
+
+        self.log.info("Check that invalid requests result in disconnection.")
+        requests = [
+            # Requesting unknown filter type results in disconnection.
+            msg_getcfcheckpt(
+                filter_type=255,
+                stop_hash=int(main_block_hash, 16)
+            ),
+            # Requesting unknown hash results in disconnection.
+            msg_getcfcheckpt(
+                filter_type=FILTER_TYPE_BASIC,
+                stop_hash=123456789,
+            ),
+        ]
+        for request in requests:
+            node0 = self.nodes[0].add_p2p_connection(P2PInterface())
+            node0.send_message(request)
+            node0.wait_for_disconnect()
+
+if __name__ == '__main__':
+    CompactFiltersTest().main()

test/functional/test_framework/messages.py

@@ -57,6 +57,8 @@ MSG_FILTERED_BLOCK = 3
 MSG_WITNESS_FLAG = 1 << 30
 MSG_TYPE_MASK = 0xffffffff >> 2
 
+FILTER_TYPE_BASIC = 0
+
 # Serialization/deserialization tools
 def sha256(s):
     return hashlib.new('sha256', s).digest()
@@ -1512,3 +1514,50 @@ class msg_no_witness_blocktxn(msg_blocktxn):
 
     def serialize(self):
         return self.block_transactions.serialize(with_witness=False)
+
+class msg_getcfcheckpt:
+    __slots__ = ("filter_type", "stop_hash")
+    msgtype =  b"getcfcheckpt"
+
+    def __init__(self, filter_type, stop_hash):
+        self.filter_type = filter_type
+        self.stop_hash = stop_hash
+
+    def deserialize(self, f):
+        self.filter_type = struct.unpack("<B", f.read(1))[0]
+        self.stop_hash = deser_uint256(f)
+
+    def serialize(self):
+        r = b""
+        r += struct.pack("<B", self.filter_type)
+        r += ser_uint256(self.stop_hash)
+        return r
+
+    def __repr__(self):
+        return "msg_getcfcheckpt(filter_type={:#x}, stop_hash={:x})".format(
+            self.filter_type, self.stop_hash)
+
+class msg_cfcheckpt:
+    __slots__ = ("filter_type", "stop_hash", "headers")
+    msgtype =  b"cfcheckpt"
+
+    def __init__(self, filter_type=None, stop_hash=None, headers=None):
+        self.filter_type = filter_type
+        self.stop_hash = stop_hash
+        self.headers = headers
+
+    def deserialize(self, f):
+        self.filter_type = struct.unpack("<B", f.read(1))[0]
+        self.stop_hash = deser_uint256(f)
+        self.headers = deser_uint256_vector(f)
+
+    def serialize(self):
+        r = b""
+        r += struct.pack("<B", self.filter_type)
+        r += ser_uint256(self.stop_hash)
+        r += ser_uint256_vector(self.headers)
+        return r
+
+    def __repr__(self):
+        return "msg_cfcheckpt(filter_type={:#x}, stop_hash={:x})".format(
+            self.filter_type, self.stop_hash)

test/functional/test_framework/mininode.py

@@ -31,6 +31,7 @@ from test_framework.messages import (
     msg_block,
     MSG_BLOCK,
     msg_blocktxn,
+    msg_cfcheckpt,
     msg_cmpctblock,
     msg_feefilter,
     msg_filteradd,
@@ -67,6 +68,7 @@ MESSAGEMAP = {
     b"addr": msg_addr,
     b"block": msg_block,
     b"blocktxn": msg_blocktxn,
+    b"cfcheckpt": msg_cfcheckpt,
     b"cmpctblock": msg_cmpctblock,
     b"feefilter": msg_feefilter,
     b"filteradd": msg_filteradd,
@@ -328,6 +330,7 @@ class P2PInterface(P2PConnection):
     def on_addr(self, message): pass
     def on_block(self, message): pass
     def on_blocktxn(self, message): pass
+    def on_cfcheckpt(self, message): pass
     def on_cmpctblock(self, message): pass
     def on_feefilter(self, message): pass
     def on_filteradd(self, message): pass

test/functional/test_runner.py

@@ -226,6 +226,7 @@ BASE_SCRIPTS = [
     'feature_loadblock.py',
     'p2p_dos_header_tree.py',
     'p2p_unrequested_blocks.py',
+    'p2p_blockfilters.py',
     'feature_includeconf.py',
     'feature_asmap.py',
     'mempool_unbroadcast.py',