Merge #19010: net processing: Add support for getcfheaders

5308c97 [test] Add test for cfheaders (Jim Posen)
f6b58c1 [net processing] Message handling for getcfheaders. (Jim Posen)
3bdc7c2 [doc] Add comment for m_headers_cache (John Newbery)

Pull request description:

  Support `getcfheaders` requests when `-peerblockfilters` is set.

  Does not advertise compact filter support in version messages.

ACKs for top commit:
  jkczyz:
    ACK 5308c97
  MarcoFalke:
    re-ACK 5308c97 , only change is doc related 🗂
  theStack:
    ACK 5308c97 🚀

Tree-SHA512: 240fc654f6f634c191d9f7628b6c4801f87ed514a1dd55c7de5d454d4012d1c09509a2d5a246bc7da445cd920252b4cd56a493c060cdb207b04af4ffe53b95f7

Author: MarcoFalke

src/index/blockfilterindex.h

@@ -39,6 +39,7 @@ class BlockFilterIndex final : public BaseIndex
     size_t WriteFilterToDisk(FlatFilePos& pos, const BlockFilter& filter);
 
     Mutex m_cs_headers_cache;
+    /** cache of block hash to filter header, to avoid disk access when responding to getcfcheckpt. */
     std::unordered_map<uint256, uint256, FilterHeaderHasher> m_headers_cache GUARDED_BY(m_cs_headers_cache);
 
 protected:

src/net_processing.cpp

@@ -129,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;
+/** Maximum number of cf hashes that may be requested with one getcfheaders. See BIP 157. */
+static constexpr uint32_t MAX_GETCFHEADERS_SIZE = 2000;
 
 struct COrphanTx {
     // When modifying, adapt the copy of this definition in tests/DoS_tests.
@@ -1989,14 +1991,16 @@ void static ProcessOrphanTx(CConnman* connman, CTxMemPool& mempool, std::set<uin
  * @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]   start_height    The start height for the request
  * @param[in]   stop_hash       The stop_hash for the request
+ * @param[in]   max_height_diff The maximum number of items permitted to request, as specified in BIP 157
  * @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,
+                                      BlockFilterType filter_type, uint32_t start_height,
+                                      const uint256& stop_hash, uint32_t max_height_diff,
                                       const CBlockIndex*& stop_index,
                                       BlockFilterIndex*& filter_index)
 {
@@ -2023,6 +2027,21 @@ static bool PrepareBlockFilterRequest(CNode* pfrom, const CChainParams& chain_pa
         }
     }
 
+    uint32_t stop_height = stop_index->nHeight;
+    if (start_height > stop_height) {
+        LogPrint(BCLog::NET, "peer %d sent invalid getcfilters/getcfheaders with " /* Continued */
+                 "start height %d and stop height %d\n",
+                 pfrom->GetId(), start_height, stop_height);
+        pfrom->fDisconnect = true;
+        return false;
+    }
+    if (stop_height - start_height >= max_height_diff) {
+        LogPrint(BCLog::NET, "peer %d requested too many cfilters/cfheaders: %d / %d\n",
+                 pfrom->GetId(), stop_height - start_height + 1, max_height_diff);
+        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));
@@ -2032,6 +2051,61 @@ static bool PrepareBlockFilterRequest(CNode* pfrom, const CChainParams& chain_pa
     return true;
 }
 
+/**
+ * Handle a cfheaders 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 ProcessGetCFHeaders(CNode* pfrom, CDataStream& vRecv, const CChainParams& chain_params,
+                                CConnman* connman)
+{
+    uint8_t filter_type_ser;
+    uint32_t start_height;
+    uint256 stop_hash;
+
+    vRecv >> filter_type_ser >> start_height >> stop_hash;
+
+    const BlockFilterType filter_type = static_cast<BlockFilterType>(filter_type_ser);
+
+    const CBlockIndex* stop_index;
+    BlockFilterIndex* filter_index;
+    if (!PrepareBlockFilterRequest(pfrom, chain_params, filter_type, start_height, stop_hash,
+                                   MAX_GETCFHEADERS_SIZE, stop_index, filter_index)) {
+        return;
+    }
+
+    uint256 prev_header;
+    if (start_height > 0) {
+        const CBlockIndex* const prev_block =
+            stop_index->GetAncestor(static_cast<int>(start_height - 1));
+        if (!filter_index->LookupFilterHeader(prev_block, prev_header)) {
+            LogPrint(BCLog::NET, "Failed to find block filter header in index: filter_type=%s, block_hash=%s\n",
+                         BlockFilterTypeName(filter_type), prev_block->GetBlockHash().ToString());
+            return;
+        }
+    }
+
+    std::vector<uint256> filter_hashes;
+    if (!filter_index->LookupFilterHashRange(start_height, stop_index, filter_hashes)) {
+        LogPrint(BCLog::NET, "Failed to find block filter hashes in index: filter_type=%s, start_height=%d, stop_hash=%s\n",
+                     BlockFilterTypeName(filter_type), start_height, stop_hash.ToString());
+        return;
+    }
+
+    CSerializedNetMsg msg = CNetMsgMaker(pfrom->GetSendVersion())
+        .Make(NetMsgType::CFHEADERS,
+              filter_type_ser,
+              stop_index->GetBlockHash(),
+              prev_header,
+              filter_hashes);
+    connman->PushMessage(pfrom, std::move(msg));
+}
+
 /**
  * Handle a getcfcheckpt request.
  *
@@ -2054,7 +2128,8 @@ static void ProcessGetCFCheckPt(CNode* pfrom, CDataStream& vRecv, const CChainPa
 
     const CBlockIndex* stop_index;
     BlockFilterIndex* filter_index;
-    if (!PrepareBlockFilterRequest(pfrom, chain_params, filter_type, stop_hash,
+    if (!PrepareBlockFilterRequest(pfrom, chain_params, filter_type, /*start_height=*/0, stop_hash,
+                                   /*max_height_diff=*/std::numeric_limits<uint32_t>::max(),
                                    stop_index, filter_index)) {
         return;
     }
@@ -3391,6 +3466,11 @@ bool ProcessMessage(CNode* pfrom, const std::string& msg_type, CDataStream& vRec
         return true;
     }
 
+    if (msg_type == NetMsgType::GETCFHEADERS) {
+        ProcessGetCFHeaders(pfrom, vRecv, chainparams, connman);
+        return true;
+    }
+
     if (msg_type == NetMsgType::GETCFCHECKPT) {
         ProcessGetCFCheckPt(pfrom, vRecv, chainparams, connman);
         return true;

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 *GETCFHEADERS="getcfheaders";
+const char *CFHEADERS="cfheaders";
 const char *GETCFCHECKPT="getcfcheckpt";
 const char *CFCHECKPT="cfcheckpt";
 } // namespace NetMsgType
@@ -73,6 +75,8 @@ const static std::string allNetMessageTypes[] = {
     NetMsgType::CMPCTBLOCK,
     NetMsgType::GETBLOCKTXN,
     NetMsgType::BLOCKTXN,
+    NetMsgType::GETCFHEADERS,
+    NetMsgType::CFHEADERS,
     NetMsgType::GETCFCHECKPT,
     NetMsgType::CFCHECKPT,
 };

src/protocol.h

@@ -225,6 +225,19 @@ extern const char* GETBLOCKTXN;
  * @since protocol version 70014 as described by BIP 152
  */
 extern const char* BLOCKTXN;
+/**
+ * getcfheaders requests a compact filter header and the filter hashes for a
+ * range of blocks, which can then be used to reconstruct the filter headers
+ * for those blocks.
+ * Only available with service bit NODE_COMPACT_FILTERS as described by
+ * BIP 157 & 158.
+ */
+extern const char* GETCFHEADERS;
+/**
+ * cfheaders is a response to a getcfheaders request containing a filter header
+ * and a vector of filter hashes for each subsequent block in the requested range.
+ */
+extern const char* CFHEADERS;
 /**
  * getcfcheckpt requests evenly spaced compact filter headers, enabling
  * parallelized download and validation of the headers between them.
@@ -235,8 +248,6 @@ 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;
 }; // namespace NetMsgType

test/functional/p2p_blockfilters.py

@@ -5,12 +5,16 @@
 """Tests NODE_COMPACT_FILTERS (BIP 157/158).
 
 Tests that a node configured with -blockfilterindex and -peerblockfilters can serve
-cfcheckpts.
+cfheaders and cfcheckpts.
 """
 
 from test_framework.messages import (
     FILTER_TYPE_BASIC,
+    hash256,
     msg_getcfcheckpt,
+    msg_getcfheaders,
+    ser_uint256,
+    uint256_from_str,
 )
 from test_framework.mininode import P2PInterface
 from test_framework.test_framework import BitcoinTestFramework
@@ -100,12 +104,45 @@ def run_test(self):
             [int(header, 16) for header in (stale_cfcheckpt,)]
         )
 
+        self.log.info("Check that peers can fetch cfheaders on active chain.")
+        request = msg_getcfheaders(
+            filter_type=FILTER_TYPE_BASIC,
+            start_height=1,
+            stop_hash=int(main_block_hash, 16)
+        )
+        node0.send_and_ping(request)
+        response = node0.last_message['cfheaders']
+        assert_equal(len(response.hashes), 1000)
+        assert_equal(
+            compute_last_header(response.prev_header, response.hashes),
+            int(main_cfcheckpt, 16)
+        )
+
+        self.log.info("Check that peers can fetch cfheaders on stale chain.")
+        request = msg_getcfheaders(
+            filter_type=FILTER_TYPE_BASIC,
+            start_height=1,
+            stop_hash=int(stale_block_hash, 16)
+        )
+        node0.send_and_ping(request)
+        response = node0.last_message['cfheaders']
+        assert_equal(len(response.hashes), 1000)
+        assert_equal(
+            compute_last_header(response.prev_header, response.hashes),
+            int(stale_cfcheckpt, 16)
+        )
+
         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)
             ),
+            msg_getcfheaders(
+                filter_type=FILTER_TYPE_BASIC,
+                start_height=1000,
+                stop_hash=int(main_block_hash, 16)
+            ),
         ]
         for request in requests:
             node1 = self.nodes[1].add_p2p_connection(P2PInterface())
@@ -114,6 +151,12 @@ def run_test(self):
 
         self.log.info("Check that invalid requests result in disconnection.")
         requests = [
+            # Requesting too many filter headers results in disconnection.
+            msg_getcfheaders(
+                filter_type=FILTER_TYPE_BASIC,
+                start_height=0,
+                stop_hash=int(tip_hash, 16)
+            ),
             # Requesting unknown filter type results in disconnection.
             msg_getcfcheckpt(
                 filter_type=255,
@@ -130,5 +173,12 @@ def run_test(self):
             node0.send_message(request)
             node0.wait_for_disconnect()
 
+def compute_last_header(prev_header, hashes):
+    """Compute the last filter header from a starting header and a sequence of filter hashes."""
+    header = ser_uint256(prev_header)
+    for filter_hash in hashes:
+        header = hash256(ser_uint256(filter_hash) + header)
+    return uint256_from_str(header)
+
 if __name__ == '__main__':
     CompactFiltersTest().main()

test/functional/test_framework/messages.py

@@ -1516,6 +1516,59 @@ class msg_no_witness_blocktxn(msg_blocktxn):
     def serialize(self):
         return self.block_transactions.serialize(with_witness=False)
 
+class msg_getcfheaders:
+    __slots__ = ("filter_type", "start_height", "stop_hash")
+    msgtype =  b"getcfheaders"
+
+    def __init__(self, filter_type, start_height, stop_hash):
+        self.filter_type = filter_type
+        self.start_height = start_height
+        self.stop_hash = stop_hash
+
+    def deserialize(self, f):
+        self.filter_type = struct.unpack("<B", f.read(1))[0]
+        self.start_height = struct.unpack("<I", f.read(4))[0]
+        self.stop_hash = deser_uint256(f)
+
+    def serialize(self):
+        r = b""
+        r += struct.pack("<B", self.filter_type)
+        r += struct.pack("<I", self.start_height)
+        r += ser_uint256(self.stop_hash)
+        return r
+
+    def __repr__(self):
+        return "msg_getcfheaders(filter_type={:#x}, start_height={}, stop_hash={:x})".format(
+            self.filter_type, self.start_height, self.stop_hash)
+
+class msg_cfheaders:
+    __slots__ = ("filter_type", "stop_hash", "prev_header", "hashes")
+    msgtype =  b"cfheaders"
+
+    def __init__(self, filter_type=None, stop_hash=None, prev_header=None, hashes=None):
+        self.filter_type = filter_type
+        self.stop_hash = stop_hash
+        self.prev_header = prev_header
+        self.hashes = hashes
+
+    def deserialize(self, f):
+        self.filter_type = struct.unpack("<B", f.read(1))[0]
+        self.stop_hash = deser_uint256(f)
+        self.prev_header = deser_uint256(f)
+        self.hashes = 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(self.prev_header)
+        r += ser_uint256_vector(self.hashes)
+        return r
+
+    def __repr__(self):
+        return "msg_cfheaders(filter_type={:#x}, stop_hash={:x})".format(
+            self.filter_type, self.stop_hash)
+
 class msg_getcfcheckpt:
     __slots__ = ("filter_type", "stop_hash")
     msgtype =  b"getcfcheckpt"

test/functional/test_framework/mininode.py

@@ -31,6 +31,7 @@
     msg_block,
     MSG_BLOCK,
     msg_blocktxn,
+    msg_cfheaders,
     msg_cfcheckpt,
     msg_cmpctblock,
     msg_feefilter,
@@ -68,6 +69,7 @@
     b"addr": msg_addr,
     b"block": msg_block,
     b"blocktxn": msg_blocktxn,
+    b"cfheaders": msg_cfheaders,
     b"cfcheckpt": msg_cfcheckpt,
     b"cmpctblock": msg_cmpctblock,
     b"feefilter": msg_feefilter,
@@ -330,6 +332,7 @@ def on_close(self):
     def on_addr(self, message): pass
     def on_block(self, message): pass
     def on_blocktxn(self, message): pass
+    def on_cfheaders(self, message): pass
     def on_cfcheckpt(self, message): pass
     def on_cmpctblock(self, message): pass
     def on_feefilter(self, message): pass