Improve handling of unconnecting headers

qa/rpc-tests/sendheaders.py

@@ -63,6 +63,21 @@
    Expect: getdata request for 14 more blocks.
 f. Announce 1 more header that builds on that fork.
    Expect: no response.
+
+Part 5: Test handling of headers that don't connect.
+a. Repeat 10 times:
+   1. Announce a header that doesn't connect.
+      Expect: getheaders message
+   2. Send headers chain.
+      Expect: getdata for the missing blocks, tip update.
+b. Then send 9 more headers that don't connect.
+   Expect: getheaders message each time.
+c. Announce a header that does connect.
+   Expect: no response.
+d. Announce 49 headers that don't connect.
+   Expect: getheaders message each time.
+e. Announce one more that doesn't connect.
+   Expect: disconnect.
 '''
 
 class BaseNode(NodeConnCB):
@@ -77,6 +92,8 @@ def __init__(self):
         self.last_getdata = None
         self.sleep_time = 0.05
         self.block_announced = False
+        self.last_getheaders = None
+        self.disconnected = False
 
     def clear_last_announcement(self):
         with mininode_lock:
@@ -127,6 +144,12 @@ def on_getdata(self, conn, message):
     def on_pong(self, conn, message):
         self.last_pong = message
 
+    def on_getheaders(self, conn, message):
+        self.last_getheaders = message
+
+    def on_close(self, conn):
+        self.disconnected = True
+
     # Test whether the last announcement we received had the
     # right header or the right inv
     # inv and headers should be lists of block hashes
@@ -178,6 +201,11 @@ def wait_for_block(self, blockhash, timeout=60):
         self.sync(test_function, timeout)
         return
 
+    def wait_for_getheaders(self, timeout=60):
+        test_function = lambda: self.last_getheaders != None
+        self.sync(test_function, timeout)
+        return
+
     def wait_for_getdata(self, hash_list, timeout=60):
         if hash_list == []:
             return
@@ -186,6 +214,11 @@ def wait_for_getdata(self, hash_list, timeout=60):
         self.sync(test_function, timeout)
         return
 
+    def wait_for_disconnect(self, timeout=60):
+        test_function = lambda: self.disconnected
+        self.sync(test_function, timeout)
+        return
+
     def send_header_for_blocks(self, new_blocks):
         headers_message = msg_headers()
         headers_message.headers = [ CBlockHeader(b) for b in new_blocks ]
@@ -510,6 +543,78 @@ def run_test(self):
 
         print("Part 4: success!")
 
+        # Now deliver all those blocks we announced.
+        [ test_node.send_message(msg_block(x)) for x in blocks ]
+
+        print("Part 5: Testing handling of unconnecting headers")
+        # First we test that receipt of an unconnecting header doesn't prevent
+        # chain sync.
+        for i in range(10):
+            test_node.last_getdata = None
+            blocks = []
+            # Create two more blocks.
+            for j in range(2):
+                blocks.append(create_block(tip, create_coinbase(height), block_time))
+                blocks[-1].solve()
+                tip = blocks[-1].sha256
+                block_time += 1
+                height += 1
+            # Send the header of the second block -> this won't connect.
+            with mininode_lock:
+                test_node.last_getheaders = None
+            test_node.send_header_for_blocks([blocks[1]])
+            test_node.wait_for_getheaders(timeout=1)
+            test_node.send_header_for_blocks(blocks)
+            test_node.wait_for_getdata([x.sha256 for x in blocks])
+            [ test_node.send_message(msg_block(x)) for x in blocks ]
+            test_node.sync_with_ping()
+            assert_equal(int(self.nodes[0].getbestblockhash(), 16), blocks[1].sha256)
+
+        blocks = []
+        # Now we test that if we repeatedly don't send connecting headers, we
+        # don't go into an infinite loop trying to get them to connect.
+        MAX_UNCONNECTING_HEADERS = 10
+        for j in range(MAX_UNCONNECTING_HEADERS+1):
+            blocks.append(create_block(tip, create_coinbase(height), block_time))
+            blocks[-1].solve()
+            tip = blocks[-1].sha256
+            block_time += 1
+            height += 1
+
+        for i in range(1, MAX_UNCONNECTING_HEADERS):
+            # Send a header that doesn't connect, check that we get a getheaders.
+            with mininode_lock:
+                test_node.last_getheaders = None
+            test_node.send_header_for_blocks([blocks[i]])
+            test_node.wait_for_getheaders(timeout=1)
+
+        # Next header will connect, should re-set our count:
+        test_node.send_header_for_blocks([blocks[0]])
+
+        # Remove the first two entries (blocks[1] would connect):
+        blocks = blocks[2:]
+
+        # Now try to see how many unconnecting headers we can send
+        # before we get disconnected.  Should be 5*MAX_UNCONNECTING_HEADERS
+        for i in range(5*MAX_UNCONNECTING_HEADERS - 1):
+            # Send a header that doesn't connect, check that we get a getheaders.
+            with mininode_lock:
+                test_node.last_getheaders = None
+            test_node.send_header_for_blocks([blocks[i%len(blocks)]])
+            test_node.wait_for_getheaders(timeout=1)
+
+        # Eventually this stops working.
+        with mininode_lock:
+            self.last_getheaders = None
+        test_node.send_header_for_blocks([blocks[-1]])
+
+        # Should get disconnected
+        test_node.wait_for_disconnect()
+        with mininode_lock:
+            self.last_getheaders = True
+
+        print("Part 5: success!")
+
         # Finally, check that the inv node never received a getdata request,
         # throughout the test
         assert_equal(inv_node.last_getdata, None)

src/main.cpp

@@ -276,6 +276,8 @@ struct CNodeState {
     CBlockIndex *pindexLastCommonBlock;
     //! The best header we have sent our peer.
     CBlockIndex *pindexBestHeaderSent;
+    //! Length of current-streak of unconnecting headers announcements
+    int nUnconnectingHeaders;
     //! Whether we've started headers synchronization with this peer.
     bool fSyncStarted;
     //! Since when we're stalling block download progress (in microseconds), or 0.
@@ -304,6 +306,7 @@ struct CNodeState {
         hashLastUnknownBlock.SetNull();
         pindexLastCommonBlock = NULL;
         pindexBestHeaderSent = NULL;
+        nUnconnectingHeaders = 0;
         fSyncStarted = false;
         nStallingSince = 0;
         nDownloadingSince = 0;
@@ -5773,6 +5776,35 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv,
             return true;
         }
 
+        CNodeState *nodestate = State(pfrom->GetId());
+
+        // If this looks like it could be a block announcement (nCount <
+        // MAX_BLOCKS_TO_ANNOUNCE), use special logic for handling headers that
+        // don't connect:
+        // - Send a getheaders message in response to try to connect the chain.
+        // - The peer can send up to MAX_UNCONNECTING_HEADERS in a row that
+        //   don't connect before giving DoS points
+        // - Once a headers message is received that is valid and does connect,
+        //   nUnconnectingHeaders gets reset back to 0.
+        if (mapBlockIndex.find(headers[0].hashPrevBlock) == mapBlockIndex.end() && nCount < MAX_BLOCKS_TO_ANNOUNCE) {
+            nodestate->nUnconnectingHeaders++;
+            pfrom->PushMessage(NetMsgType::GETHEADERS, chainActive.GetLocator(pindexBestHeader), uint256());
+            LogPrint("net", "received header %s: missing prev block %s, sending getheaders (%d) to end (peer=%d, nUnconnectingHeaders=%d)\n",
+                    headers[0].GetHash().ToString(),
+                    headers[0].hashPrevBlock.ToString(),
+                    pindexBestHeader->nHeight,
+                    pfrom->id, nodestate->nUnconnectingHeaders);
+            // Set hashLastUnknownBlock for this peer, so that if we
+            // eventually get the headers - even from a different peer -
+            // we can use this peer to download.
+            UpdateBlockAvailability(pfrom->GetId(), headers.back().GetHash());
+
+            if (nodestate->nUnconnectingHeaders % MAX_UNCONNECTING_HEADERS == 0) {
+                Misbehaving(pfrom->GetId(), 20);
+            }
+            return true;
+        }
+
         CBlockIndex *pindexLast = NULL;
         BOOST_FOREACH(const CBlockHeader& header, headers) {
             CValidationState state;
@@ -5790,6 +5822,11 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv,
             }
         }
 
+        if (nodestate->nUnconnectingHeaders > 0) {
+            LogPrint("net", "peer=%d: resetting nUnconnectingHeaders (%d -> 0)\n", pfrom->id, nodestate->nUnconnectingHeaders);
+        }
+        nodestate->nUnconnectingHeaders = 0;
+
         assert(pindexLast);
         UpdateBlockAvailability(pfrom->GetId(), pindexLast->GetBlockHash());
 
@@ -5802,7 +5839,6 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv,
         }
 
         bool fCanDirectFetch = CanDirectFetch(chainparams.GetConsensus());
-        CNodeState *nodestate = State(pfrom->GetId());
         // If this set of headers is valid and ends in a block with at least as
         // much work as our tip, download as much as possible.
         if (fCanDirectFetch && pindexLast->IsValid(BLOCK_VALID_TREE) && chainActive.Tip()->nChainWork <= pindexLast->nChainWork) {

src/main.h

@@ -138,6 +138,9 @@ static const bool DEFAULT_FEEFILTER = true;
 /** Maximum number of headers to announce when relaying blocks with headers message.*/
 static const unsigned int MAX_BLOCKS_TO_ANNOUNCE = 8;
 
+/** Maximum number of unconnecting headers announcements before DoS score */
+static const int MAX_UNCONNECTING_HEADERS = 10;
+
 static const bool DEFAULT_PEERBLOOMFILTERS = true;
 
 struct BlockHasher