Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/connector.py: 17%
676 statements
« prev ^ index » next coverage.py v7.3.1, created at 2023-09-27 06:09 +0000
1import asyncio
2import dataclasses
3import functools
4import logging
5import random
6import sys
7import traceback
8import warnings
9from collections import defaultdict, deque
10from contextlib import suppress
11from http import HTTPStatus
12from http.cookies import SimpleCookie
13from itertools import cycle, islice
14from time import monotonic
15from types import TracebackType
16from typing import ( # noqa
17 TYPE_CHECKING,
18 Any,
19 Awaitable,
20 Callable,
21 DefaultDict,
22 Dict,
23 Iterator,
24 List,
25 Literal,
26 Optional,
27 Set,
28 Tuple,
29 Type,
30 Union,
31 cast,
32)
34from . import hdrs, helpers
35from .abc import AbstractResolver
36from .client_exceptions import (
37 ClientConnectionError,
38 ClientConnectorCertificateError,
39 ClientConnectorError,
40 ClientConnectorSSLError,
41 ClientHttpProxyError,
42 ClientProxyConnectionError,
43 ServerFingerprintMismatch,
44 UnixClientConnectorError,
45 cert_errors,
46 ssl_errors,
47)
48from .client_proto import ResponseHandler
49from .client_reqrep import SSL_ALLOWED_TYPES, ClientRequest, Fingerprint
50from .helpers import _SENTINEL, ceil_timeout, is_ip_address, sentinel, set_result
51from .locks import EventResultOrError
52from .resolver import DefaultResolver
54try:
55 import ssl
57 SSLContext = ssl.SSLContext
58except ImportError: # pragma: no cover
59 ssl = None # type: ignore[assignment]
60 SSLContext = object # type: ignore[misc,assignment]
63__all__ = ("BaseConnector", "TCPConnector", "UnixConnector", "NamedPipeConnector")
66if TYPE_CHECKING: # pragma: no cover
67 from .client import ClientTimeout
68 from .client_reqrep import ConnectionKey
69 from .tracing import Trace
72class Connection:
73 _source_traceback = None
74 _transport = None
76 def __init__(
77 self,
78 connector: "BaseConnector",
79 key: "ConnectionKey",
80 protocol: ResponseHandler,
81 loop: asyncio.AbstractEventLoop,
82 ) -> None:
83 self._key = key
84 self._connector = connector
85 self._loop = loop
86 self._protocol: Optional[ResponseHandler] = protocol
87 self._callbacks: List[Callable[[], None]] = []
89 if loop.get_debug():
90 self._source_traceback = traceback.extract_stack(sys._getframe(1))
92 def __repr__(self) -> str:
93 return f"Connection<{self._key}>"
95 def __del__(self, _warnings: Any = warnings) -> None:
96 if self._protocol is not None:
97 _warnings.warn(
98 f"Unclosed connection {self!r}", ResourceWarning, source=self
99 )
100 if self._loop.is_closed():
101 return
103 self._connector._release(self._key, self._protocol, should_close=True)
105 context = {"client_connection": self, "message": "Unclosed connection"}
106 if self._source_traceback is not None:
107 context["source_traceback"] = self._source_traceback
108 self._loop.call_exception_handler(context)
110 @property
111 def transport(self) -> Optional[asyncio.Transport]:
112 if self._protocol is None:
113 return None
114 return self._protocol.transport
116 @property
117 def protocol(self) -> Optional[ResponseHandler]:
118 return self._protocol
120 def add_callback(self, callback: Callable[[], None]) -> None:
121 if callback is not None:
122 self._callbacks.append(callback)
124 def _notify_release(self) -> None:
125 callbacks, self._callbacks = self._callbacks[:], []
127 for cb in callbacks:
128 with suppress(Exception):
129 cb()
131 def close(self) -> None:
132 self._notify_release()
134 if self._protocol is not None:
135 self._connector._release(self._key, self._protocol, should_close=True)
136 self._protocol = None
138 def release(self) -> None:
139 self._notify_release()
141 if self._protocol is not None:
142 self._connector._release(
143 self._key, self._protocol, should_close=self._protocol.should_close
144 )
145 self._protocol = None
147 @property
148 def closed(self) -> bool:
149 return self._protocol is None or not self._protocol.is_connected()
152class _TransportPlaceholder:
153 """placeholder for BaseConnector.connect function"""
155 def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
156 fut = loop.create_future()
157 fut.set_result(None)
158 self.closed: asyncio.Future[Optional[Exception]] = fut
160 def close(self) -> None:
161 pass
164class BaseConnector:
165 """Base connector class.
167 keepalive_timeout - (optional) Keep-alive timeout.
168 force_close - Set to True to force close and do reconnect
169 after each request (and between redirects).
170 limit - The total number of simultaneous connections.
171 limit_per_host - Number of simultaneous connections to one host.
172 enable_cleanup_closed - Enables clean-up closed ssl transports.
173 Disabled by default.
174 timeout_ceil_threshold - Trigger ceiling of timeout values when
175 it's above timeout_ceil_threshold.
176 loop - Optional event loop.
177 """
179 _closed = True # prevent AttributeError in __del__ if ctor was failed
180 _source_traceback = None
182 # abort transport after 2 seconds (cleanup broken connections)
183 _cleanup_closed_period = 2.0
185 def __init__(
186 self,
187 *,
188 keepalive_timeout: Union[_SENTINEL, None, float] = sentinel,
189 force_close: bool = False,
190 limit: int = 100,
191 limit_per_host: int = 0,
192 enable_cleanup_closed: bool = False,
193 timeout_ceil_threshold: float = 5,
194 ) -> None:
195 if force_close:
196 if keepalive_timeout is not None and keepalive_timeout is not sentinel:
197 raise ValueError(
198 "keepalive_timeout cannot " "be set if force_close is True"
199 )
200 else:
201 if keepalive_timeout is sentinel:
202 keepalive_timeout = 15.0
204 self._timeout_ceil_threshold = timeout_ceil_threshold
206 loop = asyncio.get_running_loop()
208 self._closed = False
209 if loop.get_debug():
210 self._source_traceback = traceback.extract_stack(sys._getframe(1))
212 self._conns: Dict[ConnectionKey, List[Tuple[ResponseHandler, float]]] = {}
213 self._limit = limit
214 self._limit_per_host = limit_per_host
215 self._acquired: Set[ResponseHandler] = set()
216 self._acquired_per_host: DefaultDict[
217 ConnectionKey, Set[ResponseHandler]
218 ] = defaultdict(set)
219 self._keepalive_timeout = cast(float, keepalive_timeout)
220 self._force_close = force_close
222 # {host_key: FIFO list of waiters}
223 self._waiters = defaultdict(deque) # type: ignore[var-annotated]
225 self._loop = loop
226 self._factory = functools.partial(ResponseHandler, loop=loop)
228 self.cookies: SimpleCookie[str] = SimpleCookie()
230 # start keep-alive connection cleanup task
231 self._cleanup_handle: Optional[asyncio.TimerHandle] = None
233 # start cleanup closed transports task
234 self._cleanup_closed_handle: Optional[asyncio.TimerHandle] = None
235 self._cleanup_closed_disabled = not enable_cleanup_closed
236 self._cleanup_closed_transports: List[Optional[asyncio.Transport]] = []
237 self._cleanup_closed()
239 def __del__(self, _warnings: Any = warnings) -> None:
240 if self._closed:
241 return
242 if not self._conns:
243 return
245 conns = [repr(c) for c in self._conns.values()]
247 self._close_immediately()
249 _warnings.warn(f"Unclosed connector {self!r}", ResourceWarning, source=self)
250 context = {
251 "connector": self,
252 "connections": conns,
253 "message": "Unclosed connector",
254 }
255 if self._source_traceback is not None:
256 context["source_traceback"] = self._source_traceback
257 self._loop.call_exception_handler(context)
259 async def __aenter__(self) -> "BaseConnector":
260 return self
262 async def __aexit__(
263 self,
264 exc_type: Optional[Type[BaseException]] = None,
265 exc_value: Optional[BaseException] = None,
266 exc_traceback: Optional[TracebackType] = None,
267 ) -> None:
268 await self.close()
270 @property
271 def force_close(self) -> bool:
272 """Ultimately close connection on releasing if True."""
273 return self._force_close
275 @property
276 def limit(self) -> int:
277 """The total number for simultaneous connections.
279 If limit is 0 the connector has no limit.
280 The default limit size is 100.
281 """
282 return self._limit
284 @property
285 def limit_per_host(self) -> int:
286 """The limit for simultaneous connections to the same endpoint.
288 Endpoints are the same if they are have equal
289 (host, port, is_ssl) triple.
290 """
291 return self._limit_per_host
293 def _cleanup(self) -> None:
294 """Cleanup unused transports."""
295 if self._cleanup_handle:
296 self._cleanup_handle.cancel()
297 # _cleanup_handle should be unset, otherwise _release() will not
298 # recreate it ever!
299 self._cleanup_handle = None
301 now = self._loop.time()
302 timeout = self._keepalive_timeout
304 if self._conns:
305 connections = {}
306 deadline = now - timeout
307 for key, conns in self._conns.items():
308 alive = []
309 for proto, use_time in conns:
310 if proto.is_connected():
311 if use_time - deadline < 0:
312 transport = proto.transport
313 proto.close()
314 if key.is_ssl and not self._cleanup_closed_disabled:
315 self._cleanup_closed_transports.append(transport)
316 else:
317 alive.append((proto, use_time))
318 else:
319 transport = proto.transport
320 proto.close()
321 if key.is_ssl and not self._cleanup_closed_disabled:
322 self._cleanup_closed_transports.append(transport)
324 if alive:
325 connections[key] = alive
327 self._conns = connections
329 if self._conns:
330 self._cleanup_handle = helpers.weakref_handle(
331 self,
332 "_cleanup",
333 timeout,
334 self._loop,
335 timeout_ceil_threshold=self._timeout_ceil_threshold,
336 )
338 def _drop_acquired_per_host(
339 self, key: "ConnectionKey", val: ResponseHandler
340 ) -> None:
341 acquired_per_host = self._acquired_per_host
342 if key not in acquired_per_host:
343 return
344 conns = acquired_per_host[key]
345 conns.remove(val)
346 if not conns:
347 del self._acquired_per_host[key]
349 def _cleanup_closed(self) -> None:
350 """Double confirmation for transport close.
352 Some broken ssl servers may leave socket open without proper close.
353 """
354 if self._cleanup_closed_handle:
355 self._cleanup_closed_handle.cancel()
357 for transport in self._cleanup_closed_transports:
358 if transport is not None:
359 transport.abort()
361 self._cleanup_closed_transports = []
363 if not self._cleanup_closed_disabled:
364 self._cleanup_closed_handle = helpers.weakref_handle(
365 self,
366 "_cleanup_closed",
367 self._cleanup_closed_period,
368 self._loop,
369 timeout_ceil_threshold=self._timeout_ceil_threshold,
370 )
372 async def close(self) -> None:
373 """Close all opened transports."""
374 waiters = self._close_immediately()
375 if waiters:
376 results = await asyncio.gather(*waiters, return_exceptions=True)
377 for res in results:
378 if isinstance(res, Exception):
379 err_msg = "Error while closing connector: " + repr(res)
380 logging.error(err_msg)
382 def _close_immediately(self) -> List["asyncio.Future[None]"]:
383 waiters: List["asyncio.Future[None]"] = []
385 if self._closed:
386 return waiters
388 self._closed = True
390 try:
391 if self._loop.is_closed():
392 return waiters
394 # cancel cleanup task
395 if self._cleanup_handle:
396 self._cleanup_handle.cancel()
398 # cancel cleanup close task
399 if self._cleanup_closed_handle:
400 self._cleanup_closed_handle.cancel()
402 for data in self._conns.values():
403 for proto, t0 in data:
404 proto.close()
405 waiters.append(proto.closed)
407 for proto in self._acquired:
408 proto.close()
409 waiters.append(proto.closed)
411 # TODO (A.Yushovskiy, 24-May-2019) collect transp. closing futures
412 for transport in self._cleanup_closed_transports:
413 if transport is not None:
414 transport.abort()
416 return waiters
418 finally:
419 self._conns.clear()
420 self._acquired.clear()
421 self._waiters.clear()
422 self._cleanup_handle = None
423 self._cleanup_closed_transports.clear()
424 self._cleanup_closed_handle = None
426 @property
427 def closed(self) -> bool:
428 """Is connector closed.
430 A readonly property.
431 """
432 return self._closed
434 def _available_connections(self, key: "ConnectionKey") -> int:
435 """
436 Return number of available connections.
438 The limit, limit_per_host and the connection key are taken into account.
440 If it returns less than 1 means that there are no connections
441 available.
442 """
443 if self._limit:
444 # total calc available connections
445 available = self._limit - len(self._acquired)
447 # check limit per host
448 if (
449 self._limit_per_host
450 and available > 0
451 and key in self._acquired_per_host
452 ):
453 acquired = self._acquired_per_host.get(key)
454 assert acquired is not None
455 available = self._limit_per_host - len(acquired)
457 elif self._limit_per_host and key in self._acquired_per_host:
458 # check limit per host
459 acquired = self._acquired_per_host.get(key)
460 assert acquired is not None
461 available = self._limit_per_host - len(acquired)
462 else:
463 available = 1
465 return available
467 async def connect(
468 self, req: ClientRequest, traces: List["Trace"], timeout: "ClientTimeout"
469 ) -> Connection:
470 """Get from pool or create new connection."""
471 key = req.connection_key
472 available = self._available_connections(key)
474 # Wait if there are no available connections or if there are/were
475 # waiters (i.e. don't steal connection from a waiter about to wake up)
476 if available <= 0 or key in self._waiters:
477 fut = self._loop.create_future()
479 # This connection will now count towards the limit.
480 self._waiters[key].append(fut)
482 if traces:
483 for trace in traces:
484 await trace.send_connection_queued_start()
486 try:
487 await fut
488 except BaseException as e:
489 if key in self._waiters:
490 # remove a waiter even if it was cancelled, normally it's
491 # removed when it's notified
492 try:
493 self._waiters[key].remove(fut)
494 except ValueError: # fut may no longer be in list
495 pass
497 raise e
498 finally:
499 if key in self._waiters and not self._waiters[key]:
500 del self._waiters[key]
502 if traces:
503 for trace in traces:
504 await trace.send_connection_queued_end()
506 proto = self._get(key)
507 if proto is None:
508 placeholder = cast(ResponseHandler, _TransportPlaceholder(self._loop))
509 self._acquired.add(placeholder)
510 self._acquired_per_host[key].add(placeholder)
512 if traces:
513 for trace in traces:
514 await trace.send_connection_create_start()
516 try:
517 proto = await self._create_connection(req, traces, timeout)
518 if self._closed:
519 proto.close()
520 raise ClientConnectionError("Connector is closed.")
521 except BaseException:
522 if not self._closed:
523 self._acquired.remove(placeholder)
524 self._drop_acquired_per_host(key, placeholder)
525 self._release_waiter()
526 raise
527 else:
528 if not self._closed:
529 self._acquired.remove(placeholder)
530 self._drop_acquired_per_host(key, placeholder)
532 if traces:
533 for trace in traces:
534 await trace.send_connection_create_end()
535 else:
536 if traces:
537 # Acquire the connection to prevent race conditions with limits
538 placeholder = cast(ResponseHandler, _TransportPlaceholder(self._loop))
539 self._acquired.add(placeholder)
540 self._acquired_per_host[key].add(placeholder)
541 for trace in traces:
542 await trace.send_connection_reuseconn()
543 self._acquired.remove(placeholder)
544 self._drop_acquired_per_host(key, placeholder)
546 self._acquired.add(proto)
547 self._acquired_per_host[key].add(proto)
548 return Connection(self, key, proto, self._loop)
550 def _get(self, key: "ConnectionKey") -> Optional[ResponseHandler]:
551 try:
552 conns = self._conns[key]
553 except KeyError:
554 return None
556 t1 = self._loop.time()
557 while conns:
558 proto, t0 = conns.pop()
559 if proto.is_connected():
560 if t1 - t0 > self._keepalive_timeout:
561 transport = proto.transport
562 proto.close()
563 # only for SSL transports
564 if key.is_ssl and not self._cleanup_closed_disabled:
565 self._cleanup_closed_transports.append(transport)
566 else:
567 if not conns:
568 # The very last connection was reclaimed: drop the key
569 del self._conns[key]
570 return proto
571 else:
572 transport = proto.transport
573 proto.close()
574 if key.is_ssl and not self._cleanup_closed_disabled:
575 self._cleanup_closed_transports.append(transport)
577 # No more connections: drop the key
578 del self._conns[key]
579 return None
581 def _release_waiter(self) -> None:
582 """
583 Iterates over all waiters until one to be released is found.
585 The one to be released is not finished and
586 belongs to a host that has available connections.
587 """
588 if not self._waiters:
589 return
591 # Having the dict keys ordered this avoids to iterate
592 # at the same order at each call.
593 queues = list(self._waiters.keys())
594 random.shuffle(queues)
596 for key in queues:
597 if self._available_connections(key) < 1:
598 continue
600 waiters = self._waiters[key]
601 while waiters:
602 waiter = waiters.popleft()
603 if not waiter.done():
604 waiter.set_result(None)
605 return
607 def _release_acquired(self, key: "ConnectionKey", proto: ResponseHandler) -> None:
608 if self._closed:
609 # acquired connection is already released on connector closing
610 return
612 try:
613 self._acquired.remove(proto)
614 self._drop_acquired_per_host(key, proto)
615 except KeyError: # pragma: no cover
616 # this may be result of undetermenistic order of objects
617 # finalization due garbage collection.
618 pass
619 else:
620 self._release_waiter()
622 def _release(
623 self,
624 key: "ConnectionKey",
625 protocol: ResponseHandler,
626 *,
627 should_close: bool = False,
628 ) -> None:
629 if self._closed:
630 # acquired connection is already released on connector closing
631 return
633 self._release_acquired(key, protocol)
635 if self._force_close:
636 should_close = True
638 if should_close or protocol.should_close:
639 transport = protocol.transport
640 protocol.close()
641 # TODO: Remove once fixed: https://bugs.python.org/issue39951
642 # See PR #6321
643 set_result(protocol.closed, None)
645 if key.is_ssl and not self._cleanup_closed_disabled:
646 self._cleanup_closed_transports.append(transport)
647 else:
648 conns = self._conns.get(key)
649 if conns is None:
650 conns = self._conns[key] = []
651 conns.append((protocol, self._loop.time()))
653 if self._cleanup_handle is None:
654 self._cleanup_handle = helpers.weakref_handle(
655 self,
656 "_cleanup",
657 self._keepalive_timeout,
658 self._loop,
659 timeout_ceil_threshold=self._timeout_ceil_threshold,
660 )
662 async def _create_connection(
663 self, req: ClientRequest, traces: List["Trace"], timeout: "ClientTimeout"
664 ) -> ResponseHandler:
665 raise NotImplementedError()
668class _DNSCacheTable:
669 def __init__(self, ttl: Optional[float] = None) -> None:
670 self._addrs_rr: Dict[Tuple[str, int], Tuple[Iterator[Dict[str, Any]], int]] = {}
671 self._timestamps: Dict[Tuple[str, int], float] = {}
672 self._ttl = ttl
674 def __contains__(self, host: object) -> bool:
675 return host in self._addrs_rr
677 def add(self, key: Tuple[str, int], addrs: List[Dict[str, Any]]) -> None:
678 self._addrs_rr[key] = (cycle(addrs), len(addrs))
680 if self._ttl is not None:
681 self._timestamps[key] = monotonic()
683 def remove(self, key: Tuple[str, int]) -> None:
684 self._addrs_rr.pop(key, None)
686 if self._ttl is not None:
687 self._timestamps.pop(key, None)
689 def clear(self) -> None:
690 self._addrs_rr.clear()
691 self._timestamps.clear()
693 def next_addrs(self, key: Tuple[str, int]) -> List[Dict[str, Any]]:
694 loop, length = self._addrs_rr[key]
695 addrs = list(islice(loop, length))
696 # Consume one more element to shift internal state of `cycle`
697 next(loop)
698 return addrs
700 def expired(self, key: Tuple[str, int]) -> bool:
701 if self._ttl is None:
702 return False
704 return self._timestamps[key] + self._ttl < monotonic()
707class TCPConnector(BaseConnector):
708 """TCP connector.
710 verify_ssl - Set to True to check ssl certifications.
711 fingerprint - Pass the binary sha256
712 digest of the expected certificate in DER format to verify
713 that the certificate the server presents matches. See also
714 https://en.wikipedia.org/wiki/Transport_Layer_Security#Certificate_pinning
715 resolver - Enable DNS lookups and use this
716 resolver
717 use_dns_cache - Use memory cache for DNS lookups.
718 ttl_dns_cache - Max seconds having cached a DNS entry, None forever.
719 family - socket address family
720 local_addr - local tuple of (host, port) to bind socket to
722 keepalive_timeout - (optional) Keep-alive timeout.
723 force_close - Set to True to force close and do reconnect
724 after each request (and between redirects).
725 limit - The total number of simultaneous connections.
726 limit_per_host - Number of simultaneous connections to one host.
727 enable_cleanup_closed - Enables clean-up closed ssl transports.
728 Disabled by default.
729 loop - Optional event loop.
730 """
732 def __init__(
733 self,
734 *,
735 use_dns_cache: bool = True,
736 ttl_dns_cache: Optional[int] = 10,
737 family: int = 0,
738 ssl: Union[None, Literal[False], Fingerprint, SSLContext] = None,
739 local_addr: Optional[Tuple[str, int]] = None,
740 resolver: Optional[AbstractResolver] = None,
741 keepalive_timeout: Union[None, float, _SENTINEL] = sentinel,
742 force_close: bool = False,
743 limit: int = 100,
744 limit_per_host: int = 0,
745 enable_cleanup_closed: bool = False,
746 timeout_ceil_threshold: float = 5,
747 ) -> None:
748 super().__init__(
749 keepalive_timeout=keepalive_timeout,
750 force_close=force_close,
751 limit=limit,
752 limit_per_host=limit_per_host,
753 enable_cleanup_closed=enable_cleanup_closed,
754 timeout_ceil_threshold=timeout_ceil_threshold,
755 )
757 if not isinstance(ssl, SSL_ALLOWED_TYPES):
758 raise TypeError(
759 "ssl should be SSLContext, bool, Fingerprint, "
760 "or None, got {!r} instead.".format(ssl)
761 )
762 self._ssl = ssl
763 if resolver is None:
764 resolver = DefaultResolver()
765 self._resolver: AbstractResolver = resolver
767 self._use_dns_cache = use_dns_cache
768 self._cached_hosts = _DNSCacheTable(ttl=ttl_dns_cache)
769 self._throttle_dns_events: Dict[Tuple[str, int], EventResultOrError] = {}
770 self._family = family
771 self._local_addr = local_addr
773 def _close_immediately(self) -> List["asyncio.Future[None]"]:
774 for ev in self._throttle_dns_events.values():
775 ev.cancel()
776 return super()._close_immediately()
778 @property
779 def family(self) -> int:
780 """Socket family like AF_INET."""
781 return self._family
783 @property
784 def use_dns_cache(self) -> bool:
785 """True if local DNS caching is enabled."""
786 return self._use_dns_cache
788 def clear_dns_cache(
789 self, host: Optional[str] = None, port: Optional[int] = None
790 ) -> None:
791 """Remove specified host/port or clear all dns local cache."""
792 if host is not None and port is not None:
793 self._cached_hosts.remove((host, port))
794 elif host is not None or port is not None:
795 raise ValueError("either both host and port " "or none of them are allowed")
796 else:
797 self._cached_hosts.clear()
799 async def _resolve_host(
800 self, host: str, port: int, traces: Optional[List["Trace"]] = None
801 ) -> List[Dict[str, Any]]:
802 if is_ip_address(host):
803 return [
804 {
805 "hostname": host,
806 "host": host,
807 "port": port,
808 "family": self._family,
809 "proto": 0,
810 "flags": 0,
811 }
812 ]
814 if not self._use_dns_cache:
815 if traces:
816 for trace in traces:
817 await trace.send_dns_resolvehost_start(host)
819 res = await self._resolver.resolve(host, port, family=self._family)
821 if traces:
822 for trace in traces:
823 await trace.send_dns_resolvehost_end(host)
825 return res
827 key = (host, port)
829 if (key in self._cached_hosts) and (not self._cached_hosts.expired(key)):
830 # get result early, before any await (#4014)
831 result = self._cached_hosts.next_addrs(key)
833 if traces:
834 for trace in traces:
835 await trace.send_dns_cache_hit(host)
836 return result
838 if key in self._throttle_dns_events:
839 # get event early, before any await (#4014)
840 event = self._throttle_dns_events[key]
841 if traces:
842 for trace in traces:
843 await trace.send_dns_cache_hit(host)
844 await event.wait()
845 else:
846 # update dict early, before any await (#4014)
847 self._throttle_dns_events[key] = EventResultOrError(self._loop)
848 if traces:
849 for trace in traces:
850 await trace.send_dns_cache_miss(host)
851 try:
852 if traces:
853 for trace in traces:
854 await trace.send_dns_resolvehost_start(host)
856 addrs = await self._resolver.resolve(host, port, family=self._family)
857 if traces:
858 for trace in traces:
859 await trace.send_dns_resolvehost_end(host)
861 self._cached_hosts.add(key, addrs)
862 self._throttle_dns_events[key].set()
863 except BaseException as e:
864 # any DNS exception, independently of the implementation
865 # is set for the waiters to raise the same exception.
866 self._throttle_dns_events[key].set(exc=e)
867 raise
868 finally:
869 self._throttle_dns_events.pop(key)
871 return self._cached_hosts.next_addrs(key)
873 async def _create_connection(
874 self, req: ClientRequest, traces: List["Trace"], timeout: "ClientTimeout"
875 ) -> ResponseHandler:
876 """Create connection.
878 Has same keyword arguments as BaseEventLoop.create_connection.
879 """
880 if req.proxy:
881 _, proto = await self._create_proxy_connection(req, traces, timeout)
882 else:
883 _, proto = await self._create_direct_connection(req, traces, timeout)
885 return proto
887 @staticmethod
888 @functools.lru_cache(None)
889 def _make_ssl_context(verified: bool) -> SSLContext:
890 if verified:
891 return ssl.create_default_context()
892 else:
893 sslcontext = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
894 sslcontext.options |= ssl.OP_NO_SSLv2
895 sslcontext.options |= ssl.OP_NO_SSLv3
896 sslcontext.check_hostname = False
897 sslcontext.verify_mode = ssl.CERT_NONE
898 try:
899 sslcontext.options |= ssl.OP_NO_COMPRESSION
900 except AttributeError as attr_err:
901 warnings.warn(
902 "{!s}: The Python interpreter is compiled "
903 "against OpenSSL < 1.0.0. Ref: "
904 "https://docs.python.org/3/library/ssl.html"
905 "#ssl.OP_NO_COMPRESSION".format(attr_err),
906 )
907 sslcontext.set_default_verify_paths()
908 return sslcontext
910 def _get_ssl_context(self, req: ClientRequest) -> Optional[SSLContext]:
911 """Logic to get the correct SSL context
913 0. if req.ssl is false, return None
915 1. if ssl_context is specified in req, use it
916 2. if _ssl_context is specified in self, use it
917 3. otherwise:
918 1. if verify_ssl is not specified in req, use self.ssl_context
919 (will generate a default context according to self.verify_ssl)
920 2. if verify_ssl is True in req, generate a default SSL context
921 3. if verify_ssl is False in req, generate a SSL context that
922 won't verify
923 """
924 if req.is_ssl():
925 if ssl is None: # pragma: no cover
926 raise RuntimeError("SSL is not supported.")
927 sslcontext = req.ssl
928 if isinstance(sslcontext, ssl.SSLContext):
929 return sslcontext
930 if sslcontext is not None:
931 # not verified or fingerprinted
932 return self._make_ssl_context(False)
933 sslcontext = self._ssl
934 if isinstance(sslcontext, ssl.SSLContext):
935 return sslcontext
936 if sslcontext is not None:
937 # not verified or fingerprinted
938 return self._make_ssl_context(False)
939 return self._make_ssl_context(True)
940 else:
941 return None
943 def _get_fingerprint(self, req: ClientRequest) -> Optional["Fingerprint"]:
944 ret = req.ssl
945 if isinstance(ret, Fingerprint):
946 return ret
947 ret = self._ssl
948 if isinstance(ret, Fingerprint):
949 return ret
950 return None
952 async def _wrap_create_connection(
953 self,
954 *args: Any,
955 req: ClientRequest,
956 timeout: "ClientTimeout",
957 client_error: Type[Exception] = ClientConnectorError,
958 **kwargs: Any,
959 ) -> Tuple[asyncio.Transport, ResponseHandler]:
960 try:
961 async with ceil_timeout(
962 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
963 ):
964 return await self._loop.create_connection(*args, **kwargs)
965 except cert_errors as exc:
966 raise ClientConnectorCertificateError(req.connection_key, exc) from exc
967 except ssl_errors as exc:
968 raise ClientConnectorSSLError(req.connection_key, exc) from exc
969 except OSError as exc:
970 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
971 raise
972 raise client_error(req.connection_key, exc) from exc
974 def _warn_about_tls_in_tls(
975 self,
976 underlying_transport: asyncio.Transport,
977 req: ClientRequest,
978 ) -> None:
979 """Issue a warning if the requested URL has HTTPS scheme."""
980 if req.request_info.url.scheme != "https":
981 return
983 asyncio_supports_tls_in_tls = getattr(
984 underlying_transport,
985 "_start_tls_compatible",
986 False,
987 )
989 if asyncio_supports_tls_in_tls:
990 return
992 warnings.warn(
993 "An HTTPS request is being sent through an HTTPS proxy. "
994 "This support for TLS in TLS is known to be disabled "
995 "in the stdlib asyncio. This is why you'll probably see "
996 "an error in the log below.\n\n"
997 "It is possible to enable it via monkeypatching. "
998 "For more details, see:\n"
999 "* https://bugs.python.org/issue37179\n"
1000 "* https://github.com/python/cpython/pull/28073\n\n"
1001 "You can temporarily patch this as follows:\n"
1002 "* https://docs.aiohttp.org/en/stable/client_advanced.html#proxy-support\n"
1003 "* https://github.com/aio-libs/aiohttp/discussions/6044\n",
1004 RuntimeWarning,
1005 source=self,
1006 # Why `4`? At least 3 of the calls in the stack originate
1007 # from the methods in this class.
1008 stacklevel=3,
1009 )
1011 async def _start_tls_connection(
1012 self,
1013 underlying_transport: asyncio.Transport,
1014 req: ClientRequest,
1015 timeout: "ClientTimeout",
1016 client_error: Type[Exception] = ClientConnectorError,
1017 ) -> Tuple[asyncio.BaseTransport, ResponseHandler]:
1018 """Wrap the raw TCP transport with TLS."""
1019 tls_proto = self._factory() # Create a brand new proto for TLS
1021 # Safety of the `cast()` call here is based on the fact that
1022 # internally `_get_ssl_context()` only returns `None` when
1023 # `req.is_ssl()` evaluates to `False` which is never gonna happen
1024 # in this code path. Of course, it's rather fragile
1025 # maintainability-wise but this is to be solved separately.
1026 sslcontext = cast(ssl.SSLContext, self._get_ssl_context(req))
1028 try:
1029 async with ceil_timeout(
1030 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1031 ):
1032 try:
1033 tls_transport = await self._loop.start_tls(
1034 underlying_transport,
1035 tls_proto,
1036 sslcontext,
1037 server_hostname=req.server_hostname or req.host,
1038 ssl_handshake_timeout=timeout.total,
1039 )
1040 except BaseException:
1041 # We need to close the underlying transport since
1042 # `start_tls()` probably failed before it had a
1043 # chance to do this:
1044 underlying_transport.close()
1045 raise
1046 except cert_errors as exc:
1047 raise ClientConnectorCertificateError(req.connection_key, exc) from exc
1048 except ssl_errors as exc:
1049 raise ClientConnectorSSLError(req.connection_key, exc) from exc
1050 except OSError as exc:
1051 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1052 raise
1053 raise client_error(req.connection_key, exc) from exc
1054 except TypeError as type_err:
1055 # Example cause looks like this:
1056 # TypeError: transport <asyncio.sslproto._SSLProtocolTransport
1057 # object at 0x7f760615e460> is not supported by start_tls()
1059 raise ClientConnectionError(
1060 "Cannot initialize a TLS-in-TLS connection to host "
1061 f"{req.host!s}:{req.port:d} through an underlying connection "
1062 f"to an HTTPS proxy {req.proxy!s} ssl:{req.ssl or 'default'} "
1063 f"[{type_err!s}]"
1064 ) from type_err
1065 else:
1066 if tls_transport is None:
1067 msg = "Failed to start TLS (possibly caused by closing transport)"
1068 raise client_error(req.connection_key, OSError(msg))
1069 tls_proto.connection_made(
1070 tls_transport
1071 ) # Kick the state machine of the new TLS protocol
1073 return tls_transport, tls_proto
1075 async def _create_direct_connection(
1076 self,
1077 req: ClientRequest,
1078 traces: List["Trace"],
1079 timeout: "ClientTimeout",
1080 *,
1081 client_error: Type[Exception] = ClientConnectorError,
1082 ) -> Tuple[asyncio.Transport, ResponseHandler]:
1083 sslcontext = self._get_ssl_context(req)
1084 fingerprint = self._get_fingerprint(req)
1086 host = req.url.raw_host
1087 assert host is not None
1088 # Replace multiple trailing dots with a single one.
1089 # A trailing dot is only present for fully-qualified domain names.
1090 # See https://github.com/aio-libs/aiohttp/pull/7364.
1091 if host.endswith(".."):
1092 host = host.rstrip(".") + "."
1093 port = req.port
1094 assert port is not None
1095 host_resolved = asyncio.ensure_future(
1096 self._resolve_host(host, port, traces=traces), loop=self._loop
1097 )
1098 try:
1099 # Cancelling this lookup should not cancel the underlying lookup
1100 # or else the cancel event will get broadcast to all the waiters
1101 # across all connections.
1102 hosts = await asyncio.shield(host_resolved)
1103 except asyncio.CancelledError:
1105 def drop_exception(fut: "asyncio.Future[List[Dict[str, Any]]]") -> None:
1106 with suppress(Exception, asyncio.CancelledError):
1107 fut.result()
1109 host_resolved.add_done_callback(drop_exception)
1110 raise
1111 except OSError as exc:
1112 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1113 raise
1114 # in case of proxy it is not ClientProxyConnectionError
1115 # it is problem of resolving proxy ip itself
1116 raise ClientConnectorError(req.connection_key, exc) from exc
1118 last_exc: Optional[Exception] = None
1120 for hinfo in hosts:
1121 host = hinfo["host"]
1122 port = hinfo["port"]
1124 # Strip trailing dots, certificates contain FQDN without dots.
1125 # See https://github.com/aio-libs/aiohttp/issues/3636
1126 server_hostname = (
1127 (req.server_hostname or hinfo["hostname"]).rstrip(".")
1128 if sslcontext
1129 else None
1130 )
1132 try:
1133 transp, proto = await self._wrap_create_connection(
1134 self._factory,
1135 host,
1136 port,
1137 timeout=timeout,
1138 ssl=sslcontext,
1139 family=hinfo["family"],
1140 proto=hinfo["proto"],
1141 flags=hinfo["flags"],
1142 server_hostname=server_hostname,
1143 local_addr=self._local_addr,
1144 req=req,
1145 client_error=client_error,
1146 )
1147 except ClientConnectorError as exc:
1148 last_exc = exc
1149 continue
1151 if req.is_ssl() and fingerprint:
1152 try:
1153 fingerprint.check(transp)
1154 except ServerFingerprintMismatch as exc:
1155 transp.close()
1156 if not self._cleanup_closed_disabled:
1157 self._cleanup_closed_transports.append(transp)
1158 last_exc = exc
1159 continue
1161 return transp, proto
1162 assert last_exc is not None
1163 raise last_exc
1165 async def _create_proxy_connection(
1166 self, req: ClientRequest, traces: List["Trace"], timeout: "ClientTimeout"
1167 ) -> Tuple[asyncio.BaseTransport, ResponseHandler]:
1168 headers: Dict[str, str] = {}
1169 if req.proxy_headers is not None:
1170 headers = req.proxy_headers # type: ignore[assignment]
1171 headers[hdrs.HOST] = req.headers[hdrs.HOST]
1173 url = req.proxy
1174 assert url is not None
1175 proxy_req = ClientRequest(
1176 hdrs.METH_GET,
1177 url,
1178 headers=headers,
1179 auth=req.proxy_auth,
1180 loop=self._loop,
1181 ssl=req.ssl,
1182 )
1184 # create connection to proxy server
1185 transport, proto = await self._create_direct_connection(
1186 proxy_req, [], timeout, client_error=ClientProxyConnectionError
1187 )
1189 # Many HTTP proxies has buggy keepalive support. Let's not
1190 # reuse connection but close it after processing every
1191 # response.
1192 proto.force_close()
1194 auth = proxy_req.headers.pop(hdrs.AUTHORIZATION, None)
1195 if auth is not None:
1196 if not req.is_ssl():
1197 req.headers[hdrs.PROXY_AUTHORIZATION] = auth
1198 else:
1199 proxy_req.headers[hdrs.PROXY_AUTHORIZATION] = auth
1201 if req.is_ssl():
1202 self._warn_about_tls_in_tls(transport, req)
1204 # For HTTPS requests over HTTP proxy
1205 # we must notify proxy to tunnel connection
1206 # so we send CONNECT command:
1207 # CONNECT www.python.org:443 HTTP/1.1
1208 # Host: www.python.org
1209 #
1210 # next we must do TLS handshake and so on
1211 # to do this we must wrap raw socket into secure one
1212 # asyncio handles this perfectly
1213 proxy_req.method = hdrs.METH_CONNECT
1214 proxy_req.url = req.url
1215 key = dataclasses.replace(
1216 req.connection_key, proxy=None, proxy_auth=None, proxy_headers_hash=None
1217 )
1218 conn = Connection(self, key, proto, self._loop)
1219 proxy_resp = await proxy_req.send(conn)
1220 try:
1221 protocol = conn._protocol
1222 assert protocol is not None
1224 # read_until_eof=True will ensure the connection isn't closed
1225 # once the response is received and processed allowing
1226 # START_TLS to work on the connection below.
1227 protocol.set_response_params(
1228 read_until_eof=True,
1229 timeout_ceil_threshold=self._timeout_ceil_threshold,
1230 )
1231 resp = await proxy_resp.start(conn)
1232 except BaseException:
1233 proxy_resp.close()
1234 conn.close()
1235 raise
1236 else:
1237 conn._protocol = None
1238 conn._transport = None
1239 try:
1240 if resp.status != 200:
1241 message = resp.reason
1242 if message is None:
1243 message = HTTPStatus(resp.status).phrase
1244 raise ClientHttpProxyError(
1245 proxy_resp.request_info,
1246 resp.history,
1247 status=resp.status,
1248 message=message,
1249 headers=resp.headers,
1250 )
1251 except BaseException:
1252 # It shouldn't be closed in `finally` because it's fed to
1253 # `loop.start_tls()` and the docs say not to touch it after
1254 # passing there.
1255 transport.close()
1256 raise
1258 return await self._start_tls_connection(
1259 # Access the old transport for the last time before it's
1260 # closed and forgotten forever:
1261 transport,
1262 req=req,
1263 timeout=timeout,
1264 )
1265 finally:
1266 proxy_resp.close()
1268 return transport, proto
1271class UnixConnector(BaseConnector):
1272 """Unix socket connector.
1274 path - Unix socket path.
1275 keepalive_timeout - (optional) Keep-alive timeout.
1276 force_close - Set to True to force close and do reconnect
1277 after each request (and between redirects).
1278 limit - The total number of simultaneous connections.
1279 limit_per_host - Number of simultaneous connections to one host.
1280 loop - Optional event loop.
1281 """
1283 def __init__(
1284 self,
1285 path: str,
1286 force_close: bool = False,
1287 keepalive_timeout: Union[_SENTINEL, float, None] = sentinel,
1288 limit: int = 100,
1289 limit_per_host: int = 0,
1290 ) -> None:
1291 super().__init__(
1292 force_close=force_close,
1293 keepalive_timeout=keepalive_timeout,
1294 limit=limit,
1295 limit_per_host=limit_per_host,
1296 )
1297 self._path = path
1299 @property
1300 def path(self) -> str:
1301 """Path to unix socket."""
1302 return self._path
1304 async def _create_connection(
1305 self, req: ClientRequest, traces: List["Trace"], timeout: "ClientTimeout"
1306 ) -> ResponseHandler:
1307 try:
1308 async with ceil_timeout(
1309 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1310 ):
1311 _, proto = await self._loop.create_unix_connection(
1312 self._factory, self._path
1313 )
1314 except OSError as exc:
1315 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1316 raise
1317 raise UnixClientConnectorError(self.path, req.connection_key, exc) from exc
1319 return proto
1322class NamedPipeConnector(BaseConnector):
1323 """Named pipe connector.
1325 Only supported by the proactor event loop.
1326 See also: https://docs.python.org/3/library/asyncio-eventloop.html
1328 path - Windows named pipe path.
1329 keepalive_timeout - (optional) Keep-alive timeout.
1330 force_close - Set to True to force close and do reconnect
1331 after each request (and between redirects).
1332 limit - The total number of simultaneous connections.
1333 limit_per_host - Number of simultaneous connections to one host.
1334 loop - Optional event loop.
1335 """
1337 def __init__(
1338 self,
1339 path: str,
1340 force_close: bool = False,
1341 keepalive_timeout: Union[_SENTINEL, float, None] = sentinel,
1342 limit: int = 100,
1343 limit_per_host: int = 0,
1344 ) -> None:
1345 super().__init__(
1346 force_close=force_close,
1347 keepalive_timeout=keepalive_timeout,
1348 limit=limit,
1349 limit_per_host=limit_per_host,
1350 )
1351 if not isinstance(
1352 self._loop, asyncio.ProactorEventLoop # type: ignore[attr-defined]
1353 ):
1354 raise RuntimeError(
1355 "Named Pipes only available in proactor " "loop under windows"
1356 )
1357 self._path = path
1359 @property
1360 def path(self) -> str:
1361 """Path to the named pipe."""
1362 return self._path
1364 async def _create_connection(
1365 self, req: ClientRequest, traces: List["Trace"], timeout: "ClientTimeout"
1366 ) -> ResponseHandler:
1367 try:
1368 async with ceil_timeout(
1369 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
1370 ):
1371 _, proto = await self._loop.create_pipe_connection( # type: ignore[attr-defined]
1372 self._factory, self._path
1373 )
1374 # the drain is required so that the connection_made is called
1375 # and transport is set otherwise it is not set before the
1376 # `assert conn.transport is not None`
1377 # in client.py's _request method
1378 await asyncio.sleep(0)
1379 # other option is to manually set transport like
1380 # `proto.transport = trans`
1381 except OSError as exc:
1382 if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
1383 raise
1384 raise ClientConnectorError(req.connection_key, exc) from exc
1386 return cast(ResponseHandler, proto)