1#!/usr/bin/env python3
2# Copyright (c) 2015-2019 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"""Test bitcoind with different proxy configuration.
6
7Test plan:
8- Start bitcoind's with different proxy configurations
9- Use addnode to initiate connections
10- Verify that proxies are connected to, and the right connection command is given
11- Proxy configurations to test on bitcoind side:
12    - `-proxy` (proxy everything)
13    - `-onion` (proxy just onions)
14    - `-proxyrandomize` Circuit randomization
15- Proxy configurations to test on proxy side,
16    - support no authentication (other proxy)
17    - support no authentication + user/pass authentication (Tor)
18    - proxy on IPv6
19
20- Create various proxies (as threads)
21- Create bitcoinds that connect to them
22- Manipulate the bitcoinds using addnode (onetry) an observe effects
23
24addnode connect to IPv4
25addnode connect to IPv6
26addnode connect to onion
27addnode connect to generic DNS name
28"""
29
30import socket
31import os
32
33from test_framework.socks5 import Socks5Configuration, Socks5Command, Socks5Server, AddressType
34from test_framework.test_framework import BitcoinTestFramework
35from test_framework.util import (
36    PORT_MIN,
37    PORT_RANGE,
38    assert_equal,
39)
40from test_framework.netutil import test_ipv6_local
41
42RANGE_BEGIN = PORT_MIN + 2 * PORT_RANGE  # Start after p2p and rpc ports
43
44class ProxyTest(BitcoinTestFramework):
45    def set_test_params(self):
46        self.num_nodes = 4
47        self.setup_clean_chain = True
48
49    def setup_nodes(self):
50        self.have_ipv6 = test_ipv6_local()
51        # Create two proxies on different ports
52        # ... one unauthenticated
53        self.conf1 = Socks5Configuration()
54        self.conf1.addr = ('127.0.0.1', RANGE_BEGIN + (os.getpid() % 1000))
55        self.conf1.unauth = True
56        self.conf1.auth = False
57        # ... one supporting authenticated and unauthenticated (Tor)
58        self.conf2 = Socks5Configuration()
59        self.conf2.addr = ('127.0.0.1', RANGE_BEGIN + 1000 + (os.getpid() % 1000))
60        self.conf2.unauth = True
61        self.conf2.auth = True
62        if self.have_ipv6:
63            # ... one on IPv6 with similar configuration
64            self.conf3 = Socks5Configuration()
65            self.conf3.af = socket.AF_INET6
66            self.conf3.addr = ('::1', RANGE_BEGIN + 2000 + (os.getpid() % 1000))
67            self.conf3.unauth = True
68            self.conf3.auth = True
69        else:
70            self.log.warning("Testing without local IPv6 support")
71
72        self.serv1 = Socks5Server(self.conf1)
73        self.serv1.start()
74        self.serv2 = Socks5Server(self.conf2)
75        self.serv2.start()
76        if self.have_ipv6:
77            self.serv3 = Socks5Server(self.conf3)
78            self.serv3.start()
79
80        # Note: proxies are not used to connect to local nodes
81        # this is because the proxy to use is based on CService.GetNetwork(), which return NET_UNROUTABLE for localhost
82        args = [
83            ['-listen', '-proxy=%s:%i' % (self.conf1.addr),'-proxyrandomize=1'],
84            ['-listen', '-proxy=%s:%i' % (self.conf1.addr),'-onion=%s:%i' % (self.conf2.addr),'-proxyrandomize=0'],
85            ['-listen', '-proxy=%s:%i' % (self.conf2.addr),'-proxyrandomize=1'],
86            []
87            ]
88        if self.have_ipv6:
89            args[3] = ['-listen', '-proxy=[%s]:%i' % (self.conf3.addr),'-proxyrandomize=0', '-noonion']
90        self.add_nodes(self.num_nodes, extra_args=args)
91        self.start_nodes()
92
93    def node_test(self, node, proxies, auth, test_onion=True):
94        rv = []
95        # Test: outgoing IPv4 connection through node
96        node.addnode("15.61.23.23:1234", "onetry")
97        cmd = proxies[0].queue.get()
98        assert isinstance(cmd, Socks5Command)
99        # Note: bitcoind's SOCKS5 implementation only sends atyp DOMAINNAME, even if connecting directly to IPv4/IPv6
100        assert_equal(cmd.atyp, AddressType.DOMAINNAME)
101        assert_equal(cmd.addr, b"15.61.23.23")
102        assert_equal(cmd.port, 1234)
103        if not auth:
104            assert_equal(cmd.username, None)
105            assert_equal(cmd.password, None)
106        rv.append(cmd)
107
108        if self.have_ipv6:
109            # Test: outgoing IPv6 connection through node
110            node.addnode("[1233:3432:2434:2343:3234:2345:6546:4534]:5443", "onetry")
111            cmd = proxies[1].queue.get()
112            assert isinstance(cmd, Socks5Command)
113            # Note: bitcoind's SOCKS5 implementation only sends atyp DOMAINNAME, even if connecting directly to IPv4/IPv6
114            assert_equal(cmd.atyp, AddressType.DOMAINNAME)
115            assert_equal(cmd.addr, b"1233:3432:2434:2343:3234:2345:6546:4534")
116            assert_equal(cmd.port, 5443)
117            if not auth:
118                assert_equal(cmd.username, None)
119                assert_equal(cmd.password, None)
120            rv.append(cmd)
121
122        if test_onion:
123            # Test: outgoing onion connection through node
124            node.addnode("bitcoinostk4e4re.onion:8333", "onetry")
125            cmd = proxies[2].queue.get()
126            assert isinstance(cmd, Socks5Command)
127            assert_equal(cmd.atyp, AddressType.DOMAINNAME)
128            assert_equal(cmd.addr, b"bitcoinostk4e4re.onion")
129            assert_equal(cmd.port, 8333)
130            if not auth:
131                assert_equal(cmd.username, None)
132                assert_equal(cmd.password, None)
133            rv.append(cmd)
134
135        # Test: outgoing DNS name connection through node
136        node.addnode("node.noumenon:8333", "onetry")
137        cmd = proxies[3].queue.get()
138        assert isinstance(cmd, Socks5Command)
139        assert_equal(cmd.atyp, AddressType.DOMAINNAME)
140        assert_equal(cmd.addr, b"node.noumenon")
141        assert_equal(cmd.port, 8333)
142        if not auth:
143            assert_equal(cmd.username, None)
144            assert_equal(cmd.password, None)
145        rv.append(cmd)
146
147        return rv
148
149    def run_test(self):
150        # basic -proxy
151        self.node_test(self.nodes[0], [self.serv1, self.serv1, self.serv1, self.serv1], False)
152
153        # -proxy plus -onion
154        self.node_test(self.nodes[1], [self.serv1, self.serv1, self.serv2, self.serv1], False)
155
156        # -proxy plus -onion, -proxyrandomize
157        rv = self.node_test(self.nodes[2], [self.serv2, self.serv2, self.serv2, self.serv2], True)
158        # Check that credentials as used for -proxyrandomize connections are unique
159        credentials = set((x.username,x.password) for x in rv)
160        assert_equal(len(credentials), len(rv))
161
162        if self.have_ipv6:
163            # proxy on IPv6 localhost
164            self.node_test(self.nodes[3], [self.serv3, self.serv3, self.serv3, self.serv3], False, False)
165
166        def networks_dict(d):
167            r = {}
168            for x in d['networks']:
169                r[x['name']] = x
170            return r
171
172        # test RPC getnetworkinfo
173        n0 = networks_dict(self.nodes[0].getnetworkinfo())
174        for net in ['ipv4','ipv6','onion']:
175            assert_equal(n0[net]['proxy'], '%s:%i' % (self.conf1.addr))
176            assert_equal(n0[net]['proxy_randomize_credentials'], True)
177        assert_equal(n0['onion']['reachable'], True)
178
179        n1 = networks_dict(self.nodes[1].getnetworkinfo())
180        for net in ['ipv4','ipv6']:
181            assert_equal(n1[net]['proxy'], '%s:%i' % (self.conf1.addr))
182            assert_equal(n1[net]['proxy_randomize_credentials'], False)
183        assert_equal(n1['onion']['proxy'], '%s:%i' % (self.conf2.addr))
184        assert_equal(n1['onion']['proxy_randomize_credentials'], False)
185        assert_equal(n1['onion']['reachable'], True)
186
187        n2 = networks_dict(self.nodes[2].getnetworkinfo())
188        for net in ['ipv4','ipv6','onion']:
189            assert_equal(n2[net]['proxy'], '%s:%i' % (self.conf2.addr))
190            assert_equal(n2[net]['proxy_randomize_credentials'], True)
191        assert_equal(n2['onion']['reachable'], True)
192
193        if self.have_ipv6:
194            n3 = networks_dict(self.nodes[3].getnetworkinfo())
195            for net in ['ipv4','ipv6']:
196                assert_equal(n3[net]['proxy'], '[%s]:%i' % (self.conf3.addr))
197                assert_equal(n3[net]['proxy_randomize_credentials'], False)
198            assert_equal(n3['onion']['reachable'], False)
199
200if __name__ == '__main__':
201    ProxyTest().main()
202