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