1  #!/usr/bin/env python3
  2  # Copyright (c) 2025-present 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  """
  6  Test that a node sends a self-announcement with its external IP to
  7  in- and outbound peers after connection open and again sometime later.
  8  
  9  Additionally, this checks that the first self-announcement arrives
 10  in its own message and that this message is the first we receive.
 11  """
 12  
 13  import time
 14  
 15  from test_framework.messages import (
 16      CAddress,
 17      from_hex,
 18      msg_headers,
 19      CBlockHeader,
 20  )
 21  from test_framework.p2p import P2PInterface
 22  from test_framework.test_framework import BitcoinTestFramework
 23  from test_framework.util import assert_equal, assert_greater_than
 24  
 25  IP_TO_ANNOUNCE = "42.42.42.42"
 26  ONE_DAY = 60 * 60 * 24
 27  
 28  
 29  class SelfAnnouncementReceiver(P2PInterface):
 30      self_announcements_received = 0
 31      addresses_received = 0
 32      addr_messages_received = 0
 33  
 34      expected = None
 35      addrv2_test = False
 36  
 37      def __init__(self, *, expected, addrv2):
 38          super().__init__(support_addrv2=addrv2)
 39          self.expected = expected
 40          self.addrv2_test = addrv2
 41  
 42      def handle_addr_message(self, message):
 43          self.addr_messages_received += 1
 44          for addr in message.addrs:
 45              self.addresses_received += 1
 46              if addr == self.expected:
 47                  self.self_announcements_received += 1
 48                  if self.self_announcements_received == 1:
 49                      # If it's the first self-announcement, check that it is
 50                      # in the first addr message we receive, and that this message
 51                      # only contains one address. This also implies that it is
 52                      # the first address we receive.
 53                      assert_equal(self.addr_messages_received, 1)
 54                      assert_equal(len(message.addrs), 1)
 55  
 56      def on_addrv2(self, message):
 57          assert (self.addrv2_test)
 58          self.handle_addr_message(message)
 59  
 60      def on_addr(self, message):
 61          assert (not self.addrv2_test)
 62          self.handle_addr_message(message)
 63  
 64  
 65  class AddrSelfAnnouncementTest(BitcoinTestFramework):
 66      def set_test_params(self):
 67          self.num_nodes = 1
 68          self.extra_args = [[f"-externalip={IP_TO_ANNOUNCE}"]]
 69  
 70      def run_test(self):
 71          # populate addrman to have some addresses for a GETADDR response
 72          for i in range(50):
 73              a = f"{1 + i}.{i}.1.1"
 74              self.nodes[0].addpeeraddress(a, 8333)
 75  
 76          self.self_announcement_test(outbound=False, addrv2=False)
 77          self.self_announcement_test(outbound=False, addrv2=True)
 78          self.self_announcement_test(outbound=True, addrv2=False)
 79          self.self_announcement_test(outbound=True, addrv2=True)
 80  
 81      @staticmethod
 82      def inbound_connection_open_assertions(addr_receiver):
 83          # In response to a GETADDR, we expect a message with the self-announcement
 84          # and an addr message containing the GETADDR response.
 85          assert_equal(addr_receiver.self_announcements_received, 1)
 86          assert_equal(addr_receiver.addr_messages_received, 2)
 87          assert_greater_than(addr_receiver.addresses_received, 1)
 88  
 89      @staticmethod
 90      def outbound_connection_open_assertions(addr_receiver):
 91          # We expect only the self-announcement.
 92          assert_equal(addr_receiver.self_announcements_received, 1)
 93          assert_equal(addr_receiver.addr_messages_received, 1)
 94          assert_equal(addr_receiver.addresses_received, 1)
 95  
 96      def self_announcement_test(self, *, outbound, addrv2):
 97          connection_type = "outbound" if outbound else "inbound"
 98          addr_version = "addrv2" if addrv2 else "addrv1"
 99          self.log.info(f"Test that the node does an address self-announcement to {connection_type} connections ({addr_version})")
100  
101          # We only self-announce after initial block download is done
102          assert (not self.nodes[0].getblockchaininfo()["initialblockdownload"])
103  
104          netinfo = self.nodes[0].getnetworkinfo()
105          port = netinfo["localaddresses"][0]["port"]
106          self.nodes[0].setmocktime(int(time.time()))
107  
108          expected = CAddress()
109          expected.nServices = int(netinfo["localservices"], 16)
110          expected.ip = IP_TO_ANNOUNCE
111          expected.port = port
112          expected.time = self.nodes[0].mocktime
113  
114          with self.nodes[0].assert_debug_log([f'Advertising address {IP_TO_ANNOUNCE}:{port}']):
115              if outbound:
116                  self.log.info(f"Check that we get an initial self-announcement on an outbound connection from the node ({connection_type}, {addr_version})")
117                  addr_receiver = self.nodes[0].add_outbound_p2p_connection(SelfAnnouncementReceiver(expected=expected, addrv2=addrv2), p2p_idx=0, connection_type="outbound-full-relay")
118              else:
119                  self.log.info(f"Check that we get an initial self-announcement when connecting to a node and sending a GETADDR ({connection_type}, {addr_version})")
120                  addr_receiver = self.nodes[0].add_p2p_connection(SelfAnnouncementReceiver(expected=expected, addrv2=addrv2))
121              addr_receiver.sync_with_ping()
122  
123          if outbound:
124              self.outbound_connection_open_assertions(addr_receiver)
125          else:
126              self.inbound_connection_open_assertions(addr_receiver)
127  
128          if outbound:
129              # to avoid the node evicting the outbound peer, protect it by announcing the most recent header to it
130              tip_header = from_hex(CBlockHeader(), self.nodes[0].getblockheader(self.nodes[0].getbestblockhash(), False))
131              addr_receiver.send_and_ping(msg_headers([tip_header]))
132  
133          self.log.info(f"Check that we get more self-announcements sometime later ({connection_type}, {addr_version})")
134          for _ in range(5):
135              last_self_announcements_received = addr_receiver.self_announcements_received
136              last_addr_messages_received = addr_receiver.addr_messages_received
137              last_addresses_received = addr_receiver.addresses_received
138              with self.nodes[0].assert_debug_log([f'Advertising address {IP_TO_ANNOUNCE}:{port}']):
139                  # m_next_local_addr_send and AVG_LOCAL_ADDRESS_BROADCAST_INTERVAL:
140                  # self-announcements are sent on an exponential distribution with mean interval of 24h.
141                  # Setting the mocktime 20d forward gives a probability of (1 - e^-(480/24)) that
142                  # the event will occur (i.e. this fails once in ~500 million repeats).
143                  addr_receiver.expected.time = self.nodes[0].mocktime + 20 * ONE_DAY
144                  self.nodes[0].bumpmocktime(20 * ONE_DAY)
145                  addr_receiver.sync_with_ping()
146  
147              assert_equal(addr_receiver.self_announcements_received, last_self_announcements_received + 1)
148              assert_equal(addr_receiver.addr_messages_received, last_addr_messages_received + 1)
149              assert_equal(addr_receiver.addresses_received, last_addresses_received + 1)
150  
151          self.nodes[0].disconnect_p2ps()
152  
153  if __name__ == '__main__':
154      AddrSelfAnnouncementTest(__file__).main()