Merge #19315: [tests] Allow outbound & block-relay-only connections in functional tests.

b4dd2ef800 [test] Test the add_outbound_p2p_connection functionality (Amiti Uttarwar)
602e69e427 [test] P2PBlocksOnly - Test block-relay-only connections. (Amiti Uttarwar)
8bb6beacb1 [test/refactor] P2PBlocksOnly - Extract transaction violation test into helper. (Amiti Uttarwar)
99791e7560 [test/refactor] P2PBlocksOnly - simplify transaction creation using blocktool helper. (Amiti Uttarwar)
3997ab9154 [test] Add test framework support to create outbound connections. (Amiti Uttarwar)
5bc04e8837 [rpc/net] Introduce addconnection to test outbounds & blockrelay (Amiti Uttarwar)

Pull request description:

  The existing functional test framework uses the `addnode` RPC to spin up manual connections between bitcoind nodes. This limits our ability to add integration tests for our networking code, which often executes different code paths for different connection types.

  **This PR enables creating `outbound` & `block-relay-only` P2P connections in the functional tests.** This allows us to increase our p2p test coverage, since we can now verify expectations around these connection types.

  This builds out the [prototype](https://github.com/bitcoin/bitcoin/issues/14210#issuecomment-527421978) proposed by ajtowns in #14210. 🙌🏽

  An overview of this branch:
  - introduces a new test-only RPC function `addconnection` which initiates opening an `outbound` or `block-relay-only` connection. (conceptually similar to `addnode` but for different connection types & restricted to regtest)
  - adds `test_framework` support so a mininode can open an `outbound`/`block-relay-only` connection to a `P2PInterface`/`P2PConnection`.
  - updates `p2p_blocksonly` tests to create a `block-relay-only` connection & verify expectations around transaction relay.
  - introduces `p2p_add_connections` test that checks the behaviors of the newly introduced `add_outbound_p2p_connection` test framework function.

  With these changes, there are many more behaviors that we can add integration tests for. The blocksonly updates is just one example.

  Huge props to ajtowns for conceiving the approach & providing me feedback as I've built out this branch. Also thank you to jnewbery for lots of thoughtful input along the way.

ACKs for top commit:
  troygiorshev:
    reACK b4dd2ef800
  jnewbery:
    utACK b4dd2ef800
  MarcoFalke:
    Approach ACK b4dd2ef800 🍢

Tree-SHA512: d1cba768c19c9c80e6a38b1c340cc86a90701b14772c4a0791c458f9097f6a4574b4a4acc7d13d6790c7b1f1f197e2c3d87996270f177402145f084ef8519a6b

src/net.cpp

@@ -1122,6 +1122,27 @@ void CConnman::AcceptConnection(const ListenSocket& hListenSocket) {
     RandAddEvent((uint32_t)id);
 }
 
+bool CConnman::AddConnection(const std::string& address, ConnectionType conn_type)
+{
+    if (conn_type != ConnectionType::OUTBOUND_FULL_RELAY && conn_type != ConnectionType::BLOCK_RELAY) return false;
+
+    const int max_connections = conn_type == ConnectionType::OUTBOUND_FULL_RELAY ? m_max_outbound_full_relay : m_max_outbound_block_relay;
+
+    // Count existing connections
+    int existing_connections = WITH_LOCK(cs_vNodes,
+                                         return std::count_if(vNodes.begin(), vNodes.end(), [conn_type](CNode* node) { return node->m_conn_type == conn_type; }););
+
+    // Max connections of specified type already exist
+    if (existing_connections >= max_connections) return false;
+
+    // Max total outbound connections already exist
+    CSemaphoreGrant grant(*semOutbound, true);
+    if (!grant) return false;
+
+    OpenNetworkConnection(CAddress(), false, &grant, address.c_str(), conn_type);
+    return true;
+}
+
 void CConnman::DisconnectNodes()
 {
     {

src/net.h

@@ -938,6 +938,19 @@ public:
     bool RemoveAddedNode(const std::string& node);
     std::vector<AddedNodeInfo> GetAddedNodeInfo();
 
+    /**
+     * Attempts to open a connection. Currently only used from tests.
+     *
+     * @param[in]   address     Address of node to try connecting to
+     * @param[in]   conn_type   ConnectionType::OUTBOUND or ConnectionType::BLOCK_RELAY
+     * @return      bool        Returns false if there are no available
+     *                          slots for this connection:
+     *                          - conn_type not a supported ConnectionType
+     *                          - Max total outbound connection capacity filled
+     *                          - Max connection capacity for type is filled
+     */
+    bool AddConnection(const std::string& address, ConnectionType conn_type);
+
     size_t GetNodeCount(NumConnections num);
     void GetNodeStats(std::vector<CNodeStats>& vstats);
     bool DisconnectNode(const std::string& node);

src/rpc/net.cpp

@@ -5,6 +5,7 @@
 #include <rpc/server.h>
 
 #include <banman.h>
+#include <chainparams.h>
 #include <clientversion.h>
 #include <core_io.h>
 #include <net.h>
@@ -314,6 +315,61 @@ static RPCHelpMan addnode()
     };
 }
 
+static RPCHelpMan addconnection()
+{
+    return RPCHelpMan{"addconnection",
+        "\nOpen an outbound connection to a specified node. This RPC is for testing only.\n",
+        {
+            {"address", RPCArg::Type::STR, RPCArg::Optional::NO, "The IP address and port to attempt connecting to."},
+            {"connection_type", RPCArg::Type::STR, RPCArg::Optional::NO, "Type of connection to open, either \"outbound-full-relay\" or \"block-relay-only\"."},
+        },
+        RPCResult{
+            RPCResult::Type::OBJ, "", "",
+            {
+                { RPCResult::Type::STR, "address", "Address of newly added connection." },
+                { RPCResult::Type::STR, "connection_type", "Type of connection opened." },
+            }},
+        RPCExamples{
+            HelpExampleCli("addconnection", "\"192.168.0.6:8333\" \"outbound-full-relay\"")
+            + HelpExampleRpc("addconnection", "\"192.168.0.6:8333\" \"outbound-full-relay\"")
+        },
+        [&](const RPCHelpMan& self, const JSONRPCRequest& request) -> UniValue
+{
+    if (Params().NetworkIDString() != CBaseChainParams::REGTEST) {
+        throw std::runtime_error("addconnection is for regression testing (-regtest mode) only.");
+    }
+
+    RPCTypeCheck(request.params, {UniValue::VSTR, UniValue::VSTR});
+    const std::string address = request.params[0].get_str();
+    const std::string conn_type_in{TrimString(request.params[1].get_str())};
+    ConnectionType conn_type{};
+    if (conn_type_in == "outbound-full-relay") {
+        conn_type = ConnectionType::OUTBOUND_FULL_RELAY;
+    } else if (conn_type_in == "block-relay-only") {
+        conn_type = ConnectionType::BLOCK_RELAY;
+    } else {
+        throw JSONRPCError(RPC_INVALID_PARAMETER, self.ToString());
+    }
+
+    NodeContext& node = EnsureNodeContext(request.context);
+    if (!node.connman) {
+        throw JSONRPCError(RPC_CLIENT_P2P_DISABLED, "Error: Peer-to-peer functionality missing or disabled.");
+    }
+
+    const bool success = node.connman->AddConnection(address, conn_type);
+    if (!success) {
+        throw JSONRPCError(RPC_CLIENT_NODE_CAPACITY_REACHED, "Error: Already at capacity for specified connection type.");
+    }
+
+    UniValue info(UniValue::VOBJ);
+    info.pushKV("address", address);
+    info.pushKV("connection_type", conn_type_in);
+
+    return info;
+},
+    };
+}
+
 static RPCHelpMan disconnectnode()
 {
     return RPCHelpMan{"disconnectnode",
@@ -900,6 +956,8 @@ static const CRPCCommand commands[] =
     { "network",            "clearbanned",            &clearbanned,            {} },
     { "network",            "setnetworkactive",       &setnetworkactive,       {"state"} },
     { "network",            "getnodeaddresses",       &getnodeaddresses,       {"count"} },
+
+    { "hidden",             "addconnection",          &addconnection,          {"address", "connection_type"} },
     { "hidden",             "addpeeraddress",         &addpeeraddress,         {"address", "port"} },
 };
 // clang-format on

src/rpc/protocol.h

@@ -62,6 +62,7 @@ enum RPCErrorCode
     RPC_CLIENT_NODE_NOT_CONNECTED   = -29, //!< Node to disconnect not found in connected nodes
     RPC_CLIENT_INVALID_IP_OR_SUBNET = -30, //!< Invalid IP/Subnet
     RPC_CLIENT_P2P_DISABLED         = -31, //!< No valid connection manager instance found
+    RPC_CLIENT_NODE_CAPACITY_REACHED= -34, //!< Max number of outbound or block-relay connections already open
 
     //! Chain errors
     RPC_CLIENT_MEMPOOL_DISABLED     = -33, //!< No mempool instance found

test/functional/p2p_add_connections.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+# Copyright (c) 2020 The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Test add_outbound_p2p_connection test framework functionality"""
+
+from test_framework.p2p import P2PInterface
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import assert_equal
+
+
+def check_node_connections(*, node, num_in, num_out):
+    info = node.getnetworkinfo()
+    assert_equal(info["connections_in"], num_in)
+    assert_equal(info["connections_out"], num_out)
+
+
+class P2PAddConnections(BitcoinTestFramework):
+    def set_test_params(self):
+        self.setup_clean_chain = False
+        self.num_nodes = 2
+
+    def setup_network(self):
+        self.setup_nodes()
+        # Don't connect the nodes
+
+    def run_test(self):
+        self.log.info("Add 8 outbounds to node 0")
+        for i in range(8):
+            self.log.info(f"outbound: {i}")
+            self.nodes[0].add_outbound_p2p_connection(P2PInterface(), p2p_idx=i, connection_type="outbound-full-relay")
+
+        self.log.info("Add 2 block-relay-only connections to node 0")
+        for i in range(2):
+            self.log.info(f"block-relay-only: {i}")
+            # set p2p_idx based on the outbound connections already open to the
+            # node, so add 8 to account for the previous full-relay connections
+            self.nodes[0].add_outbound_p2p_connection(P2PInterface(), p2p_idx=i + 8, connection_type="block-relay-only")
+
+        self.log.info("Add 2 block-relay-only connections to node 1")
+        for i in range(2):
+            self.log.info(f"block-relay-only: {i}")
+            self.nodes[1].add_outbound_p2p_connection(P2PInterface(), p2p_idx=i, connection_type="block-relay-only")
+
+        self.log.info("Add 5 inbound connections to node 1")
+        for i in range(5):
+            self.log.info(f"inbound: {i}")
+            self.nodes[1].add_p2p_connection(P2PInterface())
+
+        self.log.info("Add 8 outbounds to node 1")
+        for i in range(8):
+            self.log.info(f"outbound: {i}")
+            # bump p2p_idx to account for the 2 existing outbounds on node 1
+            self.nodes[1].add_outbound_p2p_connection(P2PInterface(), p2p_idx=i + 2)
+
+        self.log.info("Check the connections opened as expected")
+        check_node_connections(node=self.nodes[0], num_in=0, num_out=10)
+        check_node_connections(node=self.nodes[1], num_in=5, num_out=10)
+
+        self.log.info("Disconnect p2p connections & try to re-open")
+        self.nodes[0].disconnect_p2ps()
+        check_node_connections(node=self.nodes[0], num_in=0, num_out=0)
+
+        self.log.info("Add 8 outbounds to node 0")
+        for i in range(8):
+            self.log.info(f"outbound: {i}")
+            self.nodes[0].add_outbound_p2p_connection(P2PInterface(), p2p_idx=i)
+        check_node_connections(node=self.nodes[0], num_in=0, num_out=8)
+
+        self.log.info("Add 2 block-relay-only connections to node 0")
+        for i in range(2):
+            self.log.info(f"block-relay-only: {i}")
+            # bump p2p_idx to account for the 8 existing outbounds on node 0
+            self.nodes[0].add_outbound_p2p_connection(P2PInterface(), p2p_idx=i + 8, connection_type="block-relay-only")
+        check_node_connections(node=self.nodes[0], num_in=0, num_out=10)
+
+        self.log.info("Restart node 0 and try to reconnect to p2ps")
+        self.restart_node(0)
+
+        self.log.info("Add 4 outbounds to node 0")
+        for i in range(4):
+            self.log.info(f"outbound: {i}")
+            self.nodes[0].add_outbound_p2p_connection(P2PInterface(), p2p_idx=i)
+        check_node_connections(node=self.nodes[0], num_in=0, num_out=4)
+
+        self.log.info("Add 2 block-relay-only connections to node 0")
+        for i in range(2):
+            self.log.info(f"block-relay-only: {i}")
+            # bump p2p_idx to account for the 4 existing outbounds on node 0
+            self.nodes[0].add_outbound_p2p_connection(P2PInterface(), p2p_idx=i + 4, connection_type="block-relay-only")
+        check_node_connections(node=self.nodes[0], num_in=0, num_out=6)
+
+        check_node_connections(node=self.nodes[1], num_in=5, num_out=10)
+
+
+if __name__ == '__main__':
+    P2PAddConnections().main()

test/functional/p2p_blocksonly.py

@@ -2,10 +2,13 @@
 # Copyright (c) 2019-2020 The Bitcoin Core developers
 # Distributed under the MIT software license, see the accompanying
 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
-"""Test p2p blocksonly"""
+"""Test p2p blocksonly mode & block-relay-only connections."""
 
-from test_framework.messages import msg_tx, CTransaction, FromHex
-from test_framework.p2p import P2PInterface
+import time
+
+from test_framework.blocktools import create_transaction
+from test_framework.messages import msg_tx
+from test_framework.p2p import P2PInterface, P2PTxInvStore
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import assert_equal
 
@@ -16,48 +19,30 @@ class P2PBlocksOnly(BitcoinTestFramework):
         self.num_nodes = 1
         self.extra_args = [["-blocksonly"]]
 
+    def skip_test_if_missing_module(self):
+        self.skip_if_no_wallet()
+
     def run_test(self):
-        block_relay_peer = self.nodes[0].add_p2p_connection(P2PInterface())
+        self.blocksonly_mode_tests()
+        self.blocks_relay_conn_tests()
 
-        self.log.info('Check that txs from p2p are rejected and result in disconnect')
-        prevtx = self.nodes[0].getblock(self.nodes[0].getblockhash(1), 2)['tx'][0]
-        rawtx = self.nodes[0].createrawtransaction(
-            inputs=[{
-                'txid': prevtx['txid'],
-                'vout': 0
-            }],
-            outputs=[{
-                self.nodes[0].get_deterministic_priv_key().address: 50 - 0.00125
-            }],
-        )
-        sigtx = self.nodes[0].signrawtransactionwithkey(
-            hexstring=rawtx,
-            privkeys=[self.nodes[0].get_deterministic_priv_key().key],
-            prevtxs=[{
-                'txid': prevtx['txid'],
-                'vout': 0,
-                'scriptPubKey': prevtx['vout'][0]['scriptPubKey']['hex'],
-            }],
-        )['hex']
+    def blocksonly_mode_tests(self):
+        self.log.info("Tests with node running in -blocksonly mode")
         assert_equal(self.nodes[0].getnetworkinfo()['localrelay'], False)
-        with self.nodes[0].assert_debug_log(['transaction sent in violation of protocol peer=0']):
-            block_relay_peer.send_message(msg_tx(FromHex(CTransaction(), sigtx)))
-            block_relay_peer.wait_for_disconnect()
-            assert_equal(self.nodes[0].getmempoolinfo()['size'], 0)
 
-        # Remove the disconnected peer and add a new one.
-        del self.nodes[0].p2ps[0]
-        tx_relay_peer = self.nodes[0].add_p2p_connection(P2PInterface())
+        self.nodes[0].add_p2p_connection(P2PInterface())
+        tx, txid, tx_hex = self.check_p2p_tx_violation()
 
         self.log.info('Check that txs from rpc are not rejected and relayed to other peers')
+        tx_relay_peer = self.nodes[0].add_p2p_connection(P2PInterface())
         assert_equal(self.nodes[0].getpeerinfo()[0]['relaytxes'], True)
-        txid = self.nodes[0].testmempoolaccept([sigtx])[0]['txid']
+
+        assert_equal(self.nodes[0].testmempoolaccept([tx_hex])[0]['allowed'], True)
         with self.nodes[0].assert_debug_log(['received getdata for: wtx {} peer=1'.format(txid)]):
-            self.nodes[0].sendrawtransaction(sigtx)
+            self.nodes[0].sendrawtransaction(tx_hex)
             tx_relay_peer.wait_for_tx(txid)
             assert_equal(self.nodes[0].getmempoolinfo()['size'], 1)
 
-        self.log.info('Check that txs from peers with relay-permission are not rejected and relayed to others')
         self.log.info("Restarting node 0 with relay permission and blocksonly")
         self.restart_node(0, ["-persistmempool=0", "-whitelist=relay@127.0.0.1", "-blocksonly"])
         assert_equal(self.nodes[0].getrawmempool(), [])
@@ -67,8 +52,7 @@ class P2PBlocksOnly(BitcoinTestFramework):
         assert_equal(peer_1_info['permissions'], ['relay'])
         peer_2_info = self.nodes[0].getpeerinfo()[1]
         assert_equal(peer_2_info['permissions'], ['relay'])
-        assert_equal(self.nodes[0].testmempoolaccept([sigtx])[0]['allowed'], True)
-        txid = self.nodes[0].testmempoolaccept([sigtx])[0]['txid']
+        assert_equal(self.nodes[0].testmempoolaccept([tx_hex])[0]['allowed'], True)
 
         self.log.info('Check that the tx from first_peer with relay-permission is relayed to others (ie.second_peer)')
         with self.nodes[0].assert_debug_log(["received getdata"]):
@@ -78,13 +62,58 @@ class P2PBlocksOnly(BitcoinTestFramework):
             # But if, for some reason, first_peer decides to relay transactions to us anyway, we should relay them to
             # second_peer since we gave relay permission to first_peer.
             # See https://github.com/bitcoin/bitcoin/issues/19943 for details.
-            first_peer.send_message(msg_tx(FromHex(CTransaction(), sigtx)))
+            first_peer.send_message(msg_tx(tx))
             self.log.info('Check that the peer with relay-permission is still connected after sending the transaction')
             assert_equal(first_peer.is_connected, True)
             second_peer.wait_for_tx(txid)
             assert_equal(self.nodes[0].getmempoolinfo()['size'], 1)
         self.log.info("Relay-permission peer's transaction is accepted and relayed")
 
+        self.nodes[0].disconnect_p2ps()
+        self.nodes[0].generate(1)
+
+    def blocks_relay_conn_tests(self):
+        self.log.info('Tests with node in normal mode with block-relay-only connections')
+        self.restart_node(0, ["-noblocksonly"])  # disables blocks only mode
+        assert_equal(self.nodes[0].getnetworkinfo()['localrelay'], True)
+
+        # Ensure we disconnect if a block-relay-only connection sends us a transaction
+        self.nodes[0].add_outbound_p2p_connection(P2PInterface(), p2p_idx=0, connection_type="block-relay-only")
+        assert_equal(self.nodes[0].getpeerinfo()[0]['relaytxes'], False)
+        _, txid, tx_hex = self.check_p2p_tx_violation(index=2)
+
+        self.log.info("Check that txs from RPC are not sent to blockrelay connection")
+        conn = self.nodes[0].add_outbound_p2p_connection(P2PTxInvStore(), p2p_idx=1, connection_type="block-relay-only")
+
+        self.nodes[0].sendrawtransaction(tx_hex)
+
+        # Bump time forward to ensure nNextInvSend timer pops
+        self.nodes[0].setmocktime(int(time.time()) + 60)
+
+        # Calling sync_with_ping twice requires that the node calls
+        # `ProcessMessage` twice, and thus ensures `SendMessages` must have
+        # been called at least once
+        conn.sync_with_ping()
+        conn.sync_with_ping()
+        assert(int(txid, 16) not in conn.get_invs())
+
+    def check_p2p_tx_violation(self, index=1):
+        self.log.info('Check that txs from P2P are rejected and result in disconnect')
+        input_txid = self.nodes[0].getblock(self.nodes[0].getblockhash(index), 2)['tx'][0]['txid']
+        tx = create_transaction(self.nodes[0], input_txid, self.nodes[0].getnewaddress(), amount=(50 - 0.001))
+        txid = tx.rehash()
+        tx_hex = tx.serialize().hex()
+
+        with self.nodes[0].assert_debug_log(['transaction sent in violation of protocol peer=0']):
+            self.nodes[0].p2ps[0].send_message(msg_tx(tx))
+            self.nodes[0].p2ps[0].wait_for_disconnect()
+            assert_equal(self.nodes[0].getmempoolinfo()['size'], 0)
+
+        # Remove the disconnected peer
+        del self.nodes[0].p2ps[0]
+
+        return tx, txid, tx_hex
+
 
 if __name__ == '__main__':
     P2PBlocksOnly().main()

test/functional/test_framework/p2p.py

@@ -71,7 +71,11 @@ from test_framework.messages import (
     NODE_WITNESS,
     sha256,
 )
-from test_framework.util import wait_until_helper
+from test_framework.util import (
+    MAX_NODES,
+    p2p_port,
+    wait_until_helper,
+)
 
 logger = logging.getLogger("TestFramework.p2p")
 
@@ -139,7 +143,7 @@ class P2PConnection(asyncio.Protocol):
     def is_connected(self):
         return self._transport is not None
 
-    def peer_connect(self, dstaddr, dstport, *, net, timeout_factor):
+    def peer_connect_helper(self, dstaddr, dstport, net, timeout_factor):
         assert not self.is_connected
         self.timeout_factor = timeout_factor
         self.dstaddr = dstaddr
@@ -148,12 +152,20 @@ class P2PConnection(asyncio.Protocol):
         self.on_connection_send_msg = None
         self.recvbuf = b""
         self.magic_bytes = MAGIC_BYTES[net]
-        logger.debug('Connecting to Bitcoin Node: %s:%d' % (self.dstaddr, self.dstport))
+
+    def peer_connect(self, dstaddr, dstport, *, net, timeout_factor):
+        self.peer_connect_helper(dstaddr, dstport, net, timeout_factor)
 
         loop = NetworkThread.network_event_loop
-        conn_gen_unsafe = loop.create_connection(lambda: self, host=self.dstaddr, port=self.dstport)
-        conn_gen = lambda: loop.call_soon_threadsafe(loop.create_task, conn_gen_unsafe)
-        return conn_gen
+        logger.debug('Connecting to Bitcoin Node: %s:%d' % (self.dstaddr, self.dstport))
+        coroutine = loop.create_connection(lambda: self, host=self.dstaddr, port=self.dstport)
+        return lambda: loop.call_soon_threadsafe(loop.create_task, coroutine)
+
+    def peer_accept_connection(self, connect_id, connect_cb=lambda: None, *, net, timeout_factor):
+        self.peer_connect_helper('0', 0, net, timeout_factor)
+
+        logger.debug('Listening for Bitcoin Node with id: {}'.format(connect_id))
+        return lambda: NetworkThread.listen(self, connect_cb, idx=connect_id)
 
     def peer_disconnect(self):
         # Connection could have already been closed by other end.
@@ -312,18 +324,27 @@ class P2PInterface(P2PConnection):
         # If the peer supports wtxid-relay
         self.wtxidrelay = wtxidrelay
 
-    def peer_connect(self, *args, services=NODE_NETWORK|NODE_WITNESS, send_version=True, **kwargs):
+    def peer_connect_send_version(self, services):
+        # Send a version msg
+        vt = msg_version()
+        vt.nServices = services
+        vt.addrTo.ip = self.dstaddr
+        vt.addrTo.port = self.dstport
+        vt.addrFrom.ip = "0.0.0.0"
+        vt.addrFrom.port = 0
+        self.on_connection_send_msg = vt  # Will be sent in connection_made callback
+
+    def peer_connect(self, *args, services=NODE_NETWORK | NODE_WITNESS, send_version=True, **kwargs):
         create_conn = super().peer_connect(*args, **kwargs)
 
         if send_version:
-            # Send a version msg
-            vt = msg_version()
-            vt.nServices = services
-            vt.addrTo.ip = self.dstaddr
-            vt.addrTo.port = self.dstport
-            vt.addrFrom.ip = "0.0.0.0"
-            vt.addrFrom.port = 0
-            self.on_connection_send_msg = vt  # Will be sent soon after connection_made
+            self.peer_connect_send_version(services)
+
+        return create_conn
+
+    def peer_accept_connection(self, *args, services=NODE_NETWORK | NODE_WITNESS, **kwargs):
+        create_conn = super().peer_accept_connection(*args, **kwargs)
+        self.peer_connect_send_version(services)
 
         return create_conn
 
@@ -414,6 +435,10 @@ class P2PInterface(P2PConnection):
 
         wait_until_helper(test_function, timeout=timeout, lock=p2p_lock, timeout_factor=self.timeout_factor)
 
+    def wait_for_connect(self, timeout=60):
+        test_function = lambda: self.is_connected
+        wait_until_helper(test_function, timeout=timeout, lock=p2p_lock)
+
     def wait_for_disconnect(self, timeout=60):
         test_function = lambda: not self.is_connected
         self.wait_until(test_function, timeout=timeout, check_connected=False)
@@ -527,6 +552,8 @@ class NetworkThread(threading.Thread):
         # There is only one event loop and no more than one thread must be created
         assert not self.network_event_loop
 
+        NetworkThread.listeners = {}
+        NetworkThread.protos = {}
         NetworkThread.network_event_loop = asyncio.new_event_loop()
 
     def run(self):
@@ -542,6 +569,48 @@ class NetworkThread(threading.Thread):
         # Safe to remove event loop.
         NetworkThread.network_event_loop = None
 
+    @classmethod
+    def listen(cls, p2p, callback, port=None, addr=None, idx=1):
+        """ Ensure a listening server is running on the given port, and run the
+        protocol specified by `p2p` on the next connection to it. Once ready
+        for connections, call `callback`."""
+
+        if port is None:
+            assert 0 < idx <= MAX_NODES
+            port = p2p_port(MAX_NODES - idx)
+        if addr is None:
+            addr = '127.0.0.1'
+
+        coroutine = cls.create_listen_server(addr, port, callback, p2p)
+        cls.network_event_loop.call_soon_threadsafe(cls.network_event_loop.create_task, coroutine)
+
+    @classmethod
+    async def create_listen_server(cls, addr, port, callback, proto):
+        def peer_protocol():
+            """Returns a function that does the protocol handling for a new
+            connection. To allow different connections to have different
+            behaviors, the protocol function is first put in the cls.protos
+            dict. When the connection is made, the function removes the
+            protocol function from that dict, and returns it so the event loop
+            can start executing it."""
+            response = cls.protos.get((addr, port))
+            cls.protos[(addr, port)] = None
+            return response
+
+        if (addr, port) not in cls.listeners:
+            # When creating a listener on a given (addr, port) we only need to
+            # do it once. If we want different behaviors for different
+            # connections, we can accomplish this by providing different
+            # `proto` functions
+
+            listener = await cls.network_event_loop.create_server(peer_protocol, addr, port)
+            logger.debug("Listening server on %s:%d should be started" % (addr, port))
+            cls.listeners[(addr, port)] = listener
+
+        cls.protos[(addr, port)] = proto
+        callback(addr, port)
+
+
 class P2PDataStore(P2PInterface):
     """A P2P data store class.
 

test/functional/test_framework/test_node.py

@@ -71,6 +71,7 @@ class TestNode():
         """
 
         self.index = i
+        self.p2p_conn_index = 1
         self.datadir = datadir
         self.bitcoinconf = os.path.join(self.datadir, "bitcoin.conf")
         self.stdout_dir = os.path.join(self.datadir, "stdout")
@@ -517,7 +518,7 @@ class TestNode():
                 self._raise_assertion_error(assert_msg)
 
     def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, **kwargs):
-        """Add a p2p connection to the node.
+        """Add an inbound p2p connection to the node.
 
         This method adds the p2p connection to the self.p2ps list and also
         returns the connection to the caller."""
@@ -546,6 +547,29 @@ class TestNode():
 
         return p2p_conn
 
+    def add_outbound_p2p_connection(self, p2p_conn, *, p2p_idx, connection_type="outbound-full-relay", **kwargs):
+        """Add an outbound p2p connection from node. Either
+        full-relay("outbound-full-relay") or
+        block-relay-only("block-relay-only") connection.
+
+        This method adds the p2p connection to the self.p2ps list and returns
+        the connection to the caller.
+        """
+
+        def addconnection_callback(address, port):
+            self.log.debug("Connecting to %s:%d %s" % (address, port, connection_type))
+            self.addconnection('%s:%d' % (address, port), connection_type)
+
+        p2p_conn.peer_accept_connection(connect_cb=addconnection_callback, connect_id=p2p_idx + 1, net=self.chain, timeout_factor=self.timeout_factor, **kwargs)()
+
+        p2p_conn.wait_for_connect()
+        self.p2ps.append(p2p_conn)
+
+        p2p_conn.wait_for_verack()
+        p2p_conn.sync_with_ping()
+
+        return p2p_conn
+
     def num_test_p2p_connections(self):
         """Return number of test framework p2p connections to the node."""
         return len([peer for peer in self.getpeerinfo() if peer['subver'] == MY_SUBVERSION.decode("utf-8")])
@@ -555,6 +579,7 @@ class TestNode():
         for p in self.p2ps:
             p.peer_disconnect()
         del self.p2ps[:]
+
         wait_until_helper(lambda: self.num_test_p2p_connections() == 0, timeout_factor=self.timeout_factor)
 
 

test/functional/test_runner.py

@@ -261,6 +261,7 @@ BASE_SCRIPTS = [
     'feature_filelock.py',
     'feature_loadblock.py',
     'p2p_dos_header_tree.py',
+    'p2p_add_connections.py',
     'p2p_unrequested_blocks.py',
     'p2p_blockfilters.py',
     'feature_includeconf.py',