From 50e27abdd35da2c5ae4e93e7e44ede2f847d300e Mon Sep 17 00:00:00 2001 From: ZundamonnoVRChatkaisetu Date: Wed, 29 Jul 2026 08:57:36 -0700 Subject: [PATCH] fix(computer-use): hide Windows cua-driver subprocess consoles Apply windows_hide_flags() (CREATE_NO_WINDOW; 0 on POSIX) at the Windows-reachable cua-driver subprocess boundaries: manifest probe, update checker, CLI fallback transport, doctor health-report spawn, and the permissions/status runner. Prevents OpenConsole/Windows Terminal windows flashing into the foreground when spawned from GUI-backed Gateway/Desktop processes. The env-probe half of the original PR was already implemented on main and is not re-applied here. Salvaged from #62821 by @ZundamonnoVRChatkaisetu (original commits carried a placeholder 'Claude Code Enterprise' identity; re-authored to the contributor's GitHub identity). --- .../test_cua_spawn_env_sanitization.py | 110 ++++++++++++++++++ tools/computer_use/cua_backend.py | 4 + tools/computer_use/doctor.py | 3 + tools/computer_use/permissions.py | 3 + 4 files changed, 120 insertions(+) diff --git a/tests/computer_use/test_cua_spawn_env_sanitization.py b/tests/computer_use/test_cua_spawn_env_sanitization.py index 8954d94e42a..83e354ff606 100644 --- a/tests/computer_use/test_cua_spawn_env_sanitization.py +++ b/tests/computer_use/test_cua_spawn_env_sanitization.py @@ -17,6 +17,7 @@ import json from unittest.mock import MagicMock SECRET = "sk-super-secret-should-not-leak" +CREATE_NO_WINDOW = 0x08000000 def _fake_completed_process(stdout: str) -> MagicMock: @@ -31,6 +32,7 @@ def _capture_run(captured, stdout=""): def fake_run(cmd, **kwargs): captured["cmd"] = cmd captured["env"] = kwargs.get("env") + captured["creationflags"] = kwargs.get("creationflags") return _fake_completed_process(stdout) return fake_run @@ -45,6 +47,13 @@ def _assert_sanitized(captured): assert env.get("CUA_DRIVER_RS_TELEMETRY_ENABLED") == "0" +def _patch_windows_hide_flags(monkeypatch, module): + monkeypatch.setattr(module, "IS_WINDOWS", True, raising=False) + monkeypatch.setattr( + module, "windows_hide_flags", lambda: CREATE_NO_WINDOW, raising=False + ) + + def test_resolve_mcp_invocation_sanitizes_env(monkeypatch): monkeypatch.setenv("ANTHROPIC_API_KEY", SECRET) monkeypatch.setenv("PATH", "/usr/bin:/bin") @@ -53,6 +62,7 @@ def test_resolve_mcp_invocation_sanitizes_env(monkeypatch): from tools.computer_use import cua_backend captured = {} + _patch_windows_hide_flags(monkeypatch, cua_backend) manifest = json.dumps({"mcp_invocation": {"command": "cua-driver", "args": ["mcp"]}}) monkeypatch.setattr( cua_backend.subprocess, "run", _capture_run(captured, stdout=manifest) @@ -61,6 +71,7 @@ def test_resolve_mcp_invocation_sanitizes_env(monkeypatch): cmd, args = cua_backend._resolve_mcp_invocation("cua-driver") assert cmd == "cua-driver" _assert_sanitized(captured) + assert captured["creationflags"] == CREATE_NO_WINDOW def test_update_check_sanitizes_env(monkeypatch): @@ -71,6 +82,7 @@ def test_update_check_sanitizes_env(monkeypatch): from tools.computer_use import cua_backend captured = {} + _patch_windows_hide_flags(monkeypatch, cua_backend) payload = json.dumps({ "current_version": "1.0.0", "latest_version": "1.0.0", @@ -87,6 +99,30 @@ def test_update_check_sanitizes_env(monkeypatch): cua_backend.cua_driver_update_check(timeout=1.0) _assert_sanitized(captured) + assert captured["creationflags"] == CREATE_NO_WINDOW + + +def test_cli_fallback_sanitizes_env_and_hides_console_on_windows(monkeypatch): + monkeypatch.setenv("ANTHROPIC_API_KEY", SECRET) + monkeypatch.setenv("PATH", "/usr/bin:/bin") + monkeypatch.delenv("HERMES_CUA_TELEMETRY", raising=False) + + from tools.computer_use import cua_backend + + captured = {} + _patch_windows_hide_flags(monkeypatch, cua_backend) + monkeypatch.setattr( + cua_backend.subprocess, + "run", + _capture_run(captured, stdout=json.dumps({"tree_markdown": "root"})), + ) + + session = object.__new__(cua_backend._CuaDriverSession) + result = session._call_tool_via_cli("list_windows", {}, timeout=5.0) + + assert result["isError"] is False + _assert_sanitized(captured) + assert captured["creationflags"] == CREATE_NO_WINDOW def test_permissions_run_sanitizes_env(monkeypatch): @@ -105,6 +141,80 @@ def test_permissions_run_sanitizes_env(monkeypatch): _assert_sanitized(captured) +def test_windows_status_hides_every_reachable_subprocess(monkeypatch): + """The Desktop status API reaches only version + doctor spawns on Windows. + + The permissions grant subprocess is intentionally excluded: its public + entry point returns before spawning anywhere except macOS, where + ``CREATE_NO_WINDOW`` is not applicable. + """ + from tools.computer_use import permissions + + binary = r"C:\Program Files\cua-driver\cua-driver.exe" + calls = [] + stdout_by_args = { + ("--version",): "cua-driver 1.2.3\n", + ("doctor", "--json"): json.dumps({"ok": True, "probes": []}), + } + + def fake_run(cmd, **kwargs): + calls.append((cmd, kwargs)) + return _fake_completed_process(stdout_by_args[tuple(cmd[1:])]) + + monkeypatch.setattr(permissions.sys, "platform", "win32") + monkeypatch.setattr(permissions, "windows_hide_flags", lambda: CREATE_NO_WINDOW) + monkeypatch.setattr(permissions.shutil, "which", lambda command: binary) + monkeypatch.setattr(permissions.subprocess, "run", fake_run) + + status = permissions.computer_use_status("cua-driver") + + assert status["version"] == "cua-driver 1.2.3" + assert status["ready"] is True + assert [cmd[1:] for cmd, _ in calls] == [ + ["--version"], + ["doctor", "--json"], + ] + assert calls, "Windows status must exercise at least one subprocess boundary" + for cmd, kwargs in calls: + assert kwargs.get("creationflags") == CREATE_NO_WINDOW, cmd + + +def test_doctor_spawn_sanitizes_env_and_hides_console_on_windows(monkeypatch): + monkeypatch.setenv("ANTHROPIC_API_KEY", SECRET) + monkeypatch.setenv("PATH", "/usr/bin:/bin") + monkeypatch.delenv("HERMES_CUA_TELEMETRY", raising=False) + + from tools.computer_use import doctor + + captured = {} + _patch_windows_hide_flags(monkeypatch, doctor) + proc = MagicMock() + proc.stdout.readline.side_effect = [ + json.dumps({"jsonrpc": "2.0", "id": 1, "result": {}}), + json.dumps({ + "jsonrpc": "2.0", + "id": 2, + "result": { + "structuredContent": {"schema_version": "1", "overall": "ok"} + }, + }), + ] + + def fake_popen(cmd, **kwargs): + captured["cmd"] = cmd + captured["env"] = kwargs.get("env") + captured["creationflags"] = kwargs.get("creationflags") + return proc + + monkeypatch.setattr(doctor.subprocess, "Popen", fake_popen) + + report = doctor._drive_health_report("cua-driver", timeout=1.0) + + assert report["overall"] == "ok" + _assert_sanitized(captured) + assert captured["creationflags"] == CREATE_NO_WINDOW + + def test_doctor_sanitized_env_helper(monkeypatch): """The doctor MCP spawn site must pass the sanitized env to Popen. diff --git a/tools/computer_use/cua_backend.py b/tools/computer_use/cua_backend.py index 9d09a4e2657..b029071f4b5 100644 --- a/tools/computer_use/cua_backend.py +++ b/tools/computer_use/cua_backend.py @@ -50,6 +50,7 @@ import threading import uuid from typing import Any, Dict, List, Optional, Tuple +from hermes_cli._subprocess_compat import windows_hide_flags from tools.computer_use.backend import ( ActionResult, CaptureResult, @@ -381,6 +382,7 @@ def _resolve_mcp_invocation( [driver_cmd, "manifest"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, stdin=subprocess.DEVNULL, + creationflags=windows_hide_flags(), # cua-driver is a third-party binary — never hand it provider # API keys via inherited env (same policy as the MCP and CLI # fallback spawns below; #53503/#55709/#58889 lineage). @@ -589,6 +591,7 @@ def cua_driver_update_check(*, timeout: Optional[float] = None) -> Optional[Dict # stdin-reading mode rather than erroring — DEVNULL gives them EOF # so they exit fast instead of blocking until the timeout. stdin=subprocess.DEVNULL, + creationflags=windows_hide_flags(), # Sanitized like every other cua-driver spawn: third-party # binary, no inherited provider keys (#53503/#55709/#58889). env=_sanitize_subprocess_env(cua_driver_child_env()), @@ -1330,6 +1333,7 @@ class _CuaDriverSession: try: proc = _subprocess.run( cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=max(15.0, timeout), + creationflags=windows_hide_flags(), env=_sanitize_subprocess_env(cua_driver_child_env()), ) except Exception as e: # pragma: no cover - subprocess spawn failure diff --git a/tools/computer_use/doctor.py b/tools/computer_use/doctor.py index f621a0e8410..a69cc3b564f 100644 --- a/tools/computer_use/doctor.py +++ b/tools/computer_use/doctor.py @@ -30,6 +30,8 @@ import subprocess import sys from typing import Any, Dict, List, Optional, Sequence, Tuple +from hermes_cli._subprocess_compat import windows_hide_flags + # Match the ALLOWED_STATUS_VALUES + ALLOWED_OVERALL_VALUES the cua-driver # integration test pins. If health_report widens its vocabulary, add here. @@ -215,6 +217,7 @@ def _open_mcp(binary: str) -> subprocess.Popen: encoding="utf-8", errors="replace", bufsize=1, + creationflags=windows_hide_flags(), env=_sanitized_cua_env(), ) diff --git a/tools/computer_use/permissions.py b/tools/computer_use/permissions.py index 541fd284c0c..65eac7877c1 100644 --- a/tools/computer_use/permissions.py +++ b/tools/computer_use/permissions.py @@ -29,6 +29,8 @@ import subprocess import sys from typing import Any, Dict, List, Optional +from hermes_cli._subprocess_compat import windows_hide_flags + # Platforms with a cua-driver runtime backend (mirrors the toolset platform_gate). _RUNTIME_PLATFORMS = frozenset({"darwin", "win32", "linux"}) _BOOLS = ("accessibility", "screen_recording", "screen_recording_capturable") @@ -70,6 +72,7 @@ def _run(binary: str, *args: str, timeout: float) -> subprocess.CompletedProcess timeout=timeout, env=_child_env(), stdin=subprocess.DEVNULL, + creationflags=windows_hide_flags(), )