-
Notifications
You must be signed in to change notification settings - Fork 779
/
rpc_bind.py
executable file
·146 lines (130 loc) · 6.08 KB
/
rpc_bind.py
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
#!/usr/bin/env python3
# Copyright (c) 2014-2019 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 running bitcoind with the -rpcbind and -rpcallowip options."""
from platform import uname
import socket
import sys
from test_framework.netutil import addr_to_hex, all_interfaces, get_bind_addrs
from test_framework.test_framework import BitcoinTestFramework, SkipTest
from test_framework.util import (
assert_equal,
assert_raises_rpc_error,
get_rpc_proxy,
rpc_port,
rpc_url,
)
class RPCBindTest(BitcoinTestFramework):
def set_test_params(self):
self.setup_clean_chain = True
self.bind_to_localhost_only = False
self.num_nodes = 1
def setup_network(self):
self.add_nodes(self.num_nodes, None)
def run_bind_test(self, 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.
'''
self.log.info("Bind test for {}".format(str(addresses)))
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]
parts = connect_to.split(':')
if len(parts) == 2:
self.nodes[0].host = parts[0]
self.nodes[0].rpc_port = parts[1]
else:
self.nodes[0].host = connect_to
self.nodes[0].rpc_port = rpc_port(self.nodes[0].index)
self.start_node(0, base_args + binds)
pid = self.nodes[0].process.pid
assert_equal(set(get_bind_addrs(pid)), set(expected))
self.stop_nodes()
def run_allowip_test(self, allow_ips, rpchost, rpcport):
'''
Start a node with rpcallow IP, and request getnetworkinfo
at a non-localhost IP.
'''
self.log.info("Allow IP test for {}:{}".format(rpchost, rpcport))
node_args = \
['-disablewallet', '-nolisten'] + \
['-rpcallowip=' + x for x in allow_ips] + \
['-rpcbind=' + addr for addr in ['127.0.0.1',
"{}:{}".format(rpchost,
rpcport)]] # Bind to localhost as well so start_nodes doesn't hang
self.nodes[0].host = None
self.start_nodes([node_args])
# connect to node through non-loopback interface
url = rpc_url(self.nodes[0].datadir, self.chain, rpchost, rpcport)
node = get_rpc_proxy(url, 0, coveragedir=self.options.coveragedir)
node.getnetworkinfo()
self.stop_nodes()
def run_test(self):
# due to OS-specific network stats queries, this test works only on
# Linux
if not sys.platform.startswith('linux'):
raise SkipTest("This test can only be run on Linux.")
# WSL in currently not supported (refer to
# https://reviews.bitcoinabc.org/T400 for details).
# This condition should be removed once netstat support is provided by
# Microsoft.
if "microsoft" in uname().version.lower():
raise SkipTest(
"Running this test on WSL is currently not supported")
# 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:
raise SkipTest(
"This test requires at least one non-loopback IPv4 interface.")
try:
s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
s.connect(("::1", 1))
s.close
except OSError:
raise SkipTest("This test requires IPv6 support.")
self.log.info("Using interface {} for testing".format(non_loopback_ip))
defaultport = rpc_port(0)
# check default without rpcallowip (IPv4 and IPv6 localhost)
self.run_bind_test(None, '127.0.0.1', [],
[('127.0.0.1', defaultport), ('::1', defaultport)])
# check default with rpcallowip (IPv4 and IPv6 localhost)
self.run_bind_test(['127.0.0.1'], '127.0.0.1', [],
[('127.0.0.1', defaultport), ('::1', defaultport)])
# check only IPv4 localhost (explicit)
self.run_bind_test(['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
self.run_bind_test(
['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
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)])
# check only IPv6 localhost (explicit)
self.run_bind_test(['[::1]'], '[::1]', ['[::1]'],
[('::1', defaultport)])
# check both IPv4 and IPv6 localhost (explicit)
self.run_bind_test(['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
self.run_bind_test(
[non_loopback_ip], non_loopback_ip, [non_loopback_ip],
[(non_loopback_ip, defaultport)])
# Check that with invalid rpcallowip, we are denied
self.run_allowip_test([non_loopback_ip], non_loopback_ip, defaultport)
assert_raises_rpc_error(-342, "non-JSON HTTP response with '403 Forbidden' from server",
self.run_allowip_test, ['1.1.1.1'], non_loopback_ip, defaultport)
if __name__ == '__main__':
RPCBindTest().main()