Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiofastnet/api

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.

7from __future__ import annotations

9import asyncio

10import errno

11import socket

12import ssl

13import weakref

14from asyncio.trsock import TransportSocket

15from logging import getLogger

16from typing import Any, Callable

18from . import constants, openssl_compat

19from .constants import SSL_BIO_SIZE_DEFAULTS, SSL_TIMEOUT_DEFAULTS

20from .selector_transport import SelectorSocketTransport

21from .ssl_transport import SSLTransport_Socket, SSLTransport_Transport

22from .transport import aiofn_is_buffered_protocol

23from .wrapped_transport import _get_original_loop_method, _should_fallback_to_asyncio, _WrappedBufferedProtocol, _WrappedProtocol

25_HAS_IPv6 = hasattr(socket, 'AF_INET6')

26_logger = getLogger('aiofastnet')

29def _is_asyncio_loop(loop: asyncio.AbstractEventLoop) -> bool:

30 return type(loop).__module__.startswith("asyncio.")

33def _validate_ssl_timeout(name: str, value: float | None, ssl_or_sslcontext: bool | ssl.SSLContext | None) -> float:

34 if value is not None and not ssl_or_sslcontext:

35 raise ValueError(

36 f'{name} is only meaningful with ssl')

38 if value is not None and value <= 0:

39 raise ValueError(f"{name} should be a positive number, got {value}")

41 if value is None:

42 return SSL_TIMEOUT_DEFAULTS[name]

44 return value

47def _validate_bio_size(name: str, value: int | None, ssl_or_sslcontext: bool | ssl.SSLContext | None) -> int:

48 if value is not None and not ssl_or_sslcontext:

49 raise ValueError(

50 f'{name} is only meaningful with ssl')

52 if value is not None and value < 16384:

53 raise ValueError(f"{name} should be a positive number >= 16384, got {value}")

55 if value is None:

56 return SSL_BIO_SIZE_DEFAULTS[name]

58 return value

61def _ssl_needs_fallback_engine(sslcontext: ssl.SSLContext) -> bool:

62 return openssl_compat.OPENSSL_DYN_LIBS is None or getattr(sslcontext, "_aiofastnet_force_fallback_ssl", False)

65async def _wait_and_close_transport_on_exc(waiter: asyncio.Future[Any], transport: Any) -> Any:

66 try:

67 return await waiter

68 except:

69 transport.close()

70 raise

73async def _create_connection_transport(

74 loop: asyncio.AbstractEventLoop,

75 sock: socket.socket,

76 protocol_factory: Callable[[], asyncio.BaseProtocol],

77 ssl: bool | ssl.SSLContext | None,

78 server_hostname: str | None=None,

79 server_side: bool=False,

80 ssl_handshake_timeout: float | None=None,

81 ssl_shutdown_timeout: float | None=None,

82 ssl_incoming_bio_size: int | None=None,

83 ssl_outgoing_bio_size: int | None=None,

84 server=None

85) -> tuple[asyncio.Transport, asyncio.BaseProtocol]:

86 sock.setblocking(False)

88 # The following big nested if-else should set transport, protocol, and

89 # optionally waiter variables

90 waiter = None

91 if _should_fallback_to_asyncio(loop):

92 if ssl:

93 protocol = protocol_factory()

94 waiter = loop.create_future() if server is None else None

95 sslcontext = None if isinstance(ssl, bool) else ssl

97 ssl_transport = SSLTransport_Transport(

98 loop, protocol, sslcontext,

99 server_side,

100 ssl_handshake_timeout,

101 ssl_shutdown_timeout,

102 ssl_incoming_bio_size,

103 ssl_outgoing_bio_size,

104 waiter=waiter,

105 server_hostname=server_hostname

106 )

107

108 ssl_protocol_factory = ssl_transport.get_tls_protocol

109

110 create_connection = _get_original_loop_method(loop, "create_connection")

111 await create_connection(ssl_protocol_factory, None, None, sock=sock)

112

113 transport = ssl_transport

114 else:

115 def wrapped_protocol_factory():

116 user_protocol = protocol_factory()

117 if aiofn_is_buffered_protocol(user_protocol):

118 return _WrappedBufferedProtocol(user_protocol)

119 else:

120 return _WrappedProtocol(user_protocol)

121

122 create_connection = _get_original_loop_method(loop, "create_connection")

123 loop_transport, wrapped_protocol = await create_connection(

124 wrapped_protocol_factory, None, None, sock=sock)

125

126 transport = wrapped_protocol._wrapped_transport

127 protocol = wrapped_protocol._protocol

128 wrapped_protocol._wrapped_transport = None

129

130 if server is not None:

131 # asyncio Transport needs _server in order to detach itself on disconnect

132 loop_transport._server = server

133 # and the Server must attach it, not WrappedTransport

134 transport = loop_transport

135 else:

136 protocol = protocol_factory()

137 waiter = loop.create_future() if server is None else None

138 if ssl:

139 sslcontext = openssl_compat.create_transport_context(server_side, server_hostname) if isinstance(ssl, bool) else ssl

140 if _ssl_needs_fallback_engine(sslcontext):

141 transport = SSLTransport_Transport(

142 loop, protocol, sslcontext,

143 server_side,

144 ssl_handshake_timeout,

145 ssl_shutdown_timeout,

146 ssl_incoming_bio_size,

147 ssl_outgoing_bio_size,

148 waiter=waiter,

149 server_hostname=server_hostname,

150 server=server

151 )

152 SelectorSocketTransport(loop, sock, transport.get_tls_protocol())

153 else:

154 transport = SSLTransport_Socket(

155 loop, protocol, sslcontext,

156 server_side,

157 ssl_handshake_timeout,

158 ssl_shutdown_timeout,

159 ssl_incoming_bio_size,

160 ssl_outgoing_bio_size,

161 sock,

162 waiter=waiter,

163 server_hostname=server_hostname,

164 server=server

165 )

166 else:

167 transport = SelectorSocketTransport(loop, sock, protocol,

168 waiter=waiter, server=server)

169

170 if waiter is not None:

171 try:

172 await waiter

173 except:

174 transport.close()

175 # gh-109534: When an exception is raised by the SSLProtocol object the

176 # exception set in this future can keep the protocol object alive and

177 # cause a reference cycle.

178 waiter = None

179 raise

180

181 return transport, protocol

182

183

184def _check_non_ssl_socket(sock):

185 if isinstance(sock, ssl.SSLSocket):

186 raise TypeError("Socket cannot be of type SSLSocket")

187

188

189async def _ensure_resolved(address, *,

190 family=0, type=socket.SOCK_STREAM,

191 proto=0, flags=0, loop):

192 host, port = address[:2]

193 info = _ipaddr_info(host, port, family, type, proto, *address[2:])

194 if info is not None:

195 # "host" is already a resolved IP.

196 return [info]

197 else:

198 return await loop.getaddrinfo(host, port, family=family, type=type,

199 proto=proto, flags=flags)

200

201

202def _ipaddr_info(host, port, family, type, proto, flowinfo=0, scopeid=0):

203 # Try to skip getaddrinfo if "host" is already an IP. Users might have

204 # handled name resolution in their own code and pass in resolved IPs.

205 if not hasattr(socket, 'inet_pton'):

206 return

207

208 if proto not in {0, socket.IPPROTO_TCP, socket.IPPROTO_UDP} or \

209 host is None:

210 return None

211

212 if type == socket.SOCK_STREAM:

213 proto = socket.IPPROTO_TCP

214 elif type == socket.SOCK_DGRAM:

215 proto = socket.IPPROTO_UDP

216 else:

217 return None

218

219 if port is None:

220 port = 0

221 elif isinstance(port, bytes) and port == b'':

222 port = 0

223 elif isinstance(port, str) and port == '':

224 port = 0

225 else:

226 # If port's a service name like "http", don't skip getaddrinfo.

227 try:

228 port = int(port)

229 except (TypeError, ValueError):

230 return None

231

232 if family == socket.AF_UNSPEC:

233 afs = [socket.AF_INET]

234 if _HAS_IPv6:

235 afs.append(socket.AF_INET6)

236 else:

237 afs = [family]

238

239 if isinstance(host, bytes):

240 host = host.decode('idna')

241 if '%' in host:

242 # Linux's inet_pton doesn't accept an IPv6 zone index after host,

243 # like '::1%lo0'.

244 return None

245

246 for af in afs:

247 try:

248 socket.inet_pton(af, host)

249 # The host has already been resolved.

250 if _HAS_IPv6 and af == socket.AF_INET6:

251 return af, type, proto, '', (host, port, flowinfo, scopeid)

252 else:

253 return af, type, proto, '', (host, port)

254 except OSError:

255 pass

256

257 # "host" is not an IP address.

258 return None

259

260

261class Server(asyncio.AbstractServer):

262 def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog,

263 ssl_handshake_timeout, ssl_shutdown_timeout,

264 ssl_incoming_bio_size, ssl_outgoing_bio_size

265 ):

266 self._loop = loop

267 self._sockets = sockets

268 # Weak references so we don't break Transport's ability to

269 # detect abandoned transports

270 self._clients = weakref.WeakSet()

271 self._waiters = []

272 self._protocol_factory = protocol_factory

273 self._backlog = backlog

274 self._ssl_context = ssl_context

275 self._ssl_handshake_timeout = ssl_handshake_timeout

276 self._ssl_shutdown_timeout = ssl_shutdown_timeout

277 self._ssl_incoming_bio_size = ssl_incoming_bio_size

278 self._ssl_outgoing_bio_size = ssl_outgoing_bio_size

279 self._serving = False

280 self._serving_forever_fut = None

281

282 def __repr__(self):

283 return f'<{self.__class__.__name__} sockets={self.sockets!r}>'

284

285 def _attach(self, transport):

286 assert self._sockets is not None

287 self._clients.add(transport)

288

289 def _detach(self, transport):

290 self._clients.discard(transport)

291 if len(self._clients) == 0 and self._sockets is None:

292 self._wakeup()

293

294 def _wakeup(self):

295 waiters = self._waiters

296 if waiters is None:

297 return

298 self._waiters = None

299 for waiter in waiters:

300 if not waiter.done():

301 waiter.set_result(None)

302

303 def _start_serving(self):

304 if self._serving:

305 return

306 self._serving = True

307 for sock in self._sockets:

308 sock.listen(self._backlog)

309 self._start_serving_one_listener(sock)

310

311 def _start_serving_one_listener(self, listening_sock):

312 self._loop.add_reader(listening_sock.fileno(), self._accept_connection, listening_sock)

313

314 def _accept_connection(self, listening_sock):

315 # This method is only called once for each event loop tick where the

316 # listening socket has triggered an EVENT_READ. There may be multiple

317 # connections waiting for an .accept() so it is called in a loop.

318 # See https://bugs.python.org/issue27906 for more details.

319 for _ in range(self._backlog + 1):

320 try:

321 conn, addr = listening_sock.accept()

322 if self._loop.get_debug():

323 _logger.debug("%r got a new connection from %r: %r", self, addr, conn)

324 conn.setblocking(False)

325 except ConnectionAbortedError:

326 # Discard connections that were aborted before accept().

327 continue

328 except (BlockingIOError, InterruptedError):

329 # Early exit because of a signal or

330 # the socket accept buffer is empty.

331 return

332 except OSError as exc:

333 # There's nowhere to send the error, so just log it.

334 if exc.errno in (errno.EMFILE, errno.ENFILE, errno.ENOBUFS, errno.ENOMEM):

335 # Some platforms (e.g. Linux keep reporting the FD as

336 # ready, so we remove the read handler temporarily.

337 # We'll try again in a while.

338 self._loop.call_exception_handler(

339 {

340 "message": "socket.accept() out of system resource",

341 "exception": exc,

342 "socket": TransportSocket(listening_sock),

343 }

344 )

345 listening_sock.remove_reader(listening_sock.fileno())

346 self._loop.call_later(constants.ACCEPT_RETRY_DELAY, self._start_serving_one_listener, listening_sock)

347 else:

348 raise # The event loop will catch, log and ignore it.

349 else:

350 asyncio.create_task(self._accept_connection2(conn))

351

352 async def _accept_connection2(self, sock):

353 # By the time _accept_connection2 is called, server can be already closed

354 # In such case we just close socket and return

355 if self._sockets is None:

356 sock.close()

357 return

358

359 try:

360 transport, _ = await _create_connection_transport(

361 self._loop,

362 sock,

363 self._protocol_factory,

364 self._ssl_context,

365 server_hostname=None,

366 server_side=True,

367 ssl_handshake_timeout=self._ssl_handshake_timeout,

368 ssl_shutdown_timeout=self._ssl_shutdown_timeout,

369 ssl_incoming_bio_size=self._ssl_incoming_bio_size,

370 ssl_outgoing_bio_size=self._ssl_outgoing_bio_size,

371 server=self,

372 )

373 except (SystemExit, KeyboardInterrupt):

374 raise

375 except BaseException as exc:

376 sock.close()

377 if self._loop.get_debug():

378 context = {

379 "message": "Error on transport creation for incoming connection",

380 "exception": exc,

381 }

382 self._loop.call_exception_handler(context)

383

384 return

385

386 # After await _create_connection_transport the server can be already closed

387 # Abort transport and return then.

388 if self._sockets is None:

389 transport.abort()

390 return

391

392 self._attach(transport)

393

394 def get_loop(self):

395 return self._loop

396

397 def is_serving(self):

398 return self._serving

399

400 @property

401 def sockets(self):

402 if self._sockets is None:

403 return ()

404 return tuple(asyncio.trsock.TransportSocket(s) for s in self._sockets)

405

406 def close(self):

407 sockets = self._sockets

408 if sockets is None:

409 return

410 self._sockets = None

411

412 for sock in sockets:

413 self._loop.remove_reader(sock.fileno())

414 sock.close()

415

416 self._serving = False

417

418 if (self._serving_forever_fut is not None and

419 not self._serving_forever_fut.done()):

420 self._serving_forever_fut.cancel()

421 self._serving_forever_fut = None

422

423 if len(self._clients) == 0:

424 self._wakeup()

425

426 def close_clients(self):

427 for transport in self._clients.copy():

428 transport.close()

429

430 def abort_clients(self):

431 for transport in self._clients.copy():

432 transport.abort()

433

434 async def start_serving(self):

435 self._start_serving()

436 # Skip one loop iteration so that all 'loop.add_reader'

437 # go through.

438 await asyncio.sleep(0)

439

440 async def serve_forever(self):

441 if self._serving_forever_fut is not None:

442 raise RuntimeError(

443 f'server {self!r} is already being awaited on serve_forever()')

444 if self._sockets is None:

445 raise RuntimeError(f'server {self!r} is closed')

446

447 self._start_serving()

448 self._serving_forever_fut = self._loop.create_future()

449

450 try:

451 await self._serving_forever_fut

452 except asyncio.CancelledError:

453 try:

454 self.close()

455 await self.wait_closed()

456 finally:

457 raise

458 finally:

459 self._serving_forever_fut = None

460

461 async def wait_closed(self):

462 """Wait until server is closed and all connections are dropped.

463

464 - If the server is not closed, wait.

465 - If it is closed, but there are still active connections, wait.

466

467 Anyone waiting here will be unblocked once both conditions

468 (server is closed and all connections have been dropped)

469 have become true, in either order.

470

471 Historical note: In 3.11 and before, this was broken, returning

472 immediately if the server was already closed, even if there

473 were still active connections. An attempted fix in 3.12.0 was

474 still broken, returning immediately if the server was still

475 open and there were no active connections. Hopefully in 3.12.1

476 we have it right.

477 """

478 # Waiters are unblocked by self._wakeup(), which is called

479 # from two places: self.close() and self._detach(), but only

480 # when both conditions have become true. To signal that this

481 # has happened, self._wakeup() sets self._waiters to None.

482 if self._waiters is None:

483 return

484 waiter = self._loop.create_future()

485 self._waiters.append(waiter)

486 await waiter

487

488

489def _stop_serving(loop, sock):

490 loop.remove_reader(sock.fileno())

491 sock.close()

492

493

494def _set_reuseport(sock):

495 if not hasattr(socket, 'SO_REUSEPORT'):

496 raise ValueError('reuse_port not supported by socket module')

497 else:

498 try:

499 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)

500 except OSError:

501 raise ValueError('reuse_port not supported by socket module, '

502 'SO_REUSEPORT defined but not implemented.')

503

504

505def _check_nonblocking_socket(py_sock):

506 if py_sock.getblocking():

507 raise ValueError("the socket must be non-blocking")

Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiofastnet/api_utils.py: 5%

277 statements