Merge bitcoin/bitcoin#22569: [0.21] Rate limit the processing of rumoured addresses

2a5710805195ca54a02aff3540ceaefb9cb3b3e2 Avoid Appveyor compilation failure (Pieter Wuille)
a653aacbd66a47edd6d14ddc62fec2d4038456b8 Add logging and addr rate limiting statistics (Pieter Wuille)
aaa4833fc9c3d44378e232002f4b9b447a2d18cb Functional tests for addr rate limiting (Pieter Wuille)
8df3e5bd84f2b2b60730034cbd71fe8f3276d434 Randomize the order of addr processing (Pieter Wuille)
83dfe6c65ef6c30ca01348ee5059c3d76e03d1d3 Rate limit the processing of incoming addr messages (Pieter Wuille)

Pull request description:

  Backport of #22387.

  The rate at which IP addresses are rumoured (through ADDR and ADDRV2 messages) on the network seems to vary from 0 for some non-participating nodes, to 0.005-0.025 addr/s for recent Bitcoin Core nodes. However, the current codebase will happily accept and process an effectively unbounded rate from attackers. There are measures to limit the influence attackers can have on the addrman database (bucket restrictions based on source IPs), but still - there is no need to permit them to feed us addresses at a rate that's orders of magnitude larger than what is common on the network today, especially as it will cause us to spam our peers too.

  This PR implements a [token bucket](https://en.wikipedia.org/wiki/Token_bucket) based rate limiter, allowing an average of 0.1 addr/s per connection, with bursts up to 1000 addresses at once. Whitelisted peers as well as responses to GETADDR requests are exempt from the limit. New connections start with 1 token, so as to not interfere with the common practice of peers' self-announcement.

  Due to the lack of the `Peer` struct in 0.21, the relevant fields have been added to `CNodeState` instead, necessitating additional locks, and slightly different structure to avoid too much `cs_main` grabbing. The last test-improving commit has also been dropped, as the code has changed too much. Most of the behavior is still tested however, just not the part that compares with RPC statistics.

ACKs for top commit:
  achow101:
    ACK 2a5710805195ca54a02aff3540ceaefb9cb3b3e2
  GeneFerneau:
    Approach + code review ACK [2a57108](https://github.com/bitcoin/bitcoin/pull/22569/commits/2a5710805195ca54a02aff3540ceaefb9cb3b3e2)
  jnewbery:
    reACK 2a5710805195ca54a02aff3540ceaefb9cb3b3e2

Tree-SHA512: ecf4891ac6173d732aa40b4d05fc0dce94127a613cb9051bf6188a2f95824f8234b17d386dd0b352ddf3d352202cc2ff07915ae35657d8e64907e3f80703d1d9

8 files changed, 144 insertions, 2 deletions

diff --git a/src/net_permissions.h b/src/net_permissions.h
index bba0ea1695..3b841ab138 100644
--- a/src/net_permissions.h
+++ b/src/net_permissions.h
@@ -30,7 +30,8 @@ enum NetPermissionFlags {
     PF_NOBAN = (1U << 4) | PF_DOWNLOAD,
     // Can query the mempool
     PF_MEMPOOL = (1U << 5),
-    // Can request addrs without hitting a privacy-preserving cache
+    // Can request addrs without hitting a privacy-preserving cache, and send us
+    // unlimited amounts of addrs.
     PF_ADDR = (1U << 7),
 
     // True if the user did not specifically set fine grained permissions
diff --git a/src/net_processing.cpp b/src/net_processing.cpp
index f29be8d8a3..f926fd652f 100644
--- a/src/net_processing.cpp
+++ b/src/net_processing.cpp
@@ -146,6 +146,13 @@ static constexpr uint32_t MAX_GETCFILTERS_SIZE = 1000;
 static constexpr uint32_t MAX_GETCFHEADERS_SIZE = 2000;
 /** the maximum percentage of addresses from our addrman to return in response to a getaddr message. */
 static constexpr size_t MAX_PCT_ADDR_TO_SEND = 23;
+/** The maximum rate of address records we're willing to process on average. Can be bypassed using
+ *  the NetPermissionFlags::Addr permission. */
+static constexpr double MAX_ADDR_RATE_PER_SECOND{0.1};
+/** The soft limit of the address processing token bucket (the regular MAX_ADDR_RATE_PER_SECOND
+ *  based increments won't go above this, but the MAX_ADDR_TO_SEND increment following GETADDR
+ *  is exempt from this limit. */
+static constexpr size_t MAX_ADDR_PROCESSING_TOKEN_BUCKET{MAX_ADDR_TO_SEND};
 
 struct COrphanTx {
     // When modifying, adapt the copy of this definition in tests/DoS_tests.
@@ -454,6 +461,16 @@ struct Peer {
     /** Work queue of items requested by this peer **/
     std::deque<CInv> m_getdata_requests GUARDED_BY(m_getdata_requests_mutex);
 
+    /** Number of addr messages that can be processed from this peer. Start at 1 to
+     *  permit self-announcement. */
+    double m_addr_token_bucket{1.0};
+    /** When m_addr_token_bucket was last updated */
+    std::chrono::microseconds m_addr_token_timestamp{GetTime<std::chrono::microseconds>()};
+    /** Total number of addresses that were dropped due to rate limiting. */
+    std::atomic<uint64_t> m_addr_rate_limited{0};
+    /** Total number of addresses that were processed (excludes rate limited ones). */
+    std::atomic<uint64_t> m_addr_processed{0};
+
     Peer(NodeId id) : m_id(id) {}
 };
 
@@ -906,6 +923,8 @@ bool GetNodeStateStats(NodeId nodeid, CNodeStateStats &stats) {
     PeerRef peer = GetPeerRef(nodeid);
     if (peer == nullptr) return false;
     stats.m_misbehavior_score = WITH_LOCK(peer->m_misbehavior_mutex, return peer->m_misbehavior_score);
+    stats.m_addr_processed = peer->m_addr_processed.load();
+    stats.m_addr_rate_limited = peer->m_addr_rate_limited.load();
 
     return true;
 }
@@ -2438,6 +2457,9 @@ void PeerManager::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDat
             // Get recent addresses
             m_connman.PushMessage(&pfrom, CNetMsgMaker(greatest_common_version).Make(NetMsgType::GETADDR));
             pfrom.fGetAddr = true;
+            // When requesting a getaddr, accept an additional MAX_ADDR_TO_SEND addresses in response
+            // (bypassing the MAX_ADDR_PROCESSING_TOKEN_BUCKET limit).
+            peer->m_addr_token_bucket += MAX_ADDR_TO_SEND;
         }
 
         if (!pfrom.IsInboundConn()) {
@@ -2591,11 +2613,34 @@ void PeerManager::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDat
         std::vector<CAddress> vAddrOk;
         int64_t nNow = GetAdjustedTime();
         int64_t nSince = nNow - 10 * 60;
+
+        // Update/increment addr rate limiting bucket.
+        const auto current_time = GetTime<std::chrono::microseconds>();
+        if (peer->m_addr_token_bucket < MAX_ADDR_PROCESSING_TOKEN_BUCKET) {
+            // Don't increment bucket if it's already full
+            const auto time_diff = std::max(current_time - peer->m_addr_token_timestamp, std::chrono::microseconds{0});
+            const double increment = std::chrono::duration<double>(time_diff).count() * MAX_ADDR_RATE_PER_SECOND;
+            peer->m_addr_token_bucket = std::min<double>(peer->m_addr_token_bucket + increment, MAX_ADDR_PROCESSING_TOKEN_BUCKET);
+        }
+        peer->m_addr_token_timestamp = current_time;
+
+        const bool rate_limited = !pfrom.HasPermission(NetPermissionFlags::PF_ADDR);
+        uint64_t num_proc = 0;
+        uint64_t num_rate_limit = 0;
+        Shuffle(vAddr.begin(), vAddr.end(), FastRandomContext());
         for (CAddress& addr : vAddr)
         {
             if (interruptMsgProc)
                 return;
 
+            // Apply rate limiting.
+            if (rate_limited) {
+                if (peer->m_addr_token_bucket < 1.0) {
+                    ++num_rate_limit;
+                    continue;
+                }
+                peer->m_addr_token_bucket -= 1.0;
+            }
             // We only bother storing full nodes, though this may include
             // things which we would not make an outbound connection to, in
             // part because we may make feeler connections to them.
@@ -2609,6 +2654,7 @@ void PeerManager::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDat
                 // Do not process banned/discouraged addresses beyond remembering we received them
                 continue;
             }
+            ++num_proc;
             bool fReachable = IsReachable(addr);
             if (addr.nTime > nSince && !pfrom.fGetAddr && vAddr.size() <= 10 && addr.IsRoutable())
             {
@@ -2619,6 +2665,15 @@ void PeerManager::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDat
             if (fReachable)
                 vAddrOk.push_back(addr);
         }
+        peer->m_addr_processed += num_proc;
+        peer->m_addr_rate_limited += num_rate_limit;
+        LogPrint(BCLog::NET, "Received addr: %u addresses (%u processed, %u rate-limited) from peer=%d%s\n",
+                 vAddr.size(),
+                 num_proc,
+                 num_rate_limit,
+                 pfrom.GetId(),
+                 fLogIPs ? ", peeraddr=" + pfrom.addr.ToString() : "");
+
         m_connman.AddNewAddresses(vAddrOk, pfrom.addr, 2 * 60 * 60);
         if (vAddr.size() < 1000)
             pfrom.fGetAddr = false;
diff --git a/src/net_processing.h b/src/net_processing.h
index 87eee566de..1982551dd4 100644
--- a/src/net_processing.h
+++ b/src/net_processing.h
@@ -32,6 +32,7 @@ static const bool DEFAULT_PEERBLOCKFILTERS = false;
 /** Threshold for marking a node to be discouraged, e.g. disconnected and added to the discouragement filter. */
 static const int DISCOURAGEMENT_THRESHOLD{100};
 
+
 class PeerManager final : public CValidationInterface, public NetEventsInterface {
 public:
     PeerManager(const CChainParams& chainparams, CConnman& connman, BanMan* banman,
@@ -150,6 +151,8 @@ struct CNodeStateStats {
     int nSyncHeight = -1;
     int nCommonHeight = -1;
     std::vector<int> vHeightInFlight;
+    uint64_t m_addr_processed = 0;
+    uint64_t m_addr_rate_limited = 0;
 };
 
 /** Get statistics from node state */
diff --git a/src/rpc/misc.cpp b/src/rpc/misc.cpp
index 00f7445cfa..cca4c6787f 100644
--- a/src/rpc/misc.cpp
+++ b/src/rpc/misc.cpp
@@ -387,7 +387,8 @@ static RPCHelpMan setmocktime()
     }
     SetMockTime(time);
     if (request.context.Has<NodeContext>()) {
-        for (const auto& chain_client : request.context.Get<NodeContext>().chain_clients) {
+        const auto& chain_clients = request.context.Get<NodeContext>().chain_clients;
+        for (const auto& chain_client : chain_clients) {
             chain_client->setMockTime(time);
         }
     }
diff --git a/src/rpc/net.cpp b/src/rpc/net.cpp
index 298529e4e7..49d480c5f5 100644
--- a/src/rpc/net.cpp
+++ b/src/rpc/net.cpp
@@ -236,6 +236,8 @@ static RPCHelpMan getpeerinfo()
                 heights.push_back(height);
             }
             obj.pushKV("inflight", heights);
+            obj.pushKV("addr_processed", statestats.m_addr_processed);
+            obj.pushKV("addr_rate_limited", statestats.m_addr_rate_limited);
         }
         if (IsDeprecatedRPCEnabled("whitelisted")) {
             // whitelisted is deprecated in v0.21 for removal in v0.22
diff --git a/test/functional/p2p_addr_relay.py b/test/functional/p2p_addr_relay.py
index 80f262d0d3..1b5c9533f7 100755
--- a/test/functional/p2p_addr_relay.py
+++ b/test/functional/p2p_addr_relay.py
@@ -16,7 +16,10 @@ from test_framework.p2p import P2PInterface
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
+    assert_greater_than_or_equal,
 )
+import os
+import random
 import time
 
 ADDRS = []
@@ -41,6 +44,20 @@ class AddrTest(BitcoinTestFramework):
     def set_test_params(self):
         self.setup_clean_chain = False
         self.num_nodes = 1
+        self.extra_args = [["-whitelist=addr@127.0.0.1"]]
+
+    def setup_rand_addr_msg(self, num):
+        addrs = []
+        for i in range(num):
+            addr = CAddress()
+            addr.time = self.mocktime + i
+            addr.nServices = NODE_NETWORK | NODE_WITNESS
+            addr.ip = "%i.%i.%i.%i" % (random.randrange(128,169), random.randrange(1,255), random.randrange(1,255), random.randrange(1,255))
+            addr.port = 8333
+            addrs.append(addr)
+        msg = msg_addr()
+        msg.addrs = addrs
+        return msg
 
     def run_test(self):
         self.log.info('Create connection that sends addr messages')
@@ -64,6 +81,67 @@ class AddrTest(BitcoinTestFramework):
             self.nodes[0].setmocktime(int(time.time()) + 30 * 60)
             addr_receiver.sync_with_ping()
 
+        # The following test is backported. The original test also verified behavior for
+        # outbound peers, but lacking add_outbound_p2p_connection, those tests have been
+        # removed here.
+        for contype, tokens, no_relay in [("inbound", 1, False)]:
+            self.log.info('Test rate limiting of addr processing for %s peers' % contype)
+            self.stop_node(0)
+            os.remove(os.path.join(self.nodes[0].datadir, "regtest", "peers.dat"))
+            self.start_node(0, [])
+            self.mocktime = int(time.time())
+            self.nodes[0].setmocktime(self.mocktime)
+            peer = self.nodes[0].add_p2p_connection(AddrReceiver())
+
+            # Check that we start off with empty addrman
+            addr_count_0 = len(self.nodes[0].getnodeaddresses(0))
+            assert_equal(addr_count_0, 0)
+
+            # Send 600 addresses. For all but the block-relay-only peer this should result in at least 1 address.
+            peer.send_and_ping(self.setup_rand_addr_msg(600))
+            addr_count_1 = len(self.nodes[0].getnodeaddresses(0))
+            assert_greater_than_or_equal(tokens, addr_count_1)
+            assert_greater_than_or_equal(addr_count_0 + 600, addr_count_1)
+            assert_equal(addr_count_1 > addr_count_0, tokens > 0)
+
+            # Send 600 more addresses. For the outbound-full-relay peer (which we send a GETADDR, and thus will
+            # process up to 1001 incoming addresses), this means more entries will appear.
+            peer.send_and_ping(self.setup_rand_addr_msg(600))
+            addr_count_2 = len(self.nodes[0].getnodeaddresses(0))
+            assert_greater_than_or_equal(tokens, addr_count_2)
+            assert_greater_than_or_equal(addr_count_1 + 600, addr_count_2)
+            assert_equal(addr_count_2 > addr_count_1, tokens > 600)
+
+            # Send 10 more. As we reached the processing limit for all nodes, this should have no effect.
+            peer.send_and_ping(self.setup_rand_addr_msg(10))
+            addr_count_3 = len(self.nodes[0].getnodeaddresses(0))
+            assert_greater_than_or_equal(tokens, addr_count_3)
+            assert_equal(addr_count_2, addr_count_3)
+
+            # Advance the time by 100 seconds, permitting the processing of 10 more addresses. Send 200,
+            # but verify that no more than 10 are processed.
+            self.mocktime += 100
+            self.nodes[0].setmocktime(self.mocktime)
+            new_tokens = 0 if no_relay else 10
+            tokens += new_tokens
+            peer.send_and_ping(self.setup_rand_addr_msg(200))
+            addr_count_4 = len(self.nodes[0].getnodeaddresses(0))
+            assert_greater_than_or_equal(tokens, addr_count_4)
+            assert_greater_than_or_equal(addr_count_3 + new_tokens, addr_count_4)
+
+            # Advance the time by 1000 seconds, permitting the processing of 100 more addresses. Send 200,
+            # but verify that no more than 100 are processed (and at least some).
+            self.mocktime += 1000
+            self.nodes[0].setmocktime(self.mocktime)
+            new_tokens = 0 if no_relay else 100
+            tokens += new_tokens
+            peer.send_and_ping(self.setup_rand_addr_msg(200))
+            addr_count_5 = len(self.nodes[0].getnodeaddresses(0))
+            assert_greater_than_or_equal(tokens, addr_count_5)
+            assert_greater_than_or_equal(addr_count_4 + new_tokens, addr_count_5)
+            assert_equal(addr_count_5 > addr_count_4, not no_relay)
+
+            self.nodes[0].disconnect_p2ps()
 
 if __name__ == '__main__':
     AddrTest().main()
diff --git a/test/functional/p2p_addrv2_relay.py b/test/functional/p2p_addrv2_relay.py
index 23ce3e5d04..5abeabb5fe 100755
--- a/test/functional/p2p_addrv2_relay.py
+++ b/test/functional/p2p_addrv2_relay.py
@@ -49,6 +49,7 @@ class AddrTest(BitcoinTestFramework):
     def set_test_params(self):
         self.setup_clean_chain = True
         self.num_nodes = 1
+        self.extra_args = [["-whitelist=addr@127.0.0.1"]]
 
     def run_test(self):
         self.log.info('Create connection that sends addrv2 messages')
diff --git a/test/functional/p2p_invalid_messages.py b/test/functional/p2p_invalid_messages.py
index db72a361d9..3934e7611e 100755
--- a/test/functional/p2p_invalid_messages.py
+++ b/test/functional/p2p_invalid_messages.py
@@ -57,6 +57,7 @@ class InvalidMessagesTest(BitcoinTestFramework):
     def set_test_params(self):
         self.num_nodes = 1
         self.setup_clean_chain = True
+        self.extra_args = [["-whitelist=addr@127.0.0.1"]]
 
     def run_test(self):
         self.test_buffer()