Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/connector.py: 19%
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
1import asyncio
2import functools
3import random
4import socket
5import sys
6import traceback
7import warnings
8from collections import OrderedDict, defaultdict, deque
9from collections.abc import Awaitable, Callable, Iterator, Sequence
10from contextlib import suppress
11from http import HTTPStatus
12from itertools import chain, cycle, islice
13from time import monotonic
14from types import TracebackType
15from typing import TYPE_CHECKING, Any, Literal, cast
17import aiohappyeyeballs
18from aiohappyeyeballs import AddrInfoType, SocketFactoryType
19from multidict import CIMultiDict
21from . import hdrs, helpers
22from .abc import AbstractResolver, ResolveResult
23from .client_exceptions import (
24 ClientConnectionError,
25 ClientConnectorCertificateError,
26 ClientConnectorDNSError,
27 ClientConnectorError,
28 ClientConnectorSSLError,
29 ClientHttpProxyError,
30 ClientProxyConnectionError,
31 InvalidUrlClientError,
32 ServerFingerprintMismatch,
33 UnixClientConnectorError,
34 cert_errors,
35 ssl_errors,
36)
37from .client_proto import ResponseHandler
38from .client_reqrep import (
39 SSL_ALLOWED_TYPES,
40 ClientRequest,
41 ClientRequestBase,
42 Fingerprint,
43)
44from .helpers import (
45 _SENTINEL,
46 HIGH_LEVEL_SCHEMA_SET,
47 ceil_timeout,
48 is_canonical_ipv4_address,
49 is_ip_address,
50 sentinel,
51 set_exception,
52 set_result,
53)
54from .log import client_logger
55from .resolver import DefaultResolver
57try:
58 import aiofastnet
59except ImportError:
60 aiofastnet = None # type: ignore[assignment]
63if sys.version_info >= (3, 12):
64 from collections.abc import Buffer
65else:
66 Buffer = "bytes | bytearray | memoryview[int] | memoryview[bytes]"
68try:
69 import ssl
71 SSLContext = ssl.SSLContext
72except ImportError: # pragma: no cover
73 ssl = None # type: ignore[assignment]
74 SSLContext = object # type: ignore[misc,assignment]
76NEEDS_CLEANUP_CLOSED = (3, 13, 0) <= sys.version_info < (
77 3,
78 13,
79 1,
80) or sys.version_info < (3, 12, 8)
81# Cleanup closed is no longer needed after https://github.com/python/cpython/pull/118960
82# which first appeared in Python 3.12.8 and 3.13.1
85__all__ = (
86 "BaseConnector",
87 "TCPConnector",
88 "UnixConnector",
89 "NamedPipeConnector",
90 "AddrInfoType",
91 "SocketFactoryType",
92)
95if TYPE_CHECKING:
96 from .client import ClientTimeout
97 from .client_reqrep import ConnectionKey
98 from .tracing import Trace
101async def create_connection(
102 loop: asyncio.AbstractEventLoop,
103 protocol_factory: Callable[[], ResponseHandler],
104 *,
105 ssl: SSLContext | None,
106 sock: socket.socket,
107 server_hostname: str | None,
108 ssl_shutdown_timeout: float | None = None,
109) -> tuple[asyncio.Transport, ResponseHandler]:
110 if aiofastnet is not None:
111 return await aiofastnet.create_connection(
112 loop,
113 protocol_factory,
114 ssl=ssl,
115 sock=sock,
116 server_hostname=server_hostname,
117 ssl_shutdown_timeout=ssl_shutdown_timeout,
118 )
119 else:
120 if sys.version_info >= (3, 11): # type: ignore[unreachable]
121 return await loop.create_connection(
122 protocol_factory,
123 ssl=ssl,
124 sock=sock,
125 server_hostname=server_hostname,
126 ssl_shutdown_timeout=ssl_shutdown_timeout,
127 )
128 else:
129 return await loop.create_connection(
130 protocol_factory,
131 ssl=ssl,
132 sock=sock,
133 server_hostname=server_hostname,
134 )
137async def start_tls(
138 loop: asyncio.AbstractEventLoop,
139 transport: asyncio.Transport,
140 protocol: ResponseHandler,
141 sslcontext: SSLContext,
142 *,
143 server_hostname: str | None,
144 ssl_handshake_timeout: float | None,
145 ssl_shutdown_timeout: float | None = None,
146) -> asyncio.BaseTransport | None:
147 if aiofastnet is not None:
148 return await aiofastnet.start_tls(
149 loop,
150 transport,
151 protocol,
152 sslcontext,
153 server_hostname=server_hostname,
154 ssl_handshake_timeout=ssl_handshake_timeout,
155 ssl_shutdown_timeout=ssl_shutdown_timeout,
156 )
157 else:
158 if sys.version_info >= (3, 11): # type: ignore[unreachable]
159 return await loop.start_tls(
160 transport,
161 protocol,
162 sslcontext,
163 server_hostname=server_hostname,
164 ssl_handshake_timeout=ssl_handshake_timeout,
165 ssl_shutdown_timeout=ssl_shutdown_timeout,
166 )
167 else:
168 return await loop.start_tls(
169 transport,
170 protocol,
171 sslcontext,
172 server_hostname=server_hostname,
173 ssl_handshake_timeout=ssl_handshake_timeout,
174 )
177class Connection:
178 """Represents a single connection."""
180 __slots__ = (
181 "_key",
182 "_connector",
183 "_loop",
184 "_protocol",
185 "_callbacks",
186 "_source_traceback",
187 )
189 def __init__(
190 self,
191 connector: "BaseConnector",
192 key: "ConnectionKey",
193 protocol: ResponseHandler,
194 loop: asyncio.AbstractEventLoop,
195 ) -> None:
196 self._key = key
197 self._connector = connector
198 self._loop = loop
199 self._protocol: ResponseHandler | None = protocol
200 self._callbacks: list[Callable[[], None]] = []
201 self._source_traceback = (
202 traceback.extract_stack(sys._getframe(1)) if loop.get_debug() else None
203 )
205 def __repr__(self) -> str:
206 return f"Connection<{self._key}>"
208 def __del__(self, _warnings: Any = warnings) -> None:
209 if self._protocol is not None:
210 _warnings.warn(
211 f"Unclosed connection {self!r}", ResourceWarning, source=self
212 )
213 if self._loop.is_closed():
214 return
216 self._connector._release(self._key, self._protocol, should_close=True)
218 context = {"client_connection": self, "message": "Unclosed connection"}
219 if self._source_traceback is not None:
220 context["source_traceback"] = self._source_traceback
221 self._loop.call_exception_handler(context)
223 def __bool__(self) -> Literal[True]:
224 """Force subclasses to not be falsy, to make checks simpler."""
225 return True
227 @property
228 def transport(self) -> asyncio.Transport | None:
229 if self._protocol is None:
230 return None
231 return self._protocol.transport
233 @property
234 def protocol(self) -> ResponseHandler | None:
235 return self._protocol
237 def add_callback(self, callback: Callable[[], None]) -> None:
238 if callback is not None:
239 self._callbacks.append(callback)
241 def _notify_release(self) -> None:
242 callbacks, self._callbacks = self._callbacks[:], []
244 for cb in callbacks:
245 with suppress(Exception):
246 cb()
248 def close(self) -> None:
249 self._notify_release()
251 if self._protocol is not None:
252 self._connector._release(self._key, self._protocol, should_close=True)
253 self._protocol = None
255 def release(self) -> None:
256 self._notify_release()
258 if self._protocol is not None:
259 self._connector._release(self._key, self._protocol)
260 self._protocol = None
262 @property
263 def closed(self) -> bool:
264 return self._protocol is None or not self._protocol.is_connected()
267class _ConnectTunnelConnection(Connection):
268 """Special connection wrapper for CONNECT tunnels that must never be pooled.
270 This connection wraps the proxy connection that will be upgraded with TLS.
271 It must never be released to the pool because:
272 1. Its 'closed' future will never complete, causing session.close() to hang
273 2. It represents an intermediate state, not a reusable connection
274 3. The real connection (with TLS) will be created separately
275 """
277 def release(self) -> None:
278 """Do nothing - don't pool or close the connection.
280 These connections are an intermediate state during the CONNECT tunnel
281 setup and will be cleaned up naturally after the TLS upgrade. If they
282 were to be pooled, they would never be properly closed, causing
283 session.close() to wait forever for their 'closed' future.
284 """
287class _TransportPlaceholder:
288 """placeholder for BaseConnector.connect function"""
290 __slots__ = ("closed", "transport")
292 def __init__(self, closed_future: asyncio.Future[Exception | None]) -> None:
293 """Initialize a placeholder for a transport."""
294 self.closed = closed_future
295 self.transport = None
297 def close(self) -> None:
298 """Close the placeholder."""
300 def abort(self) -> None:
301 """Abort the placeholder (does nothing)."""
304class BaseConnector:
305 """Base connector class.
307 keepalive_timeout - (optional) Keep-alive timeout.
308 force_close - Set to True to force close and do reconnect
309 after each request (and between redirects).
310 limit - The total number of simultaneous connections.
311 limit_per_host - Number of simultaneous connections to one host.
312 enable_cleanup_closed - Enables clean-up closed ssl transports.
313 Disabled by default.
314 timeout_ceil_threshold - Trigger ceiling of timeout values when
315 it's above timeout_ceil_threshold.
316 loop - Optional event loop.
317 """
319 _closed = True # prevent AttributeError in __del__ if ctor was failed
320 _source_traceback = None
322 # abort transport after 2 seconds (cleanup broken connections)
323 _cleanup_closed_period = 2.0
325 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET
327 def __init__(
328 self,
329 *,
330 keepalive_timeout: _SENTINEL | None | float = sentinel,
331 force_close: bool = False,
332 limit: int = 100,
333 limit_per_host: int = 0,
334 enable_cleanup_closed: bool = False,
335 timeout_ceil_threshold: float = 5,
336 ) -> None:
337 if force_close:
338 if keepalive_timeout is not None and keepalive_timeout is not sentinel:
339 raise ValueError(
340 "keepalive_timeout cannot be set if force_close is True"
341 )
342 else:
343 if keepalive_timeout is sentinel:
344 keepalive_timeout = 15.0
346 self._timeout_ceil_threshold = timeout_ceil_threshold
348 loop = asyncio.get_running_loop()
350 self._closed = False
351 if loop.get_debug():
352 self._source_traceback = traceback.extract_stack(sys._getframe(1))
354 # Connection pool of reusable connections.
355 # We use a deque to store connections because it has O(1) popleft()
356 # and O(1) append() operations to implement a FIFO queue.
357 self._conns: defaultdict[
358 ConnectionKey, deque[tuple[ResponseHandler, float]]
359 ] = defaultdict(deque)
360 self._limit = limit
361 self._limit_per_host = limit_per_host
362 self._acquired: set[ResponseHandler] = set()
363 self._acquired_per_host: defaultdict[ConnectionKey, set[ResponseHandler]] = (
364 defaultdict(set)
365 )
366 self._keepalive_timeout = cast(float, keepalive_timeout)
367 self._force_close = force_close
369 # {host_key: FIFO list of waiters}
370 # The FIFO is implemented with an OrderedDict with None keys because
371 # python does not have an ordered set.
372 self._waiters: defaultdict[
373 ConnectionKey, OrderedDict[asyncio.Future[None], None]
374 ] = defaultdict(OrderedDict)
376 self._loop = loop
377 self._factory = functools.partial(ResponseHandler, loop=loop)
379 # start keep-alive connection cleanup task
380 self._cleanup_handle: asyncio.TimerHandle | None = None
382 # start cleanup closed transports task
383 self._cleanup_closed_handle: asyncio.TimerHandle | None = None
385 if enable_cleanup_closed and not NEEDS_CLEANUP_CLOSED:
386 warnings.warn(
387 "enable_cleanup_closed ignored because "
388 "https://github.com/python/cpython/pull/118960 is fixed "
389 f"in Python version {sys.version_info}",
390 DeprecationWarning,
391 stacklevel=2,
392 )
393 enable_cleanup_closed = False
395 self._cleanup_closed_disabled = not enable_cleanup_closed
396 self._cleanup_closed_transports: list[asyncio.Transport | None] = []
398 self._placeholder_future: asyncio.Future[Exception | None] = (
399 loop.create_future()
400 )
401 self._placeholder_future.set_result(None)
402 self._cleanup_closed()
404 def __del__(self, _warnings: Any = warnings) -> None:
405 if self._closed:
406 return
407 if not self._conns:
408 return
410 conns = [repr(c) for c in self._conns.values()]
412 self._close_immediately()
414 _warnings.warn(f"Unclosed connector {self!r}", ResourceWarning, source=self)
415 context = {
416 "connector": self,
417 "connections": conns,
418 "message": "Unclosed connector",
419 }
420 if self._source_traceback is not None:
421 context["source_traceback"] = self._source_traceback
422 self._loop.call_exception_handler(context)
424 async def __aenter__(self) -> "BaseConnector":
425 return self
427 async def __aexit__(
428 self,
429 exc_type: type[BaseException] | None = None,
430 exc_value: BaseException | None = None,
431 exc_traceback: TracebackType | None = None,
432 ) -> None:
433 await self.close()
435 @property
436 def force_close(self) -> bool:
437 """Ultimately close connection on releasing if True."""
438 return self._force_close
440 @property
441 def limit(self) -> int:
442 """The total number for simultaneous connections.
444 If limit is 0 the connector has no limit.
445 The default limit size is 100.
446 """
447 return self._limit
449 @property
450 def limit_per_host(self) -> int:
451 """The limit for simultaneous connections to the same endpoint.
453 Endpoints are the same if they are have equal
454 (host, port, is_ssl) triple.
455 """
456 return self._limit_per_host
458 def _cleanup(self) -> None:
459 """Cleanup unused transports."""
460 if self._cleanup_handle:
461 self._cleanup_handle.cancel()
462 # _cleanup_handle should be unset, otherwise _release() will not
463 # recreate it ever!
464 self._cleanup_handle = None
466 now = monotonic()
467 timeout = self._keepalive_timeout
469 if self._conns:
470 connections = defaultdict(deque)
471 deadline = now - timeout
472 for key, conns in self._conns.items():
473 alive: deque[tuple[ResponseHandler, float]] = deque()
474 for proto, use_time in conns:
475 if proto.is_connected() and use_time - deadline >= 0:
476 alive.append((proto, use_time))
477 continue
478 transport = proto.transport
479 proto.close()
480 if not self._cleanup_closed_disabled and key.is_ssl:
481 self._cleanup_closed_transports.append(transport)
483 if alive:
484 connections[key] = alive
486 self._conns = connections
488 if self._conns:
489 self._cleanup_handle = helpers.weakref_handle(
490 self,
491 "_cleanup",
492 timeout,
493 self._loop,
494 timeout_ceil_threshold=self._timeout_ceil_threshold,
495 )
497 def _cleanup_closed(self) -> None:
498 """Double confirmation for transport close.
500 Some broken ssl servers may leave socket open without proper close.
501 """
502 if self._cleanup_closed_handle:
503 self._cleanup_closed_handle.cancel()
505 for transport in self._cleanup_closed_transports:
506 if transport is not None:
507 transport.abort()
509 self._cleanup_closed_transports = []
511 if not self._cleanup_closed_disabled:
512 self._cleanup_closed_handle = helpers.weakref_handle(
513 self,
514 "_cleanup_closed",
515 self._cleanup_closed_period,
516 self._loop,
517 timeout_ceil_threshold=self._timeout_ceil_threshold,
518 )
520 async def close(self, *, abort_ssl: bool = False) -> None:
521 """Close all opened transports.
523 :param abort_ssl: If True, SSL connections will be aborted immediately
524 without performing the shutdown handshake. This provides
525 faster cleanup at the cost of less graceful disconnection.
526 """
527 waiters = self._close_immediately(abort_ssl=abort_ssl)
528 if waiters:
529 results = await asyncio.gather(*waiters, return_exceptions=True)
530 for res in results:
531 if isinstance(res, Exception):
532 err_msg = "Error while closing connector: " + repr(res)
533 client_logger.debug(err_msg)
535 def _close_immediately(self, *, abort_ssl: bool = False) -> list[Awaitable[object]]:
536 waiters: list[Awaitable[object]] = []
538 if self._closed:
539 return waiters
541 self._closed = True
543 try:
544 if self._loop.is_closed():
545 return waiters
547 # cancel cleanup task
548 if self._cleanup_handle:
549 self._cleanup_handle.cancel()
551 # cancel cleanup close task
552 if self._cleanup_closed_handle:
553 self._cleanup_closed_handle.cancel()
555 for data in self._conns.values():
556 for proto, _ in data:
557 if (
558 abort_ssl
559 and proto.transport
560 and proto.transport.get_extra_info("sslcontext") is not None
561 ):
562 proto.abort()
563 else:
564 proto.close()
565 if closed := proto.closed:
566 waiters.append(closed)
568 for proto in self._acquired:
569 if (
570 abort_ssl
571 and proto.transport
572 and proto.transport.get_extra_info("sslcontext") is not None
573 ):
574 proto.abort()
575 else:
576 proto.close()
577 if closed := proto.closed:
578 waiters.append(closed)
580 # TODO (A.Yushovskiy, 24-May-2019) collect transp. closing futures
581 for transport in self._cleanup_closed_transports:
582 if transport is not None:
583 transport.abort()
585 return waiters
587 finally:
588 self._conns.clear()
589 self._acquired.clear()
590 for keyed_waiters in self._waiters.values():
591 for keyed_waiter in keyed_waiters:
592 keyed_waiter.cancel()
593 self._waiters.clear()
594 self._cleanup_handle = None
595 self._cleanup_closed_transports.clear()
596 self._cleanup_closed_handle = None
598 @property
599 def closed(self) -> bool:
600 """Is connector closed.
602 A readonly property.
603 """
604 return self._closed
606 def _available_connections(self, key: "ConnectionKey") -> int:
607 """
608 Return number of available connections.
610 The limit, limit_per_host and the connection key are taken into account.
612 If it returns less than 1 means that there are no connections
613 available.
614 """
615 # check total available connections
616 # If there are no limits, this will always return 1
617 total_remain = 1
619 if self._limit and (total_remain := self._limit - len(self._acquired)) <= 0:
620 return total_remain
622 # check limit per host
623 if host_remain := self._limit_per_host:
624 if acquired := self._acquired_per_host.get(key):
625 host_remain -= len(acquired)
626 if total_remain > host_remain:
627 return host_remain
629 return total_remain
631 def _update_proxy_auth_header_and_build_proxy_req(
632 self, req: ClientRequest
633 ) -> ClientRequestBase:
634 """Set Proxy-Authorization header for non-SSL proxy requests and builds the proxy request for SSL proxy requests."""
635 url = req.proxy
636 assert url is not None
637 headers = req.proxy_headers or CIMultiDict[str]()
638 headers[hdrs.HOST] = req.headers[hdrs.HOST]
639 proxy_req = ClientRequestBase(
640 hdrs.METH_GET,
641 url,
642 headers=headers,
643 loop=self._loop,
644 ssl=req.ssl,
645 )
646 if not req.is_ssl():
647 # For non-SSL proxies the request goes directly through the proxy,
648 # so any Proxy-Authorization belongs on the request itself, not on
649 # the synthetic proxy request used for SSL CONNECT.
650 proxy_auth = proxy_req.headers.pop(hdrs.PROXY_AUTHORIZATION, None)
651 if proxy_auth is not None:
652 req.headers[hdrs.PROXY_AUTHORIZATION] = proxy_auth
653 return proxy_req
655 async def connect(
656 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
657 ) -> Connection:
658 """Get from pool or create new connection."""
659 key = req.connection_key
660 if (conn := await self._get(key, traces)) is not None:
661 # If we do not have to wait and we can get a connection from the pool
662 # we can avoid the timeout ceil logic and directly return the connection
663 if req.proxy:
664 self._update_proxy_auth_header_and_build_proxy_req(req)
665 return conn
667 async with ceil_timeout(timeout.connect, timeout.ceil_threshold):
668 if self._available_connections(key) <= 0:
669 await self._wait_for_available_connection(key, traces)
670 if (conn := await self._get(key, traces)) is not None:
671 if req.proxy:
672 self._update_proxy_auth_header_and_build_proxy_req(req)
673 return conn
675 placeholder = cast(
676 ResponseHandler, _TransportPlaceholder(self._placeholder_future)
677 )
678 self._acquired.add(placeholder)
679 if self._limit_per_host:
680 self._acquired_per_host[key].add(placeholder)
682 try:
683 # Traces are done inside the try block to ensure that the
684 # that the placeholder is still cleaned up if an exception
685 # is raised.
686 if traces:
687 for trace in traces:
688 await trace.send_connection_create_start()
689 proto = await self._create_connection(req, traces, timeout)
690 if traces:
691 for trace in traces:
692 await trace.send_connection_create_end()
693 except BaseException:
694 self._release_acquired(key, placeholder)
695 raise
696 else:
697 if self._closed:
698 proto.close()
699 raise ClientConnectionError("Connector is closed.")
701 # The connection was successfully created, drop the placeholder
702 # and add the real connection to the acquired set. There should
703 # be no awaits after the proto is added to the acquired set
704 # to ensure that the connection is not left in the acquired set
705 # on cancellation.
706 self._acquired.remove(placeholder)
707 self._acquired.add(proto)
708 if self._limit_per_host:
709 acquired_per_host = self._acquired_per_host[key]
710 acquired_per_host.remove(placeholder)
711 acquired_per_host.add(proto)
712 return Connection(self, key, proto, self._loop)
714 async def _wait_for_available_connection(
715 self, key: "ConnectionKey", traces: list["Trace"]
716 ) -> None:
717 """Wait for an available connection slot."""
718 # We loop here because there is a race between
719 # the connection limit check and the connection
720 # being acquired. If the connection is acquired
721 # between the check and the await statement, we
722 # need to loop again to check if the connection
723 # slot is still available.
724 attempts = 0
725 while True:
726 fut: asyncio.Future[None] = self._loop.create_future()
727 keyed_waiters = self._waiters[key]
728 keyed_waiters[fut] = None
729 if attempts:
730 # If we have waited before, we need to move the waiter
731 # to the front of the queue as otherwise we might get
732 # starved and hit the timeout.
733 keyed_waiters.move_to_end(fut, last=False)
735 try:
736 # Traces happen in the try block to ensure that the
737 # the waiter is still cleaned up if an exception is raised.
738 if traces:
739 for trace in traces:
740 await trace.send_connection_queued_start()
741 await fut
742 if traces:
743 for trace in traces:
744 await trace.send_connection_queued_end()
745 finally:
746 # pop the waiter from the queue if its still
747 # there and not already removed by _release_waiter
748 keyed_waiters.pop(fut, None)
749 if not self._waiters.get(key, True):
750 del self._waiters[key]
752 if self._available_connections(key) > 0:
753 break
754 attempts += 1
756 async def _get(
757 self, key: "ConnectionKey", traces: list["Trace"]
758 ) -> Connection | None:
759 """Get next reusable connection for the key or None.
761 The connection will be marked as acquired.
762 """
763 if (conns := self._conns.get(key)) is None:
764 return None
766 t1 = monotonic()
767 while conns:
768 proto, t0 = conns.popleft()
769 # We will we reuse the connection if its connected and
770 # the keepalive timeout has not been exceeded
771 if proto.is_connected() and t1 - t0 <= self._keepalive_timeout:
772 if not conns:
773 # The very last connection was reclaimed: drop the key
774 del self._conns[key]
775 self._acquired.add(proto)
776 if self._limit_per_host:
777 self._acquired_per_host[key].add(proto)
778 if traces:
779 for trace in traces:
780 try:
781 await trace.send_connection_reuseconn()
782 except BaseException:
783 self._release_acquired(key, proto)
784 raise
785 return Connection(self, key, proto, self._loop)
787 # Connection cannot be reused, close it
788 transport = proto.transport
789 proto.close()
790 # only for SSL transports
791 if not self._cleanup_closed_disabled and key.is_ssl:
792 self._cleanup_closed_transports.append(transport)
794 # No more connections: drop the key
795 del self._conns[key]
796 return None
798 def _release_waiter(self) -> None:
799 """
800 Iterates over all waiters until one to be released is found.
802 The one to be released is not finished and
803 belongs to a host that has available connections.
804 """
805 if not self._waiters:
806 return
808 # Having the dict keys ordered this avoids to iterate
809 # at the same order at each call.
810 queues = list(self._waiters)
811 random.shuffle(queues)
813 for key in queues:
814 if self._available_connections(key) < 1:
815 continue
817 waiters = self._waiters[key]
818 while waiters:
819 waiter, _ = waiters.popitem(last=False)
820 if not waiter.done():
821 waiter.set_result(None)
822 return
824 def _release_acquired(self, key: "ConnectionKey", proto: ResponseHandler) -> None:
825 """Release acquired connection."""
826 if self._closed:
827 # acquired connection is already released on connector closing
828 return
830 self._acquired.discard(proto)
831 if self._limit_per_host and (conns := self._acquired_per_host.get(key)):
832 conns.discard(proto)
833 if not conns:
834 del self._acquired_per_host[key]
835 self._release_waiter()
837 def _release(
838 self,
839 key: "ConnectionKey",
840 protocol: ResponseHandler,
841 *,
842 should_close: bool = False,
843 ) -> None:
844 if self._closed:
845 # acquired connection is already released on connector closing
846 return
848 self._release_acquired(key, protocol)
850 if self._force_close or should_close or protocol.should_close:
851 transport = protocol.transport
852 protocol.close()
853 if key.is_ssl and not self._cleanup_closed_disabled:
854 self._cleanup_closed_transports.append(transport)
855 return
857 self._conns[key].append((protocol, monotonic()))
859 if self._cleanup_handle is None:
860 self._cleanup_handle = helpers.weakref_handle(
861 self,
862 "_cleanup",
863 self._keepalive_timeout,
864 self._loop,
865 timeout_ceil_threshold=self._timeout_ceil_threshold,
866 )
868 async def _create_connection(
869 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
870 ) -> ResponseHandler:
871 raise NotImplementedError()
874class _DNSCacheTable:
875 def __init__(self, ttl: float | None = None, max_size: int = 1000) -> None:
876 self._addrs_rr: OrderedDict[
877 tuple[str, int], tuple[Iterator[ResolveResult], int]
878 ] = OrderedDict()
879 self._timestamps: dict[tuple[str, int], float] = {}
880 self._ttl = ttl
881 self._max_size = max_size
883 def __contains__(self, host: object) -> bool:
884 return host in self._addrs_rr
886 def add(self, key: tuple[str, int], addrs: list[ResolveResult]) -> None:
887 if key in self._addrs_rr:
888 self._addrs_rr.move_to_end(key)
890 self._addrs_rr[key] = (cycle(addrs), len(addrs))
892 if self._ttl is not None:
893 self._timestamps[key] = monotonic()
895 if len(self._addrs_rr) > self._max_size:
896 oldest_key, _ = self._addrs_rr.popitem(last=False)
897 self._timestamps.pop(oldest_key, None)
899 def remove(self, key: tuple[str, int]) -> None:
900 self._addrs_rr.pop(key, None)
901 self._timestamps.pop(key, None)
903 def clear(self) -> None:
904 self._addrs_rr.clear()
905 self._timestamps.clear()
907 def next_addrs(self, key: tuple[str, int]) -> list[ResolveResult]:
908 loop, length = self._addrs_rr[key]
909 addrs = list(islice(loop, length))
910 # Consume one more element to shift internal state of `cycle`
911 next(loop)
912 self._addrs_rr.move_to_end(key)
913 return addrs
915 def expired(self, key: tuple[str, int]) -> bool:
916 if self._ttl is None:
917 return False
919 return self._timestamps[key] + self._ttl < monotonic()
922def _make_ssl_context(verified: bool) -> SSLContext:
923 """Create SSL context.
925 This method is not async-friendly and should be called from a thread
926 because it will load certificates from disk and do other blocking I/O.
927 """
928 if ssl is None:
929 # No ssl support
930 return None # type: ignore[unreachable]
931 if verified:
932 sslcontext = ssl.create_default_context()
933 else:
934 sslcontext = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
935 sslcontext.options |= ssl.OP_NO_SSLv2
936 sslcontext.options |= ssl.OP_NO_SSLv3
937 sslcontext.check_hostname = False
938 sslcontext.verify_mode = ssl.CERT_NONE
939 sslcontext.options |= ssl.OP_NO_COMPRESSION
940 sslcontext.set_default_verify_paths()
941 sslcontext.set_alpn_protocols(("http/1.1",))
942 return sslcontext
945# The default SSLContext objects are created at import time
946# since they do blocking I/O to load certificates from disk,
947# and imports should always be done before the event loop starts
948# or in a thread.
949_SSL_CONTEXT_VERIFIED = _make_ssl_context(True)
950_SSL_CONTEXT_UNVERIFIED = _make_ssl_context(False)
953class TCPConnector(BaseConnector):
954 """TCP connector.
956 verify_ssl - Set to True to check ssl certifications.
957 fingerprint - Pass the binary sha256
958 digest of the expected certificate in DER format to verify
959 that the certificate the server presents matches. See also
960 https://en.wikipedia.org/wiki/HTTP_Public_Key_Pinning
961 resolver - Enable DNS lookups and use this
962 resolver
963 use_dns_cache - Use memory cache for DNS lookups.
964 ttl_dns_cache - Max seconds having cached a DNS entry, None forever.
965 family - socket address family
966 local_addr - local tuple of (host, port) to bind socket to
968 keepalive_timeout - (optional) Keep-alive timeout.
969 force_close - Set to True to force close and do reconnect
970 after each request (and between redirects).
971 limit - The total number of simultaneous connections.
972 limit_per_host - Number of simultaneous connections to one host.
973 enable_cleanup_closed - Enables clean-up closed ssl transports.
974 Disabled by default.
975 happy_eyeballs_delay - This is the “Connection Attempt Delay”
976 as defined in RFC 8305. To disable
977 the happy eyeballs algorithm, set to None.
978 interleave - “First Address Family Count” as defined in RFC 8305
979 loop - Optional event loop.
980 socket_factory - A SocketFactoryType function that, if supplied,
981 will be used to create sockets given an
982 AddrInfoType.
983 ssl_shutdown_timeout - DEPRECATED. Will be removed in aiohttp 4.0.
984 Grace period for SSL shutdown handshake on TLS
985 connections. Default is 0 seconds (immediate abort).
986 This parameter allowed for a clean SSL shutdown by
987 notifying the remote peer of connection closure,
988 while avoiding excessive delays during connector cleanup.
989 Note: Only takes effect on Python 3.11+.
990 """
992 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET | frozenset({"tcp"})
994 def __init__(
995 self,
996 *,
997 use_dns_cache: bool = True,
998 ttl_dns_cache: int | None = 10,
999 dns_cache_max_size: int = 1000,
1000 family: socket.AddressFamily = socket.AddressFamily.AF_UNSPEC,
1001 ssl: bool | Fingerprint | SSLContext = True,
1002 local_addr: tuple[str, int] | None = None,
1003 resolver: AbstractResolver | None = None,
1004 keepalive_timeout: None | float | _SENTINEL = sentinel,
1005 force_close: bool = False,
1006 limit: int = 100,
1007 limit_per_host: int = 0,
1008 enable_cleanup_closed: bool = False,
1009 timeout_ceil_threshold: float = 5,
1010 happy_eyeballs_delay: float | None = 0.25,
1011 interleave: int | None = None,
1012 socket_factory: SocketFactoryType | None = None,
1013 ssl_shutdown_timeout: _SENTINEL | None | float = sentinel,
1014 ):
1015 super().__init__(
1016 keepalive_timeout=keepalive_timeout,
1017 force_close=force_close,
1018 limit=limit,
1019 limit_per_host=limit_per_host,
1020 enable_cleanup_closed=enable_cleanup_closed,
1021 timeout_ceil_threshold=timeout_ceil_threshold,
1022 )
1024 if not isinstance(ssl, SSL_ALLOWED_TYPES):
1025 raise TypeError(
1026 "ssl should be SSLContext, Fingerprint, or bool, "
1027 f"got {ssl!r} instead."
1028 )
1029 self._ssl = ssl
1031 self._resolver: AbstractResolver
1032 if resolver is None:
1033 self._resolver = DefaultResolver()
1034 self._resolver_owner = True
1035 else:
1036 self._resolver = resolver
1037 self._resolver_owner = False
1039 self._use_dns_cache = use_dns_cache
1040 self._cached_hosts = _DNSCacheTable(
1041 ttl=ttl_dns_cache, max_size=dns_cache_max_size
1042 )
1043 self._throttle_dns_futures: dict[tuple[str, int], set[asyncio.Future[None]]] = (
1044 {}
1045 )
1046 self._family = family
1047 self._local_addr_infos = aiohappyeyeballs.addr_to_addr_infos(local_addr)
1048 self._happy_eyeballs_delay = happy_eyeballs_delay
1049 self._interleave = interleave
1050 self._resolve_host_tasks: set[asyncio.Task[list[ResolveResult]]] = set()
1051 self._socket_factory = socket_factory
1052 self._ssl_shutdown_timeout: float | None
1054 # Handle ssl_shutdown_timeout with warning for Python < 3.11
1055 if ssl_shutdown_timeout is sentinel:
1056 self._ssl_shutdown_timeout = 0
1057 else:
1058 # Deprecation warning for ssl_shutdown_timeout parameter
1059 warnings.warn(
1060 "The ssl_shutdown_timeout parameter is deprecated and will be removed in aiohttp 4.0",
1061 DeprecationWarning,
1062 stacklevel=2,
1063 )
1064 if (
1065 sys.version_info < (3, 11)
1066 and ssl_shutdown_timeout is not None
1067 and ssl_shutdown_timeout != 0
1068 ):
1069 warnings.warn(
1070 f"ssl_shutdown_timeout={ssl_shutdown_timeout} is ignored on Python < 3.11; "
1071 "only ssl_shutdown_timeout=0 is supported. The timeout will be ignored.",
1072 RuntimeWarning,
1073 stacklevel=2,
1074 )
1075 self._ssl_shutdown_timeout = ssl_shutdown_timeout
1077 async def close(self, *, abort_ssl: bool = False) -> None:
1078 """Close all opened transports.
1080 :param abort_ssl: If True, SSL connections will be aborted immediately
1081 without performing the shutdown handshake. If False (default),
1082 the behavior is determined by ssl_shutdown_timeout:
1083 - If ssl_shutdown_timeout=0: connections are aborted
1084 - If ssl_shutdown_timeout>0: graceful shutdown is performed
1085 """
1086 # Use abort_ssl param if explicitly set, otherwise use ssl_shutdown_timeout default
1087 await super().close(abort_ssl=abort_ssl or self._ssl_shutdown_timeout == 0)
1088 if self._resolver_owner:
1089 await self._resolver.close()
1091 def _close_immediately(self, *, abort_ssl: bool = False) -> list[Awaitable[object]]:
1092 for fut in chain.from_iterable(self._throttle_dns_futures.values()):
1093 fut.cancel()
1095 waiters = super()._close_immediately(abort_ssl=abort_ssl)
1097 for t in self._resolve_host_tasks:
1098 t.cancel()
1099 waiters.append(t)
1101 return waiters
1103 @property
1104 def family(self) -> int:
1105 """Socket family like AF_INET."""
1106 return self._family
1108 @property
1109 def use_dns_cache(self) -> bool:
1110 """True if local DNS caching is enabled."""
1111 return self._use_dns_cache
1113 def clear_dns_cache(self, host: str | None = None, port: int | None = None) -> None:
1114 """Remove specified host/port or clear all dns local cache."""
1115 if host is not None and port is not None:
1116 self._cached_hosts.remove((host, port))
1117 elif host is not None or port is not None:
1118 raise ValueError("either both host and port or none of them are allowed")
1119 else:
1120 self._cached_hosts.clear()
1122 async def _resolve_host(
1123 self, host: str, port: int, traces: Sequence["Trace"] | None = None
1124 ) -> list[ResolveResult]:
1125 """Resolve host and return list of addresses."""
1126 if is_ip_address(host):
1127 # Reject legacy numeric IPv4 forms (e.g. 2130706433, 127.1) that
1128 # socket would map onto an address, slipping past a connector-level
1129 # policy that only sees the raw host.
1130 if ":" not in host and not is_canonical_ipv4_address(host):
1131 raise InvalidUrlClientError(host, "is not a canonical IPv4 address")
1132 return [
1133 {
1134 "hostname": host,
1135 "host": host,
1136 "port": port,
1137 "family": self._family,
1138 "proto": 0,
1139 "flags": 0,
1140 }
1141 ]
1143 if not self._use_dns_cache:
1144 if traces:
1145 for trace in traces:
1146 await trace.send_dns_resolvehost_start(host)
1148 if self._closed:
1149 raise ClientConnectionError("Connector is closed")
1151 res = await self._resolver.resolve(host, port, family=self._family)
1153 if traces:
1154 for trace in traces:
1155 await trace.send_dns_resolvehost_end(host)
1157 return res
1159 key = (host, port)
1160 if key in self._cached_hosts and not self._cached_hosts.expired(key):
1161 # get result early, before any await (#4014)
1162 result = self._cached_hosts.next_addrs(key)
1164 if traces:
1165 for trace in traces:
1166 await trace.send_dns_cache_hit(host)
1167 return result
1169 futures: set[asyncio.Future[None]]
1170 #
1171 # If multiple connectors are resolving the same host, we wait
1172 # for the first one to resolve and then use the result for all of them.
1173 # We use a throttle to ensure that we only resolve the host once
1174 # and then use the result for all the waiters.
1175 #
1176 if key in self._throttle_dns_futures:
1177 # get futures early, before any await (#4014)
1178 futures = self._throttle_dns_futures[key]
1179 future: asyncio.Future[None] = self._loop.create_future()
1180 futures.add(future)
1181 if traces:
1182 for trace in traces:
1183 await trace.send_dns_cache_hit(host)
1184 try:
1185 await future
1186 finally:
1187 futures.discard(future)
1188 return self._cached_hosts.next_addrs(key)
1190 # update dict early, before any await (#4014)
1191 self._throttle_dns_futures[key] = futures = set()
1192 # In this case we need to create a task to ensure that we can shield
1193 # the task from cancellation as cancelling this lookup should not cancel
1194 # the underlying lookup or else the cancel event will get broadcast to
1195 # all the waiters across all connections.
1196 #
1197 coro = self._resolve_host_with_throttle(key, host, port, futures, traces)
1198 loop = asyncio.get_running_loop()
1199 if sys.version_info >= (3, 12):
1200 # Optimization for Python 3.12, try to send immediately
1201 resolved_host_task = asyncio.Task(coro, loop=loop, eager_start=True)
1202 else:
1203 resolved_host_task = loop.create_task(coro)
1205 if not resolved_host_task.done():
1206 self._resolve_host_tasks.add(resolved_host_task)
1207 resolved_host_task.add_done_callback(self._resolve_host_tasks.discard)
1209 try:
1210 return await asyncio.shield(resolved_host_task)
1211 except asyncio.CancelledError:
1213 def drop_exception(fut: "asyncio.Future[list[ResolveResult]]") -> None:
1214 with suppress(Exception, asyncio.CancelledError):
1215 fut.result()
1217 resolved_host_task.add_done_callback(drop_exception)
1218 raise
1220 async def _resolve_host_with_throttle(
1221 self,
1222 key: tuple[str, int],
1223 host: str,
1224 port: int,
1225 futures: set[asyncio.Future[None]],
1226 traces: Sequence["Trace"] | None,
1227 ) -> list[ResolveResult]:
1228 """Resolve host and set result for all waiters.
1230 This method must be run in a task and shielded from cancellation
1231 to avoid cancelling the underlying lookup.
1232 """
1233 try:
1234 if traces:
1235 for trace in traces:
1236 await trace.send_dns_cache_miss(host)
1238 for trace in traces:
1239 await trace.send_dns_resolvehost_start(host)
1241 addrs = await self._resolver.resolve(host, port, family=self._family)
1242 if traces:
1243 for trace in traces:
1244 await trace.send_dns_resolvehost_end(host)
1246 self._cached_hosts.add(key, addrs)
1247 for fut in futures:
1248 set_result(fut, None)
1249 except BaseException as e:
1250 # any DNS exception is set for the waiters to raise the same exception.
1251 # This coro is always run in task that is shielded from cancellation so
1252 # we should never be propagating cancellation here.
1253 for fut in futures:
1254 set_exception(fut, e)
1255 raise
1256 finally:
1257 self._throttle_dns_futures.pop(key)
1259 return self._cached_hosts.next_addrs(key)
1261 async def _create_connection(
1262 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1263 ) -> ResponseHandler:
1264 """Create connection.
1266 Has same keyword arguments as BaseEventLoop.create_connection.
1267 """
1268 if req.proxy:
1269 _, proto = await self._create_proxy_connection(req, traces, timeout)
1270 else:
1271 _, proto = await self._create_direct_connection(req, traces, timeout)
1273 return proto
1275 def _get_ssl_context(self, req: ClientRequestBase) -> SSLContext | None:
1276 """Logic to get the correct SSL context
1278 0. if req.ssl is false, return None
1280 1. if ssl_context is specified in req, use it
1281 2. if _ssl_context is specified in self, use it
1282 3. otherwise:
1283 1. if verify_ssl is not specified in req, use self.ssl_context
1284 (will generate a default context according to self.verify_ssl)
1285 2. if verify_ssl is True in req, generate a default SSL context
1286 3. if verify_ssl is False in req, generate a SSL context that
1287 won't verify
1288 """
1289 if not req.is_ssl():
1290 return None
1292 if ssl is None: # pragma: no cover
1293 raise RuntimeError("SSL is not supported.")
1294 sslcontext = req.ssl
1295 if isinstance(sslcontext, ssl.SSLContext):
1296 return sslcontext
1297 if sslcontext is not True:
1298 # not verified or fingerprinted
1299 return _SSL_CONTEXT_UNVERIFIED
1300 sslcontext = self._ssl
1301 if isinstance(sslcontext, ssl.SSLContext):
1302 return sslcontext
1303 if sslcontext is not True:
1304 # not verified or fingerprinted
1305 return _SSL_CONTEXT_UNVERIFIED
1306 return _SSL_CONTEXT_VERIFIED
1308 def _get_fingerprint(self, req: ClientRequestBase) -> "Fingerprint | None":
1309 ret = req.ssl
1310 if isinstance(ret, Fingerprint):
1311 return ret
1312 ret = self._ssl
1313 if isinstance(ret, Fingerprint):
1314 return ret
1315 return None
1317 async def _wrap_create_connection(
1318 self,
1319 *args: Any,
1320 addr_infos: list[AddrInfoType],
1321 req: ClientRequestBase,
1322 timeout: "ClientTimeout",
1323 client_error: type[Exception] = ClientConnectorError,
1324 **kwargs: Any,
1325 ) -> tuple[asyncio.Transport, ResponseHandler]:
1326 try:
1327 async with ceil_timeout(
1328 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1329 ):
1330 sock = await aiohappyeyeballs.start_connection(
1331 addr_infos=addr_infos,
1332 local_addr_infos=self._local_addr_infos,
1333 happy_eyeballs_delay=self._happy_eyeballs_delay,
1334 interleave=self._interleave,
1335 loop=self._loop,
1336 socket_factory=self._socket_factory,
1337 )
1338 # Add ssl_shutdown_timeout for Python 3.11+ when SSL is used
1339 if (
1340 kwargs.get("ssl")
1341 and self._ssl_shutdown_timeout
1342 and sys.version_info >= (3, 11)
1343 ):
1344 kwargs["ssl_shutdown_timeout"] = self._ssl_shutdown_timeout
1345 return await create_connection(self._loop, *args, **kwargs, sock=sock)
1346 except cert_errors as exc:
1347 raise ClientConnectorCertificateError(req.connection_key, exc) from exc
1348 except ssl_errors as exc:
1349 raise ClientConnectorSSLError(req.connection_key, exc) from exc
1350 except OSError as exc:
1351 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1352 raise
1353 raise client_error(req.connection_key, exc) from exc
1355 def _warn_about_tls_in_tls(
1356 self,
1357 underlying_transport: asyncio.Transport,
1358 req: ClientRequest,
1359 ) -> None:
1360 """Issue a warning if the requested URL has HTTPS scheme."""
1361 if req.url.scheme != "https":
1362 return
1364 # TLS-in-TLS only applies when the proxy itself is HTTPS.
1365 # When the proxy is HTTP, start_tls upgrades a plain TCP connection,
1366 # which is standard TLS and works on all event loops and Python versions.
1367 if req.proxy is None or req.proxy.scheme != "https":
1368 return
1370 # Check if uvloop is being used, which supports TLS in TLS,
1371 # otherwise assume that asyncio's native transport is being used.
1372 if type(underlying_transport).__module__.startswith("uvloop"):
1373 return
1375 # Check if aiofastnet is being used, which supports TLS in TLS
1376 if aiofastnet is not None:
1377 return
1379 # Support in asyncio was added in Python 3.11 (bpo-44011)
1380 asyncio_supports_tls_in_tls = sys.version_info >= (3, 11) or getattr( # type: ignore[unreachable]
1381 underlying_transport,
1382 "_start_tls_compatible",
1383 False,
1384 )
1386 if asyncio_supports_tls_in_tls:
1387 return
1389 warnings.warn(
1390 "An HTTPS request is being sent through an HTTPS proxy. "
1391 "This support for TLS in TLS is known to be disabled "
1392 "in the stdlib asyncio. This is why you'll probably see "
1393 "an error in the log below.\n\n"
1394 "It is possible to enable it via monkeypatching. "
1395 "For more details, see:\n"
1396 "* https://bugs.python.org/issue37179\n"
1397 "* https://github.com/python/cpython/pull/28073\n\n"
1398 "You can temporarily patch this as follows:\n"
1399 "* https://docs.aiohttp.org/en/stable/client_advanced.html#proxy-support\n"
1400 "* https://github.com/aio-libs/aiohttp/discussions/6044\n",
1401 RuntimeWarning,
1402 source=self,
1403 # Why `4`? At least 3 of the calls in the stack originate
1404 # from the methods in this class.
1405 stacklevel=3,
1406 )
1408 async def _start_tls_connection(
1409 self,
1410 underlying_transport: asyncio.Transport,
1411 req: ClientRequest,
1412 timeout: "ClientTimeout",
1413 client_error: type[Exception] = ClientConnectorError,
1414 ) -> tuple[asyncio.BaseTransport, ResponseHandler]:
1415 """Wrap the raw TCP transport with TLS."""
1416 tls_proto = self._factory() # Create a brand new proto for TLS
1417 sslcontext = self._get_ssl_context(req)
1418 if TYPE_CHECKING:
1419 # _start_tls_connection is unreachable in the current code path
1420 # if sslcontext is None.
1421 assert sslcontext is not None
1423 try:
1424 async with ceil_timeout(
1425 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1426 ):
1427 try:
1428 # ssl_shutdown_timeout is only available in Python 3.11+
1429 if sys.version_info >= (3, 11) and self._ssl_shutdown_timeout:
1430 tls_transport = await start_tls(
1431 self._loop,
1432 underlying_transport,
1433 tls_proto,
1434 sslcontext,
1435 server_hostname=req.server_hostname or req.url.raw_host,
1436 ssl_handshake_timeout=timeout.total,
1437 ssl_shutdown_timeout=self._ssl_shutdown_timeout,
1438 )
1439 else:
1440 tls_transport = await start_tls(
1441 self._loop,
1442 underlying_transport,
1443 tls_proto,
1444 sslcontext,
1445 server_hostname=req.server_hostname or req.url.raw_host,
1446 ssl_handshake_timeout=timeout.total,
1447 )
1448 except BaseException:
1449 # We need to close the underlying transport since
1450 # `start_tls()` probably failed before it had a
1451 # chance to do this:
1452 if self._ssl_shutdown_timeout == 0:
1453 underlying_transport.abort()
1454 else:
1455 underlying_transport.close()
1456 raise
1457 if isinstance(tls_transport, asyncio.Transport):
1458 fingerprint = self._get_fingerprint(req)
1459 if fingerprint:
1460 try:
1461 fingerprint.check(tls_transport)
1462 except ServerFingerprintMismatch:
1463 tls_transport.close()
1464 if not self._cleanup_closed_disabled:
1465 self._cleanup_closed_transports.append(tls_transport)
1466 raise
1467 except cert_errors as exc:
1468 raise ClientConnectorCertificateError(req.connection_key, exc) from exc
1469 except ssl_errors as exc:
1470 raise ClientConnectorSSLError(req.connection_key, exc) from exc
1471 except OSError as exc:
1472 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1473 raise
1474 raise client_error(req.connection_key, exc) from exc
1475 except TypeError as type_err:
1476 # Example cause looks like this:
1477 # TypeError: transport <asyncio.sslproto._SSLProtocolTransport
1478 # object at 0x7f760615e460> is not supported by start_tls()
1480 raise ClientConnectionError(
1481 "Cannot initialize a TLS-in-TLS connection to host "
1482 f"{req.url.host!s}:{req.url.port:d} through an underlying connection "
1483 f"to an HTTPS proxy {req.proxy!s} ssl:{req.ssl or 'default'} "
1484 f"[{type_err!s}]"
1485 ) from type_err
1486 else:
1487 if tls_transport is None:
1488 msg = "Failed to start TLS (possibly caused by closing transport)"
1489 raise client_error(req.connection_key, OSError(msg))
1490 tls_proto.connection_made(
1491 tls_transport
1492 ) # Kick the state machine of the new TLS protocol
1494 return tls_transport, tls_proto
1496 def _convert_hosts_to_addr_infos(
1497 self, hosts: list[ResolveResult]
1498 ) -> list[AddrInfoType]:
1499 """Converts the list of hosts to a list of addr_infos.
1501 The list of hosts is the result of a DNS lookup. The list of
1502 addr_infos is the result of a call to `socket.getaddrinfo()`.
1503 """
1504 addr_infos: list[AddrInfoType] = []
1505 for hinfo in hosts:
1506 host = hinfo["host"]
1507 is_ipv6 = ":" in host
1508 family = socket.AF_INET6 if is_ipv6 else socket.AF_INET
1509 if self._family and self._family != family:
1510 continue
1511 addr = (host, hinfo["port"], 0, 0) if is_ipv6 else (host, hinfo["port"])
1512 addr_infos.append(
1513 (family, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", addr)
1514 )
1515 return addr_infos
1517 async def _create_direct_connection(
1518 self,
1519 req: ClientRequestBase,
1520 traces: list["Trace"],
1521 timeout: "ClientTimeout",
1522 *,
1523 client_error: type[Exception] = ClientConnectorError,
1524 ) -> tuple[asyncio.Transport, ResponseHandler]:
1525 sslcontext = self._get_ssl_context(req)
1526 fingerprint = self._get_fingerprint(req)
1528 host = req.url.raw_host
1529 assert host is not None
1530 # Replace multiple trailing dots with a single one.
1531 # A trailing dot is only present for fully-qualified domain names.
1532 # See https://github.com/aio-libs/aiohttp/pull/7364.
1533 if host.endswith(".."):
1534 host = host.rstrip(".") + "."
1535 port = req.url.port
1536 assert port is not None
1537 try:
1538 # Cancelling this lookup should not cancel the underlying lookup
1539 # or else the cancel event will get broadcast to all the waiters
1540 # across all connections.
1541 hosts = await self._resolve_host(host, port, traces=traces)
1542 except OSError as exc:
1543 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1544 raise
1545 # in case of proxy it is not ClientProxyConnectionError
1546 # it is problem of resolving proxy ip itself
1547 raise ClientConnectorDNSError(req.connection_key, exc) from exc
1549 last_exc: Exception | None = None
1550 addr_infos = self._convert_hosts_to_addr_infos(hosts)
1551 while addr_infos:
1552 # Strip trailing dots, certificates contain FQDN without dots.
1553 # See https://github.com/aio-libs/aiohttp/issues/3636
1554 server_hostname = (
1555 (req.server_hostname or host).rstrip(".") if sslcontext else None
1556 )
1558 try:
1559 transp, proto = await self._wrap_create_connection(
1560 self._factory,
1561 timeout=timeout,
1562 ssl=sslcontext,
1563 addr_infos=addr_infos,
1564 server_hostname=server_hostname,
1565 req=req,
1566 client_error=client_error,
1567 )
1568 except (ClientConnectorError, asyncio.TimeoutError) as exc:
1569 last_exc = exc
1570 aiohappyeyeballs.pop_addr_infos_interleave(addr_infos, self._interleave)
1571 continue
1573 if req.is_ssl() and fingerprint:
1574 try:
1575 fingerprint.check(transp)
1576 except ServerFingerprintMismatch as exc:
1577 transp.close()
1578 if not self._cleanup_closed_disabled:
1579 self._cleanup_closed_transports.append(transp)
1580 last_exc = exc
1581 # Remove the bad peer from the list of addr_infos
1582 sock: socket.socket = transp.get_extra_info("socket")
1583 bad_peer = sock.getpeername()
1584 aiohappyeyeballs.remove_addr_infos(addr_infos, bad_peer)
1585 continue
1587 return transp, proto
1588 assert last_exc is not None
1589 raise last_exc
1591 async def _create_proxy_connection(
1592 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1593 ) -> tuple[asyncio.BaseTransport, ResponseHandler]:
1594 proxy_req = self._update_proxy_auth_header_and_build_proxy_req(req)
1596 # create connection to proxy server
1597 transport, proto = await self._create_direct_connection(
1598 proxy_req, [], timeout, client_error=ClientProxyConnectionError
1599 )
1601 if req.is_ssl():
1602 self._warn_about_tls_in_tls(transport, req)
1604 # For HTTPS requests over HTTP proxy
1605 # we must notify proxy to tunnel connection
1606 # so we send CONNECT command:
1607 # CONNECT www.python.org:443 HTTP/1.1
1608 # Host: www.python.org
1609 #
1610 # next we must do TLS handshake and so on
1611 # to do this we must wrap raw socket into secure one
1612 # asyncio handles this perfectly
1613 proxy_req.method = hdrs.METH_CONNECT
1614 proxy_req.url = req.url
1615 key = req.connection_key._replace(proxy=None, proxy_headers_hash=None)
1616 conn = _ConnectTunnelConnection(self, key, proto, self._loop)
1617 proxy_resp = await proxy_req._send(conn)
1618 try:
1619 protocol = conn._protocol
1620 assert protocol is not None
1622 # read_until_eof=True will ensure the connection isn't closed
1623 # once the response is received and processed allowing
1624 # START_TLS to work on the connection below.
1625 protocol.set_response_params(
1626 read_until_eof=True,
1627 timeout_ceil_threshold=self._timeout_ceil_threshold,
1628 )
1629 resp = await proxy_resp.start(conn)
1630 except BaseException:
1631 proxy_resp.close()
1632 conn.close()
1633 raise
1634 else:
1635 conn._protocol = None
1636 try:
1637 if resp.status != 200:
1638 message = resp.reason
1639 if message is None:
1640 message = HTTPStatus(resp.status).phrase
1641 raise ClientHttpProxyError(
1642 proxy_resp.request_info,
1643 resp.history,
1644 status=resp.status,
1645 message=message,
1646 headers=resp.headers,
1647 )
1648 except BaseException:
1649 # It shouldn't be closed in `finally` because it's fed to
1650 # `loop.start_tls()` and the docs say not to touch it after
1651 # passing there.
1652 transport.close()
1653 raise
1655 return await self._start_tls_connection(
1656 # Access the old transport for the last time before it's
1657 # closed and forgotten forever:
1658 transport,
1659 req=req,
1660 timeout=timeout,
1661 )
1662 finally:
1663 proxy_resp.close()
1665 return transport, proto
1668class UnixConnector(BaseConnector):
1669 """Unix socket connector.
1671 path - Unix socket path.
1672 keepalive_timeout - (optional) Keep-alive timeout.
1673 force_close - Set to True to force close and do reconnect
1674 after each request (and between redirects).
1675 limit - The total number of simultaneous connections.
1676 limit_per_host - Number of simultaneous connections to one host.
1677 loop - Optional event loop.
1678 """
1680 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET | frozenset({"unix"})
1682 def __init__(
1683 self,
1684 path: str,
1685 force_close: bool = False,
1686 keepalive_timeout: _SENTINEL | float | None = sentinel,
1687 limit: int = 100,
1688 limit_per_host: int = 0,
1689 ) -> None:
1690 super().__init__(
1691 force_close=force_close,
1692 keepalive_timeout=keepalive_timeout,
1693 limit=limit,
1694 limit_per_host=limit_per_host,
1695 )
1696 self._path = path
1698 @property
1699 def path(self) -> str:
1700 """Path to unix socket."""
1701 return self._path
1703 async def _create_connection(
1704 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1705 ) -> ResponseHandler:
1706 try:
1707 async with ceil_timeout(
1708 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1709 ):
1710 _, proto = await self._loop.create_unix_connection(
1711 self._factory, self._path
1712 )
1713 except OSError as exc:
1714 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1715 raise
1716 raise UnixClientConnectorError(self.path, req.connection_key, exc) from exc
1718 return proto
1721class NamedPipeConnector(BaseConnector):
1722 """Named pipe connector.
1724 Only supported by the proactor event loop.
1725 See also: https://docs.python.org/3/library/asyncio-eventloop.html
1727 path - Windows named pipe path.
1728 keepalive_timeout - (optional) Keep-alive timeout.
1729 force_close - Set to True to force close and do reconnect
1730 after each request (and between redirects).
1731 limit - The total number of simultaneous connections.
1732 limit_per_host - Number of simultaneous connections to one host.
1733 loop - Optional event loop.
1734 """
1736 allowed_protocol_schema_set = HIGH_LEVEL_SCHEMA_SET | frozenset({"npipe"})
1738 def __init__(
1739 self,
1740 path: str,
1741 force_close: bool = False,
1742 keepalive_timeout: _SENTINEL | float | None = sentinel,
1743 limit: int = 100,
1744 limit_per_host: int = 0,
1745 ) -> None:
1746 super().__init__(
1747 force_close=force_close,
1748 keepalive_timeout=keepalive_timeout,
1749 limit=limit,
1750 limit_per_host=limit_per_host,
1751 )
1752 if not isinstance(
1753 self._loop,
1754 asyncio.ProactorEventLoop, # type: ignore[attr-defined]
1755 ):
1756 raise RuntimeError(
1757 "Named Pipes only available in proactor loop under windows"
1758 )
1759 self._path = path
1761 @property
1762 def path(self) -> str:
1763 """Path to the named pipe."""
1764 return self._path
1766 async def _create_connection(
1767 self, req: ClientRequest, traces: list["Trace"], timeout: "ClientTimeout"
1768 ) -> ResponseHandler:
1769 try:
1770 async with ceil_timeout(
1771 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1772 ):
1773 _, proto = await self._loop.create_pipe_connection( # type: ignore[attr-defined]
1774 self._factory, self._path
1775 )
1776 # the drain is required so that the connection_made is called
1777 # and transport is set otherwise it is not set before the
1778 # `assert conn.transport is not None`
1779 # in client.py's _request method
1780 await asyncio.sleep(0)
1781 # other option is to manually set transport like
1782 # `proto.transport = trans`
1783 except OSError as exc:
1784 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1785 raise
1786 raise ClientConnectorError(req.connection_key, exc) from exc
1788 return cast(ResponseHandler, proto)