Page MenuHomePhabricator
Differential D8133 Diff 25139 test/functional/rpc_bind.py

Changeset View

Standalone View

View Options

test/functional/rpc_bind.py

#!/usr/bin/env python3 #!/usr/bin/env python3
# Copyright (c) 2014-2019 The Bitcoin Core developers # Copyright (c) 2014-2019 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying # Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php. # file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test running bitcoind with the -rpcbind and -rpcallowip options.""" """Test running bitcoind with the -rpcbind and -rpcallowip options."""
from platform import uname from platform import uname
import socket
import sys import sys
from test_framework.netutil import addr_to_hex, all_interfaces, get_bind_addrs from test_framework.netutil import addr_to_hex, all_interfaces, get_bind_addrs, test_ipv6_local
from test_framework.test_framework import BitcoinTestFramework, SkipTest from test_framework.test_framework import BitcoinTestFramework, SkipTest
from test_framework.util import ( from test_framework.util import (
assert_equal, assert_equal,
assert_raises_rpc_error, assert_raises_rpc_error,
get_rpc_proxy, get_rpc_proxy,
rpc_port, rpc_port,
rpc_url, rpc_url,
) )
class RPCBindTest(BitcoinTestFramework): class RPCBindTest(BitcoinTestFramework):
def set_test_params(self): def set_test_params(self):
self.setup_clean_chain = True self.setup_clean_chain = True
self.bind_to_localhost_only = False self.bind_to_localhost_only = False
self.num_nodes = 1 self.num_nodes = 1
def setup_network(self): def setup_network(self):
self.add_nodes(self.num_nodes, None) self.add_nodes(self.num_nodes, None)
def add_options(self, parser):
parser.add_argument(
"--ipv4",
action='store_true',
dest="run_ipv4",
help="Run ipv4 tests only",
default=False)
parser.add_argument(
"--ipv6",
action='store_true',
dest="run_ipv6",
help="Run ipv6 tests only",
default=False)
parser.add_argument(
"--nonloopback",
action='store_true',
dest="run_nonloopback",
help="Run non-loopback tests only",
default=False)
def run_bind_test(self, allow_ips, connect_to, addresses, expected): def run_bind_test(self, allow_ips, connect_to, addresses, expected):
''' '''
Start a node with requested rpcallowip and rpcbind parameters, Start a node with requested rpcallowip and rpcbind parameters,
then try to connect, and check if the set of bound addresses then try to connect, and check if the set of bound addresses
matches the expected set. matches the expected set.
''' '''
self.log.info("Bind test for {}".format(str(addresses))) self.log.info("Bind test for {}".format(str(addresses)))
expected = [(addr_to_hex(addr), port) for (addr, port) in expected] expected = [(addr_to_hex(addr), port) for (addr, port) in expected]
Show All 31 Lines def run_allowip_test(self, allow_ips, rpchost, rpcport):
url = rpc_url(self.nodes[0].datadir, self.chain, rpchost, rpcport) url = rpc_url(self.nodes[0].datadir, self.chain, rpchost, rpcport)
node = get_rpc_proxy(url, 0, coveragedir=self.options.coveragedir) node = get_rpc_proxy(url, 0, coveragedir=self.options.coveragedir)
node.getnetworkinfo() node.getnetworkinfo()
self.stop_nodes() self.stop_nodes()
def run_test(self): def run_test(self):
# due to OS-specific network stats queries, this test works only on # due to OS-specific network stats queries, this test works only on
# Linux # Linux
if sum([self.options.run_ipv4, self.options.run_ipv6,
self.options.run_nonloopback]) > 1:
raise AssertionError(
"Only one of --ipv4, --ipv6 and --nonloopback can be set")
self.log.info("Check for linux")
if not sys.platform.startswith('linux'): if not sys.platform.startswith('linux'):
raise SkipTest("This test can only be run on Linux.") raise SkipTest("This test can only be run on linux.")
# WSL in currently not supported (refer to # WSL in currently not supported (refer to
# https://reviews.bitcoinabc.org/T400 for details). # https://reviews.bitcoinabc.org/T400 for details).
# This condition should be removed once netstat support is provided by # This condition should be removed once netstat support is provided by
# Microsoft. # Microsoft.
if "microsoft" in uname().version.lower(): if "microsoft" in uname().version.lower():
raise SkipTest( raise SkipTest(
"Running this test on WSL is currently not supported") "Running this test on WSL is currently not supported")
# find the first non-loopback interface for testing self.log.info("Check for ipv6")
non_loopback_ip = None have_ipv6 = test_ipv6_local()
if not have_ipv6 and not self.options.run_ipv4:
raise SkipTest("This test requires ipv6 support.")
self.log.info("Check for non-loopback interface")
self.non_loopback_ip = None
for name, ip in all_interfaces(): for name, ip in all_interfaces():
if ip != '127.0.0.1': if ip != '127.0.0.1':
non_loopback_ip = ip self.non_loopback_ip = ip
break break
if non_loopback_ip is None: if self.non_loopback_ip is None and self.options.run_nonloopback:
raise SkipTest( raise SkipTest("This test requires a non-loopback ip address.")
"This test requires at least one non-loopback IPv4 interface.")
try:
s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
s.connect(("::1", 1))
s.close
except OSError:
raise SkipTest("This test requires IPv6 support.")
self.log.info("Using interface {} for testing".format(non_loopback_ip)) self.defaultport = rpc_port(0)
defaultport = rpc_port(0) if not self.options.run_nonloopback:
self._run_loopback_tests()
if not self.options.run_ipv4 and not self.options.run_ipv6:
self._run_nonloopback_tests()
# check default without rpcallowip (IPv4 and IPv6 localhost) def _run_loopback_tests(self):
self.run_bind_test(None, '127.0.0.1', [], if self.options.run_ipv4:
[('127.0.0.1', defaultport), ('::1', defaultport)])
# check default with rpcallowip (IPv4 and IPv6 localhost)
self.run_bind_test(['127.0.0.1'], '127.0.0.1', [],
[('127.0.0.1', defaultport), ('::1', defaultport)])
# check only IPv4 localhost (explicit) # check only IPv4 localhost (explicit)
self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1'], self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1'],
[('127.0.0.1', defaultport)]) [('127.0.0.1', self.defaultport)])
# check only IPv4 localhost (explicit) with alternative port # check only IPv4 localhost (explicit) with alternative port
self.run_bind_test( self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171'],
['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171'],
[('127.0.0.1', 32171)]) [('127.0.0.1', 32171)])
# check only IPv4 localhost (explicit) with multiple alternative ports # check only IPv4 localhost (explicit) with multiple alternative
# on same host # ports on same host
self.run_bind_test( self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171', '127.0.0.1:32172'],
['127.0.0.1'], '127.0.0.1:32171', [
'127.0.0.1:32171', '127.0.0.1:32172'],
[('127.0.0.1', 32171), ('127.0.0.1', 32172)]) [('127.0.0.1', 32171), ('127.0.0.1', 32172)])
else:
# check default without rpcallowip (IPv4 and IPv6 localhost)
self.run_bind_test(None, '127.0.0.1', [],
[('127.0.0.1', self.defaultport), ('::1', self.defaultport)])
# check default with rpcallowip (IPv4 and IPv6 localhost)
self.run_bind_test(
['127.0.0.1'], '127.0.0.1', [], [
('127.0.0.1', self.defaultport), ('::1', self.defaultport)])
# check only IPv6 localhost (explicit) # check only IPv6 localhost (explicit)
self.run_bind_test(['[::1]'], '[::1]', ['[::1]'], self.run_bind_test(['[::1]'], '[::1]', ['[::1]'],
[('::1', defaultport)]) [('::1', self.defaultport)])
# check both IPv4 and IPv6 localhost (explicit) # check both IPv4 and IPv6 localhost (explicit)
self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1', '[::1]'], self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1', '[::1]'],
[('127.0.0.1', defaultport), ('::1', defaultport)]) [('127.0.0.1', self.defaultport), ('::1', self.defaultport)])
def _run_nonloopback_tests(self):
self.log.info(
"Using interface {} for testing".format(
self.non_loopback_ip))
# check only non-loopback interface # check only non-loopback interface
self.run_bind_test( self.run_bind_test([self.non_loopback_ip], self.non_loopback_ip, [self.non_loopback_ip],
[non_loopback_ip], non_loopback_ip, [non_loopback_ip], [(self.non_loopback_ip, self.defaultport)])
[(non_loopback_ip, defaultport)])
# Check that with invalid rpcallowip, we are denied # Check that with invalid rpcallowip, we are denied
self.run_allowip_test([non_loopback_ip], non_loopback_ip, defaultport) self.run_allowip_test([self.non_loopback_ip],
assert_raises_rpc_error(-342, "non-JSON HTTP response with '403 Forbidden' from server", self.non_loopback_ip, self.defaultport)
self.run_allowip_test, ['1.1.1.1'], non_loopback_ip, defaultport) 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)
if __name__ == '__main__': if __name__ == '__main__':
RPCBindTest().main() RPCBindTest().main()