Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/urllib3/connection.py: 26%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import datetime
4import http.client
5import logging
6import os
7import re
8import socket
9import sys
10import threading
11import typing
12import warnings
13from http.client import HTTPConnection as _HTTPConnection
14from http.client import HTTPException as HTTPException # noqa: F401
15from http.client import ResponseNotReady
16from socket import timeout as SocketTimeout
18if typing.TYPE_CHECKING:
19 from .response import HTTPResponse
20 from .util.ssl_ import _TYPE_PEER_CERT_RET_DICT
21 from .util.ssltransport import SSLTransport
23from ._collections import HTTPHeaderDict
24from .http2 import probe as http2_probe
25from .util.response import assert_header_parsing
26from .util.timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT, Timeout
27from .util.util import to_str
28from .util.wait import wait_for_read
30try: # Compiled with SSL?
31 import ssl
33 BaseSSLError = ssl.SSLError
34except (ImportError, AttributeError):
35 ssl = None # type: ignore[assignment]
37 class BaseSSLError(BaseException): # type: ignore[no-redef]
38 pass
41from ._base_connection import _TYPE_BODY
42from ._base_connection import ProxyConfig as ProxyConfig
43from ._base_connection import _ResponseOptions as _ResponseOptions
44from ._version import __version__
45from .exceptions import (
46 ConnectTimeoutError,
47 HeaderParsingError,
48 NameResolutionError,
49 NewConnectionError,
50 ProxyError,
51 SystemTimeWarning,
52)
53from .util import SKIP_HEADER, SKIPPABLE_HEADERS, connection, ssl_
54from .util.request import body_to_chunks
55from .util.ssl_ import assert_fingerprint as _assert_fingerprint
56from .util.ssl_ import (
57 create_urllib3_context,
58 is_ipaddress,
59 resolve_cert_reqs,
60 resolve_ssl_version,
61 ssl_wrap_socket,
62)
63from .util.ssl_match_hostname import CertificateError, match_hostname
64from .util.url import Url
66# Not a no-op, we're adding this to the namespace so it can be imported.
67ConnectionError = ConnectionError
68BrokenPipeError = BrokenPipeError
71log = logging.getLogger(__name__)
73port_by_scheme = {"http": 80, "https": 443}
75# When it comes time to update this value as a part of regular maintenance
76# (ie test_recent_date is failing) update it to ~6 months before the current date.
77RECENT_DATE = datetime.date(2025, 1, 1)
79_CONTAINS_CONTROL_CHAR_RE = re.compile(r"[^-!#$%&'*+.^_`|~0-9a-zA-Z]")
82class HTTPConnection(_HTTPConnection):
83 """
84 Based on :class:`http.client.HTTPConnection` but provides an extra constructor
85 backwards-compatibility layer between older and newer Pythons.
87 Additional keyword parameters are used to configure attributes of the connection.
88 Accepted parameters include:
90 - ``source_address``: Set the source address for the current connection.
91 - ``socket_options``: Set specific options on the underlying socket. If not specified, then
92 defaults are loaded from ``HTTPConnection.default_socket_options`` which includes disabling
93 Nagle's algorithm (sets TCP_NODELAY to 1) unless the connection is behind a proxy.
95 For example, if you wish to enable TCP Keep Alive in addition to the defaults,
96 you might pass:
98 .. code-block:: python
100 HTTPConnection.default_socket_options + [
101 (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
102 ]
104 Or you may want to disable the defaults by passing an empty list (e.g., ``[]``).
105 """
107 default_port: typing.ClassVar[int] = port_by_scheme["http"] # type: ignore[misc]
109 #: Disable Nagle's algorithm by default.
110 #: ``[(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]``
111 #:
112 #: Use the ``socket_options`` parameter of :class:`~urllib3.PoolManager`,
113 #: :class:`~urllib3.ProxyManager`, or :class:`~urllib3.HTTPConnectionPool`
114 #: to change this behavior.
115 default_socket_options: typing.ClassVar[
116 typing.Final[connection._TYPE_SOCKET_OPTIONS]
117 ] = [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]
119 #: Whether this connection verifies the host's certificate.
120 is_verified: bool = False
122 #: Whether this proxy connection verified the proxy host's certificate.
123 # If no proxy is currently connected to the value will be ``None``.
124 proxy_is_verified: bool | None = None
126 blocksize: int
127 source_address: tuple[str, int] | None
128 socket_options: connection._TYPE_SOCKET_OPTIONS | None
130 _has_connected_to_proxy: bool
131 _response_options: _ResponseOptions | None
132 _tunnel_host: str | None
133 _tunnel_port: int | None
134 _tunnel_scheme: str | None
136 def __init__(
137 self,
138 host: str,
139 port: int | None = None,
140 *,
141 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
142 source_address: tuple[str, int] | None = None,
143 blocksize: int = 16384,
144 socket_options: None | (
145 connection._TYPE_SOCKET_OPTIONS
146 ) = default_socket_options,
147 proxy: Url | None = None,
148 proxy_config: ProxyConfig | None = None,
149 ) -> None:
150 super().__init__(
151 host=host,
152 port=port,
153 timeout=Timeout.resolve_default_timeout(timeout),
154 source_address=source_address,
155 blocksize=blocksize,
156 )
157 self.socket_options = socket_options
158 self.proxy = proxy
159 self.proxy_config = proxy_config
161 self._has_connected_to_proxy = False
162 self._response_options = None
163 self._tunnel_host: str | None = None
164 self._tunnel_port: int | None = None
165 self._tunnel_scheme: str | None = None
167 def __str__(self) -> str:
168 return f"{type(self).__name__}(host={self.host!r}, port={self.port!r})"
170 def __repr__(self) -> str:
171 return f"<{self} at {id(self):#x}>"
173 @property
174 def host(self) -> str:
175 """
176 Getter method to remove any trailing dots that indicate the hostname is an FQDN.
178 In general, SSL certificates don't include the trailing dot indicating a
179 fully-qualified domain name, and thus, they don't validate properly when
180 checked against a domain name that includes the dot. In addition, some
181 servers may not expect to receive the trailing dot when provided.
183 However, the hostname with trailing dot is critical to DNS resolution; doing a
184 lookup with the trailing dot will properly only resolve the appropriate FQDN,
185 whereas a lookup without a trailing dot will search the system's search domain
186 list. Thus, it's important to keep the original host around for use only in
187 those cases where it's appropriate (i.e., when doing DNS lookup to establish the
188 actual TCP connection across which we're going to send HTTP requests).
189 """
190 return self._dns_host.rstrip(".")
192 @host.setter
193 def host(self, value: str) -> None:
194 """
195 Setter for the `host` property.
197 We assume that only urllib3 uses the _dns_host attribute; httplib itself
198 only uses `host`, and it seems reasonable that other libraries follow suit.
199 """
200 self._dns_host = value
202 def _new_conn(self) -> socket.socket:
203 """Establish a socket connection and set nodelay settings on it.
205 :return: New socket connection.
206 """
207 try:
208 sock = connection.create_connection(
209 (self._dns_host, self.port),
210 self.timeout,
211 source_address=self.source_address,
212 socket_options=self.socket_options,
213 )
214 except socket.gaierror as e:
215 raise NameResolutionError(self.host, self, e) from e
216 except SocketTimeout as e:
217 raise ConnectTimeoutError(
218 self,
219 f"Connection to {self.host} timed out. (connect timeout={self.timeout})",
220 ) from e
222 except OSError as e:
223 raise NewConnectionError(
224 self, f"Failed to establish a new connection: {e}"
225 ) from e
227 sys.audit("http.client.connect", self, self.host, self.port)
229 return sock
231 def set_tunnel(
232 self,
233 host: str,
234 port: int | None = None,
235 headers: typing.Mapping[str, str] | None = None,
236 scheme: str = "http",
237 ) -> None:
238 if scheme not in ("http", "https"):
239 raise ValueError(
240 f"Invalid proxy scheme for tunneling: {scheme!r}, must be either 'http' or 'https'"
241 )
242 super().set_tunnel(host, port=port, headers=headers)
243 self._tunnel_scheme = scheme
245 if sys.version_info < (3, 11, 16) or ((3, 12) <= sys.version_info < (3, 12, 14)):
246 # Taken from python/cpython#100986 which was backported in 3.11.9 and 3.12.3.
247 # When using connection_from_host, host will come without brackets.
248 # With a security patch from python/cpython#146211.
249 def _wrap_ipv6(self, ip: bytes) -> bytes:
250 if b":" in ip and ip[0] != b"["[0]:
251 return b"[" + ip + b"]"
252 return ip
254 # Copied from CPython 3.12.13 Lib/http/client.py
255 _is_legal_header_name = staticmethod(re.compile(rb"[^:\s][^:\r\n]*").fullmatch)
256 _is_illegal_header_value = staticmethod(
257 re.compile(rb"\n(?![ \t])|\r(?![ \t\n])").search
258 )
259 _contains_disallowed_url_pchar_re = re.compile("[\x00-\x20\x7f]")
261 if sys.version_info < (3, 11, 16):
262 # `_tunnel` copied from 3.11.15 backporting
263 # https://github.com/python/cpython/commit/0d4026432591d43185568dd31cef6a034c4b9261
264 # and https://github.com/python/cpython/commit/6fbc61070fda2ffb8889e77e3b24bca4249ab4d1
265 # plus a fix from https://github.com/python/cpython/commit/56b7100b04e44ea27989242b176beb8f016b2c53
266 def _tunnel(self) -> None:
267 if self._contains_disallowed_url_pchar_re.search(self._tunnel_host): # type: ignore[arg-type]
268 raise ValueError(
269 "Tunnel host can't contain control characters %r"
270 % (self._tunnel_host,)
271 )
272 _MAXLINE = http.client._MAXLINE # type: ignore[attr-defined]
273 connect = b"CONNECT %s:%d HTTP/1.0\r\n" % ( # type: ignore[str-format]
274 self._wrap_ipv6(self._tunnel_host.encode("ascii")), # type: ignore[union-attr]
275 self._tunnel_port,
276 )
277 headers = [connect]
278 for header, value in self._tunnel_headers.items(): # type: ignore[attr-defined]
279 header_bytes = header.encode("latin-1")
280 value_bytes = value.encode("latin-1")
281 if not self._is_legal_header_name(header_bytes):
282 raise ValueError(f"Invalid header name {header_bytes!r}")
283 if self._is_illegal_header_value(value_bytes):
284 raise ValueError(f"Invalid header value {value_bytes!r}")
285 headers.append(b"%s: %s\r\n" % (header_bytes, value_bytes))
286 headers.append(b"\r\n")
287 # Making a single send() call instead of one per line encourages
288 # the host OS to use a more optimal packet size instead of
289 # potentially emitting a series of small packets.
290 self.send(b"".join(headers))
291 del headers
293 response = self.response_class(self.sock, method=self._method) # type: ignore[attr-defined]
294 try:
295 (version, code, message) = response._read_status() # type: ignore[attr-defined]
297 if code != http.HTTPStatus.OK:
298 self.close()
299 raise OSError(
300 f"Tunnel connection failed: {code} {message.strip()}"
301 )
302 while True:
303 line = response.fp.readline(_MAXLINE + 1)
304 if len(line) > _MAXLINE:
305 raise http.client.LineTooLong("header line")
306 if not line:
307 # for sites which EOF without sending a trailer
308 break
309 if line in (b"\r\n", b"\n", b""):
310 break
312 if self.debuglevel > 0:
313 print("header:", line.decode())
314 finally:
315 response.close()
317 elif (3, 12) <= sys.version_info < (3, 12, 14):
318 # `_tunnel` copied from 3.12.13 backporting
319 # https://github.com/python/cpython/commit/23aef575c7629abcd4aaf028ebd226fb41a4b3c8
320 # plus a fix from https://github.com/python/cpython/commit/c00c386faa579ad71196d33408644478488e43ec
321 def _tunnel(self) -> None: # noqa: F811
322 if self._contains_disallowed_url_pchar_re.search(self._tunnel_host): # type: ignore[arg-type]
323 raise ValueError(
324 "Tunnel host can't contain control characters %r"
325 % (self._tunnel_host,)
326 )
327 connect = b"CONNECT %s:%d HTTP/1.1\r\n" % ( # type: ignore[str-format]
328 self._wrap_ipv6(self._tunnel_host.encode("idna")), # type: ignore[union-attr]
329 self._tunnel_port,
330 )
331 headers = [connect]
332 for header, value in self._tunnel_headers.items(): # type: ignore[attr-defined]
333 header_bytes = header.encode("latin-1")
334 value_bytes = value.encode("latin-1")
335 if not self._is_legal_header_name(header_bytes):
336 raise ValueError(f"Invalid header name {header_bytes!r}")
337 if self._is_illegal_header_value(value_bytes):
338 raise ValueError(f"Invalid header value {value_bytes!r}")
339 headers.append(b"%s: %s\r\n" % (header_bytes, value_bytes))
340 headers.append(b"\r\n")
341 # Making a single send() call instead of one per line encourages
342 # the host OS to use a more optimal packet size instead of
343 # potentially emitting a series of small packets.
344 self.send(b"".join(headers))
345 del headers
347 response = self.response_class(self.sock, method=self._method) # type: ignore[attr-defined]
348 try:
349 (version, code, message) = response._read_status() # type: ignore[attr-defined]
351 self._raw_proxy_headers = http.client._read_headers(response.fp) # type: ignore[attr-defined]
353 if self.debuglevel > 0:
354 for header in self._raw_proxy_headers:
355 print("header:", header.decode())
357 if code != http.HTTPStatus.OK:
358 self.close()
359 raise OSError(
360 f"Tunnel connection failed: {code} {message.strip()}"
361 )
363 finally:
364 response.close()
366 def connect(self) -> None:
367 self.sock = self._new_conn()
368 if self._tunnel_host:
369 # If we're tunneling it means we're connected to our proxy.
370 self._has_connected_to_proxy = True
372 # TODO: Fix tunnel so it doesn't depend on self.sock state.
373 self._tunnel()
375 # If there's a proxy to be connected to we are fully connected.
376 # This is set twice (once above and here) due to forwarding proxies
377 # not using tunnelling.
378 self._has_connected_to_proxy = bool(self.proxy)
380 if self._has_connected_to_proxy:
381 self.proxy_is_verified = False
383 @property
384 def is_closed(self) -> bool:
385 return self.sock is None
387 @property
388 def is_connected(self) -> bool:
389 if self.sock is None:
390 return False
391 return not wait_for_read(self.sock, timeout=0.0)
393 @property
394 def has_connected_to_proxy(self) -> bool:
395 return self._has_connected_to_proxy
397 @property
398 def proxy_is_forwarding(self) -> bool:
399 """
400 Return True if a forwarding proxy is configured, else return False
401 """
402 return bool(self.proxy) and self._tunnel_host is None
404 @property
405 def proxy_is_tunneling(self) -> bool:
406 """
407 Return True if a tunneling proxy is configured, else return False
408 """
409 return self._tunnel_host is not None
411 def close(self) -> None:
412 try:
413 super().close()
414 finally:
415 # Reset all stateful properties so connection
416 # can be re-used without leaking prior configs.
417 self.sock = None
418 self.is_verified = False
419 self.proxy_is_verified = None
420 self._has_connected_to_proxy = False
421 self._response_options = None
422 self._tunnel_host = None
423 self._tunnel_port = None
424 self._tunnel_scheme = None
426 def putrequest(
427 self,
428 method: str,
429 url: str,
430 skip_host: bool = False,
431 skip_accept_encoding: bool = False,
432 ) -> None:
433 """"""
434 # Empty docstring because the indentation of CPython's implementation
435 # is broken but we don't want this method in our documentation.
436 match = _CONTAINS_CONTROL_CHAR_RE.search(method)
437 if match:
438 raise ValueError(
439 f"Method cannot contain non-token characters {method!r} (found at least {match.group()!r})"
440 )
442 return super().putrequest(
443 method, url, skip_host=skip_host, skip_accept_encoding=skip_accept_encoding
444 )
446 def putheader(self, header: str, *values: str) -> None: # type: ignore[override]
447 """"""
448 if not any(isinstance(v, str) and v == SKIP_HEADER for v in values):
449 super().putheader(header, *values)
450 elif to_str(header.lower()) not in SKIPPABLE_HEADERS:
451 skippable_headers = "', '".join(
452 [str.title(header) for header in sorted(SKIPPABLE_HEADERS)]
453 )
454 raise ValueError(
455 f"urllib3.util.SKIP_HEADER only supports '{skippable_headers}'"
456 )
458 # `request` method's signature intentionally violates LSP.
459 # urllib3's API is different from `http.client.HTTPConnection` and the subclassing is only incidental.
460 def request( # type: ignore[override]
461 self,
462 method: str,
463 url: str,
464 body: _TYPE_BODY | None = None,
465 headers: typing.Mapping[str, str] | None = None,
466 *,
467 chunked: bool = False,
468 preload_content: bool = True,
469 decode_content: bool = True,
470 enforce_content_length: bool = True,
471 ) -> None:
472 # Update the inner socket's timeout value to send the request.
473 # This only triggers if the connection is re-used.
474 if self.sock is not None:
475 self.sock.settimeout(self.timeout)
477 # Store these values to be fed into the HTTPResponse
478 # object later. TODO: Remove this in favor of a real
479 # HTTP lifecycle mechanism.
481 # We have to store these before we call .request()
482 # because sometimes we can still salvage a response
483 # off the wire even if we aren't able to completely
484 # send the request body.
485 self._response_options = _ResponseOptions(
486 request_method=method,
487 request_url=url,
488 preload_content=preload_content,
489 decode_content=decode_content,
490 enforce_content_length=enforce_content_length,
491 )
493 if headers is None:
494 headers = {}
495 header_keys = frozenset(to_str(k.lower()) for k in headers)
496 skip_accept_encoding = "accept-encoding" in header_keys
497 skip_host = "host" in header_keys
498 self.putrequest(
499 method, url, skip_accept_encoding=skip_accept_encoding, skip_host=skip_host
500 )
502 # Transform the body into an iterable of sendall()-able chunks
503 # and detect if an explicit Content-Length is doable.
504 chunks_and_cl = body_to_chunks(body, method=method, blocksize=self.blocksize)
505 chunks = chunks_and_cl.chunks
506 content_length = chunks_and_cl.content_length
508 # When chunked is explicit set to 'True' we respect that.
509 if chunked:
510 if "transfer-encoding" not in header_keys:
511 self.putheader("Transfer-Encoding", "chunked")
512 else:
513 # Detect whether a framing mechanism is already in use. If so
514 # we respect that value, otherwise we pick chunked vs content-length
515 # depending on the type of 'body'.
516 if "content-length" in header_keys:
517 chunked = False
518 elif "transfer-encoding" in header_keys:
519 chunked = True
521 # Otherwise we go off the recommendation of 'body_to_chunks()'.
522 else:
523 chunked = False
524 if content_length is None:
525 if chunks is not None:
526 chunked = True
527 self.putheader("Transfer-Encoding", "chunked")
528 else:
529 self.putheader("Content-Length", str(content_length))
531 # Now that framing headers are out of the way we send all the other headers.
532 if "user-agent" not in header_keys:
533 self.putheader("User-Agent", _get_default_user_agent())
534 for header, value in headers.items():
535 self.putheader(header, value)
536 self.endheaders()
538 # If we're given a body we start sending that in chunks.
539 if chunks is not None:
540 for chunk in chunks:
541 # Sending empty chunks isn't allowed for TE: chunked
542 # as it indicates the end of the body.
543 if not chunk:
544 continue
545 if isinstance(chunk, str):
546 chunk = chunk.encode("utf-8")
547 if chunked:
548 self.send(b"%x\r\n%b\r\n" % (len(chunk), chunk))
549 else:
550 self.send(chunk)
552 # Regardless of whether we have a body or not, if we're in
553 # chunked mode we want to send an explicit empty chunk.
554 if chunked:
555 self.send(b"0\r\n\r\n")
557 def request_chunked(
558 self,
559 method: str,
560 url: str,
561 body: _TYPE_BODY | None = None,
562 headers: typing.Mapping[str, str] | None = None,
563 ) -> None:
564 """
565 Alternative to the common request method, which sends the
566 body with chunked encoding and not as one block
567 """
568 warnings.warn(
569 "HTTPConnection.request_chunked() is deprecated and will be removed "
570 "in urllib3 v3.0. Instead use HTTPConnection.request(..., chunked=True).",
571 category=FutureWarning,
572 stacklevel=2,
573 )
574 self.request(method, url, body=body, headers=headers, chunked=True)
576 def getresponse( # type: ignore[override]
577 self,
578 ) -> HTTPResponse:
579 """
580 Get the response from the server.
582 If the HTTPConnection is in the correct state, returns an instance of HTTPResponse or of whatever object is returned by the response_class variable.
584 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.
585 """
586 # Raise the same error as http.client.HTTPConnection
587 if self._response_options is None:
588 raise ResponseNotReady()
590 # Reset this attribute for being used again.
591 resp_options = self._response_options
592 self._response_options = None
594 # Since the connection's timeout value may have been updated
595 # we need to set the timeout on the socket.
596 self.sock.settimeout(self.timeout)
598 # This is needed here to avoid circular import errors
599 from .response import HTTPResponse
601 # Save a reference to the shutdown function before ownership is passed
602 # to httplib_response
603 # TODO should we implement it everywhere?
604 _shutdown = getattr(self.sock, "shutdown", None)
606 # Get the response from http.client.HTTPConnection
607 httplib_response = super().getresponse()
609 try:
610 assert_header_parsing(httplib_response.msg)
611 except (HeaderParsingError, TypeError) as hpe:
612 log.warning(
613 "Failed to parse headers (url=%s): %s",
614 _url_from_connection(self, resp_options.request_url),
615 hpe,
616 exc_info=True,
617 )
619 headers = HTTPHeaderDict(httplib_response.msg.items())
621 response = HTTPResponse(
622 body=httplib_response,
623 headers=headers,
624 status=httplib_response.status,
625 version=httplib_response.version,
626 version_string=getattr(self, "_http_vsn_str", "HTTP/?"),
627 reason=httplib_response.reason,
628 preload_content=resp_options.preload_content,
629 decode_content=resp_options.decode_content,
630 original_response=httplib_response,
631 enforce_content_length=resp_options.enforce_content_length,
632 request_method=resp_options.request_method,
633 request_url=resp_options.request_url,
634 sock_shutdown=_shutdown,
635 )
636 return response
639class HTTPSConnection(HTTPConnection):
640 """
641 Many of the parameters to this constructor are passed to the underlying SSL
642 socket by means of :py:func:`urllib3.util.ssl_wrap_socket`.
643 """
645 default_port = port_by_scheme["https"] # type: ignore[misc]
647 cert_reqs: int | str | None = None
648 ca_certs: str | None = None
649 ca_cert_dir: str | None = None
650 ca_cert_data: None | str | bytes = None
651 ssl_version: int | str | None = None
652 ssl_minimum_version: int | None = None
653 ssl_maximum_version: int | None = None
654 assert_fingerprint: str | None = None
655 _connect_callback: typing.Callable[..., None] | None = None
657 def __init__(
658 self,
659 host: str,
660 port: int | None = None,
661 *,
662 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
663 source_address: tuple[str, int] | None = None,
664 blocksize: int = 16384,
665 socket_options: None | (
666 connection._TYPE_SOCKET_OPTIONS
667 ) = HTTPConnection.default_socket_options,
668 proxy: Url | None = None,
669 proxy_config: ProxyConfig | None = None,
670 cert_reqs: int | str | None = None,
671 assert_hostname: None | str | typing.Literal[False] = None,
672 assert_fingerprint: str | None = None,
673 server_hostname: str | None = None,
674 ssl_context: ssl.SSLContext | None = None,
675 ca_certs: str | None = None,
676 ca_cert_dir: str | None = None,
677 ca_cert_data: None | str | bytes = None,
678 ssl_minimum_version: int | None = None,
679 ssl_maximum_version: int | None = None,
680 ssl_version: int | str | None = None, # Deprecated
681 cert_file: str | None = None,
682 key_file: str | None = None,
683 key_password: str | None = None,
684 ) -> None:
685 super().__init__(
686 host,
687 port=port,
688 timeout=timeout,
689 source_address=source_address,
690 blocksize=blocksize,
691 socket_options=socket_options,
692 proxy=proxy,
693 proxy_config=proxy_config,
694 )
696 self.key_file = key_file
697 self.cert_file = cert_file
698 self.key_password = key_password
699 self.ssl_context = ssl_context
700 self.server_hostname = server_hostname
701 self.assert_hostname = assert_hostname
702 self.assert_fingerprint = assert_fingerprint
703 self.ssl_version = ssl_version
704 self.ssl_minimum_version = ssl_minimum_version
705 self.ssl_maximum_version = ssl_maximum_version
706 self.ca_certs = ca_certs and os.path.expanduser(ca_certs)
707 self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir)
708 self.ca_cert_data = ca_cert_data
710 # cert_reqs depends on ssl_context so calculate last.
711 if cert_reqs is None:
712 if self.ssl_context is not None:
713 cert_reqs = self.ssl_context.verify_mode
714 else:
715 cert_reqs = resolve_cert_reqs(None)
716 self.cert_reqs = cert_reqs
717 self._connect_callback = None
719 def set_cert(
720 self,
721 key_file: str | None = None,
722 cert_file: str | None = None,
723 cert_reqs: int | str | None = None,
724 key_password: str | None = None,
725 ca_certs: str | None = None,
726 assert_hostname: None | str | typing.Literal[False] = None,
727 assert_fingerprint: str | None = None,
728 ca_cert_dir: str | None = None,
729 ca_cert_data: None | str | bytes = None,
730 ) -> None:
731 """
732 This method should only be called once, before the connection is used.
733 """
734 warnings.warn(
735 "HTTPSConnection.set_cert() is deprecated and will be removed "
736 "in urllib3 v3.0. Instead provide the parameters to the "
737 "HTTPSConnection constructor.",
738 category=FutureWarning,
739 stacklevel=2,
740 )
742 # If cert_reqs is not provided we'll assume CERT_REQUIRED unless we also
743 # have an SSLContext object in which case we'll use its verify_mode.
744 if cert_reqs is None:
745 if self.ssl_context is not None:
746 cert_reqs = self.ssl_context.verify_mode
747 else:
748 cert_reqs = resolve_cert_reqs(None)
750 self.key_file = key_file
751 self.cert_file = cert_file
752 self.cert_reqs = cert_reqs
753 self.key_password = key_password
754 self.assert_hostname = assert_hostname
755 self.assert_fingerprint = assert_fingerprint
756 self.ca_certs = ca_certs and os.path.expanduser(ca_certs)
757 self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir)
758 self.ca_cert_data = ca_cert_data
760 def connect(self) -> None:
761 # Today we don't need to be doing this step before the /actual/ socket
762 # connection, however in the future we'll need to decide whether to
763 # create a new socket or re-use an existing "shared" socket as a part
764 # of the HTTP/2 handshake dance.
765 if self._tunnel_host is not None and self._tunnel_port is not None:
766 probe_http2_host = self._tunnel_host
767 probe_http2_port = self._tunnel_port
768 else:
769 probe_http2_host = self.host
770 probe_http2_port = self.port
772 # Check if the target origin supports HTTP/2.
773 # If the value comes back as 'None' it means that the current thread
774 # is probing for HTTP/2 support. Otherwise, we're waiting for another
775 # probe to complete, or we get a value right away.
776 target_supports_http2: bool | None
777 if "h2" in ssl_.ALPN_PROTOCOLS:
778 target_supports_http2 = http2_probe.acquire_and_get(
779 host=probe_http2_host, port=probe_http2_port
780 )
781 else:
782 # If HTTP/2 isn't going to be offered it doesn't matter if
783 # the target supports HTTP/2. Don't want to make a probe.
784 target_supports_http2 = False
786 if self._connect_callback is not None:
787 self._connect_callback(
788 "before connect",
789 thread_id=threading.get_ident(),
790 target_supports_http2=target_supports_http2,
791 )
793 try:
794 sock: socket.socket | ssl.SSLSocket
795 self.sock = sock = self._new_conn()
796 server_hostname: str = self.host
797 tls_in_tls = False
799 # Do we need to establish a tunnel?
800 if self.proxy_is_tunneling:
801 # We're tunneling to an HTTPS origin so need to do TLS-in-TLS.
802 if self._tunnel_scheme == "https":
803 # _connect_tls_proxy will verify and assign proxy_is_verified
804 self.sock = sock = self._connect_tls_proxy(self.host, sock)
805 tls_in_tls = True
806 elif self._tunnel_scheme == "http":
807 self.proxy_is_verified = False
809 # If we're tunneling it means we're connected to our proxy.
810 self._has_connected_to_proxy = True
812 self._tunnel()
813 # Override the host with the one we're requesting data from.
814 server_hostname = typing.cast(str, self._tunnel_host)
816 if self.server_hostname is not None:
817 server_hostname = self.server_hostname
819 is_time_off = datetime.date.today() < RECENT_DATE
820 if is_time_off:
821 warnings.warn(
822 (
823 f"System time is way off (before {RECENT_DATE}). This will probably "
824 "lead to SSL verification errors"
825 ),
826 SystemTimeWarning,
827 )
829 # Remove trailing '.' from fqdn hostnames to allow certificate validation
830 server_hostname_rm_dot = server_hostname.rstrip(".")
832 # Forwarding proxies should use proxy SSL context for
833 # wrapping since that's the connection being established,
834 # whereas tunneling proxies should use the connection's SSL
835 # context.
836 # However, for backwards compatibility reasons, if the proxy
837 # is forwarding but no proxy SSL context is provided, we
838 # fall back to using the connection's SSL context until
839 # urllib3 v3.0. Appropriate warning is emitted in
840 # ``ProxyManager.__init__``.
841 if self.proxy_is_forwarding and self.proxy_config is not None:
842 ssl_context = self.proxy_config.ssl_context
843 else:
844 ssl_context = self.ssl_context
846 sock_and_verified = _ssl_wrap_socket_and_match_hostname(
847 sock=sock,
848 cert_reqs=self.cert_reqs,
849 ssl_version=self.ssl_version,
850 ssl_minimum_version=self.ssl_minimum_version,
851 ssl_maximum_version=self.ssl_maximum_version,
852 ca_certs=self.ca_certs,
853 ca_cert_dir=self.ca_cert_dir,
854 ca_cert_data=self.ca_cert_data,
855 cert_file=self.cert_file,
856 key_file=self.key_file,
857 key_password=self.key_password,
858 server_hostname=server_hostname_rm_dot,
859 ssl_context=ssl_context,
860 tls_in_tls=tls_in_tls,
861 assert_hostname=self.assert_hostname,
862 assert_fingerprint=self.assert_fingerprint,
863 )
864 self.sock = sock_and_verified.socket
866 # If an error occurs during connection/handshake we may need to release
867 # our lock so another connection can probe the origin.
868 except BaseException:
869 if self._connect_callback is not None:
870 self._connect_callback(
871 "after connect failure",
872 thread_id=threading.get_ident(),
873 target_supports_http2=target_supports_http2,
874 )
876 if target_supports_http2 is None:
877 http2_probe.set_and_release(
878 host=probe_http2_host, port=probe_http2_port, supports_http2=None
879 )
880 raise
882 # If this connection doesn't know if the origin supports HTTP/2
883 # we report back to the HTTP/2 probe our result.
884 if target_supports_http2 is None:
885 supports_http2 = sock_and_verified.socket.selected_alpn_protocol() == "h2"
886 http2_probe.set_and_release(
887 host=probe_http2_host,
888 port=probe_http2_port,
889 supports_http2=supports_http2,
890 )
892 # Forwarding proxies can never have a verified target since
893 # the proxy is the one doing the verification. Should instead
894 # use a CONNECT tunnel in order to verify the target.
895 # See: https://github.com/urllib3/urllib3/issues/3267.
896 if self.proxy_is_forwarding:
897 self.is_verified = False
898 else:
899 self.is_verified = sock_and_verified.is_verified
901 # If there's a proxy to be connected to we are fully connected.
902 # This is set twice (once above and here) due to forwarding proxies
903 # not using tunnelling.
904 self._has_connected_to_proxy = bool(self.proxy)
906 # Set `self.proxy_is_verified` unless it's already set while
907 # establishing a tunnel.
908 if self._has_connected_to_proxy and self.proxy_is_verified is None:
909 self.proxy_is_verified = sock_and_verified.is_verified
911 def _connect_tls_proxy(self, hostname: str, sock: socket.socket) -> ssl.SSLSocket:
912 """
913 Establish a TLS connection to the proxy using the provided SSL context.
914 """
915 # `_connect_tls_proxy` is called when self._tunnel_host is truthy.
916 proxy_config = typing.cast(ProxyConfig, self.proxy_config)
917 ssl_context = proxy_config.ssl_context
918 sock_and_verified = _ssl_wrap_socket_and_match_hostname(
919 sock,
920 cert_reqs=self.cert_reqs,
921 ssl_version=self.ssl_version,
922 ssl_minimum_version=self.ssl_minimum_version,
923 ssl_maximum_version=self.ssl_maximum_version,
924 ca_certs=self.ca_certs,
925 ca_cert_dir=self.ca_cert_dir,
926 ca_cert_data=self.ca_cert_data,
927 server_hostname=hostname,
928 ssl_context=ssl_context,
929 assert_hostname=proxy_config.assert_hostname,
930 assert_fingerprint=proxy_config.assert_fingerprint,
931 # Features that aren't implemented for proxies yet:
932 cert_file=None,
933 key_file=None,
934 key_password=None,
935 tls_in_tls=False,
936 )
937 self.proxy_is_verified = sock_and_verified.is_verified
938 return sock_and_verified.socket # type: ignore[return-value]
941class _WrappedAndVerifiedSocket(typing.NamedTuple):
942 """
943 Wrapped socket and whether the connection is
944 verified after the TLS handshake
945 """
947 socket: ssl.SSLSocket | SSLTransport
948 is_verified: bool
951def _ssl_wrap_socket_and_match_hostname(
952 sock: socket.socket,
953 *,
954 cert_reqs: None | str | int,
955 ssl_version: None | str | int,
956 ssl_minimum_version: int | None,
957 ssl_maximum_version: int | None,
958 cert_file: str | None,
959 key_file: str | None,
960 key_password: str | None,
961 ca_certs: str | None,
962 ca_cert_dir: str | None,
963 ca_cert_data: None | str | bytes,
964 assert_hostname: None | str | typing.Literal[False],
965 assert_fingerprint: str | None,
966 server_hostname: str | None,
967 ssl_context: ssl.SSLContext | None,
968 tls_in_tls: bool = False,
969) -> _WrappedAndVerifiedSocket:
970 """Logic for constructing an SSLContext from all TLS parameters, passing
971 that down into ssl_wrap_socket, and then doing certificate verification
972 either via hostname or fingerprint. This function exists to guarantee
973 that both proxies and targets have the same behavior when connecting via TLS.
974 """
975 default_ssl_context = False
976 if ssl_context is None:
977 default_ssl_context = True
978 context = create_urllib3_context(
979 ssl_version=resolve_ssl_version(ssl_version),
980 ssl_minimum_version=ssl_minimum_version,
981 ssl_maximum_version=ssl_maximum_version,
982 cert_reqs=resolve_cert_reqs(cert_reqs),
983 )
984 else:
985 context = ssl_context
987 context.verify_mode = resolve_cert_reqs(cert_reqs)
989 # In some cases, we want to verify hostnames ourselves
990 if (
991 # `ssl` can't verify fingerprints or alternate hostnames
992 assert_fingerprint
993 or assert_hostname
994 # assert_hostname can be set to False to disable hostname checking
995 or assert_hostname is False
996 # We still support OpenSSL 1.0.2, which prevents us from verifying
997 # hostnames easily: https://github.com/pyca/pyopenssl/pull/933
998 or ssl_.IS_PYOPENSSL
999 or not ssl_.HAS_NEVER_CHECK_COMMON_NAME
1000 ):
1001 context.check_hostname = False
1003 # Try to load OS default certs if none are given. We need to do the hasattr() check
1004 # for custom pyOpenSSL SSLContext objects because they don't support
1005 # load_default_certs().
1006 if (
1007 not ca_certs
1008 and not ca_cert_dir
1009 and not ca_cert_data
1010 and default_ssl_context
1011 and hasattr(context, "load_default_certs")
1012 ):
1013 context.load_default_certs()
1015 # Ensure that IPv6 addresses are in the proper format and don't have a
1016 # scope ID. Python's SSL module fails to recognize scoped IPv6 addresses
1017 # and interprets them as DNS hostnames.
1018 if server_hostname is not None:
1019 normalized = server_hostname.strip("[]")
1020 if "%" in normalized:
1021 normalized = normalized[: normalized.rfind("%")]
1022 if is_ipaddress(normalized):
1023 server_hostname = normalized
1025 ssl_sock = ssl_wrap_socket(
1026 sock=sock,
1027 keyfile=key_file,
1028 certfile=cert_file,
1029 key_password=key_password,
1030 ca_certs=ca_certs,
1031 ca_cert_dir=ca_cert_dir,
1032 ca_cert_data=ca_cert_data,
1033 server_hostname=server_hostname,
1034 ssl_context=context,
1035 tls_in_tls=tls_in_tls,
1036 )
1038 try:
1039 if assert_fingerprint:
1040 _assert_fingerprint(
1041 ssl_sock.getpeercert(binary_form=True), assert_fingerprint
1042 )
1043 elif (
1044 context.verify_mode != ssl.CERT_NONE
1045 and not context.check_hostname
1046 and assert_hostname is not False
1047 ):
1048 cert: _TYPE_PEER_CERT_RET_DICT = ssl_sock.getpeercert() # type: ignore[assignment]
1050 # Need to signal to our match_hostname whether to use 'commonName' or not.
1051 # If we're using our own constructed SSLContext we explicitly set 'False'
1052 # because PyPy hard-codes 'True' from SSLContext.hostname_checks_common_name.
1053 if default_ssl_context:
1054 hostname_checks_common_name = False
1055 else:
1056 hostname_checks_common_name = (
1057 getattr(context, "hostname_checks_common_name", False) or False
1058 )
1060 _match_hostname(
1061 cert,
1062 assert_hostname or server_hostname, # type: ignore[arg-type]
1063 hostname_checks_common_name,
1064 )
1066 return _WrappedAndVerifiedSocket(
1067 socket=ssl_sock,
1068 is_verified=context.verify_mode == ssl.CERT_REQUIRED
1069 or bool(assert_fingerprint),
1070 )
1071 except BaseException:
1072 ssl_sock.close()
1073 raise
1076def _match_hostname(
1077 cert: _TYPE_PEER_CERT_RET_DICT | None,
1078 asserted_hostname: str,
1079 hostname_checks_common_name: bool = False,
1080) -> None:
1081 # Our upstream implementation of ssl.match_hostname()
1082 # only applies this normalization to IP addresses so it doesn't
1083 # match DNS SANs so we do the same thing!
1084 stripped_hostname = asserted_hostname.strip("[]")
1085 if is_ipaddress(stripped_hostname):
1086 asserted_hostname = stripped_hostname
1088 try:
1089 match_hostname(cert, asserted_hostname, hostname_checks_common_name)
1090 except CertificateError as e:
1091 log.warning(
1092 "Certificate did not match expected hostname: %s. Certificate: %s",
1093 asserted_hostname,
1094 cert,
1095 )
1096 # Add cert to exception and reraise so client code can inspect
1097 # the cert when catching the exception, if they want to
1098 e._peer_cert = cert # type: ignore[attr-defined]
1099 raise
1102def _wrap_proxy_error(err: Exception, proxy_scheme: str | None) -> ProxyError:
1103 # Look for the phrase 'wrong version number', if found
1104 # then we should warn the user that we're very sure that
1105 # this proxy is HTTP-only and they have a configuration issue.
1106 error_normalized = " ".join(re.split("[^a-z]", str(err).lower()))
1107 is_likely_http_proxy = (
1108 "wrong version number" in error_normalized
1109 or "unknown protocol" in error_normalized
1110 or "record layer failure" in error_normalized
1111 )
1112 http_proxy_warning = (
1113 ". Your proxy appears to only use HTTP and not HTTPS, "
1114 "try changing your proxy URL to be HTTP. See: "
1115 "https://urllib3.readthedocs.io/en/latest/advanced-usage.html"
1116 "#https-proxy-error-http-proxy"
1117 )
1118 new_err = ProxyError(
1119 f"Unable to connect to proxy"
1120 f"{http_proxy_warning if is_likely_http_proxy and proxy_scheme == 'https' else ''}",
1121 err,
1122 )
1123 new_err.__cause__ = err
1124 return new_err
1127def _get_default_user_agent() -> str:
1128 return f"python-urllib3/{__version__}"
1131class DummyConnection:
1132 """Used to detect a failed ConnectionCls import."""
1135if not ssl:
1136 HTTPSConnection = DummyConnection # type: ignore[misc, assignment] # noqa: F811
1139VerifiedHTTPSConnection = HTTPSConnection
1142def _url_from_connection(
1143 conn: HTTPConnection | HTTPSConnection, path: str | None = None
1144) -> str:
1145 """Returns the URL from a given connection. This is mainly used for testing and logging."""
1147 scheme = "https" if isinstance(conn, HTTPSConnection) else "http"
1149 return Url(scheme=scheme, host=conn.host, port=conn.port, path=path).url