fix(computer-use): preserve macOS TCC daemon identity

Launch private computer-use daemons through CuaDriver.app so Screen
Recording authorization remains attached to its stable bundle identity
instead of Hermes' ad-hoc signature.

Co-Authored-By: GPT-5.6 Codex <noreply@openai.com>
This commit is contained in:
projetsjsl
2026-08-25 18:34:30 -04:00
committed by Teknium
parent 1fe0f2f3ac
commit 4746f614be
4 changed files with 118 additions and 13 deletions

View File

@@ -1535,7 +1535,8 @@ class TestCaptureAppFilterNoMatch:
"structuredContent": None},
]
backend.capture(mode="ax")
with patch("tools.computer_use.cua_backend.sys.platform", "linux"):
backend.capture(mode="ax")
assert backend._active_pid == 200
assert backend._active_window_id == 2

View File

@@ -248,6 +248,43 @@ def test_schema_does_not_expose_approval_token():
# ── bounded embedded daemon ─────────────────────────────────────────────
def test_macos_embedded_daemon_launches_through_cuadriver_app():
command = cb._embedded_daemon_spawn_command(
"/tmp/cua-driver",
["serve", "--embedded", "--socket", "/tmp/private.sock"],
platform="darwin",
app_path="/Applications/CuaDriver.app",
)
assert command == [
"/usr/bin/open",
"-n",
"-a",
"/Applications/CuaDriver.app",
"--args",
"serve",
"--embedded",
"--socket",
"/tmp/private.sock",
]
def test_non_macos_embedded_daemon_keeps_direct_binary_launch():
command = cb._embedded_daemon_spawn_command(
"/tmp/cua-driver",
["serve", "--embedded", "--socket", "/tmp/private.sock"],
platform="linux",
)
assert command == [
"/tmp/cua-driver",
"serve",
"--embedded",
"--socket",
"/tmp/private.sock",
]
def test_bounded_daemon_requires_a_manifest():
with pytest.raises(ValueError, match="capability_manifest"):
_EmbeddedCuaDaemon("cua-driver", "bounded")

View File

@@ -174,7 +174,7 @@ def test_unrestricted_embedded_daemon_uses_private_socket_and_two_part_ack():
stopped = SimpleNamespace(returncode=0, stdout="", stderr="")
daemon = cua_backend._EmbeddedCuaDaemon("cua-driver", "unrestricted")
with patch.object(
with patch.object(cua_backend.sys, "platform", "linux"), patch.object(
cua_backend,
"_resolve_mcp_invocation",
return_value=("/opt/cua-driver", ["mcp"]),

View File

@@ -592,13 +592,61 @@ def _wsl_windows_path_to_posix(path: str) -> str:
return os.path.join("/mnt", drive, *(str(part) for part in win.parts[1:]))
def _resolve_cua_driver_app_path(driver_cmd: str) -> Optional[str]:
"""Return the installed CuaDriver.app carrying *driver_cmd*, if present."""
marker = ".app/Contents/MacOS/"
candidates: List[str] = []
marker_index = driver_cmd.find(marker)
if marker_index >= 0:
candidates.append(driver_cmd[: marker_index + len(".app")])
candidates.extend(
[
"/Applications/CuaDriver.app",
os.path.expanduser("~/Applications/CuaDriver.app"),
]
)
for candidate in candidates:
executable = os.path.join(candidate, "Contents", "MacOS", "cua-driver")
if os.path.isfile(executable) and os.access(executable, os.X_OK):
return candidate
return None
def _embedded_daemon_spawn_command(
driver_cmd: str,
serve_args: List[str],
*,
platform: str,
app_path: Optional[str] = None,
) -> List[str]:
"""Build the private-daemon launch while preserving macOS TCC identity."""
if platform != "darwin":
return [driver_cmd, *serve_args]
resolved_app = app_path or _resolve_cua_driver_app_path(driver_cmd)
if not resolved_app:
raise RuntimeError(
"CuaDriver.app is required for private computer-use sessions on macOS. "
"Run `hermes computer-use install` to restore it."
)
return [
"/usr/bin/open",
"-n",
"-a",
resolved_app,
"--args",
*serve_args,
]
class _EmbeddedCuaDaemon:
"""Private host-owned daemon for a non-standard permission mode.
"""Private daemon for a non-standard permission mode.
Cua Driver permission mode is immutable after daemon startup. Reusing the
machine-wide daemon would therefore let one Hermes session's YOLO choice
affect another session. A private embedded daemon gives the requesting
session its own socket, process, and launch-time authorization:
session its own socket, runtime, and launch-time authorization. On macOS
the runtime is launched through CuaDriver.app so TCC remains attached to
``com.trycua.driver`` instead of the embedding host's ad-hoc signature:
* ``unrestricted`` — explicit Hermes YOLO; launch-time risk
acknowledgement via ``--dangerously-bypass-approvals``.
@@ -661,6 +709,9 @@ class _EmbeddedCuaDaemon:
self._command = driver_cmd
self._mcp_args: List[str] = list(_CUA_DRIVER_ARGS)
self._process: Any = None
self._owns_runtime = False
self._running = False
self._launch_via_app = False
self._stderr_tail: deque[str] = deque(maxlen=20)
self._stderr_thread: Optional[threading.Thread] = None
token = uuid.uuid4().hex[:12]
@@ -692,7 +743,7 @@ class _EmbeddedCuaDaemon:
pass
def start(self) -> None:
if self._process is not None and self._process.poll() is None:
if self._running:
return
from tools.environments.local import _sanitize_subprocess_env
@@ -702,8 +753,7 @@ class _EmbeddedCuaDaemon:
raise RuntimeError(cua_driver_install_hint())
self._command, self._mcp_args = _resolve_mcp_invocation(self._driver_cmd)
env = _sanitize_subprocess_env(self.child_env())
command = [
self._command,
serve_args = [
"serve",
"--embedded",
"--socket",
@@ -713,7 +763,7 @@ class _EmbeddedCuaDaemon:
self.permission_mode,
]
if self.permission_mode == "unrestricted":
command.append("--dangerously-bypass-approvals")
serve_args.append("--dangerously-bypass-approvals")
# A v3 manifest is a ceiling, not a mode: cua-driver accepts it
# alongside any permission mode and it "can narrow a profile but never
# widen it". Attaching it to unrestricted is what bounds an
@@ -721,7 +771,7 @@ class _EmbeddedCuaDaemon:
# applies — not only for bounded, which used to drop it for every
# other mode.
if self.manifest_applies:
command.extend(
serve_args.extend(
[
"--capability-manifest",
str(self.capability_manifest),
@@ -731,7 +781,15 @@ class _EmbeddedCuaDaemon:
# The private daemon owns the platform cursor overlay. Applying the
# policy only to its MCP proxy leaves this long-lived serve process
# free to create a full-screen overlay before session tuning runs.
command = _mcp_args_with_overlay_flag(command, driver_cmd=self._command)
# Must be appended BEFORE the macOS app-launch wrapping so the flag
# travels inside `open ... --args` with the rest of the serve args.
serve_args = _mcp_args_with_overlay_flag(serve_args, driver_cmd=self._command)
self._launch_via_app = sys.platform == "darwin"
command = _embedded_daemon_spawn_command(
self._command,
serve_args,
platform=sys.platform,
)
self._process = subprocess.Popen(
command,
stdin=subprocess.DEVNULL,
@@ -740,6 +798,7 @@ class _EmbeddedCuaDaemon:
text=True,
env=env,
)
self._owns_runtime = True
self._stderr_thread = threading.Thread(
target=self._drain_stderr,
args=(self._process,),
@@ -750,7 +809,10 @@ class _EmbeddedCuaDaemon:
deadline = time.monotonic() + self._START_TIMEOUT_SECONDS
while time.monotonic() < deadline:
if self._process.poll() is not None:
return_code = self._process.poll()
if return_code is not None and (
not self._launch_via_app or return_code != 0
):
detail = "; ".join(self._stderr_tail) or "no diagnostic output"
raise RuntimeError(
f"embedded cua-driver exited during startup: {detail}"
@@ -767,6 +829,7 @@ class _EmbeddedCuaDaemon:
except (OSError, subprocess.SubprocessError):
probe = None
if probe is not None and probe.returncode == 0:
self._running = True
return
time.sleep(0.1)
@@ -775,7 +838,7 @@ class _EmbeddedCuaDaemon:
raise RuntimeError(f"embedded cua-driver startup timed out: {detail}")
def proxy_invocation(self) -> Tuple[str, List[str]]:
if self._process is None or self._process.poll() is not None:
if not self._running:
raise RuntimeError("embedded cua-driver daemon is not running")
return self._command, [
*self._mcp_args,
@@ -787,7 +850,10 @@ class _EmbeddedCuaDaemon:
def stop(self) -> None:
process = self._process
self._process = None
if process is not None and process.poll() is None:
owns_runtime = self._owns_runtime
self._owns_runtime = False
self._running = False
if owns_runtime:
from tools.environments.local import _sanitize_subprocess_env
try:
@@ -801,6 +867,7 @@ class _EmbeddedCuaDaemon:
)
except (OSError, subprocess.SubprocessError):
pass
if process is not None:
try:
process.wait(timeout=5.0)
except subprocess.TimeoutExpired: