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 """Class for bitcoind node under test"""
6 7 import contextlib
8 import decimal
9 import errno
10 from enum import Enum
11 import json
12 import logging
13 import os
14 import platform
15 import re
16 import subprocess
17 import tempfile
18 import time
19 import urllib.parse
20 import collections
21 import sys
22 from collections.abc import Iterable
23 from pathlib import Path
24 25 from .authproxy import (
26 AuthServiceProxy,
27 JSONRPCException,
28 serialization_fallback,
29 )
30 from . import coverage
31 from .messages import NODE_P2P_V2
32 from .p2p import P2P_SERVICES, P2P_SUBVERSION
33 from .util import (
34 MAX_NODES,
35 assert_equal,
36 assert_not_equal,
37 append_config,
38 delete_cookie_file,
39 get_auth_cookie,
40 rpc_port,
41 wait_until_helper_internal,
42 p2p_port,
43 tor_port,
44 )
45 46 BITCOIND_PROC_WAIT_TIMEOUT = 60
47 # The size of the blocks xor key
48 # from InitBlocksdirXorKey::xor_key.size()
49 NUM_XOR_BYTES = 8
50 # Many systems have a 128kB limit for a command size. Depending on the
51 # platform, this limit may be larger or smaller. Moreover, when using the
52 # 'bitcoin' command, it may internally insert more args, which must be
53 # accounted for. There is no need to pick the largest possible value here
54 # anyway and it should be fine to set it to 1kB in tests.
55 TEST_CLI_MAX_ARG_SIZE = 1024
56 57 # The null blocks key (all 0s)
58 NULL_BLK_XOR_KEY = bytes([0] * NUM_XOR_BYTES)
59 BITCOIN_PID_FILENAME_DEFAULT = "bitcoind.pid"
60 61 if sys.platform.startswith("linux"):
62 UNIX_PATH_MAX = 108 # includes the trailing NUL
63 elif sys.platform.startswith(("darwin", "freebsd", "netbsd", "openbsd")):
64 UNIX_PATH_MAX = 104
65 else: # safest portable value
66 UNIX_PATH_MAX = 92
67 68 69 class FailedToStartError(Exception):
70 """Raised when a node fails to start correctly."""
71 72 73 class ErrorMatch(Enum):
74 FULL_TEXT = 1
75 FULL_REGEX = 2
76 PARTIAL_REGEX = 3
77 78 79 RPCConnectionType = Enum("RPCConnectionType", ["AUTO", "AUTHPROXY", "CLI"])
80 81 82 class TestNode():
83 """A class for representing a bitcoind node under test.
84 85 This class contains:
86 87 - state about the node (whether it's running, etc)
88 - a Python subprocess.Popen object representing the running process
89 - an RPC connection to the node
90 - one or more P2P connections to the node
91 92 93 To make things easier for the test writer, any unrecognised messages will
94 be dispatched to the RPC connection."""
95 96 def __init__(
97 self,
98 i,
99 datadir_path,
100 *,
101 chain,
102 rpchost,
103 timewait,
104 timeout_factor,
105 binaries,
106 coverage_dir,
107 cwd,
108 extra_conf=None,
109 extra_args=None,
110 use_cli=False,
111 version=None,
112 v2transport=False,
113 uses_wallet=False,
114 ipcbind=False,
115 ):
116 self.index = i
117 self.datadir_path = datadir_path
118 self.bitcoinconf = self.datadir_path / "bitcoin.conf"
119 self.stdout_dir = self.datadir_path / "stdout"
120 self.stderr_dir = self.datadir_path / "stderr"
121 self.chain = chain
122 self.rpchost = rpchost
123 self.rpc_timeout = timewait # Already multiplied by timeout_factor
124 self.timeout_factor = timeout_factor
125 self.binaries = binaries
126 self.coverage_dir = coverage_dir
127 self.cwd = cwd
128 self.has_explicit_bind = False
129 if extra_conf is not None:
130 append_config(self.datadir_path, extra_conf)
131 # Remember if there is bind=... in the config file.
132 self.has_explicit_bind = any(e.startswith("bind=") for e in extra_conf)
133 # Most callers will just need to add extra args to the standard list below.
134 # For those callers that need more flexibility, they can just set the args property directly.
135 # Note that common args are set in the config file (see initialize_datadir)
136 self.extra_args = extra_args
137 self.version = version
138 # Configuration for logging is set as command-line args rather than in the bitcoin.conf file.
139 # This means that starting a bitcoind using the temp dir to debug a failed test won't
140 # spam debug.log.
141 self.args = self.binaries.node_argv(need_ipc=ipcbind) + [
142 f"-datadir={self.datadir_path}",
143 "-logtimemicros",
144 "-debug",
145 "-debugexclude=leveldb",
146 "-debugexclude=rand",
147 "-uacomment=testnode%d" % i, # required for subversion uniqueness across peers
148 ]
149 if uses_wallet is not None and not uses_wallet:
150 self.args.append("-disablewallet")
151 152 self.ipc_tmp_dir = None
153 if ipcbind:
154 self.ipc_socket_path = self.chain_path / "node.sock"
155 if len(os.fsencode(self.ipc_socket_path)) < UNIX_PATH_MAX:
156 self.args.append("-ipcbind=unix")
157 else:
158 # Work around default CI path exceeding maximum socket path length.
159 self.ipc_tmp_dir = tempfile.TemporaryDirectory(prefix="test-ipc-")
160 self.ipc_socket_path = Path(self.ipc_tmp_dir.name) / "node.sock"
161 self.args.append(f"-ipcbind=unix:{self.ipc_socket_path}")
162 163 if self.version_is_at_least(190000):
164 self.args.append("-logthreadnames")
165 if self.version_is_at_least(219900):
166 self.args.append("-logsourcelocations")
167 if self.version_is_at_least(239000):
168 self.args.append("-loglevel=trace")
169 if self.version_is_at_least(290100):
170 self.args.append("-nologratelimit")
171 172 # Default behavior from global -v2transport flag is added to args to persist it over restarts.
173 # May be overwritten in individual tests, using extra_args.
174 self.default_to_v2 = v2transport
175 if self.version_is_at_least(260000):
176 # 26.0 and later support v2transport
177 if v2transport:
178 self.args.append("-v2transport=1")
179 else:
180 self.args.append("-v2transport=0")
181 # if v2transport is requested via global flag but not supported for node version, ignore it
182 183 self.cli = None
184 self.use_cli = use_cli
185 186 self.running = False
187 self.process = None
188 self.rpc_connected = False
189 self._rpc = None # Should usually not be accessed directly in tests to allow for --usecli mode
190 self.reuse_http_connections = True # Must be set before create_new_rpc_connection(), i.e. before restarting node
191 self.url = None
192 self.log = logging.getLogger('TestFramework.node%d' % i)
193 194 self.p2ps = []
195 196 self.mocktime = None
197 198 AddressKeyPair = collections.namedtuple('AddressKeyPair', ['address', 'key'])
199 PRIV_KEYS = [
200 # address , privkey
201 AddressKeyPair('mjTkW3DjgyZck4KbiRusZsqTgaYTxdSz6z', 'cVpF924EspNh8KjYsfhgY96mmxvT6DgdWiTYMtMjuM74hJaU5psW'),
202 AddressKeyPair('msX6jQXvxiNhx3Q62PKeLPrhrqZQdSimTg', 'cUxsWyKyZ9MAQTaAhUQWJmBbSvHMwSmuv59KgxQV7oZQU3PXN3KE'),
203 AddressKeyPair('mnonCMyH9TmAsSj3M59DsbH8H63U3RKoFP', 'cTrh7dkEAeJd6b3MRX9bZK8eRmNqVCMH3LSUkE3dSFDyzjU38QxK'),
204 AddressKeyPair('mqJupas8Dt2uestQDvV2NH3RU8uZh2dqQR', 'cVuKKa7gbehEQvVq717hYcbE9Dqmq7KEBKqWgWrYBa2CKKrhtRim'),
205 AddressKeyPair('msYac7Rvd5ywm6pEmkjyxhbCDKqWsVeYws', 'cQDCBuKcjanpXDpCqacNSjYfxeQj8G6CAtH1Dsk3cXyqLNC4RPuh'),
206 AddressKeyPair('n2rnuUnwLgXqf9kk2kjvVm8R5BZK1yxQBi', 'cQakmfPSLSqKHyMFGwAqKHgWUiofJCagVGhiB4KCainaeCSxeyYq'),
207 AddressKeyPair('myzuPxRwsf3vvGzEuzPfK9Nf2RfwauwYe6', 'cQMpDLJwA8DBe9NcQbdoSb1BhmFxVjWD5gRyrLZCtpuF9Zi3a9RK'),
208 AddressKeyPair('mumwTaMtbxEPUswmLBBN3vM9oGRtGBrys8', 'cSXmRKXVcoouhNNVpcNKFfxsTsToY5pvB9DVsFksF1ENunTzRKsy'),
209 AddressKeyPair('mpV7aGShMkJCZgbW7F6iZgrvuPHjZjH9qg', 'cSoXt6tm3pqy43UMabY6eUTmR3eSUYFtB2iNQDGgb3VUnRsQys2k'),
210 AddressKeyPair('mq4fBNdckGtvY2mijd9am7DRsbRB4KjUkf', 'cN55daf1HotwBAgAKWVgDcoppmUNDtQSfb7XLutTLeAgVc3u8hik'),
211 AddressKeyPair('mpFAHDjX7KregM3rVotdXzQmkbwtbQEnZ6', 'cT7qK7g1wkYEMvKowd2ZrX1E5f6JQ7TM246UfqbCiyF7kZhorpX3'),
212 AddressKeyPair('mzRe8QZMfGi58KyWCse2exxEFry2sfF2Y7', 'cPiRWE8KMjTRxH1MWkPerhfoHFn5iHPWVK5aPqjW8NxmdwenFinJ'),
213 ]
214 215 def get_deterministic_priv_key(self):
216 """Return a deterministic priv key in base58, that only depends on the node's index"""
217 assert_equal(len(self.PRIV_KEYS), MAX_NODES)
218 return self.PRIV_KEYS[self.index]
219 220 def _node_msg(self, msg: str) -> str:
221 """Return a modified msg that identifies this node by its index as a debugging aid."""
222 return "[node %d] %s" % (self.index, msg)
223 224 def _raise_assertion_error(self, msg: str):
225 """Raise an AssertionError with msg modified to identify this node."""
226 raise AssertionError(self._node_msg(msg))
227 228 def __del__(self):
229 # Ensure that we don't leave any bitcoind processes lying around after
230 # the test ends
231 if self.process:
232 # Should only happen on test failure
233 # Avoid using logger, as that may have already been shutdown when
234 # this destructor is called.
235 print(self._node_msg("Cleaning up leftover process"), file=sys.stderr)
236 self.process.kill()
237 238 def __getattr__(self, name):
239 """Dispatches any unrecognised messages to the RPC connection or a CLI instance."""
240 if self.use_cli:
241 return getattr(self.cli, name)
242 else:
243 assert self.rpc_connected and self._rpc is not None, self._node_msg("Error: no RPC connection")
244 return getattr(self._rpc, name)
245 246 def start(self, extra_args=None, *, cwd=None, stdout=None, stderr=None, env=None, **kwargs):
247 """Start the node."""
248 if extra_args is None:
249 extra_args = self.extra_args
250 251 # If listening and no -bind is given, then bitcoind would bind P2P ports on
252 # 0.0.0.0:P and 127.0.0.1:P+1 (for incoming Tor connections), where P is
253 # a unique port chosen by the test framework and configured as port=P in
254 # bitcoin.conf. To avoid collisions, change it to 127.0.0.1:tor_port().
255 will_listen = all(e != "-nolisten" and e != "-listen=0" for e in extra_args)
256 has_explicit_bind = self.has_explicit_bind or any(e.startswith("-bind=") for e in extra_args)
257 if will_listen and not has_explicit_bind:
258 extra_args.append(f"-bind=0.0.0.0:{p2p_port(self.index)}")
259 extra_args.append(f"-bind=127.0.0.1:{tor_port(self.index)}=onion")
260 261 self.use_v2transport = "-v2transport=1" in extra_args or (self.default_to_v2 and "-v2transport=0" not in extra_args)
262 263 # Add a new stdout and stderr file each time bitcoind is started
264 if stderr is None:
265 stderr = tempfile.NamedTemporaryFile(dir=self.stderr_dir, delete=False)
266 if stdout is None:
267 stdout = tempfile.NamedTemporaryFile(dir=self.stdout_dir, delete=False)
268 self.stderr = stderr
269 self.stdout = stdout
270 271 if cwd is None:
272 cwd = self.cwd
273 274 # Delete any existing cookie file -- if such a file exists (eg due to
275 # unclean shutdown), it will get overwritten anyway by bitcoind, and
276 # potentially interfere with our attempt to authenticate
277 delete_cookie_file(self.datadir_path, self.chain)
278 279 # add environment variable LIBC_FATAL_STDERR_=1 so that libc errors are written to stderr and not the terminal
280 subp_env = dict(os.environ, LIBC_FATAL_STDERR_="1")
281 if env is not None:
282 subp_env.update(env)
283 284 self.process = subprocess.Popen(self.args + extra_args, env=subp_env, stdout=stdout, stderr=stderr, cwd=cwd, **kwargs)
285 286 self.running = True
287 self.log.debug("bitcoind started, waiting for RPC to come up")
288 289 def create_new_rpc_connection(self, *, mode="AUTO", client_timeout=None):
290 """Create an additional RPC connection, likely to be used in a new thread."""
291 mode = RPCConnectionType[mode]
292 if mode == RPCConnectionType.AUTO:
293 mode = RPCConnectionType.CLI if self.use_cli else RPCConnectionType.AUTHPROXY
294 client_timeout = client_timeout or (self.rpc_timeout // 2) # Shorter timeout to allow for one retry in case of ETIMEDOUT
295 host = "127.0.0.1"
296 port = rpc_port(self.index)
297 if self.rpchost:
298 parts = self.rpchost.split(":")
299 if len(parts) == 2:
300 host, port = parts
301 else:
302 host = self.rpchost
303 if mode == RPCConnectionType.AUTHPROXY:
304 rpc_u, rpc_p = get_auth_cookie(self.datadir_path, self.chain)
305 url = f"http://{rpc_u}:{rpc_p}@{host}:{port}"
306 proxy = AuthServiceProxy(url, timeout=int(client_timeout))
307 coverage_logfile = coverage.get_filename(self.coverage_dir, self.index) if self.coverage_dir else None
308 rpc = coverage.AuthServiceProxyWrapper(proxy, url, coverage_logfile)
309 rpc.auth_service_proxy_instance.reuse_http_connections = self.reuse_http_connections
310 return rpc
311 else: # mode==CLI
312 return TestNodeCLI(self.binaries)(
313 f"-datadir={self.datadir_path}",
314 f"-rpcclienttimeout={client_timeout}",
315 f"-rpcconnect={host}",
316 f"-rpcport={port}",
317 )
318 319 def wait_for_rpc_connection(self, *, wait_for_import=True):
320 """Sets up an RPC connection to the bitcoind process. Returns False if unable to connect."""
321 # Poll at a rate of four times per second
322 poll_per_s = 4
323 324 suppressed_errors = collections.defaultdict(int)
325 latest_error = None
326 def suppress_error(category: str, e: Exception):
327 suppressed_errors[category] += 1
328 return (category, repr(e))
329 330 for _ in range(poll_per_s * self.rpc_timeout):
331 if self.process.poll() is not None:
332 # Attach abrupt shutdown error/s to the exception message
333 self.stderr.seek(0)
334 str_error = ''.join(line.decode('utf-8') for line in self.stderr)
335 str_error += "************************\n" if str_error else ''
336 337 raise FailedToStartError(self._node_msg(
338 f'bitcoind exited with status {self.process.returncode} during initialization. {str_error}'))
339 try:
340 rpc = self.create_new_rpc_connection(mode="AUTHPROXY")
341 rpc.getblockcount()
342 # If the call to getblockcount() succeeds then the RPC connection is up
343 if self.version_is_at_least(190000) and wait_for_import:
344 # getmempoolinfo.loaded is available since commit
345 # bb8ae2c (version 0.19.0)
346 self.wait_until(lambda: rpc.getmempoolinfo()['loaded'])
347 # Wait for the node to finish reindex, block import, and
348 # loading the mempool. Usually importing happens fast or
349 # even "immediate" when the node is started. However, there
350 # is no guarantee and sometimes ImportBlocks might finish
351 # later. This is going to cause intermittent test failures,
352 # because generally the tests assume the node is fully
353 # ready after being started.
354 #
355 # For example, the node will reject block messages from p2p
356 # when it is still importing with the error "Unexpected
357 # block message received"
358 #
359 # The wait is done here to make tests as robust as possible
360 # and prevent racy tests and intermittent failures as much
361 # as possible. Some tests might not need this, but the
362 # overhead is trivial, and the added guarantees are worth
363 # the minimal performance cost.
364 self.log.debug("RPC successfully started")
365 # Set rpc_connected even if we are in use_cli mode so that we know we can call self.stop() if needed.
366 self.rpc_connected = True
367 self.url = rpc.rpc_url
368 self.cli = self.create_new_rpc_connection(mode="CLI")
369 if self.use_cli:
370 return
371 self._rpc = rpc
372 return
373 except JSONRPCException as e:
374 # Suppress these as they are expected during initialization.
375 # -28 RPC in warmup
376 # -342 Service unavailable, could be starting up or shutting down
377 if e.error['code'] not in [-28, -342]:
378 raise # unknown JSON RPC exception
379 latest_error = suppress_error(f"JSONRPCException {e.error['code']}", e)
380 except OSError as e:
381 error_num = e.errno
382 if error_num is None:
383 # Work around issue where socket timeouts don't have errno set.
384 # https://github.com/python/cpython/issues/109601
385 if isinstance(e, TimeoutError):
386 error_num = errno.ETIMEDOUT
387 # http.client.RemoteDisconnected inherits from this type and
388 # doesn't specify errno.
389 elif isinstance(e, ConnectionResetError):
390 error_num = errno.ECONNRESET
391 # Windows can raise this while bitcoind shuts down during startup.
392 elif isinstance(e, ConnectionAbortedError):
393 error_num = errno.ECONNABORTED
394 395 # Suppress similarly to the above JSONRPCException errors.
396 if error_num not in [
397 errno.ECONNABORTED, # Treat identical to ECONNRESET
398 errno.ECONNRESET, # This might happen when the RPC server is in warmup,
399 # but shut down before the call to getblockcount succeeds.
400 errno.ETIMEDOUT, # Treat identical to ECONNRESET
401 errno.ECONNREFUSED # Port not yet open?
402 ]:
403 raise # unknown OS error
404 latest_error = suppress_error(f"OSError {errno.errorcode[error_num]}", e)
405 except ValueError as e:
406 # Suppress if cookie file isn't generated yet and no rpcuser or rpcpassword; bitcoind may be starting.
407 if "No RPC credentials" not in str(e):
408 raise
409 latest_error = suppress_error("missing_credentials", e)
410 time.sleep(1.0 / poll_per_s)
411 self._raise_assertion_error(f"Unable to connect to bitcoind after {self.rpc_timeout}s (ignored errors: {dict(suppressed_errors)!s}{'' if latest_error is None else f', latest: {latest_error[0]!r}/{latest_error[1]}'})")
412 413 def wait_for_cookie_credentials(self):
414 """Ensures auth cookie credentials can be read, e.g. for testing CLI with -rpcwait before RPC connection is up."""
415 self.log.debug("Waiting for cookie credentials")
416 # Poll at a rate of four times per second.
417 poll_per_s = 4
418 for _ in range(poll_per_s * self.rpc_timeout):
419 try:
420 get_auth_cookie(self.datadir_path, self.chain)
421 self.log.debug("Cookie credentials successfully retrieved")
422 return
423 except ValueError: # cookie file not found and no rpcuser or rpcpassword; bitcoind is still starting
424 pass # so we continue polling until RPC credentials are retrieved
425 time.sleep(1.0 / poll_per_s)
426 self._raise_assertion_error("Unable to retrieve cookie credentials after {}s".format(self.rpc_timeout))
427 428 def generate(self, nblocks, maxtries=1000000, **kwargs):
429 self.log.debug("TestNode.generate() dispatches `generate` call to `generatetoaddress`")
430 return self.generatetoaddress(nblocks=nblocks, address=self.get_deterministic_priv_key().address, maxtries=maxtries, **kwargs)
431 432 def generateblock(self, *args, called_by_framework, **kwargs):
433 assert called_by_framework, "Direct call of this mining RPC is discouraged. Please use one of the self.generate* methods on the test framework, which sync the nodes to avoid intermittent test issues. You may use sync_fun=self.no_op to disable the sync explicitly."
434 return self.__getattr__('generateblock')(*args, **kwargs)
435 436 def generatetoaddress(self, *args, called_by_framework, **kwargs):
437 assert called_by_framework, "Direct call of this mining RPC is discouraged. Please use one of the self.generate* methods on the test framework, which sync the nodes to avoid intermittent test issues. You may use sync_fun=self.no_op to disable the sync explicitly."
438 return self.__getattr__('generatetoaddress')(*args, **kwargs)
439 440 def generatetodescriptor(self, *args, called_by_framework, **kwargs):
441 assert called_by_framework, "Direct call of this mining RPC is discouraged. Please use one of the self.generate* methods on the test framework, which sync the nodes to avoid intermittent test issues. You may use sync_fun=self.no_op to disable the sync explicitly."
442 return self.__getattr__('generatetodescriptor')(*args, **kwargs)
443 444 def setmocktime(self, timestamp):
445 """Wrapper for setmocktime RPC, sets self.mocktime"""
446 if timestamp == 0:
447 # setmocktime(0) resets to system time.
448 self.mocktime = None
449 else:
450 self.mocktime = timestamp
451 return self.__getattr__('setmocktime')(timestamp)
452 453 def get_wallet_rpc(self, wallet_name):
454 if self.use_cli:
455 return self.cli("-rpcwallet={}".format(wallet_name))
456 else:
457 assert self.rpc_connected and self._rpc, self._node_msg("RPC not connected")
458 wallet_path = "wallet/{}".format(urllib.parse.quote(wallet_name))
459 return self._rpc / wallet_path
460 461 def version_is_at_least(self, ver):
462 return self.version is None or self.version >= ver
463 464 def stop_node(self, expected_stderr='', *, wait=0, wait_until_stopped=True):
465 """Stop the node."""
466 if not self.running:
467 return
468 assert self.rpc_connected, self._node_msg(
469 "Should only call stop_node() on a running node after wait_for_rpc_connection() succeeded. "
470 f"Did you forget to call the latter after start()? Not connected to process: {self.process.pid}")
471 self.log.debug("Stopping node")
472 # Do not use wait argument when testing older nodes, e.g. in wallet_backwards_compatibility.py
473 if self.version_is_at_least(180000):
474 self.stop(wait=wait)
475 else:
476 self.stop()
477 478 del self.p2ps[:]
479 480 assert (not expected_stderr) or wait_until_stopped # Must wait to check stderr
481 if wait_until_stopped:
482 self.wait_until_stopped(expected_stderr=expected_stderr)
483 484 def is_node_stopped(self, *, expected_stderr="", expected_ret_code=0):
485 """Checks whether the node has stopped.
486 487 Returns True if the node has stopped. False otherwise.
488 489 If the process has exited, asserts that the exit code matches
490 `expected_ret_code` (which may be a single value or an iterable of values),
491 and that stderr matches `expected_stderr` exactly or, if a regex pattern is
492 provided, contains the pattern.
493 494 This method is responsible for freeing resources (self.process)."""
495 if not self.running:
496 return True
497 return_code = self.process.poll()
498 if return_code is None:
499 return False
500 501 # process has stopped. Assert that it didn't return an error code.
502 if not isinstance(expected_ret_code, Iterable):
503 expected_ret_code = (expected_ret_code,)
504 assert return_code in expected_ret_code, self._node_msg(
505 f"Node returned unexpected exit code ({return_code}) vs ({expected_ret_code}) when stopping")
506 # Check that stderr is as expected
507 self.stderr.seek(0)
508 stderr = self.stderr.read().decode('utf-8').strip()
509 if isinstance(expected_stderr, re.Pattern):
510 if not expected_stderr.search(stderr):
511 raise AssertionError(f"Unexpected stderr {stderr!r} does not contain {expected_stderr.pattern!r}")
512 elif stderr != expected_stderr:
513 raise AssertionError("Unexpected stderr {} != {}".format(stderr, expected_stderr))
514 515 self.stdout.close()
516 self.stderr.close()
517 518 self.running = False
519 self.process = None
520 self.rpc_connected = False
521 self._rpc = None
522 self.log.debug("Node stopped")
523 return True
524 525 def wait_until_stopped(self, *, timeout=BITCOIND_PROC_WAIT_TIMEOUT, expect_error=False, **kwargs):
526 if "expected_ret_code" not in kwargs:
527 kwargs["expected_ret_code"] = 1 if expect_error else 0 # Whether node shutdown return EXIT_FAILURE or EXIT_SUCCESS
528 self.wait_until(lambda: self.is_node_stopped(**kwargs), timeout=timeout)
529 530 def kill_process(self):
531 self.process.kill()
532 self.wait_until_stopped(expected_ret_code=1 if platform.system() == "Windows" else -9)
533 assert self.is_node_stopped()
534 535 def replace_in_config(self, replacements):
536 """
537 Perform replacements in the configuration file.
538 The substitutions are passed as a list of search-replace-tuples, e.g.
539 [("old", "new"), ("foo", "bar"), ...]
540 """
541 with open(self.bitcoinconf, 'r') as conf:
542 conf_data = conf.read()
543 for replacement in replacements:
544 assert_equal(len(replacement), 2)
545 old, new = replacement[0], replacement[1]
546 conf_data = conf_data.replace(old, new)
547 with open(self.bitcoinconf, 'w') as conf:
548 conf.write(conf_data)
549 550 @property
551 def chain_path(self) -> Path:
552 return self.datadir_path / self.chain
553 554 @property
555 def debug_log_path(self) -> Path:
556 return self.chain_path / 'debug.log'
557 558 @property
559 def blocks_path(self) -> Path:
560 return self.chain_path / "blocks"
561 562 @property
563 def blocks_key_path(self) -> Path:
564 return self.blocks_path / "xor.dat"
565 566 def read_xor_key(self) -> bytes:
567 with open(self.blocks_key_path, "rb") as xor_f:
568 return xor_f.read(NUM_XOR_BYTES)
569 570 @property
571 def wallets_path(self) -> Path:
572 return self.chain_path / "wallets"
573 574 def debug_log_size(self, **kwargs) -> int:
575 with open(self.debug_log_path, **kwargs) as dl:
576 dl.seek(0, 2)
577 return dl.tell()
578 579 @contextlib.contextmanager
580 def assert_debug_log(self, expected_msgs, unexpected_msgs=None, *, timeout=0):
581 if unexpected_msgs is None:
582 unexpected_msgs = []
583 assert_equal(type(expected_msgs), list)
584 assert_equal(type(unexpected_msgs), list)
585 remaining_expected = list(expected_msgs)
586 587 time_end = time.time() + timeout * self.timeout_factor
588 prev_size = self.debug_log_size(encoding="utf-8") # Must use same encoding that is used to read() below
589 590 def join_log(log):
591 return " - " + "\n - ".join(log.splitlines())
592 593 yield
594 595 while True:
596 with open(self.debug_log_path, encoding="utf-8", errors="replace") as dl:
597 dl.seek(prev_size)
598 log = dl.read()
599 for unexpected_msg in unexpected_msgs:
600 if unexpected_msg in log:
601 self._raise_assertion_error(f'Unexpected message "{unexpected_msg}" '
602 f'found in log:\n\n{join_log(log)}\n\n')
603 while remaining_expected and remaining_expected[-1] in log:
604 remaining_expected.pop()
605 if not remaining_expected:
606 return
607 if time.time() >= time_end:
608 break
609 time.sleep(0.05)
610 remaining_expected = [e for e in remaining_expected if e not in log]
611 self._raise_assertion_error(f'Expected message(s) {remaining_expected!s} '
612 f'not found in log:\n\n{join_log(log)}\n\n')
613 614 @contextlib.contextmanager
615 def busy_wait_for_debug_log(self, expected_msgs, timeout=60):
616 """
617 Block until we see a particular debug log message fragment or until we exceed the timeout.
618 """
619 time_end = time.time() + timeout * self.timeout_factor
620 prev_size = self.debug_log_size(mode="rb") # Must use same mode that is used to read() below
621 remaining_expected = list(expected_msgs)
622 623 yield
624 625 while True:
626 with open(self.debug_log_path, "rb") as dl:
627 dl.seek(prev_size)
628 log = dl.read()
629 630 while remaining_expected and remaining_expected[-1] in log:
631 remaining_expected.pop()
632 if not remaining_expected:
633 return
634 635 if time.time() >= time_end:
636 print_log = " - " + "\n - ".join(log.decode("utf8", errors="replace").splitlines())
637 break
638 639 # No sleep here because we want to detect the message fragment as fast as
640 # possible.
641 642 remaining_expected = [e for e in remaining_expected if e not in log]
643 self._raise_assertion_error(f'Expected message(s) {remaining_expected!s} '
644 f'not found in log:\n\n{print_log}\n\n')
645 646 @contextlib.contextmanager
647 def wait_for_new_peer(self, timeout=5):
648 """
649 Wait until the node is connected to at least one new peer. We detect this
650 by watching for an increased highest peer id, using the `getpeerinfo` RPC call.
651 Note that the simpler approach of only accounting for the number of peers
652 suffers from race conditions, as disconnects from unrelated previous peers
653 could happen anytime in-between.
654 """
655 def get_highest_peer_id():
656 peer_info = self.getpeerinfo()
657 return peer_info[-1]["id"] if peer_info else -1
658 659 initial_peer_id = get_highest_peer_id()
660 yield
661 self.wait_until(lambda: get_highest_peer_id() > initial_peer_id, timeout=timeout)
662 663 def assert_start_raises_init_error(self, extra_args=None, expected_msg=None, match=ErrorMatch.FULL_TEXT, *args, **kwargs):
664 """Attempt to start the node and expect it to raise an error.
665 666 extra_args: extra arguments to pass through to bitcoind
667 expected_msg: regex that stderr should match when bitcoind fails
668 669 Will raise if bitcoind starts without an error.
670 Will raise if an expected_msg is provided and it does not match bitcoind's stdout."""
671 assert not self.running
672 with tempfile.NamedTemporaryFile(dir=self.stderr_dir, delete=False) as log_stderr, \
673 tempfile.NamedTemporaryFile(dir=self.stdout_dir, delete=False) as log_stdout:
674 assert_msg = None
675 try:
676 self.start(extra_args, stdout=log_stdout, stderr=log_stderr, *args, **kwargs)
677 ret = self.process.wait(timeout=self.rpc_timeout)
678 self.log.debug(self._node_msg(f'bitcoind exited with status {ret} during initialization'))
679 assert_not_equal(ret, 0) # Exit code must indicate failure
680 self.running = False
681 self.process = None
682 # Check stderr for expected message
683 if expected_msg is not None:
684 log_stderr.seek(0)
685 stderr = log_stderr.read().decode('utf-8').strip()
686 if match == ErrorMatch.PARTIAL_REGEX:
687 if re.search(expected_msg, stderr, flags=re.MULTILINE) is None:
688 self._raise_assertion_error(
689 'Expected message "{}" does not partially match stderr:\n"{}"'.format(expected_msg, stderr))
690 elif match == ErrorMatch.FULL_REGEX:
691 if re.fullmatch(expected_msg, stderr) is None:
692 self._raise_assertion_error(
693 'Expected message "{}" does not fully match stderr:\n"{}"'.format(expected_msg, stderr))
694 elif match == ErrorMatch.FULL_TEXT:
695 if expected_msg != stderr:
696 self._raise_assertion_error(
697 'Expected message "{}" does not fully match stderr:\n"{}"'.format(expected_msg, stderr))
698 except subprocess.TimeoutExpired as e:
699 self.process.kill()
700 self.running = False
701 self.process = None
702 assert_msg = f'bitcoind should have exited within {self.rpc_timeout}s '
703 if expected_msg is None:
704 assert_msg += "with an error"
705 else:
706 assert_msg += "with expected error " + expected_msg
707 assert_msg += f" (cmd: {e.cmd})"
708 709 # Raise assertion outside of except-block above in order for it not to be treated as a knock-on exception.
710 if assert_msg:
711 self._raise_assertion_error(assert_msg)
712 713 def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, send_version=True, supports_v2_p2p=None, wait_for_v2_handshake=True, expect_success=True, **kwargs):
714 """Add an inbound p2p connection to the node.
715 716 This method adds the p2p connection to the self.p2ps list and also
717 returns the connection to the caller.
718 719 When self.use_v2transport is True, TestNode advertises NODE_P2P_V2 service flag
720 721 An inbound connection is made from TestNode <------ P2PConnection
722 - if TestNode doesn't advertise NODE_P2P_V2 service, P2PConnection sends version message and v1 P2P is followed
723 - if TestNode advertises NODE_P2P_V2 service, (and if P2PConnections supports v2 P2P)
724 P2PConnection sends ellswift bytes and v2 P2P is followed
725 """
726 if 'dstport' not in kwargs:
727 kwargs['dstport'] = p2p_port(self.index)
728 if 'dstaddr' not in kwargs:
729 kwargs['dstaddr'] = '127.0.0.1'
730 if supports_v2_p2p is None:
731 supports_v2_p2p = self.use_v2transport
732 733 if self.use_v2transport:
734 kwargs['services'] = kwargs.get('services', P2P_SERVICES) | NODE_P2P_V2
735 supports_v2_p2p = self.use_v2transport and supports_v2_p2p
736 p2p_conn.peer_connect(**kwargs, send_version=send_version, net=self.chain, timeout_factor=self.timeout_factor, supports_v2_p2p=supports_v2_p2p)()
737 738 self.p2ps.append(p2p_conn)
739 if not expect_success:
740 return p2p_conn
741 p2p_conn.wait_until(lambda: p2p_conn.is_connected, check_connected=False)
742 if supports_v2_p2p and wait_for_v2_handshake:
743 p2p_conn.wait_until(lambda: p2p_conn.v2_state.tried_v2_handshake)
744 if send_version:
745 p2p_conn.wait_until(lambda: not p2p_conn.on_connection_send_msg)
746 if wait_for_verack:
747 # Wait for the node to send us the version and verack
748 p2p_conn.wait_for_verack()
749 # At this point we have sent our version message and received the version and verack, however the full node
750 # has not yet received the verack from us (in reply to their version). So, the connection is not yet fully
751 # established (fSuccessfullyConnected).
752 #
753 # This shouldn't lead to any issues when sending messages, since the verack will be in-flight before the
754 # message we send. However, it might lead to races where we are expecting to receive a message. E.g. a
755 # transaction that will be added to the mempool as soon as we return here.
756 #
757 # So syncing here is redundant when we only want to send a message, but the cost is low (a few milliseconds)
758 # in comparison to the upside of making tests less fragile and unexpected intermittent errors less likely.
759 p2p_conn.sync_with_ping()
760 761 # Consistency check that the node received our user agent string.
762 # Find our connection in getpeerinfo by our address:port and theirs, as this combination is unique.
763 sockname = p2p_conn._transport.get_extra_info("socket").getsockname()
764 our_addr_and_port = f"{sockname[0]}:{sockname[1]}"
765 dst_addr_and_port = f"{p2p_conn.dstaddr}:{p2p_conn.dstport}"
766 info = [peer for peer in self.getpeerinfo() if peer["addr"] == our_addr_and_port and peer["addrbind"] == dst_addr_and_port]
767 assert_equal(len(info), 1)
768 assert_equal(info[0]["subver"], P2P_SUBVERSION)
769 770 return p2p_conn
771 772 def add_outbound_p2p_connection(self, p2p_conn, *, wait_for_verack=True, wait_for_disconnect=False, p2p_idx, connection_type="outbound-full-relay", supports_v2_p2p=None, advertise_v2_p2p=None, **kwargs):
773 """Add an outbound p2p connection from node. Must be an
774 "outbound-full-relay", "block-relay-only", "addr-fetch" or "feeler" connection.
775 776 This method adds the p2p connection to the self.p2ps list and returns
777 the connection to the caller.
778 779 p2p_idx must be different for simultaneously connected peers. When reusing it for the next peer
780 after disconnecting the previous one, it is necessary to wait for the disconnect to finish to avoid
781 a race condition.
782 783 Parameters:
784 supports_v2_p2p: whether p2p_conn supports v2 P2P or not
785 advertise_v2_p2p: whether p2p_conn is advertised to support v2 P2P or not
786 787 An outbound connection is made from TestNode -------> P2PConnection
788 - if P2PConnection doesn't advertise_v2_p2p, TestNode sends version message and v1 P2P is followed
789 - if P2PConnection both supports_v2_p2p and advertise_v2_p2p, TestNode sends ellswift bytes and v2 P2P is followed
790 - if P2PConnection doesn't supports_v2_p2p but advertise_v2_p2p,
791 TestNode sends ellswift bytes and P2PConnection disconnects,
792 TestNode reconnects by sending version message and v1 P2P is followed
793 """
794 795 def addconnection_callback(address, port):
796 self.log.debug("Connecting to %s:%d %s" % (address, port, connection_type))
797 self.addconnection('%s:%d' % (address, port), connection_type, advertise_v2_p2p)
798 799 if supports_v2_p2p is None:
800 supports_v2_p2p = self.use_v2transport
801 if advertise_v2_p2p is None:
802 advertise_v2_p2p = self.use_v2transport
803 804 if advertise_v2_p2p:
805 kwargs['services'] = kwargs.get('services', P2P_SERVICES) | NODE_P2P_V2
806 assert self.use_v2transport # only a v2 TestNode could make a v2 outbound connection
807 808 # if P2PConnection is advertised to support v2 P2P when it doesn't actually support v2 P2P,
809 # reconnection needs to be attempted using v1 P2P by sending version message
810 reconnect = advertise_v2_p2p and not supports_v2_p2p
811 # P2PConnection needs to be advertised to support v2 P2P so that ellswift bytes are sent instead of msg_version
812 supports_v2_p2p = supports_v2_p2p and advertise_v2_p2p
813 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)()
814 815 if reconnect:
816 p2p_conn.wait_for_reconnect()
817 818 if connection_type == "feeler" or wait_for_disconnect:
819 # feeler connections are closed as soon as the node receives a `version` message
820 p2p_conn.wait_until(lambda: p2p_conn.message_count["version"] == 1, check_connected=False)
821 p2p_conn.wait_until(lambda: not p2p_conn.is_connected, check_connected=False)
822 else:
823 p2p_conn.wait_for_connect()
824 self.p2ps.append(p2p_conn)
825 826 if supports_v2_p2p:
827 p2p_conn.wait_until(lambda: p2p_conn.v2_state.tried_v2_handshake)
828 p2p_conn.wait_until(lambda: not p2p_conn.on_connection_send_msg)
829 if wait_for_verack:
830 p2p_conn.wait_for_verack()
831 p2p_conn.sync_with_ping()
832 833 return p2p_conn
834 835 def num_test_p2p_connections(self):
836 """Return number of test framework p2p connections to the node."""
837 return len([peer for peer in self.getpeerinfo() if peer['subver'] == P2P_SUBVERSION])
838 839 def disconnect_p2ps(self):
840 """Close all p2p connections to the node.
841 The state of the peers (such as txrequests) may not be fully cleared
842 yet, even after this method returns."""
843 for p in self.p2ps:
844 p.peer_disconnect()
845 del self.p2ps[:]
846 847 self.wait_until(lambda: self.num_test_p2p_connections() == 0)
848 849 def is_connected_to(self, other):
850 assert isinstance(other, TestNode)
851 other_subver = other.getnetworkinfo()["subversion"]
852 return any(peer["subver"] == other_subver for peer in self.getpeerinfo())
853 854 def bumpmocktime(self, seconds):
855 """Fast forward using setmocktime to self.mocktime + seconds. Requires setmocktime to have
856 been called at some point in the past."""
857 assert self.mocktime
858 self.mocktime += seconds
859 self.setmocktime(self.mocktime)
860 861 def wait_until(self, test_function, timeout=60, check_interval=0.05):
862 return wait_until_helper_internal(test_function, timeout=timeout, timeout_factor=self.timeout_factor, check_interval=check_interval)
863 864 class TestNodeCLIAttr:
865 def __init__(self, cli, command):
866 self.cli = cli
867 self.command = command
868 869 def __call__(self, *args, **kwargs):
870 return self.cli.send_cli(self.command, *args, **kwargs)
871 872 def get_request(self, *args, **kwargs):
873 return lambda: self(*args, **kwargs)
874 875 876 def arg_to_cli(arg):
877 if isinstance(arg, bool):
878 return str(arg).lower()
879 elif arg is None:
880 return 'null'
881 elif isinstance(arg, dict) or isinstance(arg, list) or isinstance(arg, tuple):
882 return json.dumps(arg, default=serialization_fallback)
883 else:
884 return str(arg)
885 886 887 class TestNodeCLI():
888 """Interface to bitcoin-cli for an individual node"""
889 def __init__(self, binaries):
890 self.options = []
891 self.binaries = binaries
892 self.input = None
893 self.log = logging.getLogger('TestFramework.bitcoincli')
894 895 def __call__(self, *options, input=None):
896 # TestNodeCLI is callable with bitcoin-cli command-line options
897 cli = TestNodeCLI(self.binaries)
898 cli.options = self.options + [str(o) for o in options]
899 cli.input = input
900 return cli
901 902 def __getattr__(self, command):
903 return TestNodeCLIAttr(self, command)
904 905 def batch(self, requests):
906 results = []
907 for request in requests:
908 try:
909 results.append(dict(result=request()))
910 except JSONRPCException as e:
911 results.append(dict(error=e))
912 return results
913 914 def send_cli(self, clicommand=None, *args, **kwargs):
915 """Run bitcoin-cli command. Deserializes returned string as python object."""
916 pos_args = [arg_to_cli(arg) for arg in args]
917 named_args = [key + "=" + arg_to_cli(value) for (key, value) in kwargs.items() if value is not None]
918 p_args = self.binaries.rpc_argv() + self.options
919 if named_args:
920 p_args += ["-named"]
921 base_arg_pos = len(p_args)
922 if clicommand is not None:
923 p_args += [clicommand]
924 p_args += pos_args + named_args
925 926 # TEST_CLI_MAX_ARG_SIZE is set low enough that checking the string
927 # length is enough and encoding to bytes is not needed before
928 # calculating the sum.
929 sum_arg_size = sum(len(arg) for arg in p_args)
930 stdin_data = self.input
931 if sum_arg_size >= TEST_CLI_MAX_ARG_SIZE:
932 self.log.debug(f"Cli: Command size {sum_arg_size} too large, using stdin")
933 rpc_args = "\n".join([arg for arg in p_args[base_arg_pos:]])
934 if stdin_data is not None:
935 stdin_data += "\n" + rpc_args
936 else:
937 stdin_data = rpc_args
938 p_args = p_args[:base_arg_pos] + ['-stdin']
939 940 self.log.debug("Running bitcoin-cli {}".format(p_args[2:]))
941 process = subprocess.Popen(p_args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
942 cli_stdout, cli_stderr = process.communicate(input=stdin_data)
943 returncode = process.poll()
944 if returncode:
945 match = re.match(r'error code: ([-0-9]+)\nerror message:\n(.*)', cli_stderr)
946 if match:
947 code, message = match.groups()
948 raise JSONRPCException(dict(code=int(code), message=message))
949 # Ignore cli_stdout, raise with cli_stderr
950 raise subprocess.CalledProcessError(returncode, p_args, output=cli_stderr)
951 try:
952 if not cli_stdout.strip():
953 return None
954 return json.loads(cli_stdout, parse_float=decimal.Decimal)
955 except (json.JSONDecodeError, decimal.InvalidOperation):
956 return cli_stdout.rstrip("\n")
957