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