aboutsummaryrefslogtreecommitdiff
path: root/qa/rpc-tests/rpcbind_test.py
blob: 7b7c01f993cf4cf1fcf3b105c5bef90955cb263b (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
#!/usr/bin/env python3
# Copyright (c) 2014-2016 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

# TODO extend this test from the test framework (like all other tests)

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.startswith('linux')) # 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()