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 """
6 Test how locally submitted transactions are sent to the network when private broadcast is used.
7 """
8 9 import re
10 import time
11 import threading
12 13 from test_framework.p2p import (
14 P2PDataStore,
15 P2PInterface,
16 P2P_SERVICES,
17 start_p2p_listener,
18 )
19 from test_framework.messages import (
20 CAddress,
21 CInv,
22 MSG_WTX,
23 malleate_tx_to_invalid_witness,
24 msg_inv,
25 msg_tx,
26 )
27 from test_framework.netutil import (
28 format_addr_port
29 )
30 from test_framework.script_util import build_malleated_tx_package
31 from test_framework.socks5 import (
32 start_socks5_server,
33 )
34 from test_framework.test_framework import (
35 BitcoinTestFramework,
36 )
37 from test_framework.util import (
38 assert_equal,
39 assert_greater_than_or_equal,
40 assert_not_equal,
41 assert_raises_rpc_error,
42 tor_port,
43 )
44 from test_framework.wallet import (
45 MiniWallet,
46 )
47 48 P2P_PRIVATE_VERSION = 70016
49 NUM_PRIVATE_BROADCAST_PER_TX = 3
50 51 52 class P2PPrivateBroadcast(BitcoinTestFramework):
53 def set_test_params(self):
54 self.disable_autoconnect = False
55 self.num_nodes = 2
56 57 def setup_nodes(self):
58 self.destinations = []
59 60 self.destinations_lock = threading.Lock()
61 62 def find_connection_type_in_debug_log(to_addr, to_port):
63 """
64 Scan the debug log of tx_originator for a connection attempt to to_addr:to_port.
65 Return the connection type (outbound-full-relay, private-broadcast, etc) or
66 None if there is no connection attempt to to_addr:to_port.
67 """
68 with open(self.tx_originator_debug_log_path, mode="r", encoding="utf-8") as debug_log:
69 for line in debug_log.readlines():
70 match = re.match(f".*trying v. connection \\((.+)\\) to \\[?{to_addr}]?:{to_port},.*", line)
71 if match:
72 return match.group(1)
73 return None
74 75 def destinations_factory(requested_to_addr, requested_to_port):
76 """
77 Instruct the SOCKS5 proxy to redirect connections:
78 * The first automatic outbound connection -> P2PDataStore
79 * The first private broadcast connection -> nodes[1]
80 * Anything else -> P2PInterface
81 """
82 conn_type = None
83 def found_connection_in_debug_log():
84 nonlocal conn_type
85 conn_type = find_connection_type_in_debug_log(requested_to_addr, requested_to_port)
86 return conn_type is not None
87 88 self.wait_until(found_connection_in_debug_log)
89 90 with self.destinations_lock:
91 i = len(self.destinations)
92 actual_to_addr = ""
93 actual_to_port = 0
94 listener = None
95 target_name = ""
96 if conn_type == "private-broadcast" and not any(dest["conn_type"] == "private-broadcast" for dest in self.destinations):
97 # Instruct the SOCKS5 server to redirect the first private
98 # broadcast connection from nodes[0] to nodes[1]
99 actual_to_addr = "127.0.0.1" # nodes[1] listen address
100 actual_to_port = tor_port(1) # nodes[1] listen port for Tor
101 target_name = "nodes[1]"
102 else:
103 # Create a Python P2P listening node and instruct the SOCKS5 proxy to
104 # redirect the connection to it. The first outbound connection is used
105 # later to serve GETDATA, thus make it P2PDataStore().
106 if conn_type == "outbound-full-relay" and not any(dest["conn_type"] == "outbound-full-relay" for dest in self.destinations):
107 listener = P2PDataStore()
108 target_name = "Python P2PDataStore"
109 else:
110 listener = P2PInterface()
111 target_name = "Python P2PInterface"
112 listener.peer_connect_helper(dstaddr="0.0.0.0", dstport=0, net=self.chain, timeout_factor=self.options.timeout_factor)
113 listener.peer_connect_send_version(services=P2P_SERVICES)
114 115 actual_to_addr, actual_to_port = start_p2p_listener(self.network_thread, listener)
116 117 self.log.debug(f"Instructing the SOCKS5 proxy to redirect connection i={i} ({conn_type}) for "
118 f"{format_addr_port(requested_to_addr, requested_to_port)} to "
119 f"{format_addr_port(actual_to_addr, actual_to_port)} ({target_name})")
120 121 self.destinations.append({
122 "requested_to": format_addr_port(requested_to_addr, requested_to_port),
123 "conn_type": conn_type,
124 "node": listener,
125 })
126 assert_equal(len(self.destinations), i + 1)
127 128 return {
129 "actual_to_addr": actual_to_addr,
130 "actual_to_port": actual_to_port,
131 }
132 133 self.socks5_server = start_socks5_server(destinations_factory)
134 135 self.extra_args = [
136 [
137 # Needed to be able to add CJDNS addresses to addrman (otherwise they are unroutable).
138 "-cjdnsreachable",
139 # Connecting, sending garbage, being disconnected messes up with this test's
140 # check_broadcasts() which waits for a particular Python node to receive a connection.
141 "-v2transport=0",
142 "-test=addrman",
143 "-privatebroadcast",
144 f"-proxy={self.socks5_server.conf.addr[0]}:{self.socks5_server.conf.addr[1]}",
145 # To increase coverage, make it think that the I2P network is reachable so that it
146 # selects such addresses as well. Pick a proxy address where nobody is listening
147 # and connection attempts fail quickly.
148 "-i2psam=127.0.0.1:1",
149 ],
150 [
151 "-connect=0",
152 f"-bind=127.0.0.1:{tor_port(1)}=onion",
153 ],
154 ]
155 super().setup_nodes()
156 157 def setup_network(self):
158 self.setup_nodes()
159 160 def check_broadcasts(self, label, tx, broadcasts_to_expect, skip_destinations):
161 def wait_and_get_destination(n):
162 """Wait for self.destinations[] to have at least n elements and return the 'n'th."""
163 def get_destinations_len():
164 with self.destinations_lock:
165 return len(self.destinations)
166 self.wait_until(lambda: get_destinations_len() > n)
167 with self.destinations_lock:
168 return self.destinations[n]
169 170 broadcasts_done = 0
171 i = skip_destinations - 1
172 while broadcasts_done < broadcasts_to_expect:
173 i += 1
174 self.log.debug(f"{label}: waiting for outbound connection i={i}")
175 # At this point the connection may not yet have been established (A),
176 # may be active (B), or may have already been closed (C).
177 dest = wait_and_get_destination(i)
178 peer = dest["node"]
179 if peer is None:
180 continue # That is the first private broadcast connection, redirected to nodes[1]
181 peer.wait_until(lambda: peer.message_count["version"] == 1, check_connected=False)
182 # Now it is either (B) or (C).
183 if peer.last_message["version"].nServices != 0:
184 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)")
185 continue
186 self.log.debug(f"{label}: outbound connection i={i} to {dest['requested_to']} must be a private broadcast, checking it")
187 peer.wait_for_disconnect()
188 # Now it is (C).
189 assert_equal(peer.message_count, {
190 "version": 1,
191 "verack": 1,
192 "inv": 1,
193 "tx": 1,
194 "ping": 1
195 })
196 dummy_address = CAddress()
197 dummy_address.nServices = 0
198 assert_equal(peer.last_message["version"].nVersion, P2P_PRIVATE_VERSION)
199 assert_equal(peer.last_message["version"].nServices, 0)
200 assert_equal(peer.last_message["version"].nTime, 0)
201 assert_equal(peer.last_message["version"].addrTo, dummy_address)
202 assert_equal(peer.last_message["version"].addrFrom, dummy_address)
203 assert_equal(peer.last_message["version"].strSubVer, "/pynode:0.0.1/")
204 assert_equal(peer.last_message["version"].nStartingHeight, 0)
205 assert_equal(peer.last_message["version"].relay, 0)
206 assert_equal(peer.last_message["tx"].tx.txid_hex, tx["txid"])
207 self.log.info(f"{label}: ok: outbound connection i={i} is private broadcast of txid={tx['txid']}")
208 broadcasts_done += 1
209 210 # Verify the tx we just observed is tracked in getprivatebroadcastinfo.
211 pbinfo = self.nodes[0].getprivatebroadcastinfo()
212 pending = [t for t in pbinfo["transactions"] if t["txid"] == tx["txid"] and t["wtxid"] == tx["wtxid"]]
213 assert_equal(len(pending), 1)
214 assert_equal(pending[0]["hex"].lower(), tx["hex"].lower())
215 peers = pending[0]["peers"]
216 assert len(peers) >= NUM_PRIVATE_BROADCAST_PER_TX
217 assert all("address" in p and "sent" in p for p in peers)
218 assert_greater_than_or_equal(sum(1 for p in peers if "received" in p), broadcasts_to_expect)
219 220 def run_test(self):
221 tx_originator = self.nodes[0]
222 self.tx_originator_debug_log_path = tx_originator.debug_log_path
223 tx_receiver = self.nodes[1]
224 far_observer = tx_receiver.add_p2p_connection(P2PInterface())
225 226 self.log.info("Test getprivatebroadcastinfo and abortprivatebroadcast fails if the node is running without -privatebroadcast set")
227 assert_raises_rpc_error(-32601, "Private broadcast is not enabled. Ensure you're running Bitcoin Core with -privatebroadcast=1.",
228 tx_receiver.getprivatebroadcastinfo)
229 assert_raises_rpc_error(-32601, "Private broadcast is not enabled. Ensure you're running Bitcoin Core with -privatebroadcast=1.",
230 tx_receiver.abortprivatebroadcast, "00" * 32)
231 232 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])
233 234 wallet = MiniWallet(tx_originator)
235 236 txs = wallet.create_self_transfer_chain(chain_length=3)
237 self.log.info(f"Created txid={txs[0]['txid']}: for basic test")
238 self.log.info(f"Created txid={txs[1]['txid']}: for broadcast with dependency in mempool + rebroadcast")
239 self.log.info(f"Created txid={txs[2]['txid']}: for broadcast with dependency not in mempool")
240 tx_originator.sendrawtransaction(hexstring=txs[0]["hex"], maxfeerate=0.1)
241 242 self.log.info("First private broadcast: waiting for the transaction to reach the recipient")
243 self.wait_until(lambda: len(tx_receiver.getrawmempool()) > 0)
244 self.log.info("First private broadcast: the recipient received the transaction")
245 far_observer.wait_for_tx(txs[0]["txid"])
246 self.log.info("First private broadcast: the recipient further relayed the transaction")
247 248 # One already checked above, check the other NUM_PRIVATE_BROADCAST_PER_TX - 1 broadcasts.
249 self.check_broadcasts("Basic", txs[0], NUM_PRIVATE_BROADCAST_PER_TX - 1, 0)
250 251 self.log.info("Resending the same transaction via RPC again (it is not in the mempool yet)")
252 ignoring_msg = f"Ignoring unnecessary request to schedule an already scheduled transaction: txid={txs[0]['txid']}, wtxid={txs[0]['wtxid']}"
253 with tx_originator.busy_wait_for_debug_log(expected_msgs=[ignoring_msg.encode()]):
254 tx_originator.sendrawtransaction(hexstring=txs[0]["hex"], maxfeerate=0)
255 256 self.log.info("Sending a malleated transaction with an invalid witness via RPC")
257 malleated_invalid = malleate_tx_to_invalid_witness(txs[0])
258 assert_raises_rpc_error(-26, "mempool-script-verify-flag-failed",
259 tx_originator.sendrawtransaction,
260 hexstring=malleated_invalid.serialize_with_witness().hex(),
261 maxfeerate=0.1)
262 263 self.log.info("Checking that the transaction is not in the originator node's mempool")
264 assert_equal(len(tx_originator.getrawmempool()), 0)
265 266 wtxid_int = int(txs[0]["wtxid"], 16)
267 inv = CInv(MSG_WTX, wtxid_int)
268 269 tx_returner = None # First outbound-full-relay, will be P2PDataStore.
270 other_peer = None # Any other outbound-full-relay, we use the second one.
271 272 def set_tx_returner_and_other():
273 nonlocal tx_returner
274 nonlocal other_peer
275 tx_returner = None
276 other_peer = None
277 with self.destinations_lock:
278 for dest in self.destinations:
279 if dest["conn_type"] == "outbound-full-relay" and dest["node"] is not None:
280 if tx_returner is None:
281 assert(type(dest["node"]) is P2PDataStore)
282 tx_returner = dest["node"]
283 else:
284 assert(type(dest["node"]) is P2PInterface)
285 other_peer = dest["node"]
286 return True
287 return False
288 289 self.wait_until(set_tx_returner_and_other)
290 291 tx_returner.wait_for_connect()
292 other_peer.wait_for_connect()
293 294 self.log.info("Sending INV and waiting for GETDATA from node")
295 tx_returner.tx_store[wtxid_int] = txs[0]["tx"]
296 assert "getdata" not in tx_returner.last_message
297 received_back_msg = f"Received our privately broadcast transaction (txid={txs[0]['txid']}) from the network"
298 with tx_originator.assert_debug_log(expected_msgs=[received_back_msg]):
299 tx_returner.send_without_ping(msg_inv([inv]))
300 tx_returner.wait_until(lambda: "getdata" in tx_returner.last_message)
301 self.wait_until(lambda: len(tx_originator.getrawmempool()) > 0)
302 303 self.log.info("Waiting for normal broadcast to another peer")
304 other_peer.wait_for_inv([inv])
305 306 self.log.info("Checking getprivatebroadcastinfo no longer reports the transaction after it is received back")
307 pbinfo = tx_originator.getprivatebroadcastinfo()
308 pending = [t for t in pbinfo["transactions"] if t["txid"] == txs[0]["txid"] and t["wtxid"] == txs[0]["wtxid"]]
309 assert_equal(len(pending), 0)
310 311 self.log.info("Sending a transaction that is already in the mempool")
312 skip_destinations = len(self.destinations)
313 tx_originator.sendrawtransaction(hexstring=txs[0]["hex"], maxfeerate=0)
314 self.check_broadcasts("Broadcast of mempool transaction", txs[0], NUM_PRIVATE_BROADCAST_PER_TX, skip_destinations)
315 316 self.log.info("Sending a transaction with a dependency in the mempool")
317 skip_destinations = len(self.destinations)
318 tx_originator.sendrawtransaction(hexstring=txs[1]["hex"], maxfeerate=0.1)
319 self.check_broadcasts("Dependency in mempool", txs[1], NUM_PRIVATE_BROADCAST_PER_TX, skip_destinations)
320 321 self.log.info("Sending a transaction with a dependency not in the mempool (should be rejected)")
322 assert_equal(len(tx_originator.getrawmempool()), 1)
323 assert_raises_rpc_error(-25, "bad-txns-inputs-missingorspent",
324 tx_originator.sendrawtransaction, hexstring=txs[2]["hex"], maxfeerate=0.1)
325 assert_raises_rpc_error(-25, "bad-txns-inputs-missingorspent",
326 tx_originator.sendrawtransaction, hexstring=txs[2]["hex"], maxfeerate=0)
327 328 # Since txs[1] has not been received back by tx_originator,
329 # it should be re-broadcast after a while. Advance tx_originator's clock
330 # to trigger a re-broadcast. Should be more than the maximum returned by
331 # NextTxBroadcast() in net_processing.cpp.
332 self.log.info("Checking that rebroadcast works")
333 delta = 20 * 60 # 20min
334 skip_destinations = len(self.destinations)
335 rebroadcast_msg = f"Reattempting broadcast of stale txid={txs[1]['txid']}"
336 with tx_originator.busy_wait_for_debug_log(expected_msgs=[rebroadcast_msg.encode()]):
337 tx_originator.setmocktime(int(time.time()) + delta)
338 tx_originator.mockscheduler(delta)
339 self.check_broadcasts("Rebroadcast", txs[1], 1, skip_destinations)
340 tx_originator.setmocktime(0) # Let the clock tick again (it will go backwards due to this).
341 342 self.log.info("Sending a pair of transactions with the same txid but different valid wtxids via RPC")
343 parent = wallet.create_self_transfer()["tx"]
344 parent_amount = parent.vout[0].nValue - 10000
345 child_amount = parent_amount - 10000
346 siblings_parent, sibling1, sibling2 = build_malleated_tx_package(
347 parent=parent,
348 rebalance_parent_output_amount=parent_amount,
349 child_amount=child_amount)
350 self.log.info(f" - sibling1: txid={sibling1.txid_hex}, wtxid={sibling1.wtxid_hex}")
351 self.log.info(f" - sibling2: txid={sibling2.txid_hex}, wtxid={sibling2.wtxid_hex}")
352 assert_equal(sibling1.txid_hex, sibling2.txid_hex)
353 assert_not_equal(sibling1.wtxid_hex, sibling2.wtxid_hex)
354 assert_equal(len(tx_originator.getrawmempool()), 1)
355 tx_returner.send_without_ping(msg_tx(siblings_parent))
356 self.wait_until(lambda: len(tx_originator.getrawmempool()) > 1)
357 self.log.info(" - siblings' parent added to the mempool")
358 tx_originator.sendrawtransaction(hexstring=sibling1.serialize_with_witness().hex(), maxfeerate=0.1)
359 self.log.info(" - sent sibling1: ok")
360 tx_originator.sendrawtransaction(hexstring=sibling2.serialize_with_witness().hex(), maxfeerate=0.1)
361 self.log.info(" - sent sibling2: ok")
362 363 self.log.info("Checking abortprivatebroadcast removes a pending private-broadcast transaction")
364 tx_abort = wallet.create_self_transfer()
365 tx_originator.sendrawtransaction(hexstring=tx_abort["hex"], maxfeerate=0.1)
366 assert tx_abort["wtxid"] in [t["wtxid"] for t in tx_originator.getprivatebroadcastinfo()["transactions"]]
367 abort_res = tx_originator.abortprivatebroadcast(tx_abort["txid"])
368 assert_equal(len(abort_res["removed_transactions"]), 1)
369 assert_equal(abort_res["removed_transactions"][0]["txid"], tx_abort["txid"])
370 assert_equal(abort_res["removed_transactions"][0]["wtxid"], tx_abort["wtxid"])
371 assert_equal(abort_res["removed_transactions"][0]["hex"].lower(), tx_abort["hex"].lower())
372 assert all(t["wtxid"] != tx_abort["wtxid"] for t in tx_originator.getprivatebroadcastinfo()["transactions"])
373 374 self.log.info("Checking abortprivatebroadcast fails for non-existent transaction")
375 assert_raises_rpc_error(
376 -5,
377 "Transaction not in private broadcast queue",
378 tx_originator.abortprivatebroadcast,
379 "0" * 64,
380 )
381 382 # Stop the SOCKS5 proxy server to avoid it being upset by the bitcoin
383 # node disconnecting in the middle of the SOCKS5 handshake when we
384 # restart below.
385 self.socks5_server.stop()
386 387 self.log.info("Trying to send a transaction when none of Tor or I2P is reachable")
388 self.restart_node(0, extra_args=[
389 "-privatebroadcast",
390 "-v2transport=0",
391 # A location where definitely a Tor control is not listening. This would allow
392 # Bitcoin Core to start, hoping/assuming that the location of the Tor proxy
393 # may be retrieved after startup from the Tor control, but it will not be, so
394 # the RPC should throw.
395 "-torcontrol=127.0.0.1:1",
396 "-listenonion",
397 ])
398 assert_raises_rpc_error(-1, "none of the Tor or I2P networks is reachable",
399 tx_originator.sendrawtransaction, hexstring=txs[0]["hex"], maxfeerate=0.1)
400 401 402 if __name__ == "__main__":
403 P2PPrivateBroadcast(__file__).main()
404