rpc_bind.py
1 #!/usr/bin/env python3 2 # Copyright (c) 2014-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 """Test running bitcoind with the -rpcbind and -rpcallowip options.""" 6 7 from test_framework.netutil import all_interfaces, addr_to_hex, get_bind_addrs, test_ipv6_local 8 from test_framework.test_framework import BitcoinTestFramework, SkipTest 9 from test_framework.test_node import ErrorMatch 10 from test_framework.util import assert_equal, assert_raises_rpc_error, get_rpc_proxy, rpc_port, rpc_url 11 12 class RPCBindTest(BitcoinTestFramework): 13 def set_test_params(self): 14 self.setup_clean_chain = True 15 self.bind_to_localhost_only = False 16 self.num_nodes = 1 17 self.supports_cli = False 18 19 def skip_test_if_missing_module(self): 20 # due to OS-specific network stats queries, this test works only on Linux 21 self.skip_if_platform_not_linux() 22 23 def setup_network(self): 24 self.add_nodes(self.num_nodes, None) 25 26 def add_options(self, parser): 27 parser.add_argument("--ipv4", action='store_true', dest="run_ipv4", help="Run ipv4 tests only", default=False) 28 parser.add_argument("--ipv6", action='store_true', dest="run_ipv6", help="Run ipv6 tests only", default=False) 29 parser.add_argument("--nonloopback", action='store_true', dest="run_nonloopback", help="Run non-loopback tests only", default=False) 30 31 def run_bind_test(self, allow_ips, connect_to, addresses, expected): 32 ''' 33 Start a node with requested rpcallowip and rpcbind parameters, 34 then try to connect, and check if the set of bound addresses 35 matches the expected set. 36 ''' 37 self.log.info("Bind test for %s" % str(addresses)) 38 expected = [(addr_to_hex(addr), port) for (addr, port) in expected] 39 base_args = ['-disablewallet', '-nolisten'] 40 if allow_ips: 41 base_args += ['-rpcallowip=' + x for x in allow_ips] 42 binds = ['-rpcbind='+addr for addr in addresses] 43 self.nodes[0].rpchost = connect_to 44 self.start_node(0, base_args + binds) 45 pid = self.nodes[0].process.pid 46 assert_equal(set(get_bind_addrs(pid)), set(expected)) 47 self.stop_nodes() 48 49 def run_invalid_bind_test(self, allow_ips, addresses): 50 ''' 51 Attempt to start a node with requested rpcallowip and rpcbind 52 parameters, expecting that the node will fail. 53 ''' 54 self.log.info(f'Invalid bind test for {addresses}') 55 base_args = ['-disablewallet', '-nolisten'] 56 if allow_ips: 57 base_args += ['-rpcallowip=' + x for x in allow_ips] 58 init_error = 'Error: Invalid port specified in -rpcbind: ' 59 for addr in addresses: 60 self.nodes[0].assert_start_raises_init_error(base_args + [f'-rpcbind={addr}'], init_error + f"'{addr}'") 61 62 def run_allowip_test(self, allow_ips, rpchost, rpcport): 63 ''' 64 Start a node with rpcallow IP, and request getnetworkinfo 65 at a non-localhost IP. 66 ''' 67 self.log.info("Allow IP test for %s:%d" % (rpchost, rpcport)) 68 node_args = \ 69 ['-disablewallet', '-nolisten'] + \ 70 ['-rpcallowip='+x for x in allow_ips] + \ 71 ['-rpcbind='+addr for addr in ['127.0.0.1', "%s:%d" % (rpchost, rpcport)]] # Bind to localhost as well so start_nodes doesn't hang 72 self.nodes[0].rpchost = None 73 self.start_nodes([node_args]) 74 # connect to node through non-loopback interface 75 node = get_rpc_proxy(rpc_url(self.nodes[0].datadir_path, 0, self.chain, "%s:%d" % (rpchost, rpcport)), 0, coveragedir=self.options.coveragedir) 76 node.getnetworkinfo() 77 self.stop_nodes() 78 79 def run_invalid_allowip_test(self): 80 ''' 81 Check parameter interaction with -rpcallowip and -cjdnsreachable. 82 RFC4193 addresses are fc00::/7 like CJDNS but have an optional 83 "local" L bit making them fd00:: which should always be OK. 84 ''' 85 self.log.info("Allow RFC4193 when compatible with CJDNS options") 86 # Don't rpcallow RFC4193 with L-bit=0 if CJDNS is enabled 87 self.nodes[0].assert_start_raises_init_error( 88 ["-rpcallowip=fc00:db8:c0:ff:ee::/80","-cjdnsreachable"], 89 "Invalid -rpcallowip subnet specification", 90 match=ErrorMatch.PARTIAL_REGEX) 91 # OK to rpcallow RFC4193 with L-bit=1 if CJDNS is enabled 92 self.start_node(0, ["-rpcallowip=fd00:db8:c0:ff:ee::/80","-cjdnsreachable"]) 93 self.stop_nodes() 94 # OK to rpcallow RFC4193 with L-bit=0 if CJDNS is not enabled 95 self.start_node(0, ["-rpcallowip=fc00:db8:c0:ff:ee::/80"]) 96 self.stop_nodes() 97 98 def run_test(self): 99 if sum([self.options.run_ipv4, self.options.run_ipv6, self.options.run_nonloopback]) > 1: 100 raise AssertionError("Only one of --ipv4, --ipv6 and --nonloopback can be set") 101 102 self.log.info("Check for ipv6") 103 have_ipv6 = test_ipv6_local() 104 if not have_ipv6 and not (self.options.run_ipv4 or self.options.run_nonloopback): 105 raise SkipTest("This test requires ipv6 support.") 106 107 self.log.info("Check for non-loopback interface") 108 self.non_loopback_ip = None 109 for name,ip in all_interfaces(): 110 if ip != '127.0.0.1': 111 self.non_loopback_ip = ip 112 break 113 if self.non_loopback_ip is None and self.options.run_nonloopback: 114 raise SkipTest("This test requires a non-loopback ip address.") 115 116 self.defaultport = rpc_port(0) 117 118 if not self.options.run_nonloopback: 119 self._run_loopback_tests() 120 if self.options.run_ipv4: 121 self.run_invalid_bind_test(['127.0.0.1'], ['127.0.0.1:notaport', '127.0.0.1:-18443', '127.0.0.1:0', '127.0.0.1:65536']) 122 if self.options.run_ipv6: 123 self.run_invalid_bind_test(['[::1]'], ['[::1]:notaport', '[::1]:-18443', '[::1]:0', '[::1]:65536']) 124 self.run_invalid_allowip_test() 125 if not self.options.run_ipv4 and not self.options.run_ipv6: 126 if self.non_loopback_ip: 127 self._run_nonloopback_tests() 128 else: 129 self.log.info('Non-loopback IP address not found, skipping non-loopback tests') 130 131 def _run_loopback_tests(self): 132 if self.options.run_ipv4: 133 # check only IPv4 localhost (explicit) 134 self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1'], 135 [('127.0.0.1', self.defaultport)]) 136 # check only IPv4 localhost (explicit) with alternative port 137 self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171'], 138 [('127.0.0.1', 32171)]) 139 # check only IPv4 localhost (explicit) with multiple alternative ports on same host 140 self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171', '127.0.0.1:32172'], 141 [('127.0.0.1', 32171), ('127.0.0.1', 32172)]) 142 else: 143 # check default without rpcallowip (IPv4 and IPv6 localhost) 144 self.run_bind_test(None, '127.0.0.1', [], 145 [('127.0.0.1', self.defaultport), ('::1', self.defaultport)]) 146 # check default with rpcallowip (IPv4 and IPv6 localhost) 147 self.run_bind_test(['127.0.0.1'], '127.0.0.1', [], 148 [('127.0.0.1', self.defaultport), ('::1', self.defaultport)]) 149 # check only IPv6 localhost (explicit) 150 self.run_bind_test(['[::1]'], '[::1]', ['[::1]'], 151 [('::1', self.defaultport)]) 152 # check both IPv4 and IPv6 localhost (explicit) 153 self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1', '[::1]'], 154 [('127.0.0.1', self.defaultport), ('::1', self.defaultport)]) 155 156 def _run_nonloopback_tests(self): 157 self.log.info("Using interface %s for testing" % self.non_loopback_ip) 158 159 # check only non-loopback interface 160 self.run_bind_test([self.non_loopback_ip], self.non_loopback_ip, [self.non_loopback_ip], 161 [(self.non_loopback_ip, self.defaultport)]) 162 163 # Check that with invalid rpcallowip, we are denied 164 self.run_allowip_test([self.non_loopback_ip], self.non_loopback_ip, self.defaultport) 165 assert_raises_rpc_error(-342, "non-JSON HTTP response with '403 Forbidden' from server", self.run_allowip_test, ['1.1.1.1'], self.non_loopback_ip, self.defaultport) 166 167 if __name__ == '__main__': 168 RPCBindTest(__file__).main()