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

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 

74try: 

75 from typing import TypedDict 

76 

77 class _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 

85except ImportError: # Python 3.7 

86 _TYPE_SOCKS_OPTIONS = typing.Dict[str, typing.Any] # type: ignore[misc, assignment] 

87 

88 

89class SOCKSConnection(HTTPConnection): 

90 """ 

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

92 """ 

93 

94 def __init__( 

95 self, 

96 _socks_options: _TYPE_SOCKS_OPTIONS, 

97 *args: typing.Any, 

98 **kwargs: typing.Any, 

99 ) -> None: 

100 self._socks_options = _socks_options 

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

102 

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

104 """ 

105 Establish a new connection via the SOCKS proxy. 

106 """ 

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

108 if self.source_address: 

109 extra_kw["source_address"] = self.source_address 

110 

111 if self.socket_options: 

112 extra_kw["socket_options"] = self.socket_options 

113 

114 try: 

115 conn = socks.create_connection( 

116 (self.host, self.port), 

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

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

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

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

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

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

123 timeout=self.timeout, 

124 **extra_kw, 

125 ) 

126 

127 except SocketTimeout as e: 

128 raise ConnectTimeoutError( 

129 self, 

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

131 ) from e 

132 

133 except socks.ProxyError as e: 

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

135 # useful errors here. 

136 if e.socket_err: 

137 error = e.socket_err 

138 if isinstance(error, SocketTimeout): 

139 raise ConnectTimeoutError( 

140 self, 

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

142 ) from e 

143 else: 

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

145 # See #2386. 

146 raise NewConnectionError( 

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

148 ) 

149 else: 

150 raise NewConnectionError( 

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

152 ) from e 

153 

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

155 raise NewConnectionError( 

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

157 ) from e 

158 

159 return conn 

160 

161 

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

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

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

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

166class SOCKSHTTPSConnection(SOCKSConnection, HTTPSConnection): 

167 pass 

168 

169 

170class SOCKSHTTPConnectionPool(HTTPConnectionPool): 

171 ConnectionCls = SOCKSConnection 

172 

173 

174class SOCKSHTTPSConnectionPool(HTTPSConnectionPool): 

175 ConnectionCls = SOCKSHTTPSConnection 

176 

177 

178class SOCKSProxyManager(PoolManager): 

179 """ 

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

181 defined SOCKS proxy. 

182 """ 

183 

184 pool_classes_by_scheme = { 

185 "http": SOCKSHTTPConnectionPool, 

186 "https": SOCKSHTTPSConnectionPool, 

187 } 

188 

189 def __init__( 

190 self, 

191 proxy_url: str, 

192 username: str | None = None, 

193 password: str | None = None, 

194 num_pools: int = 10, 

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

196 **connection_pool_kw: typing.Any, 

197 ): 

198 parsed = parse_url(proxy_url) 

199 

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

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

202 if len(split) == 2: 

203 username, password = split 

204 if parsed.scheme == "socks5": 

205 socks_version = socks.PROXY_TYPE_SOCKS5 

206 rdns = False 

207 elif parsed.scheme == "socks5h": 

208 socks_version = socks.PROXY_TYPE_SOCKS5 

209 rdns = True 

210 elif parsed.scheme == "socks4": 

211 socks_version = socks.PROXY_TYPE_SOCKS4 

212 rdns = False 

213 elif parsed.scheme == "socks4a": 

214 socks_version = socks.PROXY_TYPE_SOCKS4 

215 rdns = True 

216 else: 

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

218 

219 self.proxy_url = proxy_url 

220 

221 socks_options = { 

222 "socks_version": socks_version, 

223 "proxy_host": parsed.host, 

224 "proxy_port": parsed.port, 

225 "username": username, 

226 "password": password, 

227 "rdns": rdns, 

228 } 

229 connection_pool_kw["_socks_options"] = socks_options 

230 

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

232 

233 self.pool_classes_by_scheme = SOCKSProxyManager.pool_classes_by_scheme