diff --git a/hermes_cli/tools_config.py b/hermes_cli/tools_config.py index 9a74877a0ca..f91abb33f8a 100644 --- a/hermes_cli/tools_config.py +++ b/hermes_cli/tools_config.py @@ -971,19 +971,51 @@ def _run_cua_driver_installer(label: str = "Installing", verbose: bool = True) - "powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ps_oneliner, ] - use_shell = False manual_hint = ( 'powershell -NoProfile -ExecutionPolicy Bypass -Command ' f'"{ps_oneliner}"' ) + script_path = None else: - install_cmd = ( - "/bin/bash -c \"$(curl -fsSL " + # Download-then-exec instead of `bash -c "$(curl …)"`: no shell=True, + # no command substitution, and the script lands in a mkstemp file + # (unpredictable name, 0600) rather than a fixed /tmp path — avoiding + # both the shell-injection surface and a symlink/TOCTOU race on + # multi-user machines. The manual hint stays the upstream one-liner + # since that's what the docs/README teach. + import tempfile as _tempfile + + install_url = ( "https://raw.githubusercontent.com/trycua/cua/main/" - "libs/cua-driver/scripts/install.sh)\"" + "libs/cua-driver/scripts/install.sh" ) - use_shell = True - manual_hint = install_cmd + manual_hint = f'/bin/bash -c "$(curl -fsSL {install_url})"' + fd, script_path = _tempfile.mkstemp(prefix="cua-driver-install-", suffix=".sh") + os.close(fd) + try: + dl = subprocess.run( + ["curl", "-fsSL", "-o", script_path, install_url], + capture_output=True, text=True, timeout=120, + ) + except (subprocess.TimeoutExpired, OSError) as e: + _print_warning(f" cua-driver installer download failed: {e}") + try: + os.remove(script_path) + except OSError: + pass + return False + if dl.returncode != 0: + _print_warning( + " cua-driver installer download failed: " + f"{(dl.stderr or '').strip()[:200]}" + ) + try: + os.remove(script_path) + except OSError: + pass + return False + install_cmd = ["/bin/bash", script_path] + use_shell = False if verbose: _print_info(f" {label} cua-driver (background computer-use)...") @@ -1099,6 +1131,12 @@ def _run_cua_driver_installer(label: str = "Installing", verbose: bool = True) - except Exception as e: _print_warning(f" cua-driver {label.lower()} failed: {e}") return False + finally: + if script_path: + try: + os.remove(script_path) + except OSError: + pass def _run_post_setup(post_setup_key: str): diff --git a/tests/hermes_cli/test_install_cua_driver.py b/tests/hermes_cli/test_install_cua_driver.py index 3b7dfa36601..bf69f7fae89 100644 --- a/tests/hermes_cli/test_install_cua_driver.py +++ b/tests/hermes_cli/test_install_cua_driver.py @@ -283,6 +283,7 @@ class TestInstallerTimeoutKillsProcessGroup: killed["sig"] = sig with patch("platform.system", return_value="Linux"), \ + patch("subprocess.run", return_value=MagicMock(returncode=0, stderr="")), \ patch("subprocess.Popen", return_value=fake_proc), \ patch.object(tools_config.os, "getpgid", return_value=99999), \ patch.object(tools_config.os, "killpg", side_effect=fake_killpg), \ @@ -319,6 +320,7 @@ class TestInstallerTimeoutKillsProcessGroup: return fake_proc with patch("platform.system", return_value="Linux"), \ + patch("subprocess.run", return_value=MagicMock(returncode=0, stderr="")), \ patch("subprocess.Popen", side_effect=fake_popen), \ patch.object(tools_config, "_clear_stale_cua_install_lock"), \ patch.object(tools_config, "_print_warning"), \ @@ -326,3 +328,96 @@ class TestInstallerTimeoutKillsProcessGroup: tools_config._run_cua_driver_installer(label="Refreshing", verbose=False) assert captured.get("start_new_session") is True + + +class TestInstallerNoShell: + """The POSIX installer path must not use shell=True or command + substitution: the script is downloaded to a mkstemp file and exec'd + as a plain argv list (salvage of #34974's intent, without the fixed + /tmp path TOCTOU that PR introduced).""" + + def _run(self, download_rc=0): + import subprocess + from unittest.mock import MagicMock + from hermes_cli import tools_config + + calls = [] + fake_proc = MagicMock() + fake_proc.pid = 1 + fake_proc.returncode = 0 + fake_proc.communicate.return_value = ("", None) + + def fake_run(cmd, **kw): + calls.append(("run", cmd, kw)) + m = MagicMock() + m.returncode = download_rc + m.stderr = "curl: (6) could not resolve" if download_rc else "" + return m + + def fake_popen(cmd, **kw): + calls.append(("popen", cmd, kw)) + return fake_proc + + with patch("platform.system", return_value="Linux"), \ + patch("subprocess.run", side_effect=fake_run), \ + patch("subprocess.Popen", side_effect=fake_popen), \ + patch.object(tools_config.shutil, "which", return_value="/usr/local/bin/cua-driver"), \ + patch.object(tools_config, "_clear_stale_cua_install_lock"), \ + patch.object(tools_config, "_print_warning"), \ + patch.object(tools_config, "_print_info"), \ + patch.object(tools_config, "_print_success"): + ok = tools_config._run_cua_driver_installer(label="Refreshing", verbose=False) + return ok, calls + + def test_posix_path_downloads_then_execs_argv_list(self): + ok, calls = self._run() + assert ok is True + run_calls = [c for c in calls if c[0] == "run"] + popen_calls = [c for c in calls if c[0] == "popen"] + assert len(run_calls) == 1 and len(popen_calls) == 1 + # Download: plain argv curl, no shell. + dl_cmd = run_calls[0][1] + assert isinstance(dl_cmd, list) and dl_cmd[0] == "curl" + # Exec: argv list ["/bin/bash", ], shell=False. + exec_cmd, exec_kw = popen_calls[0][1], popen_calls[0][2] + assert isinstance(exec_cmd, list) and exec_cmd[0] == "/bin/bash" + assert "cua-driver-install-" in exec_cmd[1] + assert exec_kw.get("shell") is False + + def test_download_failure_returns_false_without_exec(self): + ok, calls = self._run(download_rc=6) + assert ok is False + assert not [c for c in calls if c[0] == "popen"] + + def test_temp_script_removed_after_run(self, tmp_path): + import os + captured = {} + import subprocess + from unittest.mock import MagicMock + from hermes_cli import tools_config + + fake_proc = MagicMock() + fake_proc.pid = 1 + fake_proc.returncode = 0 + fake_proc.communicate.return_value = ("", None) + + def fake_run(cmd, **kw): + m = MagicMock(); m.returncode = 0; m.stderr = "" + return m + + def fake_popen(cmd, **kw): + captured["script"] = cmd[1] + return fake_proc + + with patch("platform.system", return_value="Linux"), \ + patch("subprocess.run", side_effect=fake_run), \ + patch("subprocess.Popen", side_effect=fake_popen), \ + patch.object(tools_config.shutil, "which", return_value="/usr/local/bin/cua-driver"), \ + patch.object(tools_config, "_clear_stale_cua_install_lock"), \ + patch.object(tools_config, "_print_warning"), \ + patch.object(tools_config, "_print_info"), \ + patch.object(tools_config, "_print_success"): + tools_config._run_cua_driver_installer(label="Refreshing", verbose=False) + + assert "script" in captured + assert not os.path.exists(captured["script"])