Implement The Annoying Operating System named pipes (a840d8d) - derivepassphrase.git

src/derivepassphrase/ssh_agent/socketprovider.py

@@ -8,15 +8,50 @@ from __future__ import annotations
 
 import collections
 import ctypes
+import ctypes.wintypes
+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",)
 
 
@@ -32,6 +67,311 @@ class TheAnnoyingOsNamedPipesNotAvailableError(NotImplementedError):
     """This Python installation does not support 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
+
+
+# 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:
+    import ctypes
+    import ctypes.wintypes
+
+    try:  # pragma: unless the-annoying-os no cover
+        crypt32 = ctypes.WinDLL("crypt32.dll")  # type: ignore[attr-defined]
+        CryptProtectMemory = crypt32.CryptProtectMemory  # type: ignore[attr-defined]
+    except (AttributeError, FileNotFoundError):  # 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: HANDLE,
+        func: Callable,
+        arguments: Any,  # noqa: ANN401
+    ) -> HANDLE:  # pragma: no cover [external]
+        del func
+        del arguments
+        if result == ctypes.wintypes.HANDLE(INVALID_HANDLE_VALUE):
+            raise ctypes.WinError()  # type: ignore[attr-defined]
+        return result
+
+    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 TheAnnoyingOsNamedPipesNotAvailableError(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]
+
+except ImportError:  # pragma: no cover [failsafe]
+    pass
+
+
+class TheAnnoyingOsNamedPipeHandle:
+    """A named pipe handle on The Annoying OS (Microsoft Windows).
+
+    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:
+        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_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 Annoying OS named pipe name that Pageant 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:
+                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/pageant.{os.getlogin()}.{digest}"
+
+
 class SocketProvider:
     """Static functionality for providing sockets."""
 
@@ -98,15 +438,16 @@ class SocketProvider:
             [windows-ssh-agent-support]: https://the13thletter.info/derivepassphrase/0.x/wishlist/windows-ssh-agent-support/
 
         """
+        try:  # pragma: no cover [external]
+            import ctypes  # noqa: PLC0415
+        except ImportError as exc:  # pragma: no cover [external]
+            msg = "This Python version does not support Windows named pipes."
+            raise TheAnnoyingOsNamedPipesNotAvailableError(msg) from exc
         if not hasattr(ctypes, "WinDLL"):
             msg = "This Python version does not support Windows named pipes."
             raise TheAnnoyingOsNamedPipesNotAvailableError(msg)
         else:  # noqa: RET506  # pragma: unless the-annoying-os no cover
-            msg = (
-                "Communicating with Pageant or OpenSSH on Windows "
-                "is not implemented yet."
-            )
-            raise NotImplementedError(msg)
+            return TheAnnoyingOsNamedPipeHandle.for_pageant()
 
     registry: ClassVar[
         dict[str, _types.SSHAgentSocketProvider | str | None]

...	...	@@ -535,7 +535,7 @@ for key, handler in spawn_handlers.items():
535	535	reason="agent excluded via PERMITTED_AGENTS environment variable",
536	536	),
537	537	]
538		- if key in {"pageant", "ssh-agent", "(system)"}:
	538	+ if key in {"pageant", "ssh-agent"}:
539	539	marks.append(
540	540	pytest.mark.skipif(
541	541	not hasattr(socket, "AF_UNIX"),
542	542

...	...	@@ -8,15 +8,50 @@ from __future__ import annotations
8	8
9	9	import collections
10	10	import ctypes
	11	+import ctypes.wintypes
	12	+import errno
	13	+import hashlib
11	14	import os
12	15	import socket
13		-from typing import TYPE_CHECKING, ClassVar, cast
	16	+from ctypes.wintypes import ( # type: ignore[attr-defined]
	17	+ BOOL,
	18	+ DWORD,
	19	+ HANDLE,
	20	+ LPCVOID,
	21	+ LPCWSTR,
	22	+ LPDWORD,
	23	+ LPVOID,
	24	+)
	25	+from typing import TYPE_CHECKING, cast
14	26
15	27	if TYPE_CHECKING:
16	28	from collections.abc import Callable
	29	+ from typing import ClassVar
	30	+
	31	+ from typing_extensions import Any, Buffer, Literal, Self
17	32
18	33	from derivepassphrase import _types
19	34
	35	+ kernel32: ctypes.WinDLL # type: ignore[name-defined]
	36	+ crypt32: ctypes.WinDLL # type: ignore[name-defined]
	37	+
	38	+ CryptProtectMemory: Callable[[LPVOID, DWORD, DWORD], BOOL]
	39	+ CreateFile: Callable[
	40	+ [
	41	+ LPCWSTR,
	42	+ DWORD,
	43	+ DWORD,
	44	+ None,
	45	+ DWORD,
	46	+ DWORD,
	47	+ HANDLE \| None,
	48	+ ],
	49	+ HANDLE,
	50	+ ]
	51	+ ReadFile: Callable[[HANDLE, LPCVOID, DWORD, LPDWORD, None], BOOL]
	52	+ WriteFile: Callable[[HANDLE, LPCVOID, DWORD, LPDWORD, None], BOOL]
	53	+ CloseHandle: Callable[[HANDLE], BOOL]
	54	+
20	55	__all__ = ("SocketProvider",)
21	56
22	57
...	...	@@ -32,6 +67,311 @@ class TheAnnoyingOsNamedPipesNotAvailableError(NotImplementedError):
32	67	"""This Python installation does not support Windows named pipes."""
33	68
34	69
	70	+GENERIC_READ = 0x80000000
	71	+GENERIC_WRITE = 0x40000000
	72	+
	73	+FILE_SHARE_READ = 0x00000001
	74	+FILE_SHARE_WRITE = 0x00000002
	75	+
	76	+OPEN_EXISTING = 0x3
	77	+
	78	+INVALID_HANDLE_VALUE = -1
	79	+
	80	+CRYPTPROTECTMEMORY_BLOCK_SIZE = 0x10
	81	+CRYPTPROTECTMEMORY_CROSS_PROCESS = 0x1
	82	+
	83	+
	84	+# The type checker and the standard library use different
	85	+# (finer-grained?) types and type definitions on The Annoying Operating
	86	+# System to annotate things derived from ctypes.WinDLL. Furthermore, on
	87	+# other OSes, ctypes.WinDLL is missing entirely. So the type checker
	88	+# attempts to validate our code against very different type
	89	+# hierarchies, depending on what OS it is running on, and so we require
	90	+# a lot of type checking exclusions to keep the types reasonably narrow.
	91	+#
	92	+# Additionally, the official documentation for kernel32.dll and
	93	+# crypt32.dll (and The Annoying OS in general) uses CamelCase for both
	94	+# function names and function parameters. This clashes with the
	95	+# standard Python naming conventions, so we need linting exclusions for
	96	+# each such name.
	97	+#
	98	+# Finally, since this code is mostly platform-specific, we use coverage
	99	+# pragmas for the respective branches to avoid surprises about missing
	100	+# coverage.
	101	+#
	102	+# All these factors necessitate a lot of "type: ignore", "noqa:" and
	103	+# "pragma: ... no cover" comments below.
	104	+try:
	105	+ import ctypes
	106	+ import ctypes.wintypes
	107	+
	108	+ try: # pragma: unless the-annoying-os no cover
	109	+ crypt32 = ctypes.WinDLL("crypt32.dll") # type: ignore[attr-defined]
	110	+ CryptProtectMemory = crypt32.CryptProtectMemory # type: ignore[attr-defined]
	111	+ except (AttributeError, FileNotFoundError): # pragma unless posix no cover
	112	+
	113	+ def CryptProtectMemory( # noqa: N802
	114	+ pDataIn: LPVOID, # noqa: N803
	115	+ cbDataIn: DWORD, # noqa: N803
	116	+ dwFlags: DWORD, # noqa: N803
	117	+ ) -> BOOL: # pragma: no cover [external]
	118	+ raise NotImplementedError
	119	+
	120	+ else: # pragma: unless the-annoying-os no cover
	121	+ CryptProtectMemory.argtypes = [LPVOID, DWORD, DWORD] # type: ignore[attr-defined]
	122	+ CryptProtectMemory.restype = BOOL # type: ignore[attr-defined]
	123	+
	124	+ def errcheck(
	125	+ result: HANDLE,
	126	+ func: Callable,
	127	+ arguments: Any, # noqa: ANN401
	128	+ ) -> HANDLE: # pragma: no cover [external]
	129	+ del func
	130	+ del arguments
	131	+ if result == ctypes.wintypes.HANDLE(INVALID_HANDLE_VALUE):
	132	+ raise ctypes.WinError() # type: ignore[attr-defined]
	133	+ return result
	134	+
	135	+ try: # pragma: unless the-annoying-os no cover
	136	+ kernel32 = ctypes.WinDLL("Kernel32.dll") # type: ignore[attr-defined]
	137	+ CreateFile = kernel32.CreateFileW # type: ignore[attr-defined]
	138	+ ReadFile = kernel32.ReadFile # type: ignore[attr-defined]
	139	+ WriteFile = kernel32.WriteFile # type: ignore[attr-defined]
	140	+ CloseHandle = kernel32.CloseHandle # type: ignore[attr-defined]
	141	+ except (
	142	+ AttributeError,
	143	+ FileNotFoundError,
	144	+ ): # pragma: unless posix no cover
	145	+
	146	+ def CreateFile( # noqa: N802, PLR0913, PLR0917
	147	+ lpFilename: LPCWSTR, # noqa: N803
	148	+ dwDesiredAccess: DWORD, # noqa: N803
	149	+ dwShareMode: DWORD, # noqa: N803
	150	+ lpSecurityAttributes: None, # noqa: N803
	151	+ dwCreationDisposition: DWORD, # noqa: N803
	152	+ dwFlagsAndAttributes: DWORD, # noqa: N803
	153	+ hTemplateFile: HANDLE \| None, # noqa: N803
	154	+ ) -> HANDLE: # pragma: no cover [external]
	155	+ del (
	156	+ lpFilename,
	157	+ dwDesiredAccess,
	158	+ dwShareMode,
	159	+ lpSecurityAttributes,
	160	+ dwCreationDisposition,
	161	+ dwFlagsAndAttributes,
	162	+ hTemplateFile,
	163	+ )
	164	+ msg = "This Python version does not support Windows named pipes."
	165	+ raise TheAnnoyingOsNamedPipesNotAvailableError(msg)
	166	+
	167	+ def ReadFile( # noqa: N802
	168	+ hFile: HANDLE, # noqa: N803
	169	+ lpBuffer: LPVOID, # noqa: N803
	170	+ nNumberOfBytesToRead: DWORD, # noqa: N803
	171	+ lpNumberOfBytesRead: LPDWORD, # noqa: N803
	172	+ lpOverlapped: None, # noqa: N803
	173	+ ) -> BOOL: # pragma: no cover [external]
	174	+ del (
	175	+ hFile,
	176	+ lpBuffer,
	177	+ nNumberOfBytesToRead,
	178	+ lpNumberOfBytesRead,
	179	+ lpOverlapped,
	180	+ )
	181	+ raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
	182	+
	183	+ def WriteFile( # noqa: N802
	184	+ hFile: HANDLE, # noqa: N803
	185	+ lpBuffer: LPVOID, # noqa: N803
	186	+ nNumberOfBytesToWrite: DWORD, # noqa: N803
	187	+ lpNumberOfBytesWritten: LPDWORD, # noqa: N803
	188	+ lpOverlapped: None, # noqa: N803
	189	+ ) -> BOOL: # pragma: no cover [external]
	190	+ del (
	191	+ hFile,
	192	+ lpBuffer,
	193	+ nNumberOfBytesToWrite,
	194	+ lpNumberOfBytesWritten,
	195	+ lpOverlapped,
	196	+ )
	197	+ raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
	198	+
	199	+ else: # pragma: unless the-annoying-os no cover
	200	+ CreateFile.argtypes = [ # type: ignore[attr-defined]
	201	+ LPCWSTR,
	202	+ DWORD,
	203	+ DWORD,
	204	+ # Actually, LPSECURITY_ATTRIBUTES, but we always pass
	205	+ # None/NULL.
	206	+ ctypes.c_void_p,
	207	+ DWORD,
	208	+ DWORD,
	209	+ # We always pass None/NULL.
	210	+ HANDLE,
	211	+ ]
	212	+ CreateFile.restype = HANDLE # type: ignore[attr-defined]
	213	+ CreateFile.errcheck = errcheck # type: ignore[assignment, attr-defined]
	214	+ ReadFile.argtypes = [ # type: ignore[attr-defined]
	215	+ HANDLE,
	216	+ LPVOID,
	217	+ DWORD,
	218	+ LPDWORD,
	219	+ # Actually, LPOVERLAPPED, but we always pass None/NULL.
	220	+ ctypes.c_void_p,
	221	+ ]
	222	+ WriteFile.argtypes = [ # type: ignore[attr-defined]
	223	+ HANDLE,
	224	+ LPVOID,
	225	+ DWORD,
	226	+ LPDWORD,
	227	+ # Actually, LPOVERLAPPED, but we always pass None/NULL.
	228	+ ctypes.c_void_p,
	229	+ ]
	230	+ CloseHandle.argtypes = [HANDLE] # type: ignore[attr-defined]
	231	+ ReadFile.restype = BOOL # type: ignore[attr-defined]
	232	+ WriteFile.restype = BOOL # type: ignore[attr-defined]
	233	+ CloseHandle.restype = BOOL # type: ignore[attr-defined]
	234	+
	235	+except ImportError: # pragma: no cover [failsafe]
	236	+ pass
	237	+
	238	+
	239	+class TheAnnoyingOsNamedPipeHandle:
	240	+ """A named pipe handle on The Annoying OS (Microsoft Windows).
	241	+
	242	+ This handle implements the [`SSHAgentSocket`][_types.SSHAgentSocket]
	243	+ interface. It is only constructable if the Python installation can
	244	+ successfully call into The Annoying OS `kernel32.dll` library via
	245	+ [`ctypes`][].
	246	+
	247	+ """
	248	+
	249	+ _IO_ON_CLOSED_FILE = "I/O operation on closed file."
	250	+
	251	+ def __init__(self, name: str) -> None:
	252	+ self.handle: HANDLE \| None = CreateFile(
	253	+ ctypes.c_wchar_p(name),
	254	+ ctypes.c_ulong(GENERIC_READ \| GENERIC_WRITE),
	255	+ ctypes.c_ulong(FILE_SHARE_READ \| FILE_SHARE_WRITE),
	256	+ None,
	257	+ ctypes.c_ulong(OPEN_EXISTING),
	258	+ ctypes.c_ulong(0),
	259	+ None,
	260	+ )
	261	+
	262	+ def __enter__(self) -> Self:
	263	+ return self
	264	+
	265	+ def __exit__(self, *args: object) -> Literal[False]:
	266	+ try:
	267	+ if self.handle is not None:
	268	+ CloseHandle(self.handle)
	269	+ return False
	270	+ finally:
	271	+ self.handle = None
	272	+
	273	+ def recv(self, data: int, flags: int = 0, /) -> bytes:
	274	+ if self.handle is None:
	275	+ raise ValueError(self._IO_ON_CLOSED_FILE)
	276	+ del flags
	277	+ result = bytearray()
	278	+ read_count = DWORD(0)
	279	+ buffer = (ctypes.c_char * 65536)()
	280	+ while data > 0:
	281	+ block_size = min(max(0, data), 65536)
	282	+ success = ReadFile(
	283	+ self.handle,
	284	+ ctypes.cast(ctypes.byref(buffer), ctypes.c_void_p),
	285	+ DWORD(block_size),
	286	+ ctypes.cast(ctypes.byref(read_count), LPDWORD),
	287	+ None,
	288	+ )
	289	+ if (
	290	+ not success or read_count.value == 0
	291	+ ): # pragma: no cover [external]
	292	+ raise ctypes.WinError() # type: ignore[attr-defined]
	293	+ result.extend(buffer.raw[:block_size])
	294	+ data -= read_count.value
	295	+ read_count.value = 0
	296	+ return bytes(result)
	297	+
	298	+ def sendall(self, data: Buffer, flags: int = 0, /) -> None:
	299	+ if self.handle is None:
	300	+ raise ValueError(self._IO_ON_CLOSED_FILE)
	301	+ del flags
	302	+ data = bytes(data)
	303	+ databuf = (ctypes.c_char * len(data))()
	304	+ for i, x in enumerate(data):
	305	+ databuf[i] = ctypes.c_char(x)
	306	+ write_count = DWORD(0)
	307	+ WriteFile(
	308	+ self.handle,
	309	+ ctypes.cast(ctypes.byref(databuf), ctypes.c_void_p),
	310	+ DWORD(len(data)),
	311	+ ctypes.cast(ctypes.byref(write_count), LPDWORD),
	312	+ None,
	313	+ )
	314	+
	315	+ @classmethod
	316	+ def for_pageant(cls) -> Self:
	317	+ """Construct a named pipe for use with Pageant.
	318	+
	319	+ Returns:
	320	+ A new named pipe handle, using Pageant's pipe name.
	321	+
	322	+ """
	323	+ return cls(cls.pageant_named_pipe_name())
	324	+
	325	+ @staticmethod
	326	+ def pageant_named_pipe_name(
	327	+ *,
	328	+ require_cryptprotectmemory: bool = False,
	329	+ ) -> str:
	330	+ """Return The Annoying OS named pipe name that Pageant would use.
	331	+
	332	+ Args:
	333	+ require_cryptprotectmemory:
	334	+ Pageant normally attempts to use the
	335	+ `CryptProtectMemory` system function from the
	336	+ `crypt32.dll` library on The Annoying OS, ignoring any
	337	+ errors. This in turn influences the resulting named
	338	+ pipe's filename.
	339	+
	340	+ If true, and if we fail to call `CryptProtectMemory`, we
	341	+ abort; otherwise we ignore any errors from calling or
	342	+ failing to call `CryptProtectMemory`.
	343	+
	344	+ Raises:
	345	+ NotImplementedError:
	346	+ This Python version cannot call `CryptProtectMemory` due
	347	+ to lack of system support.
	348	+
	349	+ """
	350	+ realname = b"Pageant\x00"
	351	+ num_blocks = (
	352	+ len(realname) + CRYPTPROTECTMEMORY_BLOCK_SIZE - 1
	353	+ ) // CRYPTPROTECTMEMORY_BLOCK_SIZE
	354	+ assert num_blocks == 1
	355	+ buf_decl = ctypes.c_char * (CRYPTPROTECTMEMORY_BLOCK_SIZE * num_blocks)
	356	+ buf = buf_decl()
	357	+ for i, x in enumerate(realname):
	358	+ buf[i] = ctypes.c_char(x)
	359	+ try:
	360	+ CryptProtectMemory(
	361	+ ctypes.cast(ctypes.byref(buf), ctypes.c_void_p),
	362	+ DWORD(len(buf.raw)),
	363	+ DWORD(CRYPTPROTECTMEMORY_CROSS_PROCESS),
	364	+ )
	365	+ except NotImplementedError: # pragma: no cover [external]
	366	+ if require_cryptprotectmemory:
	367	+ raise
	368	+ buf_as_ssh_string = bytearray()
	369	+ buf_as_ssh_string.extend(int.to_bytes(len(buf.raw), 4, "big"))
	370	+ buf_as_ssh_string.extend(buf.raw)
	371	+ digest = hashlib.sha256(buf_as_ssh_string).hexdigest()
	372	+ return f"//./pipe/pageant.{os.getlogin()}.{digest}"
	373	+
	374	+
35	375	class SocketProvider:
36	376	"""Static functionality for providing sockets."""
37	377
...	...	@@ -98,15 +438,16 @@ class SocketProvider:
98	438	[windows-ssh-agent-support]: https://the13thletter.info/derivepassphrase/0.x/wishlist/windows-ssh-agent-support/
99	439
100	440	"""
	441	+ try: # pragma: no cover [external]
	442	+ import ctypes # noqa: PLC0415
	443	+ except ImportError as exc: # pragma: no cover [external]
	444	+ msg = "This Python version does not support Windows named pipes."
	445	+ raise TheAnnoyingOsNamedPipesNotAvailableError(msg) from exc
101	446	if not hasattr(ctypes, "WinDLL"):
102	447	msg = "This Python version does not support Windows named pipes."
103	448	raise TheAnnoyingOsNamedPipesNotAvailableError(msg)
104	449	else: # noqa: RET506 # pragma: unless the-annoying-os no cover
105		- msg = (
106		- "Communicating with Pageant or OpenSSH on Windows "
107		- "is not implemented yet."
108		- )
109		- raise NotImplementedError(msg)
	450	+ return TheAnnoyingOsNamedPipeHandle.for_pageant()
110	451
111	452	registry: ClassVar[
112	453	dict[str, _types.SSHAgentSocketProvider \| str \| None]