feature_proxy.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2015-2022 The Limenka 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 limenkad with different proxy configuration.
6
7 Test plan:
8 - Start limenkad'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 limenkad side:
12 - `-proxy` (proxy everything)
13 - `-onion` (proxy just onions)
14 - `-proxyrandomize` Circuit randomization
15 - `-cjdnsreachable`
16 - Proxy configurations to test on proxy side,
17 - support no authentication (other proxy)
18 - support no authentication + user/pass authentication (Tor)
19 - proxy on IPv6
20 - proxy over unix domain sockets
21
22 - Create various proxies (as threads)
23 - Create nodes that connect to them
24 - Manipulate the peer connections using addnode (onetry) and observe effects
25 - Test the getpeerinfo `network` field for the peer
26
27 addnode connect to IPv4
28 addnode connect to IPv6
29 addnode connect to onion
30 addnode connect to generic DNS name
31 addnode connect to a CJDNS address
32
33 - Test getnetworkinfo for each node
34
35 - Test passing invalid -proxy
36 - Test passing invalid -onion
37 - Test passing invalid -i2psam
38 - Test passing -onlynet=onion without -proxy or -onion
39 - Test passing -onlynet=onion with -onion=0 and with -noonion
40 - Test passing unknown -onlynet
41 """
42
43 import os
44 import socket
45 import tempfile
46
47 from test_framework.socks5 import Socks5Configuration, Socks5Command, Socks5Server, AddressType
48 from test_framework.test_framework import LimenkaTestFramework
49 from test_framework.util import (
50 assert_equal,
51 p2p_port,
52 )
53 from test_framework.netutil import test_ipv6_local, test_unix_socket
54
55 # Networks returned by RPC getpeerinfo.
56 NET_UNROUTABLE = "not_publicly_routable"
57 NET_IPV4 = "ipv4"
58 NET_IPV6 = "ipv6"
59 NET_ONION = "onion"
60 NET_I2P = "i2p"
61 NET_CJDNS = "cjdns"
62
63 # Networks returned by RPC getnetworkinfo, defined in src/rpc/net.cpp::GetNetworksInfo()
64 NETWORKS = frozenset({NET_IPV4, NET_IPV6, NET_ONION, NET_I2P, NET_CJDNS})
65
66 # Use the shortest temp path possible since UNIX sockets may have as little as 92-char limit
67 socket_path = tempfile.NamedTemporaryFile().name
68
69 class ProxyTest(LimenkaTestFramework):
70 def set_test_params(self):
71 self.num_nodes = 7
72 self.setup_clean_chain = True
73
74 def setup_nodes(self):
75 self.have_ipv6 = test_ipv6_local()
76 self.have_unix_sockets = test_unix_socket()
77 # Create two proxies on different ports
78 # ... one unauthenticated
79 self.conf1 = Socks5Configuration()
80 self.conf1.addr = ('127.0.0.1', p2p_port(self.num_nodes))
81 self.conf1.unauth = True
82 self.conf1.auth = False
83 # ... one supporting authenticated and unauthenticated (Tor)
84 self.conf2 = Socks5Configuration()
85 self.conf2.addr = ('127.0.0.1', p2p_port(self.num_nodes + 1))
86 self.conf2.unauth = True
87 self.conf2.auth = True
88 if self.have_ipv6:
89 # ... one on IPv6 with similar configuration
90 self.conf3 = Socks5Configuration()
91 self.conf3.af = socket.AF_INET6
92 self.conf3.addr = ('::1', p2p_port(self.num_nodes + 2))
93 self.conf3.unauth = True
94 self.conf3.auth = True
95 else:
96 self.log.warning("Testing without local IPv6 support")
97
98 if self.have_unix_sockets:
99 self.conf4 = Socks5Configuration()
100 self.conf4.af = socket.AF_UNIX
101 self.conf4.addr = socket_path
102 self.conf4.unauth = True
103 self.conf4.auth = True
104 else:
105 self.log.warning("Testing without local unix domain sockets support")
106
107 self.serv1 = Socks5Server(self.conf1)
108 self.serv1.start()
109 self.serv2 = Socks5Server(self.conf2)
110 self.serv2.start()
111 if self.have_ipv6:
112 self.serv3 = Socks5Server(self.conf3)
113 self.serv3.start()
114 if self.have_unix_sockets:
115 self.serv4 = Socks5Server(self.conf4)
116 self.serv4.start()
117
118 # We will not try to connect to this.
119 self.i2p_sam = ('127.0.0.1', 7656)
120
121 # Note: proxies are not used to connect to local nodes. This is because the proxy to
122 # use is based on CService.GetNetwork(), which returns NET_UNROUTABLE for localhost.
123 args = [
124 ['-listen', f'-proxy={self.conf1.addr[0]}:{self.conf1.addr[1]}','-proxyrandomize=1'],
125 ['-listen', f'-proxy={self.conf1.addr[0]}:{self.conf1.addr[1]}',f'-onion={self.conf2.addr[0]}:{self.conf2.addr[1]}',
126 f'-i2psam={self.i2p_sam[0]}:{self.i2p_sam[1]}', '-i2pacceptincoming=0', '-proxyrandomize=0'],
127 ['-listen', f'-proxy={self.conf2.addr[0]}:{self.conf2.addr[1]}','-proxyrandomize=1'],
128 [],
129 ['-listen', f'-proxy={self.conf1.addr[0]}:{self.conf1.addr[1]}','-proxyrandomize=1',
130 '-cjdnsreachable'],
131 [],
132 []
133 ]
134 if self.have_ipv6:
135 args[3] = ['-listen', f'-proxy=[{self.conf3.addr[0]}]:{self.conf3.addr[1]}','-proxyrandomize=0', '-noonion']
136 if self.have_unix_sockets:
137 args[5] = ['-listen', f'-proxy=unix:{socket_path}']
138 args[6] = ['-listen', f'-onion=unix:{socket_path}']
139 self.add_nodes(self.num_nodes, extra_args=args)
140 self.start_nodes()
141
142 def network_test(self, node, addr, network):
143 for peer in node.getpeerinfo():
144 if peer["addr"] == addr:
145 assert_equal(peer["network"], network)
146
147 def node_test(self, node, *, proxies, auth, test_onion, test_cjdns):
148 rv = []
149 addr = "15.61.23.23:1234"
150 self.log.debug(f"Test: outgoing IPv4 connection through node {node.index} for address {addr}")
151 # v2transport=False is used to avoid reconnections with v1 being scheduled. These could interfere with later checks.
152 node.addnode(addr, "onetry", v2transport=False)
153 cmd = proxies[0].queue.get()
154 assert isinstance(cmd, Socks5Command)
155 # Note: limenkad's SOCKS5 implementation only sends atyp DOMAINNAME, even if connecting directly to IPv4/IPv6
156 assert_equal(cmd.atyp, AddressType.DOMAINNAME)
157 assert_equal(cmd.addr, b"15.61.23.23")
158 assert_equal(cmd.port, 1234)
159 if not auth:
160 assert_equal(cmd.username, None)
161 assert_equal(cmd.password, None)
162 rv.append(cmd)
163 self.network_test(node, addr, network=NET_IPV4)
164
165 if self.have_ipv6:
166 addr = "[1233:3432:2434:2343:3234:2345:6546:4534]:5443"
167 self.log.debug(f"Test: outgoing IPv6 connection through node {node.index} for address {addr}")
168 node.addnode(addr, "onetry", v2transport=False)
169 cmd = proxies[1].queue.get()
170 assert isinstance(cmd, Socks5Command)
171 # Note: limenkad's SOCKS5 implementation only sends atyp DOMAINNAME, even if connecting directly to IPv4/IPv6
172 assert_equal(cmd.atyp, AddressType.DOMAINNAME)
173 assert_equal(cmd.addr, b"1233:3432:2434:2343:3234:2345:6546:4534")
174 assert_equal(cmd.port, 5443)
175 if not auth:
176 assert_equal(cmd.username, None)
177 assert_equal(cmd.password, None)
178 rv.append(cmd)
179 self.network_test(node, addr, network=NET_IPV6)
180
181 if test_onion:
182 addr = "pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion:8333"
183 self.log.debug(f"Test: outgoing onion connection through node {node.index} for address {addr}")
184 node.addnode(addr, "onetry", v2transport=False)
185 cmd = proxies[2].queue.get()
186 assert isinstance(cmd, Socks5Command)
187 assert_equal(cmd.atyp, AddressType.DOMAINNAME)
188 assert_equal(cmd.addr, b"pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion")
189 assert_equal(cmd.port, 8333)
190 if not auth:
191 assert_equal(cmd.username, None)
192 assert_equal(cmd.password, None)
193 rv.append(cmd)
194 self.network_test(node, addr, network=NET_ONION)
195
196 if test_cjdns:
197 addr = "[fc00:1:2:3:4:5:6:7]:8888"
198 self.log.debug(f"Test: outgoing CJDNS connection through node {node.index} for address {addr}")
199 node.addnode(addr, "onetry", v2transport=False)
200 cmd = proxies[1].queue.get()
201 assert isinstance(cmd, Socks5Command)
202 assert_equal(cmd.atyp, AddressType.DOMAINNAME)
203 assert_equal(cmd.addr, b"fc00:1:2:3:4:5:6:7")
204 assert_equal(cmd.port, 8888)
205 if not auth:
206 assert_equal(cmd.username, None)
207 assert_equal(cmd.password, None)
208 rv.append(cmd)
209 self.network_test(node, addr, network=NET_CJDNS)
210
211 addr = "node.noumenon:8333"
212 self.log.debug(f"Test: outgoing DNS name connection through node {node.index} for address {addr}")
213 node.addnode(addr, "onetry", v2transport=False)
214 cmd = proxies[3].queue.get()
215 assert isinstance(cmd, Socks5Command)
216 assert_equal(cmd.atyp, AddressType.DOMAINNAME)
217 assert_equal(cmd.addr, b"node.noumenon")
218 assert_equal(cmd.port, 8333)
219 if not auth:
220 assert_equal(cmd.username, None)
221 assert_equal(cmd.password, None)
222 rv.append(cmd)
223 self.network_test(node, addr, network=NET_UNROUTABLE)
224
225 return rv
226
227 def run_test(self):
228 # basic -proxy
229 self.node_test(self.nodes[0],
230 proxies=[self.serv1, self.serv1, self.serv1, self.serv1],
231 auth=False, test_onion=True, test_cjdns=False)
232
233 # -proxy plus -onion
234 self.node_test(self.nodes[1],
235 proxies=[self.serv1, self.serv1, self.serv2, self.serv1],
236 auth=False, test_onion=True, test_cjdns=False)
237
238 # -proxy plus -onion, -proxyrandomize
239 rv = self.node_test(self.nodes[2],
240 proxies=[self.serv2, self.serv2, self.serv2, self.serv2],
241 auth=True, test_onion=True, test_cjdns=False)
242 # Check that credentials as used for -proxyrandomize connections are unique
243 credentials = set((x.username,x.password) for x in rv)
244 assert_equal(len(credentials), len(rv))
245
246 if self.have_ipv6:
247 # proxy on IPv6 localhost
248 self.node_test(self.nodes[3],
249 proxies=[self.serv3, self.serv3, self.serv3, self.serv3],
250 auth=False, test_onion=False, test_cjdns=False)
251
252 # -proxy=unauth -proxyrandomize=1 -cjdnsreachable
253 self.node_test(self.nodes[4],
254 proxies=[self.serv1, self.serv1, self.serv1, self.serv1],
255 auth=False, test_onion=True, test_cjdns=True)
256
257 if self.have_unix_sockets:
258 self.node_test(self.nodes[5],
259 proxies=[self.serv4, self.serv4, self.serv4, self.serv4],
260 auth=True, test_onion=True, test_cjdns=False)
261
262
263 def networks_dict(d):
264 r = {}
265 for x in d['networks']:
266 r[x['name']] = x
267 return r
268
269 self.log.info("Test RPC getnetworkinfo")
270 nodes_network_info = []
271
272 self.log.debug("Test that setting -proxy disables local address discovery, i.e. -discover=0")
273 for node in self.nodes:
274 network_info = node.getnetworkinfo()
275 assert_equal(network_info["localaddresses"], [])
276 nodes_network_info.append(network_info)
277
278 n0 = networks_dict(nodes_network_info[0])
279 assert_equal(NETWORKS, n0.keys())
280 for net in NETWORKS:
281 if net == NET_I2P:
282 expected_proxy = ''
283 expected_randomize = False
284 else:
285 expected_proxy = '%s:%i' % (self.conf1.addr)
286 expected_randomize = True
287 assert_equal(n0[net]['proxy'], expected_proxy)
288 assert_equal(n0[net]['proxy_randomize_credentials'], expected_randomize)
289 assert_equal(n0['onion']['reachable'], True)
290 assert_equal(n0['i2p']['reachable'], False)
291 assert_equal(n0['cjdns']['reachable'], False)
292
293 n1 = networks_dict(nodes_network_info[1])
294 assert_equal(NETWORKS, n1.keys())
295 for net in ['ipv4', 'ipv6']:
296 assert_equal(n1[net]['proxy'], f'{self.conf1.addr[0]}:{self.conf1.addr[1]}')
297 assert_equal(n1[net]['proxy_randomize_credentials'], False)
298 assert_equal(n1['onion']['proxy'], f'{self.conf2.addr[0]}:{self.conf2.addr[1]}')
299 assert_equal(n1['onion']['proxy_randomize_credentials'], False)
300 assert_equal(n1['onion']['reachable'], True)
301 assert_equal(n1['i2p']['proxy'], f'{self.i2p_sam[0]}:{self.i2p_sam[1]}')
302 assert_equal(n1['i2p']['proxy_randomize_credentials'], False)
303 assert_equal(n1['i2p']['reachable'], True)
304
305 n2 = networks_dict(nodes_network_info[2])
306 assert_equal(NETWORKS, n2.keys())
307 proxy = f'{self.conf2.addr[0]}:{self.conf2.addr[1]}'
308 for net in NETWORKS:
309 if net == NET_I2P:
310 expected_proxy = ''
311 expected_randomize = False
312 else:
313 expected_proxy = proxy
314 expected_randomize = True
315 assert_equal(n2[net]['proxy'], expected_proxy)
316 assert_equal(n2[net]['proxy_randomize_credentials'], expected_randomize)
317 assert_equal(n2['onion']['reachable'], True)
318 assert_equal(n2['i2p']['reachable'], False)
319 assert_equal(n2['cjdns']['reachable'], False)
320
321 if self.have_ipv6:
322 n3 = networks_dict(nodes_network_info[3])
323 assert_equal(NETWORKS, n3.keys())
324 proxy = f'[{self.conf3.addr[0]}]:{self.conf3.addr[1]}'
325 for net in NETWORKS:
326 expected_proxy = '' if net == NET_I2P or net == NET_ONION else proxy
327 assert_equal(n3[net]['proxy'], expected_proxy)
328 assert_equal(n3[net]['proxy_randomize_credentials'], False)
329 assert_equal(n3['onion']['reachable'], False)
330 assert_equal(n3['i2p']['reachable'], False)
331 assert_equal(n3['cjdns']['reachable'], False)
332
333 n4 = networks_dict(nodes_network_info[4])
334 assert_equal(NETWORKS, n4.keys())
335 for net in NETWORKS:
336 if net == NET_I2P:
337 expected_proxy = ''
338 expected_randomize = False
339 else:
340 expected_proxy = '%s:%i' % (self.conf1.addr)
341 expected_randomize = True
342 assert_equal(n4[net]['proxy'], expected_proxy)
343 assert_equal(n4[net]['proxy_randomize_credentials'], expected_randomize)
344 assert_equal(n4['onion']['reachable'], True)
345 assert_equal(n4['i2p']['reachable'], False)
346 assert_equal(n4['cjdns']['reachable'], True)
347
348 if self.have_unix_sockets:
349 n5 = networks_dict(nodes_network_info[5])
350 assert_equal(NETWORKS, n5.keys())
351 for net in NETWORKS:
352 if net == NET_I2P:
353 expected_proxy = ''
354 expected_randomize = False
355 else:
356 expected_proxy = 'unix:' + self.conf4.addr # no port number
357 expected_randomize = True
358 assert_equal(n5[net]['proxy'], expected_proxy)
359 assert_equal(n5[net]['proxy_randomize_credentials'], expected_randomize)
360 assert_equal(n5['onion']['reachable'], True)
361 assert_equal(n5['i2p']['reachable'], False)
362 assert_equal(n5['cjdns']['reachable'], False)
363
364 n6 = networks_dict(nodes_network_info[6])
365 assert_equal(NETWORKS, n6.keys())
366 for net in NETWORKS:
367 if net != NET_ONION:
368 expected_proxy = ''
369 expected_randomize = False
370 else:
371 expected_proxy = 'unix:' + self.conf4.addr # no port number
372 expected_randomize = True
373 assert_equal(n6[net]['proxy'], expected_proxy)
374 assert_equal(n6[net]['proxy_randomize_credentials'], expected_randomize)
375 assert_equal(n6['onion']['reachable'], True)
376 assert_equal(n6['i2p']['reachable'], False)
377 assert_equal(n6['cjdns']['reachable'], False)
378
379 self.stop_node(1)
380
381 self.log.info("Test passing invalid -proxy hostname raises expected init error")
382 self.nodes[1].extra_args = ["-proxy=abc..abc:23456"]
383 msg = "Error: Invalid -proxy address or hostname: 'abc..abc:23456'"
384 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
385
386 self.log.info("Test passing invalid -proxy port raises expected init error")
387 self.nodes[1].extra_args = ["-proxy=192.0.0.1:def"]
388 msg = "Error: Invalid port specified in -proxy: '192.0.0.1:def'"
389 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
390
391 self.log.info("Test passing invalid -onion hostname raises expected init error")
392 self.nodes[1].extra_args = ["-onion=xyz..xyz:23456"]
393 msg = "Error: Invalid -onion address or hostname: 'xyz..xyz:23456'"
394 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
395
396 self.log.info("Test passing invalid -onion port raises expected init error")
397 self.nodes[1].extra_args = ["-onion=192.0.0.1:def"]
398 msg = "Error: Invalid port specified in -onion: '192.0.0.1:def'"
399 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
400
401 self.log.info("Test passing invalid -i2psam hostname raises expected init error")
402 self.nodes[1].extra_args = ["-i2psam=def..def:23456"]
403 msg = "Error: Invalid -i2psam address or hostname: 'def..def:23456'"
404 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
405
406 self.log.info("Test passing invalid -i2psam port raises expected init error")
407 self.nodes[1].extra_args = ["-i2psam=192.0.0.1:def"]
408 msg = "Error: Invalid port specified in -i2psam: '192.0.0.1:def'"
409 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
410
411 self.log.info("Test passing invalid -onlynet=i2p without -i2psam raises expected init error")
412 self.nodes[1].extra_args = ["-onlynet=i2p"]
413 msg = "Error: Outbound connections restricted to i2p (-onlynet=i2p) but -i2psam is not provided"
414 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
415
416 self.log.info("Test passing invalid -onlynet=cjdns without -cjdnsreachable raises expected init error")
417 self.nodes[1].extra_args = ["-onlynet=cjdns"]
418 msg = "Error: Outbound connections restricted to CJDNS (-onlynet=cjdns) but -cjdnsreachable is not provided"
419 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
420
421 self.log.info("Test passing -onlynet=onion with -onion=0/-noonion raises expected init error")
422 msg = (
423 "Error: Outbound connections restricted to Tor (-onlynet=onion) but "
424 "the proxy for reaching the Tor network is explicitly forbidden: -onion=0"
425 )
426 for arg in ["-onion=0", "-noonion"]:
427 self.nodes[1].extra_args = ["-onlynet=onion", arg]
428 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
429
430 self.log.info("Test passing -onlynet=onion without -proxy, -onion or -listenonion raises expected init error")
431 self.nodes[1].extra_args = ["-onlynet=onion", "-listenonion=0"]
432 msg = (
433 "Error: Outbound connections restricted to Tor (-onlynet=onion) but the proxy for "
434 "reaching the Tor network is not provided: none of -proxy, -onion or -listenonion is given"
435 )
436 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
437
438 self.log.info("Test passing -onlynet=onion without -proxy or -onion but with -listenonion=1 is ok")
439 self.start_node(1, extra_args=["-onlynet=onion", "-listenonion=1"])
440 self.stop_node(1, expected_stderr=f'Warning: You are using a common listening port (127.0.0.1:{p2p_port(1)}) for both Tor and local connections. All connections to this port will be assumed to be Tor connections, and will be denied any whitelist permissions. If this is not your intent, setup a separate -bind=<addr>[:<port>]=onion configuration, or set -listenonion=0.')
441
442 self.log.info("Test passing unknown network to -onlynet raises expected init error")
443 self.nodes[1].extra_args = ["-onlynet=abc"]
444 msg = "Error: Unknown network specified in -onlynet: 'abc'"
445 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
446
447 self.log.info("Test passing trailing '=' raises expected init error")
448 self.nodes[1].extra_args = ["-proxy=127.0.0.1:9050="]
449 msg = "Error: Invalid -proxy address or hostname, ends with '=': '127.0.0.1:9050='"
450 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
451
452 self.log.info("Test passing unrecognized network raises expected init error")
453 self.nodes[1].extra_args = ["-proxy=127.0.0.1:9050=foo"]
454 msg = "Error: Unrecognized network in -proxy='127.0.0.1:9050=foo': 'foo'"
455 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
456
457 self.log.info("Test passing proxy only for IPv6")
458 self.start_node(1, extra_args=["-proxy=127.6.6.6:6666=ipv6"])
459 nets = networks_dict(self.nodes[1].getnetworkinfo())
460 assert_equal(nets["ipv4"]["proxy"], "")
461 assert_equal(nets["ipv6"]["proxy"], "127.6.6.6:6666")
462 self.stop_node(1)
463
464 self.log.info("Test passing separate proxy for IPv4 and IPv6")
465 self.start_node(1, extra_args=["-proxy=127.4.4.4:4444=ipv4", "-proxy=127.6.6.6:6666=ipv6"])
466 nets = networks_dict(self.nodes[1].getnetworkinfo())
467 assert_equal(nets["ipv4"]["proxy"], "127.4.4.4:4444")
468 assert_equal(nets["ipv6"]["proxy"], "127.6.6.6:6666")
469 self.stop_node(1)
470
471 self.log.info("Test overriding the Tor proxy")
472 self.start_node(1, extra_args=["-proxy=127.1.1.1:1111", "-proxy=127.2.2.2:2222=tor"])
473 nets = networks_dict(self.nodes[1].getnetworkinfo())
474 assert_equal(nets["ipv4"]["proxy"], "127.1.1.1:1111")
475 assert_equal(nets["ipv6"]["proxy"], "127.1.1.1:1111")
476 assert_equal(nets["onion"]["proxy"], "127.2.2.2:2222")
477 self.stop_node(1)
478
479 self.log.info("Test removing CJDNS proxy")
480 self.start_node(1, extra_args=["-proxy=127.1.1.1:1111", "-proxy=0=cjdns"])
481 nets = networks_dict(self.nodes[1].getnetworkinfo())
482 assert_equal(nets["ipv4"]["proxy"], "127.1.1.1:1111")
483 assert_equal(nets["ipv6"]["proxy"], "127.1.1.1:1111")
484 assert_equal(nets["onion"]["proxy"], "127.1.1.1:1111")
485 assert_equal(nets["cjdns"]["proxy"], "")
486 self.stop_node(1)
487
488 self.log.info("Test passing too-long unix path to -proxy raises init error")
489 self.nodes[1].extra_args = [f"-proxy=unix:{'x' * 1000}"]
490 if self.have_unix_sockets:
491 msg = f"Error: Invalid -proxy address or hostname: 'unix:{'x' * 1000}'"
492 else:
493 # If unix sockets are not supported, the file path is incorrectly interpreted as host:port
494 msg = f"Error: Invalid port specified in -proxy: 'unix:{'x' * 1000}'"
495 self.nodes[1].assert_start_raises_init_error(expected_msg=msg)
496
497 # Cleanup socket path we established outside the individual test directory.
498 if self.have_unix_sockets:
499 os.unlink(socket_path)
500
501 if __name__ == '__main__':
502 ProxyTest(__file__).main()
503