diff --git a/nanobot/agent/tools/shell.py b/nanobot/agent/tools/shell.py index e5c04eb72..b8b7b632c 100644 --- a/nanobot/agent/tools/shell.py +++ b/nanobot/agent/tools/shell.py @@ -15,6 +15,8 @@ from nanobot.agent.tools.sandbox import wrap_command from nanobot.agent.tools.schema import IntegerSchema, StringSchema, tool_parameters_schema from nanobot.config.paths import get_media_dir +_IS_WINDOWS = sys.platform == "win32" + @tool_parameters( tool_parameters_schema( @@ -88,27 +90,27 @@ class ExecTool(Tool): return guard_error if self.sandbox: - workspace = self.working_dir or cwd - command = wrap_command(self.sandbox, command, workspace, cwd) - cwd = str(Path(workspace).resolve()) + if _IS_WINDOWS: + logger.warning( + "Sandbox '{}' is not supported on Windows; running unsandboxed", + self.sandbox, + ) + else: + workspace = self.working_dir or cwd + command = wrap_command(self.sandbox, command, workspace, cwd) + cwd = str(Path(workspace).resolve()) effective_timeout = min(timeout or self.timeout, self._MAX_TIMEOUT) - env = self._build_env() if self.path_append: - command = f'export PATH="$PATH:{self.path_append}"; {command}' - - bash = shutil.which("bash") or "/bin/bash" + if _IS_WINDOWS: + env["PATH"] = env.get("PATH", "") + ";" + self.path_append + else: + command = f'export PATH="$PATH:{self.path_append}"; {command}' try: - process = await asyncio.create_subprocess_exec( - bash, "-l", "-c", command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - cwd=cwd, - env=env, - ) + process = await self._spawn(command, cwd, env) try: stdout, stderr = await asyncio.wait_for( @@ -136,7 +138,6 @@ class ExecTool(Tool): result = "\n".join(output_parts) if output_parts else "(no output)" - # Head + tail truncation to preserve both start and end of output max_len = self._MAX_OUTPUT if len(result) > max_len: half = max_len // 2 @@ -151,6 +152,29 @@ class ExecTool(Tool): except Exception as e: return f"Error executing command: {str(e)}" + @staticmethod + async def _spawn( + command: str, cwd: str, env: dict[str, str], + ) -> asyncio.subprocess.Process: + """Launch *command* in a platform-appropriate shell.""" + if _IS_WINDOWS: + comspec = env.get("COMSPEC", os.environ.get("COMSPEC", "cmd.exe")) + return await asyncio.create_subprocess_exec( + comspec, "/c", command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + env=env, + ) + bash = shutil.which("bash") or "/bin/bash" + return await asyncio.create_subprocess_exec( + bash, "-l", "-c", command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + env=env, + ) + @staticmethod async def _kill_process(process: asyncio.subprocess.Process) -> None: """Kill a subprocess and reap it to prevent zombies.""" @@ -160,7 +184,7 @@ class ExecTool(Tool): except asyncio.TimeoutError: pass finally: - if sys.platform != "win32": + if not _IS_WINDOWS: try: os.waitpid(process.pid, os.WNOHANG) except (ProcessLookupError, ChildProcessError) as e: @@ -169,11 +193,26 @@ class ExecTool(Tool): def _build_env(self) -> dict[str, str]: """Build a minimal environment for subprocess execution. - Uses HOME so that ``bash -l`` sources the user's profile (which sets - PATH and other essentials). Only PATH is extended with *path_append*; - the parent process's environment is **not** inherited, preventing - secrets in env vars from leaking to LLM-generated commands. + On Unix, only HOME/LANG/TERM are passed; ``bash -l`` sources the + user's profile which sets PATH and other essentials. + + On Windows, ``cmd.exe`` has no login-profile mechanism, so a curated + set of system variables (including PATH) is forwarded. API keys and + other secrets are still excluded. """ + if _IS_WINDOWS: + sr = os.environ.get("SYSTEMROOT", r"C:\Windows") + return { + "SYSTEMROOT": sr, + "COMSPEC": os.environ.get("COMSPEC", f"{sr}\\system32\\cmd.exe"), + "USERPROFILE": os.environ.get("USERPROFILE", ""), + "HOMEDRIVE": os.environ.get("HOMEDRIVE", "C:"), + "HOMEPATH": os.environ.get("HOMEPATH", "\\"), + "TEMP": os.environ.get("TEMP", f"{sr}\\Temp"), + "TMP": os.environ.get("TMP", f"{sr}\\Temp"), + "PATHEXT": os.environ.get("PATHEXT", ".COM;.EXE;.BAT;.CMD"), + "PATH": os.environ.get("PATH", f"{sr}\\system32;{sr}"), + } home = os.environ.get("HOME", "/tmp") return { "HOME": home, diff --git a/tests/tools/test_exec_env.py b/tests/tools/test_exec_env.py index e5c0f48bb..a05510af4 100644 --- a/tests/tools/test_exec_env.py +++ b/tests/tools/test_exec_env.py @@ -1,10 +1,15 @@ """Tests for exec tool environment isolation.""" +import sys + import pytest from nanobot.agent.tools.shell import ExecTool +_UNIX_ONLY = pytest.mark.skipif(sys.platform == "win32", reason="Unix shell commands") + +@_UNIX_ONLY @pytest.mark.asyncio async def test_exec_does_not_leak_parent_env(monkeypatch): """Env vars from the parent process must not be visible to commands.""" @@ -22,6 +27,7 @@ async def test_exec_has_working_path(): assert "hello" in result +@_UNIX_ONLY @pytest.mark.asyncio async def test_exec_path_append(): """The pathAppend config should be available in the command's PATH.""" @@ -30,6 +36,7 @@ async def test_exec_path_append(): assert "/opt/custom/bin" in result +@_UNIX_ONLY @pytest.mark.asyncio async def test_exec_path_append_preserves_system_path(): """pathAppend must not clobber standard system paths.""" diff --git a/tests/tools/test_exec_platform.py b/tests/tools/test_exec_platform.py new file mode 100644 index 000000000..aa3ffee71 --- /dev/null +++ b/tests/tools/test_exec_platform.py @@ -0,0 +1,269 @@ +"""Tests for cross-platform shell execution. + +Verifies that ExecTool selects the correct shell, environment, path-append +strategy, and sandbox behaviour per platform — without actually running +platform-specific binaries (all subprocess calls are mocked). +""" + +from unittest.mock import AsyncMock, patch + +import pytest + +from nanobot.agent.tools.shell import ExecTool + + +# --------------------------------------------------------------------------- +# _build_env +# --------------------------------------------------------------------------- + +class TestBuildEnvUnix: + + def test_expected_keys(self): + with patch("nanobot.agent.tools.shell._IS_WINDOWS", False): + env = ExecTool()._build_env() + assert set(env) == {"HOME", "LANG", "TERM"} + + def test_home_from_environ(self, monkeypatch): + monkeypatch.setenv("HOME", "/Users/dev") + with patch("nanobot.agent.tools.shell._IS_WINDOWS", False): + env = ExecTool()._build_env() + assert env["HOME"] == "/Users/dev" + + def test_secrets_excluded(self, monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-secret") + monkeypatch.setenv("NANOBOT_TOKEN", "tok-secret") + with patch("nanobot.agent.tools.shell._IS_WINDOWS", False): + env = ExecTool()._build_env() + assert "OPENAI_API_KEY" not in env + assert "NANOBOT_TOKEN" not in env + for v in env.values(): + assert "secret" not in v.lower() + + +class TestBuildEnvWindows: + + _EXPECTED_KEYS = { + "SYSTEMROOT", "COMSPEC", "USERPROFILE", "HOMEDRIVE", + "HOMEPATH", "TEMP", "TMP", "PATHEXT", "PATH", + } + + def test_expected_keys(self): + with patch("nanobot.agent.tools.shell._IS_WINDOWS", True): + env = ExecTool()._build_env() + assert set(env) == self._EXPECTED_KEYS + + def test_secrets_excluded(self, monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-secret") + monkeypatch.setenv("NANOBOT_TOKEN", "tok-secret") + with patch("nanobot.agent.tools.shell._IS_WINDOWS", True): + env = ExecTool()._build_env() + assert "OPENAI_API_KEY" not in env + assert "NANOBOT_TOKEN" not in env + for v in env.values(): + assert "secret" not in v.lower() + + def test_path_has_sensible_default(self): + with ( + patch("nanobot.agent.tools.shell._IS_WINDOWS", True), + patch.dict("os.environ", {}, clear=True), + ): + env = ExecTool()._build_env() + assert "system32" in env["PATH"].lower() + + def test_systemroot_forwarded(self, monkeypatch): + monkeypatch.setenv("SYSTEMROOT", r"D:\Windows") + with patch("nanobot.agent.tools.shell._IS_WINDOWS", True): + env = ExecTool()._build_env() + assert env["SYSTEMROOT"] == r"D:\Windows" + + +# --------------------------------------------------------------------------- +# _spawn +# --------------------------------------------------------------------------- + +class TestSpawnUnix: + + @pytest.mark.asyncio + async def test_uses_bash(self): + with ( + patch("nanobot.agent.tools.shell._IS_WINDOWS", False), + patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec, + ): + mock_exec.return_value = AsyncMock() + await ExecTool._spawn("echo hi", "/tmp", {"HOME": "/tmp"}) + + args = mock_exec.call_args[0] + assert "bash" in args[0] + assert "-l" in args + assert "-c" in args + assert "echo hi" in args + + +class TestSpawnWindows: + + @pytest.mark.asyncio + async def test_uses_comspec_from_env(self): + env = {"COMSPEC": r"C:\Windows\system32\cmd.exe", "PATH": ""} + with ( + patch("nanobot.agent.tools.shell._IS_WINDOWS", True), + patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec, + ): + mock_exec.return_value = AsyncMock() + await ExecTool._spawn("dir", r"C:\Users", env) + + args = mock_exec.call_args[0] + assert "cmd.exe" in args[0] + assert "/c" in args + assert "dir" in args + + @pytest.mark.asyncio + async def test_falls_back_to_default_comspec(self): + env = {"PATH": ""} + with ( + patch("nanobot.agent.tools.shell._IS_WINDOWS", True), + patch.dict("os.environ", {}, clear=True), + patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec, + ): + mock_exec.return_value = AsyncMock() + await ExecTool._spawn("dir", r"C:\Users", env) + + args = mock_exec.call_args[0] + assert args[0] == "cmd.exe" + + +# --------------------------------------------------------------------------- +# path_append +# --------------------------------------------------------------------------- + +class TestPathAppendPlatform: + + @pytest.mark.asyncio + async def test_unix_injects_export(self): + """On Unix, path_append is an export statement prepended to command.""" + mock_proc = AsyncMock() + mock_proc.communicate.return_value = (b"ok", b"") + mock_proc.returncode = 0 + + with ( + patch("nanobot.agent.tools.shell._IS_WINDOWS", False), + patch.object(ExecTool, "_spawn", return_value=mock_proc) as mock_spawn, + patch.object(ExecTool, "_guard_command", return_value=None), + ): + tool = ExecTool(path_append="/opt/bin") + await tool.execute(command="ls") + + spawned_cmd = mock_spawn.call_args[0][0] + assert 'export PATH="$PATH:/opt/bin"' in spawned_cmd + assert spawned_cmd.endswith("ls") + + @pytest.mark.asyncio + async def test_windows_modifies_env(self): + """On Windows, path_append is appended to PATH in the env dict.""" + mock_proc = AsyncMock() + mock_proc.communicate.return_value = (b"ok", b"") + mock_proc.returncode = 0 + + captured_env = {} + + async def capture_spawn(cmd, cwd, env): + captured_env.update(env) + return mock_proc + + with ( + patch("nanobot.agent.tools.shell._IS_WINDOWS", True), + patch.object(ExecTool, "_spawn", side_effect=capture_spawn), + patch.object(ExecTool, "_guard_command", return_value=None), + ): + tool = ExecTool(path_append=r"C:\tools\bin") + await tool.execute(command="dir") + + assert captured_env["PATH"].endswith(r";C:\tools\bin") + + +# --------------------------------------------------------------------------- +# sandbox +# --------------------------------------------------------------------------- + +class TestSandboxPlatform: + + @pytest.mark.asyncio + async def test_bwrap_skipped_on_windows(self): + """bwrap must be silently skipped on Windows, not crash.""" + mock_proc = AsyncMock() + mock_proc.communicate.return_value = (b"ok", b"") + mock_proc.returncode = 0 + + with ( + patch("nanobot.agent.tools.shell._IS_WINDOWS", True), + patch.object(ExecTool, "_spawn", return_value=mock_proc) as mock_spawn, + patch.object(ExecTool, "_guard_command", return_value=None), + ): + tool = ExecTool(sandbox="bwrap") + result = await tool.execute(command="dir") + + assert "ok" in result + spawned_cmd = mock_spawn.call_args[0][0] + assert "bwrap" not in spawned_cmd + + @pytest.mark.asyncio + async def test_bwrap_applied_on_unix(self): + """On Unix, sandbox wrapping should still happen normally.""" + mock_proc = AsyncMock() + mock_proc.communicate.return_value = (b"sandboxed", b"") + mock_proc.returncode = 0 + + with ( + patch("nanobot.agent.tools.shell._IS_WINDOWS", False), + patch("nanobot.agent.tools.shell.wrap_command", return_value="bwrap -- sh -c ls") as mock_wrap, + patch.object(ExecTool, "_spawn", return_value=mock_proc) as mock_spawn, + patch.object(ExecTool, "_guard_command", return_value=None), + ): + tool = ExecTool(sandbox="bwrap", working_dir="/workspace") + await tool.execute(command="ls") + + mock_wrap.assert_called_once() + spawned_cmd = mock_spawn.call_args[0][0] + assert "bwrap" in spawned_cmd + + +# --------------------------------------------------------------------------- +# end-to-end (mocked subprocess, full execute path) +# --------------------------------------------------------------------------- + +class TestExecuteEndToEnd: + + @pytest.mark.asyncio + async def test_windows_full_path(self): + """Full execute() flow on Windows: env, spawn, output formatting.""" + mock_proc = AsyncMock() + mock_proc.communicate.return_value = (b"hello world\r\n", b"") + mock_proc.returncode = 0 + + with ( + patch("nanobot.agent.tools.shell._IS_WINDOWS", True), + patch.object(ExecTool, "_spawn", return_value=mock_proc), + patch.object(ExecTool, "_guard_command", return_value=None), + ): + tool = ExecTool() + result = await tool.execute(command="echo hello world") + + assert "hello world" in result + assert "Exit code: 0" in result + + @pytest.mark.asyncio + async def test_unix_full_path(self): + """Full execute() flow on Unix: env, spawn, output formatting.""" + mock_proc = AsyncMock() + mock_proc.communicate.return_value = (b"hello world\n", b"") + mock_proc.returncode = 0 + + with ( + patch("nanobot.agent.tools.shell._IS_WINDOWS", False), + patch.object(ExecTool, "_spawn", return_value=mock_proc), + patch.object(ExecTool, "_guard_command", return_value=None), + ): + tool = ExecTool() + result = await tool.execute(command="echo hello world") + + assert "hello world" in result + assert "Exit code: 0" in result