Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/urllib3/connection.py: 30%
320 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:35 +0000
1from __future__ import annotations
3import datetime
4import logging
5import os
6import re
7import socket
8import typing
9import warnings
10from http.client import HTTPConnection as _HTTPConnection
11from http.client import HTTPException as HTTPException # noqa: F401
12from http.client import ResponseNotReady
13from socket import timeout as SocketTimeout
15if typing.TYPE_CHECKING:
16 from typing_extensions import Literal
18 from .response import HTTPResponse
19 from .util.ssl_ import _TYPE_PEER_CERT_RET_DICT
20 from .util.ssltransport import SSLTransport
22from ._collections import HTTPHeaderDict
23from .util.response import assert_header_parsing
24from .util.timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT, Timeout
25from .util.util import to_str
26from .util.wait import wait_for_read
28try: # Compiled with SSL?
29 import ssl
31 BaseSSLError = ssl.SSLError
32except (ImportError, AttributeError):
33 ssl = None # type: ignore[assignment]
35 class BaseSSLError(BaseException): # type: ignore[no-redef]
36 pass
39from ._base_connection import _TYPE_BODY
40from ._base_connection import ProxyConfig as ProxyConfig
41from ._base_connection import _ResponseOptions as _ResponseOptions
42from ._version import __version__
43from .exceptions import (
44 ConnectTimeoutError,
45 HeaderParsingError,
46 NameResolutionError,
47 NewConnectionError,
48 ProxyError,
49 SystemTimeWarning,
50)
51from .util import SKIP_HEADER, SKIPPABLE_HEADERS, connection, ssl_
52from .util.request import body_to_chunks
53from .util.ssl_ import assert_fingerprint as _assert_fingerprint
54from .util.ssl_ import (
55 create_urllib3_context,
56 is_ipaddress,
57 resolve_cert_reqs,
58 resolve_ssl_version,
59 ssl_wrap_socket,
60)
61from .util.ssl_match_hostname import CertificateError, match_hostname
62from .util.url import Url
64# Not a no-op, we're adding this to the namespace so it can be imported.
65ConnectionError = ConnectionError
66BrokenPipeError = BrokenPipeError
69log = logging.getLogger(__name__)
71port_by_scheme = {"http": 80, "https": 443}
73# When it comes time to update this value as a part of regular maintenance
74# (ie test_recent_date is failing) update it to ~6 months before the current date.
75RECENT_DATE = datetime.date(2022, 1, 1)
77_CONTAINS_CONTROL_CHAR_RE = re.compile(r"[^-!#$%&'*+.^_`|~0-9a-zA-Z]")
80class HTTPConnection(_HTTPConnection):
81 """
82 Based on :class:`http.client.HTTPConnection` but provides an extra constructor
83 backwards-compatibility layer between older and newer Pythons.
85 Additional keyword parameters are used to configure attributes of the connection.
86 Accepted parameters include:
88 - ``source_address``: Set the source address for the current connection.
89 - ``socket_options``: Set specific options on the underlying socket. If not specified, then
90 defaults are loaded from ``HTTPConnection.default_socket_options`` which includes disabling
91 Nagle's algorithm (sets TCP_NODELAY to 1) unless the connection is behind a proxy.
93 For example, if you wish to enable TCP Keep Alive in addition to the defaults,
94 you might pass:
96 .. code-block:: python
98 HTTPConnection.default_socket_options + [
99 (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
100 ]
102 Or you may want to disable the defaults by passing an empty list (e.g., ``[]``).
103 """
105 default_port: typing.ClassVar[int] = port_by_scheme["http"] # type: ignore[misc]
107 #: Disable Nagle's algorithm by default.
108 #: ``[(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]``
109 default_socket_options: typing.ClassVar[connection._TYPE_SOCKET_OPTIONS] = [
110 (socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
111 ]
113 #: Whether this connection verifies the host's certificate.
114 is_verified: bool = False
116 #: Whether this proxy connection verified the proxy host's certificate.
117 # If no proxy is currently connected to the value will be ``None``.
118 proxy_is_verified: bool | None = None
120 blocksize: int
121 source_address: tuple[str, int] | None
122 socket_options: connection._TYPE_SOCKET_OPTIONS | None
124 _has_connected_to_proxy: bool
125 _response_options: _ResponseOptions | None
126 _tunnel_host: str | None
127 _tunnel_port: int | None
128 _tunnel_scheme: str | None
130 def __init__(
131 self,
132 host: str,
133 port: int | None = None,
134 *,
135 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
136 source_address: tuple[str, int] | None = None,
137 blocksize: int = 8192,
138 socket_options: None
139 | (connection._TYPE_SOCKET_OPTIONS) = default_socket_options,
140 proxy: Url | None = None,
141 proxy_config: ProxyConfig | None = None,
142 ) -> None:
143 super().__init__(
144 host=host,
145 port=port,
146 timeout=Timeout.resolve_default_timeout(timeout),
147 source_address=source_address,
148 blocksize=blocksize,
149 )
150 self.socket_options = socket_options
151 self.proxy = proxy
152 self.proxy_config = proxy_config
154 self._has_connected_to_proxy = False
155 self._response_options = None
156 self._tunnel_host: str | None = None
157 self._tunnel_port: int | None = None
158 self._tunnel_scheme: str | None = None
160 # https://github.com/python/mypy/issues/4125
161 # Mypy treats this as LSP violation, which is considered a bug.
162 # If `host` is made a property it violates LSP, because a writeable attribute is overridden with a read-only one.
163 # However, there is also a `host` setter so LSP is not violated.
164 # Potentially, a `@host.deleter` might be needed depending on how this issue will be fixed.
165 @property
166 def host(self) -> str:
167 """
168 Getter method to remove any trailing dots that indicate the hostname is an FQDN.
170 In general, SSL certificates don't include the trailing dot indicating a
171 fully-qualified domain name, and thus, they don't validate properly when
172 checked against a domain name that includes the dot. In addition, some
173 servers may not expect to receive the trailing dot when provided.
175 However, the hostname with trailing dot is critical to DNS resolution; doing a
176 lookup with the trailing dot will properly only resolve the appropriate FQDN,
177 whereas a lookup without a trailing dot will search the system's search domain
178 list. Thus, it's important to keep the original host around for use only in
179 those cases where it's appropriate (i.e., when doing DNS lookup to establish the
180 actual TCP connection across which we're going to send HTTP requests).
181 """
182 return self._dns_host.rstrip(".")
184 @host.setter
185 def host(self, value: str) -> None:
186 """
187 Setter for the `host` property.
189 We assume that only urllib3 uses the _dns_host attribute; httplib itself
190 only uses `host`, and it seems reasonable that other libraries follow suit.
191 """
192 self._dns_host = value
194 def _new_conn(self) -> socket.socket:
195 """Establish a socket connection and set nodelay settings on it.
197 :return: New socket connection.
198 """
199 try:
200 sock = connection.create_connection(
201 (self._dns_host, self.port),
202 self.timeout,
203 source_address=self.source_address,
204 socket_options=self.socket_options,
205 )
206 except socket.gaierror as e:
207 raise NameResolutionError(self.host, self, e) from e
208 except SocketTimeout as e:
209 raise ConnectTimeoutError(
210 self,
211 f"Connection to {self.host} timed out. (connect timeout={self.timeout})",
212 ) from e
214 except OSError as e:
215 raise NewConnectionError(
216 self, f"Failed to establish a new connection: {e}"
217 ) from e
219 return sock
221 def set_tunnel(
222 self,
223 host: str,
224 port: int | None = None,
225 headers: typing.Mapping[str, str] | None = None,
226 scheme: str = "http",
227 ) -> None:
228 if scheme not in ("http", "https"):
229 raise ValueError(
230 f"Invalid proxy scheme for tunneling: {scheme!r}, must be either 'http' or 'https'"
231 )
232 super().set_tunnel(host, port=port, headers=headers)
233 self._tunnel_scheme = scheme
235 def connect(self) -> None:
236 self.sock = self._new_conn()
237 if self._tunnel_host:
238 # If we're tunneling it means we're connected to our proxy.
239 self._has_connected_to_proxy = True
241 # TODO: Fix tunnel so it doesn't depend on self.sock state.
242 self._tunnel() # type: ignore[attr-defined]
244 # If there's a proxy to be connected to we are fully connected.
245 # This is set twice (once above and here) due to forwarding proxies
246 # not using tunnelling.
247 self._has_connected_to_proxy = bool(self.proxy)
249 @property
250 def is_closed(self) -> bool:
251 return self.sock is None
253 @property
254 def is_connected(self) -> bool:
255 if self.sock is None:
256 return False
257 return not wait_for_read(self.sock, timeout=0.0)
259 @property
260 def has_connected_to_proxy(self) -> bool:
261 return self._has_connected_to_proxy
263 def close(self) -> None:
264 try:
265 super().close()
266 finally:
267 # Reset all stateful properties so connection
268 # can be re-used without leaking prior configs.
269 self.sock = None
270 self.is_verified = False
271 self.proxy_is_verified = None
272 self._has_connected_to_proxy = False
273 self._response_options = None
274 self._tunnel_host = None
275 self._tunnel_port = None
276 self._tunnel_scheme = None
278 def putrequest(
279 self,
280 method: str,
281 url: str,
282 skip_host: bool = False,
283 skip_accept_encoding: bool = False,
284 ) -> None:
285 """"""
286 # Empty docstring because the indentation of CPython's implementation
287 # is broken but we don't want this method in our documentation.
288 match = _CONTAINS_CONTROL_CHAR_RE.search(method)
289 if match:
290 raise ValueError(
291 f"Method cannot contain non-token characters {method!r} (found at least {match.group()!r})"
292 )
294 return super().putrequest(
295 method, url, skip_host=skip_host, skip_accept_encoding=skip_accept_encoding
296 )
298 def putheader(self, header: str, *values: str) -> None:
299 """"""
300 if not any(isinstance(v, str) and v == SKIP_HEADER for v in values):
301 super().putheader(header, *values)
302 elif to_str(header.lower()) not in SKIPPABLE_HEADERS:
303 skippable_headers = "', '".join(
304 [str.title(header) for header in sorted(SKIPPABLE_HEADERS)]
305 )
306 raise ValueError(
307 f"urllib3.util.SKIP_HEADER only supports '{skippable_headers}'"
308 )
310 # `request` method's signature intentionally violates LSP.
311 # urllib3's API is different from `http.client.HTTPConnection` and the subclassing is only incidental.
312 def request( # type: ignore[override]
313 self,
314 method: str,
315 url: str,
316 body: _TYPE_BODY | None = None,
317 headers: typing.Mapping[str, str] | None = None,
318 *,
319 chunked: bool = False,
320 preload_content: bool = True,
321 decode_content: bool = True,
322 enforce_content_length: bool = True,
323 ) -> None:
324 # Update the inner socket's timeout value to send the request.
325 # This only triggers if the connection is re-used.
326 if self.sock is not None:
327 self.sock.settimeout(self.timeout)
329 # Store these values to be fed into the HTTPResponse
330 # object later. TODO: Remove this in favor of a real
331 # HTTP lifecycle mechanism.
333 # We have to store these before we call .request()
334 # because sometimes we can still salvage a response
335 # off the wire even if we aren't able to completely
336 # send the request body.
337 self._response_options = _ResponseOptions(
338 request_method=method,
339 request_url=url,
340 preload_content=preload_content,
341 decode_content=decode_content,
342 enforce_content_length=enforce_content_length,
343 )
345 if headers is None:
346 headers = {}
347 header_keys = frozenset(to_str(k.lower()) for k in headers)
348 skip_accept_encoding = "accept-encoding" in header_keys
349 skip_host = "host" in header_keys
350 self.putrequest(
351 method, url, skip_accept_encoding=skip_accept_encoding, skip_host=skip_host
352 )
354 # Transform the body into an iterable of sendall()-able chunks
355 # and detect if an explicit Content-Length is doable.
356 chunks_and_cl = body_to_chunks(body, method=method, blocksize=self.blocksize)
357 chunks = chunks_and_cl.chunks
358 content_length = chunks_and_cl.content_length
360 # When chunked is explicit set to 'True' we respect that.
361 if chunked:
362 if "transfer-encoding" not in header_keys:
363 self.putheader("Transfer-Encoding", "chunked")
364 else:
365 # Detect whether a framing mechanism is already in use. If so
366 # we respect that value, otherwise we pick chunked vs content-length
367 # depending on the type of 'body'.
368 if "content-length" in header_keys:
369 chunked = False
370 elif "transfer-encoding" in header_keys:
371 chunked = True
373 # Otherwise we go off the recommendation of 'body_to_chunks()'.
374 else:
375 chunked = False
376 if content_length is None:
377 if chunks is not None:
378 chunked = True
379 self.putheader("Transfer-Encoding", "chunked")
380 else:
381 self.putheader("Content-Length", str(content_length))
383 # Now that framing headers are out of the way we send all the other headers.
384 if "user-agent" not in header_keys:
385 self.putheader("User-Agent", _get_default_user_agent())
386 for header, value in headers.items():
387 self.putheader(header, value)
388 self.endheaders()
390 # If we're given a body we start sending that in chunks.
391 if chunks is not None:
392 for chunk in chunks:
393 # Sending empty chunks isn't allowed for TE: chunked
394 # as it indicates the end of the body.
395 if not chunk:
396 continue
397 if isinstance(chunk, str):
398 chunk = chunk.encode("utf-8")
399 if chunked:
400 self.send(b"%x\r\n%b\r\n" % (len(chunk), chunk))
401 else:
402 self.send(chunk)
404 # Regardless of whether we have a body or not, if we're in
405 # chunked mode we want to send an explicit empty chunk.
406 if chunked:
407 self.send(b"0\r\n\r\n")
409 def request_chunked(
410 self,
411 method: str,
412 url: str,
413 body: _TYPE_BODY | None = None,
414 headers: typing.Mapping[str, str] | None = None,
415 ) -> None:
416 """
417 Alternative to the common request method, which sends the
418 body with chunked encoding and not as one block
419 """
420 warnings.warn(
421 "HTTPConnection.request_chunked() is deprecated and will be removed "
422 "in urllib3 v2.1.0. Instead use HTTPConnection.request(..., chunked=True).",
423 category=DeprecationWarning,
424 stacklevel=2,
425 )
426 self.request(method, url, body=body, headers=headers, chunked=True)
428 def getresponse( # type: ignore[override]
429 self,
430 ) -> HTTPResponse:
431 """
432 Get the response from the server.
434 If the HTTPConnection is in the correct state, returns an instance of HTTPResponse or of whatever object is returned by the response_class variable.
436 If a request has not been sent or if a previous response has not be handled, ResponseNotReady is raised. If the HTTP response indicates that the connection should be closed, then it will be closed before the response is returned. When the connection is closed, the underlying socket is closed.
437 """
438 # Raise the same error as http.client.HTTPConnection
439 if self._response_options is None:
440 raise ResponseNotReady()
442 # Reset this attribute for being used again.
443 resp_options = self._response_options
444 self._response_options = None
446 # Since the connection's timeout value may have been updated
447 # we need to set the timeout on the socket.
448 self.sock.settimeout(self.timeout)
450 # This is needed here to avoid circular import errors
451 from .response import HTTPResponse
453 # Get the response from http.client.HTTPConnection
454 httplib_response = super().getresponse()
456 try:
457 assert_header_parsing(httplib_response.msg)
458 except (HeaderParsingError, TypeError) as hpe:
459 log.warning(
460 "Failed to parse headers (url=%s): %s",
461 _url_from_connection(self, resp_options.request_url),
462 hpe,
463 exc_info=True,
464 )
466 headers = HTTPHeaderDict(httplib_response.msg.items())
468 response = HTTPResponse(
469 body=httplib_response,
470 headers=headers,
471 status=httplib_response.status,
472 version=httplib_response.version,
473 reason=httplib_response.reason,
474 preload_content=resp_options.preload_content,
475 decode_content=resp_options.decode_content,
476 original_response=httplib_response,
477 enforce_content_length=resp_options.enforce_content_length,
478 request_method=resp_options.request_method,
479 request_url=resp_options.request_url,
480 )
481 return response
484class HTTPSConnection(HTTPConnection):
485 """
486 Many of the parameters to this constructor are passed to the underlying SSL
487 socket by means of :py:func:`urllib3.util.ssl_wrap_socket`.
488 """
490 default_port = port_by_scheme["https"] # type: ignore[misc]
492 cert_reqs: int | str | None = None
493 ca_certs: str | None = None
494 ca_cert_dir: str | None = None
495 ca_cert_data: None | str | bytes = None
496 ssl_version: int | str | None = None
497 ssl_minimum_version: int | None = None
498 ssl_maximum_version: int | None = None
499 assert_fingerprint: str | None = None
501 def __init__(
502 self,
503 host: str,
504 port: int | None = None,
505 *,
506 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
507 source_address: tuple[str, int] | None = None,
508 blocksize: int = 8192,
509 socket_options: None
510 | (connection._TYPE_SOCKET_OPTIONS) = HTTPConnection.default_socket_options,
511 proxy: Url | None = None,
512 proxy_config: ProxyConfig | None = None,
513 cert_reqs: int | str | None = None,
514 assert_hostname: None | str | Literal[False] = None,
515 assert_fingerprint: str | None = None,
516 server_hostname: str | None = None,
517 ssl_context: ssl.SSLContext | None = None,
518 ca_certs: str | None = None,
519 ca_cert_dir: str | None = None,
520 ca_cert_data: None | str | bytes = None,
521 ssl_minimum_version: int | None = None,
522 ssl_maximum_version: int | None = None,
523 ssl_version: int | str | None = None, # Deprecated
524 cert_file: str | None = None,
525 key_file: str | None = None,
526 key_password: str | None = None,
527 ) -> None:
528 super().__init__(
529 host,
530 port=port,
531 timeout=timeout,
532 source_address=source_address,
533 blocksize=blocksize,
534 socket_options=socket_options,
535 proxy=proxy,
536 proxy_config=proxy_config,
537 )
539 self.key_file = key_file
540 self.cert_file = cert_file
541 self.key_password = key_password
542 self.ssl_context = ssl_context
543 self.server_hostname = server_hostname
544 self.assert_hostname = assert_hostname
545 self.assert_fingerprint = assert_fingerprint
546 self.ssl_version = ssl_version
547 self.ssl_minimum_version = ssl_minimum_version
548 self.ssl_maximum_version = ssl_maximum_version
549 self.ca_certs = ca_certs and os.path.expanduser(ca_certs)
550 self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir)
551 self.ca_cert_data = ca_cert_data
553 # cert_reqs depends on ssl_context so calculate last.
554 if cert_reqs is None:
555 if self.ssl_context is not None:
556 cert_reqs = self.ssl_context.verify_mode
557 else:
558 cert_reqs = resolve_cert_reqs(None)
559 self.cert_reqs = cert_reqs
561 def set_cert(
562 self,
563 key_file: str | None = None,
564 cert_file: str | None = None,
565 cert_reqs: int | str | None = None,
566 key_password: str | None = None,
567 ca_certs: str | None = None,
568 assert_hostname: None | str | Literal[False] = None,
569 assert_fingerprint: str | None = None,
570 ca_cert_dir: str | None = None,
571 ca_cert_data: None | str | bytes = None,
572 ) -> None:
573 """
574 This method should only be called once, before the connection is used.
575 """
576 warnings.warn(
577 "HTTPSConnection.set_cert() is deprecated and will be removed "
578 "in urllib3 v2.1.0. Instead provide the parameters to the "
579 "HTTPSConnection constructor.",
580 category=DeprecationWarning,
581 stacklevel=2,
582 )
584 # If cert_reqs is not provided we'll assume CERT_REQUIRED unless we also
585 # have an SSLContext object in which case we'll use its verify_mode.
586 if cert_reqs is None:
587 if self.ssl_context is not None:
588 cert_reqs = self.ssl_context.verify_mode
589 else:
590 cert_reqs = resolve_cert_reqs(None)
592 self.key_file = key_file
593 self.cert_file = cert_file
594 self.cert_reqs = cert_reqs
595 self.key_password = key_password
596 self.assert_hostname = assert_hostname
597 self.assert_fingerprint = assert_fingerprint
598 self.ca_certs = ca_certs and os.path.expanduser(ca_certs)
599 self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir)
600 self.ca_cert_data = ca_cert_data
602 def connect(self) -> None:
603 sock: socket.socket | ssl.SSLSocket
604 self.sock = sock = self._new_conn()
605 server_hostname: str = self.host
606 tls_in_tls = False
608 # Do we need to establish a tunnel?
609 if self._tunnel_host is not None:
610 # We're tunneling to an HTTPS origin so need to do TLS-in-TLS.
611 if self._tunnel_scheme == "https":
612 self.sock = sock = self._connect_tls_proxy(self.host, sock)
613 tls_in_tls = True
615 # If we're tunneling it means we're connected to our proxy.
616 self._has_connected_to_proxy = True
618 self._tunnel() # type: ignore[attr-defined]
619 # Override the host with the one we're requesting data from.
620 server_hostname = self._tunnel_host
622 if self.server_hostname is not None:
623 server_hostname = self.server_hostname
625 is_time_off = datetime.date.today() < RECENT_DATE
626 if is_time_off:
627 warnings.warn(
628 (
629 f"System time is way off (before {RECENT_DATE}). This will probably "
630 "lead to SSL verification errors"
631 ),
632 SystemTimeWarning,
633 )
635 sock_and_verified = _ssl_wrap_socket_and_match_hostname(
636 sock=sock,
637 cert_reqs=self.cert_reqs,
638 ssl_version=self.ssl_version,
639 ssl_minimum_version=self.ssl_minimum_version,
640 ssl_maximum_version=self.ssl_maximum_version,
641 ca_certs=self.ca_certs,
642 ca_cert_dir=self.ca_cert_dir,
643 ca_cert_data=self.ca_cert_data,
644 cert_file=self.cert_file,
645 key_file=self.key_file,
646 key_password=self.key_password,
647 server_hostname=server_hostname,
648 ssl_context=self.ssl_context,
649 tls_in_tls=tls_in_tls,
650 assert_hostname=self.assert_hostname,
651 assert_fingerprint=self.assert_fingerprint,
652 )
653 self.sock = sock_and_verified.socket
654 self.is_verified = sock_and_verified.is_verified
656 # If there's a proxy to be connected to we are fully connected.
657 # This is set twice (once above and here) due to forwarding proxies
658 # not using tunnelling.
659 self._has_connected_to_proxy = bool(self.proxy)
661 def _connect_tls_proxy(self, hostname: str, sock: socket.socket) -> ssl.SSLSocket:
662 """
663 Establish a TLS connection to the proxy using the provided SSL context.
664 """
665 # `_connect_tls_proxy` is called when self._tunnel_host is truthy.
666 proxy_config = typing.cast(ProxyConfig, self.proxy_config)
667 ssl_context = proxy_config.ssl_context
668 sock_and_verified = _ssl_wrap_socket_and_match_hostname(
669 sock,
670 cert_reqs=self.cert_reqs,
671 ssl_version=self.ssl_version,
672 ssl_minimum_version=self.ssl_minimum_version,
673 ssl_maximum_version=self.ssl_maximum_version,
674 ca_certs=self.ca_certs,
675 ca_cert_dir=self.ca_cert_dir,
676 ca_cert_data=self.ca_cert_data,
677 server_hostname=hostname,
678 ssl_context=ssl_context,
679 assert_hostname=proxy_config.assert_hostname,
680 assert_fingerprint=proxy_config.assert_fingerprint,
681 # Features that aren't implemented for proxies yet:
682 cert_file=None,
683 key_file=None,
684 key_password=None,
685 tls_in_tls=False,
686 )
687 self.proxy_is_verified = sock_and_verified.is_verified
688 return sock_and_verified.socket # type: ignore[return-value]
691class _WrappedAndVerifiedSocket(typing.NamedTuple):
692 """
693 Wrapped socket and whether the connection is
694 verified after the TLS handshake
695 """
697 socket: ssl.SSLSocket | SSLTransport
698 is_verified: bool
701def _ssl_wrap_socket_and_match_hostname(
702 sock: socket.socket,
703 *,
704 cert_reqs: None | str | int,
705 ssl_version: None | str | int,
706 ssl_minimum_version: int | None,
707 ssl_maximum_version: int | None,
708 cert_file: str | None,
709 key_file: str | None,
710 key_password: str | None,
711 ca_certs: str | None,
712 ca_cert_dir: str | None,
713 ca_cert_data: None | str | bytes,
714 assert_hostname: None | str | Literal[False],
715 assert_fingerprint: str | None,
716 server_hostname: str | None,
717 ssl_context: ssl.SSLContext | None,
718 tls_in_tls: bool = False,
719) -> _WrappedAndVerifiedSocket:
720 """Logic for constructing an SSLContext from all TLS parameters, passing
721 that down into ssl_wrap_socket, and then doing certificate verification
722 either via hostname or fingerprint. This function exists to guarantee
723 that both proxies and targets have the same behavior when connecting via TLS.
724 """
725 default_ssl_context = False
726 if ssl_context is None:
727 default_ssl_context = True
728 context = create_urllib3_context(
729 ssl_version=resolve_ssl_version(ssl_version),
730 ssl_minimum_version=ssl_minimum_version,
731 ssl_maximum_version=ssl_maximum_version,
732 cert_reqs=resolve_cert_reqs(cert_reqs),
733 )
734 else:
735 context = ssl_context
737 context.verify_mode = resolve_cert_reqs(cert_reqs)
739 # In some cases, we want to verify hostnames ourselves
740 if (
741 # `ssl` can't verify fingerprints or alternate hostnames
742 assert_fingerprint
743 or assert_hostname
744 # We still support OpenSSL 1.0.2, which prevents us from verifying
745 # hostnames easily: https://github.com/pyca/pyopenssl/pull/933
746 or ssl_.IS_PYOPENSSL
747 or not ssl_.HAS_NEVER_CHECK_COMMON_NAME
748 ):
749 context.check_hostname = False
751 # Try to load OS default certs if none are given.
752 # We need to do the hasattr() check for our custom
753 # pyOpenSSL and SecureTransport SSLContext objects
754 # because neither support load_default_certs().
755 if (
756 not ca_certs
757 and not ca_cert_dir
758 and not ca_cert_data
759 and default_ssl_context
760 and hasattr(context, "load_default_certs")
761 ):
762 context.load_default_certs()
764 # Ensure that IPv6 addresses are in the proper format and don't have a
765 # scope ID. Python's SSL module fails to recognize scoped IPv6 addresses
766 # and interprets them as DNS hostnames.
767 if server_hostname is not None:
768 normalized = server_hostname.strip("[]")
769 if "%" in normalized:
770 normalized = normalized[: normalized.rfind("%")]
771 if is_ipaddress(normalized):
772 server_hostname = normalized
774 ssl_sock = ssl_wrap_socket(
775 sock=sock,
776 keyfile=key_file,
777 certfile=cert_file,
778 key_password=key_password,
779 ca_certs=ca_certs,
780 ca_cert_dir=ca_cert_dir,
781 ca_cert_data=ca_cert_data,
782 server_hostname=server_hostname,
783 ssl_context=context,
784 tls_in_tls=tls_in_tls,
785 )
787 try:
788 if assert_fingerprint:
789 _assert_fingerprint(
790 ssl_sock.getpeercert(binary_form=True), assert_fingerprint
791 )
792 elif (
793 context.verify_mode != ssl.CERT_NONE
794 and not context.check_hostname
795 and assert_hostname is not False
796 ):
797 cert: _TYPE_PEER_CERT_RET_DICT = ssl_sock.getpeercert() # type: ignore[assignment]
799 # Need to signal to our match_hostname whether to use 'commonName' or not.
800 # If we're using our own constructed SSLContext we explicitly set 'False'
801 # because PyPy hard-codes 'True' from SSLContext.hostname_checks_common_name.
802 if default_ssl_context:
803 hostname_checks_common_name = False
804 else:
805 hostname_checks_common_name = (
806 getattr(context, "hostname_checks_common_name", False) or False
807 )
809 _match_hostname(
810 cert,
811 assert_hostname or server_hostname, # type: ignore[arg-type]
812 hostname_checks_common_name,
813 )
815 return _WrappedAndVerifiedSocket(
816 socket=ssl_sock,
817 is_verified=context.verify_mode == ssl.CERT_REQUIRED
818 or bool(assert_fingerprint),
819 )
820 except BaseException:
821 ssl_sock.close()
822 raise
825def _match_hostname(
826 cert: _TYPE_PEER_CERT_RET_DICT | None,
827 asserted_hostname: str,
828 hostname_checks_common_name: bool = False,
829) -> None:
830 # Our upstream implementation of ssl.match_hostname()
831 # only applies this normalization to IP addresses so it doesn't
832 # match DNS SANs so we do the same thing!
833 stripped_hostname = asserted_hostname.strip("[]")
834 if is_ipaddress(stripped_hostname):
835 asserted_hostname = stripped_hostname
837 try:
838 match_hostname(cert, asserted_hostname, hostname_checks_common_name)
839 except CertificateError as e:
840 log.warning(
841 "Certificate did not match expected hostname: %s. Certificate: %s",
842 asserted_hostname,
843 cert,
844 )
845 # Add cert to exception and reraise so client code can inspect
846 # the cert when catching the exception, if they want to
847 e._peer_cert = cert # type: ignore[attr-defined]
848 raise
851def _wrap_proxy_error(err: Exception, proxy_scheme: str | None) -> ProxyError:
852 # Look for the phrase 'wrong version number', if found
853 # then we should warn the user that we're very sure that
854 # this proxy is HTTP-only and they have a configuration issue.
855 error_normalized = " ".join(re.split("[^a-z]", str(err).lower()))
856 is_likely_http_proxy = (
857 "wrong version number" in error_normalized
858 or "unknown protocol" in error_normalized
859 )
860 http_proxy_warning = (
861 ". Your proxy appears to only use HTTP and not HTTPS, "
862 "try changing your proxy URL to be HTTP. See: "
863 "https://urllib3.readthedocs.io/en/latest/advanced-usage.html"
864 "#https-proxy-error-http-proxy"
865 )
866 new_err = ProxyError(
867 f"Unable to connect to proxy"
868 f"{http_proxy_warning if is_likely_http_proxy and proxy_scheme == 'https' else ''}",
869 err,
870 )
871 new_err.__cause__ = err
872 return new_err
875def _get_default_user_agent() -> str:
876 return f"python-urllib3/{__version__}"
879class DummyConnection:
880 """Used to detect a failed ConnectionCls import."""
883if not ssl:
884 HTTPSConnection = DummyConnection # type: ignore[misc, assignment] # noqa: F811
887VerifiedHTTPSConnection = HTTPSConnection
890def _url_from_connection(
891 conn: HTTPConnection | HTTPSConnection, path: str | None = None
892) -> str:
893 """Returns the URL from a given connection. This is mainly used for testing and logging."""
895 scheme = "https" if isinstance(conn, HTTPSConnection) else "http"
897 return Url(scheme=scheme, host=conn.host, port=conn.port, path=path).url