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

366@@ -367,8 +367,10 @@ static CAddress GetBindAddress(SOCKET sock)
367     return addr_bind;
368 }
369 
370-CNode* CConnman::ConnectNode(CAddress addrConnect, const char *pszDest, bool fCountFailure, bool manual_connection, bool block_relay_only)
371+CNode* CConnman::ConnectNode(CAddress addrConnect, const char *pszDest, bool fCountFailure, ConnectionType conn_type)

112@@ -113,6 +113,14 @@ struct CSerializedNetMsg
113     std::string command;
114 };
115 
116+enum class ConnectionType {
117+    INBOUND, // peer initiated connections
118+    OUTBOUND, // full relay connections (blocks, addrs, txns)
119+    MANUAL, // connections to addresses added via addnode or the connect command line argument

896@@ -841,6 +897,9 @@ static const CRPCCommand commands[] =
897     { "network",            "setnetworkactive",       &setnetworkactive,       {"state"} },
898     { "network",            "getnodeaddresses",       &getnodeaddresses,       {"count"} },
899     { "hidden",             "addpeeraddress",         &addpeeraddress,         {"address", "port"} },
900+
901+    /* Not shown in help */

64@@ -65,7 +65,7 @@
65     NODE_WITNESS,
66     sha256,
67 )
68-from test_framework.util import wait_until
69+from test_framework.util import wait_until, MAX_NODES, p2p_port

150-        return conn_gen
151+        logger.debug('Connecting to Bitcoin Node: %s:%d' % (self.dstaddr, self.dstport))
152+        coroutine = loop.create_connection(lambda: self, host=self.dstaddr, port=self.dstport)
153+        return lambda: loop.call_soon_threadsafe(loop.create_task, coroutine)
154+
155+    def peer_outbound_connect(self, dstaddr, dstport, connect_cb, connect_id=0, *, net, timeout_factor):

80+        conn = self.nodes[0].add_outbound_p2p_connection(P2PTxInvStore(), connection_type="blockrelay")
81+
82+        assert_equal(self.nodes[0].testmempoolaccept([tx_hex])[0]['allowed'], True)
83+        self.nodes[0].sendrawtransaction(tx_hex)
84+        conn.sync_with_ping()
85+        assert(txid not in conn.get_invs())

85+        assert(txid not in conn.get_invs())
86+
87+    def check_p2p_tx_violation(self, index=1):
88+        self.log.info('Check that txs from P2P are rejected and result in disconnect')
89+        input_txid = self.nodes[0].getblock(self.nodes[0].getblockhash(index), 2)['tx'][0]['txid']
90+        tx = create_transaction(self.nodes[0], input_txid, self.nodes[0].getnewaddress(), amount=50 - 0.001)

301+            HelpExampleCli("addconnection", "\"192.168.0.6:8333\" \"outbound\"")
302+            + HelpExampleRpc("addconnection", "\"192.168.0.6:8333\" \"outbound\"")
303+        },
304+    };
305+
306+    help.Check(request);

895@@ -840,7 +896,10 @@ static const CRPCCommand commands[] =
896     { "network",            "clearbanned",            &clearbanned,            {} },
897     { "network",            "setnetworkactive",       &setnetworkactive,       {"state"} },
898     { "network",            "getnodeaddresses",       &getnodeaddresses,       {"count"} },
899+
900+    /* Not shown in help */

13+from test_framework.messages import msg_tx
14+from test_framework.mininode import P2PInterface, P2PTxInvStore
15 from test_framework.test_framework import BitcoinTestFramework
16 from test_framework.util import assert_equal
17 
18-

133@@ -130,7 +134,7 @@ def __init__(self):
134     def is_connected(self):
135         return self._transport is not None
136 
137-    def peer_connect(self, dstaddr, dstport, *, net, timeout_factor):
138+    def peer_connect_helper(self, dstaddr, dstport, net, timeout_factor):

74+        self.log.info('Tests with node in normal mode with block-relay-only connections')
75+        self.restart_node(0, ["-noblocksonly"])  # disables blocks only mode
76+        assert_equal(self.nodes[0].getnetworkinfo()['localrelay'], True)
77+
78+        # Ensure we disconnect if a block-relay-only connection sends us a transaction
79+        self.nodes[0].add_outbound_p2p_connection(P2PInterface(), connection_type="blockrelay")

547+        full-relay("outbound") or block-relay-only("blockrelay") connection.
548+
549+        This method adds the p2p connection to the self.p2ps list and also
550+        returns the connection to the caller."""
551+
552+        kwargs['dstport'] = p2p_port(self.index)

550@@ -524,6 +551,33 @@ def close(self, timeout=10):
551         # Safe to remove event loop.
552         NetworkThread.network_event_loop = None
553 
554+    @classmethod
555+    def listen(cls, p2p, callback, port=None, addr=None, idx=0):
556+        if port is None:
557+            assert idx >= 0 and idx < MAX_NODES
558+            port = p2p_port(MAX_NODES-idx)

571+
572+    @classmethod
573+    async def create_server(cls, addr, port, callback, proto):
574+        if (addr, port) not in cls.listeners:
575+            listener = await cls.network_event_loop.create_server(cls.peer_listener(addr, port), addr, port)
576+            logger.debug("Listening server on %s:%d should be started" % (addr, port))

293+     *
294+     * @param[in]   address     Address of node to try connecting to
295+     * @param[in]   conn_type   ConnectionType::OUTBOUND or ConnectionType::BLOCK_RELAY
296+     * @return      bool        Returns false if at max connection capacity
297+     */
298+    bool AddConnection(const std::string& address, const ConnectionType conn_type);

316+        vt.nServices = services
317+        vt.addrTo.ip = self.dstaddr
318+        vt.addrTo.port = self.dstport
319+        vt.addrFrom.ip = "0.0.0.0"
320+        vt.addrFrom.port = 0
321+        self.on_connection_send_msg = vt  # Will be sent soon after connection_made

541@@ -542,6 +542,30 @@ def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, **kwargs):
542 
543         return p2p_conn
544 
545+    def add_outbound_p2p_connection(self, p2p_conn, *, connection_type="outbound", connect_id=0, **kwargs):
546+        """ Add an outbound p2p connection from node. Either
547+        full-relay("outbound") or block-relay-only("blockrelay") connection.
548+
549+        This method adds the p2p connection to the self.p2ps list and also
550+        returns the connection to the caller."""

324+    NodeContext& context = EnsureNodeContext(request.context);
325+    if (!context.connman) {
326+        throw JSONRPCError(RPC_CLIENT_P2P_DISABLED, "Error: Peer-to-peer functionality missing or disabled");
327+    }
328+
329+    bool success = context.connman->AddConnection(address, conn_type);

84+        conn = self.nodes[0].add_outbound_p2p_connection(P2PTxInvStore(), connection_type="blockrelay")
85+
86+        # bump time forward to ensure nNextInvSend timer pops
87+        self.nodes[0].setmocktime(int(time.time()) + 5)
88+
89+        assert_equal(self.nodes[0].testmempoolaccept([tx_hex])[0]['allowed'], True)

66             second_peer.wait_for_tx(txid)
67             assert_equal(self.nodes[0].getmempoolinfo()['size'], 1)
68         self.log.info("Forcerelay peer's transaction is accepted and relayed")
69 
70+        self.nodes[0].disconnect_p2ps()
71+        self.nodes[0].generate(1)

89+        assert_equal(self.nodes[0].testmempoolaccept([tx_hex])[0]['allowed'], True)
90+        self.nodes[0].sendrawtransaction(tx_hex)
91+        conn.sync_with_ping()
92+        assert(int(txid, 16) not in conn.get_invs())
93+
94+    def check_p2p_tx_violation(self, index=1):

1074+    int nOutboundFullRelay = 0;
1075+
1076+    {
1077+        LOCK(cs_vNodes);
1078+        for (const CNode* pnode : vNodes) {
1079+            if (pnode->m_conn_type == ConnectionType::OUTBOUND) {

1@@ -2,15 +2,16 @@
2 # Copyright (c) 2019 The Bitcoin Core developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
5-"""Test p2p blocksonly"""
6+"""Test p2p blocksonly mode & block-relay-only connections."""

24+        for i in range(8):
25+            self.log.info("outbound: {}".format(i))
26+            self.nodes[0].add_outbound_p2p_connection(P2PInterface())
27+
28+        self.log.info("Add 2 block-relay-only connections to node 0")
29+        for i in range(2):

296@@ -296,6 +297,49 @@ static RPCHelpMan addnode()
297     };
298 }
299 
300+static RPCHelpMan addconnection()
301+{
302+    return RPCHelpMan{"addconnection",
303+        "\nOpen an outbound connection to a specified node (test only)",

312+        RPCExamples{
313+            HelpExampleCli("addconnection", "\"192.168.0.6:8333\"")
314+            + HelpExampleRpc("addconnection", "\"192.168.0.6:8333\"")
315+        },
316+        [&](const RPCHelpMan& self, const JSONRPCRequest& request) -> UniValue
317+

318+{
319+    if (!Params().IsMockableChain()) {
320+        throw std::runtime_error("addconnection is for regression testing (-regtest mode) only");
321+    }
322+
323+    RPCTypeCheckArgument(request.params, UniValue::VSTR);

327+        throw JSONRPCError(RPC_CLIENT_P2P_DISABLED, "Error: Peer-to-peer functionality missing or disabled");
328+    }
329+
330+    const bool success = context.connman->AddConnection(address);
331+    if (!success) {
332+        throw JSONRPCError(RPC_CLIENT_NODE_CAPACITY_REACHED, "Error: Max number of outbound full relay connections already open.");

1079@@ -1080,6 +1080,29 @@ void CConnman::AcceptConnection(const ListenSocket& hListenSocket) {
1080     RandAddEvent((uint32_t)id);
1081 }
1082 
1083+bool CConnman::AddConnection(const std::string& address)
1084+{
1085+    int nOutboundFullRelay = 0;

1097+
1098+    CSemaphoreGrant grant(*semOutbound, true);
1099+    if (!grant) return false;
1100+
1101+    CAddress addr{};
1102+    OpenNetworkConnection(addr, false, &grant, address.c_str(), ConnectionType::OUTBOUND_FULL_RELAY);

1084+bool CConnman::AddConnection(const std::string& address, const ConnectionType conn_type)
1085 {
1086+    assert(conn_type == ConnectionType::OUTBOUND_FULL_RELAY || conn_type == ConnectionType::BLOCK_RELAY);
1087+
1088     int nOutboundFullRelay = 0;
1089+    int nOutboundBlockRelay = 0;

302@@ -303,15 +303,17 @@ static RPCHelpMan addconnection()
303         "\nOpen an outbound connection to a specified node (test only)",
304         {
305             {"address", RPCArg::Type::STR, RPCArg::Optional::NO, "The IP address and port to attempt connecting to."},
306+            {"connectiontype", RPCArg::Type::STR, RPCArg::Optional::NO, "'outbound' or 'blockrelay'"},

342 
343-    const bool success = context.connman->AddConnection(address);
344+    const bool success = context.connman->AddConnection(address, conn_type);
345     if (!success) {
346-        throw JSONRPCError(RPC_CLIENT_NODE_CAPACITY_REACHED, "Error: Max number of outbound full relay connections already open.");
347+        throw JSONRPCError(RPC_CLIENT_NODE_CAPACITY_REACHED, "Error: Already at capacity for specified connection type.");

1177+
1178+    // max total outbound connections already exist
1179+    CSemaphoreGrant grant(*semOutbound, true);
1180+    if (!grant) return false;
1181+
1182+    OpenNetworkConnection(CAddress(), false, &grant, address.c_str(), conn_type);

1171+        }
1172+    }
1173+
1174+    // max connections of specified type already exist
1175+    if (conn_type == ConnectionType::OUTBOUND_FULL_RELAY && outbound_full_relay >= m_max_outbound_full_relay) return false;
1176+    if (conn_type == ConnectionType::BLOCK_RELAY && outbound_block_relay >= m_max_outbound_block_relay) return false;

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

90+        self.nodes[0].setmocktime(int(time.time()) + 5)
91+
92+        assert_equal(self.nodes[0].testmempoolaccept([tx_hex])[0]['allowed'], True)
93+        self.nodes[0].sendrawtransaction(tx_hex)
94+        conn.sync_with_ping()
95+        assert(int(txid, 16) not in conn.get_invs())

1174+    // max connections of specified type already exist
1175+    if (conn_type == ConnectionType::OUTBOUND_FULL_RELAY && outbound_full_relay >= m_max_outbound_full_relay) return false;
1176+    if (conn_type == ConnectionType::BLOCK_RELAY && outbound_block_relay >= m_max_outbound_block_relay) return false;
1177+
1178+    // max total outbound connections already exist
1179+    CSemaphoreGrant grant(*semOutbound, true);

345@@ -346,6 +346,15 @@ class CConnman
346     bool RemoveAddedNode(const std::string& node);
347     std::vector<AddedNodeInfo> GetAddedNodeInfo();
348 
349+    /**
350+     * Attempts to open a connection. Currently only used from tests.
351+     *
352+     * @param[in]   address     Address of node to try connecting to
353+     * @param[in]   conn_type   ConnectionType::OUTBOUND or ConnectionType::BLOCK_RELAY
354+     * @return      bool        Returns false if at max connection capacity

327+{
328+    return RPCHelpMan{"addconnection",
329+        "\nOpen an outbound connection to a specified node. This RPC is for testing only.\n",
330+        {
331+            {"address", RPCArg::Type::STR, RPCArg::Optional::NO, "The IP address and port to attempt connecting to."},
332+            {"connectiontype", RPCArg::Type::STR, RPCArg::Optional::NO, "'outbound' or 'blockrelay'."},

333+        },
334+        RPCResult{
335+            RPCResult::Type::OBJ, "", "",
336+            {
337+                { RPCResult::Type::STR, "address", "Address of newly added connection." },
338+                { RPCResult::Type::STR, "connectiontype", "Type of connection opened." },

347+        throw std::runtime_error("addconnection is for regression testing (-regtest mode) only.");
348+    }
349+
350+    RPCTypeCheck(request.params, {UniValue::VSTR, UniValue::VSTR});
351+    const std::string address = request.params[0].get_str();
352+    const std::string conn_type_in = request.params[1].get_str();

369+        throw JSONRPCError(RPC_CLIENT_NODE_CAPACITY_REACHED, "Error: Already at capacity for specified connection type.");
370+    }
371+
372+    UniValue info(UniValue::VOBJ);
373+    info.pushKV("address", address);
374+    info.pushKV("connectiontype", conn_type_in);

370+    }
371+
372+    UniValue info(UniValue::VOBJ);
373+    info.pushKV("address", address);
374+    info.pushKV("connectiontype", conn_type_in);
375+

961@@ -906,7 +962,9 @@ static const CRPCCommand commands[] =
962     { "network",            "clearbanned",            &clearbanned,            {} },
963     { "network",            "setnetworkactive",       &setnetworkactive,       {"state"} },
964     { "network",            "getnodeaddresses",       &getnodeaddresses,       {"count"} },
965+
966     { "hidden",             "addpeeraddress",         &addpeeraddress,         {"address", "port"} },
967+    { "hidden",             "addconnection",          &addconnection,          {"address", "connectiontype"} },

1155@@ -1156,6 +1156,30 @@ void CConnman::AcceptConnection(const ListenSocket& hListenSocket) {
1156     RandAddEvent((uint32_t)id);
1157 }
1158 
1159+bool CConnman::AddConnection(const std::string& address, const ConnectionType conn_type)

21+        self.disconnect_nodes(0, 1)
22+
23+        self.log.info("Add 8 outbounds to node 0")
24+        for i in range(8):
25+            self.log.info("outbound: {}".format(i))
26+            self.nodes[0].add_outbound_p2p_connection(P2PInterface())

31+            self.nodes[0].add_outbound_p2p_connection(P2PInterface(), connection_type="blockrelay")
32+
33+        self.log.info("Add 2 block-relay-only connections to node 1")
34+        for i in range(2):
35+            self.log.info("block-relay-only: {}".format(i))
36+            self.nodes[1].add_outbound_p2p_connection(P2PInterface(), connection_type="blockrelay")

51+
52+        # check node0 has all outbound connections
53+        node0_peers = []
54+        for i in range(10):
55+            node0_peers.append(self.nodes[0].getpeerinfo()[i]['inbound'])
56+        assert_equal(node0_peers.count(False), 10)

59+        node1_peers = []
60+        for i in range(15):
61+            node1_peers.append(self.nodes[1].getpeerinfo()[i]['inbound'])
62+
63+        assert_equal(node1_peers.count(False), 10)
64+        assert_equal(node1_peers.count(True), 5)

85+            self.nodes[0].add_outbound_p2p_connection(P2PInterface())
86+
87+        self.log.info("Add 2 block-relay-only connections to node 0")
88+        for i in range(2):
89+            self.log.info("block-relay-only: {}".format(i))
90+            self.nodes[0].add_outbound_p2p_connection(P2PInterface(), connection_type="blockrelay")

50+            self.log.info("outbound: {}".format(i))
51+            self.nodes[1].add_outbound_p2p_connection(P2PInterface())
52+
53+        self.log.info("Check the connections opened as expected")
54+        self.check_node_connections(0, 0, 10)
55+        self.check_node_connections(1, 5, 10)

26+        self.disconnect_nodes(0, 1)
27+
28+        self.log.info("Add 8 outbounds to node 0")
29+        for i in range(8):
30+            self.log.info("outbound: {}".format(i))
31+            self.nodes[0].add_outbound_p2p_connection(P2PInterface(), connection_type="outbound")

15+        self.num_nodes = 2
16+
17+    def skip_test_if_missing_module(self):
18+        self.skip_if_no_wallet()
19+
20+    def check_node_connections(self, node=None, num_in=None, num_out=None):