test: add functional test for inbound maxconnection limits

Co-authored-by: Martin Zumsande <mzumsande@gmail.com>

test/functional/p2p_connection_limits.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+# Copyright (c) 2023-present The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.messages import (
+    msg_version,
+)
+from test_framework.p2p import (
+    P2PInterface,
+    P2P_SERVICES,
+    P2P_SUBVERSION,
+    P2P_VERSION,
+)
+
+
+class P2PConnectionLimits(BitcoinTestFramework):
+    def set_test_params(self):
+        self.num_nodes = 1
+        # scenario : we have 2 inbound slots and allow a maximum of 1 tx-relaying inbound peer
+        self.extra_args = [['-maxconnections=13']]  # 11 slots are reserved for outbounds, leaving 2 inbound slots
+
+    def run_test(self):
+        self.test_inbound_limits()
+
+    def create_blocks_only_version(self):
+        no_txrelay_version_msg = msg_version()
+        no_txrelay_version_msg.nVersion = P2P_VERSION
+        no_txrelay_version_msg.strSubVer = P2P_SUBVERSION
+        no_txrelay_version_msg.nServices = P2P_SERVICES
+        no_txrelay_version_msg.relay = 0
+        return no_txrelay_version_msg
+
+    def test_inbound_limits(self):
+        node = self.nodes[0]
+
+        self.log.info('Test with 2 inbound slots, one of which allows tx-relay')
+        node.add_p2p_connection(P2PInterface())
+
+        self.log.info('Connect a full-relay inbound peer - test that eviction is triggered')
+        # Since there is no unprotected peer to evict here, the new peer is dropped instead.
+        with node.assert_debug_log(['failed to find a tx-relaying eviction candidate - connection dropped'], timeout=2):
+            self.nodes[0].add_p2p_connection(P2PInterface(), expect_success=False, wait_for_verack=False)
+        self.wait_until(lambda: len(node.getpeerinfo()) == 1)
+        node.disconnect_p2ps()
+
+        self.log.info('Connect a block-relay inbound peer - test that second full relay peer is accepted')
+        peer1 = self.nodes[0].add_p2p_connection(P2PInterface(), send_version=False, wait_for_verack=False)
+        peer1.send_without_ping(self.create_blocks_only_version())
+        peer1.wait_for_verack()
+
+        node.add_p2p_connection(P2PInterface())
+        self.wait_until(lambda: len(node.getpeerinfo()) == 2)
+
+        self.log.info('Connecting another full-relay peer triggers non-specific eviction')
+        with node.assert_debug_log(['failed to find an eviction candidate - connection dropped (full)'], timeout=2):
+            self.nodes[0].add_p2p_connection(P2PInterface(), send_version=False, wait_for_verack=False, expect_success=False)
+        self.wait_until(lambda: len(node.getpeerinfo()) == 2)
+
+
+if __name__ == '__main__':
+    P2PConnectionLimits(__file__).main()

test/functional/test_runner.py

@@ -354,6 +354,7 @@ BASE_SCRIPTS = [
     'rpc_scanblocks.py',
     'tool_bitcoin.py',
     'p2p_sendtxrcncl.py',
+    'p2p_connection_limits.py',
     'rpc_scantxoutset.py',
     'feature_torcontrol.py',
     'feature_unsupported_utxo_db.py',