"""Explicit opt-in real Windows backend. This module is imported only when ``driver=windows`` has passed configuration gates. It does not enumerate, capture or control anything during import or construction. Every effect starts at ``execute`` after server-side mTLS, certificate-role, bearer, target and scoped-grant validation. """ from __future__ import annotations import base64 import ctypes from ctypes import wintypes import importlib.util import os from pathlib import Path import tempfile import threading import time from typing import Any, Mapping from uuid import uuid4 from .drivers import DriverResult class WindowsDriver: kind = "windows" def __init__(self, *, artifact_dir: Path, max_file_read_bytes: int = 256 * 1024) -> None: if os.name != "nt": raise RuntimeError("real desktop driver is available on Windows only") self.artifact_dir = Path(artifact_dir).expanduser().resolve() self.max_file_read_bytes = max(4096, min(max_file_read_bytes, 4 * 1024 * 1024)) self._lock = threading.RLock() self._stopped = False self._cameras: dict[str, tuple[str, Any]] = {} def execute(self, action: Mapping[str, Any]) -> DriverResult: action_id = str(action.get("id") or uuid4()) action_type = str(action.get("type") or "") arguments = action.get("arguments") if not isinstance(arguments, Mapping): arguments = {} with self._lock: if self._stopped: return DriverResult(action_id, False, False, "stopped", "kill switch is active") try: if action_type == "desktop.screenshot": output = self._screenshot(arguments) return DriverResult(action_id, True, True, "completed", output=output) if action_type == "desktop.control": output = self._desktop_control(arguments) return DriverResult(action_id, True, True, "completed", output=output) if action_type == "files.read": output = self._file_read(arguments) return DriverResult(action_id, True, True, "completed", output=output) if action_type == "files.write": output = self._file_write(arguments) return DriverResult(action_id, True, True, "completed", output=output) if action_type == "webcam.start": output = self._camera_start(action_id, arguments) return DriverResult(action_id, True, True, "running", output=output) if action_type == "webcam.frame": output = self._camera_frame(arguments) return DriverResult(action_id, True, True, "completed", output=output) if action_type == "webcam.stop": return self.stop(str(arguments.get("capture_action_id") or action_id)) return DriverResult(action_id, False, False, "unsupported", "unsupported action type") except _DriverError as exc: return DriverResult(action_id, False, False, exc.code, exc.detail) except Exception as exc: return DriverResult( action_id, False, False, "driver_error", f"{action_type or 'action'} failed: {type(exc).__name__}", ) def stop(self, action_id: str) -> DriverResult: with self._lock: entry = self._cameras.pop(action_id, None) if entry is not None: _device_id, capture = entry try: capture.release() except Exception: pass return DriverResult(action_id, True, True, "stopped", "camera released") return DriverResult(action_id, True, False, "stopped", "no active long-running resource") def emergency_stop(self) -> None: with self._lock: self._stopped = True cameras = list(self._cameras.values()) self._cameras.clear() for _device_id, capture in cameras: try: capture.release() except Exception: pass def runtime_capabilities(self) -> Mapping[str, bool]: return { "screen": importlib.util.find_spec("PIL.ImageGrab") is not None, "computer": True, "files": True, "webcam": importlib.util.find_spec("cv2") is not None, } def _screenshot(self, arguments: Mapping[str, Any]) -> dict[str, object]: self._ensure_running() try: from PIL import ImageGrab except ImportError as exc: raise _DriverError("optional_dependency_missing", "desktop screenshot requires Pillow") from exc image = ImageGrab.grab(all_screens=bool(arguments.get("all_screens", False))) path, artifact_id = self._artifact("screen", ".png") image.save(path, format="PNG") return { "artifact_id": artifact_id, "local_path": str(path), "mime_type": "image/png", "width": int(image.width), "height": int(image.height), "raw_frame": None, } def _file_read(self, arguments: Mapping[str, Any]) -> dict[str, object]: self._ensure_running() path = Path(str(arguments["path"])).expanduser().resolve(strict=True) if not path.is_file(): raise _DriverError("not_a_file", "granted path is not a regular file") offset = _bounded_int(arguments.get("offset", 0), 0, 2**63 - 1, "offset") requested = _bounded_int( arguments.get("max_bytes", self.max_file_read_bytes), 1, self.max_file_read_bytes, "max_bytes", ) with path.open("rb") as handle: handle.seek(offset) chunk = handle.read(requested) following = handle.read(1) try: content = chunk.decode(str(arguments.get("encoding") or "utf-8")) encoding = str(arguments.get("encoding") or "utf-8") except (UnicodeDecodeError, LookupError): content = base64.b64encode(chunk).decode("ascii") encoding = "base64" return { "content": content, "encoding": encoding, "offset": offset, "bytes_read": len(chunk), "next_offset": offset + len(chunk), "eof": not following, } def _file_write(self, arguments: Mapping[str, Any]) -> dict[str, object]: self._ensure_running() path = Path(str(arguments["path"])).expanduser().resolve(strict=False) if arguments.get("content_base64") is not None: try: content = base64.b64decode(str(arguments["content_base64"]), validate=True) except ValueError as exc: raise _DriverError("invalid_base64", "content_base64 is invalid") from exc else: encoding = str(arguments.get("encoding") or "utf-8") try: content = str(arguments.get("content") or "").encode(encoding) except LookupError as exc: raise _DriverError("invalid_encoding", "requested encoding is unknown") from exc if len(content) > self.max_file_read_bytes: raise _DriverError("write_too_large", "single file write exceeds configured limit") if bool(arguments.get("create_parents", False)): path.parent.mkdir(parents=True, exist_ok=True) if not path.parent.is_dir(): raise _DriverError("parent_missing", "target parent directory does not exist") mode = str(arguments.get("mode") or "overwrite") if mode == "append": with path.open("ab") as handle: handle.write(content) handle.flush() os.fsync(handle.fileno()) elif mode == "overwrite": descriptor, temporary_name = tempfile.mkstemp(prefix=".omni-write-", dir=path.parent) try: with os.fdopen(descriptor, "wb") as handle: handle.write(content) handle.flush() os.fsync(handle.fileno()) os.replace(temporary_name, path) except Exception: try: os.unlink(temporary_name) except OSError: pass raise else: raise _DriverError("invalid_write_mode", "write mode must be overwrite or append") return {"bytes_written": len(content), "mode": mode, "path": str(path)} def _camera_start(self, action_id: str, arguments: Mapping[str, Any]) -> dict[str, object]: self._ensure_running() try: import cv2 except ImportError as exc: raise _DriverError("optional_dependency_missing", "webcam capture requires opencv-python") from exc device_id = str(arguments.get("device_id") or "") try: device_index = int(device_id) except ValueError as exc: raise _DriverError("invalid_camera_device", "Windows webcam device_id must be a numeric index") from exc backend = getattr(cv2, "CAP_DSHOW", 0) capture = cv2.VideoCapture(device_index, backend) if arguments.get("width") is not None: capture.set(cv2.CAP_PROP_FRAME_WIDTH, _bounded_int(arguments["width"], 160, 7680, "width")) if arguments.get("height") is not None: capture.set(cv2.CAP_PROP_FRAME_HEIGHT, _bounded_int(arguments["height"], 120, 4320, "height")) if not capture.isOpened(): capture.release() raise _DriverError("camera_open_failed", "granted webcam could not be opened") with self._lock: if self._stopped: capture.release() raise _DriverError("stopped", "kill switch is active") self._cameras[action_id] = (device_id, capture) return {"capture_action_id": action_id, "device_id": device_id, "frame_captured": False} def _camera_frame(self, arguments: Mapping[str, Any]) -> dict[str, object]: self._ensure_running() device_id = str(arguments.get("device_id") or "") with self._lock: match = next( ((action_id, capture) for action_id, (current, capture) in self._cameras.items() if current == device_id), None, ) if match is None: raise _DriverError("camera_not_active", "webcam must be started before requesting a frame") capture_action_id, capture = match ok, frame = capture.read() if not ok or frame is None: raise _DriverError("camera_frame_failed", "webcam did not return a frame") try: import cv2 except ImportError as exc: raise _DriverError("optional_dependency_missing", "webcam capture requires opencv-python") from exc path, artifact_id = self._artifact("camera", ".jpg") if not cv2.imwrite(str(path), frame, [int(cv2.IMWRITE_JPEG_QUALITY), 92]): raise _DriverError("artifact_write_failed", "camera artifact could not be written") height, width = frame.shape[:2] return { "artifact_id": artifact_id, "local_path": str(path), "mime_type": "image/jpeg", "width": int(width), "height": int(height), "capture_action_id": capture_action_id, "raw_frame": None, } def _desktop_control(self, arguments: Mapping[str, Any]) -> dict[str, object]: self._ensure_running() action = str(arguments.get("action") or arguments.get("operation") or "") if action in {"move", "click", "double_click", "right_click", "drag"}: if arguments.get("x") is not None and arguments.get("y") is not None: self._set_cursor(arguments["x"], arguments["y"]) if action == "move": return {"action": action} if action in {"click", "double_click", "right_click"}: button = "right" if action == "right_click" else str(arguments.get("button") or "left") count = 2 if action == "double_click" else 1 for _ in range(count): self._mouse_click(button) return {"action": action, "button": button, "count": count} if action == "scroll": delta = _bounded_int(arguments.get("delta", 0), -12000, 12000, "delta") self._send_mouse(0x0800, mouse_data=delta) return {"action": action, "delta": delta} if action == "drag": end_x = arguments.get("to_x") end_y = arguments.get("to_y") if end_x is None or end_y is None: raise _DriverError("drag_target_required", "drag requires to_x and to_y") button = str(arguments.get("button") or "left") down, up = self._button_flags(button) self._send_mouse(down) try: self._set_cursor(end_x, end_y) finally: self._send_mouse(up) return {"action": action, "button": button} if action == "type": text = str(arguments.get("text") or "") if len(text) > 8192: raise _DriverError("text_too_long", "single keyboard action exceeds 8192 characters") interval = min(0.05, max(0.0, float(arguments.get("interval_seconds", 0.0)))) self._type_unicode(text, interval) return {"action": action, "characters": len(text)} if action == "hotkey": keys = arguments.get("keys") if isinstance(keys, str): keys = [part for part in keys.lower().split("+") if part] if not isinstance(keys, list) or not keys or len(keys) > 8: raise _DriverError("invalid_hotkey", "hotkey requires one to eight key names") self._hotkey([str(key).lower() for key in keys]) return {"action": action, "key_count": len(keys)} if action == "clipboard.read": text = self._clipboard_read() return {"action": action, "text": text, "characters": len(text)} if action == "clipboard.write": text = str(arguments.get("text") or "") if len(text) > 1_000_000: raise _DriverError("text_too_long", "clipboard text exceeds one million characters") self._clipboard_write(text) return {"action": action, "characters": len(text)} raise _DriverError("unsupported_computer_action", "computer action is not implemented") def _artifact(self, prefix: str, suffix: str) -> tuple[Path, str]: self.artifact_dir.mkdir(parents=True, exist_ok=True) artifact_id = f"artifact_{uuid4().hex}" return self.artifact_dir / f"{prefix}_{artifact_id}{suffix}", artifact_id def _ensure_running(self) -> None: with self._lock: if self._stopped: raise _DriverError("stopped", "kill switch is active") @staticmethod def _set_cursor(x_value: object, y_value: object) -> None: x = _bounded_int(x_value, -100_000, 100_000, "x") y = _bounded_int(y_value, -100_000, 100_000, "y") if not ctypes.windll.user32.SetCursorPos(x, y): raise _DriverError("cursor_move_failed", "Windows rejected cursor movement") @staticmethod def _button_flags(button: str) -> tuple[int, int]: if button == "left": return 0x0002, 0x0004 if button == "right": return 0x0008, 0x0010 if button == "middle": return 0x0020, 0x0040 raise _DriverError("invalid_mouse_button", "mouse button must be left, right or middle") def _mouse_click(self, button: str) -> None: down, up = self._button_flags(button) self._send_mouse(down) self._send_mouse(up) @staticmethod def _send_mouse(flags: int, *, mouse_data: int = 0) -> None: input_value = _INPUT() input_value.type = 0 # MOUSEINPUT.mouseData is a DWORD even though wheel deltas are signed. input_value.union.mi = _MOUSEINPUT(0, 0, mouse_data & 0xFFFFFFFF, flags, 0, 0) if ctypes.windll.user32.SendInput(1, ctypes.byref(input_value), ctypes.sizeof(_INPUT)) != 1: raise _DriverError("send_input_failed", "Windows rejected mouse input") def _type_unicode(self, text: str, interval: float) -> None: for unit_bytes in _utf16_units(text): self._ensure_running() unit = int.from_bytes(unit_bytes, "little") self._send_keyboard(unit, 0x0004) self._send_keyboard(unit, 0x0004 | 0x0002) if interval: time.sleep(interval) def _hotkey(self, keys: list[str]) -> None: virtual_keys = [_virtual_key(key) for key in keys] pressed: list[int] = [] try: for key in virtual_keys: self._ensure_running() self._send_keyboard(key, 0) pressed.append(key) finally: for key in reversed(pressed): self._send_keyboard(key, 0x0002) @staticmethod def _clipboard_read() -> str: user32 = ctypes.windll.user32 kernel32 = ctypes.windll.kernel32 user32.OpenClipboard.argtypes = [wintypes.HWND] user32.OpenClipboard.restype = wintypes.BOOL user32.GetClipboardData.argtypes = [wintypes.UINT] user32.GetClipboardData.restype = wintypes.HANDLE kernel32.GlobalLock.argtypes = [wintypes.HGLOBAL] kernel32.GlobalLock.restype = wintypes.LPVOID kernel32.GlobalUnlock.argtypes = [wintypes.HGLOBAL] kernel32.GlobalUnlock.restype = wintypes.BOOL if not _open_clipboard(user32): raise _DriverError("clipboard_busy", "Windows clipboard is unavailable") try: handle = user32.GetClipboardData(13) # CF_UNICODETEXT if not handle: return "" pointer = kernel32.GlobalLock(handle) if not pointer: raise _DriverError("clipboard_read_failed", "clipboard text could not be locked") try: return ctypes.wstring_at(pointer) finally: kernel32.GlobalUnlock(handle) finally: user32.CloseClipboard() @staticmethod def _clipboard_write(text: str) -> None: user32 = ctypes.windll.user32 kernel32 = ctypes.windll.kernel32 user32.OpenClipboard.argtypes = [wintypes.HWND] user32.OpenClipboard.restype = wintypes.BOOL user32.SetClipboardData.argtypes = [wintypes.UINT, wintypes.HANDLE] user32.SetClipboardData.restype = wintypes.HANDLE kernel32.GlobalAlloc.argtypes = [wintypes.UINT, ctypes.c_size_t] kernel32.GlobalAlloc.restype = wintypes.HGLOBAL kernel32.GlobalLock.argtypes = [wintypes.HGLOBAL] kernel32.GlobalLock.restype = wintypes.LPVOID kernel32.GlobalUnlock.argtypes = [wintypes.HGLOBAL] kernel32.GlobalUnlock.restype = wintypes.BOOL kernel32.GlobalFree.argtypes = [wintypes.HGLOBAL] kernel32.GlobalFree.restype = wintypes.HGLOBAL if not _open_clipboard(user32): raise _DriverError("clipboard_busy", "Windows clipboard is unavailable") handle = None transferred = False try: if not user32.EmptyClipboard(): raise _DriverError("clipboard_write_failed", "Windows clipboard could not be emptied") encoded = (text + "\0").encode("utf-16-le") handle = kernel32.GlobalAlloc(0x0002, len(encoded)) # GMEM_MOVEABLE if not handle: raise _DriverError("clipboard_write_failed", "clipboard allocation failed") pointer = kernel32.GlobalLock(handle) if not pointer: raise _DriverError("clipboard_write_failed", "clipboard allocation could not be locked") try: ctypes.memmove(pointer, encoded, len(encoded)) finally: kernel32.GlobalUnlock(handle) if not user32.SetClipboardData(13, handle): raise _DriverError("clipboard_write_failed", "Windows rejected clipboard text") transferred = True finally: # Successful SetClipboardData transfers ownership to Windows. if handle and not transferred: kernel32.GlobalFree(handle) user32.CloseClipboard() @staticmethod def _send_keyboard(code: int, flags: int) -> None: input_value = _keyboard_input(code, flags) if ctypes.windll.user32.SendInput(1, ctypes.byref(input_value), ctypes.sizeof(_INPUT)) != 1: raise _DriverError("send_input_failed", "Windows rejected keyboard input") class _DriverError(RuntimeError): def __init__(self, code: str, detail: str) -> None: super().__init__(detail) self.code = code self.detail = detail def _bounded_int(value: object, minimum: int, maximum: int, name: str) -> int: if isinstance(value, bool): raise _DriverError("invalid_argument", f"{name} must be an integer") try: result = int(value) except (TypeError, ValueError) as exc: raise _DriverError("invalid_argument", f"{name} must be an integer") from exc if not minimum <= result <= maximum: raise _DriverError("invalid_argument", f"{name} is outside the supported range") return result def _open_clipboard(user32: Any) -> bool: for _attempt in range(5): if user32.OpenClipboard(None): return True time.sleep(0.02) return False def _utf16_units(text: str): encoded = text.encode("utf-16-le") for index in range(0, len(encoded), 2): yield encoded[index : index + 2] def _virtual_key(name: str) -> int: named = { "ctrl": 0x11, "control": 0x11, "shift": 0x10, "alt": 0x12, "win": 0x5B, "enter": 0x0D, "tab": 0x09, "escape": 0x1B, "esc": 0x1B, "space": 0x20, "backspace": 0x08, "delete": 0x2E, "up": 0x26, "down": 0x28, "left": 0x25, "right": 0x27, "home": 0x24, "end": 0x23, "pageup": 0x21, "pagedown": 0x22, } if name in named: return named[name] if len(name) == 1 and name.isascii() and name.isalnum(): value = ctypes.windll.user32.VkKeyScanW(ord(name.upper())) if value == -1: raise _DriverError("invalid_hotkey", f"unsupported key: {name}") return value & 0xFF if name.startswith("f") and name[1:].isdigit() and 1 <= int(name[1:]) <= 24: return 0x70 + int(name[1:]) - 1 raise _DriverError("invalid_hotkey", f"unsupported key: {name}") _ULONG_PTR = wintypes.WPARAM class _MOUSEINPUT(ctypes.Structure): _fields_ = [ ("dx", wintypes.LONG), ("dy", wintypes.LONG), ("mouseData", wintypes.DWORD), ("dwFlags", wintypes.DWORD), ("time", wintypes.DWORD), ("dwExtraInfo", _ULONG_PTR), ] class _KEYBDINPUT(ctypes.Structure): _fields_ = [ ("wVk", wintypes.WORD), ("wScan", wintypes.WORD), ("dwFlags", wintypes.DWORD), ("time", wintypes.DWORD), ("dwExtraInfo", _ULONG_PTR), ] class _INPUTUNION(ctypes.Union): _fields_ = [("mi", _MOUSEINPUT), ("ki", _KEYBDINPUT)] class _INPUT(ctypes.Structure): _anonymous_ = ("union",) _fields_ = [("type", wintypes.DWORD), ("union", _INPUTUNION)] def _keyboard_input(code: int, flags: int) -> _INPUT: """Build a keyboard INPUT without dispatching it (also used by offline tests).""" unicode_input = bool(flags & 0x0004) input_value = _INPUT() input_value.type = 1 input_value.union.ki = _KEYBDINPUT( 0 if unicode_input else code, code if unicode_input else 0, flags, 0, 0, ) return input_value