Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/urllib3/contrib/socks.py: 10%

1""" 

2This module contains provisional support for SOCKS proxies from within 

3urllib3. This module supports SOCKS4, SOCKS4A (an extension of SOCKS4), and 

4SOCKS5. To enable its functionality, either install PySocks or install this 

5module with the ``socks`` extra. 

6 

7The SOCKS implementation supports the full range of urllib3 features. It also 

8supports the following SOCKS features: 

9 

10- SOCKS4A (``proxy_url='socks4a://...``) 

11- SOCKS4 (``proxy_url='socks4://...``) 

12- SOCKS5 with remote DNS (``proxy_url='socks5h://...``) 

13- SOCKS5 with local DNS (``proxy_url='socks5://...``) 

14- Usernames and passwords for the SOCKS proxy 

15 

16.. note:: 

17 It is recommended to use ``socks5h://`` or ``socks4a://`` schemes in 

18 your ``proxy_url`` to ensure that DNS resolution is done from the remote 

19 server instead of client-side when connecting to a domain name. 

20 

21SOCKS4 supports IPv4 and domain names with the SOCKS4A extension. SOCKS5 

22supports IPv4, IPv6, and domain names. 

23 

24When connecting to a SOCKS4 proxy the ``username`` portion of the ``proxy_url`` 

25will be sent as the ``userid`` section of the SOCKS request: 

26 

27.. code-block:: python 

28 

29 proxy_url="socks4a://<userid>@proxy-host" 

30 

31When connecting to a SOCKS5 proxy the ``username`` and ``password`` portion 

32of the ``proxy_url`` will be sent as the username/password to authenticate 

33with the proxy: 

34 

35.. code-block:: python 

36 

37 proxy_url="socks5h://<username>:<password>@proxy-host" 

38 

39""" 

40 

41from __future__ import annotations 

42 

43try: 

44 import socks # type: ignore[import] 

45except ImportError: 

46 import warnings 

47 

48 from ..exceptions import DependencyWarning 

49 

50 warnings.warn( 

51 ( 

52 "SOCKS support in urllib3 requires the installation of optional " 

53 "dependencies: specifically, PySocks. For more information, see " 

54 "https://urllib3.readthedocs.io/en/latest/contrib.html#socks-proxies" 

55 ), 

56 DependencyWarning, 

57 ) 

58 raise 

59 

60import typing 

61from socket import timeout as SocketTimeout 

62 

63from ..connection import HTTPConnection, HTTPSConnection 

64from ..connectionpool import HTTPConnectionPool, HTTPSConnectionPool 

65from ..exceptions import ConnectTimeoutError, NewConnectionError 

66from ..poolmanager import PoolManager 

67from ..util.url import parse_url 

68 

69try: 

70 import ssl 

71except ImportError: 

72 ssl = None # type: ignore[assignment] 

73 

74from typing import TypedDict 

75 

76 

77class _TYPE_SOCKS_OPTIONS(TypedDict): 

78 socks_version: int 

79 proxy_host: str | None 

80 proxy_port: str | None 

81 username: str | None 

82 password: str | None 

83 rdns: bool 

84 

85 

86class SOCKSConnection(HTTPConnection): 

87 """ 

88 A plain-text HTTP connection that connects via a SOCKS proxy. 

89 """ 

90 

91 def __init__( 

92 self, 

93 _socks_options: _TYPE_SOCKS_OPTIONS, 

94 *args: typing.Any, 

95 **kwargs: typing.Any, 

96 ) -> None: 

97 self._socks_options = _socks_options 

98 super().__init__(*args, **kwargs) 

99 

100 def _new_conn(self) -> socks.socksocket: 

101 """ 

102 Establish a new connection via the SOCKS proxy. 

103 """ 

104 extra_kw: dict[str, typing.Any] = {} 

105 if self.source_address: 

106 extra_kw["source_address"] = self.source_address 

107 

108 if self.socket_options: 

109 extra_kw["socket_options"] = self.socket_options 

110 

111 try: 

112 conn = socks.create_connection( 

113 (self.host, self.port), 

114 proxy_type=self._socks_options["socks_version"], 

115 proxy_addr=self._socks_options["proxy_host"], 

116 proxy_port=self._socks_options["proxy_port"], 

117 proxy_username=self._socks_options["username"], 

118 proxy_password=self._socks_options["password"], 

119 proxy_rdns=self._socks_options["rdns"], 

120 timeout=self.timeout, 

121 **extra_kw, 

122 ) 

123 

124 except SocketTimeout as e: 

125 raise ConnectTimeoutError( 

126 self, 

127 f"Connection to {self.host} timed out. (connect timeout={self.timeout})", 

128 ) from e 

129 

130 except socks.ProxyError as e: 

131 # This is fragile as hell, but it seems to be the only way to raise 

132 # useful errors here. 

133 if e.socket_err: 

134 error = e.socket_err 

135 if isinstance(error, SocketTimeout): 

136 raise ConnectTimeoutError( 

137 self, 

138 f"Connection to {self.host} timed out. (connect timeout={self.timeout})", 

139 ) from e 

140 else: 

141 # Adding `from e` messes with coverage somehow, so it's omitted. 

142 # See #2386. 

143 raise NewConnectionError( 

144 self, f"Failed to establish a new connection: {error}" 

145 ) 

146 else: 

147 raise NewConnectionError( 

148 self, f"Failed to establish a new connection: {e}" 

149 ) from e 

150 

151 except OSError as e: # Defensive: PySocks should catch all these. 

152 raise NewConnectionError( 

153 self, f"Failed to establish a new connection: {e}" 

154 ) from e 

155 

156 return conn 

157 

158 

159# We don't need to duplicate the Verified/Unverified distinction from 

160# urllib3/connection.py here because the HTTPSConnection will already have been 

161# correctly set to either the Verified or Unverified form by that module. This 

162# means the SOCKSHTTPSConnection will automatically be the correct type. 

163class SOCKSHTTPSConnection(SOCKSConnection, HTTPSConnection): 

164 pass 

165 

166 

167class SOCKSHTTPConnectionPool(HTTPConnectionPool): 

168 ConnectionCls = SOCKSConnection 

169 

170 

171class SOCKSHTTPSConnectionPool(HTTPSConnectionPool): 

172 ConnectionCls = SOCKSHTTPSConnection 

173 

174 

175class SOCKSProxyManager(PoolManager): 

176 """ 

177 A version of the urllib3 ProxyManager that routes connections via the 

178 defined SOCKS proxy. 

179 """ 

180 

181 pool_classes_by_scheme = { 

182 "http": SOCKSHTTPConnectionPool, 

183 "https": SOCKSHTTPSConnectionPool, 

184 } 

185 

186 def __init__( 

187 self, 

188 proxy_url: str, 

189 username: str | None = None, 

190 password: str | None = None, 

191 num_pools: int = 10, 

192 headers: typing.Mapping[str, str] | None = None, 

193 **connection_pool_kw: typing.Any, 

194 ): 

195 parsed = parse_url(proxy_url) 

196 

197 if username is None and password is None and parsed.auth is not None: 

198 split = parsed.auth.split(":") 

199 if len(split) == 2: 

200 username, password = split 

201 if parsed.scheme == "socks5": 

202 socks_version = socks.PROXY_TYPE_SOCKS5 

203 rdns = False 

204 elif parsed.scheme == "socks5h": 

205 socks_version = socks.PROXY_TYPE_SOCKS5 

206 rdns = True 

207 elif parsed.scheme == "socks4": 

208 socks_version = socks.PROXY_TYPE_SOCKS4 

209 rdns = False 

210 elif parsed.scheme == "socks4a": 

211 socks_version = socks.PROXY_TYPE_SOCKS4 

212 rdns = True 

213 else: 

214 raise ValueError(f"Unable to determine SOCKS version from {proxy_url}") 

215 

216 self.proxy_url = proxy_url 

217 

218 socks_options = { 

219 "socks_version": socks_version, 

220 "proxy_host": parsed.host, 

221 "proxy_port": parsed.port, 

222 "username": username, 

223 "password": password, 

224 "rdns": rdns, 

225 } 

226 connection_pool_kw["_socks_options"] = socks_options 

227 

228 super().__init__(num_pools, headers, **connection_pool_kw) 

229 

230 self.pool_classes_by_scheme = SOCKSProxyManager.pool_classes_by_scheme