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

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-not-found] 

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/advanced-usage.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 

74 

75class _TYPE_SOCKS_OPTIONS(typing.TypedDict): 

76 socks_version: int 

77 proxy_host: str | None 

78 proxy_port: str | None 

79 username: str | None 

80 password: str | None 

81 rdns: bool 

82 

83 

84class SOCKSConnection(HTTPConnection): 

85 """ 

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

87 """ 

88 

89 def __init__( 

90 self, 

91 _socks_options: _TYPE_SOCKS_OPTIONS, 

92 *args: typing.Any, 

93 **kwargs: typing.Any, 

94 ) -> None: 

95 self._socks_options = _socks_options 

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

97 

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

99 """ 

100 Establish a new connection via the SOCKS proxy. 

101 """ 

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

103 if self.source_address: 

104 extra_kw["source_address"] = self.source_address 

105 

106 if self.socket_options: 

107 extra_kw["socket_options"] = self.socket_options 

108 

109 try: 

110 conn = socks.create_connection( 

111 (self.host, self.port), 

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

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

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

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

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

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

118 timeout=self.timeout, 

119 **extra_kw, 

120 ) 

121 

122 except SocketTimeout as e: 

123 raise ConnectTimeoutError( 

124 self, 

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

126 ) from e 

127 

128 except socks.ProxyError as e: 

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

130 # useful errors here. 

131 if e.socket_err: 

132 error = e.socket_err 

133 if isinstance(error, SocketTimeout): 

134 raise ConnectTimeoutError( 

135 self, 

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

137 ) from e 

138 else: 

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

140 # See #2386. 

141 raise NewConnectionError( 

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

143 ) 

144 else: 

145 raise NewConnectionError( 

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

147 ) from e 

148 

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

150 raise NewConnectionError( 

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

152 ) from e 

153 

154 return conn 

155 

156 

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

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

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

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

161class SOCKSHTTPSConnection(SOCKSConnection, HTTPSConnection): 

162 pass 

163 

164 

165class SOCKSHTTPConnectionPool(HTTPConnectionPool): 

166 ConnectionCls = SOCKSConnection 

167 

168 

169class SOCKSHTTPSConnectionPool(HTTPSConnectionPool): 

170 ConnectionCls = SOCKSHTTPSConnection 

171 

172 

173class SOCKSProxyManager(PoolManager): 

174 """ 

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

176 defined SOCKS proxy. 

177 """ 

178 

179 pool_classes_by_scheme = { 

180 "http": SOCKSHTTPConnectionPool, 

181 "https": SOCKSHTTPSConnectionPool, 

182 } 

183 

184 def __init__( 

185 self, 

186 proxy_url: str, 

187 username: str | None = None, 

188 password: str | None = None, 

189 num_pools: int = 10, 

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

191 **connection_pool_kw: typing.Any, 

192 ): 

193 parsed = parse_url(proxy_url) 

194 

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

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

197 if len(split) == 2: 

198 username, password = split 

199 if parsed.scheme == "socks5": 

200 socks_version = socks.PROXY_TYPE_SOCKS5 

201 rdns = False 

202 elif parsed.scheme == "socks5h": 

203 socks_version = socks.PROXY_TYPE_SOCKS5 

204 rdns = True 

205 elif parsed.scheme == "socks4": 

206 socks_version = socks.PROXY_TYPE_SOCKS4 

207 rdns = False 

208 elif parsed.scheme == "socks4a": 

209 socks_version = socks.PROXY_TYPE_SOCKS4 

210 rdns = True 

211 else: 

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

213 

214 self.proxy_url = proxy_url 

215 

216 socks_options = { 

217 "socks_version": socks_version, 

218 "proxy_host": parsed.host, 

219 "proxy_port": parsed.port, 

220 "username": username, 

221 "password": password, 

222 "rdns": rdns, 

223 } 

224 connection_pool_kw["_socks_options"] = socks_options 

225 

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

227 

228 self.pool_classes_by_scheme = SOCKSProxyManager.pool_classes_by_scheme