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

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 except BaseException: 

267 if connected_stream is not None: 

268 await aclose_forcefully(connected_stream) 

269 

270 raise 

271 finally: 

272 oserrors.clear() 

273 

274 if tls or tls_hostname or ssl_context: 

275 try: 

276 return await TLSStream.wrap( 

277 connected_stream, 

278 server_side=False, 

279 hostname=tls_hostname or str(remote_host), 

280 ssl_context=ssl_context, 

281 standard_compatible=tls_standard_compatible, 

282 ) 

283 except BaseException: 

284 await aclose_forcefully(connected_stream) 

285 raise 

286 

287 return connected_stream 

288 

289 

290async def connect_unix( 

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

292 kind: Literal[ 

293 SocketKind.SOCK_STREAM, SocketKind.SOCK_SEQPACKET 

294 ] = socket.SOCK_STREAM, 

295) -> UNIXSocketStream: 

296 """ 

297 Connect to the given UNIX socket. 

298 

299 Not available on Windows. 

300 

301 :param path: path to the socket 

302 :param kind: socket type (``SOCK_STREAM`` or ``SOCK_SEQPACKET``) 

303 :return: a socket stream object 

304 :raises ConnectionFailed: if the connection fails 

305 

306 """ 

307 path = os.fspath(path) 

308 return await get_async_backend().connect_unix(path, kind) 

309 

310 

311async def create_tcp_listener( 

312 *, 

313 local_host: IPAddressType | None = None, 

314 local_port: int = 0, 

315 family: AnyIPAddressFamily = socket.AddressFamily.AF_UNSPEC, 

316 backlog: int = 65536, 

317 reuse_port: bool = False, 

318) -> MultiListener[SocketStream]: 

319 """ 

320 Create a TCP socket listener. 

321 

322 :param local_port: port number to listen on 

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

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

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

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

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

328 2**16, or 65536) 

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

330 address/port (not supported on Windows) 

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

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

333 interfaces failed 

334 

335 """ 

336 asynclib = get_async_backend() 

337 backlog = min(backlog, 65536) 

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

339 

340 def setup_raw_socket( 

341 fam: AddressFamily, 

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

343 *, 

344 v6only: bool = True, 

345 ) -> socket.socket: 

346 sock = socket.socket(fam) 

347 try: 

348 sock.setblocking(False) 

349 

350 if fam == AddressFamily.AF_INET6: 

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

352 

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

354 # reuse. 

355 if sys.platform == "win32": 

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

357 else: 

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

359 

360 if reuse_port: 

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

362 

363 # Workaround for #554 

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

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

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

367 

368 sock.bind(bind_addr) 

369 sock.listen(backlog) 

370 except BaseException: 

371 sock.close() 

372 raise 

373 

374 return sock 

375 

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

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

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

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

380 gai_res = await getaddrinfo( 

381 local_host, 

382 local_port, 

383 family=family, 

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

385 flags=socket.AI_PASSIVE | socket.AI_ADDRCONFIG, 

386 ) 

387 

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

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

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

391 

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

393 if ( 

394 local_host is None 

395 and family == AddressFamily.AF_UNSPEC 

396 and socket.has_dualstack_ipv6() 

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

398 ): 

399 raw_socket = setup_raw_socket( 

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

401 ) 

402 listener = asynclib.create_tcp_listener(raw_socket) 

403 return MultiListener([listener]) 

404 

405 errors: list[OSError] = [] 

406 try: 

407 for _ in range(10): # enough to absorb ephemeral port collisions 

408 listeners: list[SocketListener] = [] 

409 bound_ephemeral_port = local_port 

410 try: 

411 for fam, *_, sockaddr in sockaddrs: 

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

413 raw_socket = setup_raw_socket(fam, sockaddr) 

414 

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

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

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

418 bound_ephemeral_port = raw_socket.getsockname()[1] 

419 

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

421 except BaseException as exc: 

422 for listener in listeners: 

423 await listener.aclose() 

424 

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

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

427 if ( 

428 isinstance(exc, OSError) 

429 and exc.errno == errno.EADDRINUSE 

430 and local_port == 0 

431 and bound_ephemeral_port 

432 ): 

433 errors.append(exc) 

434 sockaddrs.append(sockaddrs.pop(0)) 

435 continue 

436 

437 raise 

438 

439 return MultiListener(listeners) 

440 

441 raise OSError( 

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

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

444 finally: 

445 del errors # Prevent reference cycles 

446 

447 

448async def create_unix_listener( 

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

450 *, 

451 mode: int | None = None, 

452 backlog: int = 65536, 

453 kind: Literal[ 

454 SocketKind.SOCK_STREAM, SocketKind.SOCK_SEQPACKET 

455 ] = SocketKind.SOCK_STREAM, 

456) -> SocketListener: 

457 """ 

458 Create a UNIX socket listener. 

459 

460 Not available on Windows. 

461 

462 :param path: path of the socket 

463 :param mode: permissions to set on the socket 

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

465 2**16, or 65536) 

466 :param kind: socket type (``SOCK_STREAM`` or ``SOCK_SEQPACKET``) 

467 :return: a listener object 

468 

469 .. versionchanged:: 3.0 

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

471 removed first. 

472 

473 """ 

474 backlog = min(backlog, 65536) 

475 raw_socket = await setup_unix_local_socket(path, mode, kind) 

476 try: 

477 raw_socket.listen(backlog) 

478 return get_async_backend().create_unix_listener(raw_socket) 

479 except BaseException: 

480 raw_socket.close() 

481 raise 

482 

483 

484async def create_udp_socket( 

485 family: AnyIPAddressFamily = AddressFamily.AF_UNSPEC, 

486 *, 

487 local_host: IPAddressType | None = None, 

488 local_port: int = 0, 

489 reuse_port: bool = False, 

490) -> UDPSocket: 

491 """ 

492 Create a UDP socket. 

493 

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

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

496 

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

498 determined from ``local_host`` if omitted 

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

500 :param local_port: local port to bind to 

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

502 address/port (not supported on Windows) 

503 :return: a UDP socket 

504 

505 """ 

506 if family is AddressFamily.AF_UNSPEC and not local_host: 

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

508 

509 if local_host: 

510 gai_res = await getaddrinfo( 

511 str(local_host), 

512 local_port, 

513 family=family, 

514 type=socket.SOCK_DGRAM, 

515 flags=socket.AI_PASSIVE | socket.AI_ADDRCONFIG, 

516 ) 

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

518 local_address = gai_res[0][-1] 

519 elif family is AddressFamily.AF_INET6: 

520 local_address = ("::", local_port) 

521 else: 

522 local_address = ("0.0.0.0", local_port) 

523 

524 sock = await get_async_backend().create_udp_socket( 

525 family, local_address, None, reuse_port 

526 ) 

527 return cast(UDPSocket, sock) 

528 

529 

530async def create_connected_udp_socket( 

531 remote_host: IPAddressType, 

532 remote_port: int, 

533 *, 

534 family: AnyIPAddressFamily = AddressFamily.AF_UNSPEC, 

535 local_host: IPAddressType | None = None, 

536 local_port: int = 0, 

537 reuse_port: bool = False, 

538) -> ConnectedUDPSocket: 

539 """ 

540 Create a connected UDP socket. 

541 

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

543 any packets sent from other sources are dropped. 

544 

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

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

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

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

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

550 :param local_port: local port to bind to 

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

552 address/port (not supported on Windows) 

553 :return: a connected UDP socket 

554 

555 """ 

556 local_address = None 

557 if local_host: 

558 gai_res = await getaddrinfo( 

559 str(local_host), 

560 local_port, 

561 family=family, 

562 type=socket.SOCK_DGRAM, 

563 flags=socket.AI_PASSIVE | socket.AI_ADDRCONFIG, 

564 ) 

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

566 local_address = gai_res[0][-1] 

567 

568 gai_res = await getaddrinfo( 

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

570 ) 

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

572 remote_address = gai_res[0][-1] 

573 

574 if local_address is None and local_port: 

575 if family is AddressFamily.AF_INET6: 

576 local_address = ("::", local_port) 

577 else: 

578 local_address = ("0.0.0.0", local_port) 

579 

580 sock = await get_async_backend().create_udp_socket( 

581 family, local_address, remote_address, reuse_port 

582 ) 

583 return cast(ConnectedUDPSocket, sock) 

584 

585 

586async def create_unix_datagram_socket( 

587 *, 

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

589 local_mode: int | None = None, 

590) -> UNIXDatagramSocket: 

591 """ 

592 Create a UNIX datagram socket. 

593 

594 Not available on Windows. 

595 

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

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

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

599 

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

601 removed first. 

602 

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

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

605 :return: a UNIX datagram socket 

606 

607 """ 

608 raw_socket = await setup_unix_local_socket( 

609 local_path, local_mode, socket.SOCK_DGRAM 

610 ) 

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

612 

613 

614async def create_connected_unix_datagram_socket( 

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

616 *, 

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

618 local_mode: int | None = None, 

619) -> ConnectedUNIXDatagramSocket: 

620 """ 

621 Create a connected UNIX datagram socket. 

622 

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

624 

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

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

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

628 

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

630 removed first. 

631 

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

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

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

635 :return: a connected UNIX datagram socket 

636 

637 """ 

638 remote_path = os.fspath(remote_path) 

639 raw_socket = await setup_unix_local_socket( 

640 local_path, local_mode, socket.SOCK_DGRAM 

641 ) 

642 return await get_async_backend().create_unix_datagram_socket( 

643 raw_socket, remote_path 

644 ) 

645 

646 

647async def getaddrinfo( 

648 host: bytes | str | None, 

649 port: str | int | None, 

650 *, 

651 family: int | AddressFamily = 0, 

652 type: int | SocketKind = 0, 

653 proto: int = 0, 

654 flags: int = 0, 

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

656 """ 

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

658 

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

660 IDNA 2008 standard. 

661 

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

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

664 

665 :param host: host name 

666 :param port: port number 

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

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

669 :param proto: protocol number 

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

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

672 

673 .. seealso:: :func:`socket.getaddrinfo` 

674 

675 """ 

676 # Handle unicode hostnames 

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

678 gai_res = await get_async_backend().getaddrinfo( 

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

680 ) 

681 return [ 

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

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

684 # filter out IPv6 results when IPv6 is disabled 

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

686 ] 

687 

688 

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

690 """ 

691 Look up the host name of an IP address. 

692 

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

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

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

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

697 current thread 

698 

699 .. seealso:: :func:`socket.getnameinfo` 

700 

701 """ 

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

703 

704 

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

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

707 """ 

708 .. deprecated:: 4.7.0 

709 Use :func:`wait_readable` instead. 

710 

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

712 

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

714 level constructs like socket streams! 

715 

716 :param sock: a socket object 

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

718 socket to become readable 

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

720 to become readable 

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

722 current thread 

723 

724 """ 

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

726 

727 

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

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

730 """ 

731 .. deprecated:: 4.7.0 

732 Use :func:`wait_writable` instead. 

733 

734 Wait until the given socket can be written to. 

735 

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

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

738 

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

740 level constructs like socket streams! 

741 

742 :param sock: a socket object 

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

744 socket to become writable 

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

746 to become writable 

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

748 current thread 

749 

750 """ 

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

752 

753 

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

755 """ 

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

757 

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

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

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

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

762 

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

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

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

766 a ``SOCKET``. 

767 

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

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

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

771 

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

773 level constructs like socket streams! 

774 

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

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

777 object to become readable 

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

779 to become readable 

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

781 current thread 

782 

783 """ 

784 return get_async_backend().wait_readable(obj) 

785 

786 

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

788 """ 

789 Wait until the given object can be written to. 

790 

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

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

793 object to become writable 

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

795 to become writable 

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

797 current thread 

798 

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

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

801 

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

803 level constructs like socket streams! 

804 

805 """ 

806 return get_async_backend().wait_writable(obj) 

807 

808 

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

810 """ 

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

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

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

814 `~anyio.ClosedResourceError`. 

815 

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

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

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

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

820 usually want to do these steps in order: 

821 

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

823 to use it will abort before they start. 

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

825 3. Actually close the object. 

826 

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

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

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

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

831 in anyway. 

832 

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

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

835 current thread 

836 

837 """ 

838 get_async_backend().notify_closing(obj) 

839 

840 

841# 

842# Private API 

843# 

844 

845 

846def convert_ipv6_sockaddr( 

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

848) -> tuple[str, int]: 

849 """ 

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

851 

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

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

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

855 

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

857 :return: the converted socket address 

858 

859 """ 

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

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

862 host, port, _flowinfo, scope_id = sockaddr 

863 if scope_id: 

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

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

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

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

868 

869 # Add scope_id to the address 

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

871 else: 

872 return host, port 

873 else: 

874 return sockaddr 

875 

876 

877async def setup_unix_local_socket( 

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

879 mode: int | None, 

880 socktype: int, 

881) -> socket.socket: 

882 """ 

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

884 exists. 

885 

886 Not available on Windows. 

887 

888 :param path: path of the socket 

889 :param mode: permissions to set on the socket 

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

891 

892 """ 

893 path_str: str | None 

894 if path is not None: 

895 path_str = os.fsdecode(path) 

896 

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

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

899 # Copied from pathlib... 

900 try: 

901 stat_result = os.stat(path) 

902 except OSError as e: 

903 if e.errno not in ( 

904 errno.ENOENT, 

905 errno.ENOTDIR, 

906 errno.EBADF, 

907 errno.ELOOP, 

908 ): 

909 raise 

910 else: 

911 if stat.S_ISSOCK(stat_result.st_mode): 

912 os.unlink(path) 

913 else: 

914 path_str = None 

915 

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

917 raw_socket.setblocking(False) 

918 

919 if path_str is not None: 

920 try: 

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

922 if mode is not None: 

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

924 except BaseException: 

925 raw_socket.close() 

926 raise 

927 

928 return raw_socket 

929 

930 

931@dataclass 

932class TCPConnectable(ByteStreamConnectable): 

933 """ 

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

935 

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

937 :param port: TCP port number of the server 

938 """ 

939 

940 host: str | IPv4Address | IPv6Address 

941 port: int 

942 

943 def __post_init__(self) -> None: 

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