mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-04-08 12:13:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(exec): add Windows support for shell command execution

Browse Source 
ExecTool hardcoded bash, breaking exec on Windows. Now uses cmd.exe
via COMSPEC on Windows with a curated minimal env (PATH, SYSTEMROOT,
etc.) that excludes secrets. bwrap sandbox gracefully skips on Windows.
This commit is contained in:
Xubin Ren 2026-04-07 08:24:11 +00:00 committed by Xubin Ren
parent 63acfc4f2f
commit ef0284a4e0
3 changed files with 335 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
nanobot/agent/tools/shell.py
View File

@ -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,

7
tests/tools/test_exec_env.py
View File

@ -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."""

269
tests/tools/test_exec_platform.py Normal file
View File

@ -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