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