Merge #2410: [Test] Add p2p invalid messages functional test (and tes…

…t framework update)

7b04c0a [Test] Clean duplicate connections creation in mining_pos_coldStaking.py (furszy)
15a799e [Test] MAX_PROTOCOL_MESSAGE_LENGTH PIVXified. (furszy)
0aedf35 [tests] Don't import asyncio to test magic bytes (John Newbery)
9bb5309 Refactor resource exhaustion test (Troy Giorshev)
589a780 Fix "invalid message size" test (Troy Giorshev)
8a0c12b Move size limits to module-global (Troy Giorshev)
b3391c5 Remove two unneeded tests (Troy Giorshev)
1404e68 test: Add various low-level p2p tests (furszy)
8aaf7e1 test: Remove fragile assert_memory_usage_stable (MarcoFalke)
f68e22c [Test] p2p_invalid_messages.py do not change msg.command in test_command and raise sync_with_ping timeout (furszy)
c851c0b Test framework: Wait for verack by default on every new connection (furszy)
c02e9a0 [Test] move framework subversion to string (furszy)
3472a39 Replace coroutine with async def in p2p_invalid_messages.py (Elichai Turkel)
33c7b19 net_processing: align msg checksum check properly. (furszy)
7f71b1b Hash P2P messages as they are received instead of at process-time (furszy)
215a638 test: Skip flaky p2p_invalid_messages test on macOS (Fabian Jahr)
c11f565 qa: Make swap_magic_bytes in p2p_invalid_messages atomic (MarcoFalke)
be979ad test: Fix race in p2p_invalid_messages (MarcoFalke)
6a72f0c qa: Add tests for invalid message headers (MarcoFalke)
51ddd3d Introduce p2p_invalid_messages.py test (furszy)
55a37b5 net: fix missing jump line in "Oversized message from peer.." error. (furszy)
0edfce5 test_node: get_mem_rss fixups (MarcoFalke)
6f21213 tests: add P2PConnection.send_raw_message (James O'Beirne)
ae68c6e tests: add utility to assert node memory usage hasn't increased (James O'Beirne)
8469afe test: forward timeouts properly in send_blocks_and_test (James O'Beirne)
db28a53 Skip is_closing() check when not available. (Daniel Kraft)
be9dacb tests: fixes mininode's P2PConnection sending messages on closing transport (marcoagner)
53599c8 qa: Avoid start/stop of the network thread mid-test (furszy)
688190c [qa] mininode: Expose connection state through is_connected (MarcoFalke)

Pull request description:

  Part of the deep and long net and ser work that I'm doing (and Tor v3 onion addresses support). Friend of #2359.

  Focused on the end goal of implementing the `p2p_invalid_messages` functional test which validates that invalid msg headers, msg types, oversized payloads and inventory msgs aren't accepted nor can cause a resource exhaustion. And an extra covered scenario, in `p2p_invalid_tx.py`, for the orphan pool overflow.

  Plus, to get up to the goal, had to work over the functional test framework as well.

  So.. adapted list:

  * bitcoin#9045.
  * bitcoin#13512.
  * bitcoin#13517.
  * bitcoin#13715.
  * bitcoin#13747.
  * bitcoin#14456.
  * bitcoin#14522.
  * bitcoin#14672.
  * bitcoin#14693.
  * bitcoin#14812.
  * bitcoin#15246.
  * bitcoin#15330.
  * bitcoin#15697.
  * bitcoin#16445.
  * bitcoin#17931.
  * bitcoin#17469.
  * bitcoin#18628 (only `p2p_invalid_tx.py` and `p2p_invalid_messages.py`. We don't support the other tests yet).
  * bitcoin#19177.
  * bitcoin#19264.

ACKs for top commit:
  random-zebra:
    utACK 7b04c0a and merging...

Tree-SHA512: 48d1f1a6acc24a9abce033fbf1f281ba4392147da39d118a694a09d63c3e0610cc1a29d711b434b16cc0d0e030dd9f1a89843870091b6999b862b9ab18a20679

src/net.cpp

@@ -706,7 +706,7 @@ bool CNode::ReceiveMsgBytes(const char* pch, unsigned int nBytes, bool& complete
             return false;
 
         if (msg.in_data && msg.hdr.nMessageSize > MAX_PROTOCOL_MESSAGE_LENGTH) {
-            LogPrint(BCLog::NET, "Oversized message from peer=%i, disconnecting", GetId());
+            LogPrint(BCLog::NET, "Oversized message from peer=%i, disconnecting\n", GetId());
             return false;
         }
 
@@ -797,12 +797,20 @@ int CNetMessage::readData(const char* pch, unsigned int nBytes)
         vRecv.resize(std::min(hdr.nMessageSize, nDataPos + nCopy + 256 * 1024));
     }
 
+    hasher.Write((const unsigned char*)pch, nCopy);
     memcpy(&vRecv[nDataPos], pch, nCopy);
     nDataPos += nCopy;
 
     return nCopy;
 }
 
+const uint256& CNetMessage::GetMessageHash() const
+{
+    assert(complete());
+    if (data_hash.IsNull())
+        hasher.Finalize(data_hash.begin());
+    return data_hash;
+}
 
 // requires LOCK(cs_vSend)
 size_t CConnman::SocketSendData(CNode* pnode)

src/net.h

@@ -486,6 +486,9 @@ class CNodeStats
 
 class CNetMessage
 {
+private:
+    mutable CHash256 hasher;
+    mutable uint256 data_hash;
 public:
     bool in_data; // parsing header (false) or data (true)
 
@@ -513,6 +516,8 @@ class CNetMessage
         return (hdr.nMessageSize == nDataPos);
     }
 
+    const uint256& GetMessageHash() const;
+
     void SetVersion(int nVersionIn)
     {
         hdrbuf.SetVersion(nVersionIn);

src/net_processing.cpp

@@ -1352,7 +1352,7 @@ bool static ProcessMessage(CNode* pfrom, std::string strCommand, CDataStream& vR
         if (vInv.size() > MAX_INV_SZ) {
             LOCK(cs_main);
             Misbehaving(pfrom->GetId(), 20);
-            return error("message inv size() = %u", vInv.size());
+            return error("peer=%d message inv size() = %u", pfrom->GetId(), vInv.size());
         }
 
         LOCK(cs_main);
@@ -1402,7 +1402,7 @@ bool static ProcessMessage(CNode* pfrom, std::string strCommand, CDataStream& vR
         if (vInv.size() > MAX_INV_SZ) {
             LOCK(cs_main);
             Misbehaving(pfrom->GetId(), 20);
-            return error("message getdata size() = %u", vInv.size());
+            return error("peer=%d message getdata size() = %u", pfrom->GetId(), vInv.size());
         }
 
         if (vInv.size() != 1)
@@ -2021,25 +2021,25 @@ bool ProcessMessages(CNode* pfrom, CConnman& connman, std::atomic<bool>& interru
     // Read header
     CMessageHeader& hdr = msg.hdr;
     if (!hdr.IsValid(Params().MessageStart())) {
-        LogPrint(BCLog::NET, "PROCESSMESSAGE: ERRORS IN HEADER %s peer=%d\n", SanitizeString(hdr.GetCommand()), pfrom->id);
+        LogPrint(BCLog::NET, "PROCESSMESSAGE: ERRORS IN HEADER '%s' peer=%d\n", SanitizeString(hdr.GetCommand()), pfrom->id);
         return fMoreWork;
     }
     std::string strCommand = hdr.GetCommand();
 
     // Message size
     unsigned int nMessageSize = hdr.nMessageSize;
 
-        // Checksum
-        CDataStream& vRecv = msg.vRecv;
-        uint256 hash = Hash(vRecv.begin(), vRecv.begin() + nMessageSize);
-        if (memcmp(hash.begin(), hdr.pchChecksum, CMessageHeader::CHECKSUM_SIZE) != 0)
-        {
-            LogPrint(BCLog::NET, "%s(%s, %u bytes): CHECKSUM ERROR expected %s was %s\n", __func__,
-               SanitizeString(strCommand), nMessageSize,
-               HexStr(hash.begin(), hash.begin()+CMessageHeader::CHECKSUM_SIZE),
-               HexStr(hdr.pchChecksum, hdr.pchChecksum+CMessageHeader::CHECKSUM_SIZE));
-            return fMoreWork;
-        }
+    // Checksum
+    CDataStream& vRecv = msg.vRecv;
+    uint256 hash = msg.GetMessageHash();
+    if (memcmp(hash.begin(), hdr.pchChecksum, CMessageHeader::CHECKSUM_SIZE) != 0)
+    {
+        LogPrint(BCLog::NET, "%s(%s, %u bytes): CHECKSUM ERROR expected %s was %s\n", __func__,
+           SanitizeString(strCommand), nMessageSize,
+           HexStr(hash.begin(), hash.begin()+CMessageHeader::CHECKSUM_SIZE),
+           HexStr(hdr.pchChecksum, hdr.pchChecksum+CMessageHeader::CHECKSUM_SIZE));
+        return fMoreWork;
+    }
 
     // Process message
     bool fRet = false;

test/functional/example_test.py

@@ -21,8 +21,6 @@
     mininode_lock,
     msg_block,
     msg_getdata,
-    network_thread_join,
-    network_thread_start,
 )
 from test_framework.test_framework import PivxTestFramework
 from test_framework.util import (
@@ -135,9 +133,6 @@ def run_test(self):
         # Create P2P connections to two of the nodes
         self.nodes[0].add_p2p_connection(BaseNode())
 
-        # Start up network handling in another thread. This needs to be called
-        # after the P2P connections have been created.
-        network_thread_start()
         # wait_for_verack ensures that the P2P connection is fully up.
         self.nodes[0].p2p.wait_for_verack()
 
@@ -189,14 +184,9 @@ def run_test(self):
         connect_nodes(self.nodes[1], 2)
 
         self.log.info("Add P2P connection to node2")
-        # We can't add additional P2P connections once the network thread has started. Disconnect the connection
-        # to node0, wait for the network thread to terminate, then connect to node2. This is specific to
-        # the current implementation of the network thread and may be improved in future.
         self.nodes[0].disconnect_p2ps()
-        network_thread_join()
 
         self.nodes[2].add_p2p_connection(BaseNode())
-        network_thread_start()
         self.nodes[2].p2p.wait_for_verack()
 
         self.log.info("Wait for node2 reach current tip. Test that it has propagated all the blocks to us")

test/functional/feature_block.py

@@ -21,7 +21,7 @@
     uint256_from_compact,
     uint256_from_str,
 )
-from test_framework.mininode import P2PDataStore, network_thread_start, network_thread_join
+from test_framework.mininode import P2PDataStore
 from test_framework.script import (
     CScript,
     MAX_SCRIPT_ELEMENT_SIZE,
@@ -84,7 +84,6 @@ def run_test(self):
         node = self.nodes[0]  # convenience reference to the node
         # reconnect_p2p() expects the network thread to be running
         self.log.info("Starting network thread...")
-        network_thread_start()
         self.reconnect_p2p()
 
         self.block_heights = {}
@@ -1207,10 +1206,8 @@ def reconnect_p2p(self):
         The node gets disconnected several times in this test. This helper
         method reconnects the p2p and restarts the network thread."""
 
-        network_thread_join()
         self.nodes[0].disconnect_p2ps()
         self.nodes[0].add_p2p_connection(P2PDataStore())
-        network_thread_start()
         self.nodes[0].p2p.wait_for_verack()
 
     def send_blocks(self, blocks, success=True, reject_reason=None, reconnect=False, timeout=60):

test/functional/feature_cltv.py

@@ -67,10 +67,6 @@ def set_test_params(self):
 
     def run_test(self):
         self.nodes[0].add_p2p_connection(P2PInterface())
-
-        network_thread_start()
-
-        # wait_for_verack ensures that the P2P connection is fully up.
         self.nodes[0].p2p.wait_for_verack()
 
         self.log.info("Mining %d blocks", CLTV_HEIGHT - 2)

test/functional/feature_nulldummy.py

@@ -15,7 +15,7 @@
 
 from test_framework.test_framework import PivxTestFramework
 from test_framework.util import *
-from test_framework.mininode import CTransaction, network_thread_start
+from test_framework.messages import CTransaction
 from test_framework.blocktools import create_coinbase, create_block, add_witness_commitment
 from test_framework.script import CScript
 from io import BytesIO
@@ -48,7 +48,6 @@ def run_test(self):
         self.address = self.nodes[0].getnewaddress()
         self.ms_address = self.nodes[0].addmultisigaddress(1,[self.address])
 
-        network_thread_start()
         self.coinbase_blocks = self.nodes[0].generate(2) # Block 2
         coinbase_txid = []
         for i in self.coinbase_blocks:

test/functional/mining_pos_coldStaking.py

@@ -8,15 +8,12 @@
 from time import sleep
 
 from test_framework.messages import CTransaction, CTxIn, CTxOut, COIN, COutPoint
-from test_framework.mininode import network_thread_start
-from test_framework.pivx_node import PivxTestNode
 from test_framework.script import CScript, OP_CHECKSIG
 from test_framework.test_framework import PivxTestFramework
 from test_framework.util import (
     assert_equal,
     assert_greater_than,
     assert_raises_rpc_error,
-    p2p_port,
     bytes_to_hex_str,
     set_node_times,
 )
@@ -34,30 +31,8 @@ def set_test_params(self):
         self.num_nodes = 3
         self.extra_args = [['-nuparams=v5_shield:201']] * self.num_nodes
         self.extra_args[0].append('-sporkkey=932HEevBSujW2ud7RfB1YF91AFygbBRQj3de3LyaCRqNzKKgWXi')
-
-    def setup_chain(self):
-        # Start with PoW cache: 200 blocks
-        self.log.info("Initializing test directory " + self.options.tmpdir)
-        self._initialize_chain()
         self.enable_mocktime()
 
-    def init_test(self):
-        title = "*** Starting %s ***" % self.__class__.__name__
-        underline = "-" * len(title)
-        self.log.info("\n\n%s\n%s\n%s\n", title, underline, self.description)
-        self.DEFAULT_FEE = 0.05
-        # Setup the p2p connections and start up the network thread.
-        self.test_nodes = []
-        for i in range(self.num_nodes):
-            self.test_nodes.append(PivxTestNode())
-            self.test_nodes[i].peer_connect('127.0.0.1', p2p_port(i))
-
-        network_thread_start()  # Start up network handling in another thread
-
-        # Let the test nodes get in sync
-        for i in range(self.num_nodes):
-            self.test_nodes[i].wait_for_verack()
-
     def setColdStakingEnforcement(self, fEnable=True):
         sporkName = "SPORK_19_COLDSTAKING_MAINTENANCE"
         # update spork 19 with node[0]
@@ -77,11 +52,12 @@ def isColdStakingEnforced(self):
         # verify from node[1]
         return not self.is_spork_active(1, "SPORK_19_COLDSTAKING_MAINTENANCE")
 
-
-
     def run_test(self):
         self.description = "Performs tests on the Cold Staking P2CS implementation"
-        self.init_test()
+        title = "*** Starting %s ***" % self.__class__.__name__
+        underline = "-" * len(title)
+        self.log.info("\n\n%s\n%s\n%s\n", title, underline, self.description)
+        self.DEFAULT_FEE = 0.05
         NUM_OF_INPUTS = 20
         INPUT_VALUE = 249
 

test/functional/p2p_feefilter.py

@@ -47,9 +47,8 @@ def run_test(self):
         node1.generate(1)
         self.sync_blocks()
 
-        # Setup the p2p connections and start up the network thread.
-        self.nodes[0].add_p2p_connection(TestNode())
-        network_thread_start()
+        # Setup the p2p connections
+        self.nodes[0].add_p2p_connection(TestP2PConn())
         self.nodes[0].p2p.wait_for_verack()
 
         # Test that invs are received for all txs at feerate of 20 sat/byte

test/functional/p2p_fingerprint.py

@@ -18,7 +18,6 @@
     msg_block,
     msg_getdata,
     msg_getheaders,
-    network_thread_start,
     wait_until,
 )
 from test_framework.test_framework import PivxTestFramework
@@ -76,8 +75,6 @@ def last_header_equals(self, expected_hash, node):
     # last month but that have over a month's worth of work are also withheld.
     def run_test(self):
         node0 = self.nodes[0].add_p2p_connection(P2PInterface())
-
-        network_thread_start()
         node0.wait_for_verack()
 
         # Set node time to 60 days ago

test/functional/p2p_invalid_block.py

@@ -14,7 +14,7 @@
 
 from test_framework.blocktools import create_block, create_coinbase, create_transaction
 from test_framework.messages import COIN
-from test_framework.mininode import network_thread_start, P2PDataStore
+from test_framework.mininode import P2PDataStore
 from test_framework.script import (
     CScript,
     OP_TRUE,
@@ -35,7 +35,6 @@ def run_test(self):
         # Add p2p connection to node0
         node = self.nodes[0]  # convenience reference to the node
         node.add_p2p_connection(P2PDataStore())
-        network_thread_start()
         node.p2p.wait_for_verack()
 
         best_block = node.getblock(node.getbestblockhash())