You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

rpcbind_test.py 4.8KB

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