You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

proxy_test.py 8.1KB

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