test/functional/rpcbind_test.py

   1 #!/usr/bin/env python3
   2 # Copyright (c) 2014-2016 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 import socket
   8 import sys
   9
  10 from test_framework.test_framework import BitcoinTestFramework, SkipTest
  11 from test_framework.util import *
  12 from test_framework.netutil import *
  13
  14 class RPCBindTest(BitcoinTestFramework):
  15     def set_test_params(self):
  16         self.setup_clean_chain = True
  17         self.num_nodes = 1
  18
  19     def setup_network(self):
  20         self.add_nodes(self.num_nodes, None)
  21
  22     def run_bind_test(self, allow_ips, connect_to, addresses, expected):
  23         '''
  24         Start a node with requested rpcallowip and rpcbind parameters,
  25         then try to connect, and check if the set of bound addresses
  26         matches the expected set.
  27         '''
  28         self.log.info("Bind test for %s" % str(addresses))
  29         expected = [(addr_to_hex(addr), port) for (addr, port) in expected]
  30         base_args = ['-disablewallet', '-nolisten']
  31         if allow_ips:
  32             base_args += ['-rpcallowip=' + x for x in allow_ips]
  33         binds = ['-rpcbind='+addr for addr in addresses]
  34         self.nodes[0].rpchost = connect_to
  35         self.start_node(0, base_args + binds)
  36         pid = self.nodes[0].process.pid
  37         assert_equal(set(get_bind_addrs(pid)), set(expected))
  38         self.stop_nodes()
  39
  40     def run_allowip_test(self, allow_ips, rpchost, rpcport):
  41         '''
  42         Start a node with rpcallow IP, and request getnetworkinfo
  43         at a non-localhost IP.
  44         '''
  45         self.log.info("Allow IP test for %s:%d" % (rpchost, rpcport))
  46         base_args = ['-disablewallet', '-nolisten'] + ['-rpcallowip='+x for x in allow_ips]
  47         self.nodes[0].rpchost = None
  48         self.start_nodes([base_args])
  49         # connect to node through non-loopback interface
  50         node = get_rpc_proxy(rpc_url(get_datadir_path(self.options.tmpdir, 0), 0, "%s:%d" % (rpchost, rpcport)), 0, coveragedir=self.options.coveragedir)
  51         node.getnetworkinfo()
  52         self.stop_nodes()
  53
  54     def run_test(self):
  55         # due to OS-specific network stats queries, this test works only on Linux
  56         if not sys.platform.startswith('linux'):
  57             raise SkipTest("This test can only be run on linux.")
  58         # find the first non-loopback interface for testing
  59         non_loopback_ip = None
  60         for name,ip in all_interfaces():
  61             if ip != '127.0.0.1':
  62                 non_loopback_ip = ip
  63                 break
  64         if non_loopback_ip is None:
  65             raise SkipTest("This test requires at least one non-loopback IPv4 interface.")
  66         try:
  67             s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
  68             s.connect(("::1",1))
  69             s.close
  70         except OSError:
  71             raise SkipTest("This test requires IPv6 support.")
  72
  73         self.log.info("Using interface %s for testing" % non_loopback_ip)
  74
  75         defaultport = rpc_port(0)
  76
  77         # check default without rpcallowip (IPv4 and IPv6 localhost)
  78         self.run_bind_test(None, '127.0.0.1', [],
  79             [('127.0.0.1', defaultport), ('::1', defaultport)])
  80         # check default with rpcallowip (IPv6 any)
  81         self.run_bind_test(['127.0.0.1'], '127.0.0.1', [],
  82             [('::0', defaultport)])
  83         # check only IPv4 localhost (explicit)
  84         self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1'],
  85             [('127.0.0.1', defaultport)])
  86         # check only IPv4 localhost (explicit) with alternative port
  87         self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171'],
  88             [('127.0.0.1', 32171)])
  89         # check only IPv4 localhost (explicit) with multiple alternative ports on same host
  90         self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171', '127.0.0.1:32172'],
  91             [('127.0.0.1', 32171), ('127.0.0.1', 32172)])
  92         # check only IPv6 localhost (explicit)
  93         self.run_bind_test(['[::1]'], '[::1]', ['[::1]'],
  94             [('::1', defaultport)])
  95         # check both IPv4 and IPv6 localhost (explicit)
  96         self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1', '[::1]'],
  97             [('127.0.0.1', defaultport), ('::1', defaultport)])
  98         # check only non-loopback interface
  99         self.run_bind_test([non_loopback_ip], non_loopback_ip, [non_loopback_ip],
 100             [(non_loopback_ip, defaultport)])
 101
 102         # Check that with invalid rpcallowip, we are denied
 103         self.run_allowip_test([non_loopback_ip], non_loopback_ip, defaultport)
 104         assert_raises_rpc_error(-342, "non-JSON HTTP response with '403 Forbidden' from server", self.run_allowip_test, ['1.1.1.1'], non_loopback_ip, defaultport)
 105
 106 if __name__ == '__main__':
 107     RPCBindTest().main()