Implement The Annoying Operating System named pipes (a637416) - 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, 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,263 @@ 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
+
+CRYPTPROTECTMEMORY_BLOCK_SIZE = 0x10
+CRYPTPROTECTMEMORY_CROSS_PROCESS = 0x1
+
+
+try:
+    import ctypes
+    import ctypes.wintypes
+
+    try:
+        crypt32 = ctypes.WinDLL("crypt32.dll")  # type: ignore[attr-defined]
+        CryptProtectMemory = crypt32.CryptProtectMemory  # type: ignore[attr-defined]
+    except (AttributeError, FileNotFoundError):
+
+        def CryptProtectMemory(  # noqa: N802
+            pDataIn: LPVOID,  # noqa: N803
+            cbDataIn: DWORD,  # noqa: N803
+            dwFlags: DWORD,  # noqa: N803
+        ) -> BOOL:
+            raise NotImplementedError
+
+    else:
+        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:
+        del func
+        del arguments
+        invalid_handle = ctypes.wintypes.HANDLE(0)
+        if result == invalid_handle:
+            raise ctypes.WinError()  # type: ignore[attr-defined]
+        return result
+
+    try:
+        kernel32 = ctypes.WinDLL("Kernel32.dll")  # type: ignore[attr-defined]
+        CreateFile = kernel32.CreateFileW  # type: ignore[attr-defined]
+        ReadFile = kernel32.ReadFileW  # type: ignore[attr-defined]
+        WriteFile = kernel32.WriteFileW  # type: ignore[attr-defined]
+        CloseHandle = kernel32.CloseHandle  # type: ignore[attr-defined]
+    except (AttributeError, FileNotFoundError):
+
+        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:
+            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:
+            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:
+            del (
+                hFile,
+                lpBuffer,
+                nNumberOfBytesToWrite,
+                lpNumberOfBytesWritten,
+                lpOverlapped,
+            )
+            raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
+
+    else:
+        CreateFile.argtypes = [  # type: ignore[attr-defined]
+            LPCWSTR,
+            DWORD,
+            DWORD,
+            None,
+            DWORD,
+            DWORD,
+            HANDLE | None,
+        ]
+        CreateFile.restype = HANDLE  # type: ignore[attr-defined]
+        CreateFile.errcheck = errcheck  # type: ignore[attr-defined]
+        ReadFile.argtypes = [HANDLE, LPVOID, DWORD, LPDWORD, None]  # type: ignore[attr-defined]
+        WriteFile.argtypes = [HANDLE, LPVOID, DWORD, LPDWORD, None]  # type: ignore[attr-defined]
+        CloseHandle.argtypes = [HANDLE]  # type: ignore[attr-defined]
+        ReadFile.argtypes = [BOOL]  # type: ignore[attr-defined]
+        WriteFile.argtypes = [BOOL]  # type: ignore[attr-defined]
+        CloseHandle.argtypes = [BOOL]  # type: ignore[attr-defined]
+
+except ImportError:
+    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`][].
+
+    """
+
+    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) -> bool | None:
+        try:
+            if self.handle is not None:
+                return CloseHandle(self.handle).value != 0
+            return False
+        finally:
+            self.handle = None
+
+    def recv(self, data: int, flags: int = 0, /) -> bytes:
+        if self.handle is None:
+            raise OSError(errno.EBADF, os.strerror(errno.EBADF))
+        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:
+                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 OSError(errno.EBADF, os.strerror(errno.EBADF))
+        del flags
+        data = bytes(data)
+        databuf = (ctypes.c_char * len(data))(data)
+        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 = "Pageant"
+        c_realname = ctypes.create_string_buffer(realname.encode("UTF-8"))
+        num_blocks = (
+            len(c_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(c_realname.raw)
+        try:
+            CryptProtectMemory(
+                ctypes.cast(ctypes.byref(buf), ctypes.c_void_p),
+                DWORD(len(buf.raw)),
+                DWORD(CRYPTPROTECTMEMORY_CROSS_PROCESS),
+            )
+        except NotImplementedError:
+            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 +390,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]


...	...	@@ -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, 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,263 @@ 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	+CRYPTPROTECTMEMORY_BLOCK_SIZE = 0x10
	79	+CRYPTPROTECTMEMORY_CROSS_PROCESS = 0x1
	80	+
	81	+
	82	+try:
	83	+ import ctypes
	84	+ import ctypes.wintypes
	85	+
	86	+ try:
	87	+ crypt32 = ctypes.WinDLL("crypt32.dll") # type: ignore[attr-defined]
	88	+ CryptProtectMemory = crypt32.CryptProtectMemory # type: ignore[attr-defined]
	89	+ except (AttributeError, FileNotFoundError):
	90	+
	91	+ def CryptProtectMemory( # noqa: N802
	92	+ pDataIn: LPVOID, # noqa: N803
	93	+ cbDataIn: DWORD, # noqa: N803
	94	+ dwFlags: DWORD, # noqa: N803
	95	+ ) -> BOOL:
	96	+ raise NotImplementedError
	97	+
	98	+ else:
	99	+ CryptProtectMemory.argtypes = [LPVOID, DWORD, DWORD] # type: ignore[attr-defined]
	100	+ CryptProtectMemory.restype = BOOL # type: ignore[attr-defined]
	101	+
	102	+ def errcheck(
	103	+ result: HANDLE,
	104	+ func: Callable,
	105	+ arguments: Any, # noqa: ANN401
	106	+ ) -> HANDLE:
	107	+ del func
	108	+ del arguments
	109	+ invalid_handle = ctypes.wintypes.HANDLE(0)
	110	+ if result == invalid_handle:
	111	+ raise ctypes.WinError() # type: ignore[attr-defined]
	112	+ return result
	113	+
	114	+ try:
	115	+ kernel32 = ctypes.WinDLL("Kernel32.dll") # type: ignore[attr-defined]
	116	+ CreateFile = kernel32.CreateFileW # type: ignore[attr-defined]
	117	+ ReadFile = kernel32.ReadFileW # type: ignore[attr-defined]
	118	+ WriteFile = kernel32.WriteFileW # type: ignore[attr-defined]
	119	+ CloseHandle = kernel32.CloseHandle # type: ignore[attr-defined]
	120	+ except (AttributeError, FileNotFoundError):
	121	+
	122	+ def CreateFile( # noqa: N802, PLR0913, PLR0917
	123	+ lpFilename: LPCWSTR, # noqa: N803
	124	+ dwDesiredAccess: DWORD, # noqa: N803
	125	+ dwShareMode: DWORD, # noqa: N803
	126	+ lpSecurityAttributes: None, # noqa: N803
	127	+ dwCreationDisposition: DWORD, # noqa: N803
	128	+ dwFlagsAndAttributes: DWORD, # noqa: N803
	129	+ hTemplateFile: HANDLE \| None, # noqa: N803
	130	+ ) -> HANDLE:
	131	+ del (
	132	+ lpFilename,
	133	+ dwDesiredAccess,
	134	+ dwShareMode,
	135	+ lpSecurityAttributes,
	136	+ dwCreationDisposition,
	137	+ dwFlagsAndAttributes,
	138	+ hTemplateFile,
	139	+ )
	140	+ msg = "This Python version does not support Windows named pipes."
	141	+ raise TheAnnoyingOsNamedPipesNotAvailableError(msg)
	142	+
	143	+ def ReadFile( # noqa: N802
	144	+ hFile: HANDLE, # noqa: N803
	145	+ lpBuffer: LPVOID, # noqa: N803
	146	+ nNumberOfBytesToRead: DWORD, # noqa: N803
	147	+ lpNumberOfBytesRead: LPDWORD, # noqa: N803
	148	+ lpOverlapped: None, # noqa: N803
	149	+ ) -> BOOL:
	150	+ del (
	151	+ hFile,
	152	+ lpBuffer,
	153	+ nNumberOfBytesToRead,
	154	+ lpNumberOfBytesRead,
	155	+ lpOverlapped,
	156	+ )
	157	+ raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
	158	+
	159	+ def WriteFile( # noqa: N802
	160	+ hFile: HANDLE, # noqa: N803
	161	+ lpBuffer: LPVOID, # noqa: N803
	162	+ nNumberOfBytesToWrite: DWORD, # noqa: N803
	163	+ lpNumberOfBytesWritten: LPDWORD, # noqa: N803
	164	+ lpOverlapped: None, # noqa: N803
	165	+ ) -> BOOL:
	166	+ del (
	167	+ hFile,
	168	+ lpBuffer,
	169	+ nNumberOfBytesToWrite,
	170	+ lpNumberOfBytesWritten,
	171	+ lpOverlapped,
	172	+ )
	173	+ raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP))
	174	+
	175	+ else:
	176	+ CreateFile.argtypes = [ # type: ignore[attr-defined]
	177	+ LPCWSTR,
	178	+ DWORD,
	179	+ DWORD,
	180	+ None,
	181	+ DWORD,
	182	+ DWORD,
	183	+ HANDLE \| None,
	184	+ ]
	185	+ CreateFile.restype = HANDLE # type: ignore[attr-defined]
	186	+ CreateFile.errcheck = errcheck # type: ignore[attr-defined]
	187	+ ReadFile.argtypes = [HANDLE, LPVOID, DWORD, LPDWORD, None] # type: ignore[attr-defined]
	188	+ WriteFile.argtypes = [HANDLE, LPVOID, DWORD, LPDWORD, None] # type: ignore[attr-defined]
	189	+ CloseHandle.argtypes = [HANDLE] # type: ignore[attr-defined]
	190	+ ReadFile.argtypes = [BOOL] # type: ignore[attr-defined]
	191	+ WriteFile.argtypes = [BOOL] # type: ignore[attr-defined]
	192	+ CloseHandle.argtypes = [BOOL] # type: ignore[attr-defined]
	193	+
	194	+except ImportError:
	195	+ pass
	196	+
	197	+
	198	+class TheAnnoyingOsNamedPipeHandle:
	199	+ """A named pipe handle on The Annoying OS (Microsoft Windows).
	200	+
	201	+ This handle implements the [`SSHAgentSocket`][_types.SSHAgentSocket]
	202	+ interface. It is only constructable if the Python installation can
	203	+ successfully call into The Annoying OS `kernel32.dll` library via
	204	+ [`ctypes`][].
	205	+
	206	+ """
	207	+
	208	+ def __init__(self, name: str) -> None:
	209	+ self.handle: HANDLE \| None = CreateFile(
	210	+ ctypes.c_wchar_p(name),
	211	+ ctypes.c_ulong(GENERIC_READ \| GENERIC_WRITE),
	212	+ ctypes.c_ulong(FILE_SHARE_READ \| FILE_SHARE_WRITE),
	213	+ None,
	214	+ ctypes.c_ulong(OPEN_EXISTING),
	215	+ ctypes.c_ulong(0),
	216	+ None,
	217	+ )
	218	+
	219	+ def __enter__(self) -> Self:
	220	+ return self
	221	+
	222	+ def __exit__(self, *args: object) -> bool \| None:
	223	+ try:
	224	+ if self.handle is not None:
	225	+ return CloseHandle(self.handle).value != 0
	226	+ return False
	227	+ finally:
	228	+ self.handle = None
	229	+
	230	+ def recv(self, data: int, flags: int = 0, /) -> bytes:
	231	+ if self.handle is None:
	232	+ raise OSError(errno.EBADF, os.strerror(errno.EBADF))
	233	+ del flags
	234	+ result = bytearray()
	235	+ read_count = DWORD(0)
	236	+ buffer = (ctypes.c_char * 65536)()
	237	+ while data > 0:
	238	+ block_size = min(max(0, data), 65536)
	239	+ success = ReadFile(
	240	+ self.handle,
	241	+ ctypes.cast(ctypes.byref(buffer), ctypes.c_void_p),
	242	+ DWORD(block_size),
	243	+ ctypes.cast(ctypes.byref(read_count), LPDWORD),
	244	+ None,
	245	+ )
	246	+ if not success or read_count.value == 0:
	247	+ raise ctypes.WinError() # type: ignore[attr-defined]
	248	+ result.extend(buffer.raw[:block_size])
	249	+ data -= read_count.value
	250	+ read_count.value = 0
	251	+ return bytes(result)
	252	+
	253	+ def sendall(self, data: Buffer, flags: int = 0, /) -> None:
	254	+ if self.handle is None:
	255	+ raise OSError(errno.EBADF, os.strerror(errno.EBADF))
	256	+ del flags
	257	+ data = bytes(data)
	258	+ databuf = (ctypes.c_char * len(data))(data)
	259	+ write_count = DWORD(0)
	260	+ WriteFile(
	261	+ self.handle,
	262	+ ctypes.cast(ctypes.byref(databuf), ctypes.c_void_p),
	263	+ DWORD(len(data)),
	264	+ ctypes.cast(ctypes.byref(write_count), LPDWORD),
	265	+ None,
	266	+ )
	267	+
	268	+ @classmethod
	269	+ def for_pageant(cls) -> Self:
	270	+ """Construct a named pipe for use with Pageant.
	271	+
	272	+ Returns:
	273	+ A new named pipe handle, using Pageant's pipe name.
	274	+
	275	+ """
	276	+ return cls(cls.pageant_named_pipe_name())
	277	+
	278	+ @staticmethod
	279	+ def pageant_named_pipe_name(
	280	+ *,
	281	+ require_cryptprotectmemory: bool = False,
	282	+ ) -> str:
	283	+ """Return The Annoying OS named pipe name that Pageant would use.
	284	+
	285	+ Args:
	286	+ require_cryptprotectmemory:
	287	+ Pageant normally attempts to use the
	288	+ `CryptProtectMemory` system function from the
	289	+ `crypt32.dll` library on The Annoying OS, ignoring any
	290	+ errors. This in turn influences the resulting named
	291	+ pipe's filename.
	292	+
	293	+ If true, and if we fail to call `CryptProtectMemory`, we
	294	+ abort; otherwise we ignore any errors from calling or
	295	+ failing to call `CryptProtectMemory`.
	296	+
	297	+ Raises:
	298	+ NotImplementedError:
	299	+ This Python version cannot call `CryptProtectMemory` due
	300	+ to lack of system support.
	301	+
	302	+ """
	303	+ realname = "Pageant"
	304	+ c_realname = ctypes.create_string_buffer(realname.encode("UTF-8"))
	305	+ num_blocks = (
	306	+ len(c_realname) + CRYPTPROTECTMEMORY_BLOCK_SIZE - 1
	307	+ ) // CRYPTPROTECTMEMORY_BLOCK_SIZE
	308	+ assert num_blocks == 1
	309	+ buf_decl = ctypes.c_char * (CRYPTPROTECTMEMORY_BLOCK_SIZE * num_blocks)
	310	+ buf = buf_decl(c_realname.raw)
	311	+ try:
	312	+ CryptProtectMemory(
	313	+ ctypes.cast(ctypes.byref(buf), ctypes.c_void_p),
	314	+ DWORD(len(buf.raw)),
	315	+ DWORD(CRYPTPROTECTMEMORY_CROSS_PROCESS),
	316	+ )
	317	+ except NotImplementedError:
	318	+ if require_cryptprotectmemory:
	319	+ raise
	320	+ buf_as_ssh_string = bytearray()
	321	+ buf_as_ssh_string.extend(int.to_bytes(len(buf.raw), 4, "big"))
	322	+ buf_as_ssh_string.extend(buf.raw)
	323	+ digest = hashlib.sha256(buf_as_ssh_string).hexdigest()
	324	+ return f"//.pipe/pageant.{os.getlogin()}.{digest}"
	325	+
	326	+
35	327	class SocketProvider:
36	328	"""Static functionality for providing sockets."""
37	329
...	...	@@ -98,15 +390,16 @@ class SocketProvider:
98	390	[windows-ssh-agent-support]: https://the13thletter.info/derivepassphrase/0.x/wishlist/windows-ssh-agent-support/
99	391
100	392	"""
	393	+ try: # pragma: no cover [external]
	394	+ import ctypes # noqa: PLC0415
	395	+ except ImportError as exc: # pragma: no cover [external]
	396	+ msg = "This Python version does not support Windows named pipes."
	397	+ raise TheAnnoyingOsNamedPipesNotAvailableError(msg) from exc
101	398	if not hasattr(ctypes, "WinDLL"):
102	399	msg = "This Python version does not support Windows named pipes."
103	400	raise TheAnnoyingOsNamedPipesNotAvailableError(msg)
104	401	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)
	402	+ return TheAnnoyingOsNamedPipeHandle.for_pageant()
110	403
111	404	registry: ClassVar[
112	405	dict[str, _types.SSHAgentSocketProvider \| str \| None]
113	406