Support for IPC for non-elevated usage in Windows

README.adoc

@@ -66,6 +66,22 @@ requires running as Administrator. This library can still be used when running
 as non-administrator, via the  `fido.client.WindowsClient` class. An example of
 this is included in the file `examples/credential.py`.
 
+To install the dependencies required for using IPC for accessing raw CTAP 
+functionality without elevation on Windows:
+
+  pip install fido2[win]
+
+To enable the IPC service elevation is required, though once started, other programs
+using the IPC may run without elevation. The following enables the service with auto
+start on boot.
+
+  python -m fido2.ipcservice.service --startup auto install
+
+To start the service once installed use the following:
+
+  sc start CTAPIPCService
+
+For stopping and deleting the service, use corresponding sc commands.
 
 Under Linux you will need to add a Udev rule to be able to access the FIDO
 device, or run as root. For example, the Udev rule may contain the following:
@@ -94,6 +110,9 @@ NFC support is optionally available via PC/SC, using the pyscard library. For
 instructions on installing this dependency, see
 https://github.com/LudovicRousseau/pyscard/blob/master/INSTALL.md.
 
+Windows IPC support is optionally available via a Windows service using the pywin32
+library. For instructions on installing this dependency, see
+https://github.com/mhammond/pywin32#installing-via-pip.
 
 === Development
 For development of the library we use https://docs.astral.sh/uv/[uv].

examples/exampleutils.py

@@ -33,7 +33,7 @@
 from getpass import getpass
 
 from fido2.client import DefaultClientDataCollector, Fido2Client, UserInteraction
-from fido2.hid import CtapHidDevice
+from fido2.hid import CtapHidDevice, ipc_available
 
 # Support NFC devices if we can
 try:
@@ -46,7 +46,7 @@
     from fido2.client.windows import WindowsClient
 
     use_winclient = (
-        WindowsClient.is_available() and not ctypes.windll.shell32.IsUserAnAdmin()
+        WindowsClient.is_available() and not (ctypes.windll.shell32.IsUserAnAdmin() or ipc_available())
     )
 except Exception:
     use_winclient = False

fido2/hid/__init__.py

@@ -31,6 +31,7 @@
 import os
 import struct
 import sys
+import ctypes
 from enum import IntEnum, IntFlag, unique
 from threading import Event
 from typing import Callable, Iterator
@@ -45,7 +46,11 @@
 if sys.platform == "linux":
     from . import linux as backend
 elif sys.platform == "win32":
-    from . import windows as backend
+    from . import ipc as ipc
+    if not ctypes.windll.shell32.IsUserAnAdmin() and ipc.IPC_Pipe.is_pipe_available():
+        from . import ipc as backend
+    else:
+        from . import windows as backend
 elif sys.platform == "darwin":
     from . import macos as backend
 # The following have version numbers at the end
@@ -63,6 +68,13 @@
 get_descriptor = backend.get_descriptor
 open_connection = backend.open_connection
 
+def ipc_available() -> bool:
+    if sys.platform != "win32":
+        return False
+    if hasattr(backend, "IPC_Pipe"):
+        return backend.IPC_Pipe.is_pipe_available()
+    return False
+
 
 class ConnectionFailure(Exception):
     """The CTAP connection failed or returned an invalid response."""

fido2/hid/ipc.py

@@ -0,0 +1,178 @@
+import win32file
+import win32pipe
+import pywintypes
+import random
+import string
+import json
+import os
+from fido2.hid.base import CtapHidConnection, HidDescriptor
+from fido2.hid.ipc_util import Commands, InputOutput, Connections, DescriptorConvertor, PIPE_NAME
+import time
+import win32security
+import ntsecuritycon as ntsec
+import win32pipe
+import win32con
+
+class IPC_Pipe:
+    def __init__(self, retry_time_ms=2000, retry_count=3):
+        for _ in range(retry_count):
+            try:
+                self.handle = win32file.CreateFile(
+                    PIPE_NAME,
+                    win32file.GENERIC_READ | win32file.GENERIC_WRITE,
+                    0,
+                    None,
+                    win32file.OPEN_EXISTING,
+                    0,
+                    None
+                )
+                break
+            except pywintypes.error as e:
+                if e.winerror == 231: #Busy handles
+                    time.sleep(retry_time_ms/1000)
+
+
+    @staticmethod
+    def is_pipe_available(timeout_ms=2000) -> bool:
+        try:
+            win32pipe.WaitNamedPipe(PIPE_NAME, timeout_ms)
+            handle = win32file.CreateFile(
+                PIPE_NAME,
+                win32file.GENERIC_READ | win32file.GENERIC_WRITE,
+                0,
+                None,
+                win32file.OPEN_EXISTING,
+                0,
+                None
+            )
+            random_data= os.urandom(16)
+            command = InputOutput(
+                command = Commands.ECHO,
+                data = random_data
+            )
+            win32file.WriteFile(handle, command.to_binary())
+            hr, response = win32file.ReadFile(handle, 65536)
+            response_data = InputOutput.from_binary(response).data
+            win32file.CloseHandle(handle)
+            assert response_data == b"echo"+random_data
+            return True
+        except (pywintypes.error, AssertionError):
+            return False
+
+    def call_pipe(self, command: InputOutput, has_response: bool = True) -> InputOutput | None:
+        data = command.to_binary()
+        win32file.WriteFile(self.handle, data)
+        if has_response:
+            hr, response = win32file.ReadFile(self.handle, 65536)
+            return InputOutput.from_binary(response)
+
+    def close_pipe(self):
+        win32file.CloseHandle(self.handle)
+
+
+class IPC_Client:
+    def __init__(self):
+        self.pipe = IPC_Pipe()
+
+    def list_descriptors(self):
+        command = InputOutput(
+            command = Commands.LIST_DESCRIPTORS
+        )
+        resp = self.pipe.call_pipe(command)
+        data = resp.data
+        assert resp.command == Commands.LIST_DESCRIPTORS
+        descriptors_list = json.loads(data.decode())
+        descriptors_list_hid = [DescriptorConvertor.str_to_descriptor(descriptor) for descriptor in descriptors_list]
+        return descriptors_list_hid
+
+    def get_descriptor(self, path: bytes):
+        command = InputOutput(
+            command = Commands.GET_DESCRIPTOR,
+            data = path
+        )
+        resp = self.pipe.call_pipe(command)
+        assert resp.command == Commands.GET_DESCRIPTOR
+        descriptor = DescriptorConvertor.str_to_descriptor(resp.data.decode())
+        return descriptor
+
+    def open_connection(self, descriptor: HidDescriptor):
+        command = InputOutput(
+            command = Commands.OPEN_CONNECTION,
+            data = DescriptorConvertor.descriptor_to_str(descriptor).encode()
+        )
+        resp = self.pipe.call_pipe(command)
+        assert resp.command == Commands.OPEN_CONNECTION
+        handle = resp.data
+        return handle
+
+    def read_packet(self, handle: bytes):
+        command = InputOutput(
+            command = Commands.READ_PACKET,
+            handle = handle
+        )
+        resp = self.pipe.call_pipe(command)
+        assert resp.command == Commands.READ_PACKET
+        data = resp.data
+        return data
+
+    def write_packet(self, handle: bytes, data: bytes):
+        command = InputOutput(
+            command = Commands.WRITE_PACKET,
+            handle = handle,
+            data = data
+        )
+
+        self.pipe.call_pipe(command, has_response = False)
+
+    def close(self, handle: bytes):
+        command = InputOutput(
+            command = Commands.CLOSE,
+            handle = handle
+        )
+        self.pipe.call_pipe(command, has_response = False)
+
+class ClientHandler:
+    def __init__(self):
+        self.ipc_client = None
+
+    def open_client(self):
+        if self.ipc_client:
+            return
+        assert IPC_Pipe.is_pipe_available()
+        self.ipc_client = IPC_Client()
+
+    def get_client(self):
+        self.open_client()
+        return self.ipc_client
+
+    def close_client(self):
+        assert self.ipc_client
+        self.ipc_client.pipe.close_pipe()
+
+client_handler = ClientHandler()
+
+def list_descriptors():
+    client = client_handler.get_client()
+    return client.list_descriptors()
+
+def get_descriptor(path):
+    client = client_handler.get_client()
+    return client.get_descriptor(path)
+
+class IPC_Connection(CtapHidConnection):
+    def __init__(self, client: IPC_Client, descriptor: HidDescriptor):
+        self.client = client
+        self.handle = self.client.open_connection(descriptor)
+
+    def read_packet(self):
+        return self.client.read_packet(self.handle)
+
+    def write_packet(self, data):
+        self.client.write_packet(self.handle, data)
+
+    def close(self):
+        self.client.close(self.handle)
+
+def open_connection(descriptor):
+    client = client_handler.get_client()
+    return IPC_Connection(client, descriptor)

fido2/hid/ipc_util.py

@@ -0,0 +1,85 @@
+import struct
+import json
+import os
+from enum import Enum
+from dataclasses import dataclass, asdict, field
+from fido2.hid.base import FIDO_USAGE, FIDO_USAGE_PAGE, CtapHidConnection, HidDescriptor
+
+HANDLE_SIZE = 16
+PIPE_NAME = r"\\.\pipe\CTAPPipe"
+
+class Commands(int, Enum):
+    LIST_DESCRIPTORS = 0x00
+    GET_DESCRIPTOR = 0x01
+    OPEN_CONNECTION = 0x02
+    READ_PACKET = 0x03
+    WRITE_PACKET = 0x04
+    CLOSE = 0x05
+    ERROR = 0x06
+    ECHO = 0x07
+
+@dataclass
+class InputOutput:
+    command: Commands
+    handle: bytes = b""
+    data: bytes = b""
+
+    @staticmethod
+    def from_binary(bindata: bytes):
+        command = Commands(bindata[0])
+        remaining = bindata[1:]
+        handle = b''
+        if command in (Commands.READ_PACKET, Commands.WRITE_PACKET, Commands.CLOSE):
+            handle = remaining[:HANDLE_SIZE]
+            remaining = remaining[HANDLE_SIZE:]
+        data = remaining
+        return InputOutput(
+            command = command,
+            handle = handle,
+            data = data
+        )
+
+    def to_binary(self):
+        if self.command in (Commands.READ_PACKET, Commands.WRITE_PACKET, Commands.CLOSE):
+            return struct.pack(
+                f"B{HANDLE_SIZE}s{len(self.data)}s",
+                self.command,
+                self.handle,
+                self.data,
+            )
+        else:
+            return struct.pack(
+                f"B{len(self.data)}s",
+                self.command,
+                self.data,
+            )
+
+@dataclass
+class Connections():
+    connections: dict[bytes, CtapHidConnection] = field(default_factory=dict)
+
+    def add_connection(self, connection: CtapHidConnection) -> bytes:
+        handle = os.urandom(HANDLE_SIZE)
+        self.connections[handle] = connection
+        return handle
+
+    def remove_connection(self, handle: bytes) -> CtapHidConnection:
+        return self.connections.pop(handle)
+
+    def get_connection(self, handle: bytes) -> CtapHidConnection:
+        return self.connections.get(handle)
+
+class DescriptorConvertor:
+
+    @staticmethod
+    def descriptor_to_str(descriptor: HidDescriptor) -> str:
+        descriptor_dict = asdict(descriptor)
+        if isinstance(descriptor_dict.get("path", None), bytes):
+            descriptor_dict["path"] = descriptor_dict["path"].decode()
+        return json.dumps(descriptor_dict)
+
+    @staticmethod
+    def str_to_descriptor(descriptor_str: str) -> HidDescriptor:
+        descriptor_dict = json.loads(descriptor_str)
+        descriptor_dict["path"] = descriptor_dict["path"].encode()
+        return HidDescriptor(**descriptor_dict)