Implement Windows named pipes on The Annoying OS (6ea7220) - derivepassphrase.git

Implement Windows named pipes on The Annoying OS

Marco Ricci commited on 2025-12-14 17:21:00
Zeige 2 geänderte Dateien mit 509 Einfügungen und 43 Löschungen.

Using the `ctypes` module, call into The Annoying OS's system libraries
and bind the functions necessary to create, read from/write to, and
close handles to existing Windows named pipes.  We also bind the
functions necessary to compute the pipe name for PuTTY/Pageant, and
provide convenience functions to connect to PuTTY/Pageant and to
OpenSSH, the de facto main SSH implementations on The Annoying OS.

One major design question remains: how to discover the (named pipe
address for) the SSH agent to use?  Tempting options are using
`SSH_AUTH_SOCK` again (whether literally or with special notation), and
listing the preferred agent addresses in the user configuration file.

For certain "well-known" addresses such as PuTTY/Pageant or OpenSSH, we
provide specific SSH agent socket provider registry entries that attempt
to connect to the respective agent. Other applications could easily
register a custom provider that respects their configuration if the
final socket (address) depends on external configuration.  It is thus
not clear to me if the system needs to be more flexible than it
currently is, e.g., if the SSH agent socket provider needs to accept
arguments that further configure the address or the connection options
to the socket or named pipe.

The formal name for this facility is "Windows named pipe", and this is
the name other developers will expect.  Thus, any mention of The
Annoying OS in the context of Windows named pipes will strictly be
called "Windows named pipe", not "The Annoying OS named pipe" or
similar.  (Some pre-existing names were adapted accordingly.)
Similarly, the other facility is called "UNIX domain socket", not
"AF_UNIX" or "AfUnix" or somesuch.

This commit contains no specific test code for this functionality; we
leave this to follow-up commits.  For ease of integration with the
existing test suite, *in this commit* we attempt to use `SSH_AUTH_SOCK`
directly as the named pipe's name (which will fail unless specifically
prepared).

Since many of the symbols are undefined on other operating systems, and
because the `ctypes` library relies on dynamically generated attributes
and is thus mostly invisible to static analysis tools, much of the code
needs type checking exemptions and extraneous explicit (C) casts at the
Python level.  Additionally, because our stub functions use the same
CamelCase naming as The Annoying OS's official documentation does, much
of the code also needs linting exceptions for the naming policy.

src/derivepassphrase/_internals/cli_helpers.py e231f24..34f258d
src/derivepassphrase/ssh_agent/socketprovider.py 30b6503..10ff68c

src/derivepassphrase/_internals/cli_helpers.py

Zeige Datei @ 6ea7220

@@ -741,12 +741,14 @@ def handle_notimplementederror(
     """  # noqa: DOC201
 
     def handle_notimplementederror(excgroup: BaseExceptionGroup) -> NoReturn:
-        if excgroup.subgroup(socketprovider.AfUnixNotAvailableError):
+        if excgroup.subgroup(
+            socketprovider.UnixDomainSocketsNotAvailableError
+        ):
             warning_callback(
                 _msg.TranslatedString(_msg.WarnMsgTemplate.NO_AF_UNIX)
             )
         if excgroup.subgroup(
-            socketprovider.TheAnnoyingOsNamedPipesNotAvailableError
+            socketprovider.WindowsNamedPipesNotAvailableError
         ):
             warning_callback(
                 _msg.TranslatedString(

src/derivepassphrase/ssh_agent/socketprovider.py

Zeige Datei @ 6ea7220

@@ -8,15 +8,50 @@ from __future__ import annotations
 
 import collections
 import ctypes
+import enum
+import errno
+import hashlib
 import os
 import socket
-from typing import TYPE_CHECKING, ClassVar, cast
+from ctypes.wintypes import (  # type: ignore[attr-defined]
+    BOOL,
+    DWORD,
+    HANDLE,
+    LPCVOID,
+    LPCWSTR,
+    LPDWORD,
+    LPVOID,
+)
+from typing import TYPE_CHECKING, cast
 
 if TYPE_CHECKING:
     from collections.abc import Callable
+    from typing import ClassVar
+
+    from typing_extensions import Any, Buffer, Literal, Self
 
     from derivepassphrase import _types
 
+    kernel32: ctypes.WinDLL  # type: ignore[name-defined]
+    crypt32: ctypes.WinDLL  # type: ignore[name-defined]
+
+    CryptProtectMemory: Callable[[LPVOID, DWORD, DWORD], BOOL]
+    CreateFile: Callable[
+        [
+            LPCWSTR,
+            DWORD,
+            DWORD,
+            None,
+            DWORD,
+            DWORD,
+            HANDLE | None,
+        ],
+        HANDLE,
+    ]
+    ReadFile: Callable[[HANDLE, LPCVOID, DWORD, LPDWORD, None], BOOL]
+    WriteFile: Callable[[HANDLE, LPCVOID, DWORD, LPDWORD, None], BOOL]
+    CloseHandle: Callable[[HANDLE], BOOL]
+
 __all__ = ("SocketProvider",)
 
 
@@ -24,12 +59,384 @@ class NoSuchProviderError(KeyError):
     """No such SSH agent socket provider is known."""
 
 
-class AfUnixNotAvailableError(NotImplementedError):
-    """This Python installation does not support socket.AF_UNIX."""
+class UnixDomainSocketsNotAvailableError(NotImplementedError):
+    """This Python installation does not support UNIX domain sockets.
+
+    On this installation, the standard library symbol
+    [`socket.AF_UNIX`][] is not available, necessary to create UNIX
+    domain sockets.
+
+    """
+
+
+class WindowsNamedPipesNotAvailableError(NotImplementedError):
+    """This Python installation does not support Windows named pipes.
+
+    On this installation, the standard library symbol
+    [`ctypes.WinDLL`][] is not available, necessary to interface with
+    the Annoying Operating System's standard library to create Windows
+    named pipes.
+
+    """
+
+
+GENERIC_READ = 0x80000000
+GENERIC_WRITE = 0x40000000
+
+FILE_SHARE_READ = 0x00000001
+FILE_SHARE_WRITE = 0x00000002
+
+OPEN_EXISTING = 0x3
+
+INVALID_HANDLE_VALUE = -1
+
+CRYPTPROTECTMEMORY_BLOCK_SIZE = 0x10
+CRYPTPROTECTMEMORY_CROSS_PROCESS = 0x1
+
+PIPE_PREFIX = "//./pipe/".replace("/", "\\")
+
+
+# The type checker and the standard library use different
+# (finer-grained?) types and type definitions on The Annoying Operating
+# System to annotate things derived from ctypes.WinDLL.  Furthermore, on
+# other OSes, ctypes.WinDLL is missing entirely.  So the type checker
+# attempts to validate our code against *very* different type
+# hierarchies, depending on what OS it is running on, and so we require
+# a lot of type checking exclusions to keep the types reasonably narrow.
+#
+# Additionally, the official documentation for kernel32.dll and
+# crypt32.dll (and The Annoying OS in general) uses CamelCase for both
+# function names and function parameters.  This clashes with the
+# standard Python naming conventions, so we need linting exclusions for
+# each such name.
+#
+# Finally, since this code is mostly platform-specific, we use coverage
+# pragmas for the respective branches to avoid surprises about missing
+# coverage.
+#
+# All these factors necessitate *a lot* of "type: ignore", "noqa:" and
+# "pragma: ... no cover" comments below.
+
+try:  # pragma: unless the-annoying-os no cover
+    import ctypes
+
+    crypt32 = ctypes.WinDLL("crypt32.dll")  # type: ignore[attr-defined]
+    CryptProtectMemory = crypt32.CryptProtectMemory  # type: ignore[attr-defined]
+except (
+    AttributeError,
+    FileNotFoundError,
+    ImportError,
+):  # pragma unless posix no cover
+
+    def CryptProtectMemory(  # noqa: N802
+        pDataIn: LPVOID,  # noqa: N803
+        cbDataIn: DWORD,  # noqa: N803
+        dwFlags: DWORD,  # noqa: N803
+    ) -> BOOL:  # pragma: no cover [external]
+        raise NotImplementedError
+
+else:  # pragma: unless the-annoying-os no cover
+    CryptProtectMemory.argtypes = [LPVOID, DWORD, DWORD]  # type: ignore[attr-defined]
+    CryptProtectMemory.restype = BOOL  # type: ignore[attr-defined]
+
+
+def _errcheck(
+    result: int,
+    func: Callable,
+    arguments: Any,  # noqa: ANN401
+) -> int:  # pragma: no cover [external]
+    del func
+    del arguments
+    # Even if they have the same value, ctypes pointers don't
+    # compare equal to each other.  We need to compare the values
+    # explicitly, keeping the ranges in mind as well.
+    result_value = HANDLE(result).value
+    invalid_value = HANDLE(INVALID_HANDLE_VALUE).value
+    if result_value == invalid_value:
+        raise ctypes.WinError()  # type: ignore[attr-defined]
+    assert result_value is not None  # for the type checker
+    return result_value
+
+
+try:  # pragma: unless the-annoying-os no cover
+    kernel32 = ctypes.WinDLL("Kernel32.dll")  # type: ignore[attr-defined]
+    CreateFile = kernel32.CreateFileW  # type: ignore[attr-defined]
+    ReadFile = kernel32.ReadFile  # type: ignore[attr-defined]
+    WriteFile = kernel32.WriteFile  # type: ignore[attr-defined]
+    CloseHandle = kernel32.CloseHandle  # type: ignore[attr-defined]
+except (
+    AttributeError,
+    FileNotFoundError,
+):  # pragma: unless posix no cover
+
+    def CreateFile(  # noqa: N802, PLR0913, PLR0917
+        lpFilename: LPCWSTR,  # noqa: N803
+        dwDesiredAccess: DWORD,  # noqa: N803
+        dwShareMode: DWORD,  # noqa: N803
+        lpSecurityAttributes: None,  # noqa: N803
+        dwCreationDisposition: DWORD,  # noqa: N803
+        dwFlagsAndAttributes: DWORD,  # noqa: N803
+        hTemplateFile: HANDLE | None,  # noqa: N803
+    ) -> HANDLE:  # pragma: no cover [external]
+        del (
+            lpFilename,
+            dwDesiredAccess,
+            dwShareMode,
+            lpSecurityAttributes,
+            dwCreationDisposition,
+            dwFlagsAndAttributes,
+            hTemplateFile,
+        )
+        msg = "This Python version does not support Windows named pipes."
+        raise WindowsNamedPipesNotAvailableError(msg)
+
+    def ReadFile(  # noqa: N802
+        hFile: HANDLE,  # noqa: N803
+        lpBuffer: LPVOID,  # noqa: N803
+        nNumberOfBytesToRead: DWORD,  # noqa: N803
+        lpNumberOfBytesRead: LPDWORD,  # noqa: N803
+        lpOverlapped: None,  # noqa: N803
+    ) -> BOOL:  # pragma: no cover [external]
+        del (
+            hFile,
+            lpBuffer,
+            nNumberOfBytesToRead,
+            lpNumberOfBytesRead,
+            lpOverlapped,
+        )
+        raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
+
+    def WriteFile(  # noqa: N802
+        hFile: HANDLE,  # noqa: N803
+        lpBuffer: LPVOID,  # noqa: N803
+        nNumberOfBytesToWrite: DWORD,  # noqa: N803
+        lpNumberOfBytesWritten: LPDWORD,  # noqa: N803
+        lpOverlapped: None,  # noqa: N803
+    ) -> BOOL:  # pragma: no cover [external]
+        del (
+            hFile,
+            lpBuffer,
+            nNumberOfBytesToWrite,
+            lpNumberOfBytesWritten,
+            lpOverlapped,
+        )
+        raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
+
+else:  # pragma: unless the-annoying-os no cover
+    CreateFile.argtypes = [  # type: ignore[attr-defined]
+        LPCWSTR,
+        DWORD,
+        DWORD,
+        # Actually, LPSECURITY_ATTRIBUTES, but we always pass
+        # None/NULL.
+        ctypes.c_void_p,
+        DWORD,
+        DWORD,
+        # We always pass None/NULL.
+        HANDLE,
+    ]
+    CreateFile.restype = HANDLE  # type: ignore[attr-defined]
+    CreateFile.errcheck = _errcheck  # type: ignore[assignment, attr-defined]
+    ReadFile.argtypes = [  # type: ignore[attr-defined]
+        HANDLE,
+        LPVOID,
+        DWORD,
+        LPDWORD,
+        # Actually, LPOVERLAPPED, but we always pass None/NULL.
+        ctypes.c_void_p,
+    ]
+    WriteFile.argtypes = [  # type: ignore[attr-defined]
+        HANDLE,
+        LPVOID,
+        DWORD,
+        LPDWORD,
+        # Actually, LPOVERLAPPED, but we always pass None/NULL.
+        ctypes.c_void_p,
+    ]
+    CloseHandle.argtypes = [HANDLE]  # type: ignore[attr-defined]
+    ReadFile.restype = BOOL  # type: ignore[attr-defined]
+    WriteFile.restype = BOOL  # type: ignore[attr-defined]
+    CloseHandle.restype = BOOL  # type: ignore[attr-defined]
+
+
+class WindowsNamedPipeHandle:
+    """A Windows named pipe handle.
+
+    This handle implements the [`SSHAgentSocket`][_types.SSHAgentSocket]
+    interface.  It is only constructable if the Python installation can
+    successfully call into the Annoying OS `kernel32.dll` library via
+    [`ctypes`][].
+
+    """
+
+    _IO_ON_CLOSED_FILE = "I/O operation on closed file."
+
+    def __init__(self, name: str) -> None:
+        """Create a named pipe handle.
+
+        Args:
+            name:
+                The named pipe's name.
+
+        Raises:
+            OSError:
+                The system failed to open the named pipe.
+            ValueError:
+                The named pipe's name does not designate a named pipe.
+            WindowsNamedPipesNotAvailableError:
+                This Python version does not support Windows named
+                pipes.
+
+        """
+        if not name.replace("/", "\\").startswith(PIPE_PREFIX):
+            raise ValueError(errno.EINVAL, os.strerror(errno.EINVAL))
+        self.handle: HANDLE | None = CreateFile(
+            ctypes.c_wchar_p(name),
+            ctypes.c_ulong(GENERIC_READ | GENERIC_WRITE),
+            ctypes.c_ulong(FILE_SHARE_READ | FILE_SHARE_WRITE),
+            None,
+            ctypes.c_ulong(OPEN_EXISTING),
+            ctypes.c_ulong(0),
+            None,
+        )
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(self, *args: object) -> Literal[False]:
+        try:
+            if self.handle is not None:
+                CloseHandle(self.handle)
+            return False
+        finally:
+            self.handle = None
+
+    def recv(self, data: int, flags: int = 0, /) -> bytes:
+        if self.handle is None:
+            raise ValueError(self._IO_ON_CLOSED_FILE)
+        del flags
+        result = bytearray()
+        read_count = DWORD(0)
+        buffer = (ctypes.c_char * 65536)()
+        while data > 0:
+            block_size = min(max(0, data), 65536)
+            success = ReadFile(
+                self.handle,
+                ctypes.cast(ctypes.byref(buffer), ctypes.c_void_p),
+                DWORD(block_size),
+                ctypes.cast(ctypes.byref(read_count), LPDWORD),
+                None,
+            )
+            if (
+                not success or read_count.value == 0
+            ):  # pragma: no cover [external]
+                raise ctypes.WinError()  # type: ignore[attr-defined]
+            result.extend(buffer.raw[:block_size])
+            data -= read_count.value
+            read_count.value = 0
+        return bytes(result)
+
+    def sendall(self, data: Buffer, flags: int = 0, /) -> None:
+        if self.handle is None:
+            raise ValueError(self._IO_ON_CLOSED_FILE)
+        del flags
+        data = bytes(data)
+        databuf = (ctypes.c_char * len(data))()
+        for i, x in enumerate(data):
+            databuf[i] = ctypes.c_char(x)
+        write_count = DWORD(0)
+        WriteFile(
+            self.handle,
+            ctypes.cast(ctypes.byref(databuf), ctypes.c_void_p),
+            DWORD(len(data)),
+            ctypes.cast(ctypes.byref(write_count), LPDWORD),
+            None,
+        )
+
+    @classmethod
+    def for_openssh(cls) -> Self:
+        """Construct a named pipe for use with OpenSSH on The Annoying OS.
+
+        Returns:
+            A new named pipe handle, using OpenSSH's pipe name.
+
+        """
+        return cls(f"{PIPE_PREFIX}openssh-ssh-agent")
+
+    @classmethod
+    def for_pageant(cls) -> Self:
+        """Construct a named pipe for use with Pageant.
+
+        Returns:
+            A new named pipe handle, using Pageant's pipe name.
+
+        """
+        return cls(cls.pageant_named_pipe_name())
+
+    @staticmethod
+    def pageant_named_pipe_name(
+        *,
+        require_cryptprotectmemory: bool = False,
+    ) -> str:
+        """Return the pipe name that Pageant on The Annoying OS would use.
+
+        Args:
+            require_cryptprotectmemory:
+                Pageant normally attempts to use the
+                `CryptProtectMemory` system function from the
+                `crypt32.dll` library on The Annoying OS, ignoring any
+                errors.  This in turn influences the resulting named
+                pipe's filename.
 
+                If true, and if we fail to call `CryptProtectMemory`, we
+                abort; otherwise we ignore any errors from calling or
+                failing to call `CryptProtectMemory`.
+
+        Raises:
+            NotImplementedError:
+                `require_cryptprotectmemory` was `True`, but this Python
+                version cannot call `CryptProtectMemory` due to lack of
+                system support.
+
+        """
+        realname = b"Pageant\x00"
+        num_blocks = (
+            len(realname) + CRYPTPROTECTMEMORY_BLOCK_SIZE - 1
+        ) // CRYPTPROTECTMEMORY_BLOCK_SIZE
+        assert num_blocks == 1
+        buf_decl = ctypes.c_char * (CRYPTPROTECTMEMORY_BLOCK_SIZE * num_blocks)
+        buf = buf_decl()
+        for i, x in enumerate(realname):
+            buf[i] = ctypes.c_char(x)
+        try:
+            CryptProtectMemory(
+                ctypes.cast(ctypes.byref(buf), ctypes.c_void_p),
+                DWORD(len(buf.raw)),
+                DWORD(CRYPTPROTECTMEMORY_CROSS_PROCESS),
+            )
+        except NotImplementedError:  # pragma: no cover [external]
+            if require_cryptprotectmemory:
+                raise
+        buf_as_ssh_string = bytearray()
+        buf_as_ssh_string.extend(int.to_bytes(len(buf.raw), 4, "big"))
+        buf_as_ssh_string.extend(buf.raw)
+        digest = hashlib.sha256(buf_as_ssh_string).hexdigest()
+        return f"{PIPE_PREFIX}pageant.{os.getlogin()}.{digest}"
 
-class TheAnnoyingOsNamedPipesNotAvailableError(NotImplementedError):
-    """This Python installation does not support Windows named pipes."""
+
+class _WindowsNamedPipeSocketAddress(enum.Enum):
+    """Internal enum for the socket address for Windows named pipes.
+
+    Attributes:
+        PAGEANT: The dynamic address for PuTTY/Pageant.
+        OPENSSH: The static address for OpenSSH-on-Windows.
+
+    """
+
+    PAGEANT = "PuTTY/Pageant on The Annoying OS"
+    """"""
+    OPENSSH = "OpenSSH on The Annoying OS"
+    """"""
 
 
 class SocketProvider:
@@ -52,13 +459,11 @@ class SocketProvider:
         Raises:
             KeyError:
                 The `SSH_AUTH_SOCK` environment variable was not found.
-            AfUnixNotAvailableError:
-                [This Python version does not support UNIX domain
-                sockets][AF_UNIX], necessary to automatically connect to
+            UnixDomainSocketsNotAvailableError:
+                This Python version does not support UNIX domain
+                sockets, necessary to automatically connect to
                 a running SSH agent via the `SSH_AUTH_SOCK` environment
                 variable.
-
-                [AF_UNIX]: https://docs.python.org/3/library/socket.html#socket.AF_UNIX
             OSError:
                 There was an error setting up a socket connection to the
                 agent.
@@ -66,7 +471,7 @@ class SocketProvider:
         """
         if not hasattr(socket, "AF_UNIX"):
             msg = "This Python version does not support UNIX domain sockets."
-            raise AfUnixNotAvailableError(msg)
+            raise UnixDomainSocketsNotAvailableError(msg)
         else:  # noqa: RET506  # pragma: unless posix no cover
             sock = socket.socket(family=socket.AF_UNIX)
             if "SSH_AUTH_SOCK" not in os.environ:
@@ -78,35 +483,92 @@ class SocketProvider:
             return sock
 
     @staticmethod
-    def the_annoying_os_named_pipes() -> _types.SSHAgentSocket:
-        """Return a socket connected to Pageant/OpenSSH on The Annoying OS.
-
-        This may be a write-through socket if the underlying connection
-        to Pageant or OpenSSH does not use an actual network socket to
-        communicate.
+    def _the_annoying_os_named_pipe(
+        pipe_name: _WindowsNamedPipeSocketAddress | str,
+    ) -> _types.SSHAgentSocket:
+        """Return a socket wrapper around a Windows named pipe.
 
         Raises:
-            TheAnnoyingOsNamedPipesNotAvailableError:
-                This functionality is not implemented yet.
-
-        Warning: Not implemented yet
-            This functionality is not implemented yet.  Specifically,
-            [we do not yet support any of the communication mechanisms
-            used by the leading SSH agent
-            implementations.][windows-ssh-agent-support]
-
-            [windows-ssh-agent-support]: https://the13thletter.info/derivepassphrase/0.x/wishlist/windows-ssh-agent-support/
+            OSError:
+                There was an error setting up a connection to the agent.
+            WindowsNamedPipesNotAvailableError:
+                This Python version does not support Windows named
+                pipes.
 
         """
         if not hasattr(ctypes, "WinDLL"):
             msg = "This Python version does not support Windows named pipes."
-            raise TheAnnoyingOsNamedPipesNotAvailableError(msg)
+            raise WindowsNamedPipesNotAvailableError(msg)
         else:  # noqa: RET506  # pragma: unless the-annoying-os no cover
-            msg = (
-                "Communicating with Pageant or OpenSSH on Windows "
-                "is not implemented yet."
+            # TODO(the-13th-letter): Rewrite using structural pattern
+            # matching.
+            # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
+            if pipe_name == _WindowsNamedPipeSocketAddress.PAGEANT:
+                return WindowsNamedPipeHandle.for_pageant()
+            if pipe_name == _WindowsNamedPipeSocketAddress.OPENSSH:
+                return WindowsNamedPipeHandle.for_openssh()
+            return WindowsNamedPipeHandle(pipe_name)
+
+    @classmethod
+    def the_annoying_os_named_pipe_for_pageant(cls) -> _types.SSHAgentSocket:
+        """Return a socket wrapper connected to Pageant on The Annoying OS.
+
+        Raises:
+            OSError:
+                There was an error setting up a connection to Pageant.
+            WindowsNamedPipesNotAvailableError:
+                This Python version does not support Windows named
+                pipes.
+
+        """
+        return cls._the_annoying_os_named_pipe(
+            _WindowsNamedPipeSocketAddress.PAGEANT
         )
-            raise NotImplementedError(msg)
+
+    @classmethod
+    def the_annoying_os_named_pipe_for_openssh(cls) -> _types.SSHAgentSocket:
+        """Return a socket wrapper connected to the OpenSSH agent on The Annoying OS.
+
+        Raises:
+            OSError:
+                There was an error setting up a connection to the OpenSSH
+                agent.
+            WindowsNamedPipesNotAvailableError:
+                This Python version does not support Windows named
+                pipes.
+
+        """  # noqa: E501
+        return cls._the_annoying_os_named_pipe(
+            _WindowsNamedPipeSocketAddress.OPENSSH
+        )
+
+    @classmethod
+    def the_annoying_os_named_pipe_ssh_auth_sock(cls) -> _types.SSHAgentSocket:
+        r"""Return a socket wrapper connected to the agent in `SSH_AUTH_SOCK`.
+
+        The `SSH_AUTH_SOCK` environment variable is assumed to contain
+        a valid named pipe name, i.e., a path starting with `\\.\pipe\`
+        or `//./pipe/`.
+
+        Raises:
+            KeyError:
+                The `SSH_AUTH_SOCK` environment variable was not found.
+            ValueError:
+                The path in `SSH_AUTH_SOCK` clearly does not name a valid
+                named pipe name.
+            OSError:
+                There was an error setting up a connection to the OpenSSH
+                agent.
+            WindowsNamedPipesNotAvailableError:
+                This Python version does not support Windows named
+                pipes.
+
+        """
+        address = os.environ["SSH_AUTH_SOCK"]
+        if not address.replace("/", "\\").startswith(PIPE_PREFIX):
+            msg = f"Invalid named pipe name: {address!r}"
+            raise ValueError(msg)
+        return cls._the_annoying_os_named_pipe(address)
 
     registry: ClassVar[
         dict[str, _types.SSHAgentSocketProvider | str | None]
@@ -337,21 +799,23 @@ class SocketProvider:
 
 
 SocketProvider.registry.update({
-    "posix": SocketProvider.unix_domain_ssh_auth_sock,
-    "the_annoying_os": SocketProvider.the_annoying_os_named_pipes,
+    "ssh_auth_sock_on_posix": SocketProvider.unix_domain_ssh_auth_sock,
+    "pageant_on_the_annoying_os": SocketProvider.the_annoying_os_named_pipe_for_pageant,  # noqa: E501
+    "openssh_on_the_annoying_os": SocketProvider.the_annoying_os_named_pipe_for_openssh,  # noqa: E501
+    "ssh_auth_sock_on_the_annoying_os": SocketProvider.the_annoying_os_named_pipe_ssh_auth_sock,  # noqa: E501
     # known instances
     "stub_agent": None,
     "stub_with_address": None,
     "stub_with_address_and_deterministic_dsa": None,
     # aliases
-    "native": "the_annoying_os" if os.name == "nt" else "posix",
-    "unix_domain": "posix",
-    "ssh_auth_sock": "posix",
+    "posix": "ssh_auth_sock_on_posix",
+    "ssh_auth_sock": "ssh_auth_sock_on_posix",
+    "unix_domain": "ssh_auth_sock_on_posix",
+    "the_annoying_os": "pageant_on_the_annoying_os",
     "the_annoying_os_named_pipe": "the_annoying_os",
-    "pageant_on_the_annoying_os": "the_annoying_os",
-    "openssh_on_the_annoying_os": "the_annoying_os",
     "windows": "the_annoying_os",
     "windows_named_pipe": "the_annoying_os",
-    "pageant_on_windows": "the_annoying_os",
-    "openssh_on_windows": "the_annoying_os",
+    "pageant_on_windows": "pageant_on_the_annoying_os",
+    "openssh_on_windows": "openssh_on_the_annoying_os",
+    "native": "the_annoying_os" if os.name == "nt" else "posix",
 })


...	...	@@ -741,12 +741,14 @@ def handle_notimplementederror(
741	741	""" # noqa: DOC201
742	742
743	743	def handle_notimplementederror(excgroup: BaseExceptionGroup) -> NoReturn:
744		- if excgroup.subgroup(socketprovider.AfUnixNotAvailableError):
	744	+ if excgroup.subgroup(
	745	+ socketprovider.UnixDomainSocketsNotAvailableError
	746	+ ):
745	747	warning_callback(
746	748	_msg.TranslatedString(_msg.WarnMsgTemplate.NO_AF_UNIX)
747	749	)
748	750	if excgroup.subgroup(
749		- socketprovider.TheAnnoyingOsNamedPipesNotAvailableError
	751	+ socketprovider.WindowsNamedPipesNotAvailableError
750	752	):
751	753	warning_callback(
752	754	_msg.TranslatedString(