_sockets.py 34 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011
  1. from __future__ import annotations
  2. import errno
  3. import os
  4. import socket
  5. import ssl
  6. import stat
  7. import sys
  8. from collections.abc import Awaitable
  9. from dataclasses import dataclass
  10. from ipaddress import IPv4Address, IPv6Address, ip_address
  11. from os import PathLike, chmod
  12. from socket import AddressFamily, SocketKind
  13. from typing import TYPE_CHECKING, Any, Literal, cast, overload
  14. from .. import ConnectionFailed, to_thread
  15. from ..abc import (
  16. ByteStreamConnectable,
  17. ConnectedUDPSocket,
  18. ConnectedUNIXDatagramSocket,
  19. IPAddressType,
  20. IPSockAddrType,
  21. SocketListener,
  22. SocketStream,
  23. UDPSocket,
  24. UNIXDatagramSocket,
  25. UNIXSocketStream,
  26. )
  27. from ..streams.stapled import MultiListener
  28. from ..streams.tls import TLSConnectable, TLSStream
  29. from ._eventloop import get_async_backend
  30. from ._resources import aclose_forcefully
  31. from ._synchronization import Event
  32. from ._tasks import create_task_group, move_on_after
  33. if TYPE_CHECKING:
  34. from _typeshed import FileDescriptorLike
  35. else:
  36. FileDescriptorLike = object
  37. if sys.version_info < (3, 11):
  38. from exceptiongroup import ExceptionGroup
  39. if sys.version_info >= (3, 12):
  40. from typing import override
  41. else:
  42. from typing_extensions import override
  43. if sys.version_info < (3, 13):
  44. from typing_extensions import deprecated
  45. else:
  46. from warnings import deprecated
  47. IPPROTO_IPV6 = getattr(socket, "IPPROTO_IPV6", 41) # https://bugs.python.org/issue29515
  48. AnyIPAddressFamily = Literal[
  49. AddressFamily.AF_UNSPEC, AddressFamily.AF_INET, AddressFamily.AF_INET6
  50. ]
  51. IPAddressFamily = Literal[AddressFamily.AF_INET, AddressFamily.AF_INET6]
  52. def idna2008_resolve(host: str) -> bytes:
  53. try:
  54. return host.encode("ascii")
  55. except UnicodeEncodeError:
  56. import idna
  57. return idna.encode(host, uts46=True)
  58. # tls_hostname given
  59. @overload
  60. async def connect_tcp(
  61. remote_host: IPAddressType,
  62. remote_port: int,
  63. *,
  64. local_host: IPAddressType | None = ...,
  65. local_port: int | None = ...,
  66. ssl_context: ssl.SSLContext | None = ...,
  67. tls_standard_compatible: bool = ...,
  68. tls_hostname: str,
  69. happy_eyeballs_delay: float = ...,
  70. ) -> TLSStream: ...
  71. # ssl_context given
  72. @overload
  73. async 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,
  80. tls_standard_compatible: bool = ...,
  81. tls_hostname: str | None = ...,
  82. happy_eyeballs_delay: float = ...,
  83. ) -> TLSStream: ...
  84. # tls=True
  85. @overload
  86. async def connect_tcp(
  87. remote_host: IPAddressType,
  88. remote_port: int,
  89. *,
  90. local_host: IPAddressType | None = ...,
  91. local_port: int | None = ...,
  92. tls: Literal[True],
  93. ssl_context: ssl.SSLContext | None = ...,
  94. tls_standard_compatible: bool = ...,
  95. tls_hostname: str | None = ...,
  96. happy_eyeballs_delay: float = ...,
  97. ) -> TLSStream: ...
  98. # tls=False
  99. @overload
  100. async def connect_tcp(
  101. remote_host: IPAddressType,
  102. remote_port: int,
  103. *,
  104. local_host: IPAddressType | None = ...,
  105. local_port: int | None = ...,
  106. tls: Literal[False],
  107. ssl_context: ssl.SSLContext | None = ...,
  108. tls_standard_compatible: bool = ...,
  109. tls_hostname: str | None = ...,
  110. happy_eyeballs_delay: float = ...,
  111. ) -> SocketStream: ...
  112. # No TLS arguments
  113. @overload
  114. async def connect_tcp(
  115. remote_host: IPAddressType,
  116. remote_port: int,
  117. *,
  118. local_host: IPAddressType | None = ...,
  119. local_port: int | None = ...,
  120. happy_eyeballs_delay: float = ...,
  121. ) -> SocketStream: ...
  122. async def connect_tcp(
  123. remote_host: IPAddressType,
  124. remote_port: int,
  125. *,
  126. local_host: IPAddressType | None = None,
  127. local_port: int | None = None,
  128. tls: bool = False,
  129. ssl_context: ssl.SSLContext | None = None,
  130. tls_standard_compatible: bool = True,
  131. tls_hostname: str | None = None,
  132. happy_eyeballs_delay: float = 0.25,
  133. ) -> SocketStream | TLSStream:
  134. """
  135. Connect to a host using the TCP protocol.
  136. This function implements the stateless version of the Happy Eyeballs algorithm (RFC
  137. 6555). If ``remote_host`` is a host name that resolves to multiple IP addresses,
  138. each one is tried until one connection attempt succeeds. If the first attempt does
  139. not connected within 250 milliseconds, a second attempt is started using the next
  140. address in the list, and so on. On IPv6 enabled systems, an IPv6 address (if
  141. available) is tried first.
  142. When the connection has been established, a TLS handshake will be done if either
  143. ``ssl_context`` or ``tls_hostname`` is not ``None``, or if ``tls`` is ``True``.
  144. :param remote_host: the IP address or host name to connect to
  145. :param remote_port: port on the target host to connect to
  146. :param local_host: the interface address or name to bind the socket to before
  147. connecting
  148. :param local_port: the local port to bind to (requires ``local_host`` to also be
  149. set)
  150. :param tls: ``True`` to do a TLS handshake with the connected stream and return a
  151. :class:`~anyio.streams.tls.TLSStream` instead
  152. :param ssl_context: the SSL context object to use (if omitted, a default context is
  153. created)
  154. :param tls_standard_compatible: If ``True``, performs the TLS shutdown handshake
  155. before closing the stream and requires that the server does this as well.
  156. Otherwise, :exc:`~ssl.SSLEOFError` may be raised during reads from the stream.
  157. Some protocols, such as HTTP, require this option to be ``False``.
  158. See :meth:`~ssl.SSLContext.wrap_socket` for details.
  159. :param tls_hostname: host name to check the server certificate against (defaults to
  160. the value of ``remote_host``)
  161. :param happy_eyeballs_delay: delay (in seconds) before starting the next connection
  162. attempt
  163. :return: a socket stream object if no TLS handshake was done, otherwise a TLS stream
  164. :raises ConnectionFailed: if the connection fails
  165. """
  166. # Placed here due to https://github.com/python/mypy/issues/7057
  167. connected_stream: SocketStream | None = None
  168. async def try_connect(remote_host: str, event: Event) -> None:
  169. nonlocal connected_stream
  170. try:
  171. stream = await asynclib.connect_tcp(remote_host, remote_port, local_address)
  172. except OSError as exc:
  173. oserrors.append(exc)
  174. return
  175. else:
  176. if connected_stream is None:
  177. connected_stream = stream
  178. tg.cancel_scope.cancel()
  179. else:
  180. await stream.aclose()
  181. finally:
  182. event.set()
  183. asynclib = get_async_backend()
  184. local_address: IPSockAddrType | None = None
  185. family = socket.AF_UNSPEC
  186. if local_host:
  187. gai_res = await getaddrinfo(str(local_host), local_port)
  188. family, *_, local_address = gai_res[0]
  189. target_host = str(remote_host)
  190. try:
  191. addr_obj = ip_address(remote_host)
  192. except ValueError:
  193. addr_obj = None
  194. if addr_obj is not None:
  195. if isinstance(addr_obj, IPv6Address):
  196. target_addrs = [(socket.AF_INET6, addr_obj.compressed)]
  197. else:
  198. target_addrs = [(socket.AF_INET, addr_obj.compressed)]
  199. else:
  200. # getaddrinfo() will raise an exception if name resolution fails
  201. gai_res = await getaddrinfo(
  202. target_host, remote_port, family=family, type=socket.SOCK_STREAM
  203. )
  204. # Organize the list so that the first address is an IPv6 address (if available)
  205. # and the second one is an IPv4 addresses. The rest can be in whatever order.
  206. v6_found = v4_found = False
  207. target_addrs = []
  208. for af, *_, sa in gai_res:
  209. if af == socket.AF_INET6 and not v6_found:
  210. v6_found = True
  211. target_addrs.insert(0, (af, sa[0]))
  212. elif af == socket.AF_INET and not v4_found and v6_found:
  213. v4_found = True
  214. target_addrs.insert(1, (af, sa[0]))
  215. else:
  216. target_addrs.append((af, sa[0]))
  217. oserrors: list[OSError] = []
  218. try:
  219. async with create_task_group() as tg:
  220. for _af, addr in target_addrs:
  221. event = Event()
  222. tg.start_soon(try_connect, addr, event)
  223. with move_on_after(happy_eyeballs_delay):
  224. await event.wait()
  225. if connected_stream is None:
  226. cause = (
  227. oserrors[0]
  228. if len(oserrors) == 1
  229. else ExceptionGroup("multiple connection attempts failed", oserrors)
  230. )
  231. raise OSError("All connection attempts failed") from cause
  232. finally:
  233. oserrors.clear()
  234. if tls or tls_hostname or ssl_context:
  235. try:
  236. return await TLSStream.wrap(
  237. connected_stream,
  238. server_side=False,
  239. hostname=tls_hostname or str(remote_host),
  240. ssl_context=ssl_context,
  241. standard_compatible=tls_standard_compatible,
  242. )
  243. except BaseException:
  244. await aclose_forcefully(connected_stream)
  245. raise
  246. return connected_stream
  247. async def connect_unix(path: str | bytes | PathLike[Any]) -> UNIXSocketStream:
  248. """
  249. Connect to the given UNIX socket.
  250. Not available on Windows.
  251. :param path: path to the socket
  252. :return: a socket stream object
  253. :raises ConnectionFailed: if the connection fails
  254. """
  255. path = os.fspath(path)
  256. return await get_async_backend().connect_unix(path)
  257. async def create_tcp_listener(
  258. *,
  259. local_host: IPAddressType | None = None,
  260. local_port: int = 0,
  261. family: AnyIPAddressFamily = socket.AddressFamily.AF_UNSPEC,
  262. backlog: int = 65536,
  263. reuse_port: bool = False,
  264. ) -> MultiListener[SocketStream]:
  265. """
  266. Create a TCP socket listener.
  267. :param local_port: port number to listen on
  268. :param local_host: IP address of the interface to listen on. If omitted, listen on
  269. all IPv4 and IPv6 interfaces. To listen on all interfaces on a specific address
  270. family, use ``0.0.0.0`` for IPv4 or ``::`` for IPv6.
  271. :param family: address family (used if ``local_host`` was omitted)
  272. :param backlog: maximum number of queued incoming connections (up to a maximum of
  273. 2**16, or 65536)
  274. :param reuse_port: ``True`` to allow multiple sockets to bind to the same
  275. address/port (not supported on Windows)
  276. :return: a multi-listener object containing one or more socket listeners
  277. :raises OSError: if there's an error creating a socket, or binding to one or more
  278. interfaces failed
  279. """
  280. asynclib = get_async_backend()
  281. backlog = min(backlog, 65536)
  282. local_host = str(local_host) if local_host is not None else None
  283. def setup_raw_socket(
  284. fam: AddressFamily,
  285. bind_addr: tuple[str, int] | tuple[str, int, int, int],
  286. *,
  287. v6only: bool = True,
  288. ) -> socket.socket:
  289. sock = socket.socket(fam)
  290. try:
  291. sock.setblocking(False)
  292. if fam == AddressFamily.AF_INET6:
  293. sock.setsockopt(IPPROTO_IPV6, socket.IPV6_V6ONLY, v6only)
  294. # For Windows, enable exclusive address use. For others, enable address
  295. # reuse.
  296. if sys.platform == "win32":
  297. sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
  298. else:
  299. sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  300. if reuse_port:
  301. sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
  302. # Workaround for #554
  303. if fam == socket.AF_INET6 and "%" in bind_addr[0]:
  304. addr, scope_id = bind_addr[0].split("%", 1)
  305. bind_addr = (addr, bind_addr[1], 0, int(scope_id))
  306. sock.bind(bind_addr)
  307. sock.listen(backlog)
  308. except BaseException:
  309. sock.close()
  310. raise
  311. return sock
  312. # We passing type=0 on non-Windows platforms as a workaround for a uvloop bug
  313. # where we don't get the correct scope ID for IPv6 link-local addresses when passing
  314. # type=socket.SOCK_STREAM to getaddrinfo():
  315. # https://github.com/MagicStack/uvloop/issues/539
  316. gai_res = await getaddrinfo(
  317. local_host,
  318. local_port,
  319. family=family,
  320. type=socket.SOCK_STREAM if sys.platform == "win32" else 0,
  321. flags=socket.AI_PASSIVE | socket.AI_ADDRCONFIG,
  322. )
  323. # The set comprehension is here to work around a glibc bug:
  324. # https://sourceware.org/bugzilla/show_bug.cgi?id=14969
  325. sockaddrs = sorted({res for res in gai_res if res[1] == SocketKind.SOCK_STREAM})
  326. # Special case for dual-stack binding on the "any" interface
  327. if (
  328. local_host is None
  329. and family == AddressFamily.AF_UNSPEC
  330. and socket.has_dualstack_ipv6()
  331. and any(fam == AddressFamily.AF_INET6 for fam, *_ in gai_res)
  332. ):
  333. raw_socket = setup_raw_socket(
  334. AddressFamily.AF_INET6, ("::", local_port), v6only=False
  335. )
  336. listener = asynclib.create_tcp_listener(raw_socket)
  337. return MultiListener([listener])
  338. errors: list[OSError] = []
  339. try:
  340. for _ in range(len(sockaddrs)):
  341. listeners: list[SocketListener] = []
  342. bound_ephemeral_port = local_port
  343. try:
  344. for fam, *_, sockaddr in sockaddrs:
  345. sockaddr = sockaddr[0], bound_ephemeral_port, *sockaddr[2:]
  346. raw_socket = setup_raw_socket(fam, sockaddr)
  347. # Store the assigned port if an ephemeral port was requested, so
  348. # we'll bind to the same port on all interfaces
  349. if local_port == 0 and len(gai_res) > 1:
  350. bound_ephemeral_port = raw_socket.getsockname()[1]
  351. listeners.append(asynclib.create_tcp_listener(raw_socket))
  352. except BaseException as exc:
  353. for listener in listeners:
  354. await listener.aclose()
  355. # If an ephemeral port was requested but binding the assigned port
  356. # failed for another interface, rotate the address list and try again
  357. if (
  358. isinstance(exc, OSError)
  359. and exc.errno == errno.EADDRINUSE
  360. and local_port == 0
  361. and bound_ephemeral_port
  362. ):
  363. errors.append(exc)
  364. sockaddrs.append(sockaddrs.pop(0))
  365. continue
  366. raise
  367. return MultiListener(listeners)
  368. raise OSError(
  369. f"Could not create {len(sockaddrs)} listeners with a consistent port"
  370. ) from ExceptionGroup("Several bind attempts failed", errors)
  371. finally:
  372. del errors # Prevent reference cycles
  373. async def create_unix_listener(
  374. path: str | bytes | PathLike[Any],
  375. *,
  376. mode: int | None = None,
  377. backlog: int = 65536,
  378. ) -> SocketListener:
  379. """
  380. Create a UNIX socket listener.
  381. Not available on Windows.
  382. :param path: path of the socket
  383. :param mode: permissions to set on the socket
  384. :param backlog: maximum number of queued incoming connections (up to a maximum of
  385. 2**16, or 65536)
  386. :return: a listener object
  387. .. versionchanged:: 3.0
  388. If a socket already exists on the file system in the given path, it will be
  389. removed first.
  390. """
  391. backlog = min(backlog, 65536)
  392. raw_socket = await setup_unix_local_socket(path, mode, socket.SOCK_STREAM)
  393. try:
  394. raw_socket.listen(backlog)
  395. return get_async_backend().create_unix_listener(raw_socket)
  396. except BaseException:
  397. raw_socket.close()
  398. raise
  399. async def create_udp_socket(
  400. family: AnyIPAddressFamily = AddressFamily.AF_UNSPEC,
  401. *,
  402. local_host: IPAddressType | None = None,
  403. local_port: int = 0,
  404. reuse_port: bool = False,
  405. ) -> UDPSocket:
  406. """
  407. Create a UDP socket.
  408. If ``port`` has been given, the socket will be bound to this port on the local
  409. machine, making this socket suitable for providing UDP based services.
  410. :param family: address family (``AF_INET`` or ``AF_INET6``) – automatically
  411. determined from ``local_host`` if omitted
  412. :param local_host: IP address or host name of the local interface to bind to
  413. :param local_port: local port to bind to
  414. :param reuse_port: ``True`` to allow multiple sockets to bind to the same
  415. address/port (not supported on Windows)
  416. :return: a UDP socket
  417. """
  418. if family is AddressFamily.AF_UNSPEC and not local_host:
  419. raise ValueError('Either "family" or "local_host" must be given')
  420. if local_host:
  421. gai_res = await getaddrinfo(
  422. str(local_host),
  423. local_port,
  424. family=family,
  425. type=socket.SOCK_DGRAM,
  426. flags=socket.AI_PASSIVE | socket.AI_ADDRCONFIG,
  427. )
  428. family = cast(AnyIPAddressFamily, gai_res[0][0])
  429. local_address = gai_res[0][-1]
  430. elif family is AddressFamily.AF_INET6:
  431. local_address = ("::", 0)
  432. else:
  433. local_address = ("0.0.0.0", 0)
  434. sock = await get_async_backend().create_udp_socket(
  435. family, local_address, None, reuse_port
  436. )
  437. return cast(UDPSocket, sock)
  438. async def create_connected_udp_socket(
  439. remote_host: IPAddressType,
  440. remote_port: int,
  441. *,
  442. family: AnyIPAddressFamily = AddressFamily.AF_UNSPEC,
  443. local_host: IPAddressType | None = None,
  444. local_port: int = 0,
  445. reuse_port: bool = False,
  446. ) -> ConnectedUDPSocket:
  447. """
  448. Create a connected UDP socket.
  449. Connected UDP sockets can only communicate with the specified remote host/port, an
  450. any packets sent from other sources are dropped.
  451. :param remote_host: remote host to set as the default target
  452. :param remote_port: port on the remote host to set as the default target
  453. :param family: address family (``AF_INET`` or ``AF_INET6``) – automatically
  454. determined from ``local_host`` or ``remote_host`` if omitted
  455. :param local_host: IP address or host name of the local interface to bind to
  456. :param local_port: local port to bind to
  457. :param reuse_port: ``True`` to allow multiple sockets to bind to the same
  458. address/port (not supported on Windows)
  459. :return: a connected UDP socket
  460. """
  461. local_address = None
  462. if local_host:
  463. gai_res = await getaddrinfo(
  464. str(local_host),
  465. local_port,
  466. family=family,
  467. type=socket.SOCK_DGRAM,
  468. flags=socket.AI_PASSIVE | socket.AI_ADDRCONFIG,
  469. )
  470. family = cast(AnyIPAddressFamily, gai_res[0][0])
  471. local_address = gai_res[0][-1]
  472. gai_res = await getaddrinfo(
  473. str(remote_host), remote_port, family=family, type=socket.SOCK_DGRAM
  474. )
  475. family = cast(AnyIPAddressFamily, gai_res[0][0])
  476. remote_address = gai_res[0][-1]
  477. sock = await get_async_backend().create_udp_socket(
  478. family, local_address, remote_address, reuse_port
  479. )
  480. return cast(ConnectedUDPSocket, sock)
  481. async def create_unix_datagram_socket(
  482. *,
  483. local_path: None | str | bytes | PathLike[Any] = None,
  484. local_mode: int | None = None,
  485. ) -> UNIXDatagramSocket:
  486. """
  487. Create a UNIX datagram socket.
  488. Not available on Windows.
  489. If ``local_path`` has been given, the socket will be bound to this path, making this
  490. socket suitable for receiving datagrams from other processes. Other processes can
  491. send datagrams to this socket only if ``local_path`` is set.
  492. If a socket already exists on the file system in the ``local_path``, it will be
  493. removed first.
  494. :param local_path: the path on which to bind to
  495. :param local_mode: permissions to set on the local socket
  496. :return: a UNIX datagram socket
  497. """
  498. raw_socket = await setup_unix_local_socket(
  499. local_path, local_mode, socket.SOCK_DGRAM
  500. )
  501. return await get_async_backend().create_unix_datagram_socket(raw_socket, None)
  502. async def create_connected_unix_datagram_socket(
  503. remote_path: str | bytes | PathLike[Any],
  504. *,
  505. local_path: None | str | bytes | PathLike[Any] = None,
  506. local_mode: int | None = None,
  507. ) -> ConnectedUNIXDatagramSocket:
  508. """
  509. Create a connected UNIX datagram socket.
  510. Connected datagram sockets can only communicate with the specified remote path.
  511. If ``local_path`` has been given, the socket will be bound to this path, making
  512. this socket suitable for receiving datagrams from other processes. Other processes
  513. can send datagrams to this socket only if ``local_path`` is set.
  514. If a socket already exists on the file system in the ``local_path``, it will be
  515. removed first.
  516. :param remote_path: the path to set as the default target
  517. :param local_path: the path on which to bind to
  518. :param local_mode: permissions to set on the local socket
  519. :return: a connected UNIX datagram socket
  520. """
  521. remote_path = os.fspath(remote_path)
  522. raw_socket = await setup_unix_local_socket(
  523. local_path, local_mode, socket.SOCK_DGRAM
  524. )
  525. return await get_async_backend().create_unix_datagram_socket(
  526. raw_socket, remote_path
  527. )
  528. async def getaddrinfo(
  529. host: bytes | str | None,
  530. port: str | int | None,
  531. *,
  532. family: int | AddressFamily = 0,
  533. type: int | SocketKind = 0,
  534. proto: int = 0,
  535. flags: int = 0,
  536. ) -> list[tuple[AddressFamily, SocketKind, int, str, tuple[str, int]]]:
  537. """
  538. Look up a numeric IP address given a host name.
  539. Internationalized domain names are translated according to the (non-transitional)
  540. IDNA 2008 standard.
  541. .. note:: 4-tuple IPv6 socket addresses are automatically converted to 2-tuples of
  542. (host, port), unlike what :func:`socket.getaddrinfo` does.
  543. :param host: host name
  544. :param port: port number
  545. :param family: socket family (`'AF_INET``, ...)
  546. :param type: socket type (``SOCK_STREAM``, ...)
  547. :param proto: protocol number
  548. :param flags: flags to pass to upstream ``getaddrinfo()``
  549. :return: list of tuples containing (family, type, proto, canonname, sockaddr)
  550. .. seealso:: :func:`socket.getaddrinfo`
  551. """
  552. # Handle unicode hostnames
  553. encoded_host = idna2008_resolve(host) if isinstance(host, str) else host
  554. gai_res = await get_async_backend().getaddrinfo(
  555. encoded_host, port, family=family, type=type, proto=proto, flags=flags
  556. )
  557. return [
  558. (family, type, proto, canonname, convert_ipv6_sockaddr(sockaddr))
  559. for family, type, proto, canonname, sockaddr in gai_res
  560. # filter out IPv6 results when IPv6 is disabled
  561. if not isinstance(sockaddr[0], int)
  562. ]
  563. def getnameinfo(sockaddr: IPSockAddrType, flags: int = 0) -> Awaitable[tuple[str, str]]:
  564. """
  565. Look up the host name of an IP address.
  566. :param sockaddr: socket address (e.g. (ipaddress, port) for IPv4)
  567. :param flags: flags to pass to upstream ``getnameinfo()``
  568. :return: a tuple of (host name, service name)
  569. :raises NoEventLoopError: if no supported asynchronous event loop is running in the
  570. current thread
  571. .. seealso:: :func:`socket.getnameinfo`
  572. """
  573. return get_async_backend().getnameinfo(sockaddr, flags)
  574. @deprecated("This function is deprecated; use `wait_readable` instead")
  575. def wait_socket_readable(sock: socket.socket) -> Awaitable[None]:
  576. """
  577. .. deprecated:: 4.7.0
  578. Use :func:`wait_readable` instead.
  579. Wait until the given socket has data to be read.
  580. .. warning:: Only use this on raw sockets that have not been wrapped by any higher
  581. level constructs like socket streams!
  582. :param sock: a socket object
  583. :raises ~anyio.ClosedResourceError: if the socket was closed while waiting for the
  584. socket to become readable
  585. :raises ~anyio.BusyResourceError: if another task is already waiting for the socket
  586. to become readable
  587. :raises NoEventLoopError: if no supported asynchronous event loop is running in the
  588. current thread
  589. """
  590. return get_async_backend().wait_readable(sock.fileno())
  591. @deprecated("This function is deprecated; use `wait_writable` instead")
  592. def wait_socket_writable(sock: socket.socket) -> Awaitable[None]:
  593. """
  594. .. deprecated:: 4.7.0
  595. Use :func:`wait_writable` instead.
  596. Wait until the given socket can be written to.
  597. This does **NOT** work on Windows when using the asyncio backend with a proactor
  598. event loop (default on py3.8+).
  599. .. warning:: Only use this on raw sockets that have not been wrapped by any higher
  600. level constructs like socket streams!
  601. :param sock: a socket object
  602. :raises ~anyio.ClosedResourceError: if the socket was closed while waiting for the
  603. socket to become writable
  604. :raises ~anyio.BusyResourceError: if another task is already waiting for the socket
  605. to become writable
  606. :raises NoEventLoopError: if no supported asynchronous event loop is running in the
  607. current thread
  608. """
  609. return get_async_backend().wait_writable(sock.fileno())
  610. def wait_readable(obj: FileDescriptorLike) -> Awaitable[None]:
  611. """
  612. Wait until the given object has data to be read.
  613. On Unix systems, ``obj`` must either be an integer file descriptor, or else an
  614. object with a ``.fileno()`` method which returns an integer file descriptor. Any
  615. kind of file descriptor can be passed, though the exact semantics will depend on
  616. your kernel. For example, this probably won't do anything useful for on-disk files.
  617. On Windows systems, ``obj`` must either be an integer ``SOCKET`` handle, or else an
  618. object with a ``.fileno()`` method which returns an integer ``SOCKET`` handle. File
  619. descriptors aren't supported, and neither are handles that refer to anything besides
  620. a ``SOCKET``.
  621. On backends where this functionality is not natively provided (asyncio
  622. ``ProactorEventLoop`` on Windows), it is provided using a separate selector thread
  623. which is set to shut down when the interpreter shuts down.
  624. .. warning:: Don't use this on raw sockets that have been wrapped by any higher
  625. level constructs like socket streams!
  626. :param obj: an object with a ``.fileno()`` method or an integer handle
  627. :raises ~anyio.ClosedResourceError: if the object was closed while waiting for the
  628. object to become readable
  629. :raises ~anyio.BusyResourceError: if another task is already waiting for the object
  630. to become readable
  631. :raises NoEventLoopError: if no supported asynchronous event loop is running in the
  632. current thread
  633. """
  634. return get_async_backend().wait_readable(obj)
  635. def wait_writable(obj: FileDescriptorLike) -> Awaitable[None]:
  636. """
  637. Wait until the given object can be written to.
  638. :param obj: an object with a ``.fileno()`` method or an integer handle
  639. :raises ~anyio.ClosedResourceError: if the object was closed while waiting for the
  640. object to become writable
  641. :raises ~anyio.BusyResourceError: if another task is already waiting for the object
  642. to become writable
  643. :raises NoEventLoopError: if no supported asynchronous event loop is running in the
  644. current thread
  645. .. seealso:: See the documentation of :func:`wait_readable` for the definition of
  646. ``obj`` and notes on backend compatibility.
  647. .. warning:: Don't use this on raw sockets that have been wrapped by any higher
  648. level constructs like socket streams!
  649. """
  650. return get_async_backend().wait_writable(obj)
  651. def notify_closing(obj: FileDescriptorLike) -> None:
  652. """
  653. Call this before closing a file descriptor (on Unix) or socket (on
  654. Windows). This will cause any `wait_readable` or `wait_writable`
  655. calls on the given object to immediately wake up and raise
  656. `~anyio.ClosedResourceError`.
  657. This doesn't actually close the object – you still have to do that
  658. yourself afterwards. Also, you want to be careful to make sure no
  659. new tasks start waiting on the object in between when you call this
  660. and when it's actually closed. So to close something properly, you
  661. usually want to do these steps in order:
  662. 1. Explicitly mark the object as closed, so that any new attempts
  663. to use it will abort before they start.
  664. 2. Call `notify_closing` to wake up any already-existing users.
  665. 3. Actually close the object.
  666. It's also possible to do them in a different order if that's more
  667. convenient, *but only if* you make sure not to have any checkpoints in
  668. between the steps. This way they all happen in a single atomic
  669. step, so other tasks won't be able to tell what order they happened
  670. in anyway.
  671. :param obj: an object with a ``.fileno()`` method or an integer handle
  672. :raises NoEventLoopError: if no supported asynchronous event loop is running in the
  673. current thread
  674. """
  675. get_async_backend().notify_closing(obj)
  676. #
  677. # Private API
  678. #
  679. def convert_ipv6_sockaddr(
  680. sockaddr: tuple[str, int, int, int] | tuple[str, int],
  681. ) -> tuple[str, int]:
  682. """
  683. Convert a 4-tuple IPv6 socket address to a 2-tuple (address, port) format.
  684. If the scope ID is nonzero, it is added to the address, separated with ``%``.
  685. Otherwise the flow id and scope id are simply cut off from the tuple.
  686. Any other kinds of socket addresses are returned as-is.
  687. :param sockaddr: the result of :meth:`~socket.socket.getsockname`
  688. :return: the converted socket address
  689. """
  690. # This is more complicated than it should be because of MyPy
  691. if isinstance(sockaddr, tuple) and len(sockaddr) == 4:
  692. host, port, flowinfo, scope_id = sockaddr
  693. if scope_id:
  694. # PyPy (as of v7.3.11) leaves the interface name in the result, so
  695. # we discard it and only get the scope ID from the end
  696. # (https://foss.heptapod.net/pypy/pypy/-/issues/3938)
  697. host = host.split("%")[0]
  698. # Add scope_id to the address
  699. return f"{host}%{scope_id}", port
  700. else:
  701. return host, port
  702. else:
  703. return sockaddr
  704. async def setup_unix_local_socket(
  705. path: None | str | bytes | PathLike[Any],
  706. mode: int | None,
  707. socktype: int,
  708. ) -> socket.socket:
  709. """
  710. Create a UNIX local socket object, deleting the socket at the given path if it
  711. exists.
  712. Not available on Windows.
  713. :param path: path of the socket
  714. :param mode: permissions to set on the socket
  715. :param socktype: socket.SOCK_STREAM or socket.SOCK_DGRAM
  716. """
  717. path_str: str | None
  718. if path is not None:
  719. path_str = os.fsdecode(path)
  720. # Linux abstract namespace sockets aren't backed by a concrete file so skip stat call
  721. if not path_str.startswith("\0"):
  722. # Copied from pathlib...
  723. try:
  724. stat_result = os.stat(path)
  725. except OSError as e:
  726. if e.errno not in (
  727. errno.ENOENT,
  728. errno.ENOTDIR,
  729. errno.EBADF,
  730. errno.ELOOP,
  731. ):
  732. raise
  733. else:
  734. if stat.S_ISSOCK(stat_result.st_mode):
  735. os.unlink(path)
  736. else:
  737. path_str = None
  738. raw_socket = socket.socket(socket.AF_UNIX, socktype)
  739. raw_socket.setblocking(False)
  740. if path_str is not None:
  741. try:
  742. await to_thread.run_sync(raw_socket.bind, path_str, abandon_on_cancel=True)
  743. if mode is not None:
  744. await to_thread.run_sync(chmod, path_str, mode, abandon_on_cancel=True)
  745. except BaseException:
  746. raw_socket.close()
  747. raise
  748. return raw_socket
  749. @dataclass
  750. class TCPConnectable(ByteStreamConnectable):
  751. """
  752. Connects to a TCP server at the given host and port.
  753. :param host: host name or IP address of the server
  754. :param port: TCP port number of the server
  755. """
  756. host: str | IPv4Address | IPv6Address
  757. port: int
  758. def __post_init__(self) -> None:
  759. if self.port < 1 or self.port > 65535:
  760. raise ValueError("TCP port number out of range")
  761. @override
  762. async def connect(self) -> SocketStream:
  763. try:
  764. return await connect_tcp(self.host, self.port)
  765. except OSError as exc:
  766. raise ConnectionFailed(
  767. f"error connecting to {self.host}:{self.port}: {exc}"
  768. ) from exc
  769. @dataclass
  770. class UNIXConnectable(ByteStreamConnectable):
  771. """
  772. Connects to a UNIX domain socket at the given path.
  773. :param path: the file system path of the socket
  774. """
  775. path: str | bytes | PathLike[str] | PathLike[bytes]
  776. @override
  777. async def connect(self) -> UNIXSocketStream:
  778. try:
  779. return await connect_unix(self.path)
  780. except OSError as exc:
  781. raise ConnectionFailed(f"error connecting to {self.path!r}: {exc}") from exc
  782. def as_connectable(
  783. remote: ByteStreamConnectable
  784. | tuple[str | IPv4Address | IPv6Address, int]
  785. | str
  786. | bytes
  787. | PathLike[str],
  788. /,
  789. *,
  790. tls: bool = False,
  791. ssl_context: ssl.SSLContext | None = None,
  792. tls_hostname: str | None = None,
  793. tls_standard_compatible: bool = True,
  794. ) -> ByteStreamConnectable:
  795. """
  796. Return a byte stream connectable from the given object.
  797. If a bytestream connectable is given, it is returned unchanged.
  798. If a tuple of (host, port) is given, a TCP connectable is returned.
  799. If a string or bytes path is given, a UNIX connectable is returned.
  800. If ``tls=True``, the connectable will be wrapped in a
  801. :class:`~.streams.tls.TLSConnectable`.
  802. :param remote: a connectable, a tuple of (host, port) or a path to a UNIX socket
  803. :param tls: if ``True``, wrap the plaintext connectable in a
  804. :class:`~.streams.tls.TLSConnectable`, using the provided TLS settings)
  805. :param ssl_context: if ``tls=True``, the SSLContext object to use (if not provided,
  806. a secure default will be created)
  807. :param tls_hostname: if ``tls=True``, host name of the server to use for checking
  808. the server certificate (defaults to the host portion of the address for TCP
  809. connectables)
  810. :param tls_standard_compatible: if ``False`` and ``tls=True``, makes the TLS stream
  811. skip the closing handshake when closing the connection, so it won't raise an
  812. exception if the server does the same
  813. """
  814. connectable: TCPConnectable | UNIXConnectable | TLSConnectable
  815. if isinstance(remote, ByteStreamConnectable):
  816. return remote
  817. elif isinstance(remote, tuple) and len(remote) == 2:
  818. connectable = TCPConnectable(*remote)
  819. elif isinstance(remote, (str, bytes, PathLike)):
  820. connectable = UNIXConnectable(remote)
  821. else:
  822. raise TypeError(f"cannot convert {remote!r} to a connectable")
  823. if tls:
  824. if not tls_hostname and isinstance(connectable, TCPConnectable):
  825. tls_hostname = str(connectable.host)
  826. connectable = TLSConnectable(
  827. connectable,
  828. ssl_context=ssl_context,
  829. hostname=tls_hostname,
  830. standard_compatible=tls_standard_compatible,
  831. )
  832. return connectable