diff --git a/src/windows_mcp/__main__.py b/src/windows_mcp/__main__.py index 061d59a3..7b02fba7 100755 --- a/src/windows_mcp/__main__.py +++ b/src/windows_mcp/__main__.py @@ -67,7 +67,9 @@ def _http_middleware( if ip_allowlist: middleware.append(Middleware(IPAllowlistMiddleware, allowlist=ip_allowlist)) if auth_key: - middleware.append(Middleware(AuthKeyMiddleware, auth_key=auth_key, oauth_validator=oauth_validator)) + middleware.append( + Middleware(AuthKeyMiddleware, auth_key=auth_key, oauth_validator=oauth_validator) + ) elif oauth_validator: middleware.append(Middleware(OAuthOnlyMiddleware, oauth_validator=oauth_validator)) return middleware @@ -169,8 +171,6 @@ def __getattr__(name: str): raise AttributeError(f"module {__name__!r} has no attribute {name!r}") - - class Transport(Enum): STDIO = "stdio" SSE = "sse" @@ -180,7 +180,9 @@ def __str__(self): return self.value -def _apply_tool_filter(mcp, explicit_tools: list[str] | None, exclude_tools: list[str] | None) -> None: +def _apply_tool_filter( + mcp, explicit_tools: list[str] | None, exclude_tools: list[str] | None +) -> None: """Remove disabled tools from the MCP registry.""" tool_mgr = getattr(mcp, "_tool_manager", None) tools_dict = getattr(tool_mgr, "_tools", None) @@ -192,14 +194,27 @@ def _apply_tool_filter(mcp, explicit_tools: list[str] | None, exclude_tools: lis for k, v in components.items() if isinstance(k, str) and k.startswith("tool:") } + def _remove(name): - keys = [k for k, v in components.items() if isinstance(k, str) and k.startswith("tool:") and (getattr(components[k], "name", None) == name or k.split(":", 1)[1].split("@", 1)[0] == name)] + keys = [ + k + for k, v in components.items() + if isinstance(k, str) + and k.startswith("tool:") + and ( + getattr(components[k], "name", None) == name + or k.split(":", 1)[1].split("@", 1)[0] == name + ) + ] for k in keys: components.pop(k, None) + registered = set(tools_dict.keys()) else: + def _remove(name): tools_dict.pop(name, None) + registered = set(tools_dict.keys()) if explicit_tools: @@ -366,7 +381,23 @@ def main(): type=str, show_default=False, ) -def serve(ctx, transport, host, port, debug, config, auth_key, allow_insecure_remote, ip_allowlist, tools, exclude_tools, ssl_certfile, ssl_keyfile, oauth_client_id, oauth_client_secret): +def serve( + ctx, + transport, + host, + port, + debug, + config, + auth_key, + allow_insecure_remote, + ip_allowlist, + tools, + exclude_tools, + ssl_certfile, + ssl_keyfile, + oauth_client_id, + oauth_client_secret, +): asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) if transport == Transport.STDIO.value: os.environ.setdefault("NO_COLOR", "1") @@ -388,24 +419,42 @@ def serve(ctx, transport, host, port, debug, config, auth_key, allow_insecure_re port = int(_choose_value(ctx, "port", port, cfg.server.port, 8000)) auth_key = _choose_value(ctx, "auth_key", auth_key, cfg.server.auth_key, None) allow_insecure_remote = bool( - _choose_value(ctx, "allow_insecure_remote", allow_insecure_remote, cfg.server.allow_insecure_remote, False) + _choose_value( + ctx, + "allow_insecure_remote", + allow_insecure_remote, + cfg.server.allow_insecure_remote, + False, + ) ) ssl_certfile = _choose_value(ctx, "ssl_certfile", ssl_certfile, cfg.server.ssl_certfile, None) ssl_keyfile = _choose_value(ctx, "ssl_keyfile", ssl_keyfile, cfg.server.ssl_keyfile, None) - oauth_client_id = _choose_value(ctx, "oauth_client_id", oauth_client_id, cfg.security.oauth_client_id, None) + oauth_client_id = _choose_value( + ctx, "oauth_client_id", oauth_client_id, cfg.security.oauth_client_id, None + ) oauth_client_secret = _choose_value( ctx, "oauth_client_secret", oauth_client_secret, cfg.security.oauth_client_secret, None ) cli_tools = [t.strip() for t in tools.split(",") if t.strip()] if tools else [] - cli_exclude = [t.strip() for t in exclude_tools.split(",") if t.strip()] if _param_explicit(ctx, "exclude_tools") and exclude_tools else list(cfg.tools.exclude) - cli_allowlist = [e.strip() for e in ip_allowlist.split(",")] if ip_allowlist and _param_explicit(ctx, "ip_allowlist") else cfg.security.ip_allowlist + cli_exclude = ( + [t.strip() for t in exclude_tools.split(",") if t.strip()] + if _param_explicit(ctx, "exclude_tools") and exclude_tools + else list(cfg.tools.exclude) + ) + cli_allowlist = ( + [e.strip() for e in ip_allowlist.split(",")] + if ip_allowlist and _param_explicit(ctx, "ip_allowlist") + else cfg.security.ip_allowlist + ) if bool(ssl_certfile) != bool(ssl_keyfile): raise click.ClickException("--ssl-certfile and --ssl-keyfile must be provided together.") if bool(oauth_client_id) != bool(oauth_client_secret): - raise click.ClickException("OAuth requires both --oauth-client-id and --oauth-client-secret.") + raise click.ClickException( + "OAuth requires both --oauth-client-id and --oauth-client-secret." + ) parsed_allowlist = None if cli_allowlist: @@ -489,7 +538,7 @@ def _gen_tls(host: str, cert_path, key_path) -> None: mkcert = subprocess.run(["where", "mkcert"], capture_output=True).returncode == 0 if mkcert: - click.echo("mkcert detected — generating a locally-trusted certificate...") + click.echo("mkcert detected -- generating a locally-trusted certificate...") install = subprocess.run(["mkcert", "-install"], capture_output=True, text=True) if install.returncode != 0: raise click.ClickException(f"mkcert -install failed:\n{install.stderr.strip()}") @@ -498,35 +547,50 @@ def _gen_tls(host: str, cert_path, key_path) -> None: result = subprocess.run( [ "mkcert", - "-cert-file", str(cert_path), - "-key-file", str(key_path), + "-cert-file", + str(cert_path), + "-key-file", + str(key_path), *sans, ], - capture_output=True, text=True, + capture_output=True, + text=True, ) if result.returncode != 0: raise click.ClickException(f"mkcert failed:\n{result.stderr.strip()}") click.echo(" Certificate is automatically trusted by Windows.") else: - click.echo("mkcert not found — falling back to openssl (self-signed)...") + click.echo("mkcert not found -- falling back to openssl (self-signed)...") click.echo(" Tip: winget install FiloSottile.mkcert for auto-trusted certs next time.") result = subprocess.run( [ - "openssl", "req", "-x509", "-newkey", "rsa:4096", - "-keyout", str(key_path), - "-out", str(cert_path), - "-days", "365", "-nodes", - "-subj", f"/CN={host or 'windows-mcp'}", + "openssl", + "req", + "-x509", + "-newkey", + "rsa:4096", + "-keyout", + str(key_path), + "-out", + str(cert_path), + "-days", + "365", + "-nodes", + "-subj", + f"/CN={host or 'windows-mcp'}", ], - capture_output=True, text=True, + capture_output=True, + text=True, ) if result.returncode != 0: raise click.ClickException(f"openssl failed:\n{result.stderr.strip()}") click.echo(" To make Windows trust this cert, run in an elevated PowerShell:") - click.echo(f' Import-Certificate -FilePath "{cert_path}" -CertStoreLocation Cert:\\LocalMachine\\Root') + click.echo( + f' Import-Certificate -FilePath "{cert_path}" -CertStoreLocation Cert:\\LocalMachine\\Root' + ) - click.echo(f" cert → {cert_path}") - click.echo(f" key → {key_path}") + click.echo(f" cert -> {cert_path}") + click.echo(f" key -> {key_path}") _TASK_NAME = "windows-mcp-server" @@ -553,11 +617,7 @@ def _build_start_script(program_args: list[str]) -> str: log_out = CONFIG_DIR / "server.log" log_err = CONFIG_DIR / "server.error.log" command = subprocess.list2cmdline(program_args) - return ( - "@echo off\n" - "setlocal\n" - f"{command} 1>>\"{log_out}\" 2>>\"{log_err}\"\n" - ) + return f'@echo off\nsetlocal\n{command} 1>>"{log_out}" 2>>"{log_err}"\n' def _schtasks(*args: str) -> subprocess.CompletedProcess: @@ -604,13 +664,17 @@ def install(transport: str, host: str, port: int, force: bool) -> None: "/F", ) if result.returncode != 0: - raise click.ClickException(f"schtasks /Create failed:\n{result.stderr.strip() or result.stdout.strip()}") + raise click.ClickException( + f"schtasks /Create failed:\n{result.stderr.strip() or result.stdout.strip()}" + ) run_result = _schtasks("/Run", "/TN", _TASK_NAME) if run_result.returncode != 0: - raise click.ClickException(f"schtasks /Run failed:\n{run_result.stderr.strip() or run_result.stdout.strip()}") + raise click.ClickException( + f"schtasks /Run failed:\n{run_result.stderr.strip() or run_result.stdout.strip()}" + ) - click.echo("Scheduled task installed — server is starting now.") + click.echo("Scheduled task installed -- server is starting now.") click.echo(f" Task : {_TASK_NAME}") click.echo(f" Transport : {transport}") click.echo(f" Address : {host}:{port}") @@ -746,7 +810,7 @@ def auth(transport: str, host: str, port: int, with_tls: bool, force: bool) -> N # --------------------------------------------------------------------------- -# `windows-mcp service` command group +# `windows-mcp service secure-desktop` command group # --------------------------------------------------------------------------- _SERVICE_NAME = "WindowsMCPHost" @@ -758,33 +822,105 @@ def _require_win32(): import win32serviceutil # noqa: F401 except ImportError: raise click.ClickException( - "pywin32 is required for service management. " - "Install it with: pip install pywin32" + "pywin32 is required for service management. Install it with: pip install pywin32" ) +def _admin_only_prefixes() -> list[str]: + """Paths under which Windows defaults to admin-only write access.""" + return [ + os.environ.get("ProgramFiles", r"C:\Program Files"), + os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)"), + os.environ.get("SystemRoot", r"C:\Windows"), + ] + + +def _path_is_admin_only(path: str) -> bool: + """Return True if *path* lives under a default admin-only prefix. + + This is a *heuristic*, not a permission check -- but it covers 99% of + real installs. Users on truly custom layouts can override with + --allow-user-binary-path. + """ + norm = os.path.normcase(os.path.normpath(path)) + for prefix in _admin_only_prefixes(): + if not prefix: + continue + prefix_norm = os.path.normcase(os.path.normpath(prefix)) + if norm.startswith(prefix_norm + os.sep) or norm == prefix_norm: + return True + return False + + _UAC_POLICY_KEY = r"SOFTWARE\Microsoft\Windows\CurrentVersion\Policies\System" -_UAC_POLICY_VALUE = "PromptOnSecureDesktop" -def _set_prompt_on_secure_desktop(enabled: bool) -> None: - """Set or clear the 'Switch to secure desktop' UAC policy. +def _set_uac_secure_desktop_off(off: bool) -> tuple[int, int]: + """Toggle the UAC secure-desktop policy. Returns the (POSD, CPB) readback + so callers can verify the writes stuck. + + off=True -> POSD=0, CPB=4 (UAC on Default desktop). + off=False -> POSD=1, CPB=5 (modern Win 11 default). - When disabled (enabled=False) UAC prompts appear on the normal Default - desktop where user-mode UIA and SendInput can reach them, instead of the - isolated Winlogon desktop. Requires elevation. + CPB=4 specifically (not 5) is needed on Win 11 25H2 -- see + docs/win11-uac-investigation.md iter-7 vs iter-8. """ import winreg + + posd_target = 0 if off else 1 + cpba_target = 4 if off else 5 + with winreg.OpenKey( winreg.HKEY_LOCAL_MACHINE, _UAC_POLICY_KEY, - access=winreg.KEY_SET_VALUE, + access=winreg.KEY_SET_VALUE | winreg.KEY_QUERY_VALUE, ) as key: - winreg.SetValueEx(key, _UAC_POLICY_VALUE, 0, winreg.REG_DWORD, 1 if enabled else 0) + winreg.SetValueEx(key, "PromptOnSecureDesktop", 0, winreg.REG_DWORD, posd_target) + winreg.SetValueEx(key, "ConsentPromptBehaviorAdmin", 0, winreg.REG_DWORD, cpba_target) + posd, _ = winreg.QueryValueEx(key, "PromptOnSecureDesktop") + cpba, _ = winreg.QueryValueEx(key, "ConsentPromptBehaviorAdmin") + return int(posd), int(cpba) + + +def _verify_install_paths_are_admin_only() -> None: + """Raise ClickException if the Python interpreter or windows_mcp package live + in a user-writable location. + + The Windows SCM will launch the binary path as SYSTEM. If any component + of that path is under user-writable storage (a uv tool cache, a venv in + %LOCALAPPDATA%, a per-user pip install), then any process running as the + user can replace files there and gain SYSTEM the next time the service + starts. Refuse the install rather than register an unsafe service. + """ + import windows_mcp + + py_exe = sys.executable + pkg_path = os.path.dirname(os.path.abspath(windows_mcp.__file__)) + + unsafe: list[str] = [] + if not _path_is_admin_only(py_exe): + unsafe.append(f" Python interpreter : {py_exe}") + if not _path_is_admin_only(pkg_path): + unsafe.append(f" windows_mcp package: {pkg_path}") + + if not unsafe: + return + + raise click.ClickException( + "Refusing to install the LocalSystem service: the binary path lives in a\n" + "user-writable location. Anyone who can write to that path will obtain\n" + "SYSTEM the next time the service starts.\n\n" + "\n".join(unsafe) + "\n\n" + "Install Python system-wide (e.g. `winget install Python.Python.3.13`,\n" + "which lands under %ProgramFiles%) and then `pip install windows-mcp`\n" + "into that system Python. Re-run this command.\n\n" + "If you accept the risk (e.g. testing inside a disposable VM), pass\n" + "--allow-user-binary-path." + ) def _sc_state_name(state: int) -> str: import win32service + return { win32service.SERVICE_STOPPED: "STOPPED", win32service.SERVICE_START_PENDING: "START_PENDING", @@ -798,30 +934,102 @@ def _sc_state_name(state: int) -> str: @main.group() def service(): - """Manage the Windows MCP host service (LocalSystem, required for UAC access). + """Manage Windows MCP optional privileged services. + + Privileged services run as NT AUTHORITY\\SYSTEM and expose a local named + pipe to the user-mode broker. They are opt-in because they require + elevation to install. + + Sub-groups: + + secure-desktop Host service that lets the agent see and click UAC + consent prompts (Secure Desktop / Winlogon). + """ + + +@service.group("secure-desktop") +def service_secure_desktop(): + """Manage the Secure Desktop host service (handles UAC consent prompts). The host service runs as NT AUTHORITY\\SYSTEM and exposes a local named pipe - so the MCP broker can capture screenshots even while a UAC prompt is on screen. + so the MCP broker can capture screenshots and route input across the + Winlogon (Secure Desktop) boundary that fires during UAC consent prompts. + + UAC remains fully enabled -- the service does NOT weaken the Secure Desktop + policy. Whether the broker may auto-click a UAC prompt is governed by the + ``WINDOWS_MCP_SECURE_DESKTOP_POLICY`` env var (``block`` by default). - Installing the service also disables the "Switch to secure desktop" UAC policy - (PromptOnSecureDesktop=0) so that UAC prompts appear on the normal Default - desktop. This allows user-mode UIA and SendInput to reach the Yes/No buttons - directly, without needing cross-session tricks. The policy is restored when - the service is uninstalled. + Must be installed once from an elevated (Administrator) prompt: - Must be installed once from an elevated (Administrator) prompt. + uv run windows-mcp service secure-desktop install """ -@service.command("install") +@service_secure_desktop.command("install") @click.option("--force", is_flag=True, help="Uninstall then reinstall if already present.") -def service_install(force: bool): - """Install and start the Windows MCP host service (requires elevation).""" +@click.option( + "--policy", + type=click.Choice(["block", "allow_with_match", "allow_all"]), + default=None, + help=( + "Persist a Secure Desktop consent policy on install. " + "If omitted, falls back to WINDOWS_MCP_SECURE_DESKTOP_POLICY, " + "then config.toml, then 'block'." + ), +) +@click.option( + "--allow-publisher", + "allow_publisher", + multiple=True, + help=( + "Publisher substring to allow under --policy=allow_with_match. " + "Repeat to add multiple. Comma-separated also works." + ), +) +@click.option( + "--allow-user-binary-path", + is_flag=True, + default=False, + help=( + "Allow installing even if Python or windows_mcp live in a user-writable " + "location. Unsafe outside a disposable VM -- any local process running as " + "the user can replace the binary and gain SYSTEM at next service start." + ), +) +def service_secure_desktop_install( + force: bool, + policy: str | None, + allow_publisher: tuple[str, ...], + allow_user_binary_path: bool, +): + """Install and start the Secure Desktop host service (requires elevation).""" _require_win32() import win32serviceutil import win32service import pywintypes from windows_mcp.service.host import WindowsMCPHostService + from windows_mcp.service import policy as policy_mod + + if not allow_user_binary_path: + _verify_install_paths_are_admin_only() + else: + click.echo( + "WARNING: --allow-user-binary-path was passed. The service binary " + "path may be user-writable, which is a privilege-escalation risk. " + "Use only in disposable VMs." + ) + + # Resolve effective policy: CLI flag > env var > config.toml > default ("block"). + cfg = load_config(discover_config_path(None)) + cli_allowlist: list[str] = [] + for raw in allow_publisher: + cli_allowlist.extend(s.strip() for s in raw.split(",") if s.strip()) + effective_policy = policy_mod.resolve_install_time_policy( + cli_policy=policy, + cli_allowlist=cli_allowlist or None, + config_policy=cfg.secure_desktop.policy, + config_allowlist=cfg.secure_desktop.publishers_allowlist, + ) # Check whether the service already exists. already_installed = False @@ -853,11 +1061,6 @@ def service_install(force: bool): # against the system Python and cannot import windows_mcp, causing 1053. # Using sys.executable guarantees the exact interpreter that has the # package is what the SCM launches. - # - # Binary path format: "" -m windows_mcp.service.host - # When the SCM starts this with no extra args, host.py calls - # servicemanager.StartServiceCtrlDispatcher() to enter the service loop. - from windows_mcp.service.host import WindowsMCPHostService binary_path = f'"{sys.executable}" -m windows_mcp.service.host' hscm = None @@ -873,11 +1076,11 @@ def service_install(force: bool): win32service.SERVICE_AUTO_START, win32service.SERVICE_ERROR_NORMAL, binary_path, - None, # load order group - 0, # tag id - None, # dependencies - None, # service account → LocalSystem - None, # password + None, # load order group + 0, # tag id + None, # dependencies + None, # service account -> LocalSystem + None, # password ) win32service.ChangeServiceConfig2( hs, @@ -905,33 +1108,60 @@ def service_install(force: bool): else: raise click.ClickException(f"Failed to start service: {exc}") - # Disable "Switch to secure desktop" so UAC prompts appear on the normal - # Default desktop, where user-mode UIA and SendInput can reach them. try: - _set_prompt_on_secure_desktop(False) - click.echo("UAC policy : PromptOnSecureDesktop disabled (UAC on Default desktop).") + policy_mod.write_to_registry(effective_policy) + click.echo(f"UAC consent policy : {effective_policy.policy}") + if effective_policy.publishers_allowlist: + click.echo(f" publishers allowlist: {effective_policy.publishers_allowlist}") + except Exception as exc: + click.echo(f"Warning: could not persist UAC policy: {exc}") + click.echo(" Service will refuse auto-clicks until policy is set.") + + # Route UAC to the user's Default desktop. Without this every UAC + # prompt lands on Winlogon where consent.exe is unreachable from + # user-mode UIA (see docs/win11-uac-investigation.md). Restored on + # uninstall. + try: + posd, cpba = _set_uac_secure_desktop_off(True) + if posd == 0 and cpba == 4: + click.echo( + "UAC policy : PromptOnSecureDesktop=0, ConsentPromptBehaviorAdmin=4 " + "(UAC will render on Default desktop)." + ) + else: + click.echo( + f"Warning: UAC policy writes did not stick: readback PromptOnSecureDesktop={posd}, " + f"ConsentPromptBehaviorAdmin={cpba}. Group Policy or another layer is " + "overriding the registry values." + ) except Exception as exc: - click.echo(f"Warning: could not update UAC policy: {exc}") - click.echo(" UAC dialogs may not be accessible to the agent.") + click.echo(f"Warning: could not disable secure-desktop UAC policy: {exc}") + click.echo( + " UAC will continue to render on the Secure Desktop, where the " + "dialog is unreachable from user-mode UIA on Win11. WaitForUACPrompt " + "will return an empty tree." + ) click.echo("\nThe host service is now running as NT AUTHORITY\\SYSTEM.") click.echo("It will restart automatically at each boot.") - click.echo("Run `windows-mcp service uninstall` to remove it.") + click.echo( + "Run `windows-mcp service secure-desktop set-policy ` to change without reinstalling." + ) + click.echo("Run `windows-mcp service secure-desktop uninstall` to remove it.") -@service.command("uninstall") -def service_uninstall(): - """Stop and remove the Windows MCP host service (requires elevation).""" +@service_secure_desktop.command("uninstall") +def service_secure_desktop_uninstall(): + """Stop and remove the Secure Desktop host service (requires elevation).""" _require_win32() import win32serviceutil - import win32service import pywintypes try: win32serviceutil.StopService(_SERVICE_NAME) click.echo(f"Service '{_SERVICE_NAME}' stopped.") except pywintypes.error: - pass # Not running — that's fine + pass # Not running -- that's fine try: win32serviceutil.RemoveService(_SERVICE_NAME) @@ -939,19 +1169,61 @@ def service_uninstall(): except pywintypes.error as exc: raise click.ClickException(f"Failed to remove service: {exc}") - # Restore the secure desktop UAC policy. try: - _set_prompt_on_secure_desktop(True) - click.echo("UAC policy : PromptOnSecureDesktop restored.") + from windows_mcp.service import policy as policy_mod + + policy_mod.delete_from_registry() + click.echo("UAC consent policy : cleared from registry.") + except Exception as exc: + click.echo(f"Warning: could not clear UAC policy registry key: {exc}") + + try: + _set_uac_secure_desktop_off(False) + click.echo( + "UAC policy : PromptOnSecureDesktop=1, ConsentPromptBehaviorAdmin=5 " + "(Secure Desktop restored)." + ) + except Exception as exc: + click.echo(f"Warning: could not restore UAC secure-desktop policy: {exc}") + + + +@service_secure_desktop.command("set-policy") +@click.argument("policy_name", type=click.Choice(["block", "allow_with_match", "allow_all"])) +@click.option( + "--allow-publisher", + "allow_publisher", + multiple=True, + help="Publisher substring(s) for allow_with_match. Repeat or comma-separate.", +) +def service_secure_desktop_set_policy(policy_name: str, allow_publisher: tuple[str, ...]): + """Update the persisted Secure Desktop consent policy without reinstalling.""" + _require_win32() + from windows_mcp.service import policy as policy_mod + + allowlist: list[str] = [] + for raw in allow_publisher: + allowlist.extend(s.strip() for s in raw.split(",") if s.strip()) + new_policy = policy_mod.SecureDesktopPolicy(policy=policy_name, publishers_allowlist=allowlist) + try: + policy_mod.write_to_registry(new_policy) + except PermissionError as exc: + raise click.ClickException( + f"Permission denied writing policy to HKLM: {exc}. Run as Administrator." + ) except Exception as exc: - click.echo(f"Warning: could not restore UAC policy: {exc}") + raise click.ClickException(f"Failed to write policy: {exc}") + click.echo(f"Policy updated -> {policy_name}") + if allowlist: + click.echo(f" publishers allowlist: {allowlist}") -@service.command("start") -def service_start(): - """Start the Windows MCP host service.""" +@service_secure_desktop.command("start") +def service_secure_desktop_start(): + """Start the Secure Desktop host service.""" _require_win32() import win32serviceutil + try: win32serviceutil.StartService(_SERVICE_NAME) click.echo(f"Service '{_SERVICE_NAME}' started.") @@ -959,11 +1231,12 @@ def service_start(): raise click.ClickException(f"Failed to start service: {exc}") -@service.command("stop") -def service_stop(): - """Stop the Windows MCP host service.""" +@service_secure_desktop.command("stop") +def service_secure_desktop_stop(): + """Stop the Secure Desktop host service.""" _require_win32() import win32serviceutil + try: win32serviceutil.StopService(_SERVICE_NAME) click.echo(f"Service '{_SERVICE_NAME}' stopped.") @@ -971,9 +1244,9 @@ def service_stop(): raise click.ClickException(f"Failed to stop service: {exc}") -@service.command("status") -def service_status(): - """Show the current status of the Windows MCP host service.""" +@service_secure_desktop.command("status") +def service_secure_desktop_status(): + """Show the current status of the Secure Desktop host service.""" _require_win32() import win32serviceutil import win32service @@ -989,16 +1262,17 @@ def service_status(): if status[1] == win32service.SERVICE_RUNNING: try: from windows_mcp.service.pipe import get_client + client = get_client() client.invalidate_cache() if client.is_available(): desktop = client.desktop_name() - click.echo(f"Pipe : reachable") + click.echo("Pipe : reachable") click.echo(f"Desktop : {desktop}") else: click.echo("Pipe : not reachable (service may still be starting)") except Exception as exc: - click.echo(f"Pipe : error — {exc}") + click.echo(f"Pipe : error -- {exc}") except pywintypes.error: click.echo(f"Service '{_SERVICE_NAME}' is not installed.") diff --git a/src/windows_mcp/desktop/screenshot.py b/src/windows_mcp/desktop/screenshot.py index 537f4ac0..6c5a868f 100644 --- a/src/windows_mcp/desktop/screenshot.py +++ b/src/windows_mcp/desktop/screenshot.py @@ -1,6 +1,3 @@ -import ctypes -import ctypes.wintypes -import io import logging import os @@ -216,69 +213,6 @@ def capture(self, capture_rect: uia.Rect | None) -> Image.Image: return image -# --------------------------------------------------------------------------- -# UAC / Secure Desktop detection (user-mode, no elevation needed) -# --------------------------------------------------------------------------- - -_UOI_NAME = 2 -_DESKTOP_READOBJECTS = 0x0001 - - -def is_secure_desktop_active() -> bool: - """Return True if the Secure Desktop (UAC prompt) is currently the input desktop. - - Works from any privilege level — we only need DESKTOP_READOBJECTS to read - the desktop name. Returns False on any error so callers degrade safely. - """ - try: - hdesk = ctypes.windll.user32.OpenInputDesktop(0, False, _DESKTOP_READOBJECTS) - if not hdesk: - return False - try: - buf = ctypes.create_unicode_buffer(256) - needed = ctypes.wintypes.DWORD() - ctypes.windll.user32.GetUserObjectInformationW( - hdesk, _UOI_NAME, buf, ctypes.sizeof(buf), ctypes.byref(needed) - ) - return buf.value.lower() == "winlogon" - finally: - ctypes.windll.user32.CloseDesktop(hdesk) - except Exception: - return False - - -# --------------------------------------------------------------------------- -# Service backend (routes through the LocalSystem host service when installed) -# --------------------------------------------------------------------------- - - -class _ServiceBackend(_ScreenshotBackend): - """Capture via the Windows MCP host service. - - Used automatically when the service is installed AND the Secure Desktop is - active (UAC prompt). During normal desktop use this backend is skipped so - there is zero pipe overhead on every screenshot. - """ - - name = "service" - priority = 5 # lower number = tried first; beats dxcam (10), mss (20), pillow (100) - - def is_available(self, capture_rect: "uia.Rect | None") -> bool: - if not is_secure_desktop_active(): - return False - try: - from windows_mcp.service.pipe import get_client - return get_client().is_available() - except Exception: - return False - - def capture(self, capture_rect: "uia.Rect | None") -> Image.Image: - from windows_mcp.service.pipe import get_client - png_bytes = get_client().screenshot() - img = Image.open(io.BytesIO(png_bytes)) - return _crop_screenshot(img, capture_rect) - - # --------------------------------------------------------------------------- # Instance management # --------------------------------------------------------------------------- diff --git a/src/windows_mcp/desktop/service.py b/src/windows_mcp/desktop/service.py index 8fd422df..378e7ee3 100755 --- a/src/windows_mcp/desktop/service.py +++ b/src/windows_mcp/desktop/service.py @@ -38,6 +38,50 @@ logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) + +def _is_consent_target(x: int, y: int) -> bool: + """Return True if the screen pixel at (x, y) is owned by consent.exe. + + Used to detect when a Click should be routed through the SYSTEM host + service's UIAccess worker instead of the broker's medium-integrity + uia.Click, which UIPI blocks against the UAC dialog. + """ + try: + hwnd = win32gui.WindowFromPoint((x, y)) + if not hwnd: + return False + # WindowFromPoint may return a child HWND; walk up to the top. + top = win32gui.GetAncestor(hwnd, win32con.GA_ROOT) + _, pid = win32process.GetWindowThreadProcessId(top or hwnd) + if not pid: + return False + try: + name = Process(pid).name().lower() + except Exception: # noqa: BLE001 + return False + return name == "consent.exe" + except Exception: # noqa: BLE001 + return False + + +def _route_click_through_host(x: int, y: int) -> bool: + """Best-effort click via the SYSTEM host service. Returns True if it took. + + Lazy-imports the pipe client so non-UAC code paths don't pay the cost. + """ + try: + from windows_mcp.service.pipe import get_client + client = get_client() + if not client.is_available(): + return False + ok = bool(client.uia_click_at(x, y)) + if ok: + logger.info("Click(%d,%d) routed through host UIAccess worker", x, y) + return ok + except Exception as exc: # noqa: BLE001 + logger.warning("host-routed click failed, falling back to local: %s", exc) + return False + import windows_mcp.uia as uia # noqa: E402 # Key name aliases for shortcut keys that differ from UIA SpecialKeyNames @@ -121,15 +165,11 @@ def get_state( capture_rect = self.get_display_union_rect(display_indices) if display_indices else None screenshot_region = self._rect_to_bounding_box(capture_rect) if capture_rect else None - # Fast path for Screenshot tool (use_ui_tree=False): skip window enumeration. - # UIAutomation calls (get_controls_handles / get_windows / get_active_window) - # can hang when an app is launching and not responding to WM messages. - # Also skip when the Secure Desktop is active (UAC prompt): UIA cannot - # cross the Winlogon desktop boundary from user mode, so all calls - # return None and crash attempting to walk the accessibility tree. - from windows_mcp.desktop.screenshot import is_secure_desktop_active - uac_active = is_secure_desktop_active() - if use_ui_tree and not uac_active: + # Fast path for Screenshot tool (use_ui_tree=False): skip window + # enumeration. UIAutomation calls (get_controls_handles / get_windows + # / get_active_window) can hang when an app is launching and not + # responding to WM messages. + if use_ui_tree: controls_handles = self.get_controls_handles() # Taskbar,Program Manager,Apps, Dialogs windows, windows_handles = self.get_windows(controls_handles=controls_handles) # Apps active_window = self.get_active_window(windows=windows) # Active Window @@ -163,10 +203,7 @@ def get_state( logger.debug(f"Active window: {active_window or 'No Active Window Found'}") logger.debug(f"Windows: {windows}") - if uac_active and use_ui_tree: - # UIA tree via the host service — the only way to read the UAC dialog. - tree_state = self._build_uac_tree_state() - elif use_ui_tree: + if use_ui_tree: other_windows_handles = list(controls_handles - windows_handles) tree_state = self.tree.get_state( active_window_handle, other_windows_handles, use_dom=use_dom @@ -647,23 +684,17 @@ def click(self, loc: tuple[int, int] | list[int], button: str = "left", clicks: else: x, y = loc - # With PromptOnSecureDesktop disabled (set by `windows-mcp service install`), - # UAC prompts appear on the Default desktop and uia.Click() reaches them via - # SendInput — hardware-level input that bypasses UIPI. - # The service route below is a fallback for the rare case where the policy - # wasn't applied (secure desktop still active). - from windows_mcp.desktop.screenshot import is_secure_desktop_active - if button == "left" and is_secure_desktop_active(): - from windows_mcp.service import get_host_client - try: - get_host_client().uia_click_at(x, y) - except Exception as exc: - logger.warning("UAC click via service failed: %s", exc) - return - if clicks == 0: uia.SetCursorPos(x, y) return + # UAC routing: consent.exe runs at System integrity. UIPI blocks + # mouse input from medium-integrity processes, so a normal uia.Click + # at the dialog's Yes/No coords silently fails to dismiss UAC. The + # SYSTEM host service has a UIAccess-tokened user-session worker + # that CAN invoke across integrity; route the click there if the + # target pixel is owned by consent.exe. + if _is_consent_target(x, y) and _route_click_through_host(x, y): + return match button: case "left": if clicks >= 2: @@ -688,6 +719,7 @@ def type( press_enter: bool | str = False, ): x, y = loc + uia.Click(x, y) if caret_position == "start": uia.SendKeys("{Home}", waitTime=0.05) @@ -738,11 +770,12 @@ def scroll( return 'Invalid type. Use "horizontal" or "vertical".' return None - def drag(self, loc: tuple[int, int]|list[int]): + def drag(self, loc: tuple[int, int] | list[int]): if isinstance(loc, list): x, y = loc[0], loc[1] else: x, y = loc + sleep(0.5) cx, cy = uia.GetCursorPos() uia.DragDrop(cx, cy, x, y, moveSpeed=1) @@ -847,48 +880,6 @@ def callback(hwnd, _): handles.add(secondary_taskbar_hwnd) return handles - def _build_uac_tree_state(self) -> TreeState: - """Build a TreeState from the host service UIA tree (used during UAC).""" - from windows_mcp.service import get_host_client - try: - raw = get_host_client().uia_tree() - except Exception as exc: - logger.warning("_build_uac_tree_state: service call failed: %s", exc) - return TreeState(status=False) - - interactive: list[TreeElementNode] = [] - - def _walk(node: dict, window_name: str) -> None: - bbox = node.get("bbox", {}) - center = node.get("center", {}) - ctrl = node.get("control_type", "") - name = node.get("name", "") - can_invoke = node.get("can_invoke", False) - - # Include buttons and any invokable element as interactive nodes. - if bbox and center and (can_invoke or ctrl.lower() in ("button",)): - bb = BoundingBox( - left=bbox["left"], top=bbox["top"], - right=bbox["right"], bottom=bbox["bottom"], - width=bbox["width"], height=bbox["height"], - ) - c = Center(x=center["x"], y=center["y"]) - interactive.append(TreeElementNode( - name=name, - control_type=ctrl + "Control" if ctrl else "Control", - window_name=window_name, - bounding_box=bb, - center=c, - metadata={}, - )) - for child in node.get("children", []): - _walk(child, window_name) - - for top in raw: - _walk(top, top.get("name", "Secure Desktop")) - - return TreeState(status=True, interactive_nodes=interactive) - def get_active_window(self, windows: list[Window] | None = None) -> Window | None: try: if windows is None: diff --git a/src/windows_mcp/infrastructure/__init__.py b/src/windows_mcp/infrastructure/__init__.py index e8a0984d..242d3c00 100644 --- a/src/windows_mcp/infrastructure/__init__.py +++ b/src/windows_mcp/infrastructure/__init__.py @@ -12,6 +12,8 @@ ServerConfig, SecurityConfig, ToolsConfig, + SecureDesktopConfig, + SECURE_DESKTOP_POLICIES, CONFIG_DIR, CONFIG_FILE, discover_config_path, @@ -34,6 +36,8 @@ "ServerConfig", "SecurityConfig", "ToolsConfig", + "SecureDesktopConfig", + "SECURE_DESKTOP_POLICIES", "CONFIG_DIR", "CONFIG_FILE", "discover_config_path", diff --git a/src/windows_mcp/infrastructure/config.py b/src/windows_mcp/infrastructure/config.py index e4c65807..7960af3a 100644 --- a/src/windows_mcp/infrastructure/config.py +++ b/src/windows_mcp/infrastructure/config.py @@ -30,11 +30,33 @@ class ToolsConfig: exclude: list[str] = field(default_factory=list) +SECURE_DESKTOP_POLICIES = ("block", "allow_with_match", "allow_all") + + +@dataclass +class SecureDesktopConfig: + """Policy for how the LocalSystem host service may handle UAC consent prompts. + + ``policy`` values: + - ``block`` — service exposes the dialog to the agent but + REFUSES to auto-click Yes/No. Human approval + still required. (default) + - ``allow_with_match``— auto-click only if the requesting binary's + publisher (CommonName from the Authenticode + signature) matches one of ``publishers_allowlist``. + - ``allow_all`` — auto-click any UAC prompt. Only safe in sandboxed + VMs. Opt-in. + """ + policy: str = "block" + publishers_allowlist: list[str] = field(default_factory=list) + + @dataclass class WindowsMCPConfig: server: ServerConfig = field(default_factory=ServerConfig) security: SecurityConfig = field(default_factory=SecurityConfig) tools: ToolsConfig = field(default_factory=ToolsConfig) + secure_desktop: SecureDesktopConfig = field(default_factory=SecureDesktopConfig) source_path: Path | None = None @@ -117,6 +139,19 @@ def load_config(path: Path | None) -> WindowsMCPConfig: if "exclude" in tools: cfg.tools.exclude = _list_of_strings(tools["exclude"], "tools.exclude") + secure_desktop = data.get("secure_desktop", {}) + if "policy" in secure_desktop: + p = str(secure_desktop["policy"]) + if p not in SECURE_DESKTOP_POLICIES: + raise ValueError( + f"secure_desktop.policy must be one of {SECURE_DESKTOP_POLICIES}, got {p!r}" + ) + cfg.secure_desktop.policy = p + if "publishers_allowlist" in secure_desktop: + cfg.secure_desktop.publishers_allowlist = _list_of_strings( + secure_desktop["publishers_allowlist"], "secure_desktop.publishers_allowlist" + ) + cfg.source_path = path return cfg @@ -160,4 +195,14 @@ def write_config(cfg: WindowsMCPConfig, path: Path) -> None: items = ', '.join(f'"{t}"' for t in cfg.tools.exclude) lines += ['[tools]', f'exclude = [{items}]', ''] + sd_cfg, sd_def = cfg.secure_desktop, SecureDesktopConfig() + sd_lines: list[str] = [] + if sd_cfg.policy != sd_def.policy: + sd_lines.append(f'policy = "{sd_cfg.policy}"') + if sd_cfg.publishers_allowlist: + items = ', '.join(f'"{p}"' for p in sd_cfg.publishers_allowlist) + sd_lines.append(f'publishers_allowlist = [{items}]') + if sd_lines: + lines += ['[secure_desktop]'] + sd_lines + [''] + path.write_text('\n'.join(lines), encoding='utf-8') diff --git a/src/windows_mcp/service/host.py b/src/windows_mcp/service/host.py index 34e00427..f76345cc 100644 --- a/src/windows_mcp/service/host.py +++ b/src/windows_mcp/service/host.py @@ -3,7 +3,7 @@ This module serves two purposes: 1. **Service class** (``WindowsMCPHostService``) — a ``pywin32`` - ``ServiceFramework`` subclass installed via ``windows-mcp service install``. + ``ServiceFramework`` subclass installed via ``windows-mcp service secure-desktop install``. It starts a named pipe server that handles privileged desktop operations requested by the user-mode broker. @@ -24,21 +24,44 @@ from __future__ import annotations -import base64 import logging import threading from typing import Any from .protocol import PIPE_NAME, PIPE_BUFFER_SIZE, Request, Response -from . import secure_desktop +from . import policy, secure_desktop logger = logging.getLogger(__name__) def _setup_file_logging() -> None: - """Write service logs to %TEMP%\\windows-mcp-host.log (no stdout in a service).""" + """Write service logs to a world-readable file (no stdout in a service). + + The service runs as SYSTEM. Logging under %TEMP% (SYSTEM's profile temp) + makes the file unreadable by the interactive user, so diagnosing the + service used to require an elevated dump. Prefer + ``%ProgramData%\\windows-mcp\\host.log`` — ProgramData is readable by + Users by default, so an ordinary (non-elevated) process can tail the + service log. Fall back to %TEMP% if ProgramData can't be created. + """ import os - log_path = os.path.join(os.environ.get("TEMP", "C:\\Temp"), "windows-mcp-host.log") + candidates = [] + program_data = os.environ.get("ProgramData") + if program_data: + candidates.append(os.path.join(program_data, "windows-mcp")) + candidates.append(os.environ.get("TEMP", "C:\\Temp")) + + log_path = None + for base in candidates: + try: + os.makedirs(base, exist_ok=True) + log_path = os.path.join(base, "windows-mcp-host.log") + break + except Exception: + continue + if log_path is None: + log_path = os.path.join(os.environ.get("TEMP", "C:\\Temp"), "windows-mcp-host.log") + logging.basicConfig( filename=log_path, level=logging.DEBUG, @@ -64,34 +87,176 @@ def _setup_file_logging() -> None: # Pipe security # --------------------------------------------------------------------------- -def _build_pipe_sa() -> Any: - """Return a SECURITY_ATTRIBUTES with a NULL DACL (allows all local access). +# Pipe-specific access flags. FILE_ALL_ACCESS = 0x1F01FF — full read/write on +# the pipe handle. We grant this to the two principals we trust: +# - SYSTEM: the service itself +# - Active console user SID: the broker process running in their session +_FILE_ALL_ACCESS = 0x1F01FF + +# Identifiers for the well-known SIDs we need. +# win32security.WinLocalSystemSid → S-1-5-18 (NT AUTHORITY\SYSTEM) +# win32security.WinInteractiveSid → S-1-5-4 (NT AUTHORITY\INTERACTIVE) +_SID_TYPE_SYSTEM = "WinLocalSystemSid" +_SID_TYPE_INTERACTIVE = "WinInteractiveSid" + - A NULL DACL is intentional here: the pipe is local-only (no network - listener), so allowing any local process to connect is fine. The tighter - SYSTEM+user DACL can be added later once the basic pipe works; for now, - complexity in the DACL was causing CreateNamedPipe to fail silently. +def _console_user_sid() -> Any | None: + """Return the SID of the user logged on to the physical console, or None. - Note: SECURITY_ATTRIBUTES lives in pywintypes, not win32security. + Pattern: + WTSGetActiveConsoleSessionId → WTSQueryUserToken → GetTokenInformation + with TokenUser → SID. + + Returns None on services like CI where no interactive session exists yet. """ if not _WIN32_AVAILABLE: return None try: - import pywintypes + import win32ts + import win32api + session_id = win32ts.WTSGetActiveConsoleSessionId() + # 0xFFFFFFFF (~0) means no active session. + if session_id is None or session_id == 0xFFFFFFFF: + logger.info("No active console session yet") + return None + token = win32ts.WTSQueryUserToken(session_id) + try: + token_user = win32security.GetTokenInformation( + token, win32security.TokenUser + ) + # GetTokenInformation(TokenUser) returns a tuple (SID, attrs). + sid = token_user[0] + logger.info( + "Console user SID for session %d: %s", + session_id, win32security.ConvertSidToStringSid(sid), + ) + return sid + finally: + win32api.CloseHandle(token) + except Exception as exc: + logger.warning("Could not resolve console user SID: %s", exc) + return None + + +def _build_pipe_sa() -> Any: + """Return a SECURITY_ATTRIBUTES that allows only SYSTEM + the console user. + + Falls back to SYSTEM + NT AUTHORITY\\INTERACTIVE (S-1-5-4) when no console + user is logged in yet — the typical case at boot, because the service is + SERVICE_AUTO_START and comes up before anyone logs in. The pipe server + loop rebuilds this SD for every instance and then blocks on + ConnectNamedPipe, so the *first* boot-time instance keeps whatever DACL it + was created with until a client actually connects. If that fallback were + BUILTIN\\Administrators (the previous behaviour), the broker — which runs + in the interactive session under the user's *filtered* medium-integrity + token, where Administrators is a deny-only SID — could never open the pipe, + the ConnectNamedPipe would never complete, and the loop would never + recreate the instance with the correct console-user DACL. It would hang + until a manual service restart. INTERACTIVE is present in every + interactively-logged-on token (filtered or not), so the boot-time pipe is + reachable by the interactive user as soon as they log in; once a console + user is resolved, later instances tighten to SYSTEM + that specific SID. + The privileged operations behind the pipe are policy-gated regardless. + Never falls back to a NULL DACL — that was the original mistake. + + Raising would prevent the service from starting; instead, on failure we + return a SECURITY_ATTRIBUTES with a *deny-all* DACL so the pipe is created + but unreachable, making the failure obvious in logs rather than silent. + """ + if not _WIN32_AVAILABLE: + return None + + import pywintypes + + try: + # SYSTEM SID — always allowed; the service runs as SYSTEM. + sid_system = win32security.CreateWellKnownSid( + getattr(win32security, _SID_TYPE_SYSTEM), None + ) + + # Console user SID (if anyone is logged in); else fall back to the + # INTERACTIVE group so the pipe is still reachable by the interactive + # user once they log in (see _build_pipe_sa docstring for why Admins + # would deadlock the boot-time instance). + sid_user = _console_user_sid() + if sid_user is None: + sid_user = win32security.CreateWellKnownSid( + getattr(win32security, _SID_TYPE_INTERACTIVE), None + ) + logger.info("Pipe DACL fallback: SYSTEM + NT AUTHORITY\\INTERACTIVE") + else: + logger.info( + "Pipe DACL: SYSTEM + console user %s", + win32security.ConvertSidToStringSid(sid_user), + ) + + dacl = win32security.ACL() + dacl.AddAccessAllowedAce( + win32security.ACL_REVISION, _FILE_ALL_ACCESS, sid_system + ) + dacl.AddAccessAllowedAce( + win32security.ACL_REVISION, _FILE_ALL_ACCESS, sid_user + ) + sd = win32security.SECURITY_DESCRIPTOR() - sd.SetSecurityDescriptorDacl(True, None, False) # NULL DACL = everyone + sd.SetSecurityDescriptorDacl(True, dacl, False) + sd.SetSecurityDescriptorOwner(sid_system, False) + sa = pywintypes.SECURITY_ATTRIBUTES() sa.SECURITY_DESCRIPTOR = sd return sa + except Exception as exc: - logger.warning("Could not build pipe SA, falling back to None: %s", exc) - return None + # Defensive: empty DACL = deny everyone except the SD owner. The pipe + # will still be created, but clients won't be able to connect — and + # the exception is logged loudly so the failure mode is discoverable. + logger.exception("Failed to build restrictive pipe DACL: %s", exc) + try: + empty_dacl = win32security.ACL() + sd = win32security.SECURITY_DESCRIPTOR() + sd.SetSecurityDescriptorDacl(True, empty_dacl, False) + sa = pywintypes.SECURITY_ATTRIBUTES() + sa.SECURITY_DESCRIPTOR = sd + return sa + except Exception: + return None # --------------------------------------------------------------------------- # Request dispatcher # --------------------------------------------------------------------------- +def _enforce_policy(operation: str) -> tuple[bool, str]: + """Return (allowed, reason) for an auto-click on the UAC consent dialog. + + Every click routed to the host is a consent-dialog click — the broker only + forwards clicks whose target pixel is owned by ``consent.exe`` — so the + consent policy is always the deciding factor. + + Read-only ops (tree walks, publisher lookups) are not gated; agents always + need visibility into UAC. Only the auto-click is gated, because that is the + action that would bypass the human. + + Earlier revisions additionally short-circuited to "allowed" unless the input + desktop reported ``Winlogon``. That read is racy while UAC is up with + ``PromptOnSecureDesktop=0`` (it flips between ``Default`` and ``Winlogon``), + which made ``block`` / ``allow_with_match`` enforce only intermittently, so + the desktop check is dropped in favour of unconditional enforcement. + """ + pol = policy.read_from_registry() + try: + publisher = secure_desktop._spawn_in_user_session("publisher", timeout=15.0) + except Exception as exc: + logger.warning("policy: user-session publisher lookup failed: %s", exc) + publisher = None + allowed, reason = pol.allows_auto_click(publisher) + logger.info( + "policy check: op=%s policy=%s publisher=%r → %s (%s)", + operation, pol.policy, publisher, allowed, reason, + ) + return allowed, reason + + def _dispatch(req: Request) -> Response: """Execute a single request and return a response.""" try: @@ -103,24 +268,25 @@ def _dispatch(req: Request) -> Response: name = secure_desktop.get_input_desktop_name() return Response(id=req.id, result=name) - case "screenshot": - png = secure_desktop.capture_screenshot() - return Response(id=req.id, result=base64.b64encode(png).decode()) - - case "uia_windows": - titles = secure_desktop.uia_get_window_titles() - return Response(id=req.id, result=titles) + case "wait_for_uac_prompt": + timeout_ms = int(req.params.get("timeout_ms", 60_000)) + result = secure_desktop.wait_for_uac_prompt(timeout_ms=timeout_ms) + return Response(id=req.id, result=result) - case "uia_tree": - tree = secure_desktop.uia_get_tree() - return Response(id=req.id, result=tree) - - case "uia_invoke": - ok = secure_desktop.uia_invoke_element(req.params["name"]) - return Response(id=req.id, result=ok) + case "policy_state": + pol = policy.read_from_registry() + return Response(id=req.id, result={ + "policy": pol.policy, + "publishers_allowlist": pol.publishers_allowlist, + }) case "uia_click_at": - ok = secure_desktop.uia_click_at(req.params["x"], req.params["y"]) + allowed, reason = _enforce_policy("uia_click_at") + if not allowed: + return Response(id=req.id, error=f"policy denied: {reason}") + ok = secure_desktop._spawn_in_user_session( + "click_at", str(req.params["x"]), str(req.params["y"]) + ) return Response(id=req.id, result=ok) case _: @@ -197,6 +363,15 @@ def _serve_one_client(handle: Any) -> None: req = Request.decode(data) resp = _dispatch(req) win32file.WriteFile(handle, resp.encode()) + # FlushFileBuffers blocks until the client has read the response. + # Without it, the DisconnectNamedPipe below will discard the bytes + # we just wrote — MSDN: "DisconnectNamedPipe forces all the data + # that has not been read out of the pipe to be discarded." The + # race shows up under timing variance: small responses fit the + # pipe buffer instantly so WriteFile returns before the client + # has drained it, and the client then reads 0 bytes from a + # disconnected pipe instead of the actual response. + win32file.FlushFileBuffers(handle) except Exception as exc: logger.warning("Error serving pipe client: %s", exc) finally: diff --git a/src/windows_mcp/service/pipe.py b/src/windows_mcp/service/pipe.py index a05e5108..5698da9b 100644 --- a/src/windows_mcp/service/pipe.py +++ b/src/windows_mcp/service/pipe.py @@ -6,13 +6,12 @@ client = get_client() if client.is_available(): - png = client.screenshot() # bytes name = client.desktop_name() # "Default" | "Winlogon" + result = client.wait_for_uac_prompt(timeout_ms=60_000) """ from __future__ import annotations -import base64 import logging import time from typing import Any @@ -68,27 +67,22 @@ def desktop_name(self) -> str: """Return the name of the current input desktop ('Default' or 'Winlogon').""" return self._call("desktop_name", {}) - def screenshot(self) -> bytes: - """Capture the current input desktop (incl. UAC). Returns PNG bytes.""" - encoded = self._call("screenshot", {}) - return base64.b64decode(encoded) - - def uia_windows(self) -> list[str]: - """Return top-level window titles visible on the input desktop.""" - return self._call("uia_windows", {}) - - def uia_tree(self) -> list[dict]: - """Return the full UIA tree of the input desktop as a list of window dicts.""" - return self._call("uia_tree", {}) - - def uia_invoke(self, name: str) -> bool: - """Find and invoke a named element (e.g. 'Yes', 'No') on the input desktop.""" - return self._call("uia_invoke", {"name": name}) - def uia_click_at(self, x: int, y: int) -> bool: """Invoke the element at screen coordinates (x, y) on the input desktop.""" return self._call("uia_click_at", {"x": x, "y": y}) + def wait_for_uac_prompt(self, timeout_ms: int = 60_000) -> dict | None: + """Block until UAC fires on the input desktop, or until *timeout_ms* elapses. + + Returns a dict ``{"desktop": "Winlogon", "publisher": str|None, "tree": [...]}`` + on success, or ``None`` on timeout. + """ + return self._call("wait_for_uac_prompt", {"timeout_ms": timeout_ms}) + + def policy_state(self) -> dict: + """Return the persisted Secure-Desktop policy and allowlist.""" + return self._call("policy_state", {}) + # ------------------------------------------------------------------ # Internal # ------------------------------------------------------------------ @@ -98,22 +92,7 @@ def _call(self, method: str, params: dict[str, Any]) -> Any: raise RuntimeError("pywin32 is not available") req = Request(method=method, params=params) - - try: - # Block until the pipe is available (or timeout). - win32pipe.WaitNamedPipe(PIPE_NAME, CALL_TIMEOUT_MS) - - handle = win32file.CreateFile( - PIPE_NAME, - win32file.GENERIC_READ | win32file.GENERIC_WRITE, - 0, - None, - win32file.OPEN_EXISTING, - 0, - None, - ) - except pywintypes.error as exc: - raise RuntimeError(f"Cannot connect to host service pipe: {exc}") from exc + handle = self._open_with_retry() try: # Switch to message read mode so we get whole messages back. @@ -138,6 +117,40 @@ def _call(self, method: str, params: dict[str, Any]) -> Any: raise RuntimeError(f"Host service error ({method}): {resp.error}") return resp.result + def _open_with_retry(self, *, attempts: int = 30, gap_ms: int = 100) -> Any: + """Open the named pipe, retrying on the brief race window where the + server has connected one instance but not yet recreated the next. + + WaitNamedPipe returns ERROR_FILE_NOT_FOUND (not ERROR_SEM_TIMEOUT) + when no instance of the pipe is currently in WAITING_FOR_CONNECT + state. The host service spends ~20-50 ms between accepting one + connection and creating the next instance; back-to-back broker + calls (e.g. is_available() ping immediately followed by an actual + operation) often land inside that window. Retry with a short gap. + """ + last_exc: Any = None + for _ in range(attempts): + try: + win32pipe.WaitNamedPipe(PIPE_NAME, CALL_TIMEOUT_MS) + return win32file.CreateFile( + PIPE_NAME, + win32file.GENERIC_READ | win32file.GENERIC_WRITE, + 0, + None, + win32file.OPEN_EXISTING, + 0, + None, + ) + except pywintypes.error as exc: + last_exc = exc + if exc.winerror in (2, 231): # FILE_NOT_FOUND, PIPE_BUSY + time.sleep(gap_ms / 1000.0) + continue + raise RuntimeError(f"Cannot connect to host service pipe: {exc}") from exc + raise RuntimeError( + f"Cannot connect to host service pipe after {attempts} retries: {last_exc}" + ) + _client: HostServiceClient | None = None diff --git a/src/windows_mcp/service/policy.py b/src/windows_mcp/service/policy.py new file mode 100644 index 00000000..8bd665da --- /dev/null +++ b/src/windows_mcp/service/policy.py @@ -0,0 +1,173 @@ +"""Secure-Desktop consent policy — read from HKLM, written on install/set-policy. + +Three policies (per issue #236): + +* ``block`` — service exposes the UAC dialog to the agent but + REFUSES to auto-click Yes/No. Default. +* ``allow_with_match`` — auto-click only if the requesting binary's publisher + (as it appears in the UAC dialog) substring-matches + one of ``publishers_allowlist``. +* ``allow_all`` — auto-click any UAC prompt. Only for sandboxed VMs. + +The policy is persisted in the registry so that the LocalSystem service can +read it without any inheritance from the broker's user environment. The +broker also reads it to pre-filter requests before they ever leave the user +session (defense in depth — even a tampered broker cannot bypass the SYSTEM +service's check). + +Registry layout:: + + HKLM\\SOFTWARE\\Windows-MCP\\SecureDesktop + Policy REG_SZ "block" | "allow_with_match" | "allow_all" + PublishersAllowlist REG_MULTI_SZ ["Microsoft Corporation", ...] +""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass, field + +logger = logging.getLogger(__name__) + +POLICIES = ("block", "allow_with_match", "allow_all") +DEFAULT_POLICY = "block" + +ENV_POLICY = "WINDOWS_MCP_SECURE_DESKTOP_POLICY" +ENV_ALLOWLIST = "WINDOWS_MCP_SECURE_DESKTOP_ALLOWLIST" + +_REG_PATH = r"SOFTWARE\Windows-MCP\SecureDesktop" +_REG_POLICY = "Policy" +_REG_ALLOWLIST = "PublishersAllowlist" + + +@dataclass +class SecureDesktopPolicy: + policy: str = DEFAULT_POLICY + publishers_allowlist: list[str] = field(default_factory=list) + + def allows_auto_click(self, publisher: str | None) -> tuple[bool, str]: + """Return ``(allowed, reason)`` for an auto-click attempt on the Secure Desktop. + + ``publisher`` is the "Verified publisher" string from the UAC dialog + (or ``None`` if it could not be determined). + """ + if self.policy == "allow_all": + return True, "policy=allow_all" + if self.policy == "block": + return False, "policy=block" + # allow_with_match + if publisher is None: + return False, "publisher unknown; allow_with_match requires a match" + for needle in self.publishers_allowlist: + if needle and needle.lower() in publisher.lower(): + return True, f"publisher {publisher!r} matched allowlist entry {needle!r}" + return False, f"publisher {publisher!r} not in allowlist" + + +def _validate_policy(value: str) -> str: + v = value.strip().lower() + if v not in POLICIES: + raise ValueError( + f"Invalid policy {value!r}; must be one of {POLICIES}" + ) + return v + + +def from_env() -> SecureDesktopPolicy | None: + """Build a policy from environment variables, or return None if unset.""" + raw = os.environ.get(ENV_POLICY) + if not raw: + return None + policy = _validate_policy(raw) + raw_list = os.environ.get(ENV_ALLOWLIST, "") + allowlist = [s.strip() for s in raw_list.split(",") if s.strip()] + return SecureDesktopPolicy(policy=policy, publishers_allowlist=allowlist) + + +def read_from_registry() -> SecureDesktopPolicy: + """Read the persisted policy. Returns the default policy on any failure.""" + try: + import winreg + except ImportError: + return SecureDesktopPolicy() + try: + with winreg.OpenKey( + winreg.HKEY_LOCAL_MACHINE, _REG_PATH, access=winreg.KEY_READ + ) as key: + try: + policy_raw, _ = winreg.QueryValueEx(key, _REG_POLICY) + policy = _validate_policy(str(policy_raw)) + except FileNotFoundError: + policy = DEFAULT_POLICY + try: + allowlist_raw, _ = winreg.QueryValueEx(key, _REG_ALLOWLIST) + allowlist = [s for s in (allowlist_raw or []) if s] + except FileNotFoundError: + allowlist = [] + except FileNotFoundError: + return SecureDesktopPolicy() + except OSError as exc: + logger.warning("Reading policy from registry failed: %s", exc) + return SecureDesktopPolicy() + return SecureDesktopPolicy(policy=policy, publishers_allowlist=allowlist) + + +def write_to_registry(policy: SecureDesktopPolicy) -> None: + """Persist *policy* to HKLM. Requires elevation.""" + import winreg + _validate_policy(policy.policy) + with winreg.CreateKeyEx( + winreg.HKEY_LOCAL_MACHINE, _REG_PATH, access=winreg.KEY_SET_VALUE + ) as key: + winreg.SetValueEx(key, _REG_POLICY, 0, winreg.REG_SZ, policy.policy) + winreg.SetValueEx( + key, _REG_ALLOWLIST, 0, winreg.REG_MULTI_SZ, list(policy.publishers_allowlist) + ) + logger.info( + "Wrote secure-desktop policy=%s allowlist=%s", policy.policy, policy.publishers_allowlist + ) + + +def delete_from_registry() -> None: + """Remove the persisted policy. Used on service uninstall.""" + try: + import winreg + except ImportError: + return + try: + winreg.DeleteKey(winreg.HKEY_LOCAL_MACHINE, _REG_PATH) + except FileNotFoundError: + pass + except OSError as exc: + logger.warning("Could not delete policy registry key: %s", exc) + + +def resolve_install_time_policy( + cli_policy: str | None, + cli_allowlist: list[str] | None, + config_policy: str | None, + config_allowlist: list[str] | None, +) -> SecureDesktopPolicy: + """Merge CLI flag, env var, TOML config, and default — in that precedence order.""" + env = from_env() + + if cli_policy is not None: + policy = _validate_policy(cli_policy) + elif env is not None: + policy = env.policy + elif config_policy is not None: + policy = _validate_policy(config_policy) + else: + policy = DEFAULT_POLICY + + if cli_allowlist: + allowlist = list(cli_allowlist) + elif env is not None and env.publishers_allowlist: + allowlist = env.publishers_allowlist + elif config_allowlist: + allowlist = list(config_allowlist) + else: + allowlist = [] + + return SecureDesktopPolicy(policy=policy, publishers_allowlist=allowlist) diff --git a/src/windows_mcp/service/protocol.py b/src/windows_mcp/service/protocol.py index e2046eb6..2ab6440e 100644 --- a/src/windows_mcp/service/protocol.py +++ b/src/windows_mcp/service/protocol.py @@ -5,10 +5,11 @@ Methods ------- -ping → "pong" -desktop_name → str ("Default" | "Winlogon") -screenshot → base64-encoded PNG bytes (full virtual screen) -uia_windows → list[str] of top-level window titles on input desktop +ping → "pong" +desktop_name → str ("Default" | "Winlogon") +uia_click_at → bool — invoke element at (x, y); policy-gated when desktop=Winlogon +wait_for_uac_prompt → dict | None — block until UAC fires (or timeout) +policy_state → dict — persisted Secure-Desktop policy + allowlist """ from __future__ import annotations diff --git a/src/windows_mcp/service/secure_desktop.py b/src/windows_mcp/service/secure_desktop.py index 53ef590f..add66aeb 100644 --- a/src/windows_mcp/service/secure_desktop.py +++ b/src/windows_mcp/service/secure_desktop.py @@ -24,9 +24,13 @@ import ctypes import ctypes.wintypes -import io +import json import logging +import re +import subprocess +import sys import threading +import time from contextlib import contextmanager from typing import Any @@ -40,13 +44,56 @@ _WINSTA_ALL_ACCESS = 0x037F _DESKTOP_ALL_ACCESS = 0x01FF _DESKTOP_READOBJECTS = 0x0001 +_DESKTOP_ENUMERATE = 0x0040 +_DESKTOP_SWITCHDESKTOP = 0x0100 +# Minimum rights to SetThreadDesktop + walk UIA on the input desktop. +# Winlogon's DACL doesn't grant ALL_ACCESS to admin tokens, so the read-only +# attach is the only one that actually succeeds for the user-session worker +# enumerating consent.exe. +_DESKTOP_READ_ATTACH = _DESKTOP_SWITCHDESKTOP | _DESKTOP_ENUMERATE | _DESKTOP_READOBJECTS _user32 = ctypes.windll.user32 _kernel32 = ctypes.windll.kernel32 +# Declare argtypes/restype on the user32 desktop/window-station functions +# we call. Without these, ctypes passes Python str as a c_char_p (ASCII +# bytes) but the *W APIs expect LPCWSTR (UTF-16) -- silently corrupted +# calls return NULL with no obvious cause. +_user32.OpenWindowStationW.argtypes = [ + ctypes.wintypes.LPCWSTR, + ctypes.wintypes.BOOL, + ctypes.wintypes.DWORD, +] +_user32.OpenWindowStationW.restype = ctypes.wintypes.HANDLE +_user32.OpenInputDesktop.argtypes = [ + ctypes.wintypes.DWORD, + ctypes.wintypes.BOOL, + ctypes.wintypes.DWORD, +] +_user32.OpenInputDesktop.restype = ctypes.wintypes.HANDLE +_user32.SetThreadDesktop.argtypes = [ctypes.wintypes.HANDLE] +_user32.SetThreadDesktop.restype = ctypes.wintypes.BOOL +_user32.SetProcessWindowStation.argtypes = [ctypes.wintypes.HANDLE] +_user32.SetProcessWindowStation.restype = ctypes.wintypes.BOOL +_user32.GetProcessWindowStation.restype = ctypes.wintypes.HANDLE +_user32.GetThreadDesktop.argtypes = [ctypes.wintypes.DWORD] +_user32.GetThreadDesktop.restype = ctypes.wintypes.HANDLE +_user32.CloseDesktop.argtypes = [ctypes.wintypes.HANDLE] +_user32.CloseDesktop.restype = ctypes.wintypes.BOOL +_user32.CloseWindowStation.argtypes = [ctypes.wintypes.HANDLE] +_user32.CloseWindowStation.restype = ctypes.wintypes.BOOL +_user32.GetUserObjectInformationW.argtypes = [ + ctypes.wintypes.HANDLE, + ctypes.c_int, + ctypes.c_void_p, + ctypes.wintypes.DWORD, + ctypes.POINTER(ctypes.wintypes.DWORD), +] +_user32.GetUserObjectInformationW.restype = ctypes.wintypes.BOOL +_kernel32.GetCurrentThreadId.restype = ctypes.wintypes.DWORD + # UIA constants _UIA_InvokePatternId = 10000 -_UIA_NamePropertyId = 30005 _UIA_TreeScope_Descendants = 4 @@ -54,6 +101,7 @@ # Internal helpers # --------------------------------------------------------------------------- + def _open_winsta0() -> int: handle = _user32.OpenWindowStationW("WinSta0", False, _WINSTA_ALL_ACCESS) return handle or 0 @@ -75,21 +123,46 @@ def _get_desktop_name(hdesk: int) -> str: @contextmanager def _input_desktop(): - """Switch process/thread to WinSta0\\, then restore on exit.""" + """Switch the thread to the input (Default) desktop, then restore on exit. + + The secure-desktop install routes UAC to the Default desktop, so this + only ever needs to attach to the interactive input desktop. Tries + ALL_ACCESS first (needed for synthetic input), falls back to + DESKTOP_READ_ATTACH if that's all we can get. + """ hwinsta_prev = _user32.GetProcessWindowStation() hdesk_prev = _user32.GetThreadDesktop(_kernel32.GetCurrentThreadId()) hwinsta = _open_winsta0() if hwinsta: _user32.SetProcessWindowStation(hwinsta) - hdesk = _open_input_desktop(_DESKTOP_ALL_ACCESS) + hdesk = 0 + own_hdesk = True + attached_via = None + for access in (_DESKTOP_ALL_ACCESS, _DESKTOP_READ_ATTACH): + hdesk = _open_input_desktop(access) + if not hdesk: + continue + if _user32.SetThreadDesktop(hdesk): + attached_via = access + break + _user32.CloseDesktop(hdesk) + hdesk = 0 if hdesk: - _user32.SetThreadDesktop(hdesk) + name = _get_desktop_name(hdesk) or "(unknown)" + logger.info("_input_desktop: attached to %r access=0x%04x", name, attached_via) + else: + logger.warning( + "_input_desktop: could not attach to the input desktop. The " + "thread will stay on the worker's initial desktop and UIA " + "enumeration will reflect that desktop." + ) try: yield finally: if hdesk: _user32.SetThreadDesktop(hdesk_prev) - _user32.CloseDesktop(hdesk) + if own_hdesk: + _user32.CloseDesktop(hdesk) if hwinsta: _user32.SetProcessWindowStation(hwinsta_prev) _user32.CloseWindowStation(hwinsta) @@ -131,6 +204,7 @@ def _wrapper(): def _create_uia() -> tuple[Any, Any]: """Return (IUIAutomation, uia_core) — must be called on a thread with no prior COM init.""" import comtypes.client + ctypes.windll.ole32.CoInitialize(None) # STA — matches the existing user-mode UIA code uia_core = comtypes.client.GetModule("UIAutomationCore.dll") iuia = comtypes.client.CreateObject( @@ -172,9 +246,12 @@ def _serialize_element(element: Any, walker: Any, depth: int = 0) -> dict | None "name": name, "control_type": ctrl, "bbox": { - "left": rect.left, "top": rect.top, - "right": rect.right, "bottom": rect.bottom, - "width": w, "height": h, + "left": rect.left, + "top": rect.top, + "right": rect.right, + "bottom": rect.bottom, + "width": w, + "height": h, }, "center": {"x": rect.left + w // 2, "y": rect.top + h // 2}, "can_invoke": can_invoke, @@ -185,163 +262,1325 @@ def _serialize_element(element: Any, walker: Any, depth: int = 0) -> dict | None return None +def _tree_contains_consent(tree: list[dict], _pid: int) -> bool: + """Return True if the worker's serialized UIA tree contains a window that + looks like consent.exe's UAC dialog. We match on the dialog title + "User Account Control" rather than process id (the serializer doesn't + record pid). _pid is accepted for future use.""" + + target = "user account control" + + def _walk(node: dict) -> bool: + if not isinstance(node, dict): + return False + if target in (node.get("name") or "").lower(): + return True + for child in node.get("children") or []: + if _walk(child): + return True + return False + + return any(_walk(n) for n in tree) + + +def _find_consent_pid() -> int | None: + """Walk Toolhelp32Snapshot looking for ``consent.exe``. Returns the first + matching PID or ``None``. Used to confirm element identity from a + UIAccess worker that can't OpenDesktop('Winlogon') -- if an element's + CurrentProcessId matches consent.exe's PID, it is the UAC dialog. + """ + import ctypes + from ctypes import wintypes + + TH32CS_SNAPPROCESS = 0x00000002 + INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value + + class PROCESSENTRY32W(ctypes.Structure): + _fields_ = [ + ("dwSize", wintypes.DWORD), + ("cntUsage", wintypes.DWORD), + ("th32ProcessID", wintypes.DWORD), + ("th32DefaultHeapID", ctypes.c_size_t), + ("th32ModuleID", wintypes.DWORD), + ("cntThreads", wintypes.DWORD), + ("th32ParentProcessID", wintypes.DWORD), + ("pcPriClassBase", ctypes.c_long), + ("dwFlags", wintypes.DWORD), + ("szExeFile", wintypes.WCHAR * 260), + ] + + k32 = ctypes.windll.kernel32 + snap = k32.CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0) + if snap == INVALID_HANDLE_VALUE or snap == 0: + return None + try: + k32.Process32FirstW.argtypes = [ + wintypes.HANDLE, + ctypes.POINTER(PROCESSENTRY32W), + ] + k32.Process32FirstW.restype = wintypes.BOOL + k32.Process32NextW.argtypes = [ + wintypes.HANDLE, + ctypes.POINTER(PROCESSENTRY32W), + ] + k32.Process32NextW.restype = wintypes.BOOL + pe = PROCESSENTRY32W() + pe.dwSize = ctypes.sizeof(PROCESSENTRY32W) + ok = k32.Process32FirstW(snap, ctypes.byref(pe)) + while ok: + if (pe.szExeFile or "").lower() == "consent.exe": + return int(pe.th32ProcessID) + ok = k32.Process32NextW(snap, ctypes.byref(pe)) + finally: + k32.CloseHandle(snap) + return None + + # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- + def get_input_desktop_name() -> str: """Return the name of the current input desktop. Returns ``"Default"`` during normal desktop use and ``"Winlogon"`` while a - UAC prompt is active. Works from user-mode too (used for detection in the - broker via :func:`~windows_mcp.desktop.screenshot.is_secure_desktop_active`). + UAC prompt is on the secure desktop. Works from user-mode too. + + When called from a LocalSystem service the process window station is + ``Service-0x0-3e7$``, not ``WinSta0`` — and ``OpenInputDesktop`` on the + service winstation never returns the user's input desktop. We first try a + plain ``OpenInputDesktop`` (cheap, works from user mode) and fall back to + momentarily attaching to ``WinSta0`` when that returns nothing useful. """ hdesk = _open_input_desktop(_DESKTOP_READOBJECTS) - if not hdesk: + if hdesk: + try: + name = _get_desktop_name(hdesk) + finally: + _user32.CloseDesktop(hdesk) + if name: + return name + + hwinsta_prev = _user32.GetProcessWindowStation() + hwinsta = _open_winsta0() + if not hwinsta: return "Default" try: - return _get_desktop_name(hdesk) + _user32.SetProcessWindowStation(hwinsta) + hdesk = _open_input_desktop(_DESKTOP_READOBJECTS) + if not hdesk: + return "Default" + try: + return _get_desktop_name(hdesk) or "Default" + finally: + _user32.CloseDesktop(hdesk) finally: - _user32.CloseDesktop(hdesk) + _user32.SetProcessWindowStation(hwinsta_prev) + _user32.CloseWindowStation(hwinsta) -def capture_screenshot() -> bytes: - """Capture the current input desktop as PNG bytes. +def _send_tab_key() -> None: + """Send a single Tab keypress via SendInput on the current input desktop. - Uses GDI (Pillow ImageGrab) after SetThreadDesktop — DXGI is unavailable - from Session 0, but GDI BitBlt works once the thread is on the right desktop. + Used to provoke a UIA FocusChanged event from consent.exe: the worker + has TokenUIAccess=1, so UIPI permits SendInput to the higher-integrity + UAC dialog. Tab only moves focus between the Yes/No buttons -- it never + activates one -- so it is a safe nudge to make the dialog re-emit a + focus event that a registered handler can capture. """ - with _input_desktop(): - from PIL import ImageGrab - img = ImageGrab.grab(all_screens=True) - buf = io.BytesIO() - img.save(buf, format="PNG") - return buf.getvalue() - - -def uia_get_window_titles() -> list[str]: - """Return names of top-level windows on the current input desktop.""" - def _work() -> list[str]: - titles: list[str] = [] - with _input_desktop(): - iuia, _ = _create_uia() - root = iuia.GetRootElement() - walker = iuia.RawViewWalker - child = walker.GetFirstChildElement(root) - while child: + INPUT_KEYBOARD = 1 + KEYEVENTF_KEYUP = 0x0002 + VK_TAB = 0x09 + + ULONG_PTR = ctypes.c_size_t # pointer-sized, matches ULONG_PTR on x86/x64 + + class _KEYBDINPUT(ctypes.Structure): + _fields_ = [ + ("wVk", ctypes.wintypes.WORD), + ("wScan", ctypes.wintypes.WORD), + ("dwFlags", ctypes.wintypes.DWORD), + ("time", ctypes.wintypes.DWORD), + ("dwExtraInfo", ULONG_PTR), + ] + + class _MOUSEINPUT(ctypes.Structure): + # Largest union member -- present only so the union (and thus INPUT) + # gets the correct size. SendInput validates cbSize == sizeof(INPUT), + # which is 40 bytes on x64; without MOUSEINPUT the union shrinks to + # KEYBDINPUT and the call fails with ERROR_INVALID_PARAMETER. + _fields_ = [ + ("dx", ctypes.c_long), + ("dy", ctypes.c_long), + ("mouseData", ctypes.wintypes.DWORD), + ("dwFlags", ctypes.wintypes.DWORD), + ("time", ctypes.wintypes.DWORD), + ("dwExtraInfo", ULONG_PTR), + ] + + class _INPUT_UNION(ctypes.Union): + _fields_ = [("ki", _KEYBDINPUT), ("mi", _MOUSEINPUT)] + + class _INPUT(ctypes.Structure): + _fields_ = [("type", ctypes.wintypes.DWORD), ("u", _INPUT_UNION)] + + SendInput = _user32.SendInput + SendInput.argtypes = [ctypes.wintypes.UINT, ctypes.POINTER(_INPUT), ctypes.c_int] + SendInput.restype = ctypes.wintypes.UINT + + down = _INPUT(type=INPUT_KEYBOARD, u=_INPUT_UNION(ki=_KEYBDINPUT(VK_TAB, 0, 0, 0, 0))) + up = _INPUT( + type=INPUT_KEYBOARD, + u=_INPUT_UNION(ki=_KEYBDINPUT(VK_TAB, 0, KEYEVENTF_KEYUP, 0, 0)), + ) + arr = (_INPUT * 2)(down, up) + sent = SendInput(2, arr, ctypes.sizeof(_INPUT)) + logger.info( + "_send_tab_key: SendInput sent=%d gle=%d sizeof(INPUT)=%d", + sent, ctypes.GetLastError(), ctypes.sizeof(_INPUT), + ) + + +def _send_mouse_click(x: int, y: int) -> bool: + """Synthesize an absolute left-click at (x, y) via SendInput. + + consent.exe renders its Yes/No buttons as XAML/Composition that may not + expose a UIA InvokePattern, so ``uia_click_at``'s programmatic invoke + can't activate them. But a physical mouse click does -- and because the + worker holds TokenUIAccess=1, UIPI exempts its SendInput from the + medium->System integrity block, so the click reaches the higher- + integrity UAC dialog. Coordinates are absolute virtual-screen pixels, + normalised to the 0..65535 range SendInput expects. + + Returns True if SendInput accepted all three events (move, down, up). + """ + INPUT_MOUSE = 0 + MOUSEEVENTF_MOVE = 0x0001 + MOUSEEVENTF_ABSOLUTE = 0x8000 + MOUSEEVENTF_VIRTUALDESK = 0x4000 + MOUSEEVENTF_LEFTDOWN = 0x0002 + MOUSEEVENTF_LEFTUP = 0x0004 + SM_XVIRTUALSCREEN = 76 + SM_YVIRTUALSCREEN = 77 + SM_CXVIRTUALSCREEN = 78 + SM_CYVIRTUALSCREEN = 79 + + ULONG_PTR = ctypes.c_size_t + + class _KEYBDINPUT(ctypes.Structure): + _fields_ = [ + ("wVk", ctypes.wintypes.WORD), + ("wScan", ctypes.wintypes.WORD), + ("dwFlags", ctypes.wintypes.DWORD), + ("time", ctypes.wintypes.DWORD), + ("dwExtraInfo", ULONG_PTR), + ] + + class _MOUSEINPUT(ctypes.Structure): + _fields_ = [ + ("dx", ctypes.c_long), + ("dy", ctypes.c_long), + ("mouseData", ctypes.wintypes.DWORD), + ("dwFlags", ctypes.wintypes.DWORD), + ("time", ctypes.wintypes.DWORD), + ("dwExtraInfo", ULONG_PTR), + ] + + class _INPUT_UNION(ctypes.Union): + _fields_ = [("mi", _MOUSEINPUT), ("ki", _KEYBDINPUT)] + + class _INPUT(ctypes.Structure): + _fields_ = [("type", ctypes.wintypes.DWORD), ("u", _INPUT_UNION)] + + # Normalise absolute pixel (x, y) into the 0..65535 range across the + # whole virtual desktop (all monitors), matching MOUSEEVENTF_VIRTUALDESK. + vx = _user32.GetSystemMetrics(SM_XVIRTUALSCREEN) + vy = _user32.GetSystemMetrics(SM_YVIRTUALSCREEN) + vw = _user32.GetSystemMetrics(SM_CXVIRTUALSCREEN) or 1 + vh = _user32.GetSystemMetrics(SM_CYVIRTUALSCREEN) or 1 + nx = int(round((x - vx) * 65535 / vw)) + ny = int(round((y - vy) * 65535 / vh)) + + SendInput = _user32.SendInput + SendInput.argtypes = [ctypes.wintypes.UINT, ctypes.POINTER(_INPUT), ctypes.c_int] + SendInput.restype = ctypes.wintypes.UINT + + move_flags = MOUSEEVENTF_MOVE | MOUSEEVENTF_ABSOLUTE | MOUSEEVENTF_VIRTUALDESK + move = _INPUT(type=INPUT_MOUSE, u=_INPUT_UNION(mi=_MOUSEINPUT(nx, ny, 0, move_flags, 0, 0))) + down = _INPUT( + type=INPUT_MOUSE, + u=_INPUT_UNION(mi=_MOUSEINPUT(nx, ny, 0, MOUSEEVENTF_LEFTDOWN | MOUSEEVENTF_ABSOLUTE | MOUSEEVENTF_VIRTUALDESK, 0, 0)), + ) + up = _INPUT( + type=INPUT_MOUSE, + u=_INPUT_UNION(mi=_MOUSEINPUT(nx, ny, 0, MOUSEEVENTF_LEFTUP | MOUSEEVENTF_ABSOLUTE | MOUSEEVENTF_VIRTUALDESK, 0, 0)), + ) + arr = (_INPUT * 3)(move, down, up) + sent = SendInput(3, arr, ctypes.sizeof(_INPUT)) + logger.info( + "_send_mouse_click(%d,%d)->norm(%d,%d): SendInput sent=%d gle=%d", + x, y, nx, ny, sent, ctypes.GetLastError(), + ) + return sent == 3 + + +def _capture_consent_via_focus_events(iuia, uia_core, walker, consent_pid, wait_ms=3000): + """Strategy A: subscribe to UIA FocusChanged events and capture the + consent.exe element when one fires. + + Win 11 25H2 blocks every *pull* path to consent.exe (root walk, + FindAll(ProcessId), ElementFromHandle gives only the frame). But UIA + event delivery is *push* and is what Narrator/Magnifier rely on for + cross-integrity access. A UIAccess process receives focus events from + higher-integrity processes. We register a no-filter FocusChanged + handler, nudge the dialog with a Tab keypress to force a fresh focus + event, pump the STA message queue, and capture any sender whose + ProcessId matches consent.exe. Returns the serialized dialog-root node + or None. + """ + import comtypes + import pythoncom + + cap_lock = threading.Lock() + captured: list[Any] = [] + cap_event = threading.Event() + + class _FocusHandler(comtypes.COMObject): + _com_interfaces_ = [uia_core.IUIAutomationFocusChangedEventHandler] + + def HandleFocusChangedEvent(self, sender): + try: + spid = sender.CurrentProcessId if sender is not None else None + if spid == consent_pid: + with cap_lock: + captured.append(sender) + cap_event.set() + except Exception: + pass + return 0 + + handler = _FocusHandler() + registered = False + try: + iuia.AddFocusChangedEventHandler(None, handler) + registered = True + logger.info("focus-events: AddFocusChangedEventHandler registered") + except Exception as exc: # noqa: BLE001 + logger.warning("focus-events: AddFocusChangedEventHandler failed: %s", exc) + return None + + try: + # Nudge the dialog so it emits a fresh focus event. + try: + _send_tab_key() + except Exception as exc: # noqa: BLE001 + logger.warning("focus-events: _send_tab_key failed: %s", exc) + + deadline = time.monotonic() + (wait_ms / 1000.0) + nudges = 0 + while time.monotonic() < deadline: + pythoncom.PumpWaitingMessages() + if cap_event.wait(timeout=0.05): + break + # Re-nudge every ~1s in case the first Tab landed before the + # handler was fully wired through the COM marshaller. + nudges += 1 + if nudges % 20 == 0: try: - name = child.CurrentName - if name: - titles.append(name) + _send_tab_key() except Exception: pass - try: - child = walker.GetNextSiblingElement(child) - except Exception: - break - return titles - try: - return _run_on_fresh_thread(_work) or [] - except Exception as exc: - logger.warning("uia_get_window_titles failed: %s", exc) + with cap_lock: + sender = captured[0] if captured else None + if sender is None: + logger.info("focus-events: no consent.exe focus event captured") + return None + # Walk up to the dialog root so Yes AND No are both descendants. + cur = sender + for _ in range(8): + try: + parent = walker.GetParentElement(cur) + except Exception: + break + if parent is None: + break + try: + ppid = parent.CurrentProcessId + except Exception: + ppid = None + if ppid != consent_pid: + break + cur = parent + node = _serialize_element(cur, walker) + if node: + logger.info("focus-events: captured + serialized consent dialog") + return node + finally: + if registered: + try: + iuia.RemoveFocusChangedEventHandler(handler) + except Exception: + pass + + +def _find_visible_hwnds_for_pid(pid: int) -> list[int]: + """Return every TOP-LEVEL + CHILD HWND owned by *pid* on the current + desktop, regardless of visibility flag. + + consent.exe's dialog content (the Yes/No buttons, the prompt text, + publisher info) is rendered into a CHILD window of the frame HWND -- + Win32 EnumWindows returns the top-level frame but not its children. + We use EnumWindows + EnumChildWindows together to get every HWND + consent.exe owns, then walk each via ElementFromHandle. + """ + EnumWindows = _user32.EnumWindows + EnumWindows.argtypes = [ctypes.c_void_p, ctypes.wintypes.LPARAM] + EnumWindows.restype = ctypes.wintypes.BOOL + EnumChildWindows = _user32.EnumChildWindows + EnumChildWindows.argtypes = [ctypes.wintypes.HWND, ctypes.c_void_p, ctypes.wintypes.LPARAM] + EnumChildWindows.restype = ctypes.wintypes.BOOL + GetWindowThreadProcessId = _user32.GetWindowThreadProcessId + GetWindowThreadProcessId.argtypes = [ + ctypes.wintypes.HWND, + ctypes.POINTER(ctypes.wintypes.DWORD), + ] + GetWindowThreadProcessId.restype = ctypes.wintypes.DWORD + + WNDENUMPROC = ctypes.WINFUNCTYPE( + ctypes.wintypes.BOOL, + ctypes.wintypes.HWND, + ctypes.wintypes.LPARAM, + ) + top: list[int] = [] + children: list[int] = [] + + @WNDENUMPROC + def _on_top(hwnd, _lparam): + owner = ctypes.wintypes.DWORD(0) + GetWindowThreadProcessId(hwnd, ctypes.byref(owner)) + if owner.value == pid: + top.append(int(hwnd)) + return True + + @WNDENUMPROC + def _on_child(hwnd, _lparam): + owner = ctypes.wintypes.DWORD(0) + GetWindowThreadProcessId(hwnd, ctypes.byref(owner)) + if owner.value == pid: + children.append(int(hwnd)) + return True + + EnumWindows(ctypes.cast(_on_top, ctypes.c_void_p), 0) + for parent_hwnd in top: + EnumChildWindows(parent_hwnd, ctypes.cast(_on_child, ctypes.c_void_p), 0) + # De-duplicate while preserving order (top first, then children). + seen: set[int] = set() + out: list[int] = [] + for h in top + children: + if h not in seen: + seen.add(h) + out.append(h) + return out + + +def _find_top_hwnd_for_pid(pid: int) -> int: + """Legacy helper retained for the publisher walker. Returns the first + visible top-level HWND, or 0.""" + hwnds = _find_visible_hwnds_for_pid(pid) + return hwnds[0] if hwnds else 0 + + +def _synthesize_uac_buttons(dialog_bbox: dict) -> list[dict]: + """Build synthesized Yes/No button nodes from the dialog bounding box. + + The Win 11 consent.exe dialog renders its action buttons as XAML inside + a Composition layer that the UIA tree walker can't descend into across + the integrity boundary. The buttons ARE clickable -- ElementFromPoint + at click time resolves them across integrity. So we synthesize nodes + with predicted center coords and ``can_invoke=True`` so the LLM's + button-finding pass (name + can_invoke) succeeds and the subsequent + Click(loc=[cx, cy]) lands on the real XAML element. + + The layout for the modern Win 11 consent dialog (observed at 1280x720): + dialog bbox left=420 right=860 top=178 bottom=535 (440x357) + Yes button center ~ (538, 507) -> 0.27 from left, 28px from bottom + No button center ~ (745, 507) -> 0.74 from left, 28px from bottom + The same relative positions hold for the taller "expanded details" + layout because the Yes/No strip is anchored to the dialog bottom. + """ + left = dialog_bbox.get("left", 0) + top = dialog_bbox.get("top", 0) + right = dialog_bbox.get("right", 0) + bottom = dialog_bbox.get("bottom", 0) + if right <= left or bottom <= top: return [] + w = right - left + yes_cx = left + int(w * 0.27) + no_cx = left + int(w * 0.74) + cy = bottom - 28 + def _btn(name: str, cx: int, cy: int) -> dict: + # 80x32 hit-rect approximation centered on (cx, cy). + return { + "name": name, + "control_type": "button", + "bbox": { + "left": cx - 40, "top": cy - 16, + "right": cx + 40, "bottom": cy + 16, + "width": 80, "height": 32, + }, + "center": {"x": cx, "y": cy}, + "can_invoke": True, + "children": [], + "_synthesized": True, + } + + return [_btn("Yes", yes_cx, cy), _btn("No", no_cx, cy)] + + +def _ensure_uac_buttons(node: dict | None) -> None: + """Attach synthesized Yes/No buttons to a childless UAC dialog window node. -def uia_get_tree() -> list[dict]: + consent.exe renders its Yes/No buttons as XAML/Composition the UIA walker + can't descend into cross-integrity, so *whichever* strategy locates the + dialog window (FocusChanged event, GetFocusedElement walk-up, FindAll, or + ElementFromHandle) ends up with a window node that has no walkable + children. Apply this to every such return path so downstream callers can + always find the buttons by name + center coords; the Click tool resolves + the real XAML element via ElementFromPoint at click time. + """ + if ( + node + and not node.get("children") + and "Account Control" in (node.get("name") or "") + ): + node["children"] = _synthesize_uac_buttons(node.get("bbox") or {}) + + +def uia_get_tree(consent_pid: int = 0) -> list[dict]: """Return the full UIA tree of the current input desktop. - Each entry is a top-level window serialized as a nested dict. Elements with - ``can_invoke=True`` support ``IUIAutomationInvokePattern`` — the broker uses - this to identify clickable buttons (Yes/No on a UAC dialog) without - re-walking the tree. + Each entry is a top-level window serialized as a nested dict. Elements + with ``can_invoke=True`` support ``IUIAutomationInvokePattern`` — the + broker uses this to identify clickable buttons (Yes/No on a UAC dialog) + without re-walking the tree. + + If *consent_pid* is non-zero, find that pid's top HWND via EnumWindows + and walk from ``ElementFromHandle(hwnd)`` instead of the desktop root. - Runs on a fresh thread so COM initialises *after* SetThreadDesktop, binding - IUIAutomation to the correct desktop (Winlogon during UAC). + The first node may carry a ``_diag`` key with the lookup outcome + (``hwnd=...``, fallback reason) so the broker / client can see what + happened without needing access to the SYSTEM broker log. """ + def _work() -> list[dict]: nodes: list[dict] = [] + diag_lines: list[str] = [] with _input_desktop(): - iuia, _ = _create_uia() - root = iuia.GetRootElement() + iuia, uia_core = _create_uia() walker = iuia.RawViewWalker + if consent_pid: + # Win 11 25H2 blocks every UIA *pull* path to consent.exe + # (System integrity): root walk, FindAll(ProcessId)=0, + # ElementFromHandle returns only the OS frame. UIA event + # delivery is *push* and is what Narrator/Magnifier use for + # cross-integrity access -- try that first (strategy A). Keep + # the wait short: in practice consent.exe rarely delivers a + # focus event to the UIAccess worker, so a long wait mostly + # just burns the per-spawn budget before the more reliable + # ElementFromHandle path gets to run. + try: + focus_node = _capture_consent_via_focus_events( + iuia, uia_core, walker, consent_pid, wait_ms=2000 + ) + if focus_node: + _ensure_uac_buttons(focus_node) + focus_node["_diag"] = ( + f"consent_pid={consent_pid} | path=FocusEvent" + ) + nodes.append(focus_node) + return nodes + diag_lines.append(f"consent_pid={consent_pid} FocusEvent=empty") + except Exception as exc: # noqa: BLE001 + diag_lines.append(f"FocusEvent raised: {exc}") + logger.warning("uia_get_tree: focus-event strategy failed: %s", exc) + + # GetFocusedElement is a cheap pull-path retry now that a Tab + # nudge may have moved focus onto a consent button. + UIA_ProcessIdPropertyId = 30005 - 3 # 30002 + try: + focused = iuia.GetFocusedElement() + fpid = None + try: + fpid = focused.CurrentProcessId if focused is not None else None + except Exception: + fpid = None + diag_lines.append( + f"consent_pid={consent_pid} GetFocusedElement pid={fpid}" + ) + if focused is not None and fpid == consent_pid: + # Walk up to the dialog root so the serializer + # captures both Yes and No as descendants. + cur = focused + for _ in range(8): + try: + parent = walker.GetParentElement(cur) + except Exception: + break + if parent is None: + break + try: + ppid = parent.CurrentProcessId + except Exception: + ppid = None + if ppid != consent_pid: + break + cur = parent + node = _serialize_element(cur, walker) + if node: + _ensure_uac_buttons(node) + node["_diag"] = " | ".join(diag_lines) + " | path=GetFocusedElement+walk-up" + nodes.append(node) + return nodes + else: + diag_lines.append("focus walk-up: _serialize_element returned None") + except Exception as exc: # noqa: BLE001 + diag_lines.append(f"GetFocusedElement raised: {exc}") + logger.warning("uia_get_tree: GetFocusedElement failed: %s", exc) + # Then try FindAll(ProcessId) as a backup + try: + pid_cond = iuia.CreatePropertyCondition(UIA_ProcessIdPropertyId, consent_pid) + matches = iuia.GetRootElement().FindAll( + _UIA_TreeScope_Descendants, pid_cond + ) + length = matches.Length if matches is not None else 0 + diag_lines.append(f"FindAll(ProcessId)={length}") + if length > 0: + for i in range(length): + elem = matches.GetElement(i) + if elem is None: + continue + node = _serialize_element(elem, walker) + if node: + _ensure_uac_buttons(node) + if not nodes: + node["_diag"] = " | ".join(diag_lines) + " | path=FindAll(ProcessId)" + nodes.append(node) + if nodes: + return nodes + except Exception as exc: # noqa: BLE001 + diag_lines.append(f"FindAll(ProcessId) raised: {exc}") + logger.warning("uia_get_tree: FindAll(ProcessId) failed: %s", exc) + # Fallback to the HWND walk in case FindAll returned 0 (e.g. + # consent.exe registered nothing or the call was denied). + hwnds = _find_visible_hwnds_for_pid(consent_pid) + diag_lines.append( + f"hwnds=[{','.join(f'0x{h:x}' for h in hwnds)}]" + ) + walked_any = False + for hwnd in hwnds: + try: + elem = iuia.ElementFromHandle(hwnd) + if elem is None: + diag_lines.append(f"ElementFromHandle(0x{hwnd:x}) None") + continue + node = _serialize_element(elem, walker) + if node: + _ensure_uac_buttons(node) + if node.get("children"): + diag_lines.append( + f"synthesized_yes_no={len(node['children'])} from bbox={node['bbox']}" + ) + if not nodes: + node["_diag"] = " | ".join(diag_lines) + f" | path=ElementFromHandle(0x{hwnd:x})" + nodes.append(node) + walked_any = True + except Exception as exc: # noqa: BLE001 + diag_lines.append(f"ElementFromHandle(0x{hwnd:x}) raised: {exc}") + logger.warning( + "uia_get_tree: ElementFromHandle(0x%x) failed: %s", hwnd, exc + ) + if walked_any: + return nodes + # Every scoped strategy failed to reach consent.exe. Do NOT + # fall through to the full-desktop root walk below: that walks + # and serializes every top-level window on the Default desktop + # (recursively), which is pointless here (consent.exe isn't in + # it) and slow enough on a TCG VM to blow the thread timeout, + # turning a clean "not found" into an empty result. Return a + # diagnostic placeholder immediately so the host can retry. + return [{ + "name": "(consent-not-reached)", + "control_type": "", + "bbox": {"left": 0, "top": 0, "right": 0, "bottom": 0, "width": 0, "height": 0}, + "center": {"x": 0, "y": 0}, + "can_invoke": False, + "children": [], + "_diag": " | ".join(diag_lines) + " | path=consent-scoped-empty", + }] + root = iuia.GetRootElement() child = walker.GetFirstChildElement(root) + first = True while child: node = _serialize_element(child, walker) if node: + if first and diag_lines: + node["_diag"] = " | ".join(diag_lines) + " | path=desktop-root" + first = False nodes.append(node) try: child = walker.GetNextSiblingElement(child) except Exception: break + if not nodes and diag_lines: + # Walker returned nothing at all -- bubble the diag up as a + # placeholder node so the caller can see why. + nodes.append({ + "name": "(empty)", + "control_type": "", + "bbox": {"left": 0, "top": 0, "right": 0, "bottom": 0, "width": 0, "height": 0}, + "center": {"x": 0, "y": 0}, + "can_invoke": False, + "children": [], + "_diag": " | ".join(diag_lines) + " | path=walker-empty", + }) return nodes try: - return _run_on_fresh_thread(_work) or [] + # The default 15s is too short once we're doing the multi-strategy + # consent walk (a ~4s FocusChanged wait plus several cross-integrity + # COM calls) on a slow TCG VM -- the thread would time out and return + # None, which surfaces as an empty tree. Give it room (kept below the + # broker's per-spawn timeout so the broker, not this thread, owns the + # hard deadline). + thread_timeout = 40.0 if consent_pid else 15.0 + return _run_on_fresh_thread(_work, timeout=thread_timeout) or [] except Exception as exc: logger.error("uia_get_tree failed: %s", exc) return [] -def uia_invoke_element(name: str) -> bool: - """Find a named element on the input desktop and invoke it via UIA. +def uia_click_at(x: int, y: int) -> bool: + """Invoke the element at (x, y) on the input desktop via UIA ElementFromPoint. - Uses ``IUIAutomation.FindFirst`` + ``IUIAutomationInvokePattern.Invoke()``. - Direct COM call — no input injection needed, works from Session 0. - Runs on a fresh thread so COM binds to the Winlogon desktop. + Callers can pass coordinates straight from the screenshot. Runs on a fresh + thread so COM binds to the correct (Winlogon) desktop. """ + def _work() -> bool: with _input_desktop(): iuia, uia_core = _create_uia() - root = iuia.GetRootElement() - condition = iuia.CreatePropertyCondition(_UIA_NamePropertyId, name) - element = root.FindFirst(_UIA_TreeScope_Descendants, condition) - if element is None: - logger.warning("uia_invoke_element: no element named %r", name) - return False - pattern = element.GetCurrentPattern(_UIA_InvokePatternId) - if pattern is None: - logger.warning("uia_invoke_element: %r has no InvokePattern", name) - return False - invoke = pattern.QueryInterface(uia_core.IUIAutomationInvokePattern) - invoke.Invoke() - logger.info("uia_invoke_element: invoked %r", name) - return True + # ElementFromPoint wants the comtypes-generated POINT type; pass a + # plain tuple (comtypes marshals it) rather than a hand-rolled + # ctypes struct, which comtypes rejects. If the point query fails + # for any reason just fall straight through to the SendInput path -- + # for the consent.exe XAML buttons (the reason this routes here at + # all) there is no InvokePattern to use anyway. + element = None + try: + element = iuia.ElementFromPoint((x, y)) + except Exception as exc: # noqa: BLE001 + logger.info( + "uia_click_at(%d,%d): ElementFromPoint failed (%s); using SendInput", + x, y, exc, + ) + pattern = None + if element is not None: + try: + pattern = element.GetCurrentPattern(_UIA_InvokePatternId) + except Exception: # noqa: BLE001 + pattern = None + if pattern is not None: + invoke = pattern.QueryInterface(uia_core.IUIAutomationInvokePattern) + invoke.Invoke() + logger.info("uia_click_at(%d,%d): invoked %r via InvokePattern", x, y, element.CurrentName) + return True + # No InvokePattern -- consent.exe's XAML Yes/No buttons expose no + # UIA pattern the walker/point-query can activate. Fall back to a + # synthesized physical click; the worker's UIAccess token exempts + # its SendInput from UIPI, so it reaches the System-integrity UAC + # dialog. + logger.info( + "uia_click_at(%d,%d): no InvokePattern (element=%r) -- falling back to SendInput click", + x, y, (element.CurrentName if element is not None else None), + ) + return _send_mouse_click(x, y) try: return _run_on_fresh_thread(_work) or False except Exception as exc: - logger.error("uia_invoke_element(%r) failed: %s", name, exc) + logger.error("uia_click_at(%d,%d) failed: %s", x, y, exc) return False -def uia_click_at(x: int, y: int) -> bool: - """Invoke the element at (x, y) on the input desktop via UIA ElementFromPoint. +# --------------------------------------------------------------------------- +# UAC dialog inspection +# --------------------------------------------------------------------------- - Callers can pass coordinates straight from the screenshot. Runs on a fresh - thread so COM binds to the correct (Winlogon) desktop. +# Patterns the Windows UAC dialog uses for its "verified publisher" line. +# These are localised on non-English Windows; if no pattern matches we return None +# and the allow_with_match policy refuses on caller side. +_PUBLISHER_PATTERNS = [ + re.compile(r"Verified publisher:\s*(.+)", re.IGNORECASE), + re.compile(r"Program name:\s*(.+)", re.IGNORECASE), + re.compile(r"Publisher:\s*(.+)", re.IGNORECASE), +] + + +def get_uac_publisher(consent_pid: int = 0) -> str | None: + """Inspect the active UAC consent dialog and return its publisher string. + + Returns ``None`` if no UAC dialog is currently displayed, if its layout does + not match the expected English pattern, or if reading the UIA tree fails. + + When *consent_pid* is non-zero, scope the walk to that pid's top-level + HWND via ElementFromHandle -- same rationale as uia_get_tree. """ - class _POINT(ctypes.Structure): - _fields_ = [("x", ctypes.c_long), ("y", ctypes.c_long)] - def _work() -> bool: + def _work() -> str | None: with _input_desktop(): - iuia, uia_core = _create_uia() - element = iuia.ElementFromPoint(_POINT(x, y)) - if element is None: - logger.warning("uia_click_at(%d,%d): no element found", x, y) - return False - pattern = element.GetCurrentPattern(_UIA_InvokePatternId) - if pattern is None: - logger.warning("uia_click_at(%d,%d): no InvokePattern", x, y) - return False - invoke = pattern.QueryInterface(uia_core.IUIAutomationInvokePattern) - invoke.Invoke() - logger.info("uia_click_at(%d,%d): invoked %r", x, y, element.CurrentName) - return True + iuia, _ = _create_uia() + walker = iuia.RawViewWalker + collected: list[str] = [] + + def _collect(elem: Any, depth: int = 0) -> None: + if depth > 8: + return + try: + name = elem.CurrentName or "" + if name: + collected.append(name) + except Exception: + return + try: + child = walker.GetFirstChildElement(elem) + while child: + _collect(child, depth + 1) + try: + child = walker.GetNextSiblingElement(child) + except Exception: + break + except Exception: + pass + + roots: list[Any] = [] + if consent_pid: + hwnd = _find_top_hwnd_for_pid(consent_pid) + if hwnd: + try: + elem = iuia.ElementFromHandle(hwnd) + if elem is not None: + roots.append(elem) + except Exception as exc: # noqa: BLE001 + logger.warning( + "get_uac_publisher: ElementFromHandle(0x%x) failed: %s", + hwnd, exc, + ) + if not roots: + root = iuia.GetRootElement() + child = walker.GetFirstChildElement(root) + while child: + roots.append(child) + try: + child = walker.GetNextSiblingElement(child) + except Exception: + break + + for r in roots: + _collect(r) + + text = "\n".join(collected) + for pat in _PUBLISHER_PATTERNS: + match = pat.search(text) + if match: + return match.group(1).strip() + return None try: - return _run_on_fresh_thread(_work) or False + return _run_on_fresh_thread(_work) except Exception as exc: - logger.error("uia_click_at(%d,%d) failed: %s", x, y, exc) - return False + logger.warning("get_uac_publisher failed: %s", exc) + return None + + +# --------------------------------------------------------------------------- +# User-session worker spawn +# --------------------------------------------------------------------------- + + +def _spawn_in_user_session(*op_args: str, timeout: float = 30.0) -> Any: + """Run one ``user_session_worker`` op inside the active console user's session. + + Session 0 isolation blocks the LocalSystem service from walking UIA trees + owned by user-session processes (consent.exe is the case that matters for + UAC). We side-step that by ``CreateProcessAsUser``-ing a one-shot helper + into the interactive session — UIA from inside the user's session sees + Winlogon normally — and parse the JSON it writes to stdout. + + Uses the user's *linked* elevated token when available so the helper has + enough access to enumerate consent.exe; falls back to the standard user + token otherwise. + """ + import pywintypes + import win32api + import win32con + import win32event + import win32file + import win32pipe + import win32process + import win32profile # CreateEnvironmentBlock lives here, not in win32process + import win32security + import win32ts + + session_id = win32ts.WTSGetActiveConsoleSessionId() + if session_id in (0xFFFFFFFF, 0): + raise RuntimeError( + "no interactive console session is active " + "(WTSGetActiveConsoleSessionId returned no user session)" + ) + + user_token = win32ts.WTSQueryUserToken(session_id) + elevated_token = None + try: + elevated_token = win32security.GetTokenInformation( + user_token, win32security.TokenLinkedToken + ) + except Exception: + elevated_token = None + spawn_token = elevated_token or user_token + using_elevated = bool(elevated_token) + + # Enable SeTcbPrivilege on the broker's process token. SYSTEM has it, + # but it isn't enabled by default. SetTokenInformation(TokenUIAccess) + # requires this privilege to be ENABLED on the caller, not just held. + try: + TOKEN_ADJUST_PRIVILEGES = 0x0020 + TOKEN_QUERY = 0x0008 + SE_PRIVILEGE_ENABLED = 0x00000002 + + # Declare argtypes so ctypes doesn't truncate the GetCurrentProcess + # pseudo-handle (-1) into ERROR_INVALID_HANDLE on 64-bit. + kernel32 = ctypes.windll.kernel32 + advapi32 = ctypes.windll.advapi32 + kernel32.GetCurrentProcess.restype = ctypes.wintypes.HANDLE + advapi32.OpenProcessToken.argtypes = [ + ctypes.wintypes.HANDLE, + ctypes.wintypes.DWORD, + ctypes.POINTER(ctypes.wintypes.HANDLE), + ] + advapi32.OpenProcessToken.restype = ctypes.wintypes.BOOL + advapi32.LookupPrivilegeValueW.argtypes = [ + ctypes.wintypes.LPCWSTR, + ctypes.wintypes.LPCWSTR, + ctypes.POINTER(ctypes.c_uint64), + ] + advapi32.LookupPrivilegeValueW.restype = ctypes.wintypes.BOOL + advapi32.AdjustTokenPrivileges.argtypes = [ + ctypes.wintypes.HANDLE, + ctypes.wintypes.BOOL, + ctypes.c_void_p, + ctypes.wintypes.DWORD, + ctypes.c_void_p, + ctypes.c_void_p, + ] + advapi32.AdjustTokenPrivileges.restype = ctypes.wintypes.BOOL + + h_proc_token = ctypes.wintypes.HANDLE() + ok = advapi32.OpenProcessToken( + kernel32.GetCurrentProcess(), + TOKEN_ADJUST_PRIVILEGES | TOKEN_QUERY, + ctypes.byref(h_proc_token), + ) + if ok: + luid = ctypes.c_uint64(0) + if advapi32.LookupPrivilegeValueW(None, "SeTcbPrivilege", ctypes.byref(luid)): + # TOKEN_PRIVILEGES { DWORD count; LUID_AND_ATTRIBUTES privs[1]; } + # LUID_AND_ATTRIBUTES { LUID(8 bytes); DWORD attrs; } + tp_buf = (ctypes.c_uint32 * 4)() + tp_buf[0] = 1 # PrivilegeCount + tp_buf[1] = luid.value & 0xFFFFFFFF # LUID.LowPart + tp_buf[2] = (luid.value >> 32) & 0xFFFFFFFF # LUID.HighPart + tp_buf[3] = SE_PRIVILEGE_ENABLED # Attributes + ok2 = advapi32.AdjustTokenPrivileges( + h_proc_token, False, ctypes.byref(tp_buf), 0, None, None + ) + gle = ctypes.GetLastError() if not ok2 else 0 + logger.info( + "AdjustTokenPrivileges(SeTcbPrivilege=ENABLED) ok=%s gle=%d", + bool(ok2), + gle, + ) + else: + logger.warning( + "LookupPrivilegeValueW(SeTcbPrivilege) failed gle=%d", ctypes.GetLastError() + ) + kernel32.CloseHandle(h_proc_token) + else: + logger.warning("OpenProcessToken failed gle=%d", ctypes.GetLastError()) + except Exception as exc: + logger.warning("SeTcbPrivilege enable raised: %s", exc) + + # Set TokenUIAccess=1 on the token *before* CreateProcessAsUser. Without + # this the spawned process always boots with TokenUIAccess=0 — Windows + # only checks the manifest's uiAccess attribute as a *request*, the + # privilege itself comes from this flag on the primary token, and + # CreateProcessAsUser does not set it for us based on the exe's manifest. + # SetTokenInformation(TokenUIAccess) from a SYSTEM caller with + # SeTcbPrivilege enabled bypasses the signature + trusted-path checks + # AppInfo normally enforces; see https://learn.microsoft.com/en-us/answers/questions/1009084/ + # and Tyranid's notes at https://www.tiraniddo.dev/2019/02/ + try: + TOKEN_UI_ACCESS = 26 + # Declare argtypes -- without them, ctypes treats the HANDLE arg as + # c_int (4 bytes) and truncates the high 32 bits of the 8-byte + # PyHANDLE. The call then operates on a corrupted handle (which on + # this box happened to "succeed" -- gle=0, ok=1 -- presumably because + # the truncated value collided with some other open handle), so the + # real spawn token never gets its UIAccess bit set and the spawned + # worker boots with TokenUIAccess=0. + advapi32.SetTokenInformation.argtypes = [ + ctypes.wintypes.HANDLE, + ctypes.c_int, # TOKEN_INFORMATION_CLASS enum + ctypes.c_void_p, # LPVOID TokenInformation + ctypes.wintypes.DWORD, # TokenInformationLength + ] + advapi32.SetTokenInformation.restype = ctypes.wintypes.BOOL + + ui_access = ctypes.c_uint32(1) + ok = advapi32.SetTokenInformation( + int(spawn_token), + TOKEN_UI_ACCESS, + ctypes.cast(ctypes.byref(ui_access), ctypes.c_void_p), + ctypes.sizeof(ui_access), + ) + if not ok: + gle = ctypes.GetLastError() + logger.warning( + "SetTokenInformation(TokenUIAccess=1) failed (gle=%d) - " + "worker will spawn without UIAccess and won't be able to " + "walk Winlogon", + gle, + ) + else: + logger.info( + "SetTokenInformation(TokenUIAccess=1) on spawn token (handle=0x%x) OK", + int(spawn_token), + ) + except Exception as exc: + logger.warning("SetTokenInformation(TokenUIAccess) raised: %s", exc) + + sa = win32security.SECURITY_ATTRIBUTES() + sa.bInheritHandle = True + stdout_r, stdout_w = win32pipe.CreatePipe(sa, 0) + stderr_r, stderr_w = win32pipe.CreatePipe(sa, 0) + stdin_r, stdin_w = win32pipe.CreatePipe(sa, 0) + # Read ends stay in the service; do not let them leak into the child. + win32api.SetHandleInformation(stdout_r, win32con.HANDLE_FLAG_INHERIT, 0) + win32api.SetHandleInformation(stderr_r, win32con.HANDLE_FLAG_INHERIT, 0) + # Worker reads stdin; the broker's write end must stay non-inheritable. + win32api.SetHandleInformation(stdin_w, win32con.HANDLE_FLAG_INHERIT, 0) + + argv = [ + sys.executable, + "-m", + "windows_mcp.service.user_session_worker", + *op_args, + ] + cmd_line = subprocess.list2cmdline(argv) + + startup = win32process.STARTUPINFO() + startup.dwFlags = win32con.STARTF_USESTDHANDLES + startup.hStdInput = stdin_r + startup.hStdOutput = stdout_w + startup.hStdError = stderr_w + # Spawn on the interactive Default desktop. Tried lpDesktop="winsta0\winlogon" + # for read ops: even with TokenUIAccess=1 + signed worker in + # %ProgramFiles%, the OS rejects user32.dll init against Winlogon's + # DACL and the worker crashes with STATUS_DLL_INIT_FAILED + # (exit=-1073741502). The worker's _input_desktop later attaches to + # whichever desktop it can. + startup.lpDesktop = r"winsta0\default" + + user_env = win32profile.CreateEnvironmentBlock(spawn_token, False) + + creation_flags = win32con.CREATE_NO_WINDOW | win32process.CREATE_UNICODE_ENVIRONMENT + + proc_handle = thread_handle = None + try: + proc_info = win32process.CreateProcessAsUser( + spawn_token, + None, + cmd_line, + None, + None, + True, + creation_flags, + user_env, + None, + startup, + ) + proc_handle, thread_handle, _pid, _tid = proc_info + finally: + # Now that the child has inherited the write ends we can drop ours. + try: + win32file.CloseHandle(stdout_w) + except Exception: + pass + try: + win32file.CloseHandle(stderr_w) + except Exception: + pass + try: + win32file.CloseHandle(stdin_r) + except Exception: + pass + try: + win32api.CloseHandle(user_token) + except Exception: + pass + if elevated_token: + try: + win32api.CloseHandle(elevated_token) + except Exception: + pass + + # Worker reads its stdin once and treats an empty line as "no + # pre-attach" -- close the write end so the child doesn't block. + try: + win32file.WriteFile(stdin_w, b"\n") + except Exception as exc: + logger.warning("writing handoff line to worker stdin failed: %s", exc) + finally: + try: + win32file.CloseHandle(stdin_w) + except Exception: + pass + + logger.info( + "spawned user-session worker pid=? session=%d elevated=%s op=%s", + session_id, + using_elevated, + " ".join(op_args), + ) + + stdout_chunks: list[bytes] = [] + stderr_chunks: list[bytes] = [] + + def _drain(handle: Any, sink: list[bytes]) -> None: + while True: + try: + _, chunk = win32file.ReadFile(handle, 4096) + except pywintypes.error as exc: + if exc.winerror in (109, 233): # BROKEN_PIPE, NO_DATA + return + raise + if not chunk: + return + sink.append(bytes(chunk)) + + import threading as _threading + + err_thread = _threading.Thread(target=_drain, args=(stderr_r, stderr_chunks), daemon=True) + err_thread.start() + try: + _drain(stdout_r, stdout_chunks) + finally: + err_thread.join(timeout=1.0) + + try: + win32event.WaitForSingleObject(proc_handle, int(timeout * 1000)) + exit_code = win32process.GetExitCodeProcess(proc_handle) + finally: + try: + win32file.CloseHandle(stdout_r) + except Exception: + pass + try: + win32file.CloseHandle(stderr_r) + except Exception: + pass + try: + win32api.CloseHandle(proc_handle) + except Exception: + pass + try: + win32api.CloseHandle(thread_handle) + except Exception: + pass + + stdout_text = b"".join(stdout_chunks).decode("utf-8", errors="replace").strip() + stderr_text = b"".join(stderr_chunks).decode("utf-8", errors="replace").strip() + if stderr_text: + logger.info("user-session worker stderr: %s", stderr_text) + + if not stdout_text: + raise RuntimeError( + f"user-session worker produced no stdout (exit={exit_code}, stderr={stderr_text!r})" + ) + try: + payload = json.loads(stdout_text) + except json.JSONDecodeError as exc: + raise RuntimeError( + f"user-session worker stdout not JSON (exit={exit_code}): {stdout_text!r}" + ) from exc + if not payload.get("ok"): + raise RuntimeError(f"user-session worker error: {payload.get('error', 'unknown')}") + return payload.get("result") + + +# --------------------------------------------------------------------------- +# WaitForUACPrompt +# --------------------------------------------------------------------------- + + +def wait_for_uac_prompt(timeout_ms: int = 60_000, poll_ms: int = 250) -> dict | None: + """Block until the Secure Desktop becomes the input desktop, then return the dialog. + + Returns a dict with the UIA tree of the consent dialog plus the extracted + publisher, or ``None`` if the timeout expires without UAC firing. + + Attaches the calling process to ``WinSta0`` once for the duration of the + poll loop — the LocalSystem host service starts on ``Service-0x0-3e7$``, + and ``OpenInputDesktop`` on that station never returns the interactive + user's input desktop. Restoring the original window station on exit + keeps subsequent pipe handlers on their original station. + """ + deadline = time.monotonic() + (timeout_ms / 1000.0) + hwinsta_prev = _user32.GetProcessWindowStation() + hwinsta = _open_winsta0() + if hwinsta: + _user32.SetProcessWindowStation(hwinsta) + logger.info( + "wait_for_uac_prompt: polling winsta=%s for up to %dms (hwinsta=%s)", + "WinSta0" if hwinsta else "(failed-open)", + timeout_ms, + hwinsta, + ) + # Diagnostic: log the current PromptOnSecureDesktop registry value so + # we can tell whether iter-5's "policy was set but UAC still went to + # Winlogon" is the registry read returning 0 (Windows ignoring the + # value) or returning 1 (the write didn't actually stick). + try: + import winreg + + with winreg.OpenKey( + winreg.HKEY_LOCAL_MACHINE, + r"SOFTWARE\Microsoft\Windows\CurrentVersion\Policies\System", + access=winreg.KEY_QUERY_VALUE, + ) as _key: + posd, _ = winreg.QueryValueEx(_key, "PromptOnSecureDesktop") + elua, _ = winreg.QueryValueEx(_key, "EnableLUA") + cpba, _ = winreg.QueryValueEx(_key, "ConsentPromptBehaviorAdmin") + logger.info( + "wait_for_uac_prompt: registry policy: " + "PromptOnSecureDesktop=%s EnableLUA=%s ConsentPromptBehaviorAdmin=%s", + posd, + elua, + cpba, + ) + except Exception as exc: + logger.warning("wait_for_uac_prompt: could not read UAC policy registry: %s", exc) + + seen: dict[str, int] = {} + try: + while time.monotonic() < deadline: + name = "" + hdesk = _open_input_desktop(_DESKTOP_READOBJECTS) + if hdesk: + try: + name = _get_desktop_name(hdesk) or "" + finally: + _user32.CloseDesktop(hdesk) + seen[name] = seen.get(name, 0) + 1 + + # Primary path: PromptOnSecureDesktop=0 (set by `service + # secure-desktop install`) makes UAC render on the user's + # Default desktop. consent.exe is then a regular top-level + # window reachable via plain UIA from a same-session worker -- + # no UIAccess, no DACL loosening, no screenshot trickery. + # Detect by polling for consent.exe in the process list. + # + # We previously had an early-bail branch on input-desktop name + # == "Winlogon" -- but Windows briefly transitions the input + # desktop to Winlogon during the UAC dispatch sequence even + # when POSD=0 keeps the actual dialog on Default. That early + # bail returned tree=[] before the consent.exe check ran, so + # the worker never got a chance to walk the dialog. Now we + # always check consent.exe first regardless of the input + # desktop name; only after the polling window expires without + # ever seeing consent.exe will we report "Winlogon detected" + # as a diagnostic. + consent_pid = _find_consent_pid() + if consent_pid: + logger.info( + "wait_for_uac_prompt: consent.exe pid=%d on desktop=%r -- using Default-desktop UIA path", + consent_pid, + name, + ) + # Pass consent_pid to the worker so uia_get_tree / + # get_uac_publisher can scope via EnumWindows + + # ElementFromHandle / FocusChanged events. consent.exe at + # System integrity is not returned by GetRootElement walk + # even from a UIAccess caller; cross-integrity access + # requires the per-pid scoped path. + pid_arg = f"--consent-pid={consent_pid}" + tree: list[dict] = [] + publisher = None + # Retry the worker walk until it returns the consent window, + # but ALWAYS stay inside the overall deadline. Each spawn is a + # fresh Python + COM init, which is slow on a KVM-less VM, so + # give it a generous per-spawn timeout -- capped to whatever + # time is left so we never blow past the caller's deadline (a + # bounded loop here is what keeps WaitForUACPrompt returning + # within timeout_ms instead of hanging until the client's read + # timeout fires). + attempt = 0 + while True: + remaining = deadline - time.monotonic() + if remaining <= 2.0: + logger.warning( + "wait_for_uac_prompt: deadline reached after %d worker attempts; " + "returning whatever the worker last saw", + attempt, + ) + break + attempt += 1 + spawn_to = min(45.0, remaining - 1.0) + try: + tree = _spawn_in_user_session("tree", pid_arg, timeout=spawn_to) or [] + except Exception as exc: + logger.warning("Default-desktop tree spawn failed: %s", exc) + tree = [] + if _tree_contains_consent(tree, consent_pid): + logger.info( + "wait_for_uac_prompt: consent.exe in UIA tree after %d attempts (%d top windows)", + attempt, + len(tree), + ) + break + time.sleep(0.2) + # Publisher is best-effort; only spend time on it if the + # deadline hasn't already passed. + pub_remaining = deadline - time.monotonic() + if pub_remaining > 3.0: + try: + publisher = _spawn_in_user_session( + "publisher", pid_arg, timeout=min(15.0, pub_remaining - 1.0) + ) + except Exception as exc: + logger.warning("publisher spawn failed: %s", exc) + publisher = None + return { + "desktop": name or "Default", + "publisher": publisher, + "tree": tree, + } + + time.sleep(poll_ms / 1000.0) + logger.warning("wait_for_uac_prompt: timed out; saw desktops: %s", seen) + return None + finally: + if hwinsta: + _user32.SetProcessWindowStation(hwinsta_prev) + _user32.CloseWindowStation(hwinsta) diff --git a/src/windows_mcp/service/user_session_worker.py b/src/windows_mcp/service/user_session_worker.py new file mode 100644 index 00000000..ae897fee --- /dev/null +++ b/src/windows_mcp/service/user_session_worker.py @@ -0,0 +1,114 @@ +"""Worker executed inside the active console user's session. + +The LocalSystem host service spawns this helper via ``CreateProcessAsUser`` +when it needs to walk or click the UAC consent dialog. Session 0 +isolation prevents a service thread from enumerating windows owned by +user-session processes such as ``consent.exe``. A process *inside* the +user's session is not subject to that boundary, and with the user's +elevated linked token + ``SetTokenInformation(TokenUIAccess=1)`` it has +enough access to read consent.exe's higher-integrity UIA tree. + +Invocation:: + + python -m windows_mcp.service.user_session_worker [args...] + +The worker emits a single JSON line on stdout describing the result and +exits with code 0 on success / 1 on failure. The parent service reads the +pipe and forwards the payload over the named-pipe protocol. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import os +import sys + +logger = logging.getLogger(__name__) + + +def _build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser(description="windows-mcp user-session UIA worker") + sub = p.add_subparsers(dest="op", required=True) + tree = sub.add_parser("tree", help="Walk the input desktop's UIA tree.") + tree.add_argument( + "--consent-pid", + type=int, + default=0, + help=( + "If non-zero, scope the walk to the top-level HWND owned by this " + "pid (consent.exe). IUIAutomation.GetRootElement walk does NOT " + "return System-integrity windows even from a UIAccess caller, so " + "for UAC we look up the HWND via EnumWindows and use " + "ElementFromHandle which DOES cross the integrity boundary." + ), + ) + pub = sub.add_parser("publisher", help="Extract the UAC consent publisher string.") + pub.add_argument("--consent-pid", type=int, default=0) + cl = sub.add_parser("click_at", help="Invoke the UIA element at (x, y).") + cl.add_argument("x", type=int) + cl.add_argument("y", type=int) + return p + + +def _drain_stdin_handoff() -> None: + """Broker writes a single line to our stdin then closes it. Pre-iter-8 + that line carried a duplicated Winlogon HDESK and/or consent.exe HWND + for cross-desktop reads. Iter-8 routes UAC to Default so the broker no + longer passes anything, but the broker still writes an empty newline + to signal "no handoff" -- read and discard it so the worker doesn't + block on stdin in some future change.""" + try: + os.read(0, 128) + except OSError: + pass + + +def main() -> int: + # Worker diagnostics go to stderr; stdout is reserved for the JSON payload + # the parent service reads back. + logging.basicConfig( + stream=sys.stderr, + level=logging.INFO, + format="[user-session-worker pid=%(process)d] %(message)s", + ) + _drain_stdin_handoff() + args = _build_parser().parse_args() + + # Import lazily so a failed import surfaces as JSON instead of a Python + # traceback the parent can't parse. + try: + from windows_mcp.service import secure_desktop + except Exception as exc: + json.dump( + {"ok": False, "error": f"import failed: {exc}", "type": type(exc).__name__}, + sys.stdout, + ) + return 1 + + try: + if args.op == "tree": + result = secure_desktop.uia_get_tree(consent_pid=args.consent_pid) + elif args.op == "publisher": + result = secure_desktop.get_uac_publisher(consent_pid=args.consent_pid) + elif args.op == "click_at": + result = secure_desktop.uia_click_at(args.x, args.y) + else: + json.dump({"ok": False, "error": f"unknown op: {args.op}"}, sys.stdout) + return 1 + except Exception as exc: + logger.exception("op %s failed", args.op) + json.dump( + {"ok": False, "error": str(exc), "type": type(exc).__name__}, + sys.stdout, + ) + return 1 + + json.dump({"ok": True, "result": result}, sys.stdout) + sys.stdout.flush() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/windows_mcp/tools/__init__.py b/src/windows_mcp/tools/__init__.py index 4a55b46d..c99a0634 100644 --- a/src/windows_mcp/tools/__init__.py +++ b/src/windows_mcp/tools/__init__.py @@ -12,6 +12,7 @@ scrape, shell, snapshot, + uac, ) _MODULES = [ @@ -26,6 +27,7 @@ process, notification, registry, + uac, ] diff --git a/src/windows_mcp/tools/uac.py b/src/windows_mcp/tools/uac.py new file mode 100644 index 00000000..dbb2c1e2 --- /dev/null +++ b/src/windows_mcp/tools/uac.py @@ -0,0 +1,87 @@ +"""UAC handling tool — WaitForUACPrompt. + +Lets the agent block until a UAC consent prompt fires on the Secure Desktop, +returning the dialog's UIA tree and (best-effort) verified publisher so the +agent can decide whether to approve it. + +Requires the LocalSystem host service to be installed +(``uv run windows-mcp service secure-desktop install``). Without the service, +this tool returns an explanatory error rather than silently blocking — the +broker cannot see the Secure Desktop on its own. +""" + +from __future__ import annotations + +from typing import Annotated + +from fastmcp import Context +from mcp.types import ToolAnnotations +from pydantic import Field + +from windows_mcp.infrastructure import with_analytics + + +def register(mcp, *, get_desktop, get_analytics): + @mcp.tool( + name="WaitForUACPrompt", + description=( + "Block until a UAC consent prompt appears on the Secure Desktop, then " + "return the dialog as a UIA tree plus the verified publisher (if " + "detectable). Use this when you have just triggered an operation that " + "is expected to require elevation. Requires the Windows-MCP host " + "service to be installed." + ), + annotations=ToolAnnotations( + title="WaitForUACPrompt", + readOnlyHint=True, + destructiveHint=False, + idempotentHint=False, + openWorldHint=False, + ), + ) + @with_analytics(get_analytics(), "WaitForUACPrompt-Tool") + def wait_for_uac_prompt( + timeout_ms: Annotated[ + int, + Field( + description=( + "Maximum time to wait, in milliseconds, before giving up. " + "Defaults to 60000 (60 seconds)." + ), + ge=100, + le=600_000, + ), + ] = 60_000, + ctx: Context = None, + ) -> dict: + from windows_mcp.service import get_host_client + + client = get_host_client() + if not client.is_available(): + return { + "ok": False, + "error": ( + "Windows-MCP Secure Desktop host service is not installed or not " + "running. Install it with: " + "`uv run windows-mcp service secure-desktop install` " + "(requires Administrator)." + ), + } + try: + result = client.wait_for_uac_prompt(timeout_ms=timeout_ms) + except Exception as exc: + return {"ok": False, "error": f"Host service call failed: {exc}"} + if result is None: + return {"ok": True, "fired": False, "reason": "timeout"} + try: + pol = client.policy_state() + except Exception: + pol = None + return { + "ok": True, + "fired": True, + "desktop": result.get("desktop"), + "publisher": result.get("publisher"), + "tree": result.get("tree"), + "policy": pol, + }