From b5ad5e783d6f636d5ca5703919d05fd0119a34fc Mon Sep 17 00:00:00 2001 From: "Wladimir J. van der Laan" Date: Mon, 7 Apr 2014 17:29:36 +0200 Subject: Add Python test for -rpcbind and -rpcallowip Add a new test, `rpcbind_test.py`, that extensively tests the new `-rpcbind` functionality. --- qa/rpc-tests/rpcbind_test.py | 152 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 152 insertions(+) create mode 100755 qa/rpc-tests/rpcbind_test.py (limited to 'qa/rpc-tests/rpcbind_test.py') diff --git a/qa/rpc-tests/rpcbind_test.py b/qa/rpc-tests/rpcbind_test.py new file mode 100755 index 0000000000..a31f8d98ef --- /dev/null +++ b/qa/rpc-tests/rpcbind_test.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python +# Copyright (c) 2014 The Bitcoin Core developers +# Distributed under the MIT/X11 software license, see the accompanying +# file COPYING or http://www.opensource.org/licenses/mit-license.php. + +# Test for -rpcbind, as well as -rpcallowip and -rpcconnect + +# Add python-bitcoinrpc to module search path: +import os +import sys +sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), "python-bitcoinrpc")) + +import json +import shutil +import subprocess +import tempfile +import traceback + +from bitcoinrpc.authproxy import AuthServiceProxy, JSONRPCException +from util import * +from netutil import * + +def run_bind_test(tmpdir, allow_ips, connect_to, addresses, expected): + ''' + Start a node with requested rpcallowip and rpcbind parameters, + then try to connect, and check if the set of bound addresses + matches the expected set. + ''' + expected = [(addr_to_hex(addr), port) for (addr, port) in expected] + base_args = ['-disablewallet', '-nolisten'] + if allow_ips: + base_args += ['-rpcallowip=' + x for x in allow_ips] + binds = ['-rpcbind='+addr for addr in addresses] + nodes = start_nodes(1, tmpdir, [base_args + binds], connect_to) + try: + pid = bitcoind_processes[0].pid + assert_equal(set(get_bind_addrs(pid)), set(expected)) + finally: + stop_nodes(nodes) + wait_bitcoinds() + +def run_allowip_test(tmpdir, allow_ips, rpchost): + ''' + Start a node with rpcwallow IP, and request getinfo + at a non-localhost IP. + ''' + base_args = ['-disablewallet', '-nolisten'] + ['-rpcallowip='+x for x in allow_ips] + nodes = start_nodes(1, tmpdir, [base_args]) + try: + # connect to node through non-loopback interface + url = "http://rt:rt@%s:%d" % (rpchost, START_RPC_PORT,) + node = AuthServiceProxy(url) + node.getinfo() + finally: + node = None # make sure connection will be garbage collected and closed + stop_nodes(nodes) + wait_bitcoinds() + + +def run_test(tmpdir): + assert(sys.platform == 'linux2') # due to OS-specific network stats queries, this test works only on Linux + # find the first non-loopback interface for testing + non_loopback_ip = None + for name,ip in all_interfaces(): + if ip != '127.0.0.1': + non_loopback_ip = ip + break + if non_loopback_ip is None: + assert(not 'This test requires at least one non-loopback IPv4 interface') + print("Using interface %s for testing" % non_loopback_ip) + + # check default without rpcallowip (IPv4 and IPv6 localhost) + run_bind_test(tmpdir, None, '127.0.0.1', [], + [('127.0.0.1', 11100), ('::1', 11100)]) + # check default with rpcallowip (IPv6 any) + run_bind_test(tmpdir, ['127.0.0.1'], '127.0.0.1', [], + [('::0', 11100)]) + # check only IPv4 localhost (explicit) + run_bind_test(tmpdir, ['127.0.0.1'], '127.0.0.1', ['127.0.0.1'], + [('127.0.0.1', START_RPC_PORT)]) + # check only IPv4 localhost (explicit) with alternative port + run_bind_test(tmpdir, ['127.0.0.1'], '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 + run_bind_test(tmpdir, ['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)]) + # check only IPv6 localhost (explicit) + run_bind_test(tmpdir, ['[::1]'], '[::1]', ['[::1]'], + [('::1', 11100)]) + # check both IPv4 and IPv6 localhost (explicit) + run_bind_test(tmpdir, ['127.0.0.1'], '127.0.0.1', ['127.0.0.1', '[::1]'], + [('127.0.0.1', START_RPC_PORT), ('::1', START_RPC_PORT)]) + # check only non-loopback interface + run_bind_test(tmpdir, [non_loopback_ip], non_loopback_ip, [non_loopback_ip], + [(non_loopback_ip, START_RPC_PORT)]) + + # Check that with invalid rpcallowip, we are denied + run_allowip_test(tmpdir, [non_loopback_ip], non_loopback_ip) + try: + run_allowip_test(tmpdir, ['1.1.1.1'], non_loopback_ip) + assert(not 'Connection not denied by rpcallowip as expected') + except ValueError: + pass + +def main(): + import optparse + + parser = optparse.OptionParser(usage="%prog [options]") + parser.add_option("--nocleanup", dest="nocleanup", default=False, action="store_true", + help="Leave bitcoinds and test.* datadir on exit or error") + parser.add_option("--srcdir", dest="srcdir", default="../../src", + help="Source directory containing bitcoind/bitcoin-cli (default: %default%)") + parser.add_option("--tmpdir", dest="tmpdir", default=tempfile.mkdtemp(prefix="test"), + help="Root directory for datadirs") + (options, args) = parser.parse_args() + + os.environ['PATH'] = options.srcdir+":"+os.environ['PATH'] + + check_json_precision() + + success = False + nodes = [] + try: + print("Initializing test directory "+options.tmpdir) + if not os.path.isdir(options.tmpdir): + os.makedirs(options.tmpdir) + initialize_chain(options.tmpdir) + + run_test(options.tmpdir) + + success = True + + except AssertionError as e: + print("Assertion failed: "+e.message) + except Exception as e: + print("Unexpected exception caught during testing: "+str(e)) + traceback.print_tb(sys.exc_info()[2]) + + if not options.nocleanup: + print("Cleaning up") + wait_bitcoinds() + shutil.rmtree(options.tmpdir) + + if success: + print("Tests successful") + sys.exit(0) + else: + print("Failed") + sys.exit(1) + +if __name__ == '__main__': + main() -- cgit v1.2.3