Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/urllib3/connection.py: 24%
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]")
80# Starting the optional OWS match at the beginning of a whitespace run avoids
81# quadratic backtracking for long header values.
82_OBSOLETE_FOLD_RE = re.compile(r"(?:(?<![ \t])[ \t]+)?\r\n[ \t]+")
85def _normalize_header_value(value: str) -> str:
86 if "\r" in value or "\n" in value:
87 value = _OBSOLETE_FOLD_RE.sub(" ", value)
88 # A CR or LF outside of an obs-fold is invalid in a field value, but
89 # http.client can let one through, e.g. when a server ends lines with
90 # a bare LF. RFC 9110 section 5.5 requires rejecting the message or
91 # replacing each such character with SP.
92 value = value.replace("\r", " ").replace("\n", " ")
93 # RFC 9110 section 5.5 treats NUL the same way, and leading and trailing
94 # whitespace is not part of the field value.
95 if "\x00" in value:
96 value = value.replace("\x00", " ")
97 return value.strip(" \t")
100def _normalize_header_values(
101 message: http.client.HTTPMessage,
102) -> list[tuple[str, str]]:
103 header_items = message.items()
104 headers_changed = False
105 for index, (name, value) in enumerate(header_items):
106 normalized_value = _normalize_header_value(value)
107 if normalized_value != value:
108 header_items[index] = (name, normalized_value)
109 headers_changed = True
111 if headers_changed:
112 # Keep the original message in sync for downstream consumers such as
113 # cookie jars while preserving its identity and header ordering.
114 for name, _ in header_items:
115 del message[name]
116 for name, value in header_items:
117 message[name] = value
119 return header_items
122class HTTPConnection(_HTTPConnection):
123 """
124 Based on :class:`http.client.HTTPConnection` but provides an extra constructor
125 backwards-compatibility layer between older and newer Pythons.
127 Additional keyword parameters are used to configure attributes of the connection.
128 Accepted parameters include:
130 - ``source_address``: Set the source address for the current connection.
131 - ``blocksize``: Read size for file-like request bodies, in bytes. Defaults to
132 16384 (16 KiB). Does not split or resize elements of iterable bodies.
133 Larger values may improve upload throughput but use more memory per active
134 upload. See :ref:`upload_buffering` for tuning guidance.
135 - ``socket_options``: Set specific options on the underlying socket. If not specified, then
136 defaults are loaded from ``HTTPConnection.default_socket_options`` which includes disabling
137 Nagle's algorithm (sets TCP_NODELAY to 1) unless the connection is behind a proxy.
139 For example, if you wish to enable TCP Keep Alive in addition to the defaults,
140 you might pass:
142 .. code-block:: python
144 HTTPConnection.default_socket_options + [
145 (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
146 ]
148 Or you may want to disable the defaults by passing an empty list (e.g., ``[]``).
149 """
151 default_port: typing.ClassVar[int] = port_by_scheme["http"] # type: ignore[misc]
153 #: Disable Nagle's algorithm by default.
154 #: ``[(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]``
155 #:
156 #: Use the ``socket_options`` parameter of :class:`~urllib3.PoolManager`,
157 #: :class:`~urllib3.ProxyManager`, or :class:`~urllib3.HTTPConnectionPool`
158 #: to change this behavior.
159 default_socket_options: typing.ClassVar[
160 typing.Final[connection._TYPE_SOCKET_OPTIONS]
161 ] = [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]
163 #: Whether this connection verifies the host's certificate.
164 is_verified: bool = False
166 #: Whether this proxy connection verified the proxy host's certificate.
167 # If no proxy is currently connected to the value will be ``None``.
168 proxy_is_verified: bool | None = None
170 blocksize: int
171 source_address: tuple[str, int] | None
172 socket_options: connection._TYPE_SOCKET_OPTIONS | None
174 _has_connected_to_proxy: bool
175 _response_options: _ResponseOptions | None
176 _tunnel_host: str | None
177 _tunnel_port: int | None
178 _tunnel_scheme: str | None
180 def __init__(
181 self,
182 host: str,
183 port: int | None = None,
184 *,
185 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
186 source_address: tuple[str, int] | None = None,
187 blocksize: int = 16384,
188 socket_options: None | (
189 connection._TYPE_SOCKET_OPTIONS
190 ) = default_socket_options,
191 proxy: Url | None = None,
192 proxy_config: ProxyConfig | None = None,
193 ) -> None:
194 super().__init__(
195 host=host,
196 port=port,
197 timeout=Timeout.resolve_default_timeout(timeout),
198 source_address=source_address,
199 blocksize=blocksize,
200 )
201 self.socket_options = socket_options
202 self.proxy = proxy
203 self.proxy_config = proxy_config
205 self._has_connected_to_proxy = False
206 self._response_options = None
207 self._tunnel_host: str | None = None
208 self._tunnel_port: int | None = None
209 self._tunnel_scheme: str | None = None
211 def __str__(self) -> str:
212 return f"{type(self).__name__}(host={self.host!r}, port={self.port!r})"
214 def __repr__(self) -> str:
215 return f"<{self} at {id(self):#x}>"
217 @property
218 def host(self) -> str:
219 """
220 Getter method to remove any trailing dots that indicate the hostname is an FQDN.
222 In general, SSL certificates don't include the trailing dot indicating a
223 fully-qualified domain name, and thus, they don't validate properly when
224 checked against a domain name that includes the dot. In addition, some
225 servers may not expect to receive the trailing dot when provided.
227 However, the hostname with trailing dot is critical to DNS resolution; doing a
228 lookup with the trailing dot will properly only resolve the appropriate FQDN,
229 whereas a lookup without a trailing dot will search the system's search domain
230 list. Thus, it's important to keep the original host around for use only in
231 those cases where it's appropriate (i.e., when doing DNS lookup to establish the
232 actual TCP connection across which we're going to send HTTP requests).
233 """
234 return self._dns_host.rstrip(".")
236 @host.setter
237 def host(self, value: str) -> None:
238 """
239 Setter for the `host` property.
241 We assume that only urllib3 uses the _dns_host attribute; httplib itself
242 only uses `host`, and it seems reasonable that other libraries follow suit.
243 """
244 self._dns_host = value
246 def _new_conn(self) -> socket.socket:
247 """Establish a socket connection and set nodelay settings on it.
249 :return: New socket connection.
250 """
251 try:
252 sock = connection.create_connection(
253 (self._dns_host, self.port),
254 self.timeout,
255 source_address=self.source_address,
256 socket_options=self.socket_options,
257 )
258 except socket.gaierror as e:
259 raise NameResolutionError(self.host, self, e) from e
260 except SocketTimeout as e:
261 raise ConnectTimeoutError(
262 self,
263 f"Connection to {self.host} timed out. (connect timeout={self.timeout})",
264 ) from e
266 except OSError as e:
267 raise NewConnectionError(
268 self, f"Failed to establish a new connection: {e}"
269 ) from e
271 sys.audit("http.client.connect", self, self.host, self.port)
273 return sock
275 def set_tunnel(
276 self,
277 host: str,
278 port: int | None = None,
279 headers: typing.Mapping[str, str] | None = None,
280 scheme: str = "http",
281 ) -> None:
282 if scheme not in ("http", "https"):
283 raise ValueError(
284 f"Invalid proxy scheme for tunneling: {scheme!r}, must be either 'http' or 'https'"
285 )
286 super().set_tunnel(host, port=port, headers=headers)
287 self._tunnel_scheme = scheme
289 if sys.version_info < (3, 11, 16) or ((3, 12) <= sys.version_info < (3, 12, 14)):
290 # Taken from python/cpython#100986 which was backported in 3.11.9 and 3.12.3.
291 # When using connection_from_host, host will come without brackets.
292 # With a security patch from python/cpython#146211.
293 def _wrap_ipv6(self, ip: bytes) -> bytes:
294 if b":" in ip and ip[0] != b"["[0]:
295 return b"[" + ip + b"]"
296 return ip
298 # Copied from CPython 3.12.13 Lib/http/client.py
299 _is_legal_header_name = staticmethod(re.compile(rb"[^:\s][^:\r\n]*").fullmatch)
300 _is_illegal_header_value = staticmethod(
301 re.compile(rb"\n(?![ \t])|\r(?![ \t\n])").search
302 )
303 _contains_disallowed_url_pchar_re = re.compile("[\x00-\x20\x7f]")
305 if sys.version_info < (3, 11, 16):
306 # `_tunnel` copied from 3.11.15 backporting
307 # https://github.com/python/cpython/commit/0d4026432591d43185568dd31cef6a034c4b9261
308 # and https://github.com/python/cpython/commit/6fbc61070fda2ffb8889e77e3b24bca4249ab4d1
309 # plus a fix from https://github.com/python/cpython/commit/56b7100b04e44ea27989242b176beb8f016b2c53
310 def _tunnel(self) -> None:
311 if self._contains_disallowed_url_pchar_re.search(self._tunnel_host): # type: ignore[arg-type]
312 raise ValueError(
313 "Tunnel host can't contain control characters %r"
314 % (self._tunnel_host,)
315 )
316 _MAXLINE = http.client._MAXLINE # type: ignore[attr-defined]
317 connect = b"CONNECT %s:%d HTTP/1.0\r\n" % ( # type: ignore[str-format]
318 self._wrap_ipv6(self._tunnel_host.encode("ascii")), # type: ignore[union-attr]
319 self._tunnel_port,
320 )
321 headers = [connect]
322 for header, value in self._tunnel_headers.items(): # type: ignore[attr-defined]
323 header_bytes = header.encode("latin-1")
324 value_bytes = value.encode("latin-1")
325 if not self._is_legal_header_name(header_bytes):
326 raise ValueError(f"Invalid header name {header_bytes!r}")
327 if self._is_illegal_header_value(value_bytes):
328 raise ValueError(f"Invalid header value {value_bytes!r}")
329 headers.append(b"%s: %s\r\n" % (header_bytes, value_bytes))
330 headers.append(b"\r\n")
331 # Making a single send() call instead of one per line encourages
332 # the host OS to use a more optimal packet size instead of
333 # potentially emitting a series of small packets.
334 self.send(b"".join(headers))
335 del headers
337 response = self.response_class(self.sock, method=self._method) # type: ignore[attr-defined]
338 try:
339 version, code, message = response._read_status() # type: ignore[attr-defined]
341 if code != http.HTTPStatus.OK:
342 self.close()
343 raise OSError(
344 f"Tunnel connection failed: {code} {message.strip()}"
345 )
346 while True:
347 line = response.fp.readline(_MAXLINE + 1)
348 if len(line) > _MAXLINE:
349 raise http.client.LineTooLong("header line")
350 if not line:
351 # for sites which EOF without sending a trailer
352 break
353 if line in (b"\r\n", b"\n", b""):
354 break
356 if self.debuglevel > 0:
357 print("header:", line.decode())
358 finally:
359 response.close()
361 elif (3, 12) <= sys.version_info < (3, 12, 14):
362 # `_tunnel` copied from 3.12.13 backporting
363 # https://github.com/python/cpython/commit/23aef575c7629abcd4aaf028ebd226fb41a4b3c8
364 # plus a fix from https://github.com/python/cpython/commit/c00c386faa579ad71196d33408644478488e43ec
365 def _tunnel(self) -> None: # noqa: F811
366 if self._contains_disallowed_url_pchar_re.search(self._tunnel_host): # type: ignore[arg-type]
367 raise ValueError(
368 "Tunnel host can't contain control characters %r"
369 % (self._tunnel_host,)
370 )
371 connect = b"CONNECT %s:%d HTTP/1.1\r\n" % ( # type: ignore[str-format]
372 self._wrap_ipv6(self._tunnel_host.encode("idna")), # type: ignore[union-attr]
373 self._tunnel_port,
374 )
375 headers = [connect]
376 for header, value in self._tunnel_headers.items(): # type: ignore[attr-defined]
377 header_bytes = header.encode("latin-1")
378 value_bytes = value.encode("latin-1")
379 if not self._is_legal_header_name(header_bytes):
380 raise ValueError(f"Invalid header name {header_bytes!r}")
381 if self._is_illegal_header_value(value_bytes):
382 raise ValueError(f"Invalid header value {value_bytes!r}")
383 headers.append(b"%s: %s\r\n" % (header_bytes, value_bytes))
384 headers.append(b"\r\n")
385 # Making a single send() call instead of one per line encourages
386 # the host OS to use a more optimal packet size instead of
387 # potentially emitting a series of small packets.
388 self.send(b"".join(headers))
389 del headers
391 response = self.response_class(self.sock, method=self._method) # type: ignore[attr-defined]
392 try:
393 version, code, message = response._read_status() # type: ignore[attr-defined]
395 self._raw_proxy_headers = http.client._read_headers(response.fp) # type: ignore[attr-defined]
397 if self.debuglevel > 0:
398 for header in self._raw_proxy_headers:
399 print("header:", header.decode())
401 if code != http.HTTPStatus.OK:
402 self.close()
403 raise OSError(
404 f"Tunnel connection failed: {code} {message.strip()}"
405 )
407 finally:
408 response.close()
410 def connect(self) -> None:
411 self.sock = self._new_conn()
412 if self._tunnel_host:
413 # If we're tunneling it means we're connected to our proxy.
414 self._has_connected_to_proxy = True
416 # TODO: Fix tunnel so it doesn't depend on self.sock state.
417 self._tunnel()
419 # If there's a proxy to be connected to we are fully connected.
420 # This is set twice (once above and here) due to forwarding proxies
421 # not using tunnelling.
422 self._has_connected_to_proxy = bool(self.proxy)
424 if self._has_connected_to_proxy:
425 self.proxy_is_verified = False
427 @property
428 def is_closed(self) -> bool:
429 return self.sock is None
431 @property
432 def is_connected(self) -> bool:
433 if self.sock is None:
434 return False
435 return not wait_for_read(self.sock, timeout=0.0)
437 @property
438 def has_connected_to_proxy(self) -> bool:
439 return self._has_connected_to_proxy
441 @property
442 def proxy_is_forwarding(self) -> bool:
443 """
444 Return True if a forwarding proxy is configured, else return False
445 """
446 return bool(self.proxy) and self._tunnel_host is None
448 @property
449 def proxy_is_tunneling(self) -> bool:
450 """
451 Return True if a tunneling proxy is configured, else return False
452 """
453 return self._tunnel_host is not None
455 def close(self) -> None:
456 try:
457 super().close()
458 finally:
459 # Reset all stateful properties so connection
460 # can be re-used without leaking prior configs.
461 self.sock = None
462 self.is_verified = False
463 self.proxy_is_verified = None
464 self._has_connected_to_proxy = False
465 self._response_options = None
466 self._tunnel_host = None
467 self._tunnel_port = None
468 self._tunnel_scheme = None
470 def putrequest(
471 self,
472 method: str,
473 url: str,
474 skip_host: bool = False,
475 skip_accept_encoding: bool = False,
476 ) -> None:
477 """"""
478 # Empty docstring because the indentation of CPython's implementation
479 # is broken but we don't want this method in our documentation.
480 match = _CONTAINS_CONTROL_CHAR_RE.search(method)
481 if match:
482 raise ValueError(
483 f"Method cannot contain non-token characters {method!r} (found at least {match.group()!r})"
484 )
486 return super().putrequest(
487 method, url, skip_host=skip_host, skip_accept_encoding=skip_accept_encoding
488 )
490 def putheader(self, header: str, *values: str) -> None: # type: ignore[override]
491 """"""
492 if len(values) == 1:
493 # Avoid generator creation, any(), and argument unpacking
494 # on the common single-value path.
495 value = values[0]
496 if not (isinstance(value, str) and value == SKIP_HEADER):
497 super().putheader(header, value)
498 return
499 skip_header = True
500 else:
501 skip_header = any(
502 isinstance(value, str) and value == SKIP_HEADER for value in values
503 )
505 if not skip_header:
506 super().putheader(header, *values)
507 elif to_str(header.lower()) not in SKIPPABLE_HEADERS:
508 skippable_headers = "', '".join(
509 [str.title(header) for header in sorted(SKIPPABLE_HEADERS)]
510 )
511 raise ValueError(
512 f"urllib3.util.SKIP_HEADER only supports '{skippable_headers}'"
513 )
515 # `request` method's signature intentionally violates LSP.
516 # urllib3's API is different from `http.client.HTTPConnection` and the subclassing is only incidental.
517 def request( # type: ignore[override]
518 self,
519 method: str,
520 url: str,
521 body: _TYPE_BODY | None = None,
522 headers: typing.Mapping[str, str] | None = None,
523 *,
524 chunked: bool = False,
525 preload_content: bool = True,
526 decode_content: bool = True,
527 enforce_content_length: bool = True,
528 ) -> None:
529 # Update the inner socket's timeout value to send the request.
530 # This only triggers if the connection is re-used.
531 if self.sock is not None:
532 self.sock.settimeout(self.timeout)
534 # Store these values to be fed into the HTTPResponse
535 # object later. TODO: Remove this in favor of a real
536 # HTTP lifecycle mechanism.
538 # We have to store these before we call .request()
539 # because sometimes we can still salvage a response
540 # off the wire even if we aren't able to completely
541 # send the request body.
542 self._response_options = _ResponseOptions(
543 request_method=method,
544 request_url=url,
545 preload_content=preload_content,
546 decode_content=decode_content,
547 enforce_content_length=enforce_content_length,
548 )
550 if headers is None:
551 headers = {}
552 header_keys = frozenset(to_str(k.lower()) for k in headers)
553 skip_accept_encoding = "accept-encoding" in header_keys
554 skip_host = "host" in header_keys
555 self.putrequest(
556 method, url, skip_accept_encoding=skip_accept_encoding, skip_host=skip_host
557 )
559 # Transform the body into an iterable of sendall()-able chunks
560 # and detect if an explicit Content-Length is doable.
561 chunks_and_cl = body_to_chunks(body, method=method, blocksize=self.blocksize)
562 chunks = chunks_and_cl.chunks
563 content_length = chunks_and_cl.content_length
565 # When chunked is explicit set to 'True' we respect that.
566 if chunked:
567 if "transfer-encoding" not in header_keys:
568 self.putheader("Transfer-Encoding", "chunked")
569 else:
570 # Detect whether a framing mechanism is already in use. If so
571 # we respect that value, otherwise we pick chunked vs content-length
572 # depending on the type of 'body'.
573 if "content-length" in header_keys:
574 chunked = False
575 elif "transfer-encoding" in header_keys:
576 chunked = True
578 # Otherwise we go off the recommendation of 'body_to_chunks()'.
579 else:
580 chunked = False
581 if content_length is None:
582 if chunks is not None:
583 chunked = True
584 self.putheader("Transfer-Encoding", "chunked")
585 else:
586 self.putheader("Content-Length", str(content_length))
588 # Now that framing headers are out of the way we send all the other headers.
589 if "user-agent" not in header_keys:
590 self.putheader("User-Agent", _get_default_user_agent())
591 for header, value in headers.items():
592 self.putheader(header, value)
593 self.endheaders()
595 # If we're given a body we start sending that in chunks.
596 if chunks is not None:
597 for chunk in chunks:
598 # Sending empty chunks isn't allowed for TE: chunked
599 # as it indicates the end of the body.
600 if not chunk:
601 continue
602 if isinstance(chunk, str):
603 chunk = chunk.encode("utf-8")
604 if chunked:
605 self.send(b"%x\r\n%b\r\n" % (len(chunk), chunk))
606 else:
607 self.send(chunk)
609 # Regardless of whether we have a body or not, if we're in
610 # chunked mode we want to send an explicit empty chunk.
611 if chunked:
612 self.send(b"0\r\n\r\n")
614 def request_chunked(
615 self,
616 method: str,
617 url: str,
618 body: _TYPE_BODY | None = None,
619 headers: typing.Mapping[str, str] | None = None,
620 ) -> None:
621 """
622 Alternative to the common request method, which sends the
623 body with chunked encoding and not as one block
624 """
625 warnings.warn(
626 "HTTPConnection.request_chunked() is deprecated and will be removed "
627 "in urllib3 v3.0. Instead use HTTPConnection.request(..., chunked=True).",
628 category=FutureWarning,
629 stacklevel=2,
630 )
631 self.request(method, url, body=body, headers=headers, chunked=True)
633 def getresponse( # type: ignore[override]
634 self,
635 ) -> HTTPResponse:
636 """
637 Get the response from the server.
639 If the HTTPConnection is in the correct state, returns an instance of HTTPResponse or of whatever object is returned by the response_class variable.
641 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.
642 """
643 # Raise the same error as http.client.HTTPConnection
644 if self._response_options is None:
645 raise ResponseNotReady()
647 # Reset this attribute for being used again.
648 resp_options = self._response_options
649 self._response_options = None
651 # Since the connection's timeout value may have been updated
652 # we need to set the timeout on the socket.
653 self.sock.settimeout(self.timeout)
655 # This is needed here to avoid circular import errors
656 from .response import HTTPResponse
658 # Save a reference to the shutdown function before ownership is passed
659 # to httplib_response
660 # TODO should we implement it everywhere?
661 _shutdown = getattr(self.sock, "shutdown", None)
663 # Get the response from http.client.HTTPConnection
664 httplib_response = super().getresponse()
666 try:
667 assert_header_parsing(httplib_response.msg)
668 except (HeaderParsingError, TypeError) as hpe:
669 log.warning(
670 "Failed to parse headers (url=%s): %s",
671 _url_from_connection(self, resp_options.request_url),
672 hpe,
673 exc_info=True,
674 )
676 header_items = _normalize_header_values(httplib_response.msg)
677 headers = HTTPHeaderDict(header_items)
679 response = HTTPResponse(
680 body=httplib_response,
681 headers=headers,
682 status=httplib_response.status,
683 version=httplib_response.version,
684 version_string=getattr(self, "_http_vsn_str", "HTTP/?"),
685 reason=httplib_response.reason,
686 preload_content=resp_options.preload_content,
687 decode_content=resp_options.decode_content,
688 original_response=httplib_response,
689 enforce_content_length=resp_options.enforce_content_length,
690 request_method=resp_options.request_method,
691 request_url=resp_options.request_url,
692 sock_shutdown=_shutdown,
693 )
694 return response
697class HTTPSConnection(HTTPConnection):
698 """
699 Many of the parameters to this constructor are passed to the underlying SSL
700 socket by means of :py:func:`urllib3.util.ssl_wrap_socket`.
701 """
703 default_port = port_by_scheme["https"] # type: ignore[misc]
705 cert_reqs: int | str | None = None
706 ca_certs: str | None = None
707 ca_cert_dir: str | None = None
708 ca_cert_data: None | str | bytes = None
709 ssl_version: int | str | None = None
710 ssl_minimum_version: int | None = None
711 ssl_maximum_version: int | None = None
712 assert_fingerprint: str | None = None
713 _connect_callback: typing.Callable[..., None] | None = None
715 def __init__(
716 self,
717 host: str,
718 port: int | None = None,
719 *,
720 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
721 source_address: tuple[str, int] | None = None,
722 blocksize: int = 16384,
723 socket_options: None | (
724 connection._TYPE_SOCKET_OPTIONS
725 ) = HTTPConnection.default_socket_options,
726 proxy: Url | None = None,
727 proxy_config: ProxyConfig | None = None,
728 cert_reqs: int | str | None = None,
729 assert_hostname: None | str | typing.Literal[False] = None,
730 assert_fingerprint: str | None = None,
731 server_hostname: str | None = None,
732 ssl_context: ssl.SSLContext | None = None,
733 ca_certs: str | None = None,
734 ca_cert_dir: str | None = None,
735 ca_cert_data: None | str | bytes = None,
736 ssl_minimum_version: int | None = None,
737 ssl_maximum_version: int | None = None,
738 ssl_version: int | str | None = None, # Deprecated
739 cert_file: str | None = None,
740 key_file: str | None = None,
741 key_password: str | None = None,
742 ) -> None:
743 super().__init__(
744 host,
745 port=port,
746 timeout=timeout,
747 source_address=source_address,
748 blocksize=blocksize,
749 socket_options=socket_options,
750 proxy=proxy,
751 proxy_config=proxy_config,
752 )
754 self.key_file = key_file
755 self.cert_file = cert_file
756 self.key_password = key_password
757 self.ssl_context = ssl_context
758 self.server_hostname = server_hostname
759 self.assert_hostname = assert_hostname
760 self.assert_fingerprint = assert_fingerprint
761 self.ssl_version = ssl_version
762 self.ssl_minimum_version = ssl_minimum_version
763 self.ssl_maximum_version = ssl_maximum_version
764 self.ca_certs = ca_certs and os.path.expanduser(ca_certs)
765 self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir)
766 self.ca_cert_data = ca_cert_data
768 # cert_reqs depends on ssl_context so calculate last.
769 if cert_reqs is None:
770 if self.ssl_context is not None:
771 cert_reqs = self.ssl_context.verify_mode
772 else:
773 cert_reqs = resolve_cert_reqs(None)
774 self.cert_reqs = cert_reqs
775 self._connect_callback = None
777 def set_cert(
778 self,
779 key_file: str | None = None,
780 cert_file: str | None = None,
781 cert_reqs: int | str | None = None,
782 key_password: str | None = None,
783 ca_certs: str | None = None,
784 assert_hostname: None | str | typing.Literal[False] = None,
785 assert_fingerprint: str | None = None,
786 ca_cert_dir: str | None = None,
787 ca_cert_data: None | str | bytes = None,
788 ) -> None:
789 """
790 This method should only be called once, before the connection is used.
791 """
792 warnings.warn(
793 "HTTPSConnection.set_cert() is deprecated and will be removed "
794 "in urllib3 v3.0. Instead provide the parameters to the "
795 "HTTPSConnection constructor.",
796 category=FutureWarning,
797 stacklevel=2,
798 )
800 # If cert_reqs is not provided we'll assume CERT_REQUIRED unless we also
801 # have an SSLContext object in which case we'll use its verify_mode.
802 if cert_reqs is None:
803 if self.ssl_context is not None:
804 cert_reqs = self.ssl_context.verify_mode
805 else:
806 cert_reqs = resolve_cert_reqs(None)
808 self.key_file = key_file
809 self.cert_file = cert_file
810 self.cert_reqs = cert_reqs
811 self.key_password = key_password
812 self.assert_hostname = assert_hostname
813 self.assert_fingerprint = assert_fingerprint
814 self.ca_certs = ca_certs and os.path.expanduser(ca_certs)
815 self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir)
816 self.ca_cert_data = ca_cert_data
818 def connect(self) -> None:
819 # Today we don't need to be doing this step before the /actual/ socket
820 # connection, however in the future we'll need to decide whether to
821 # create a new socket or re-use an existing "shared" socket as a part
822 # of the HTTP/2 handshake dance.
823 if self._tunnel_host is not None and self._tunnel_port is not None:
824 probe_http2_host = self._tunnel_host
825 probe_http2_port = self._tunnel_port
826 else:
827 probe_http2_host = self.host
828 probe_http2_port = self.port
830 # Check if the target origin supports HTTP/2.
831 # If the value comes back as 'None' it means that the current thread
832 # is probing for HTTP/2 support. Otherwise, we're waiting for another
833 # probe to complete, or we get a value right away.
834 target_supports_http2: bool | None
835 if "h2" in ssl_.ALPN_PROTOCOLS:
836 target_supports_http2 = http2_probe.acquire_and_get(
837 host=probe_http2_host, port=probe_http2_port
838 )
839 else:
840 # If HTTP/2 isn't going to be offered it doesn't matter if
841 # the target supports HTTP/2. Don't want to make a probe.
842 target_supports_http2 = False
844 if self._connect_callback is not None:
845 self._connect_callback(
846 "before connect",
847 thread_id=threading.get_ident(),
848 target_supports_http2=target_supports_http2,
849 )
851 try:
852 sock: socket.socket | ssl.SSLSocket
853 self.sock = sock = self._new_conn()
854 server_hostname: str = self.host
855 tls_in_tls = False
857 # Do we need to establish a tunnel?
858 if self.proxy_is_tunneling:
859 # We're tunneling to an HTTPS origin so need to do TLS-in-TLS.
860 if self._tunnel_scheme == "https":
861 # _connect_tls_proxy will verify and assign proxy_is_verified
862 self.sock = sock = self._connect_tls_proxy(self.host, sock)
863 tls_in_tls = True
864 elif self._tunnel_scheme == "http":
865 self.proxy_is_verified = False
867 # If we're tunneling it means we're connected to our proxy.
868 self._has_connected_to_proxy = True
870 self._tunnel()
871 # Override the host with the one we're requesting data from.
872 server_hostname = typing.cast(str, self._tunnel_host)
874 if self.server_hostname is not None:
875 server_hostname = self.server_hostname
877 is_time_off = datetime.date.today() < RECENT_DATE
878 if is_time_off:
879 warnings.warn(
880 (
881 f"System time is way off (before {RECENT_DATE}). This will probably "
882 "lead to SSL verification errors"
883 ),
884 SystemTimeWarning,
885 )
887 # Remove trailing '.' from fqdn hostnames to allow certificate validation
888 server_hostname_rm_dot = server_hostname.rstrip(".")
890 # Forwarding proxies should use proxy SSL context for
891 # wrapping since that's the connection being established,
892 # whereas tunneling proxies should use the connection's SSL
893 # context.
894 # However, for backwards compatibility reasons, if the proxy
895 # is forwarding but no proxy SSL context is provided, we
896 # fall back to using the connection's SSL context until
897 # urllib3 v3.0. Appropriate warning is emitted in
898 # ``ProxyManager.__init__``.
899 wrapped_socket: ssl.SSLSocket | SSLTransport
900 if self.proxy_is_forwarding and self.proxy_config is not None:
901 wrapped_socket = self._connect_tls_proxy(self.host, sock)
902 is_verified = self.proxy_is_verified is True
903 else:
904 wrapped_socket, is_verified = _ssl_wrap_socket_and_match_hostname(
905 sock=sock,
906 cert_reqs=self.cert_reqs,
907 ssl_version=self.ssl_version,
908 ssl_minimum_version=self.ssl_minimum_version,
909 ssl_maximum_version=self.ssl_maximum_version,
910 ca_certs=self.ca_certs,
911 ca_cert_dir=self.ca_cert_dir,
912 ca_cert_data=self.ca_cert_data,
913 cert_file=self.cert_file,
914 key_file=self.key_file,
915 key_password=self.key_password,
916 server_hostname=server_hostname_rm_dot,
917 ssl_context=self.ssl_context,
918 tls_in_tls=tls_in_tls,
919 assert_hostname=self.assert_hostname,
920 assert_fingerprint=self.assert_fingerprint,
921 )
922 self.sock = wrapped_socket
924 # If an error occurs during connection/handshake we may need to release
925 # our lock so another connection can probe the origin.
926 except BaseException:
927 if self._connect_callback is not None:
928 self._connect_callback(
929 "after connect failure",
930 thread_id=threading.get_ident(),
931 target_supports_http2=target_supports_http2,
932 )
934 if target_supports_http2 is None:
935 http2_probe.set_and_release(
936 host=probe_http2_host, port=probe_http2_port, supports_http2=None
937 )
938 raise
940 # If this connection doesn't know if the origin supports HTTP/2
941 # we report back to the HTTP/2 probe our result.
942 if target_supports_http2 is None:
943 supports_http2 = wrapped_socket.selected_alpn_protocol() == "h2"
944 http2_probe.set_and_release(
945 host=probe_http2_host,
946 port=probe_http2_port,
947 supports_http2=supports_http2,
948 )
950 # Forwarding proxies can never have a verified target since
951 # the proxy is the one doing the verification. Should instead
952 # use a CONNECT tunnel in order to verify the target.
953 # See: https://github.com/urllib3/urllib3/issues/3267.
954 if self.proxy_is_forwarding:
955 self.is_verified = False
956 else:
957 self.is_verified = is_verified
959 # If there's a proxy to be connected to we are fully connected.
960 # This is set twice (once above and here) due to forwarding proxies
961 # not using tunnelling.
962 self._has_connected_to_proxy = bool(self.proxy)
964 # Set `self.proxy_is_verified` unless it's already set while
965 # establishing a tunnel.
966 if self._has_connected_to_proxy and self.proxy_is_verified is None:
967 self.proxy_is_verified = is_verified
969 def _connect_tls_proxy(self, hostname: str, sock: socket.socket) -> ssl.SSLSocket:
970 """
971 Establish a TLS connection to the proxy using proxy-specific policy.
972 """
973 proxy_config = typing.cast(ProxyConfig, self.proxy_config)
974 proxy_ssl_context = proxy_config.ssl_context
976 ssl_context: ssl.SSLContext | None
977 cert_reqs: int | str | None
978 if proxy_ssl_context is not None:
979 # Prefer the proxy's cert policy for the proxy connection
980 ssl_context = proxy_ssl_context
981 cert_reqs = proxy_ssl_context.verify_mode
982 ca_certs = None
983 ca_cert_dir = None
984 ca_cert_data = None
985 ssl_version = None
986 ssl_minimum_version = None
987 ssl_maximum_version = None
988 else:
989 # Otherwise we inherit the pool's cert policies
990 ssl_context = self.ssl_context if self.proxy_is_forwarding else None
991 cert_reqs = self.cert_reqs
992 ca_certs = self.ca_certs
993 ca_cert_dir = self.ca_cert_dir
994 ca_cert_data = self.ca_cert_data
995 ssl_version = self.ssl_version
996 ssl_minimum_version = self.ssl_minimum_version
997 ssl_maximum_version = self.ssl_maximum_version
999 sock_and_verified = _ssl_wrap_socket_and_match_hostname(
1000 sock,
1001 cert_reqs=cert_reqs,
1002 ssl_version=ssl_version,
1003 ssl_minimum_version=ssl_minimum_version,
1004 ssl_maximum_version=ssl_maximum_version,
1005 ca_certs=ca_certs,
1006 ca_cert_dir=ca_cert_dir,
1007 ca_cert_data=ca_cert_data,
1008 server_hostname=hostname,
1009 ssl_context=ssl_context,
1010 assert_hostname=proxy_config.assert_hostname,
1011 assert_fingerprint=proxy_config.assert_fingerprint,
1012 # Features that aren't implemented for proxies yet:
1013 cert_file=None,
1014 key_file=None,
1015 key_password=None,
1016 tls_in_tls=False,
1017 )
1018 self.proxy_is_verified = sock_and_verified.is_verified
1019 return sock_and_verified.socket # type: ignore[return-value]
1022class _WrappedAndVerifiedSocket(typing.NamedTuple):
1023 """
1024 Wrapped socket and whether the connection is
1025 verified after the TLS handshake
1026 """
1028 socket: ssl.SSLSocket | SSLTransport
1029 is_verified: bool
1032def _ssl_wrap_socket_and_match_hostname(
1033 sock: socket.socket,
1034 *,
1035 cert_reqs: None | str | int,
1036 ssl_version: None | str | int,
1037 ssl_minimum_version: int | None,
1038 ssl_maximum_version: int | None,
1039 cert_file: str | None,
1040 key_file: str | None,
1041 key_password: str | None,
1042 ca_certs: str | None,
1043 ca_cert_dir: str | None,
1044 ca_cert_data: None | str | bytes,
1045 assert_hostname: None | str | typing.Literal[False],
1046 assert_fingerprint: str | None,
1047 server_hostname: str | None,
1048 ssl_context: ssl.SSLContext | None,
1049 tls_in_tls: bool = False,
1050) -> _WrappedAndVerifiedSocket:
1051 """Logic for constructing an SSLContext from all TLS parameters, passing
1052 that down into ssl_wrap_socket, and then doing certificate verification
1053 either via hostname or fingerprint. This function exists to guarantee
1054 that both proxies and targets have the same behavior when connecting via TLS.
1055 """
1056 default_ssl_context = False
1057 if ssl_context is None:
1058 default_ssl_context = True
1059 context = create_urllib3_context(
1060 ssl_version=resolve_ssl_version(ssl_version),
1061 ssl_minimum_version=ssl_minimum_version,
1062 ssl_maximum_version=ssl_maximum_version,
1063 cert_reqs=resolve_cert_reqs(cert_reqs),
1064 )
1065 else:
1066 context = ssl_context
1068 context.verify_mode = resolve_cert_reqs(cert_reqs)
1070 # In some cases, we want to verify hostnames ourselves
1071 if (
1072 # `ssl` can't verify fingerprints or alternate hostnames
1073 assert_fingerprint
1074 or assert_hostname
1075 # assert_hostname can be set to False to disable hostname checking
1076 or assert_hostname is False
1077 # We still support OpenSSL 1.0.2, which prevents us from verifying
1078 # hostnames easily: https://github.com/pyca/pyopenssl/pull/933
1079 or ssl_.IS_PYOPENSSL
1080 or not ssl_.HAS_NEVER_CHECK_COMMON_NAME
1081 ):
1082 context.check_hostname = False
1084 # Try to load OS default certs if none are given. We need to do the hasattr() check
1085 # for custom pyOpenSSL SSLContext objects because they don't support
1086 # load_default_certs().
1087 if (
1088 not ca_certs
1089 and not ca_cert_dir
1090 and not ca_cert_data
1091 and default_ssl_context
1092 and hasattr(context, "load_default_certs")
1093 ):
1094 context.load_default_certs()
1096 # Ensure that IPv6 addresses are in the proper format and don't have a
1097 # scope ID. Python's SSL module fails to recognize scoped IPv6 addresses
1098 # and interprets them as DNS hostnames.
1099 if server_hostname is not None:
1100 normalized = server_hostname.strip("[]")
1101 if "%" in normalized:
1102 normalized = normalized[: normalized.rfind("%")]
1103 if is_ipaddress(normalized):
1104 server_hostname = normalized
1106 ssl_sock = ssl_wrap_socket(
1107 sock=sock,
1108 keyfile=key_file,
1109 certfile=cert_file,
1110 key_password=key_password,
1111 ca_certs=ca_certs,
1112 ca_cert_dir=ca_cert_dir,
1113 ca_cert_data=ca_cert_data,
1114 server_hostname=server_hostname,
1115 ssl_context=context,
1116 tls_in_tls=tls_in_tls,
1117 )
1119 try:
1120 if assert_fingerprint:
1121 _assert_fingerprint(
1122 ssl_sock.getpeercert(binary_form=True), assert_fingerprint
1123 )
1124 elif (
1125 context.verify_mode != ssl.CERT_NONE
1126 and not context.check_hostname
1127 and assert_hostname is not False
1128 ):
1129 cert: _TYPE_PEER_CERT_RET_DICT = ssl_sock.getpeercert() # type: ignore[assignment]
1131 # Need to signal to our match_hostname whether to use 'commonName' or not.
1132 # If we're using our own constructed SSLContext we explicitly set 'False'
1133 # because PyPy hard-codes 'True' from SSLContext.hostname_checks_common_name.
1134 if default_ssl_context:
1135 hostname_checks_common_name = False
1136 else:
1137 hostname_checks_common_name = (
1138 getattr(context, "hostname_checks_common_name", False) or False
1139 )
1141 _match_hostname(
1142 cert,
1143 assert_hostname or server_hostname, # type: ignore[arg-type]
1144 hostname_checks_common_name,
1145 )
1147 return _WrappedAndVerifiedSocket(
1148 socket=ssl_sock,
1149 is_verified=context.verify_mode == ssl.CERT_REQUIRED
1150 or bool(assert_fingerprint),
1151 )
1152 except BaseException:
1153 ssl_sock.close()
1154 raise
1157def _match_hostname(
1158 cert: _TYPE_PEER_CERT_RET_DICT | None,
1159 asserted_hostname: str,
1160 hostname_checks_common_name: bool = False,
1161) -> None:
1162 # Our upstream implementation of ssl.match_hostname()
1163 # only applies this normalization to IP addresses so it doesn't
1164 # match DNS SANs so we do the same thing!
1165 stripped_hostname = asserted_hostname.strip("[]")
1166 if is_ipaddress(stripped_hostname):
1167 asserted_hostname = stripped_hostname
1169 try:
1170 match_hostname(cert, asserted_hostname, hostname_checks_common_name)
1171 except CertificateError as e:
1172 log.warning(
1173 "Certificate did not match expected hostname: %s. Certificate: %s",
1174 asserted_hostname,
1175 cert,
1176 )
1177 # Add cert to exception and reraise so client code can inspect
1178 # the cert when catching the exception, if they want to
1179 e._peer_cert = cert # type: ignore[attr-defined]
1180 raise
1183def _wrap_proxy_error(err: Exception, proxy_scheme: str | None) -> ProxyError:
1184 # Look for the phrase 'wrong version number', if found
1185 # then we should warn the user that we're very sure that
1186 # this proxy is HTTP-only and they have a configuration issue.
1187 error_normalized = " ".join(re.split("[^a-z]", str(err).lower()))
1188 is_likely_http_proxy = (
1189 "wrong version number" in error_normalized
1190 or "unknown protocol" in error_normalized
1191 or "record layer failure" in error_normalized
1192 )
1193 http_proxy_warning = (
1194 ". Your proxy appears to only use HTTP and not HTTPS, "
1195 "try changing your proxy URL to be HTTP. See: "
1196 "https://urllib3.readthedocs.io/en/latest/advanced-usage.html"
1197 "#https-proxy-error-http-proxy"
1198 )
1199 new_err = ProxyError(
1200 f"Unable to connect to proxy"
1201 f"{http_proxy_warning if is_likely_http_proxy and proxy_scheme == 'https' else ''}",
1202 err,
1203 )
1204 new_err.__cause__ = err
1205 return new_err
1208def _get_default_user_agent() -> str:
1209 return f"python-urllib3/{__version__}"
1212class DummyConnection:
1213 """Used to detect a failed ConnectionCls import."""
1216if not ssl:
1217 HTTPSConnection = DummyConnection # type: ignore[misc, assignment] # noqa: F811
1220VerifiedHTTPSConnection = HTTPSConnection
1223def _url_from_connection(
1224 conn: HTTPConnection | HTTPSConnection, path: str | None = None
1225) -> str:
1226 """Returns the URL from a given connection. This is mainly used for testing and logging."""
1228 scheme = "https" if isinstance(conn, HTTPSConnection) else "http"
1230 return Url(scheme=scheme, host=conn.host, port=conn.port, path=path).url