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          self.skip_if_platform_not_posix()
 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_invalid_bind_test(self, allow_ips, addresses):
 49          '''
 50          Attempt to start a node with requested rpcallowip and rpcbind
 51          parameters, expecting that the node will fail.
 52          '''
 53          self.log.info(f'Invalid bind test for {addresses}')
 54          base_args = ['-disablewallet', '-nolisten']
 55          if allow_ips:
 56              base_args += ['-rpcallowip=' + x for x in allow_ips]
 57          init_error = 'Error: Invalid port specified in -rpcbind: '
 58          for addr in addresses:
 59              self.nodes[0].assert_start_raises_init_error(base_args + [f'-rpcbind={addr}'], init_error + f"'{addr}'")
 60  
 61      def run_allowip_test(self, allow_ips, rpchost, rpcport):
 62          '''
 63          Start a node with rpcallow IP, and request getnetworkinfo
 64          at a non-localhost IP.
 65          '''
 66          self.log.info("Allow IP test for %s:%d" % (rpchost, rpcport))
 67          node_args = \
 68              ['-disablewallet', '-nolisten'] + \
 69              ['-rpcallowip='+x for x in allow_ips] + \
 70              ['-rpcbind='+addr for addr in ['127.0.0.1', "%s:%d" % (rpchost, rpcport)]] # Bind to localhost as well so start_nodes doesn't hang
 71          self.nodes[0].rpchost = None
 72          self.start_nodes([node_args])
 73          # connect to node through non-loopback interface
 74          node = get_rpc_proxy(rpc_url(self.nodes[0].datadir_path, 0, self.chain, "%s:%d" % (rpchost, rpcport)), 0, coveragedir=self.options.coveragedir)
 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 ip != '127.0.0.1':
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          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)
168  
169  if __name__ == '__main__':
170      RPCBindTest(__file__).main()