Merge bitcoin/bitcoin#24748: test/BIP324: functional tests for v2 P2P encryption

bc9283c4415a932ec1eeb70ca2aa4399c80437b3 [test] Add functional test to test early key response behaviour in BIP 324 (stratospher)
ffe6a56d75c0b47d0729e4e0b7225a827b43ad89 [test] Check whether v2 TestNode performs downgrading (stratospher)
ba737358a37438c18f0fba723eab10ccfd9aae9b [test] Add functional tests to test v2 P2P behaviour (stratospher)
4115cf995647d1a513caecb54a4ff3f51927aa8e [test] Ignore BIP324 decoy messages (stratospher)
8c054aa04d33b247744b3747cd5bf3005a013e90 [test] Allow inbound and outbound connections supporting v2 P2P protocol (stratospher)
382894c3acd2dbf3e4198814f547c75b6fb17706  [test] Reconnect using v1 P2P when v2 P2P terminates due to magic byte mismatch (stratospher)
a94e350ac0e5b65ef23a84b05fb10d1204c98c97 [test] Build v2 P2P messages (stratospher)
bb7bffed799dc5ad8b606768164fce46d4cbf9d0 [test] Use lock for sending P2P messages in test framework (stratospher)
5b91fb14aba7d7fe45c9ac364526815bec742356 [test] Read v2 P2P messages (stratospher)
05bddb20f5cc9036fd680500bde8ece70dbf0646 [test] Perform initial v2 handshake (stratospher)
a049d1bd08c8cdb3b693520f24f8a82572dcaab1 [test] Introduce EncryptedP2PState object in P2PConnection (stratospher)
b89fa59e715a185d9fa7fce089dad4273d3b1532 [test] Construct class to handle v2 P2P protocol functions (stratospher)
8d6c848a48530893ca40be5c1285541b3e7a94f3 [test] Move MAGIC_BYTES to messages.py (stratospher)
595ad4b16880ae1f23463ca9985381c8eae945d8 [test/crypto] Add ECDH (stratospher)
4487b8051797173c7ab432e75efa370afb03b529 [rpc/net] Allow v2 p2p support in addconnection (stratospher)

Pull request description:

  This PR introduces support for v2 P2P encryption(BIP 324) in the existing functional test framework and adds functional tests for the same.

  ### commits overview
  1. introduces a new class `EncryptedP2PState` to store the keys, functions for performing the initial v2 handshake and encryption/decryption.
  3. this class is used by `P2PConnection` in inbound/outbound connections to perform the initial v2 handshake before the v1 version handshake. Only after the initial v2 handshake is performed do application layer P2P messages(version, verack etc..) get exchanged. (in a v2 connection)
      - `v2_state` is the object of class `EncryptedP2PState` in `P2PConnection` used to store its keys, session-id etc.
      - a node [advertising](https://github.com/stratospher/blogosphere/blob/main/integration_test_bip324.md#advertising-to-support-v2-p2p) support for  v2 P2P is different from a node actually [supporting v2 P2P](https://github.com/stratospher/blogosphere/blob/main/integration_test_bip324.md#supporting-v2-p2p) (differ when false advertisement of services occur)
          - introduce a boolean variable `supports_v2_p2p` in `P2PConnection` to denote if it supports v2 P2P.
          - introduce a boolean variable `advertises_v2_p2p` to denote whether `P2PConnection` which mimics peer behaviour advertises V2 P2P support. Default option is `False`.
      - In the test framework, you can create Inbound and Outbound connections to `TestNode`
          1. During **Inbound Connections**, `P2PConnection` is the initiator [`TestNode` <--------- `P2PConnection`]
              - Case 1:
                  - if the `TestNode` advertises/signals v2 P2P support (means `self.nodes[i]` set up with `"-v2transport=1"`), different behaviour will be exhibited based on whether:
                      1. `P2PConnection` supports v2 P2P
                      2. `P2PConnection` does not support v2 P2P
                 - In a real world scenario, the initiator node would intrinsically know if they support v2 P2P based on whatever code they choose to run. However, in the test scenario where we mimic peer behaviour, we have no way of knowing if `P2PConnection` should support v2 P2P or not. So `supports_v2_p2p` boolean variable is used as an option to enable support for v2 P2P in `P2PConnection`.
                - Since the `TestNode` advertises v2 P2P support (using "-v2transport=1"), our initiator `P2PConnection` would send:
                  1. (if the `P2PConnection` supports v2 P2P) ellswift + garbage bytes to initiate the connection
                  2. (if the `P2PConnection` does not support v2 P2P) version message to initiate the connection
             - Case 2:
                  - if the `TestNode` doesn't signal v2 P2P support; `P2PConnection` being the initiator would send version message to initiate a connection.
         2. During **Outbound Connections** [TestNode --------> P2PConnection]
             - initiator `TestNode` would send:
                  - (if the `P2PConnection` advertises v2 P2P) ellswift + garbage bytes to initiate the connection
                  - (if the `P2PConnection` advertises v2 P2P) version message to initiate the connection
            - Suppose `P2PConnection` advertises v2 P2P support when it actually doesn't support v2 P2P (false advertisement scenario)
                 - `TestNode` sends ellswift + garbage bytes
                 - `P2PConnection` receives but can't process it and disconnects.
                 - `TestNode` then tries using v1 P2P and sends version message
                 - `P2PConnection` receives/processes this successfully and they communicate on v1 P2P

  4. the encrypted P2P messages follow a different format - 3 byte length + 1-13 byte message_type + payload + 16 byte MAC
  5. includes support for testing decoy messages and v2 connection downgrade(using false advertisement - when a v2 node makes an outbound connection to a node which doesn't support v2 but is advertised as v2 by some malicious
  intermediary)

  ### run the tests
  * functional test - `test/functional/p2p_v2_encrypted.py` `test/functional/p2p_v2_earlykeyresponse.py`

  I'm also super grateful to @ dhruv for his really valuable feedback on this branch.
  Also written a more elaborate explanation here - https://github.com/stratospher/blogosphere/blob/main/integration_test_bip324.md

ACKs for top commit:
  naumenkogs:
    ACK bc9283c4415a932ec1eeb70ca2aa4399c80437b3
  mzumsande:
    Code Review ACK bc9283c4415a932ec1eeb70ca2aa4399c80437b3
  theStack:
    Code-review ACK bc9283c4415a932ec1eeb70ca2aa4399c80437b3
  glozow:
    ACK bc9283c4415a932ec1eeb70ca2aa4399c80437b3

Tree-SHA512: 9b54ed27e925e1775e0e0d35e959cdbf2a9a1aab7bcf5d027e66f8b59780bdd0458a7a4311ddc7dd67657a4a2a2cd5034ead75524420d58a83f642a8304c9811

4 files changed, 510 insertions, 54 deletions

diff --git a/test/functional/test_framework/messages.py b/test/functional/test_framework/messages.py
index d008cb39aa..cc30424653 100755
--- a/test/functional/test_framework/messages.py
+++ b/test/functional/test_framework/messages.py
@@ -75,6 +75,13 @@ MAX_OP_RETURN_RELAY = 83
 
 DEFAULT_MEMPOOL_EXPIRY_HOURS = 336  # hours
 
+MAGIC_BYTES = {
+    "mainnet": b"\xf9\xbe\xb4\xd9",   # mainnet
+    "testnet3": b"\x0b\x11\x09\x07",  # testnet3
+    "regtest": b"\xfa\xbf\xb5\xda",   # regtest
+    "signet": b"\x0a\x03\xcf\x40",    # signet
+}
+
 def sha256(s):
     return hashlib.sha256(s).digest()
 
diff --git a/test/functional/test_framework/p2p.py b/test/functional/test_framework/p2p.py
index 34fe467d23..c113a4c8d8 100755
--- a/test/functional/test_framework/p2p.py
+++ b/test/functional/test_framework/p2p.py
@@ -73,6 +73,7 @@ from test_framework.messages import (
     msg_wtxidrelay,
     NODE_NETWORK,
     NODE_WITNESS,
+    MAGIC_BYTES,
     sha256,
 )
 from test_framework.util import (
@@ -80,6 +81,11 @@ from test_framework.util import (
     p2p_port,
     wait_until_helper_internal,
 )
+from test_framework.v2_p2p import (
+    EncryptedP2PState,
+    MSGTYPE_TO_SHORTID,
+    SHORTID,
+)
 
 logger = logging.getLogger("TestFramework.p2p")
 
@@ -141,13 +147,6 @@ MESSAGEMAP = {
     b"wtxidrelay": msg_wtxidrelay,
 }
 
-MAGIC_BYTES = {
-    "mainnet": b"\xf9\xbe\xb4\xd9",   # mainnet
-    "testnet3": b"\x0b\x11\x09\x07",  # testnet3
-    "regtest": b"\xfa\xbf\xb5\xda",   # regtest
-    "signet": b"\x0a\x03\xcf\x40",    # signet
-}
-
 
 class P2PConnection(asyncio.Protocol):
     """A low-level connection object to a node's P2P interface.
@@ -166,11 +165,20 @@ class P2PConnection(asyncio.Protocol):
         # The underlying transport of the connection.
         # Should only call methods on this from the NetworkThread, c.f. call_soon_threadsafe
         self._transport = None
+        # This lock is acquired before sending messages over the socket. There's an implied lock order and
+        # p2p_lock must not be acquired after _send_lock as it could result in deadlocks.
+        self._send_lock = threading.Lock()
+        self.v2_state = None  # EncryptedP2PState object needed for v2 p2p connections
+        self.reconnect = False  # set if reconnection needs to happen
 
     @property
     def is_connected(self):
         return self._transport is not None
 
+    @property
+    def supports_v2_p2p(self):
+        return self.v2_state is not None
+
     def peer_connect_helper(self, dstaddr, dstport, net, timeout_factor):
         assert not self.is_connected
         self.timeout_factor = timeout_factor
@@ -181,16 +189,21 @@ class P2PConnection(asyncio.Protocol):
         self.recvbuf = b""
         self.magic_bytes = MAGIC_BYTES[net]
 
-    def peer_connect(self, dstaddr, dstport, *, net, timeout_factor):
+    def peer_connect(self, dstaddr, dstport, *, net, timeout_factor, supports_v2_p2p):
         self.peer_connect_helper(dstaddr, dstport, net, timeout_factor)
+        if supports_v2_p2p:
+            self.v2_state = EncryptedP2PState(initiating=True, net=net)
 
         loop = NetworkThread.network_event_loop
         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):
+    def peer_accept_connection(self, connect_id, connect_cb=lambda: None, *, net, timeout_factor, supports_v2_p2p, reconnect):
         self.peer_connect_helper('0', 0, net, timeout_factor)
+        self.reconnect = reconnect
+        if supports_v2_p2p:
+            self.v2_state = EncryptedP2PState(initiating=False, net=net)
 
         logger.debug('Listening for Bitcoin Node with id: {}'.format(connect_id))
         return lambda: NetworkThread.listen(self, connect_cb, idx=connect_id)
@@ -206,14 +219,22 @@ class P2PConnection(asyncio.Protocol):
         assert not self._transport
         logger.debug("Connected & Listening: %s:%d" % (self.dstaddr, self.dstport))
         self._transport = transport
-        if self.on_connection_send_msg:
+        # in an inbound connection to the TestNode with P2PConnection as the initiator, [TestNode <---- P2PConnection]
+        # send the initial handshake immediately
+        if self.supports_v2_p2p and self.v2_state.initiating and not self.v2_state.tried_v2_handshake:
+            send_handshake_bytes = self.v2_state.initiate_v2_handshake()
+            self.send_raw_message(send_handshake_bytes)
+        # if v2 connection, send `on_connection_send_msg` after initial v2 handshake.
+        # if reconnection situation, send `on_connection_send_msg` after version message is received in `on_version()`.
+        if self.on_connection_send_msg and not self.supports_v2_p2p and not self.reconnect:
             self.send_message(self.on_connection_send_msg)
             self.on_connection_send_msg = None  # Never used again
         self.on_open()
 
     def connection_lost(self, exc):
         """asyncio callback when a connection is closed."""
-        if exc:
+        # don't display warning if reconnection needs to be attempted using v1 P2P
+        if exc and not self.reconnect:
             logger.warning("Connection lost to {}:{} due to {}".format(self.dstaddr, self.dstport, exc))
         else:
             logger.debug("Closed connection to: %s:%d" % (self.dstaddr, self.dstport))
@@ -221,13 +242,62 @@ class P2PConnection(asyncio.Protocol):
         self.recvbuf = b""
         self.on_close()
 
+    # v2 handshake method
+    def v2_handshake(self):
+        """v2 handshake performed before P2P messages are exchanged (see BIP324). P2PConnection is the initiator
+        (in inbound connections to TestNode) and the responder (in outbound connections from TestNode).
+        Performed by:
+            * initiator using `initiate_v2_handshake()`, `complete_handshake()` and `authenticate_handshake()`
+            * responder using `respond_v2_handshake()`, `complete_handshake()` and `authenticate_handshake()`
+
+        `initiate_v2_handshake()` is immediately done by the initiator when the connection is established in
+        `connection_made()`. The rest of the initial v2 handshake functions are handled here.
+        """
+        if not self.v2_state.peer:
+            if not self.v2_state.initiating and not self.v2_state.sent_garbage:
+                # if the responder hasn't sent garbage yet, the responder is still reading ellswift bytes
+                # reads ellswift bytes till the first mismatch from 12 bytes V1_PREFIX
+                length, send_handshake_bytes = self.v2_state.respond_v2_handshake(BytesIO(self.recvbuf))
+                self.recvbuf = self.recvbuf[length:]
+                if send_handshake_bytes == -1:
+                    self.v2_state = None
+                    return
+                elif send_handshake_bytes:
+                    self.send_raw_message(send_handshake_bytes)
+                elif send_handshake_bytes == b"":
+                    return  # only after send_handshake_bytes are sent can `complete_handshake()` be done
+
+            # `complete_handshake()` reads the remaining ellswift bytes from recvbuf
+            # and sends response after deriving shared ECDH secret using received ellswift bytes
+            length, response = self.v2_state.complete_handshake(BytesIO(self.recvbuf))
+            self.recvbuf = self.recvbuf[length:]
+            if response:
+                self.send_raw_message(response)
+            else:
+                return  # only after response is sent can `authenticate_handshake()` be done
+
+        # `self.v2_state.peer` is instantiated only after shared ECDH secret/BIP324 derived keys and ciphers
+        # is derived in `complete_handshake()`.
+        # so `authenticate_handshake()` which uses the BIP324 derived ciphers gets called after `complete_handshake()`.
+        assert self.v2_state.peer
+        length, is_mac_auth = self.v2_state.authenticate_handshake(self.recvbuf)
+        if not is_mac_auth:
+            raise ValueError("invalid v2 mac tag in handshake authentication")
+        self.recvbuf = self.recvbuf[length:]
+        if self.v2_state.tried_v2_handshake and self.on_connection_send_msg:
+            self.send_message(self.on_connection_send_msg)
+            self.on_connection_send_msg = None
+
     # Socket read methods
 
     def data_received(self, t):
         """asyncio callback when data is read from the socket."""
         if len(t) > 0:
             self.recvbuf += t
-            self._on_data()
+            if self.supports_v2_p2p and not self.v2_state.tried_v2_handshake:
+                self.v2_handshake()
+            else:
+                self._on_data()
 
     def _on_data(self):
         """Try to read P2P messages from the recv buffer.
@@ -237,23 +307,48 @@ class P2PConnection(asyncio.Protocol):
         the on_message callback for processing."""
         try:
             while True:
-                if len(self.recvbuf) < 4:
-                    return
-                if self.recvbuf[:4] != self.magic_bytes:
-                    raise ValueError("magic bytes mismatch: {} != {}".format(repr(self.magic_bytes), repr(self.recvbuf)))
-                if len(self.recvbuf) < 4 + 12 + 4 + 4:
-                    return
-                msgtype = self.recvbuf[4:4+12].split(b"\x00", 1)[0]
-                msglen = struct.unpack("<i", self.recvbuf[4+12:4+12+4])[0]
-                checksum = self.recvbuf[4+12+4:4+12+4+4]
-                if len(self.recvbuf) < 4 + 12 + 4 + 4 + msglen:
-                    return
-                msg = self.recvbuf[4+12+4+4:4+12+4+4+msglen]
-                th = sha256(msg)
-                h = sha256(th)
-                if checksum != h[:4]:
-                    raise ValueError("got bad checksum " + repr(self.recvbuf))
-                self.recvbuf = self.recvbuf[4+12+4+4+msglen:]
+                if self.supports_v2_p2p:
+                    # v2 P2P messages are read
+                    msglen, msg = self.v2_state.v2_receive_packet(self.recvbuf)
+                    if msglen == -1:
+                        raise ValueError("invalid v2 mac tag " + repr(self.recvbuf))
+                    elif msglen == 0:  # need to receive more bytes in recvbuf
+                        return
+                    self.recvbuf = self.recvbuf[msglen:]
+
+                    if msg is None:  # ignore decoy messages
+                        return
+                    assert msg  # application layer messages (which aren't decoy messages) are non-empty
+                    shortid = msg[0]  # 1-byte short message type ID
+                    if shortid == 0:
+                        # next 12 bytes are interpreted as ASCII message type if shortid is b'\x00'
+                        if len(msg) < 13:
+                            raise IndexError("msg needs minimum required length of 13 bytes")
+                        msgtype = msg[1:13].rstrip(b'\x00')
+                        msg = msg[13:]  # msg is set to be payload
+                    else:
+                        # a 1-byte short message type ID
+                        msgtype = SHORTID.get(shortid, f"unknown-{shortid}")
+                        msg = msg[1:]
+                else:
+                    # v1 P2P messages are read
+                    if len(self.recvbuf) < 4:
+                        return
+                    if self.recvbuf[:4] != self.magic_bytes:
+                        raise ValueError("magic bytes mismatch: {} != {}".format(repr(self.magic_bytes), repr(self.recvbuf)))
+                    if len(self.recvbuf) < 4 + 12 + 4 + 4:
+                        return
+                    msgtype = self.recvbuf[4:4+12].split(b"\x00", 1)[0]
+                    msglen = struct.unpack("<i", self.recvbuf[4+12:4+12+4])[0]
+                    checksum = self.recvbuf[4+12+4:4+12+4+4]
+                    if len(self.recvbuf) < 4 + 12 + 4 + 4 + msglen:
+                        return
+                    msg = self.recvbuf[4+12+4+4:4+12+4+4+msglen]
+                    th = sha256(msg)
+                    h = sha256(th)
+                    if checksum != h[:4]:
+                        raise ValueError("got bad checksum " + repr(self.recvbuf))
+                    self.recvbuf = self.recvbuf[4+12+4+4+msglen:]
                 if msgtype not in MESSAGEMAP:
                     raise ValueError("Received unknown msgtype from %s:%d: '%s' %s" % (self.dstaddr, self.dstport, msgtype, repr(msg)))
                 f = BytesIO(msg)
@@ -262,7 +357,8 @@ class P2PConnection(asyncio.Protocol):
                 self._log_message("receive", t)
                 self.on_message(t)
         except Exception as e:
-            logger.exception('Error reading message:', repr(e))
+            if not self.reconnect:
+                logger.exception('Error reading message:', repr(e))
             raise
 
     def on_message(self, message):
@@ -271,14 +367,15 @@ class P2PConnection(asyncio.Protocol):
 
     # Socket write methods
 
-    def send_message(self, message):
+    def send_message(self, message, is_decoy=False):
         """Send a P2P message over the socket.
 
         This method takes a P2P payload, builds the P2P header and adds
         the message to the send buffer to be sent over the socket."""
-        tmsg = self.build_message(message)
-        self._log_message("send", message)
-        return self.send_raw_message(tmsg)
+        with self._send_lock:
+            tmsg = self.build_message(message, is_decoy)
+            self._log_message("send", message)
+            return self.send_raw_message(tmsg)
 
     def send_raw_message(self, raw_message_bytes):
         if not self.is_connected:
@@ -294,19 +391,29 @@ class P2PConnection(asyncio.Protocol):
 
     # Class utility methods
 
-    def build_message(self, message):
+    def build_message(self, message, is_decoy=False):
         """Build a serialized P2P message"""
         msgtype = message.msgtype
         data = message.serialize()
-        tmsg = self.magic_bytes
-        tmsg += msgtype
-        tmsg += b"\x00" * (12 - len(msgtype))
-        tmsg += struct.pack("<I", len(data))
-        th = sha256(data)
-        h = sha256(th)
-        tmsg += h[:4]
-        tmsg += data
-        return tmsg
+        if self.supports_v2_p2p:
+            if msgtype in SHORTID.values():
+                tmsg = MSGTYPE_TO_SHORTID.get(msgtype).to_bytes(1, 'big')
+            else:
+                tmsg = b"\x00"
+                tmsg += msgtype
+                tmsg += b"\x00" * (12 - len(msgtype))
+            tmsg += data
+            return self.v2_state.v2_enc_packet(tmsg, ignore=is_decoy)
+        else:
+            tmsg = self.magic_bytes
+            tmsg += msgtype
+            tmsg += b"\x00" * (12 - len(msgtype))
+            tmsg += struct.pack("<I", len(data))
+            th = sha256(data)
+            h = sha256(th)
+            tmsg += h[:4]
+            tmsg += data
+            return tmsg
 
     def _log_message(self, direction, msg):
         """Logs a message being sent or received over the connection."""
@@ -450,6 +557,12 @@ class P2PInterface(P2PConnection):
 
     def on_version(self, message):
         assert message.nVersion >= MIN_P2P_VERSION_SUPPORTED, "Version {} received. Test framework only supports versions greater than {}".format(message.nVersion, MIN_P2P_VERSION_SUPPORTED)
+        # reconnection using v1 P2P has happened since version message can be processed, previously unsent version message is sent using v1 P2P here
+        if self.reconnect:
+            if self.on_connection_send_msg:
+                self.send_message(self.on_connection_send_msg)
+                self.on_connection_send_msg = None
+            self.reconnect = False
         if message.nVersion >= 70016 and self.wtxidrelay:
             self.send_message(msg_wtxidrelay())
         if self.support_addrv2:
@@ -478,6 +591,13 @@ class P2PInterface(P2PConnection):
         test_function = lambda: not self.is_connected
         self.wait_until(test_function, timeout=timeout, check_connected=False)
 
+    def wait_for_reconnect(self, timeout=60):
+        def test_function():
+            if not (self.is_connected and self.last_message.get('version') and self.v2_state is None):
+                return False
+            return True
+        self.wait_until(test_function, timeout=timeout, check_connected=False)
+
     # Message receiving helper methods
 
     def wait_for_tx(self, txid, timeout=60):
@@ -622,6 +742,11 @@ class NetworkThread(threading.Thread):
         if addr is None:
             addr = '127.0.0.1'
 
+        def exception_handler(loop, context):
+            if not p2p.reconnect:
+                loop.default_exception_handler(context)
+
+        cls.network_event_loop.set_exception_handler(exception_handler)
         coroutine = cls.create_listen_server(addr, port, callback, p2p)
         cls.network_event_loop.call_soon_threadsafe(cls.network_event_loop.create_task, coroutine)
 
@@ -635,7 +760,9 @@ class NetworkThread(threading.Thread):
             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
+            # remove protocol function from dict only when reconnection doesn't need to happen/already happened
+            if not proto.reconnect:
+                cls.protos[(addr, port)] = None
             return response
 
         if (addr, port) not in cls.listeners:
@@ -708,7 +835,7 @@ class P2PDataStore(P2PInterface):
         if response is not None:
             self.send_message(response)
 
-    def send_blocks_and_test(self, blocks, node, *, success=True, force_send=False, reject_reason=None, expect_disconnect=False, timeout=60):
+    def send_blocks_and_test(self, blocks, node, *, success=True, force_send=False, reject_reason=None, expect_disconnect=False, timeout=60, is_decoy=False):
         """Send blocks to test node and test whether the tip advances.
 
          - add all blocks to our block_store
@@ -727,9 +854,11 @@ class P2PDataStore(P2PInterface):
 
         reject_reason = [reject_reason] if reject_reason else []
         with node.assert_debug_log(expected_msgs=reject_reason):
+            if is_decoy:  # since decoy messages are ignored by the recipient - no need to wait for response
+                force_send = True
             if force_send:
                 for b in blocks:
-                    self.send_message(msg_block(block=b))
+                    self.send_message(msg_block(block=b), is_decoy)
             else:
                 self.send_message(msg_headers([CBlockHeader(block) for block in blocks]))
                 self.wait_until(
diff --git a/test/functional/test_framework/test_node.py b/test/functional/test_framework/test_node.py
index 77f6e69e98..58956a95f7 100755
--- a/test/functional/test_framework/test_node.py
+++ b/test/functional/test_framework/test_node.py
@@ -27,7 +27,8 @@ from .authproxy import (
     serialization_fallback,
 )
 from .descriptors import descsum_create
-from .p2p import P2P_SUBVERSION
+from .messages import NODE_P2P_V2
+from .p2p import P2P_SERVICES, P2P_SUBVERSION
 from .util import (
     MAX_NODES,
     assert_equal,
@@ -659,18 +660,30 @@ class TestNode():
                     assert_msg += "with expected error " + expected_msg
                 self._raise_assertion_error(assert_msg)
 
-    def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, send_version=True, **kwargs):
+    def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, send_version=True, supports_v2_p2p=False, **kwargs):
         """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."""
+        returns the connection to the caller.
+
+        When self.use_v2transport is True, TestNode advertises NODE_P2P_V2 service flag
+
+        An inbound connection is made from TestNode <------ P2PConnection
+        - if TestNode doesn't advertise NODE_P2P_V2 service, P2PConnection sends version message and v1 P2P is followed
+        - if TestNode advertises NODE_P2P_V2 service, (and if P2PConnections supports v2 P2P)
+                P2PConnection sends ellswift bytes and v2 P2P is followed
+        """
         if 'dstport' not in kwargs:
             kwargs['dstport'] = p2p_port(self.index)
         if 'dstaddr' not in kwargs:
             kwargs['dstaddr'] = '127.0.0.1'
 
         p2p_conn.p2p_connected_to_node = True
-        p2p_conn.peer_connect(**kwargs, send_version=send_version, net=self.chain, timeout_factor=self.timeout_factor)()
+        if self.use_v2transport:
+            kwargs['services'] = kwargs.get('services', P2P_SERVICES) | NODE_P2P_V2
+        supports_v2_p2p = self.use_v2transport and supports_v2_p2p
+        p2p_conn.peer_connect(**kwargs, send_version=send_version, net=self.chain, timeout_factor=self.timeout_factor, supports_v2_p2p=supports_v2_p2p)()
+
         self.p2ps.append(p2p_conn)
         p2p_conn.wait_until(lambda: p2p_conn.is_connected, check_connected=False)
         if send_version:
@@ -701,7 +714,7 @@ class TestNode():
 
         return p2p_conn
 
-    def add_outbound_p2p_connection(self, p2p_conn, *, wait_for_verack=True, p2p_idx, connection_type="outbound-full-relay", **kwargs):
+    def add_outbound_p2p_connection(self, p2p_conn, *, wait_for_verack=True, p2p_idx, connection_type="outbound-full-relay", supports_v2_p2p=False, advertise_v2_p2p=False, **kwargs):
         """Add an outbound p2p connection from node. Must be an
         "outbound-full-relay", "block-relay-only", "addr-fetch" or "feeler" connection.
 
@@ -711,14 +724,37 @@ class TestNode():
         p2p_idx must be different for simultaneously connected peers. When reusing it for the next peer
         after disconnecting the previous one, it is necessary to wait for the disconnect to finish to avoid
         a race condition.
+
+        Parameters:
+            supports_v2_p2p: whether p2p_conn supports v2 P2P or not
+            advertise_v2_p2p: whether p2p_conn is advertised to support v2 P2P or not
+
+        An outbound connection is made from TestNode -------> P2PConnection
+            - if P2PConnection doesn't advertise_v2_p2p, TestNode sends version message and v1 P2P is followed
+            - if P2PConnection both supports_v2_p2p and advertise_v2_p2p, TestNode sends ellswift bytes and v2 P2P is followed
+            - if P2PConnection doesn't supports_v2_p2p but advertise_v2_p2p,
+                TestNode sends ellswift bytes and P2PConnection disconnects,
+                TestNode reconnects by sending version message and v1 P2P is followed
         """
 
         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)
+            self.addconnection('%s:%d' % (address, port), connection_type, advertise_v2_p2p)
 
         p2p_conn.p2p_connected_to_node = False
-        p2p_conn.peer_accept_connection(connect_cb=addconnection_callback, connect_id=p2p_idx + 1, net=self.chain, timeout_factor=self.timeout_factor, **kwargs)()
+        if advertise_v2_p2p:
+            kwargs['services'] = kwargs.get('services', P2P_SERVICES) | NODE_P2P_V2
+            assert self.use_v2transport  # only a v2 TestNode could make a v2 outbound connection
+
+        # if P2PConnection is advertised to support v2 P2P when it doesn't actually support v2 P2P,
+        # reconnection needs to be attempted using v1 P2P by sending version message
+        reconnect = advertise_v2_p2p and not supports_v2_p2p
+        # P2PConnection needs to be advertised to support v2 P2P so that ellswift bytes are sent instead of msg_version
+        supports_v2_p2p = supports_v2_p2p and advertise_v2_p2p
+        p2p_conn.peer_accept_connection(connect_cb=addconnection_callback, connect_id=p2p_idx + 1, net=self.chain, timeout_factor=self.timeout_factor, supports_v2_p2p=supports_v2_p2p, reconnect=reconnect, **kwargs)()
+
+        if reconnect:
+            p2p_conn.wait_for_reconnect()
 
         if connection_type == "feeler":
             # feeler connections are closed as soon as the node receives a `version` message
diff --git a/test/functional/test_framework/v2_p2p.py b/test/functional/test_framework/v2_p2p.py
new file mode 100644
index 0000000000..0b3979fba2
--- /dev/null
+++ b/test/functional/test_framework/v2_p2p.py
@@ -0,0 +1,284 @@
+#!/usr/bin/env python3
+# Copyright (c) 2022 The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Class for v2 P2P protocol (see BIP 324)"""
+
+import logging
+import random
+
+from .crypto.bip324_cipher import FSChaCha20Poly1305
+from .crypto.chacha20 import FSChaCha20
+from .crypto.ellswift import ellswift_create, ellswift_ecdh_xonly
+from .crypto.hkdf import hkdf_sha256
+from .key import TaggedHash
+from .messages import MAGIC_BYTES
+
+logger = logging.getLogger("TestFramework.v2_p2p")
+
+CHACHA20POLY1305_EXPANSION = 16
+HEADER_LEN = 1
+IGNORE_BIT_POS = 7
+LENGTH_FIELD_LEN = 3
+MAX_GARBAGE_LEN = 4095
+TRANSPORT_VERSION = b''
+
+SHORTID = {
+    1: b"addr",
+    2: b"block",
+    3: b"blocktxn",
+    4: b"cmpctblock",
+    5: b"feefilter",
+    6: b"filteradd",
+    7: b"filterclear",
+    8: b"filterload",
+    9: b"getblocks",
+    10: b"getblocktxn",
+    11: b"getdata",
+    12: b"getheaders",
+    13: b"headers",
+    14: b"inv",
+    15: b"mempool",
+    16: b"merkleblock",
+    17: b"notfound",
+    18: b"ping",
+    19: b"pong",
+    20: b"sendcmpct",
+    21: b"tx",
+    22: b"getcfilters",
+    23: b"cfilter",
+    24: b"getcfheaders",
+    25: b"cfheaders",
+    26: b"getcfcheckpt",
+    27: b"cfcheckpt",
+    28: b"addrv2",
+}
+
+# Dictionary which contains short message type ID for the P2P message
+MSGTYPE_TO_SHORTID = {msgtype: shortid for shortid, msgtype in SHORTID.items()}
+
+
+class EncryptedP2PState:
+    """A class for managing the state when v2 P2P protocol is used. Performs initial v2 handshake and encrypts/decrypts
+    P2P messages. P2PConnection uses an object of this class.
+
+
+    Args:
+        initiating (bool): defines whether the P2PConnection is an initiator or responder.
+            - initiating = True for inbound connections in the test framework   [TestNode <------- P2PConnection]
+            - initiating = False for outbound connections in the test framework [TestNode -------> P2PConnection]
+
+        net (string): chain used (regtest, signet etc..)
+
+    Methods:
+        perform an advanced form of diffie-hellman handshake to instantiate the encrypted transport. before exchanging
+        any P2P messages, 2 nodes perform this handshake in order to determine a shared secret that is unique to both
+        of them and use it to derive keys to encrypt/decrypt P2P messages.
+            - initial v2 handshakes is performed by: (see BIP324 section #overall-handshake-pseudocode)
+                1. initiator using initiate_v2_handshake(), complete_handshake() and authenticate_handshake()
+                2. responder using respond_v2_handshake(), complete_handshake() and authenticate_handshake()
+            - initialize_v2_transport() sets various BIP324 derived keys and ciphers.
+
+        encrypt/decrypt v2 P2P messages using v2_enc_packet() and v2_receive_packet().
+    """
+    def __init__(self, *, initiating, net):
+        self.initiating = initiating  # True if initiator
+        self.net = net
+        self.peer = {}  # object with various BIP324 derived keys and ciphers
+        self.privkey_ours = None
+        self.ellswift_ours = None
+        self.sent_garbage = b""
+        self.received_garbage = b""
+        self.received_prefix = b""  # received ellswift bytes till the first mismatch from 16 bytes v1_prefix
+        self.tried_v2_handshake = False  # True when the initial handshake is over
+        # stores length of packet contents to detect whether first 3 bytes (which contains length of packet contents)
+        # has been decrypted. set to -1 if decryption hasn't been done yet.
+        self.contents_len = -1
+        self.found_garbage_terminator = False
+
+    @staticmethod
+    def v2_ecdh(priv, ellswift_theirs, ellswift_ours, initiating):
+        """Compute BIP324 shared secret.
+
+        Returns:
+        bytes - BIP324 shared secret
+        """
+        ecdh_point_x32 = ellswift_ecdh_xonly(ellswift_theirs, priv)
+        if initiating:
+            # Initiating, place our public key encoding first.
+            return TaggedHash("bip324_ellswift_xonly_ecdh", ellswift_ours + ellswift_theirs + ecdh_point_x32)
+        else:
+            # Responding, place their public key encoding first.
+            return TaggedHash("bip324_ellswift_xonly_ecdh", ellswift_theirs + ellswift_ours + ecdh_point_x32)
+
+    def generate_keypair_and_garbage(self):
+        """Generates ellswift keypair and 4095 bytes garbage at max"""
+        self.privkey_ours, self.ellswift_ours = ellswift_create()
+        garbage_len = random.randrange(MAX_GARBAGE_LEN + 1)
+        self.sent_garbage = random.randbytes(garbage_len)
+        logger.debug(f"sending {garbage_len} bytes of garbage data")
+        return self.ellswift_ours + self.sent_garbage
+
+    def initiate_v2_handshake(self):
+        """Initiator begins the v2 handshake by sending its ellswift bytes and garbage
+
+        Returns:
+        bytes - bytes to be sent to the peer when starting the v2 handshake as an initiator
+        """
+        return self.generate_keypair_and_garbage()
+
+    def respond_v2_handshake(self, response):
+        """Responder begins the v2 handshake by sending its ellswift bytes and garbage. However, the responder
+        sends this after having received at least one byte that mismatches 16-byte v1_prefix.
+
+        Returns:
+        1. int - length of bytes that were consumed so that recvbuf can be updated
+        2. bytes - bytes to be sent to the peer when starting the v2 handshake as a responder.
+                 - returns b"" if more bytes need to be received before we can respond and start the v2 handshake.
+                 - returns -1 to downgrade the connection to v1 P2P.
+        """
+        v1_prefix = MAGIC_BYTES[self.net] + b'version\x00\x00\x00\x00\x00'
+        while len(self.received_prefix) < 16:
+            byte = response.read(1)
+            # return b"" if we need to receive more bytes
+            if not byte:
+                return len(self.received_prefix), b""
+            self.received_prefix += byte
+            if self.received_prefix[-1] != v1_prefix[len(self.received_prefix) - 1]:
+                return len(self.received_prefix), self.generate_keypair_and_garbage()
+        # return -1 to decide v1 only after all 16 bytes processed
+        return len(self.received_prefix), -1
+
+    def complete_handshake(self, response):
+        """ Instantiates the encrypted transport and
+        sends garbage terminator + optional decoy packets + transport version packet.
+        Done by both initiator and responder.
+
+        Returns:
+        1. int - length of bytes that were consumed. returns 0 if all 64 bytes from ellswift haven't been received yet.
+        2. bytes - bytes to be sent to the peer when completing the v2 handshake
+        """
+        ellswift_theirs = self.received_prefix + response.read(64 - len(self.received_prefix))
+        # return b"" if we need to receive more bytes
+        if len(ellswift_theirs) != 64:
+            return 0, b""
+        ecdh_secret = self.v2_ecdh(self.privkey_ours, ellswift_theirs, self.ellswift_ours, self.initiating)
+        self.initialize_v2_transport(ecdh_secret)
+        # Send garbage terminator
+        msg_to_send = self.peer['send_garbage_terminator']
+        # Optionally send decoy packets after garbage terminator.
+        aad = self.sent_garbage
+        for decoy_content_len in [random.randint(1, 100) for _ in range(random.randint(0, 10))]:
+            msg_to_send += self.v2_enc_packet(decoy_content_len * b'\x00', aad=aad, ignore=True)
+            aad = b''
+        # Send version packet.
+        msg_to_send += self.v2_enc_packet(TRANSPORT_VERSION, aad=aad)
+        return 64 - len(self.received_prefix), msg_to_send
+
+    def authenticate_handshake(self, response):
+        """ Ensures that the received optional decoy packets and transport version packet are authenticated.
+        Marks the v2 handshake as complete. Done by both initiator and responder.
+
+        Returns:
+        1. int - length of bytes that were processed so that recvbuf can be updated
+        2. bool - True if the authentication was successful/more bytes need to be received and False otherwise
+        """
+        processed_length = 0
+
+        # Detect garbage terminator in the received bytes
+        if not self.found_garbage_terminator:
+            received_garbage = response[:16]
+            response = response[16:]
+            processed_length = len(received_garbage)
+            for i in range(MAX_GARBAGE_LEN + 1):
+                if received_garbage[-16:] == self.peer['recv_garbage_terminator']:
+                    # Receive, decode, and ignore version packet.
+                    # This includes skipping decoys and authenticating the received garbage.
+                    self.found_garbage_terminator = True
+                    self.received_garbage = received_garbage[:-16]
+                    break
+                else:
+                    # don't update recvbuf since more bytes need to be received
+                    if len(response) == 0:
+                        return 0, True
+                    received_garbage += response[:1]
+                    processed_length += 1
+                    response = response[1:]
+            else:
+                # disconnect since garbage terminator was not seen after 4 KiB of garbage.
+                return processed_length, False
+
+        # Process optional decoy packets and transport version packet
+        while not self.tried_v2_handshake:
+            length, contents = self.v2_receive_packet(response, aad=self.received_garbage)
+            if length == -1:
+                return processed_length, False
+            elif length == 0:
+                return processed_length, True
+            processed_length += length
+            self.received_garbage = b""
+            # decoy packets have contents = None. v2 handshake is complete only when version packet
+            # (can be empty with contents = b"") with contents != None is received.
+            if contents is not None:
+                self.tried_v2_handshake = True
+                return processed_length, True
+            response = response[length:]
+
+    def initialize_v2_transport(self, ecdh_secret):
+        """Sets the peer object with various BIP324 derived keys and ciphers."""
+        peer = {}
+        salt = b'bitcoin_v2_shared_secret' + MAGIC_BYTES[self.net]
+        for name in ('initiator_L', 'initiator_P', 'responder_L', 'responder_P', 'garbage_terminators', 'session_id'):
+            peer[name] = hkdf_sha256(salt=salt, ikm=ecdh_secret, info=name.encode('utf-8'), length=32)
+        if self.initiating:
+            self.peer['send_L'] = FSChaCha20(peer['initiator_L'])
+            self.peer['send_P'] = FSChaCha20Poly1305(peer['initiator_P'])
+            self.peer['send_garbage_terminator'] = peer['garbage_terminators'][:16]
+            self.peer['recv_L'] = FSChaCha20(peer['responder_L'])
+            self.peer['recv_P'] = FSChaCha20Poly1305(peer['responder_P'])
+            self.peer['recv_garbage_terminator'] = peer['garbage_terminators'][16:]
+        else:
+            self.peer['send_L'] = FSChaCha20(peer['responder_L'])
+            self.peer['send_P'] = FSChaCha20Poly1305(peer['responder_P'])
+            self.peer['send_garbage_terminator'] = peer['garbage_terminators'][16:]
+            self.peer['recv_L'] = FSChaCha20(peer['initiator_L'])
+            self.peer['recv_P'] = FSChaCha20Poly1305(peer['initiator_P'])
+            self.peer['recv_garbage_terminator'] = peer['garbage_terminators'][:16]
+        self.peer['session_id'] = peer['session_id']
+
+    def v2_enc_packet(self, contents, aad=b'', ignore=False):
+        """Encrypt a BIP324 packet.
+
+        Returns:
+        bytes - encrypted packet contents
+        """
+        assert len(contents) <= 2**24 - 1
+        header = (ignore << IGNORE_BIT_POS).to_bytes(HEADER_LEN, 'little')
+        plaintext = header + contents
+        aead_ciphertext = self.peer['send_P'].encrypt(aad, plaintext)
+        enc_plaintext_len = self.peer['send_L'].crypt(len(contents).to_bytes(LENGTH_FIELD_LEN, 'little'))
+        return enc_plaintext_len + aead_ciphertext
+
+    def v2_receive_packet(self, response, aad=b''):
+        """Decrypt a BIP324 packet
+
+        Returns:
+        1. int - number of bytes consumed (or -1 if error)
+        2. bytes - contents of decrypted non-decoy packet if any (or None otherwise)
+        """
+        if self.contents_len == -1:
+            if len(response) < LENGTH_FIELD_LEN:
+                return 0, None
+            enc_contents_len = response[:LENGTH_FIELD_LEN]
+            self.contents_len = int.from_bytes(self.peer['recv_L'].crypt(enc_contents_len), 'little')
+        response = response[LENGTH_FIELD_LEN:]
+        if len(response) < HEADER_LEN + self.contents_len + CHACHA20POLY1305_EXPANSION:
+            return 0, None
+        aead_ciphertext = response[:HEADER_LEN + self.contents_len + CHACHA20POLY1305_EXPANSION]
+        plaintext = self.peer['recv_P'].decrypt(aad, aead_ciphertext)
+        if plaintext is None:
+            return -1, None  # disconnect
+        header = plaintext[:HEADER_LEN]
+        length = LENGTH_FIELD_LEN + HEADER_LEN + self.contents_len + CHACHA20POLY1305_EXPANSION
+        self.contents_len = -1
+        return length, None if (header[0] & (1 << IGNORE_BIT_POS)) else plaintext[HEADER_LEN:]