Repository: agronholm/anyio (python)

Issue:
add the ability to wrap stdlib sockets
### Things to check first

- [X] I have searched the existing issues and didn't find my feature already requested there


### Feature description

I'd like the ability to wrap a stdlib socket in a SocketStream/UNIXSocketStream/SocketListener/UDPSocket/ConnectedUDPSocket

Ideally it would be a synchronous function that takes a connected or listening socket of any kind and does the right thing, but asyncio only provides async methods to adopt a socket. It's also hard to detect if a socket is connected or connecting or listening securely for the purposes of wrapping with TLS

Perhaps the best interface for now would be two async functions, `async def wrap_client_sock -> SocketStream | UNIXSocketStream | ConnectedUDPSocket` and `async def wrap_server_sock -> SocketListener | UDPSocket` ?

### Use case

I plan to use it with socket.socketpair in the anyio.to_process.run_sync implementation to fix #516 

Pull request description:
Added class methods for wrapping sockets
<!-- Thank you for your contribution! -->
## Changes

Allows wrapping existing sockets as AnyIO streams or listeners.

Fixes #845. <!-- Provide issue number if exists -->

<!-- Please give a short brief about these changes. -->

## Checklist

If this is a user-facing code change, like a bugfix or a new feature, please ensure that
you've fulfilled the following conditions (where applicable):

- [X] You've added tests (in `tests/`) added which would fail without your patch
- [X] You've updated the documentation (in `docs/`, in case of behavior changes or new
features)
- [X] You've added a new changelog entry (in `docs/versionhistory.rst`).

If this is a trivial change, like a typo fix or a code reformatting, then you can ignore
these instructions.

### Updating the changelog

If there are no entries after the last release, use `**UNRELEASED**` as the version.
If, say, your patch fixes issue <span>#</span>123, the entry should look like this:

```
- Fix big bad boo-boo in task groups
  (`#123 <https://github.com/agronholm/anyio/issues/123>`_; PR by @yourgithubaccount)
```

If there's no issue linked, just link to your pull request instead by updating the
changelog after you've created the PR.


Patch:
diff --git a/docs/networking.rst b/docs/networking.rst
index 795cd9c..d39d3fb 100644
--- a/docs/networking.rst
+++ b/docs/networking.rst
@@ -241,6 +241,20 @@ having to pass the path every time you send data to the peer::
 
     run(main)
 
+Wrapping existing sockets as streams or listeners
+-------------------------------------------------
+
+In some cases, you might want to create a socket in third party code and wrap that as an
+AnyIO stream or socket listener. For that, various class methods exist:
+
+* :meth:`.abc.SocketListener.from_socket`
+* :meth:`.abc.SocketStream.from_socket`
+* :meth:`.abc.UNIXSocketStream.from_socket`
+* :meth:`.abc.UDPSocket.from_socket`
+* :meth:`.abc.ConnectedUDPSocket.from_socket`
+* :meth:`.abc.UNIXDatagramSocket.from_socket`
+* :meth:`.abc.ConnectedUNIXDatagramSocket.from_socket`
+
 Abstracting remote connections using Connectables
 -------------------------------------------------
 
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 680c971..0c30277 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -5,6 +5,15 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
 **UNRELEASED**
 
+- Added various class methods to wrap existing sockets as listeners or socket streams:
+
+  * ``SocketListener.from_socket()``
+  * ``SocketStream.from_socket()``
+  * ``UNIXSocketStream.from_socket()``
+  * ``UDPSocket.from_socket()``
+  * ``ConnectedUDPSocket.from_socket()``
+  * ``UNIXDatagramSocket.from_socket()``
+  * ``ConnectedUNIXDatagramSocket.from_socket()``
 - Added a hierarchy of connectable stream classes for transparently connecting to
   various remote or local endpoints for exchanging bytes or objects
 - Added context manager mix-in classes (``anyio.ContextManagerMixin`` and
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index 07f8a2c..ff9a7e3 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -2854,6 +2854,45 @@ class AsyncIOBackend(AsyncBackend):
 
                     get_selector().remove_reader(fd)
 
+    @classmethod
+    async def wrap_listener_socket(cls, sock: socket.socket) -> SocketListener:
+        return TCPSocketListener(sock)
+
+    @classmethod
+    async def wrap_stream_socket(cls, sock: socket.socket) -> SocketStream:
+        transport, protocol = await get_running_loop().create_connection(
+            StreamProtocol, sock=sock
+        )
+        return SocketStream(transport, protocol)
+
+    @classmethod
+    async def wrap_unix_stream_socket(cls, sock: socket.socket) -> UNIXSocketStream:
+        return UNIXSocketStream(sock)
+
+    @classmethod
+    async def wrap_udp_socket(cls, sock: socket.socket) -> UDPSocket:
+        transport, protocol = await get_running_loop().create_datagram_endpoint(
+            DatagramProtocol, sock=sock
+        )
+        return UDPSocket(transport, protocol)
+
+    @classmethod
+    async def wrap_connected_udp_socket(cls, sock: socket.socket) -> ConnectedUDPSocket:
+        transport, protocol = await get_running_loop().create_datagram_endpoint(
+            DatagramProtocol, sock=sock
+        )
+        return ConnectedUDPSocket(transport, protocol)
+
+    @classmethod
+    async def wrap_unix_datagram_socket(cls, sock: socket.socket) -> UNIXDatagramSocket:
+        return UNIXDatagramSocket(sock)
+
+    @classmethod
+    async def wrap_connected_unix_datagram_socket(
+        cls, sock: socket.socket
+    ) -> ConnectedUNIXDatagramSocket:
+        return ConnectedUNIXDatagramSocket(sock)
+
     @classmethod
     def current_default_thread_limiter(cls) -> CapacityLimiter:
         try:
diff --git a/src/anyio/_backends/_trio.py b/src/anyio/_backends/_trio.py
index 2e77e79..021f111 100644
--- a/src/anyio/_backends/_trio.py
+++ b/src/anyio/_backends/_trio.py
@@ -1286,6 +1286,42 @@ class TrioBackend(AsyncBackend):
     def notify_closing(cls, obj: FileDescriptorLike) -> None:
         notify_closing(obj)
 
+    @classmethod
+    async def wrap_listener_socket(cls, sock: socket.socket) -> abc.SocketListener:
+        return TCPSocketListener(sock)
+
+    @classmethod
+    async def wrap_stream_socket(cls, sock: socket.socket) -> SocketStream:
+        trio_sock = trio.socket.from_stdlib_socket(sock)
+        return SocketStream(trio_sock)
+
+    @classmethod
+    async def wrap_unix_stream_socket(cls, sock: socket.socket) -> UNIXSocketStream:
+        trio_sock = trio.socket.from_stdlib_socket(sock)
+        return UNIXSocketStream(trio_sock)
+
+    @classmethod
+    async def wrap_udp_socket(cls, sock: socket.socket) -> UDPSocket:
+        trio_sock = trio.socket.from_stdlib_socket(sock)
+        return UDPSocket(trio_sock)
+
+    @classmethod
+    async def wrap_connected_udp_socket(cls, sock: socket.socket) -> ConnectedUDPSocket:
+        trio_sock = trio.socket.from_stdlib_socket(sock)
+        return ConnectedUDPSocket(trio_sock)
+
+    @classmethod
+    async def wrap_unix_datagram_socket(cls, sock: socket.socket) -> UNIXDatagramSocket:
+        trio_sock = trio.socket.from_stdlib_socket(sock)
+        return UNIXDatagramSocket(trio_sock)
+
+    @classmethod
+    async def wrap_connected_unix_datagram_socket(
+        cls, sock: socket.socket
+    ) -> ConnectedUNIXDatagramSocket:
+        trio_sock = trio.socket.from_stdlib_socket(sock)
+        return ConnectedUNIXDatagramSocket(trio_sock)
+
     @classmethod
     def current_default_thread_limiter(cls) -> CapacityLimiter:
         try:
diff --git a/src/anyio/abc/_eventloop.py b/src/anyio/abc/_eventloop.py
index c74bee9..5bd9e08 100644
--- a/src/anyio/abc/_eventloop.py
+++ b/src/anyio/abc/_eventloop.py
@@ -348,6 +348,43 @@ class AsyncBackend(metaclass=ABCMeta):
     def notify_closing(cls, obj: FileDescriptorLike) -> None:
         pass
 
+    @classmethod
+    @abstractmethod
+    async def wrap_listener_socket(cls, sock: socket) -> SocketListener:
+        pass
+
+    @classmethod
+    @abstractmethod
+    async def wrap_stream_socket(cls, sock: socket) -> SocketStream:
+        pass
+
+    @classmethod
+    @abstractmethod
+    async def wrap_unix_stream_socket(cls, sock: socket) -> UNIXSocketStream:
+        pass
+
+    @classmethod
+    @abstractmethod
+    async def wrap_udp_socket(cls, sock: socket) -> UDPSocket:
+        pass
+
+    @classmethod
+    @abstractmethod
+    async def wrap_connected_udp_socket(cls, sock: socket) -> ConnectedUDPSocket:
+        pass
+
+    @classmethod
+    @abstractmethod
+    async def wrap_unix_datagram_socket(cls, sock: socket) -> UNIXDatagramSocket:
+        pass
+
+    @classmethod
+    @abstractmethod
+    async def wrap_connected_unix_datagram_socket(
+        cls, sock: socket
+    ) -> ConnectedUNIXDatagramSocket:
+        pass
+
     @classmethod
     @abstractmethod
     def current_default_thread_limiter(cls) -> CapacityLimiter:
diff --git a/src/anyio/abc/_sockets.py b/src/anyio/abc/_sockets.py
index 52d8f41..afd2500 100644
--- a/src/anyio/abc/_sockets.py
+++ b/src/anyio/abc/_sockets.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import errno
 import socket
 import sys
 from abc import abstractmethod
@@ -10,6 +11,7 @@ from ipaddress import IPv4Address, IPv6Address
 from socket import AddressFamily
 from typing import Any, TypeVar, Union
 
+from .._core._eventloop import get_async_backend
 from .._core._typedattr import (
     TypedAttributeProvider,
     TypedAttributeSet,
@@ -31,6 +33,75 @@ UNIXDatagramPacketType: TypeAlias = tuple[bytes, str]
 T_Retval = TypeVar("T_Retval")
 
 
+def _validate_socket(
+    sock_or_fd: socket.socket | int,
+    sock_type: socket.SocketKind,
+    addr_family: socket.AddressFamily = socket.AF_UNSPEC,
+    *,
+    require_connected: bool = False,
+    require_bound: bool = False,
+) -> socket.socket:
+    if isinstance(sock_or_fd, int):
+        try:
+            sock = socket.socket(fileno=sock_or_fd)
+        except OSError as exc:
+            if exc.errno == errno.ENOTSOCK:
+                raise ValueError(
+                    "the file descriptor does not refer to a socket"
+                ) from exc
+            elif require_connected:
+                raise ValueError("the socket must be connected") from exc
+            elif require_bound:
+                raise ValueError("the socket must be bound to a local address") from exc
+            else:
+                raise
+    elif isinstance(sock_or_fd, socket.socket):
+        sock = sock_or_fd
+    else:
+        raise TypeError(
+            f"expected an int or socket, got {type(sock_or_fd).__qualname__} instead"
+        )
+
+    try:
+        if require_connected:
+            try:
+                sock.getpeername()
+            except OSError as exc:
+                raise ValueError("the socket must be connected") from exc
+
+        if require_bound:
+            try:
+                if sock.family in (socket.AF_INET, socket.AF_INET6):
+                    bound_addr = sock.getsockname()[1]
+                else:
+                    bound_addr = sock.getsockname()
+            except OSError:
+                bound_addr = None
+
+            if not bound_addr:
+                raise ValueError("the socket must be bound to a local address")
+
+        if addr_family != socket.AF_UNSPEC and sock.family != addr_family:
+            raise ValueError(
+                f"address family mismatch: expected {addr_family.name}, got "
+                f"{sock.family.name}"
+            )
+
+        if sock.type != sock_type:
+            raise ValueError(
+                f"socket type mismatch: expected {sock_type.name}, got {sock.type.name}"
+            )
+    except BaseException:
+        # Avoid ResourceWarning from the locally constructed socket object
+        if isinstance(sock_or_fd, int):
+            sock.detach()
+
+        raise
+
+    sock.setblocking(False)
+    return sock
+
+
 class SocketAttribute(TypedAttributeSet):
     #: the address family of the underlying socket
     family: AddressFamily = typed_attribute()
@@ -91,8 +162,40 @@ class SocketStream(ByteStream, _SocketProvider):
     Supports all relevant extra attributes from :class:`~SocketAttribute`.
     """
 
+    @classmethod
+    async def from_socket(cls, sock_or_fd: socket.socket | int) -> SocketStream:
+        """
+        Wrap an existing socket object or file descriptor as a socket stream.
+
+        The newly created socket wrapper takes ownership of the socket being passed in.
+        The existing socket must already be connected.
+
+        :param sock_or_fd: a socket object or file descriptor
+        :return: a socket stream
+
+        """
+        sock = _validate_socket(sock_or_fd, socket.SOCK_STREAM, require_connected=True)
+        return await get_async_backend().wrap_stream_socket(sock)
+
 
 class UNIXSocketStream(SocketStream):
+    @classmethod
+    async def from_socket(cls, sock_or_fd: socket.socket | int) -> UNIXSocketStream:
+        """
+        Wrap an existing socket object or file descriptor as a UNIX socket stream.
+
+        The newly created socket wrapper takes ownership of the socket being passed in.
+        The existing socket must already be connected.
+
+        :param sock_or_fd: a socket object or file descriptor
+        :return: a UNIX socket stream
+
+        """
+        sock = _validate_socket(
+            sock_or_fd, socket.SOCK_STREAM, socket.AF_UNIX, require_connected=True
+        )
+        return await get_async_backend().wrap_unix_stream_socket(sock)
+
     @abstractmethod
     async def send_fds(self, message: bytes, fds: Collection[int | IOBase]) -> None:
         """
@@ -121,6 +224,23 @@ class SocketListener(Listener[SocketStream], _SocketProvider):
     Supports all relevant extra attributes from :class:`~SocketAttribute`.
     """
 
+    @classmethod
+    async def from_socket(
+        cls,
+        sock_or_fd: socket.socket | int,
+    ) -> SocketListener:
+        """
+        Wrap an existing socket object or file descriptor as a socket listener.
+
+        The newly created listener takes ownership of the socket being passed in.
+
+        :param sock_or_fd: a socket object or file descriptor
+        :return: a socket listener
+
+        """
+        sock = _validate_socket(sock_or_fd, socket.SOCK_STREAM, require_bound=True)
+        return await get_async_backend().wrap_listener_socket(sock)
+
     @abstractmethod
     async def accept(self) -> SocketStream:
         """Accept an incoming connection."""
@@ -148,6 +268,21 @@ class UDPSocket(UnreliableObjectStream[UDPPacketType], _SocketProvider):
     Supports all relevant extra attributes from :class:`~SocketAttribute`.
     """
 
+    @classmethod
+    async def from_socket(cls, sock_or_fd: socket.socket | int) -> UDPSocket:
+        """
+        Wrap an existing socket object or file descriptor as a UDP socket.
+
+        The newly created socket wrapper takes ownership of the socket being passed in.
+        The existing socket must be bound to a local address.
+
+        :param sock_or_fd: a socket object or file descriptor
+        :return: a UDP socket
+
+        """
+        sock = _validate_socket(sock_or_fd, socket.SOCK_DGRAM, require_bound=True)
+        return await get_async_backend().wrap_udp_socket(sock)
+
     async def sendto(self, data: bytes, host: str, port: int) -> None:
         """
         Alias for :meth:`~.UnreliableObjectSendStream.send` ((data, (host, port))).
@@ -163,6 +298,25 @@ class ConnectedUDPSocket(UnreliableObjectStream[bytes], _SocketProvider):
     Supports all relevant extra attributes from :class:`~SocketAttribute`.
     """
 
+    @classmethod
+    async def from_socket(cls, sock_or_fd: socket.socket | int) -> ConnectedUDPSocket:
+        """
+        Wrap an existing socket object or file descriptor as a connected UDP socket.
+
+        The newly created socket wrapper takes ownership of the socket being passed in.
+        The existing socket must already be connected.
+
+        :param sock_or_fd: a socket object or file descriptor
+        :return: a connected UDP socket
+
+        """
+        sock = _validate_socket(
+            sock_or_fd,
+            socket.SOCK_DGRAM,
+            require_connected=True,
+        )
+        return await get_async_backend().wrap_connected_udp_socket(sock)
+
 
 class UNIXDatagramSocket(
     UnreliableObjectStream[UNIXDatagramPacketType], _SocketProvider
@@ -173,6 +327,24 @@ class UNIXDatagramSocket(
     Supports all relevant extra attributes from :class:`~SocketAttribute`.
     """
 
+    @classmethod
+    async def from_socket(
+        cls,
+        sock_or_fd: socket.socket | int,
+    ) -> UNIXDatagramSocket:
+        """
+        Wrap an existing socket object or file descriptor as a UNIX datagram
+        socket.
+
+        The newly created socket wrapper takes ownership of the socket being passed in.
+
+        :param sock_or_fd: a socket object or file descriptor
+        :return: a UNIX datagram socket
+
+        """
+        sock = _validate_socket(sock_or_fd, socket.SOCK_DGRAM, socket.AF_UNIX)
+        return await get_async_backend().wrap_unix_datagram_socket(sock)
+
     async def sendto(self, data: bytes, path: str) -> None:
         """Alias for :meth:`~.UnreliableObjectSendStream.send` ((data, path))."""
         return await self.send((data, path))
@@ -184,3 +356,24 @@ class ConnectedUNIXDatagramSocket(UnreliableObjectStream[bytes], _SocketProvider
 
     Supports all relevant extra attributes from :class:`~SocketAttribute`.
     """
+
+    @classmethod
+    async def from_socket(
+        cls,
+        sock_or_fd: socket.socket | int,
+    ) -> ConnectedUNIXDatagramSocket:
+        """
+        Wrap an existing socket object or file descriptor as a connected UNIX datagram
+        socket.
+
+        The newly created socket wrapper takes ownership of the socket being passed in.
+        The existing socket must already be connected.
+
+        :param sock_or_fd: a socket object or file descriptor
+        :return: a connected UNIX datagram socket
+
+        """
+        sock = _validate_socket(
+            sock_or_fd, socket.SOCK_DGRAM, socket.AF_UNIX, require_connected=True
+        )
+        return await get_async_backend().wrap_connected_unix_datagram_socket(sock)


Test patch:
diff --git a/tests/test_sockets.py b/tests/test_sockets.py
index 01236e8..eacccec 100644
--- a/tests/test_sockets.py
+++ b/tests/test_sockets.py
@@ -18,7 +18,7 @@ from pathlib import Path
 from socket import AddressFamily
 from ssl import SSLContext, SSLError
 from threading import Thread
-from typing import TYPE_CHECKING, Any, Literal, NoReturn, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Literal, NoReturn, Protocol, TypeVar, cast
 from unittest import mock
 
 import psutil
@@ -61,11 +61,16 @@ from anyio import (
 )
 from anyio._core._eventloop import get_async_backend
 from anyio.abc import (
+    ConnectedUDPSocket,
+    ConnectedUNIXDatagramSocket,
     IPSockAddrType,
     Listener,
     SocketAttribute,
     SocketListener,
     SocketStream,
+    UDPSocket,
+    UNIXDatagramSocket,
+    UNIXSocketStream,
 )
 from anyio.lowlevel import checkpoint
 from anyio.streams.stapled import MultiListener
@@ -142,6 +147,60 @@ def check_asyncio_bug(anyio_backend_name: str, family: AnyIPAddressFamily) -> No
             pytest.skip("Does not work due to a known bug (39148)")
 
 
+class SockFdFactoryProtocol(Protocol):
+    def __call__(
+        self,
+        family: socket.AddressFamily,
+        kind: socket.SocketKind,
+        *,
+        bound: bool = False,
+        connected: bool = False,
+    ) -> socket.socket | int: ...
+
+
+@pytest.fixture(
+    params=[pytest.param(False, id="sock"), pytest.param(True, id="fileno")]
+)
+def sock_or_fd_factory(
+    request: SubRequest, tmp_path_factory: TempPathFactory
+) -> SockFdFactoryProtocol:
+    def factory(
+        family: socket.AddressFamily,
+        kind: socket.SocketKind,
+        *,
+        bound: bool = False,
+        connected: bool = False,
+    ) -> socket.socket | int:
+        sock = socket.socket(family, kind)
+
+        if bound or connected:
+            if family in (socket.AF_INET, socket.AF_INET6):
+                local_addr: str | tuple[str, int] = ("localhost", 0)
+            else:
+                local_addr = str(tmp_path_factory.mktemp("anyio") / "socket")
+
+        if bound:
+            sock.bind(local_addr)
+            if kind == socket.SOCK_STREAM:
+                sock.listen()
+        elif connected:
+            server_sock = socket.socket(family, kind)
+            request.addfinalizer(server_sock.close)
+            server_sock.bind(local_addr)
+            if kind == socket.SOCK_STREAM:
+                server_sock.listen()
+
+            sock.connect(server_sock.getsockname())
+
+        if request.param:
+            return sock.detach()
+        else:
+            request.addfinalizer(sock.close)
+            return sock
+
+    return factory
+
+
 _T = TypeVar("_T")
 
 
@@ -573,6 +632,48 @@ class TestTCPStream:
             )
             assert not caplog_text
 
+    async def test_from_socket(
+        self, family: AnyIPAddressFamily, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(family, socket.SOCK_STREAM, connected=True)
+        async with await SocketStream.from_socket(sock_or_fd) as stream:
+            assert isinstance(stream, SocketStream)
+            assert stream.extra(SocketAttribute.family) == family
+
+    async def test_from_socket_not_a_socket(self) -> None:
+        with pytest.raises(
+            TypeError,
+            match="expected an int or socket, got str instead",
+        ):
+            await SocketStream.from_socket("foo")  # type: ignore[arg-type]
+
+    async def test_from_socket_pass_file_fd(self, tmp_path: Path) -> None:
+        with pytest.raises(
+            ValueError,
+            match="the file descriptor does not refer to a socket",
+        ):
+            with tmp_path.joinpath("foo").open("wb") as fd:
+                await SocketStream.from_socket(fd.fileno())
+
+    async def test_from_socket_wrong_socket_type(
+        self, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(
+            socket.AF_INET, socket.SocketKind.SOCK_DGRAM, connected=True
+        )
+        with pytest.raises(
+            ValueError,
+            match="socket type mismatch: expected SOCK_STREAM, got SOCK_DGRAM",
+        ):
+            await SocketStream.from_socket(sock_or_fd)
+
+    async def test_from_socket_not_connected(
+        self, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(socket.AF_INET, socket.SOCK_STREAM, bound=True)
+        with pytest.raises(ValueError, match="the socket must be connected"):
+            await SocketStream.from_socket(sock_or_fd)
+
 
 @pytest.mark.network
 class TestTCPListener:
@@ -788,6 +889,21 @@ class TestTCPListener:
         async with await create_tcp_listener(local_host=link_local_ipv6_address):
             pass
 
+    async def test_from_socket(
+        self, family: AnyIPAddressFamily, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(family, socket.SOCK_STREAM, bound=True)
+        async with await SocketListener.from_socket(sock_or_fd) as listener:
+            assert isinstance(listener, SocketListener)
+            assert listener.extra(SocketAttribute.family) == family
+
+    async def test_from_socket_not_bound(
+        self, family: AnyIPAddressFamily, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(family, socket.SOCK_STREAM)
+        with pytest.raises(ValueError, match="the socket must be bound"):
+            await SocketListener.from_socket(sock_or_fd)
+
 
 @pytest.mark.skipif(
     sys.platform == "win32", reason="UNIX sockets are not available on Windows"
@@ -1127,6 +1243,46 @@ class TestUNIXStream:
             async with await connect_unix(actual_path):
                 pass
 
+    async def test_from_socket(
+        self, socket_path: Path, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(
+            socket.AF_UNIX, socket.SOCK_STREAM, connected=True
+        )
+        async with await UNIXSocketStream.from_socket(sock_or_fd) as stream:
+            assert isinstance(stream, UNIXSocketStream)
+
+    async def test_from_socket_wrong_socket_type(
+        self, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(
+            socket.AF_UNIX, socket.SOCK_DGRAM, connected=True
+        )
+        with pytest.raises(
+            ValueError,
+            match="socket type mismatch: expected SOCK_STREAM, got SOCK_DGRAM",
+        ):
+            await UNIXSocketStream.from_socket(sock_or_fd)
+
+    async def test_from_socket_wrong_address_family(
+        self, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(
+            socket.AF_INET, socket.SOCK_STREAM, connected=True
+        )
+        with pytest.raises(
+            ValueError,
+            match="address family mismatch: expected AF_UNIX, got AF_INET",
+        ):
+            await UNIXSocketStream.from_socket(sock_or_fd)
+
+    async def test_from_socket_not_connected(
+        self, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(socket.AF_UNIX, socket.SOCK_STREAM)
+        with pytest.raises(ValueError, match="the socket must be connected"):
+            await UNIXSocketStream.from_socket(sock_or_fd)
+
 
 @pytest.mark.skipif(
     sys.platform == "win32", reason="UNIX sockets are not available on Windows"
@@ -1187,7 +1343,7 @@ class TestUNIXListener:
     async def test_socket_options(self, socket_path: Path) -> None:
         async with await create_unix_listener(socket_path) as listener:
             listener_socket = listener.extra(SocketAttribute.raw_socket)
-            assert listener_socket.family == socket.AddressFamily.AF_UNIX
+            assert listener_socket.family == socket.AF_UNIX
             listener_socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 80000)
             assert listener_socket.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) in (
                 80000,
@@ -1250,6 +1406,12 @@ class TestUNIXListener:
         async with await create_unix_listener(real_path):
             pass
 
+    async def test_from_socket(self, sock_or_fd_factory: SockFdFactoryProtocol) -> None:
+        sock_or_fd = sock_or_fd_factory(socket.AF_UNIX, socket.SOCK_STREAM, bound=True)
+        async with await SocketListener.from_socket(sock_or_fd) as listener:
+            assert isinstance(listener, SocketListener)
+            assert listener.extra(SocketAttribute.family) == socket.AF_UNIX
+
 
 async def test_multi_listener(tmp_path_factory: TempPathFactory) -> None:
     async def handle(stream: SocketStream) -> None:
@@ -1272,8 +1434,7 @@ async def test_multi_listener(tmp_path_factory: TempPathFactory) -> None:
                     local_address = listener.extra(SocketAttribute.local_address)
                     if (
                         sys.platform != "win32"
-                        and listener.extra(SocketAttribute.family)
-                        == socket.AddressFamily.AF_UNIX
+                        and listener.extra(SocketAttribute.family) == socket.AF_UNIX
                     ):
                         assert isinstance(local_address, str)
                         stream: SocketStream = await connect_unix(local_address)
@@ -1417,6 +1578,26 @@ class TestUDPSocket:
             )
             assert local_address[1] > 0
 
+    async def test_from_socket(
+        self, family: AnyIPAddressFamily, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(family, socket.SOCK_DGRAM, bound=True)
+        async with await UDPSocket.from_socket(sock_or_fd) as udp_socket:
+            assert isinstance(udp_socket, UDPSocket)
+            assert udp_socket.extra(SocketAttribute.family) == family
+
+    async def test_from_socket_wrong_socket_type(
+        self, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(
+            socket.AF_INET, socket.SOCK_STREAM, connected=True
+        )
+        with pytest.raises(
+            ValueError,
+            match="socket type mismatch: expected SOCK_DGRAM, got SOCK_STREAM",
+        ):
+            await UDPSocket.from_socket(sock_or_fd)
+
 
 @pytest.mark.network
 @pytest.mark.usefixtures("check_asyncio_bug")
@@ -1540,6 +1721,33 @@ class TestConnectedUDPSocket:
         with pytest.raises(ClosedResourceError):
             await udp.send(b"foo")
 
+    async def test_from_socket(
+        self, family: AnyIPAddressFamily, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(family, socket.SOCK_DGRAM, connected=True)
+        async with await ConnectedUDPSocket.from_socket(sock_or_fd) as udp_socket:
+            assert isinstance(udp_socket, ConnectedUDPSocket)
+            assert udp_socket.extra(SocketAttribute.family) == family
+
+    async def test_from_socket_wrong_socket_type(
+        self, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(
+            socket.AF_INET, socket.SOCK_STREAM, connected=True
+        )
+        with pytest.raises(
+            ValueError,
+            match="socket type mismatch: expected SOCK_DGRAM, got SOCK_STREAM",
+        ):
+            await ConnectedUDPSocket.from_socket(sock_or_fd)
+
+    async def test_from_socket_not_connected(
+        self, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(socket.AF_INET, socket.SOCK_DGRAM, bound=True)
+        with pytest.raises(ValueError, match="the socket must be connected"):
+            await ConnectedUDPSocket.from_socket(sock_or_fd)
+
 
 @pytest.mark.skipif(
     sys.platform == "win32", reason="UNIX sockets are not available on Windows"
@@ -1682,6 +1890,22 @@ class TestUNIXDatagramSocket:
         async with await create_unix_datagram_socket(local_path=real_path):
             pass
 
+    async def test_from_socket(self, sock_or_fd_factory: SockFdFactoryProtocol) -> None:
+        sock_or_fd = sock_or_fd_factory(socket.AF_UNIX, socket.SOCK_DGRAM)
+        async with await UNIXDatagramSocket.from_socket(sock_or_fd) as udp_socket:
+            assert isinstance(udp_socket, UNIXDatagramSocket)
+            assert udp_socket.extra(SocketAttribute.family) == socket.AF_UNIX
+
+    async def test_from_socket_wrong_socket_type(
+        self, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(socket.AF_UNIX, socket.SOCK_STREAM)
+        with pytest.raises(
+            ValueError,
+            match="socket type mismatch: expected SOCK_DGRAM, got SOCK_STREAM",
+        ):
+            await UNIXDatagramSocket.from_socket(sock_or_fd)
+
 
 @pytest.mark.skipif(
     sys.platform == "win32", reason="UNIX sockets are not available on Windows"
@@ -1856,6 +2080,35 @@ class TestConnectedUNIXDatagramSocket:
         with pytest.raises(ClosedResourceError):
             await udp.send(b"foo")
 
+    async def test_from_socket(self, sock_or_fd_factory: SockFdFactoryProtocol) -> None:
+        sock_or_fd = sock_or_fd_factory(
+            socket.AF_UNIX, socket.SOCK_DGRAM, connected=True
+        )
+        async with await ConnectedUNIXDatagramSocket.from_socket(
+            sock_or_fd
+        ) as udp_socket:
+            assert isinstance(udp_socket, ConnectedUNIXDatagramSocket)
+            assert udp_socket.extra(SocketAttribute.family) == socket.AF_UNIX
+
+    async def test_from_socket_wrong_socket_type(
+        self, socket_path: Path, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(
+            socket.AF_UNIX, socket.SOCK_STREAM, connected=True
+        )
+        with pytest.raises(
+            ValueError,
+            match="socket type mismatch: expected SOCK_DGRAM, got SOCK_STREAM",
+        ):
+            await ConnectedUNIXDatagramSocket.from_socket(sock_or_fd)
+
+    async def test_from_socket_not_connected(
+        self, sock_or_fd_factory: SockFdFactoryProtocol
+    ) -> None:
+        sock_or_fd = sock_or_fd_factory(socket.AF_UNIX, socket.SOCK_DGRAM, bound=True)
+        with pytest.raises(ValueError, match="the socket must be connected"):
+            await ConnectedUNIXDatagramSocket.from_socket(sock_or_fd)
+
 
 async def test_tcp_connectable(mocker: MockerFixture) -> None:
     mock_connect_tcp = mocker.patch("anyio._core._sockets.connect_tcp")
@@ -1941,9 +2194,7 @@ async def test_getaddrinfo() -> None:
     assert correct != wrong
 
 
-@pytest.mark.parametrize(
-    "sock_type", [socket.SOCK_STREAM, socket.SocketKind.SOCK_STREAM]
-)
+@pytest.mark.parametrize("sock_type", [socket.SOCK_STREAM, socket.SOCK_STREAM])
 async def test_getaddrinfo_ipv6addr(
     sock_type: Literal[socket.SocketKind.SOCK_STREAM],
 ) -> None:
@@ -1951,8 +2202,8 @@ async def test_getaddrinfo_ipv6addr(
     proto = 0 if platform.system() == "Windows" else 6
     assert await getaddrinfo("::1", 0, type=sock_type) == [
         (
-            socket.AddressFamily.AF_INET6,
-            socket.SocketKind.SOCK_STREAM,
+            socket.AF_INET6,
+            socket.SOCK_STREAM,
             proto,
             "",
             ("::1", 0),
@@ -1961,9 +2212,7 @@ async def test_getaddrinfo_ipv6addr(
 
 
 async def test_getaddrinfo_ipv6_disabled() -> None:
-    gai_result = [
-        (AddressFamily.AF_INET6, socket.SocketKind.SOCK_STREAM, 6, "", (1, b""))
-    ]
+    gai_result = [(AddressFamily.AF_INET6, socket.SOCK_STREAM, 6, "", (1, b""))]
     with mock.patch.object(get_async_backend(), "getaddrinfo", return_value=gai_result):
         assert await getaddrinfo("::1", 0) == []