Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/urllib3/contrib/socks.py: 10%
83 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-08 06:45 +0000
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-08 06:45 +0000
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.
7The SOCKS implementation supports the full range of urllib3 features. It also
8supports the following SOCKS features:
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
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.
21SOCKS4 supports IPv4 and domain names with the SOCKS4A extension. SOCKS5
22supports IPv4, IPv6, and domain names.
24When connecting to a SOCKS4 proxy the ``username`` portion of the ``proxy_url``
25will be sent as the ``userid`` section of the SOCKS request:
27.. code-block:: python
29 proxy_url="socks4a://<userid>@proxy-host"
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:
35.. code-block:: python
37 proxy_url="socks5h://<username>:<password>@proxy-host"
39"""
41from __future__ import annotations
43try:
44 import socks # type: ignore[import]
45except ImportError:
46 import warnings
48 from ..exceptions import DependencyWarning
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
60import typing
61from socket import timeout as SocketTimeout
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
69try:
70 import ssl
71except ImportError:
72 ssl = None # type: ignore[assignment]
74from typing import TypedDict
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
86class SOCKSConnection(HTTPConnection):
87 """
88 A plain-text HTTP connection that connects via a SOCKS proxy.
89 """
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)
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
108 if self.socket_options:
109 extra_kw["socket_options"] = self.socket_options
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 )
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
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
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
156 return conn
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
167class SOCKSHTTPConnectionPool(HTTPConnectionPool):
168 ConnectionCls = SOCKSConnection
171class SOCKSHTTPSConnectionPool(HTTPSConnectionPool):
172 ConnectionCls = SOCKSHTTPSConnection
175class SOCKSProxyManager(PoolManager):
176 """
177 A version of the urllib3 ProxyManager that routes connections via the
178 defined SOCKS proxy.
179 """
181 pool_classes_by_scheme = {
182 "http": SOCKSHTTPConnectionPool,
183 "https": SOCKSHTTPSConnectionPool,
184 }
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)
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}")
216 self.proxy_url = proxy_url
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
228 super().__init__(num_pools, headers, **connection_pool_kw)
230 self.pool_classes_by_scheme = SOCKSProxyManager.pool_classes_by_scheme