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