Changeset View
Changeset View
Standalone View
Standalone View
qa/rpc-tests/rpcbind_test.py
Show All 27 Lines | 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. | ||||
''' | ''' | ||||
expected = [(addr_to_hex(addr), port) for (addr, port) in expected] | expected = [(addr_to_hex(addr), port) for (addr, port) in expected] | ||||
base_args = ['-disablewallet', '-nolisten'] | base_args = ['-disablewallet', '-nolisten'] | ||||
if allow_ips: | if allow_ips: | ||||
base_args += ['-rpcallowip=' + x for x in allow_ips] | base_args += ['-rpcallowip=' + x for x in allow_ips] | ||||
binds = ['-rpcbind='+addr for addr in addresses] | binds = ['-rpcbind=' + addr for addr in addresses] | ||||
self.nodes = start_nodes(self.num_nodes, self.options.tmpdir, [base_args + binds], connect_to) | self.nodes = start_nodes( | ||||
self.num_nodes, self.options.tmpdir, [base_args + binds], connect_to) | |||||
pid = bitcoind_processes[0].pid | pid = bitcoind_processes[0].pid | ||||
assert_equal(set(get_bind_addrs(pid)), set(expected)) | assert_equal(set(get_bind_addrs(pid)), set(expected)) | ||||
stop_nodes(self.nodes) | stop_nodes(self.nodes) | ||||
def run_allowip_test(self, allow_ips, rpchost, rpcport): | def run_allowip_test(self, allow_ips, rpchost, rpcport): | ||||
''' | ''' | ||||
Start a node with rpcallow IP, and request getnetworkinfo | Start a node with rpcallow IP, and request getnetworkinfo | ||||
at a non-localhost IP. | at a non-localhost IP. | ||||
''' | ''' | ||||
base_args = ['-disablewallet', '-nolisten'] + ['-rpcallowip='+x for x in allow_ips] | base_args = ['-disablewallet', '-nolisten'] + [ | ||||
self.nodes = start_nodes(self.num_nodes, self.options.tmpdir, [base_args]) | '-rpcallowip=' + x for x in allow_ips] | ||||
self.nodes = start_nodes( | |||||
self.num_nodes, self.options.tmpdir, [base_args]) | |||||
# connect to node through non-loopback interface | # connect to node through non-loopback interface | ||||
node = get_rpc_proxy(rpc_url(0, "%s:%d" % (rpchost, rpcport)), 0) | node = get_rpc_proxy(rpc_url(0, "%s:%d" % (rpchost, rpcport)), 0) | ||||
node.getnetworkinfo() | node.getnetworkinfo() | ||||
stop_nodes(self.nodes) | stop_nodes(self.nodes) | ||||
def run_test(self): | def run_test(self): | ||||
# due to OS-specific network stats queries, this test works only on Linux | # due to OS-specific network stats queries, this test works only on | ||||
# Linux | |||||
assert(sys.platform.startswith('linux')) | assert(sys.platform.startswith('linux')) | ||||
# find the first non-loopback interface for testing | # find the first non-loopback interface for testing | ||||
non_loopback_ip = None | 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 | non_loopback_ip = ip | ||||
break | break | ||||
if non_loopback_ip is None: | if non_loopback_ip is None: | ||||
assert(not 'This test requires at least one non-loopback IPv4 interface') | assert( | ||||
not 'This test requires at least one non-loopback IPv4 interface') | |||||
print("Using interface %s for testing" % non_loopback_ip) | print("Using interface %s for testing" % non_loopback_ip) | ||||
defaultport = rpc_port(0) | defaultport = rpc_port(0) | ||||
# check default without rpcallowip (IPv4 and IPv6 localhost) | # check default without rpcallowip (IPv4 and IPv6 localhost) | ||||
self.run_bind_test(None, '127.0.0.1', [], | self.run_bind_test(None, '127.0.0.1', [], | ||||
[('127.0.0.1', defaultport), ('::1', defaultport)]) | [('127.0.0.1', defaultport), ('::1', defaultport)]) | ||||
# check default with rpcallowip (IPv6 any) | # check default with rpcallowip (IPv6 any) | ||||
self.run_bind_test(['127.0.0.1'], '127.0.0.1', [], | self.run_bind_test(['127.0.0.1'], '127.0.0.1', [], | ||||
[('::0', defaultport)]) | [('::0', 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', defaultport)]) | ||||
# check only IPv4 localhost (explicit) with alternative port | # check only IPv4 localhost (explicit) with alternative port | ||||
self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171'], | self.run_bind_test( | ||||
['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 on same host | # check only IPv4 localhost (explicit) with multiple alternative ports | ||||
self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171', '127.0.0.1:32172'], | # on same host | ||||
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', 32171), ('127.0.0.1', 32172)]) | [('127.0.0.1', 32171), ('127.0.0.1', 32172)]) | ||||
# 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', 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', defaultport), ('::1', defaultport)]) | ||||
# check only non-loopback interface | # check only non-loopback interface | ||||
self.run_bind_test([non_loopback_ip], non_loopback_ip, [non_loopback_ip], | self.run_bind_test( | ||||
[non_loopback_ip], non_loopback_ip, [non_loopback_ip], | |||||
[(non_loopback_ip, 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([non_loopback_ip], non_loopback_ip, defaultport) | ||||
try: | try: | ||||
self.run_allowip_test(['1.1.1.1'], non_loopback_ip, defaultport) | self.run_allowip_test(['1.1.1.1'], non_loopback_ip, defaultport) | ||||
assert(not 'Connection not denied by rpcallowip as expected') | assert(not 'Connection not denied by rpcallowip as expected') | ||||
except JSONRPCException: | except JSONRPCException: | ||||
pass | pass | ||||
if __name__ == '__main__': | if __name__ == '__main__': | ||||
RPCBindTest().main() | RPCBindTest().main() |