Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/anyio/_core/_sockets.py: 26%

1from __future__ import annotations 

2 

3import errno 

4import os 

5import socket 

6import ssl 

7import stat 

8import sys 

9from collections.abc import Awaitable 

10from dataclasses import dataclass 

11from ipaddress import IPv4Address, IPv6Address, ip_address 

12from os import PathLike, chmod 

13from socket import AddressFamily, SocketKind 

14from typing import TYPE_CHECKING, Any, Literal, cast, overload 

15 

16from .. import ConnectionFailed, to_thread 

17from ..abc import ( 

18 ByteStreamConnectable, 

19 ConnectedUDPSocket, 

20 ConnectedUNIXDatagramSocket, 

21 IPAddressType, 

22 IPSockAddrType, 

23 SocketListener, 

24 SocketStream, 

25 UDPSocket, 

26 UNIXDatagramSocket, 

27 UNIXSocketStream, 

28) 

29from ..streams.stapled import MultiListener 

30from ..streams.tls import TLSConnectable, TLSStream 

31from ._eventloop import get_async_backend 

32from ._resources import aclose_forcefully 

33from ._synchronization import Event 

34from ._tasks import create_task_group, move_on_after 

35 

36if TYPE_CHECKING: 

37 from _typeshed import FileDescriptorLike 

38else: 

39 FileDescriptorLike = object 

40 

41if sys.version_info < (3, 11): 

42 from exceptiongroup import ExceptionGroup 

43 

44if sys.version_info >= (3, 12): 

45 from typing import override 

46else: 

47 from typing_extensions import override 

48 

49if sys.version_info < (3, 13): 

50 from typing_extensions import deprecated 

51else: 

52 from warnings import deprecated 

53 

54IPPROTO_IPV6 = getattr(socket, "IPPROTO_IPV6", 41) # https://bugs.python.org/issue29515 

55 

56AnyIPAddressFamily = Literal[ 

57 AddressFamily.AF_UNSPEC, AddressFamily.AF_INET, AddressFamily.AF_INET6 

58] 

59IPAddressFamily = Literal[AddressFamily.AF_INET, AddressFamily.AF_INET6] 

60 

61 

62def idna2008_resolve(host: str) -> bytes: 

63 try: 

64 return host.encode("ascii") 

65 except UnicodeEncodeError: 

66 import idna 

67 

68 return idna.encode(host, uts46=True) 

69 

70 

71# tls_hostname given 

72@overload 

73async def connect_tcp( 

74 remote_host: IPAddressType, 

75 remote_port: int, 

76 *, 

77 local_host: IPAddressType | None = ..., 

78 local_port: int | None = ..., 

79 ssl_context: ssl.SSLContext | None = ..., 

80 tls_standard_compatible: bool = ..., 

81 tls_hostname: str, 

82 happy_eyeballs_delay: float = ..., 

83) -> TLSStream: ... 

84 

85 

86# ssl_context given 

87@overload 

88async def connect_tcp( 

89 remote_host: IPAddressType, 

90 remote_port: int, 

91 *, 

92 local_host: IPAddressType | None = ..., 

93 local_port: int | None = ..., 

94 ssl_context: ssl.SSLContext, 

95 tls_standard_compatible: bool = ..., 

96 tls_hostname: str | None = ..., 

97 happy_eyeballs_delay: float = ..., 

98) -> TLSStream: ... 

99 

100 

101# tls=True 

102@overload 

103async def connect_tcp( 

104 remote_host: IPAddressType, 

105 remote_port: int, 

106 *, 

107 local_host: IPAddressType | None = ..., 

108 local_port: int | None = ..., 

109 tls: Literal[True], 

110 ssl_context: ssl.SSLContext | None = ..., 

111 tls_standard_compatible: bool = ..., 

112 tls_hostname: str | None = ..., 

113 happy_eyeballs_delay: float = ..., 

114) -> TLSStream: ... 

115 

116 

117# tls=False 

118@overload 

119async def connect_tcp( 

120 remote_host: IPAddressType, 

121 remote_port: int, 

122 *, 

123 local_host: IPAddressType | None = ..., 

124 local_port: int | None = ..., 

125 tls: Literal[False], 

126 ssl_context: ssl.SSLContext | None = ..., 

127 tls_standard_compatible: bool = ..., 

128 tls_hostname: str | None = ..., 

129 happy_eyeballs_delay: float = ..., 

130) -> SocketStream: ... 

131 

132 

133# No TLS arguments 

134@overload 

135async def connect_tcp( 

136 remote_host: IPAddressType, 

137 remote_port: int, 

138 *, 

139 local_host: IPAddressType | None = ..., 

140 local_port: int | None = ..., 

141 happy_eyeballs_delay: float = ..., 

142) -> SocketStream: ... 

143 

144 

145async def connect_tcp( 

146 remote_host: IPAddressType, 

147 remote_port: int, 

148 *, 

149 local_host: IPAddressType | None = None, 

150 local_port: int | None = None, 

151 tls: bool = False, 

152 ssl_context: ssl.SSLContext | None = None, 

153 tls_standard_compatible: bool = True, 

154 tls_hostname: str | None = None, 

155 happy_eyeballs_delay: float = 0.25, 

156) -> SocketStream | TLSStream: 

157 """ 

158 Connect to a host using the TCP protocol. 

159 

160 This function implements the stateless version of the Happy Eyeballs algorithm (RFC 

161 6555). If ``remote_host`` is a host name that resolves to multiple IP addresses, 

162 each one is tried until one connection attempt succeeds. If the first attempt does 

163 not connected within 250 milliseconds, a second attempt is started using the next 

164 address in the list, and so on. On IPv6 enabled systems, an IPv6 address (if 

165 available) is tried first. 

166 

167 When the connection has been established, a TLS handshake will be done if either 

168 ``ssl_context`` or ``tls_hostname`` is not ``None``, or if ``tls`` is ``True``. 

169 

170 :param remote_host: the IP address or host name to connect to 

171 :param remote_port: port on the target host to connect to 

172 :param local_host: the interface address or name to bind the socket to before 

173 connecting 

174 :param local_port: the local port to bind to (requires ``local_host`` to also be 

175 set) 

176 :param tls: ``True`` to do a TLS handshake with the connected stream and return a 

177 :class:`~anyio.streams.tls.TLSStream` instead 

178 :param ssl_context: the SSL context object to use (if omitted, a default context is 

179 created) 

180 :param tls_standard_compatible: If ``True``, performs the TLS shutdown handshake 

181 before closing the stream and requires that the server does this as well. 

182 Otherwise, :exc:`~ssl.SSLEOFError` may be raised during reads from the stream. 

183 Some protocols, such as HTTP, require this option to be ``False``. 

184 See :meth:`~ssl.SSLContext.wrap_socket` for details. 

185 :param tls_hostname: host name to check the server certificate against (defaults to 

186 the value of ``remote_host``) 

187 :param happy_eyeballs_delay: delay (in seconds) before starting the next connection 

188 attempt 

189 :return: a socket stream object if no TLS handshake was done, otherwise a TLS stream 

190 :raises ConnectionFailed: if the connection fails 

191 

192 """ 

193 # Placed here due to https://github.com/python/mypy/issues/7057 

194 connected_stream: SocketStream | None = None 

195 

196 async def try_connect(remote_host: str, event: Event) -> None: 

197 nonlocal connected_stream 

198 try: 

199 stream = await asynclib.connect_tcp(remote_host, remote_port, local_address) 

200 except OSError as exc: 

201 oserrors.append(exc) 

202 return 

203 else: 

204 if connected_stream is None: 

205 connected_stream = stream 

206 tg.cancel_scope.cancel() 

207 else: 

208 await stream.aclose() 

209 finally: 

210 event.set() 

211 

212 asynclib = get_async_backend() 

213 local_address: IPSockAddrType | None = None 

214 family = socket.AF_UNSPEC 

215 if local_host: 

216 gai_res = await getaddrinfo(str(local_host), local_port) 

217 family, *_, local_address = gai_res[0] 

218 

219 target_host = str(remote_host) 

220 try: 

221 addr_obj = ip_address(remote_host) 

222 except ValueError: 

223 addr_obj = None 

224 

225 if addr_obj is not None: 

226 if isinstance(addr_obj, IPv6Address): 

227 target_addrs = [(socket.AF_INET6, addr_obj.compressed)] 

228 else: 

229 target_addrs = [(socket.AF_INET, addr_obj.compressed)] 

230 else: 

231 # getaddrinfo() will raise an exception if name resolution fails 

232 gai_res = await getaddrinfo( 

233 target_host, remote_port, family=family, type=socket.SOCK_STREAM 

234 ) 

235 

236 # Organize the list so that the first address is an IPv6 address (if available) 

237 # and the second one is an IPv4 addresses. The rest can be in whatever order. 

238 v6_found = v4_found = False 

239 target_addrs = [] 

240 for af, *_, sa in gai_res: 

241 if af == socket.AF_INET6 and not v6_found: 

242 v6_found = True 

243 target_addrs.insert(0, (af, sa[0])) 

244 elif af == socket.AF_INET and not v4_found and v6_found: 

245 v4_found = True 

246 target_addrs.insert(1, (af, sa[0])) 

247 else: 

248 target_addrs.append((af, sa[0])) 

249 

250 oserrors: list[OSError] = [] 

251 try: 

252 async with create_task_group() as tg: 

253 for _af, addr in target_addrs: 

254 event = Event() 

255 tg.start_soon(try_connect, addr, event) 

256 with move_on_after(happy_eyeballs_delay): 

257 await event.wait() 

258 

259 if connected_stream is None: 

260 cause = ( 

261 oserrors[0] 

262 if len(oserrors) == 1 

263 else ExceptionGroup("multiple connection attempts failed", oserrors) 

264 ) 

265 raise OSError("All connection attempts failed") from cause 

266 finally: 

267 oserrors.clear() 

268 

269 if tls or tls_hostname or ssl_context: 

270 try: 

271 return await TLSStream.wrap( 

272 connected_stream, 

273 server_side=False, 

274 hostname=tls_hostname or str(remote_host), 

275 ssl_context=ssl_context, 

276 standard_compatible=tls_standard_compatible, 

277 ) 

278 except BaseException: 

279 await aclose_forcefully(connected_stream) 

280 raise 

281 

282 return connected_stream 

283 

284 

285async def connect_unix(path: str | bytes | PathLike[Any]) -> UNIXSocketStream: 

286 """ 

287 Connect to the given UNIX socket. 

288 

289 Not available on Windows. 

290 

291 :param path: path to the socket 

292 :return: a socket stream object 

293 :raises ConnectionFailed: if the connection fails 

294 

295 """ 

296 path = os.fspath(path) 

297 return await get_async_backend().connect_unix(path) 

298 

299 

300async def create_tcp_listener( 

301 *, 

302 local_host: IPAddressType | None = None, 

303 local_port: int = 0, 

304 family: AnyIPAddressFamily = socket.AddressFamily.AF_UNSPEC, 

305 backlog: int = 65536, 

306 reuse_port: bool = False, 

307) -> MultiListener[SocketStream]: 

308 """ 

309 Create a TCP socket listener. 

310 

311 :param local_port: port number to listen on 

312 :param local_host: IP address of the interface to listen on. If omitted, listen on 

313 all IPv4 and IPv6 interfaces. To listen on all interfaces on a specific address 

314 family, use ``0.0.0.0`` for IPv4 or ``::`` for IPv6. 

315 :param family: address family (used if ``local_host`` was omitted) 

316 :param backlog: maximum number of queued incoming connections (up to a maximum of 

317 2**16, or 65536) 

318 :param reuse_port: ``True`` to allow multiple sockets to bind to the same 

319 address/port (not supported on Windows) 

320 :return: a multi-listener object containing one or more socket listeners 

321 :raises OSError: if there's an error creating a socket, or binding to one or more 

322 interfaces failed 

323 

324 """ 

325 asynclib = get_async_backend() 

326 backlog = min(backlog, 65536) 

327 local_host = str(local_host) if local_host is not None else None 

328 

329 def setup_raw_socket( 

330 fam: AddressFamily, 

331 bind_addr: tuple[str, int] | tuple[str, int, int, int], 

332 *, 

333 v6only: bool = True, 

334 ) -> socket.socket: 

335 sock = socket.socket(fam) 

336 try: 

337 sock.setblocking(False) 

338 

339 if fam == AddressFamily.AF_INET6: 

340 sock.setsockopt(IPPROTO_IPV6, socket.IPV6_V6ONLY, v6only) 

341 

342 # For Windows, enable exclusive address use. For others, enable address 

343 # reuse. 

344 if sys.platform == "win32": 

345 sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1) 

346 else: 

347 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 

348 

349 if reuse_port: 

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

351 

352 # Workaround for #554 

353 if fam == socket.AF_INET6 and "%" in bind_addr[0]: 

354 addr, scope_id = bind_addr[0].split("%", 1) 

355 bind_addr = (addr, bind_addr[1], 0, int(scope_id)) 

356 

357 sock.bind(bind_addr) 

358 sock.listen(backlog) 

359 except BaseException: 

360 sock.close() 

361 raise 

362 

363 return sock 

364 

365 # We passing type=0 on non-Windows platforms as a workaround for a uvloop bug 

366 # where we don't get the correct scope ID for IPv6 link-local addresses when passing 

367 # type=socket.SOCK_STREAM to getaddrinfo(): 

368 # https://github.com/MagicStack/uvloop/issues/539 

369 gai_res = await getaddrinfo( 

370 local_host, 

371 local_port, 

372 family=family, 

373 type=socket.SOCK_STREAM if sys.platform == "win32" else 0, 

374 flags=socket.AI_PASSIVE | socket.AI_ADDRCONFIG, 

375 ) 

376 

377 # The set comprehension is here to work around a glibc bug: 

378 # https://sourceware.org/bugzilla/show_bug.cgi?id=14969 

379 sockaddrs = sorted({res for res in gai_res if res[1] == SocketKind.SOCK_STREAM}) 

380 

381 # Special case for dual-stack binding on the "any" interface 

382 if ( 

383 local_host is None 

384 and family == AddressFamily.AF_UNSPEC 

385 and socket.has_dualstack_ipv6() 

386 and any(fam == AddressFamily.AF_INET6 for fam, *_ in gai_res) 

387 ): 

388 raw_socket = setup_raw_socket( 

389 AddressFamily.AF_INET6, ("::", local_port), v6only=False 

390 ) 

391 listener = asynclib.create_tcp_listener(raw_socket) 

392 return MultiListener([listener]) 

393 

394 errors: list[OSError] = [] 

395 try: 

396 for _ in range(len(sockaddrs)): 

397 listeners: list[SocketListener] = [] 

398 bound_ephemeral_port = local_port 

399 try: 

400 for fam, *_, sockaddr in sockaddrs: 

401 sockaddr = sockaddr[0], bound_ephemeral_port, *sockaddr[2:] 

402 raw_socket = setup_raw_socket(fam, sockaddr) 

403 

404 # Store the assigned port if an ephemeral port was requested, so 

405 # we'll bind to the same port on all interfaces 

406 if local_port == 0 and len(gai_res) > 1: 

407 bound_ephemeral_port = raw_socket.getsockname()[1] 

408 

409 listeners.append(asynclib.create_tcp_listener(raw_socket)) 

410 except BaseException as exc: 

411 for listener in listeners: 

412 await listener.aclose() 

413 

414 # If an ephemeral port was requested but binding the assigned port 

415 # failed for another interface, rotate the address list and try again 

416 if ( 

417 isinstance(exc, OSError) 

418 and exc.errno == errno.EADDRINUSE 

419 and local_port == 0 

420 and bound_ephemeral_port 

421 ): 

422 errors.append(exc) 

423 sockaddrs.append(sockaddrs.pop(0)) 

424 continue 

425 

426 raise 

427 

428 return MultiListener(listeners) 

429 

430 raise OSError( 

431 f"Could not create {len(sockaddrs)} listeners with a consistent port" 

432 ) from ExceptionGroup("Several bind attempts failed", errors) 

433 finally: 

434 del errors # Prevent reference cycles 

435 

436 

437async def create_unix_listener( 

438 path: str | bytes | PathLike[Any], 

439 *, 

440 mode: int | None = None, 

441 backlog: int = 65536, 

442) -> SocketListener: 

443 """ 

444 Create a UNIX socket listener. 

445 

446 Not available on Windows. 

447 

448 :param path: path of the socket 

449 :param mode: permissions to set on the socket 

450 :param backlog: maximum number of queued incoming connections (up to a maximum of 

451 2**16, or 65536) 

452 :return: a listener object 

453 

454 .. versionchanged:: 3.0 

455 If a socket already exists on the file system in the given path, it will be 

456 removed first. 

457 

458 """ 

459 backlog = min(backlog, 65536) 

460 raw_socket = await setup_unix_local_socket(path, mode, socket.SOCK_STREAM) 

461 try: 

462 raw_socket.listen(backlog) 

463 return get_async_backend().create_unix_listener(raw_socket) 

464 except BaseException: 

465 raw_socket.close() 

466 raise 

467 

468 

469async def create_udp_socket( 

470 family: AnyIPAddressFamily = AddressFamily.AF_UNSPEC, 

471 *, 

472 local_host: IPAddressType | None = None, 

473 local_port: int = 0, 

474 reuse_port: bool = False, 

475) -> UDPSocket: 

476 """ 

477 Create a UDP socket. 

478 

479 If ``port`` has been given, the socket will be bound to this port on the local 

480 machine, making this socket suitable for providing UDP based services. 

481 

482 :param family: address family (``AF_INET`` or ``AF_INET6``) – automatically 

483 determined from ``local_host`` if omitted 

484 :param local_host: IP address or host name of the local interface to bind to 

485 :param local_port: local port to bind to 

486 :param reuse_port: ``True`` to allow multiple sockets to bind to the same 

487 address/port (not supported on Windows) 

488 :return: a UDP socket 

489 

490 """ 

491 if family is AddressFamily.AF_UNSPEC and not local_host: 

492 raise ValueError('Either "family" or "local_host" must be given') 

493 

494 if local_host: 

495 gai_res = await getaddrinfo( 

496 str(local_host), 

497 local_port, 

498 family=family, 

499 type=socket.SOCK_DGRAM, 

500 flags=socket.AI_PASSIVE | socket.AI_ADDRCONFIG, 

501 ) 

502 family = cast(AnyIPAddressFamily, gai_res[0][0]) 

503 local_address = gai_res[0][-1] 

504 elif family is AddressFamily.AF_INET6: 

505 local_address = ("::", 0) 

506 else: 

507 local_address = ("0.0.0.0", 0) 

508 

509 sock = await get_async_backend().create_udp_socket( 

510 family, local_address, None, reuse_port 

511 ) 

512 return cast(UDPSocket, sock) 

513 

514 

515async def create_connected_udp_socket( 

516 remote_host: IPAddressType, 

517 remote_port: int, 

518 *, 

519 family: AnyIPAddressFamily = AddressFamily.AF_UNSPEC, 

520 local_host: IPAddressType | None = None, 

521 local_port: int = 0, 

522 reuse_port: bool = False, 

523) -> ConnectedUDPSocket: 

524 """ 

525 Create a connected UDP socket. 

526 

527 Connected UDP sockets can only communicate with the specified remote host/port, an 

528 any packets sent from other sources are dropped. 

529 

530 :param remote_host: remote host to set as the default target 

531 :param remote_port: port on the remote host to set as the default target 

532 :param family: address family (``AF_INET`` or ``AF_INET6``) – automatically 

533 determined from ``local_host`` or ``remote_host`` if omitted 

534 :param local_host: IP address or host name of the local interface to bind to 

535 :param local_port: local port to bind to 

536 :param reuse_port: ``True`` to allow multiple sockets to bind to the same 

537 address/port (not supported on Windows) 

538 :return: a connected UDP socket 

539 

540 """ 

541 local_address = None 

542 if local_host: 

543 gai_res = await getaddrinfo( 

544 str(local_host), 

545 local_port, 

546 family=family, 

547 type=socket.SOCK_DGRAM, 

548 flags=socket.AI_PASSIVE | socket.AI_ADDRCONFIG, 

549 ) 

550 family = cast(AnyIPAddressFamily, gai_res[0][0]) 

551 local_address = gai_res[0][-1] 

552 

553 gai_res = await getaddrinfo( 

554 str(remote_host), remote_port, family=family, type=socket.SOCK_DGRAM 

555 ) 

556 family = cast(AnyIPAddressFamily, gai_res[0][0]) 

557 remote_address = gai_res[0][-1] 

558 

559 sock = await get_async_backend().create_udp_socket( 

560 family, local_address, remote_address, reuse_port 

561 ) 

562 return cast(ConnectedUDPSocket, sock) 

563 

564 

565async def create_unix_datagram_socket( 

566 *, 

567 local_path: None | str | bytes | PathLike[Any] = None, 

568 local_mode: int | None = None, 

569) -> UNIXDatagramSocket: 

570 """ 

571 Create a UNIX datagram socket. 

572 

573 Not available on Windows. 

574 

575 If ``local_path`` has been given, the socket will be bound to this path, making this 

576 socket suitable for receiving datagrams from other processes. Other processes can 

577 send datagrams to this socket only if ``local_path`` is set. 

578 

579 If a socket already exists on the file system in the ``local_path``, it will be 

580 removed first. 

581 

582 :param local_path: the path on which to bind to 

583 :param local_mode: permissions to set on the local socket 

584 :return: a UNIX datagram socket 

585 

586 """ 

587 raw_socket = await setup_unix_local_socket( 

588 local_path, local_mode, socket.SOCK_DGRAM 

589 ) 

590 return await get_async_backend().create_unix_datagram_socket(raw_socket, None) 

591 

592 

593async def create_connected_unix_datagram_socket( 

594 remote_path: str | bytes | PathLike[Any], 

595 *, 

596 local_path: None | str | bytes | PathLike[Any] = None, 

597 local_mode: int | None = None, 

598) -> ConnectedUNIXDatagramSocket: 

599 """ 

600 Create a connected UNIX datagram socket. 

601 

602 Connected datagram sockets can only communicate with the specified remote path. 

603 

604 If ``local_path`` has been given, the socket will be bound to this path, making 

605 this socket suitable for receiving datagrams from other processes. Other processes 

606 can send datagrams to this socket only if ``local_path`` is set. 

607 

608 If a socket already exists on the file system in the ``local_path``, it will be 

609 removed first. 

610 

611 :param remote_path: the path to set as the default target 

612 :param local_path: the path on which to bind to 

613 :param local_mode: permissions to set on the local socket 

614 :return: a connected UNIX datagram socket 

615 

616 """ 

617 remote_path = os.fspath(remote_path) 

618 raw_socket = await setup_unix_local_socket( 

619 local_path, local_mode, socket.SOCK_DGRAM 

620 ) 

621 return await get_async_backend().create_unix_datagram_socket( 

622 raw_socket, remote_path 

623 ) 

624 

625 

626async def getaddrinfo( 

627 host: bytes | str | None, 

628 port: str | int | None, 

629 *, 

630 family: int | AddressFamily = 0, 

631 type: int | SocketKind = 0, 

632 proto: int = 0, 

633 flags: int = 0, 

634) -> list[tuple[AddressFamily, SocketKind, int, str, tuple[str, int]]]: 

635 """ 

636 Look up a numeric IP address given a host name. 

637 

638 Internationalized domain names are translated according to the (non-transitional) 

639 IDNA 2008 standard. 

640 

641 .. note:: 4-tuple IPv6 socket addresses are automatically converted to 2-tuples of 

642 (host, port), unlike what :func:`socket.getaddrinfo` does. 

643 

644 :param host: host name 

645 :param port: port number 

646 :param family: socket family (`'AF_INET``, ...) 

647 :param type: socket type (``SOCK_STREAM``, ...) 

648 :param proto: protocol number 

649 :param flags: flags to pass to upstream ``getaddrinfo()`` 

650 :return: list of tuples containing (family, type, proto, canonname, sockaddr) 

651 

652 .. seealso:: :func:`socket.getaddrinfo` 

653 

654 """ 

655 # Handle unicode hostnames 

656 encoded_host = idna2008_resolve(host) if isinstance(host, str) else host 

657 gai_res = await get_async_backend().getaddrinfo( 

658 encoded_host, port, family=family, type=type, proto=proto, flags=flags 

659 ) 

660 return [ 

661 (family, type, proto, canonname, convert_ipv6_sockaddr(sockaddr)) 

662 for family, type, proto, canonname, sockaddr in gai_res 

663 # filter out IPv6 results when IPv6 is disabled 

664 if not isinstance(sockaddr[0], int) 

665 ] 

666 

667 

668def getnameinfo(sockaddr: IPSockAddrType, flags: int = 0) -> Awaitable[tuple[str, str]]: 

669 """ 

670 Look up the host name of an IP address. 

671 

672 :param sockaddr: socket address (e.g. (ipaddress, port) for IPv4) 

673 :param flags: flags to pass to upstream ``getnameinfo()`` 

674 :return: a tuple of (host name, service name) 

675 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

676 current thread 

677 

678 .. seealso:: :func:`socket.getnameinfo` 

679 

680 """ 

681 return get_async_backend().getnameinfo(sockaddr, flags) 

682 

683 

684@deprecated("This function is deprecated; use `wait_readable` instead") 

685def wait_socket_readable(sock: socket.socket) -> Awaitable[None]: 

686 """ 

687 .. deprecated:: 4.7.0 

688 Use :func:`wait_readable` instead. 

689 

690 Wait until the given socket has data to be read. 

691 

692 .. warning:: Only use this on raw sockets that have not been wrapped by any higher 

693 level constructs like socket streams! 

694 

695 :param sock: a socket object 

696 :raises ~anyio.ClosedResourceError: if the socket was closed while waiting for the 

697 socket to become readable 

698 :raises ~anyio.BusyResourceError: if another task is already waiting for the socket 

699 to become readable 

700 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

701 current thread 

702 

703 """ 

704 return get_async_backend().wait_readable(sock.fileno()) 

705 

706 

707@deprecated("This function is deprecated; use `wait_writable` instead") 

708def wait_socket_writable(sock: socket.socket) -> Awaitable[None]: 

709 """ 

710 .. deprecated:: 4.7.0 

711 Use :func:`wait_writable` instead. 

712 

713 Wait until the given socket can be written to. 

714 

715 This does **NOT** work on Windows when using the asyncio backend with a proactor 

716 event loop (default on py3.8+). 

717 

718 .. warning:: Only use this on raw sockets that have not been wrapped by any higher 

719 level constructs like socket streams! 

720 

721 :param sock: a socket object 

722 :raises ~anyio.ClosedResourceError: if the socket was closed while waiting for the 

723 socket to become writable 

724 :raises ~anyio.BusyResourceError: if another task is already waiting for the socket 

725 to become writable 

726 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

727 current thread 

728 

729 """ 

730 return get_async_backend().wait_writable(sock.fileno()) 

731 

732 

733def wait_readable(obj: FileDescriptorLike) -> Awaitable[None]: 

734 """ 

735 Wait until the given object has data to be read. 

736 

737 On Unix systems, ``obj`` must either be an integer file descriptor, or else an 

738 object with a ``.fileno()`` method which returns an integer file descriptor. Any 

739 kind of file descriptor can be passed, though the exact semantics will depend on 

740 your kernel. For example, this probably won't do anything useful for on-disk files. 

741 

742 On Windows systems, ``obj`` must either be an integer ``SOCKET`` handle, or else an 

743 object with a ``.fileno()`` method which returns an integer ``SOCKET`` handle. File 

744 descriptors aren't supported, and neither are handles that refer to anything besides 

745 a ``SOCKET``. 

746 

747 On backends where this functionality is not natively provided (asyncio 

748 ``ProactorEventLoop`` on Windows), it is provided using a separate selector thread 

749 which is set to shut down when the interpreter shuts down. 

750 

751 .. warning:: Don't use this on raw sockets that have been wrapped by any higher 

752 level constructs like socket streams! 

753 

754 :param obj: an object with a ``.fileno()`` method or an integer handle 

755 :raises ~anyio.ClosedResourceError: if the object was closed while waiting for the 

756 object to become readable 

757 :raises ~anyio.BusyResourceError: if another task is already waiting for the object 

758 to become readable 

759 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

760 current thread 

761 

762 """ 

763 return get_async_backend().wait_readable(obj) 

764 

765 

766def wait_writable(obj: FileDescriptorLike) -> Awaitable[None]: 

767 """ 

768 Wait until the given object can be written to. 

769 

770 :param obj: an object with a ``.fileno()`` method or an integer handle 

771 :raises ~anyio.ClosedResourceError: if the object was closed while waiting for the 

772 object to become writable 

773 :raises ~anyio.BusyResourceError: if another task is already waiting for the object 

774 to become writable 

775 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

776 current thread 

777 

778 .. seealso:: See the documentation of :func:`wait_readable` for the definition of 

779 ``obj`` and notes on backend compatibility. 

780 

781 .. warning:: Don't use this on raw sockets that have been wrapped by any higher 

782 level constructs like socket streams! 

783 

784 """ 

785 return get_async_backend().wait_writable(obj) 

786 

787 

788def notify_closing(obj: FileDescriptorLike) -> None: 

789 """ 

790 Call this before closing a file descriptor (on Unix) or socket (on 

791 Windows). This will cause any `wait_readable` or `wait_writable` 

792 calls on the given object to immediately wake up and raise 

793 `~anyio.ClosedResourceError`. 

794 

795 This doesn't actually close the object – you still have to do that 

796 yourself afterwards. Also, you want to be careful to make sure no 

797 new tasks start waiting on the object in between when you call this 

798 and when it's actually closed. So to close something properly, you 

799 usually want to do these steps in order: 

800 

801 1. Explicitly mark the object as closed, so that any new attempts 

802 to use it will abort before they start. 

803 2. Call `notify_closing` to wake up any already-existing users. 

804 3. Actually close the object. 

805 

806 It's also possible to do them in a different order if that's more 

807 convenient, *but only if* you make sure not to have any checkpoints in 

808 between the steps. This way they all happen in a single atomic 

809 step, so other tasks won't be able to tell what order they happened 

810 in anyway. 

811 

812 :param obj: an object with a ``.fileno()`` method or an integer handle 

813 :raises NoEventLoopError: if no supported asynchronous event loop is running in the 

814 current thread 

815 

816 """ 

817 get_async_backend().notify_closing(obj) 

818 

819 

820# 

821# Private API 

822# 

823 

824 

825def convert_ipv6_sockaddr( 

826 sockaddr: tuple[str, int, int, int] | tuple[str, int], 

827) -> tuple[str, int]: 

828 """ 

829 Convert a 4-tuple IPv6 socket address to a 2-tuple (address, port) format. 

830 

831 If the scope ID is nonzero, it is added to the address, separated with ``%``. 

832 Otherwise the flow id and scope id are simply cut off from the tuple. 

833 Any other kinds of socket addresses are returned as-is. 

834 

835 :param sockaddr: the result of :meth:`~socket.socket.getsockname` 

836 :return: the converted socket address 

837 

838 """ 

839 # This is more complicated than it should be because of MyPy 

840 if isinstance(sockaddr, tuple) and len(sockaddr) == 4: 

841 host, port, flowinfo, scope_id = sockaddr 

842 if scope_id: 

843 # PyPy (as of v7.3.11) leaves the interface name in the result, so 

844 # we discard it and only get the scope ID from the end 

845 # (https://foss.heptapod.net/pypy/pypy/-/issues/3938) 

846 host = host.split("%")[0] 

847 

848 # Add scope_id to the address 

849 return f"{host}%{scope_id}", port 

850 else: 

851 return host, port 

852 else: 

853 return sockaddr 

854 

855 

856async def setup_unix_local_socket( 

857 path: None | str | bytes | PathLike[Any], 

858 mode: int | None, 

859 socktype: int, 

860) -> socket.socket: 

861 """ 

862 Create a UNIX local socket object, deleting the socket at the given path if it 

863 exists. 

864 

865 Not available on Windows. 

866 

867 :param path: path of the socket 

868 :param mode: permissions to set on the socket 

869 :param socktype: socket.SOCK_STREAM or socket.SOCK_DGRAM 

870 

871 """ 

872 path_str: str | None 

873 if path is not None: 

874 path_str = os.fsdecode(path) 

875 

876 # Linux abstract namespace sockets aren't backed by a concrete file so skip stat call 

877 if not path_str.startswith("\0"): 

878 # Copied from pathlib... 

879 try: 

880 stat_result = os.stat(path) 

881 except OSError as e: 

882 if e.errno not in ( 

883 errno.ENOENT, 

884 errno.ENOTDIR, 

885 errno.EBADF, 

886 errno.ELOOP, 

887 ): 

888 raise 

889 else: 

890 if stat.S_ISSOCK(stat_result.st_mode): 

891 os.unlink(path) 

892 else: 

893 path_str = None 

894 

895 raw_socket = socket.socket(socket.AF_UNIX, socktype) 

896 raw_socket.setblocking(False) 

897 

898 if path_str is not None: 

899 try: 

900 await to_thread.run_sync(raw_socket.bind, path_str, abandon_on_cancel=True) 

901 if mode is not None: 

902 await to_thread.run_sync(chmod, path_str, mode, abandon_on_cancel=True) 

903 except BaseException: 

904 raw_socket.close() 

905 raise 

906 

907 return raw_socket 

908 

909 

910@dataclass 

911class TCPConnectable(ByteStreamConnectable): 

912 """ 

913 Connects to a TCP server at the given host and port. 

914 

915 :param host: host name or IP address of the server 

916 :param port: TCP port number of the server 

917 """ 

918 

919 host: str | IPv4Address | IPv6Address 

920 port: int 

921 

922 def __post_init__(self) -> None: 

923 if self.port < 1 or self.port > 65535: 

924 raise ValueError("TCP port number out of range") 

925 

926 @override 

927 async def connect(self) -> SocketStream: 

928 try: 

929 return await connect_tcp(self.host, self.port) 

930 except OSError as exc: 

931 raise ConnectionFailed( 

932 f"error connecting to {self.host}:{self.port}: {exc}" 

933 ) from exc 

934 

935 

936@dataclass 

937class UNIXConnectable(ByteStreamConnectable): 

938 """ 

939 Connects to a UNIX domain socket at the given path. 

940 

941 :param path: the file system path of the socket 

942 """ 

943 

944 path: str | bytes | PathLike[str] | PathLike[bytes]