1 #!/usr/bin/env python3
2 # Copyright (c) 2014-2019 The Limenka 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 limenkad 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 LimenkaTestFramework, 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(LimenkaTestFramework):
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()
169