diff --git a/doc/release-notes.md b/doc/release-notes.md --- a/doc/release-notes.md +++ b/doc/release-notes.md @@ -8,3 +8,6 @@ - Add a `-fixedseeds` option which can be set to 0 to disable the hardcoded seeds. This can be used in conjunction with `dsnssed=0` to create a trusted peer only setup. In this case the nodes need to be added manually with the `-addnode` option or the `addnode` RPC. + - The node will now limit the rate at which the addresses received via p2p messages are processed. + This can be bypassed if needed by granting the `addr` permission to a peer(see the `-whitelist` + option for details). diff --git a/src/net_permissions.h b/src/net_permissions.h --- a/src/net_permissions.h +++ b/src/net_permissions.h @@ -32,7 +32,8 @@ 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), // Bypass the limit on how many proof INVs are tracked from this peer as // well as the delay penalty when reaching the the in-flight requests limit diff --git a/src/net_processing.h b/src/net_processing.h --- a/src/net_processing.h +++ b/src/net_processing.h @@ -43,6 +43,8 @@ int m_starting_height = -1; std::chrono::microseconds m_ping_wait; std::vector vHeightInFlight; + uint64_t m_addr_processed = 0; + uint64_t m_addr_rate_limited = 0; bool m_addr_relay_enabled{false}; }; diff --git a/src/net_processing.cpp b/src/net_processing.cpp --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -280,6 +280,17 @@ static constexpr size_t MAX_PCT_ADDR_TO_SEND = 23; /** The maximum number of address records permitted in an ADDR message. */ static constexpr size_t MAX_ADDR_TO_SEND{1000}; +/** + * 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}; inline size_t GetMaxAddrToSend() { return gArgs.GetArg("-maxaddrtosend", MAX_ADDR_TO_SEND); @@ -413,6 +424,21 @@ std::atomic_bool m_wants_addrv2{false}; /** Whether this peer has already sent us a getaddr message. */ bool m_getaddr_recvd{false}; + /** + * 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()}; + /** Total number of addresses that were dropped due to rate limiting. */ + std::atomic m_addr_rate_limited{0}; + /** + * Total number of addresses that were processed (excludes rate limited + * ones). + */ + std::atomic m_addr_processed{0}; /** * Set of txids to reconsider once their parent transactions have been @@ -1626,6 +1652,8 @@ } stats.m_ping_wait = ping_wait; + stats.m_addr_processed = peer->m_addr_processed.load(); + stats.m_addr_rate_limited = peer->m_addr_rate_limited.load(); stats.m_addr_relay_enabled = peer->m_addr_relay_enabled.load(); return true; @@ -3511,6 +3539,10 @@ m_connman.PushMessage(&pfrom, CNetMsgMaker(greatest_common_version) .Make(NetMsgType::GETADDR)); peer->m_getaddr_sent = 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 += GetMaxAddrToSend(); } if (!pfrom.IsInboundConn()) { @@ -3674,11 +3706,40 @@ std::vector vAddrOk; int64_t nNow = GetAdjustedTime(); int64_t nSince = nNow - 10 * 60; + + // Update/increment addr rate limiting bucket. + const auto current_time = GetTime(); + 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, 0us); + const double increment = + CountSecondsDouble(time_diff) * MAX_ADDR_RATE_PER_SECOND; + peer->m_addr_token_bucket = + std::min(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. @@ -3697,6 +3758,7 @@ // remembering we received them continue; } + ++num_proc; bool fReachable = IsReachable(addr); if (addr.nTime > nSince && !peer->m_getaddr_sent && vAddr.size() <= 10 && addr.IsRoutable()) { @@ -3708,6 +3770,13 @@ 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) { diff --git a/src/rpc/net.cpp b/src/rpc/net.cpp --- a/src/rpc/net.cpp +++ b/src/rpc/net.cpp @@ -285,6 +285,9 @@ 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(gArgs, "whitelisted")) { // whitelisted is deprecated in v0.24.7 for removal in v0.25 diff --git a/test/functional/p2p_addr_relay.py b/test/functional/p2p_addr_relay.py --- a/test/functional/p2p_addr_relay.py +++ b/test/functional/p2p_addr_relay.py @@ -6,6 +6,7 @@ Test addr relay """ +import random import time from test_framework.messages import ( @@ -15,7 +16,7 @@ msg_getaddr, msg_verack, ) -from test_framework.p2p import P2PInterface +from test_framework.p2p import P2PInterface, p2p_lock from test_framework.test_framework import BitcoinTestFramework from test_framework.util import assert_equal, assert_greater_than @@ -23,6 +24,7 @@ class AddrReceiver(P2PInterface): num_ipv4_received = 0 test_addr_contents = False + _tokens = 1 send_getaddr = True def __init__(self, test_addr_contents=False, send_getaddr=True): @@ -42,6 +44,22 @@ "Invalid addr.port of {} (8333-8342 expected)".format(addr.port)) assert addr.ip.startswith('123.123.123.') + def on_getaddr(self, message): + # When the node sends us a getaddr, it increments the addr relay tokens + # for the connection by 1000 + self._tokens += 1000 + + @property + def tokens(self): + with p2p_lock: + return self._tokens + + def increment_tokens(self, n): + # When we move mocktime forward, the node increments the addr relay + # tokens for its peers + with p2p_lock: + self._tokens += n + def addr_received(self): return self.num_ipv4_received != 0 @@ -60,6 +78,7 @@ def set_test_params(self): self.num_nodes = 1 + self.extra_args = [["-whitelist=addr@127.0.0.1"]] def run_test(self): self.oversized_addr_test() @@ -70,6 +89,7 @@ # in subsequent tests self.getaddr_tests() self.blocksonly_mode_tests() + self.rate_limit_tests() def setup_addr_msg(self, num): addrs = [] @@ -86,6 +106,22 @@ msg.addrs = addrs return msg + def setup_rand_addr_msg(self, num): + addrs = [] + for i in range(num): + addr = CAddress() + addr.time = self.mocktime + i + addr.nServices = NODE_NETWORK + addr.ip = ( + f"{random.randrange(128,169)}.{random.randrange(1,255)}" + f".{random.randrange(1,255)}.{random.randrange(1,255)}" + ) + addr.port = 8333 + addrs.append(addr) + msg = msg_addr() + msg.addrs = addrs + return msg + def send_addr_msg(self, source, msg, receivers): source.send_and_ping(msg) # pop m_next_addr_send timer @@ -305,7 +341,7 @@ def blocksonly_mode_tests(self): self.log.info('Test addr relay in -blocksonly mode') - self.restart_node(0, ["-blocksonly"]) + self.restart_node(0, ["-blocksonly", "-whitelist=addr@127.0.0.1"]) self.mocktime = int(time.time()) self.log.info('Check that we send getaddr messages') @@ -322,6 +358,77 @@ self.nodes[0].disconnect_p2ps() + def send_addrs_and_test_rate_limiting(self, peer, no_relay, new_addrs, + total_addrs): + """Send an addr message and check that the number of addresses processed + and rate-limited is as expected + """ + peer.send_and_ping(self.setup_rand_addr_msg(new_addrs)) + + peerinfo = self.nodes[0].getpeerinfo()[0] + addrs_processed = peerinfo['addr_processed'] + addrs_rate_limited = peerinfo['addr_rate_limited'] + self.log.debug(f"addrs_processed = {addrs_processed}, " + f"addrs_rate_limited = {addrs_rate_limited}") + + if no_relay: + assert_equal(addrs_processed, 0) + assert_equal(addrs_rate_limited, 0) + else: + assert_equal(addrs_processed, min(total_addrs, peer.tokens)) + assert_equal(addrs_rate_limited, max(0, total_addrs - peer.tokens)) + + def rate_limit_tests(self): + self.mocktime = int(time.time()) + self.restart_node(0, []) + self.nodes[0].setmocktime(self.mocktime) + + for contype, no_relay in [ + ("outbound-full-relay", False), + ("block-relay-only", True), + ("inbound", False) + ]: + self.log.info( + f'Test rate limiting of addr processing for {contype} peers') + if contype == "inbound": + peer = self.nodes[0].add_p2p_connection(AddrReceiver()) + else: + peer = self.nodes[0].add_outbound_p2p_connection( + AddrReceiver(), p2p_idx=0, connection_type=contype) + + # Send 600 addresses. For all but the block-relay-only peer this + # should result in addresses being processed. + self.send_addrs_and_test_rate_limiting(peer, no_relay, 600, 600) + + # 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 addresses will be processed. + self.send_addrs_and_test_rate_limiting(peer, no_relay, 600, 1200) + + # Send 10 more. As we reached the processing limit for all nodes, + # no more addresses should be procesesd. + self.send_addrs_and_test_rate_limiting(peer, no_relay, 10, 1210) + + # Advance the time by 100 seconds, permitting the processing of 10 + # more addresses. + # Send 200 and verify that 10 are processed. + self.mocktime += 100 + self.nodes[0].setmocktime(self.mocktime) + peer.increment_tokens(10) + + self.send_addrs_and_test_rate_limiting(peer, no_relay, 200, 1410) + + # Advance the time by 1000 seconds, permitting the processing of 100 + # more addresses. + # Send 200 and verify that 100 are processed. + self.mocktime += 1000 + self.nodes[0].setmocktime(self.mocktime) + peer.increment_tokens(100) + + self.send_addrs_and_test_rate_limiting(peer, no_relay, 200, 1610) + + 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 --- a/test/functional/p2p_addrv2_relay.py +++ b/test/functional/p2p_addrv2_relay.py @@ -11,7 +11,6 @@ from test_framework.messages import NODE_NETWORK, CAddress, msg_addrv2 from test_framework.p2p import P2PInterface from test_framework.test_framework import BitcoinTestFramework -from test_framework.util import assert_equal ADDRS = [] for i in range(10): @@ -30,11 +29,10 @@ super().__init__(support_addrv2=True) def on_addrv2(self, message): - for addr in message.addrs: - assert_equal(addr.nServices, NODE_NETWORK) - assert addr.ip.startswith('123.123.123.') - assert (8333 <= addr.port < 8343) - self.addrv2_received_and_checked = True + expected_set = set((addr.ip, addr.port) for addr in ADDRS) + received_set = set((addr.ip, addr.port) for addr in message.addrs) + if expected_set == received_set: + self.addrv2_received_and_checked = True def wait_for_addrv2(self): self.wait_until(lambda: "addrv2" in self.last_message) @@ -44,6 +42,7 @@ 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 --- a/test/functional/p2p_invalid_messages.py +++ b/test/functional/p2p_invalid_messages.py @@ -54,6 +54,7 @@ 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()