rpc_bind.py
1 #!/usr/bin/env python3 2 # Copyright (c) 2014-2019 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.util import assert_equal, assert_raises_rpc_error, get_rpc_proxy, rpc_port, rpc_url 10 11 class RPCBindTest(BitcoinTestFramework): 12 def set_test_params(self): 13 self.setup_clean_chain = True 14 self.bind_to_localhost_only = False 15 self.num_nodes = 1 16 self.supports_cli = False 17 18 def skip_test_if_missing_module(self): 19 # due to OS-specific network stats queries, this test works only on Linux 20 self.skip_if_platform_not_linux() 21 22 def setup_network(self): 23 self.add_nodes(self.num_nodes, None) 24 25 def add_options(self, parser): 26 parser.add_argument("--ipv4", action='store_true', dest="run_ipv4", help="Run ipv4 tests only", default=False) 27 parser.add_argument("--ipv6", action='store_true', dest="run_ipv6", help="Run ipv6 tests only", default=False) 28 parser.add_argument("--nonloopback", action='store_true', dest="run_nonloopback", help="Run non-loopback tests only", default=False) 29 30 def run_bind_test(self, allow_ips, connect_to, addresses, expected): 31 ''' 32 Start a node with requested rpcallowip and rpcbind parameters, 33 then try to connect, and check if the set of bound addresses 34 matches the expected set. 35 ''' 36 self.log.info("Bind test for %s" % str(addresses)) 37 expected = [(addr_to_hex(addr), port) for (addr, port) in expected] 38 base_args = ['-disablewallet', '-nolisten'] 39 if allow_ips: 40 base_args += ['-rpcallowip=' + x for x in allow_ips] 41 binds = ['-rpcbind='+addr for addr in addresses] 42 self.nodes[0].rpchost = connect_to 43 self.start_node(0, base_args + binds) 44 pid = self.nodes[0].process.pid 45 assert_equal(set(get_bind_addrs(pid)), set(expected)) 46 self.stop_nodes() 47 48 def run_allowip_test(self, allow_ips, rpchost, rpcport): 49 ''' 50 Start a node with rpcallow IP, and request getnetworkinfo 51 at a non-localhost IP. 52 ''' 53 self.log.info("Allow IP test for %s:%d" % (rpchost, rpcport)) 54 node_args = \ 55 ['-disablewallet', '-nolisten'] + \ 56 ['-rpcallowip='+x for x in allow_ips] + \ 57 ['-rpcbind='+addr for addr in ['127.0.0.1', "%s:%d" % (rpchost, rpcport)]] # Bind to localhost as well so start_nodes doesn't hang 58 self.nodes[0].rpchost = None 59 self.start_nodes([node_args]) 60 # connect to node through non-loopback interface 61 node = get_rpc_proxy(rpc_url(self.nodes[0].datadir_path, 0, self.chain, "%s:%d" % (rpchost, rpcport)), 0, coveragedir=self.options.coveragedir) 62 node.getnetworkinfo() 63 self.stop_nodes() 64 65 def run_test(self): 66 if sum([self.options.run_ipv4, self.options.run_ipv6, self.options.run_nonloopback]) > 1: 67 raise AssertionError("Only one of --ipv4, --ipv6 and --nonloopback can be set") 68 69 self.log.info("Check for ipv6") 70 have_ipv6 = test_ipv6_local() 71 if not have_ipv6 and not (self.options.run_ipv4 or self.options.run_nonloopback): 72 raise SkipTest("This test requires ipv6 support.") 73 74 self.log.info("Check for non-loopback interface") 75 self.non_loopback_ip = None 76 for name,ip in all_interfaces(): 77 if ip != '127.0.0.1': 78 self.non_loopback_ip = ip 79 break 80 if self.non_loopback_ip is None and self.options.run_nonloopback: 81 raise SkipTest("This test requires a non-loopback ip address.") 82 83 self.defaultport = rpc_port(0) 84 85 if not self.options.run_nonloopback: 86 self._run_loopback_tests() 87 if not self.options.run_ipv4 and not self.options.run_ipv6: 88 self._run_nonloopback_tests() 89 90 def _run_loopback_tests(self): 91 if self.options.run_ipv4: 92 # check only IPv4 localhost (explicit) 93 self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1'], 94 [('127.0.0.1', self.defaultport)]) 95 # check only IPv4 localhost (explicit) with alternative port 96 self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171'], 97 [('127.0.0.1', 32171)]) 98 # check only IPv4 localhost (explicit) with multiple alternative ports on same host 99 self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171', '127.0.0.1:32172'], 100 [('127.0.0.1', 32171), ('127.0.0.1', 32172)]) 101 else: 102 # check default without rpcallowip (IPv4 and IPv6 localhost) 103 self.run_bind_test(None, '127.0.0.1', [], 104 [('127.0.0.1', self.defaultport), ('::1', self.defaultport)]) 105 # check default with rpcallowip (IPv4 and IPv6 localhost) 106 self.run_bind_test(['127.0.0.1'], '127.0.0.1', [], 107 [('127.0.0.1', self.defaultport), ('::1', self.defaultport)]) 108 # check only IPv6 localhost (explicit) 109 self.run_bind_test(['[::1]'], '[::1]', ['[::1]'], 110 [('::1', self.defaultport)]) 111 # check both IPv4 and IPv6 localhost (explicit) 112 self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1', '[::1]'], 113 [('127.0.0.1', self.defaultport), ('::1', self.defaultport)]) 114 115 def _run_nonloopback_tests(self): 116 self.log.info("Using interface %s for testing" % self.non_loopback_ip) 117 118 # check only non-loopback interface 119 self.run_bind_test([self.non_loopback_ip], self.non_loopback_ip, [self.non_loopback_ip], 120 [(self.non_loopback_ip, self.defaultport)]) 121 122 # Check that with invalid rpcallowip, we are denied 123 self.run_allowip_test([self.non_loopback_ip], self.non_loopback_ip, self.defaultport) 124 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) 125 126 if __name__ == '__main__': 127 RPCBindTest().main()