1#!/usr/bin/env python3
2# Copyright (c) 2014-2016 The Bitcoin Core developers
3# Distributed under the MIT software license, see the accompanying
4# file COPYING or http://www.opensource.org/licenses/mit-license.php.
5
6# Test for -rpcbind, as well as -rpcallowip and -rpcconnect
7
8import tempfile
9import traceback
10
11from test_framework.test_framework import BitcoinTestFramework
12from test_framework.util import *
13from test_framework.netutil import *
14
15class RPCBindTest(BitcoinTestFramework):
16
17    def __init__(self):
18        super().__init__()
19        self.setup_clean_chain = True
20        self.num_nodes = 1
21
22    def setup_network(self):
23        pass
24
25    def setup_nodes(self):
26        pass
27
28    def run_bind_test(self, allow_ips, connect_to, addresses, expected):
29        '''
30        Start a node with requested rpcallowip and rpcbind parameters,
31        then try to connect, and check if the set of bound addresses
32        matches the expected set.
33        '''
34        expected = [(addr_to_hex(addr), port) for (addr, port) in expected]
35        base_args = ['-disablewallet', '-nolisten']
36        if allow_ips:
37            base_args += ['-rpcallowip=' + x for x in allow_ips]
38        binds = ['-rpcbind='+addr for addr in addresses]
39        self.nodes = start_nodes(self.num_nodes, self.options.tmpdir, [base_args + binds], connect_to)
40        try:
41            pid = bitcoind_processes[0].pid
42            assert_equal(set(get_bind_addrs(pid)), set(expected))
43        finally:
44            stop_nodes(self.nodes)
45
46    def run_allowip_test(self, allow_ips, rpchost, rpcport):
47        '''
48        Start a node with rpcallow IP, and request getinfo
49        at a non-localhost IP.
50        '''
51        base_args = ['-disablewallet', '-nolisten'] + ['-rpcallowip='+x for x in allow_ips]
52        self.nodes = start_nodes(self.num_nodes, self.options.tmpdir, [base_args])
53        try:
54            # connect to node through non-loopback interface
55            node = get_rpc_proxy(rpc_url(0, "%s:%d" % (rpchost, rpcport)), 0)
56            node.getinfo()
57        finally:
58            node = None # make sure connection will be garbage collected and closed
59            stop_nodes(self.nodes)
60
61    def run_test(self):
62        # due to OS-specific network stats queries, this test works only on Linux
63        assert(sys.platform.startswith('linux'))
64        # find the first non-loopback interface for testing
65        non_loopback_ip = None
66        for name,ip in all_interfaces():
67            if ip != '127.0.0.1':
68                non_loopback_ip = ip
69                break
70        if non_loopback_ip is None:
71            assert(not 'This test requires at least one non-loopback IPv4 interface')
72        print("Using interface %s for testing" % non_loopback_ip)
73
74        defaultport = rpc_port(0)
75
76        # check default without rpcallowip (IPv4 and IPv6 localhost)
77        self.run_bind_test(None, '127.0.0.1', [],
78            [('127.0.0.1', defaultport), ('::1', defaultport)])
79        # check default with rpcallowip (IPv6 any)
80        self.run_bind_test(['127.0.0.1'], '127.0.0.1', [],
81            [('::0', defaultport)])
82        # check only IPv4 localhost (explicit)
83        self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1'],
84            [('127.0.0.1', defaultport)])
85        # check only IPv4 localhost (explicit) with alternative port
86        self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171'],
87            [('127.0.0.1', 32171)])
88        # check only IPv4 localhost (explicit) with multiple alternative ports on same host
89        self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171', '127.0.0.1:32172'],
90            [('127.0.0.1', 32171), ('127.0.0.1', 32172)])
91        # check only IPv6 localhost (explicit)
92        self.run_bind_test(['[::1]'], '[::1]', ['[::1]'],
93            [('::1', defaultport)])
94        # check both IPv4 and IPv6 localhost (explicit)
95        self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1', '[::1]'],
96            [('127.0.0.1', defaultport), ('::1', defaultport)])
97        # check only non-loopback interface
98        self.run_bind_test([non_loopback_ip], non_loopback_ip, [non_loopback_ip],
99            [(non_loopback_ip, defaultport)])
100
101        # Check that with invalid rpcallowip, we are denied
102        self.run_allowip_test([non_loopback_ip], non_loopback_ip, defaultport)
103        try:
104            self.run_allowip_test(['1.1.1.1'], non_loopback_ip, defaultport)
105            assert(not 'Connection not denied by rpcallowip as expected')
106        except JSONRPCException:
107            pass
108
109if __name__ == '__main__':
110    RPCBindTest ().main ()
111