Inside Bitcoin

1#!/usr/bin/env python3
2# Copyright (c) 2017-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"""
6Test how locally submitted transactions are sent to the network when private broadcast is used.
7"""
8
9import time
10import threading
11
12from test_framework.p2p import (
13    P2PDataStore,
14    P2PInterface,
15    P2P_SERVICES,
16    start_p2p_listener,
17)
18from test_framework.messages import (
19    CAddress,
20    CInv,
21    MSG_WTX,
22    malleate_tx_to_invalid_witness,
23    msg_inv,
24    msg_tx,
25)
26from test_framework.netutil import (
27    format_addr_port
28)
29from test_framework.script_util import build_malleated_tx_package
30from test_framework.socks5 import (
31    start_socks5_server,
32)
33from test_framework.test_framework import (
34    BitcoinTestFramework,
35)
36from test_framework.util import (
37    assert_equal,
38    assert_greater_than_or_equal,
39    assert_not_equal,
40    assert_raises_rpc_error,
41    tor_port,
42)
43from test_framework.wallet import (
44    MiniWallet,
45)
46
47P2P_PRIVATE_VERSION = 70016
48NUM_PRIVATE_BROADCAST_PER_TX = 3
49MAX_PRIVATE_BROADCAST_ATTEMPTS = 1000
50
51
52class NoRelayP2PInterface(P2PInterface):
53    def peer_connect_send_version(self, services):
54        super().peer_connect_send_version(services)
55        self.on_connection_send_msg.relay = 0
56
57
58class P2PPrivateBroadcast(BitcoinTestFramework):
59    def set_test_params(self):
60        self.disable_autoconnect = False
61        self.num_nodes = 2
62
63    def setup_nodes(self):
64        self.destinations = []
65
66        self.destinations_lock = threading.Lock()
67
68        self.trigger_no_relay_peer = False
69        self.no_relay_peer = None
70
71        def destinations_factory(requested_to_addr, requested_to_port, proxy_client):
72            """
73            Instruct the SOCKS5 proxy to redirect connections:
74            * The first automatic outbound connection -> P2PDataStore
75            * The first private broadcast connection -> nodes[1]
76            * Anything else -> P2PInterface
77
78            proxy_client is the client's socket address as seen by the proxy (host:port),
79            equal to the node's addrbind for this connection.
80            """
81            conn_type = None
82            # SOCKS handlers run in separate threads, so each needs its own RPC connection.
83            rpc = self.nodes[0].create_new_rpc_connection()
84
85            def connection_type_found():
86                nonlocal conn_type
87                # The proxy has already replied SUCCESS to the SOCKS5 request, so the node
88                # has finished ConnectNode and registered the peer (or is about to).
89                # The proxy client address equals the node's addrbind for this connection.
90                for peer in rpc.getpeerinfo():
91                    if peer.get("addrbind") == proxy_client:
92                        conn_type = peer["connection_type"]
93                        return True
94                return False
95
96            self.wait_until(connection_type_found)
97
98            with self.destinations_lock:
99                i = len(self.destinations)
100                actual_to_addr = ""
101                actual_to_port = 0
102                listener = None
103                target_name = ""
104                if conn_type == "private-broadcast" and not any(dest["conn_type"] == "private-broadcast" for dest in self.destinations):
105                    # Instruct the SOCKS5 server to redirect the first private
106                    # broadcast connection from nodes[0] to nodes[1]
107                    actual_to_addr = "127.0.0.1" # nodes[1] listen address
108                    actual_to_port = tor_port(1) # nodes[1] listen port for Tor
109                    target_name = "nodes[1]"
110                else:
111                    # Create a Python P2P listening node and instruct the SOCKS5 proxy to
112                    # redirect the connection to it. The first outbound connection is used
113                    # later to serve GETDATA, thus make it P2PDataStore().
114                    if conn_type == "outbound-full-relay" and not any(dest["conn_type"] == "outbound-full-relay" for dest in self.destinations):
115                        listener = P2PDataStore()
116                        target_name = "Python P2PDataStore"
117                    elif conn_type == "private-broadcast" and self.trigger_no_relay_peer:
118                        listener = NoRelayP2PInterface()
119                        target_name = "Python NoRelayP2PInterface"
120                        self.trigger_no_relay_peer = False
121                        self.no_relay_peer = listener
122                    else:
123                        listener = P2PInterface()
124                        target_name = "Python P2PInterface"
125                    listener.peer_connect_helper(dstaddr="0.0.0.0", dstport=0, net=self.chain, timeout_factor=self.options.timeout_factor)
126                    listener.peer_connect_send_version(services=P2P_SERVICES)
127
128                    actual_to_addr, actual_to_port = start_p2p_listener(self.network_thread, listener)
129
130                self.log.debug(f"Instructing the SOCKS5 proxy to redirect connection i={i} ({conn_type}) for "
131                               f"{format_addr_port(requested_to_addr, requested_to_port)} to "
132                               f"{format_addr_port(actual_to_addr, actual_to_port)} ({target_name})")
133
134                self.destinations.append({
135                    "requested_to": format_addr_port(requested_to_addr, requested_to_port),
136                    "conn_type": conn_type,
137                    "node": listener,
138                })
139                assert_equal(len(self.destinations), i + 1)
140
141                return {
142                    "actual_to_addr": actual_to_addr,
143                    "actual_to_port": actual_to_port,
144                }
145
146        self.socks5_server = start_socks5_server(destinations_factory)
147
148        self.extra_args = [
149            [
150                # Needed to be able to add CJDNS addresses to addrman (otherwise they are unroutable).
151                "-cjdnsreachable",
152                # Connecting, sending garbage, being disconnected messes up with this test's
153                # check_broadcasts() which waits for a particular Python node to receive a connection.
154                "-v2transport=0",
155                "-test=addrman",
156                "-privatebroadcast",
157                f"-proxy={self.socks5_server.conf.addr[0]}:{self.socks5_server.conf.addr[1]}",
158                # To increase coverage, make it think that the I2P network is reachable so that it
159                # selects such addresses as well. Pick a proxy address where nobody is listening
160                # and connection attempts fail quickly.
161                "-i2psam=127.0.0.1:1",
162            ],
163            [
164                "-connect=0",
165                f"-bind=127.0.0.1:{tor_port(1)}=onion",
166            ],
167        ]
168        super().setup_nodes()
169
170    def setup_network(self):
171        self.setup_nodes()
172
173    def check_broadcasts(self, label, tx, broadcasts_to_expect, skip_destinations):
174        def wait_and_get_destination(n):
175            """Wait for self.destinations[] to have at least n elements and return the 'n'th."""
176            def get_destinations_len():
177                with self.destinations_lock:
178                    return len(self.destinations)
179            self.wait_until(lambda: get_destinations_len() > n)
180            with self.destinations_lock:
181                return self.destinations[n]
182
183        broadcasts_done = 0
184        i = skip_destinations - 1
185        while broadcasts_done < broadcasts_to_expect:
186            i += 1
187            self.log.debug(f"{label}: waiting for outbound connection i={i}")
188            # At this point the connection may not yet have been established (A),
189            # may be active (B), or may have already been closed (C).
190            dest = wait_and_get_destination(i)
191            peer = dest["node"]
192            if peer is None:
193                continue # That is the first private broadcast connection, redirected to nodes[1]
194            peer.wait_until(lambda: peer.message_count["version"] == 1, check_connected=False)
195            # Now it is either (B) or (C).
196            if peer.last_message["version"].nServices != 0:
197                self.log.debug(f"{label}: outbound connection i={i} to {dest['requested_to']} not a private broadcast, ignoring it (maybe feeler or extra block only)")
198                continue
199            self.log.debug(f"{label}: outbound connection i={i} to {dest['requested_to']} must be a private broadcast, checking it")
200            peer.wait_for_disconnect()
201            # Now it is (C).
202            assert_equal(peer.message_count, {
203                "version": 1,
204                "verack": 1,
205                "inv": 1,
206                "tx": 1,
207                "ping": 1
208            })
209            dummy_address = CAddress()
210            dummy_address.nServices = 0
211            assert_equal(peer.last_message["version"].nVersion, P2P_PRIVATE_VERSION)
212            assert_equal(peer.last_message["version"].nServices, 0)
213            assert_equal(peer.last_message["version"].nTime, 0)
214            assert_equal(peer.last_message["version"].addrTo, dummy_address)
215            assert_equal(peer.last_message["version"].addrFrom, dummy_address)
216            assert_equal(peer.last_message["version"].strSubVer, "/pynode:0.0.1/")
217            assert_equal(peer.last_message["version"].nStartingHeight, 0)
218            assert_equal(peer.last_message["version"].relay, 0)
219            assert_equal(peer.last_message["tx"].tx.txid_hex, tx["txid"])
220            self.log.info(f"{label}: ok: outbound connection i={i} is private broadcast of txid={tx['txid']}")
221            broadcasts_done += 1
222
223        # Verify the tx we just observed is tracked in getprivatebroadcastinfo.
224        pbinfo = self.nodes[0].getprivatebroadcastinfo()
225        pending = [t for t in pbinfo["transactions"] if t["txid"] == tx["txid"] and t["wtxid"] == tx["wtxid"]]
226        assert_equal(len(pending), 1)
227        assert_equal(pending[0]["hex"].lower(), tx["hex"].lower())
228        peers = pending[0]["peers"]
229        assert_greater_than_or_equal(len(peers), NUM_PRIVATE_BROADCAST_PER_TX)
230        assert_equal(pending[0]["attempts_remaining"], MAX_PRIVATE_BROADCAST_ATTEMPTS - len(peers))
231        assert all("address" in p and "sent" in p for p in peers)
232        assert_greater_than_or_equal(sum(1 for p in peers if "received" in p), broadcasts_to_expect)
233
234    def run_test(self):
235        tx_originator = self.nodes[0]
236        tx_receiver = self.nodes[1]
237        far_observer = tx_receiver.add_p2p_connection(P2PInterface())
238
239        self.log.info("Test getprivatebroadcastinfo and abortprivatebroadcast fails if the node is running without -privatebroadcast set")
240        assert_raises_rpc_error(-32601, "Private broadcast is not enabled. Ensure you're running Bitcoin Core with -privatebroadcast=1.",
241            tx_receiver.getprivatebroadcastinfo)
242        assert_raises_rpc_error(-32601, "Private broadcast is not enabled. Ensure you're running Bitcoin Core with -privatebroadcast=1.",
243            tx_receiver.abortprivatebroadcast, "00" * 32)
244
245        self.fill_node_addrman(node_index=0, address_types_to_add=[CAddress.NET_IPV4, CAddress.NET_IPV6, CAddress.NET_TORV3, CAddress.NET_I2P, CAddress.NET_CJDNS])
246
247        wallet = MiniWallet(tx_originator)
248
249        txs = wallet.create_self_transfer_chain(chain_length=3)
250        self.log.info(f"Created txid={txs[0]['txid']}: for basic test")