Inside Bitcoin

751        When self.use_v2transport is True, TestNode advertises NODE_P2P_V2 service flag
752
753        An inbound connection is made from TestNode <------ P2PConnection
754        - if TestNode doesn't advertise NODE_P2P_V2 service, P2PConnection sends version message and v1 P2P is followed
755        - if TestNode advertises NODE_P2P_V2 service, (and if P2PConnections supports v2 P2P)
756                P2PConnection sends ellswift bytes and v2 P2P is followed
757        """
758        if 'dstport' not in kwargs:
759            kwargs['dstport'] = p2p_port(self.index)
760        if 'dstaddr' not in kwargs:
761            kwargs['dstaddr'] = '127.0.0.1'
762        if supports_v2_p2p is None:
763            supports_v2_p2p = self.use_v2transport
764
765        if self.use_v2transport:
766            kwargs['services'] = kwargs.get('services', P2P_SERVICES) | NODE_P2P_V2
767        supports_v2_p2p = self.use_v2transport and supports_v2_p2p
768        p2p_conn.peer_connect(**kwargs, send_version=send_version, net=self.chain, timeout_factor=self.timeout_factor, supports_v2_p2p=supports_v2_p2p)()
769
770        self.p2ps.append(p2p_conn)
771        if not expect_success:
772            return p2p_conn
773        p2p_conn.wait_until(lambda: p2p_conn.is_connected, check_connected=False)
774        if supports_v2_p2p and wait_for_v2_handshake:
775            p2p_conn.wait_until(lambda: p2p_conn.v2_state.tried_v2_handshake)
776        if send_version:
777            p2p_conn.wait_until(lambda: not p2p_conn.on_connection_send_msg)
778        if wait_for_verack:
779            # Wait for the node to send us the version and verack
780            p2p_conn.wait_for_verack()
781            # At this point we have sent our version message and received the version and verack, however the full node
782            # has not yet received the verack from us (in reply to their version). So, the connection is not yet fully
783            # established (fSuccessfullyConnected).
784            #
785            # This shouldn't lead to any issues when sending messages, since the verack will be in-flight before the
786            # message we send. However, it might lead to races where we are expecting to receive a message. E.g. a
787            # transaction that will be added to the mempool as soon as we return here.
788            #
789            # So syncing here is redundant when we only want to send a message, but the cost is low (a few milliseconds)
790            # in comparison to the upside of making tests less fragile and unexpected intermittent errors less likely.
791            p2p_conn.sync_with_ping()
792
793            # Consistency check that the node received our user agent string.
794            # Find our connection in getpeerinfo by our address:port and theirs, as this combination is unique.
795            sockname = p2p_conn._transport.get_extra_info("socket").getsockname()
796            our_addr_and_port = f"{sockname[0]}:{sockname[1]}"
797            dst_addr_and_port = f"{p2p_conn.dstaddr}:{p2p_conn.dstport}"
798            info = [peer for peer in self.getpeerinfo() if peer["addr"] == our_addr_and_port and peer["addrbind"] == dst_addr_and_port]
799            assert_equal(len(info), 1)
800            assert_equal(info[0]["subver"], P2P_SUBVERSION)
801
802        return p2p_conn
803
804    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):
805        """Add an outbound p2p connection from node. Must be an
806        "outbound-full-relay", "block-relay-only", "addr-fetch" or "feeler" connection.
807
808        This method adds the p2p connection to the self.p2ps list and returns
809        the connection to the caller.
810
811        p2p_idx must be different for simultaneously connected peers. When reusing it for the next peer
812        after disconnecting the previous one, it is necessary to wait for the disconnect to finish to avoid
813        a race condition.
814
815        Parameters:
816            supports_v2_p2p: whether p2p_conn supports v2 P2P or not
817            advertise_v2_p2p: whether p2p_conn is advertised to support v2 P2P or not
818
819        An outbound connection is made from TestNode -------> P2PConnection
820            - if P2PConnection doesn't advertise_v2_p2p, TestNode sends version message and v1 P2P is followed
821            - if P2PConnection both supports_v2_p2p and advertise_v2_p2p, TestNode sends ellswift bytes and v2 P2P is followed
822            - if P2PConnection doesn't supports_v2_p2p but advertise_v2_p2p,
823                TestNode sends ellswift bytes and P2PConnection disconnects,
824                TestNode reconnects by sending version message and v1 P2P is followed
825        """
826
827        def addconnection_callback(address, port):
828            self.log.debug("Connecting to %s:%d %s" % (address, port, connection_type))
829            self.addconnection('%s:%d' % (address, port), connection_type, advertise_v2_p2p)
830
831        if supports_v2_p2p is None:
832            supports_v2_p2p = self.use_v2transport
833        if advertise_v2_p2p is None:
834            advertise_v2_p2p = self.use_v2transport
835
836        if advertise_v2_p2p:
837            kwargs['services'] = kwargs.get('services', P2P_SERVICES) | NODE_P2P_V2
838            assert self.use_v2transport  # only a v2 TestNode could make a v2 outbound connection
839
840        # if P2PConnection is advertised to support v2 P2P when it doesn't actually support v2 P2P,
841        # reconnection needs to be attempted using v1 P2P by sending version message
842        reconnect = advertise_v2_p2p and not supports_v2_p2p
843        # P2PConnection needs to be advertised to support v2 P2P so that ellswift bytes are sent instead of msg_version
844        supports_v2_p2p = supports_v2_p2p and advertise_v2_p2p
845        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)()
846
847        if reconnect:
848            p2p_conn.wait_for_reconnect()
849
850        if connection_type == "feeler" or wait_for_disconnect:
851            # feeler connections are closed as soon as the node receives a `version` message
852            p2p_conn.wait_until(lambda: p2p_conn.message_count["version"] == 1, check_connected=False)
853            p2p_conn.wait_until(lambda: not p2p_conn.is_connected, check_connected=False)
854        else:
855            p2p_conn.wait_for_connect()
856            self.p2ps.append(p2p_conn)
857
858            if supports_v2_p2p:
859                p2p_conn.wait_until(lambda: p2p_conn.v2_state.tried_v2_handshake)
860            p2p_conn.wait_until(lambda: not p2p_conn.on_connection_send_msg)
861            if wait_for_verack:
862                p2p_conn.wait_for_verack()
863                p2p_conn.sync_with_ping()
864
865        return p2p_conn
866
867    def num_test_p2p_connections(self):
868        """Return number of test framework p2p connections to the node."""
869        return len([peer for peer in self.getpeerinfo() if peer['subver'] == P2P_SUBVERSION])
870
871    def disconnect_p2ps(self):
872        """Close all p2p connections to the node.
873        The state of the peers (such as txrequests) may not be fully cleared
874        yet, even after this method returns."""
875        for p in self.p2ps:
876            p.peer_disconnect()
877        del self.p2ps[:]
878
879        self.wait_until(lambda: self.num_test_p2p_connections() == 0)
880
881    def is_connected_to(self, other):
882        assert isinstance(other, TestNode)
883        other_subver = other.getnetworkinfo()["subversion"]
884        return any(peer["subver"] == other_subver for peer in self.getpeerinfo())
885
886    def bumpmocktime(self, seconds):
887        """Fast forward using setmocktime to self.mocktime + seconds. Requires setmocktime to have
888        been called at some point in the past."""
889        assert self.mocktime
890        self.mocktime += seconds
891        self.setmocktime(self.mocktime)
892
893    def wait_until(self, test_function, timeout=60, check_interval=0.05):
894        return wait_until_helper_internal(test_function, timeout=timeout, timeout_factor=self.timeout_factor, check_interval=check_interval)
895
896class TestNodeCLIAttr:
897    def __init__(self, cli, command):
898        self.cli = cli
899        self.command = command
900
901    def __call__(self, *args, **kwargs):
902        return self.cli.send_cli(self.command, *args, **kwargs)
903
904    def get_request(self, *args, **kwargs):
905        return lambda: self(*args, **kwargs)
906
907
908def arg_to_cli(arg):
909    if isinstance(arg, bool):
910        return str(arg).lower()
911    elif arg is None:
912        return 'null'
913    elif isinstance(arg, dict) or isinstance(arg, list) or isinstance(arg, tuple):
914        return json.dumps(arg, default=serialization_fallback)
915    else:
916        return str(arg)
917
918
919class TestNodeCLI():
920    """Interface to bitcoin-cli for an individual node"""
921    def __init__(self, binaries):
922        self.options = []
923        self.binaries = binaries
924        self.input = None
925        self.log = logging.getLogger('TestFramework.bitcoincli')
926
927    def __call__(self, *options, input=None):
928        # TestNodeCLI is callable with bitcoin-cli command-line options
929        cli = TestNodeCLI(self.binaries)
930        cli.options = self.options + [str(o) for o in options]
931        cli.input = input
932        return cli
933
934    def __getattr__(self, command):
935        return TestNodeCLIAttr(self, command)
936
937    def batch(self, requests):
938        results = []
939        for request in requests:
940            try:
941                results.append(dict(result=request()))
942            except JSONRPCException as e:
943                results.append(dict(error=e))
944        return results
945
946    def send_cli(self, clicommand=None, *args, **kwargs):
947        """Run bitcoin-cli command. Deserializes returned string as python object."""
948        pos_args = [arg_to_cli(arg) for arg in args]
949        named_args = [key + "=" + arg_to_cli(value) for (key, value) in kwargs.items() if value is not None]
950        p_args = self.binaries.rpc_argv() + self.options
951        if named_args:
952            p_args += ["-named"]
953        base_arg_pos = len(p_args)
954        if clicommand is not None:
955            p_args += [clicommand]
956        p_args += pos_args + named_args
957
958        # TEST_CLI_MAX_ARG_SIZE is set low enough that checking the string
959        # length is enough and encoding to bytes is not needed before
960        # calculating the sum.
961        sum_arg_size = sum(len(arg) for arg in p_args)
962        stdin_data = self.input
963        if sum_arg_size >= TEST_CLI_MAX_ARG_SIZE:
964            self.log.debug(f"Cli: Command size {sum_arg_size} too large, using stdin")
965            rpc_args = "\n".join([arg for arg in p_args[base_arg_pos:]])
966            if stdin_data is not None:
967                stdin_data += "\n" + rpc_args
968            else:
969                stdin_data = rpc_args
970            p_args = p_args[:base_arg_pos] + ['-stdin']
971
972        self.log.debug("Running bitcoin-cli {}".format(p_args[2:]))
973        process = subprocess.Popen(p_args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
974        cli_stdout, cli_stderr = process.communicate(input=stdin_data)
975        returncode = process.poll()
976        if returncode:
977            match = re.match(r'error code: ([-0-9]+)\nerror message:\n(.*)', cli_stderr)
978            if match:
979                code, message = match.groups()
980                raise JSONRPCException(dict(code=int(code), message=message))
981            match = re.match(r'error: Server response: (.*)\n?$', cli_stderr)
982            if match:
983                message = match.group(1)
984                raise JSONRPCException(dict(
985                    code=-342,
986                    message=f"non-JSON HTTP response with '503 Service Unavailable' from server: {message}",
987                ), http_status=503)
988            # Ignore cli_stdout, raise with cli_stderr
989            raise subprocess.CalledProcessError(returncode, p_args, output=cli_stderr)
990        try:
991            if not cli_stdout.strip():
992                return None
993            return json.loads(cli_stdout, parse_float=decimal.Decimal)
994        except (json.JSONDecodeError, decimal.InvalidOperation):
995            return cli_stdout.rstrip("\n")
996