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()
155