fix: harden Windows exec shell edge cases
Maintainer edit: preserve raw cmd.exe quoting for shell='cmd', propagate native exit codes through the default PowerShell path, and keep quoted Windows executable paths invokable under PowerShell.
This commit is contained in:
@@ -491,14 +491,17 @@ class ExecTool(Tool):
|
|||||||
program = shell_program or default_program
|
program = shell_program or default_program
|
||||||
program_name = PureWindowsPath(program).name.lower()
|
program_name = PureWindowsPath(program).name.lower()
|
||||||
if program_name in ("cmd", "cmd.exe"):
|
if program_name in ("cmd", "cmd.exe"):
|
||||||
return await asyncio.create_subprocess_exec(
|
cmd_env = {**env, "COMSPEC": program}
|
||||||
program, "/c", command,
|
return await asyncio.create_subprocess_shell(
|
||||||
|
command,
|
||||||
stdin=stdin,
|
stdin=stdin,
|
||||||
stdout=asyncio.subprocess.PIPE,
|
stdout=asyncio.subprocess.PIPE,
|
||||||
stderr=asyncio.subprocess.PIPE,
|
stderr=asyncio.subprocess.PIPE,
|
||||||
cwd=cwd,
|
cwd=cwd,
|
||||||
env=env,
|
env=cmd_env,
|
||||||
)
|
)
|
||||||
|
command = ExecTool._normalize_powershell_command(command)
|
||||||
|
command = f"{command}\nif ($LASTEXITCODE -ne $null) {{ exit $LASTEXITCODE }}"
|
||||||
return await asyncio.create_subprocess_exec(
|
return await asyncio.create_subprocess_exec(
|
||||||
program, "-NoProfile", "-NonInteractive", "-Command", command,
|
program, "-NoProfile", "-NonInteractive", "-Command", command,
|
||||||
stdin=stdin,
|
stdin=stdin,
|
||||||
@@ -522,6 +525,29 @@ class ExecTool(Tool):
|
|||||||
env=env,
|
env=env,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _normalize_powershell_command(command: str) -> str:
|
||||||
|
stripped = command.lstrip()
|
||||||
|
if not stripped or stripped[0] not in {"'", '"'}:
|
||||||
|
return command
|
||||||
|
|
||||||
|
quote = stripped[0]
|
||||||
|
end = stripped.find(quote, 1)
|
||||||
|
if end == -1 or end + 1 >= len(stripped) or not stripped[end + 1].isspace():
|
||||||
|
return command
|
||||||
|
|
||||||
|
executable = stripped[1:end]
|
||||||
|
looks_like_windows_executable = (
|
||||||
|
bool(re.match(r"^[A-Za-z]:[\\/]", executable))
|
||||||
|
or executable.startswith(r"\\")
|
||||||
|
or executable.lower().endswith((".exe", ".cmd", ".bat", ".ps1"))
|
||||||
|
)
|
||||||
|
if not looks_like_windows_executable:
|
||||||
|
return command
|
||||||
|
|
||||||
|
leading = command[: len(command) - len(stripped)]
|
||||||
|
return f"{leading}& {stripped}"
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _resolve_shell(shell: str | None) -> tuple[str | None, str | None]:
|
def _resolve_shell(shell: str | None) -> tuple[str | None, str | None]:
|
||||||
if not shell:
|
if not shell:
|
||||||
|
|||||||
@@ -175,23 +175,57 @@ class TestSpawnWindows:
|
|||||||
assert kwargs["env"] == env
|
assert kwargs["env"] == env
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_explicit_cmd_shell_uses_cmd_c(self):
|
async def test_explicit_cmd_shell_uses_raw_shell_string(self):
|
||||||
"""Explicit shell='cmd' should launch the resolved cmd.exe with /c."""
|
"""Explicit shell='cmd' should preserve raw cmd.exe quoting semantics."""
|
||||||
env = {"COMSPEC": r"C:\Windows\system32\cmd.exe", "PATH": ""}
|
env = {"COMSPEC": r"C:\Windows\system32\cmd.exe", "PATH": ""}
|
||||||
|
with (
|
||||||
|
patch("nanobot.agent.tools.shell._IS_WINDOWS", True),
|
||||||
|
patch("asyncio.create_subprocess_shell", new_callable=AsyncMock) as mock_shell,
|
||||||
|
):
|
||||||
|
mock_shell.return_value = AsyncMock()
|
||||||
|
await ExecTool._spawn(
|
||||||
|
'echo "a & b"', r"C:\work", env,
|
||||||
|
shell_program=r"C:\Windows\system32\cmd.exe",
|
||||||
|
)
|
||||||
|
|
||||||
|
args = mock_shell.call_args[0]
|
||||||
|
assert args == ('echo "a & b"',)
|
||||||
|
kwargs = mock_shell.call_args[1]
|
||||||
|
assert kwargs["cwd"] == r"C:\work"
|
||||||
|
assert kwargs["env"] == {
|
||||||
|
"COMSPEC": r"C:\Windows\system32\cmd.exe",
|
||||||
|
"PATH": "",
|
||||||
|
}
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_powershell_preserves_last_native_exit_code(self):
|
||||||
|
"""PowerShell -Command should forward native process exit codes."""
|
||||||
|
env = {"PATH": ""}
|
||||||
with (
|
with (
|
||||||
patch("nanobot.agent.tools.shell._IS_WINDOWS", True),
|
patch("nanobot.agent.tools.shell._IS_WINDOWS", True),
|
||||||
patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec,
|
patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec,
|
||||||
):
|
):
|
||||||
mock_exec.return_value = AsyncMock()
|
mock_exec.return_value = AsyncMock()
|
||||||
await ExecTool._spawn(
|
await ExecTool._spawn("cmd /c exit 7", r"C:\work", env)
|
||||||
"dir", r"C:\work", env,
|
|
||||||
shell_program=r"C:\Windows\system32\cmd.exe",
|
|
||||||
)
|
|
||||||
|
|
||||||
args = mock_exec.call_args[0]
|
command = mock_exec.call_args[0][-1]
|
||||||
assert args[:3] == (r"C:\Windows\system32\cmd.exe", "/c", "dir")
|
assert "cmd /c exit 7" in command
|
||||||
kwargs = mock_exec.call_args[1]
|
assert "if ($LASTEXITCODE -ne $null) { exit $LASTEXITCODE }" in command
|
||||||
assert kwargs["cwd"] == r"C:\work"
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_powershell_invokes_quoted_windows_executable_path(self):
|
||||||
|
"""PowerShell needs & before quoted executable paths with arguments."""
|
||||||
|
env = {"PATH": ""}
|
||||||
|
command = r'"D:\Program Files\Python\python.exe" -u -c "print(1)"'
|
||||||
|
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(command, r"C:\work", env)
|
||||||
|
|
||||||
|
powershell_command = mock_exec.call_args[0][-1]
|
||||||
|
assert powershell_command.startswith("& " + command)
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_prefers_pwsh_when_available(self):
|
async def test_prefers_pwsh_when_available(self):
|
||||||
@@ -645,23 +679,23 @@ class TestResolveShellWindows:
|
|||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_shell_cmd_accepted(self):
|
async def test_shell_cmd_accepted(self):
|
||||||
"""shell='cmd' should use the resolved cmd.exe with /c."""
|
"""shell='cmd' should preserve the command string for cmd.exe parsing."""
|
||||||
mock_proc = AsyncMock()
|
mock_proc = AsyncMock()
|
||||||
mock_proc.communicate.return_value = (b"hello\n", b"")
|
mock_proc.communicate.return_value = (b'"a & b"\n', b"")
|
||||||
mock_proc.returncode = 0
|
mock_proc.returncode = 0
|
||||||
|
|
||||||
with (
|
with (
|
||||||
patch("nanobot.agent.tools.shell._IS_WINDOWS", True),
|
patch("nanobot.agent.tools.shell._IS_WINDOWS", True),
|
||||||
patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec,
|
patch("asyncio.create_subprocess_shell", new_callable=AsyncMock) as mock_shell,
|
||||||
patch.object(ExecTool, "_guard_command", return_value=None),
|
patch.object(ExecTool, "_guard_command", return_value=None),
|
||||||
):
|
):
|
||||||
mock_exec.return_value = mock_proc
|
mock_shell.return_value = mock_proc
|
||||||
tool = ExecTool()
|
tool = ExecTool()
|
||||||
result = await tool.execute(command="echo hello", shell="cmd")
|
result = await tool.execute(command='echo "a & b"', shell="cmd")
|
||||||
|
|
||||||
assert "hello" in result
|
assert '"a & b"' in result
|
||||||
args = mock_exec.call_args[0]
|
args = mock_shell.call_args[0]
|
||||||
assert args[1:3] == ("/c", "echo hello")
|
assert args == ('echo "a & b"',)
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_shell_bash_rejected_on_windows(self):
|
async def test_shell_bash_rejected_on_windows(self):
|
||||||
@@ -675,11 +709,12 @@ class TestResolveShellWindows:
|
|||||||
|
|
||||||
|
|
||||||
@pytest.mark.skipif(
|
@pytest.mark.skipif(
|
||||||
sys.platform != "win32" or shutil.which("pwsh") is None,
|
sys.platform != "win32",
|
||||||
reason="requires Windows with PowerShell 7",
|
reason="requires Windows",
|
||||||
)
|
)
|
||||||
class TestWindowsRealExec:
|
class TestWindowsRealExec:
|
||||||
|
|
||||||
|
@pytest.mark.skipif(shutil.which("pwsh") is None, reason="requires PowerShell 7")
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_single_line_and_separator_uses_pwsh(self):
|
async def test_single_line_and_separator_uses_pwsh(self):
|
||||||
result = await ExecTool(timeout=10).execute(command="echo before && echo after")
|
result = await ExecTool(timeout=10).execute(command="echo before && echo after")
|
||||||
@@ -687,3 +722,17 @@ class TestWindowsRealExec:
|
|||||||
assert "before" in result
|
assert "before" in result
|
||||||
assert "after" in result
|
assert "after" in result
|
||||||
assert "Exit code: 0" in result
|
assert "Exit code: 0" in result
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_explicit_cmd_preserves_embedded_quotes(self):
|
||||||
|
result = await ExecTool(timeout=10).execute(command='echo "a & b"', shell="cmd")
|
||||||
|
|
||||||
|
assert '"a & b"' in result
|
||||||
|
assert r'\"a & b\"' not in result
|
||||||
|
assert "Exit code: 0" in result
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_default_powershell_preserves_native_exit_code(self):
|
||||||
|
result = await ExecTool(timeout=10).execute(command="cmd /c exit 7")
|
||||||
|
|
||||||
|
assert "Exit code: 7" in result
|
||||||
|
|||||||
Reference in New Issue
Block a user