Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiofastnet/api_utils.py: 4%
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
1# Portions of this file are derived from CPython's asyncio sources
2# (notably asyncio.base_events and asyncio.selector_events).
3# Copyright (c) Python Software Foundation.
4# Licensed under the Python Software Foundation License Version 2.
5# See LICENSES/PSF-2.0.txt and THIRD_PARTY_NOTICES for details.
7import asyncio
8import errno
9import socket
10import ssl
11import weakref
12from asyncio.trsock import TransportSocket
13from logging import getLogger
14from typing import Callable, Union, Optional, Tuple
16from . import constants, openssl_compat
17from .constants import SSL_TIMEOUT_DEFAULTS, SSL_BIO_SIZE_DEFAULTS
18from .ssl_transport import SSLTransport_Socket, SSLTransport_Transport
19from .transport import SocketTransport, aiofn_is_buffered_protocol
20from .wrapped_transport import _should_fallback_to_asyncio, \
21 _WrappedBufferedProtocol, _WrappedProtocol, _get_original_loop_method
24_HAS_IPv6 = hasattr(socket, 'AF_INET6')
25_logger = getLogger('aiofastnet')
28def _is_asyncio_loop(loop: asyncio.AbstractEventLoop) -> bool:
29 return type(loop).__module__.startswith("asyncio.")
32def _validate_ssl_timeout(name: str, value: Optional[float], ssl_or_sslcontext: Optional[Union[bool, ssl.SSLContext]]) -> float:
33 if value is not None and not ssl_or_sslcontext:
34 raise ValueError(
35 f'{name} is only meaningful with ssl')
37 if value is not None and value <= 0:
38 raise ValueError(f"{name} should be a positive number, got {value}")
40 if value is None:
41 return SSL_TIMEOUT_DEFAULTS[name]
43 return value
46def _validate_bio_size(name: str, value: Optional[int], ssl_or_sslcontext: Optional[Union[bool, ssl.SSLContext]]) -> int:
47 if value is not None and not ssl_or_sslcontext:
48 raise ValueError(
49 f'{name} is only meaningful with ssl')
51 if value is not None and value < 16384:
52 raise ValueError(f"{name} should be a positive number >= 16384, got {value}")
54 if value is None:
55 return SSL_BIO_SIZE_DEFAULTS[name]
57 return value
60def _ssl_needs_fallback_engine(sslcontext: ssl.SSLContext) -> bool:
61 return openssl_compat.OPENSSL_DYN_LIBS is None or getattr(sslcontext, "_aiofastnet_force_fallback_ssl", False)
64async def _create_connection_transport(
65 loop: asyncio.AbstractEventLoop,
66 sock: socket.socket,
67 protocol_factory: Callable[[], asyncio.BaseProtocol],
68 ssl: Union[bool, ssl.SSLContext, None],
69 server_hostname: Optional[str]=None,
70 server_side: bool=False,
71 ssl_handshake_timeout: Optional[float]=None,
72 ssl_shutdown_timeout: Optional[float]=None,
73 ssl_incoming_bio_size: Optional[int]=None,
74 ssl_outgoing_bio_size: Optional[int]=None,
75 server=None
76) -> Tuple[asyncio.Transport, asyncio.BaseProtocol]:
77 sock.setblocking(False)
79 # The following big nested if-else should set transport, protocol, and
80 # optionally waiter variables
81 waiter = None
82 if _should_fallback_to_asyncio(loop):
83 if ssl:
84 protocol = protocol_factory()
85 waiter = loop.create_future()
86 sslcontext = None if isinstance(ssl, bool) else ssl
88 ssl_transport = SSLTransport_Transport(
89 loop, protocol, sslcontext,
90 server_side,
91 ssl_handshake_timeout,
92 ssl_shutdown_timeout,
93 ssl_incoming_bio_size,
94 ssl_outgoing_bio_size,
95 waiter=waiter,
96 server_hostname=server_hostname
97 )
99 ssl_protocol_factory = ssl_transport.get_tls_protocol
101 create_connection = _get_original_loop_method(loop, "create_connection")
102 loop_transport, ssl_protocol = await create_connection(
103 ssl_protocol_factory, None, None, sock=sock)
104 transport = ssl_transport
105 else:
106 def wrapped_protocol_factory():
107 user_protocol = protocol_factory()
108 if aiofn_is_buffered_protocol(user_protocol):
109 return _WrappedBufferedProtocol(user_protocol)
110 else:
111 return _WrappedProtocol(user_protocol)
113 create_connection = _get_original_loop_method(loop, "create_connection")
114 loop_transport, wrapped_protocol = await create_connection(
115 wrapped_protocol_factory, None, None, sock=sock)
116 transport = wrapped_protocol._wrapped_transport
117 protocol = wrapped_protocol._protocol
118 wrapped_protocol._wrapped_transport = None
120 # Ugly but I don't know how else to attach conventional transport
121 # to my Server object
122 if server is not None:
123 loop_transport._server = server
124 server._attach(loop_transport)
125 else:
126 protocol = protocol_factory()
127 waiter = loop.create_future()
128 if ssl:
129 sslcontext = openssl_compat.create_transport_context(server_side, server_hostname) if isinstance(ssl, bool) else ssl
130 if _ssl_needs_fallback_engine(sslcontext):
131 ssl_transport = SSLTransport_Transport(
132 loop, protocol, sslcontext,
133 server_side,
134 ssl_handshake_timeout,
135 ssl_shutdown_timeout,
136 ssl_incoming_bio_size,
137 ssl_outgoing_bio_size,
138 waiter=waiter,
139 server_hostname=server_hostname
140 )
141 transport = ssl_transport
142 SocketTransport(loop, sock, ssl_transport.get_tls_protocol(), server=server)
143 else:
144 transport = SSLTransport_Socket(
145 loop, protocol, sslcontext,
146 server_side,
147 ssl_handshake_timeout,
148 ssl_shutdown_timeout,
149 ssl_incoming_bio_size,
150 ssl_outgoing_bio_size,
151 sock,
152 waiter=waiter,
153 server_hostname=server_hostname,
154 server=server
155 )
156 else:
157 transport = SocketTransport(loop, sock, protocol,
158 waiter=waiter, server=server)
160 if waiter is not None:
161 try:
162 await waiter
163 except:
164 transport.close()
165 # gh-109534: When an exception is raised by the SSLProtocol object the
166 # exception set in this future can keep the protocol object alive and
167 # cause a reference cycle.
168 waiter = None
169 raise
171 return transport, protocol
174def _check_ssl_socket(sock):
175 if isinstance(sock, ssl.SSLSocket):
176 raise TypeError("Socket cannot be of type SSLSocket")
179async def _ensure_resolved(address, *,
180 family=0, type=socket.SOCK_STREAM,
181 proto=0, flags=0, loop):
182 host, port = address[:2]
183 info = _ipaddr_info(host, port, family, type, proto, *address[2:])
184 if info is not None:
185 # "host" is already a resolved IP.
186 return [info]
187 else:
188 return await loop.getaddrinfo(host, port, family=family, type=type,
189 proto=proto, flags=flags)
192def _ipaddr_info(host, port, family, type, proto, flowinfo=0, scopeid=0):
193 # Try to skip getaddrinfo if "host" is already an IP. Users might have
194 # handled name resolution in their own code and pass in resolved IPs.
195 if not hasattr(socket, 'inet_pton'):
196 return
198 if proto not in {0, socket.IPPROTO_TCP, socket.IPPROTO_UDP} or \
199 host is None:
200 return None
202 if type == socket.SOCK_STREAM:
203 proto = socket.IPPROTO_TCP
204 elif type == socket.SOCK_DGRAM:
205 proto = socket.IPPROTO_UDP
206 else:
207 return None
209 if port is None:
210 port = 0
211 elif isinstance(port, bytes) and port == b'':
212 port = 0
213 elif isinstance(port, str) and port == '':
214 port = 0
215 else:
216 # If port's a service name like "http", don't skip getaddrinfo.
217 try:
218 port = int(port)
219 except (TypeError, ValueError):
220 return None
222 if family == socket.AF_UNSPEC:
223 afs = [socket.AF_INET]
224 if _HAS_IPv6:
225 afs.append(socket.AF_INET6)
226 else:
227 afs = [family]
229 if isinstance(host, bytes):
230 host = host.decode('idna')
231 if '%' in host:
232 # Linux's inet_pton doesn't accept an IPv6 zone index after host,
233 # like '::1%lo0'.
234 return None
236 for af in afs:
237 try:
238 socket.inet_pton(af, host)
239 # The host has already been resolved.
240 if _HAS_IPv6 and af == socket.AF_INET6:
241 return af, type, proto, '', (host, port, flowinfo, scopeid)
242 else:
243 return af, type, proto, '', (host, port)
244 except OSError:
245 pass
247 # "host" is not an IP address.
248 return None
251class Server(asyncio.AbstractServer):
252 def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog,
253 ssl_handshake_timeout, ssl_shutdown_timeout,
254 ssl_incoming_bio_size, ssl_outgoing_bio_size
255 ):
256 self._loop = loop
257 self._sockets = sockets
258 # Weak references so we don't break Transport's ability to
259 # detect abandoned transports
260 self._clients = weakref.WeakSet()
261 self._waiters = []
262 self._protocol_factory = protocol_factory
263 self._backlog = backlog
264 self._ssl_context = ssl_context
265 self._ssl_handshake_timeout = ssl_handshake_timeout
266 self._ssl_shutdown_timeout = ssl_shutdown_timeout
267 self._ssl_incoming_bio_size = ssl_incoming_bio_size
268 self._ssl_outgoing_bio_size = ssl_outgoing_bio_size
269 self._serving = False
270 self._serving_forever_fut = None
272 def __repr__(self):
273 return f'<{self.__class__.__name__} sockets={self.sockets!r}>'
275 def _attach(self, transport):
276 assert self._sockets is not None
277 self._clients.add(transport)
279 def _detach(self, transport):
280 self._clients.discard(transport)
281 if len(self._clients) == 0 and self._sockets is None:
282 self._wakeup()
284 def _wakeup(self):
285 waiters = self._waiters
286 if waiters is None:
287 return
288 self._waiters = None
289 for waiter in waiters:
290 if not waiter.done():
291 waiter.set_result(None)
293 def _start_serving(self):
294 if self._serving:
295 return
296 self._serving = True
297 for sock in self._sockets:
298 sock.listen(self._backlog)
299 _start_serving(
300 self._loop,
301 self._protocol_factory, sock, self._ssl_context,
302 self, self._backlog,
303 self._ssl_handshake_timeout,
304 self._ssl_shutdown_timeout,
305 self._ssl_incoming_bio_size,
306 self._ssl_outgoing_bio_size)
308 def get_loop(self):
309 return self._loop
311 def is_serving(self):
312 return self._serving
314 @property
315 def sockets(self):
316 if self._sockets is None:
317 return ()
318 return tuple(asyncio.trsock.TransportSocket(s) for s in self._sockets)
320 def close(self):
321 sockets = self._sockets
322 if sockets is None:
323 return
324 self._sockets = None
326 for sock in sockets:
327 _stop_serving(self._loop, sock)
329 self._serving = False
331 if (self._serving_forever_fut is not None and
332 not self._serving_forever_fut.done()):
333 self._serving_forever_fut.cancel()
334 self._serving_forever_fut = None
336 if len(self._clients) == 0:
337 self._wakeup()
339 def close_clients(self):
340 for transport in self._clients.copy():
341 transport.close()
343 def abort_clients(self):
344 for transport in self._clients.copy():
345 transport.abort()
347 async def start_serving(self):
348 self._start_serving()
349 # Skip one loop iteration so that all 'loop.add_reader'
350 # go through.
351 await asyncio.sleep(0)
353 async def serve_forever(self):
354 if self._serving_forever_fut is not None:
355 raise RuntimeError(
356 f'server {self!r} is already being awaited on serve_forever()')
357 if self._sockets is None:
358 raise RuntimeError(f'server {self!r} is closed')
360 self._start_serving()
361 self._serving_forever_fut = self._loop.create_future()
363 try:
364 await self._serving_forever_fut
365 except asyncio.CancelledError:
366 try:
367 self.close()
368 await self.wait_closed()
369 finally:
370 raise
371 finally:
372 self._serving_forever_fut = None
374 async def wait_closed(self):
375 """Wait until server is closed and all connections are dropped.
377 - If the server is not closed, wait.
378 - If it is closed, but there are still active connections, wait.
380 Anyone waiting here will be unblocked once both conditions
381 (server is closed and all connections have been dropped)
382 have become true, in either order.
384 Historical note: In 3.11 and before, this was broken, returning
385 immediately if the server was already closed, even if there
386 were still active connections. An attempted fix in 3.12.0 was
387 still broken, returning immediately if the server was still
388 open and there were no active connections. Hopefully in 3.12.1
389 we have it right.
390 """
391 # Waiters are unblocked by self._wakeup(), which is called
392 # from two places: self.close() and self._detach(), but only
393 # when both conditions have become true. To signal that this
394 # has happened, self._wakeup() sets self._waiters to None.
395 if self._waiters is None:
396 return
397 waiter = self._loop.create_future()
398 self._waiters.append(waiter)
399 await waiter
402def _accept_connection(
403 loop, protocol_factory, sock,
404 sslcontext, server,
405 backlog,
406 ssl_handshake_timeout,
407 ssl_shutdown_timeout,
408 ssl_incoming_bio_size,
409 ssl_outgoing_bio_size
410):
411 # This method is only called once for each event loop tick where the
412 # listening socket has triggered an EVENT_READ. There may be multiple
413 # connections waiting for an .accept() so it is called in a loop.
414 # See https://bugs.python.org/issue27906 for more details.
415 for _ in range(backlog + 1):
416 try:
417 conn, addr = sock.accept()
418 if loop.get_debug():
419 _logger.debug("%r got a new connection from %r: %r",
420 server, addr, conn)
421 conn.setblocking(False)
422 except ConnectionAbortedError:
423 # Discard connections that were aborted before accept().
424 continue
425 except (BlockingIOError, InterruptedError):
426 # Early exit because of a signal or
427 # the socket accept buffer is empty.
428 return
429 except OSError as exc:
430 # There's nowhere to send the error, so just log it.
431 if exc.errno in (errno.EMFILE, errno.ENFILE,
432 errno.ENOBUFS, errno.ENOMEM):
433 # Some platforms (e.g. Linux keep reporting the FD as
434 # ready, so we remove the read handler temporarily.
435 # We'll try again in a while.
436 loop.call_exception_handler({
437 'message': 'socket.accept() out of system resource',
438 'exception': exc,
439 'socket': TransportSocket(sock),
440 })
441 loop.remove_reader(sock.fileno())
442 loop.call_later(constants.ACCEPT_RETRY_DELAY,
443 _start_serving,
444 loop, protocol_factory, sock, sslcontext, server,
445 backlog,
446 ssl_handshake_timeout,
447 ssl_shutdown_timeout,
448 ssl_incoming_bio_size,
449 ssl_outgoing_bio_size
450 )
451 else:
452 raise # The event loop will catch, log and ignore it.
453 else:
454 accept = _accept_connection2(
455 loop, protocol_factory, conn, sslcontext, server,
456 ssl_handshake_timeout, ssl_shutdown_timeout,
457 ssl_incoming_bio_size, ssl_outgoing_bio_size
458 )
459 asyncio.create_task(accept)
462async def _accept_connection2(
463 loop,
464 protocol_factory,
465 sock,
466 sslcontext, server,
467 ssl_handshake_timeout,
468 ssl_shutdown_timeout,
469 ssl_incoming_bio_size,
470 ssl_outgoing_bio_size
471):
472 protocol = None
473 transport = None
474 try:
475 transport, protocol = await _create_connection_transport(
476 loop, sock, protocol_factory, sslcontext,
477 server_hostname=None, server_side=True,
478 ssl_handshake_timeout=ssl_handshake_timeout,
479 ssl_shutdown_timeout=ssl_shutdown_timeout,
480 ssl_incoming_bio_size=ssl_incoming_bio_size,
481 ssl_outgoing_bio_size=ssl_outgoing_bio_size,
482 server=server
483 )
484 except (SystemExit, KeyboardInterrupt):
485 raise
486 except BaseException as exc:
487 if transport is None:
488 sock.close()
489 if loop.get_debug():
490 context = {
491 'message':
492 'Error on transport creation for incoming connection',
493 'exception': exc,
494 }
495 if protocol is not None:
496 context['protocol'] = protocol
497 if transport is not None:
498 context['transport'] = transport
499 loop.call_exception_handler(context)
502def _start_serving(loop, protocol_factory, sock,
503 sslcontext, server, backlog,
504 ssl_handshake_timeout,
505 ssl_shutdown_timeout,
506 ssl_incoming_bio_size,
507 ssl_outgoing_bio_size,
508 ):
509 loop.add_reader(sock.fileno(), _accept_connection, loop,
510 protocol_factory, sock, sslcontext, server, backlog,
511 ssl_handshake_timeout, ssl_shutdown_timeout,
512 ssl_incoming_bio_size, ssl_outgoing_bio_size
513 )
516def _stop_serving(loop, sock):
517 loop.remove_reader(sock.fileno())
518 sock.close()