Repository: bitcoin/bitcoin
File: test/functional/p2p_v2_misbehaving.py (Python)

Code before:

    def data_received(self, t):
        if self.test_type == TestType.EARLY_KEY_RESPONSE:
            # check that data can be received on recvbuf only when mismatch from V1_PREFIX happens (magic_sent = True)
            assert self.v2_state.can_data_be_received and self.v2_state.magic_sent
        else:
            super().data_received(t)

    def on_open(self):
        self.connection_opened = True


class EncryptedP2PMisbehaving(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 1
        self.extra_args = [["-v2transport=1", "-peertimeout=3"]]

    def run_test(self):
        self.test_earlykeyresponse()
        self.test_v2disconnection()

    def test_earlykeyresponse(self):
        self.log.info('Sending ellswift bytes in parts to ensure that response from responder is received only when')
        self.log.info('ellswift bytes have a mismatch from the 16 bytes(network magic followed by "version\\x00\\x00\\x00\\x00\\x00")')
        node0 = self.nodes[0]
        self.log.info('Sending first 4 bytes of ellswift which match network magic')
        self.log.info('If a response is received, assertion failure would happen in our custom data_received() function')
        # send happens in `initiate_v2_handshake()` in `connection_made()`
        peer1 = node0.add_p2p_connection(MisbehavingV2Peer(TestType.EARLY_KEY_RESPONSE), wait_for_verack=False, send_version=False, supports_v2_p2p=True)
        self.wait_until(lambda: peer1.connection_opened)
        self.log.info('Sending remaining ellswift and garbage which are different from V1_PREFIX. Since a response is')
        self.log.info('expected now, our custom data_received() function wouldn\'t result in assertion failure')
        ellswift_and_garbage_data = peer1.v2_state.initiate_v2_handshake()
        peer1.send_raw_message(ellswift_and_garbage_data)
        with self.nodes[0].assert_debug_log(['version handshake timeout peer=0']):
            peer1.wait_for_disconnect(timeout=5)
        self.log.info('successful disconnection since modified ellswift was sent as response')

    def test_v2disconnection(self):
        # test v2 disconnection scenarios
        node0 = self.nodes[0]
        expected_debug_message = [
            [],  # EARLY_KEY_RESPONSE
            ["V2 transport error: missing garbage terminator, peer=1"],  # EXCESS_GARBAGE
            ["version handshake timeout peer=2"],  # WRONG_GARBAGE_TERMINATOR
            ["V2 transport error: packet decryption failure"],  # WRONG_GARBAGE
            ["V2 transport error: packet decryption failure"],  # SEND_NO_AAD
            [],  # SEND_NON_EMPTY_VERSION_PACKET
        ]
        for test_type in TestType:
            if test_type == TestType.EARLY_KEY_RESPONSE:

Review comment (suggestion):
do you think it would make sense to use `test_earlykeyresponse` here and then make `test_v2disconnection` into `run_test`

Code after:
        self.log.info('Sending first 4 bytes of ellswift which match network magic')
        self.log.info('If a response is received, assertion failure would happen in our custom data_received() function')
        peer1 = node0.add_p2p_connection(MisbehavingV2Peer(TestType.EARLY_KEY_RESPONSE), wait_for_verack=False, send_version=False, supports_v2_p2p=True, wait_for_v2_handshake=False)
        peer1.send_raw_message(MAGIC_BYTES['regtest'])
        self.log.info('Sending remaining ellswift and garbage which are different from V1_PREFIX. Since a response is')
        self.log.info('expected now, our custom data_received() function wouldn\'t result in assertion failure')
        peer1.v2_state.can_data_be_received = True
        peer1.send_raw_message(peer1.v2_state.ellswift_ours[4:] + peer1.v2_state.sent_garbage)
        with node0.assert_debug_log(['V2 handshake timeout peer=0']):
            peer1.wait_for_disconnect(timeout=5)
        self.log.info('successful disconnection since modified ellswift was sent as response')

    def test_v2disconnection(self):
        # test v2 disconnection scenarios
        node0 = self.nodes[0]
        expected_debug_message = [
            [],  # EARLY_KEY_RESPONSE
            ["V2 transport error: missing garbage terminator, peer=1"],  # EXCESS_GARBAGE
            ["V2 handshake timeout peer=2"],  # WRONG_GARBAGE_TERMINATOR
            ["V2 transport error: packet decryption failure"],  # WRONG_GARBAGE
            ["V2 transport error: packet decryption failure"],  # SEND_NO_AAD
            [],  # SEND_NON_EMPTY_VERSION_PACKET
        ]
        for test_type in TestType:
            if test_type == TestType.EARLY_KEY_RESPONSE:
                continue
            elif test_type == TestType.SEND_NON_EMPTY_VERSION_PACKET:
                node0.add_p2p_connection(MisbehavingV2Peer(test_type), wait_for_verack=True, send_version=True, supports_v2_p2p=True)
                self.log.info(f"No disconnection for {test_type.name}")
            else:
                with node0.assert_debug_log(expected_debug_message[test_type.value], timeout=5):
                    peer = node0.add_p2p_connection(MisbehavingV2Peer(test_type), wait_for_verack=False, send_version=False, supports_v2_p2p=True, expect_success=False)
                    peer.wait_for_disconnect()
                self.log.info(f"Expected disconnection for {test_type.name}")


if __name__ == '__main__':
    EncryptedP2PMisbehaving().main()