152 lines
5.5 KiB
152 lines
5.5 KiB
#!/usr/bin/env python2 |
|
# Copyright (c) 2014-2015 The Bitcoin Core developers |
|
# Distributed under the MIT 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 |
|
|
|
import json |
|
import shutil |
|
import subprocess |
|
import tempfile |
|
import traceback |
|
|
|
from test_framework.util import * |
|
from test_framework.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, rpcport): |
|
''' |
|
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, rpcport,) |
|
node = get_rpc_proxy(url, 1) |
|
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) |
|
|
|
defaultport = rpc_port(0) |
|
|
|
# check default without rpcallowip (IPv4 and IPv6 localhost) |
|
run_bind_test(tmpdir, None, '127.0.0.1', [], |
|
[('127.0.0.1', defaultport), ('::1', defaultport)]) |
|
# check default with rpcallowip (IPv6 any) |
|
run_bind_test(tmpdir, ['127.0.0.1'], '127.0.0.1', [], |
|
[('::0', defaultport)]) |
|
# 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', defaultport)]) |
|
# 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', defaultport)]) |
|
# 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', defaultport), ('::1', defaultport)]) |
|
# check only non-loopback interface |
|
run_bind_test(tmpdir, [non_loopback_ip], non_loopback_ip, [non_loopback_ip], |
|
[(non_loopback_ip, defaultport)]) |
|
|
|
# Check that with invalid rpcallowip, we are denied |
|
run_allowip_test(tmpdir, [non_loopback_ip], non_loopback_ip, defaultport) |
|
try: |
|
run_allowip_test(tmpdir, ['1.1.1.1'], non_loopback_ip, defaultport) |
|
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()
|
|
|