mirror of
https://github.com/AstrBotDevs/AstrBot.git
synced 2026-09-01 15:32:49 +08:00
feat: use PowerShell for Windows local shell (#9471)
This commit is contained in:
@@ -445,8 +445,9 @@ def _apply_local_env_tools(req: ProviderRequest, plugin_context: Context) -> Non
|
||||
def _build_local_mode_prompt() -> str:
|
||||
system_name = platform.system() or "Unknown"
|
||||
shell_hint = (
|
||||
"The runtime shell is Windows Command Prompt (cmd.exe). "
|
||||
"Use cmd-compatible commands and do not assume Unix commands like cat/ls/grep are available."
|
||||
"The runtime shell is Windows PowerShell 5.1 (powershell.exe). "
|
||||
"Use Windows PowerShell 5.1-compatible syntax and cmdlets; do not use "
|
||||
"PowerShell 7-only syntax or assume Unix commands like cat/ls/grep are available."
|
||||
if system_name.lower() == "windows"
|
||||
else "The runtime shell is Unix-like. Use POSIX-compatible shell commands."
|
||||
)
|
||||
|
||||
@@ -140,25 +140,35 @@ class LocalShellComponent(ShellComponent):
|
||||
if env:
|
||||
run_env.update({str(k): str(v) for k, v in env.items()})
|
||||
working_dir = os.path.abspath(cwd) if cwd else get_astrbot_root()
|
||||
if background:
|
||||
# `command` is intentionally executed through the current shell so
|
||||
# local computer-use behavior matches existing tool semantics.
|
||||
# Safety relies on `_is_safe_command()` and the allowed-root checks.
|
||||
proc = subprocess.Popen( # noqa: S602 # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit
|
||||
popen_command: str | list[str] = command
|
||||
popen_shell = shell
|
||||
if sys.platform == "win32" and shell:
|
||||
popen_command = [
|
||||
"powershell.exe",
|
||||
"-NoLogo",
|
||||
"-NoProfile",
|
||||
"-NonInteractive",
|
||||
"-Command",
|
||||
command,
|
||||
shell=shell,
|
||||
]
|
||||
popen_shell = False
|
||||
if background:
|
||||
# Shell commands use PowerShell 5.1 on Windows and the platform
|
||||
# shell elsewhere. Safety relies on `_is_safe_command()`.
|
||||
proc = subprocess.Popen( # noqa: S602 # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit
|
||||
popen_command,
|
||||
shell=popen_shell,
|
||||
cwd=working_dir,
|
||||
env=run_env,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
return {"pid": proc.pid, "stdout": "", "stderr": "", "exit_code": None}
|
||||
# `command` is intentionally executed through the current shell so
|
||||
# local computer-use behavior matches existing tool semantics.
|
||||
# Safety relies on `_is_safe_command()` and the allowed-root checks.
|
||||
# Shell commands use PowerShell 5.1 on Windows and the platform shell
|
||||
# elsewhere. Safety relies on `_is_safe_command()`.
|
||||
proc = subprocess.Popen( # noqa: S602 # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit
|
||||
command,
|
||||
shell=shell,
|
||||
popen_command,
|
||||
shell=popen_shell,
|
||||
cwd=working_dir,
|
||||
env=run_env,
|
||||
stdout=subprocess.PIPE,
|
||||
@@ -247,7 +257,7 @@ class LocalShellComponent(ShellComponent):
|
||||
output_path.touch()
|
||||
|
||||
process_kwargs: dict[str, Any] = {}
|
||||
if os.name == "nt":
|
||||
if sys.platform == "win32":
|
||||
process_kwargs["creationflags"] = getattr(
|
||||
subprocess,
|
||||
"CREATE_NEW_PROCESS_GROUP",
|
||||
@@ -257,8 +267,21 @@ class LocalShellComponent(ShellComponent):
|
||||
process_kwargs["start_new_session"] = True
|
||||
|
||||
try:
|
||||
process = await asyncio.create_subprocess_shell(
|
||||
command,
|
||||
if sys.platform == "win32":
|
||||
process_factory = asyncio.create_subprocess_exec
|
||||
process_args = (
|
||||
"powershell.exe",
|
||||
"-NoLogo",
|
||||
"-NoProfile",
|
||||
"-NonInteractive",
|
||||
"-Command",
|
||||
command,
|
||||
)
|
||||
else:
|
||||
process_factory = asyncio.create_subprocess_shell
|
||||
process_args = (command,)
|
||||
process = await process_factory(
|
||||
*process_args,
|
||||
cwd=working_dir,
|
||||
env=run_env,
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
|
||||
@@ -51,6 +51,108 @@ def test_local_shell_component_decodes_utf8_output(monkeypatch):
|
||||
assert result["exit_code"] == 0
|
||||
|
||||
|
||||
def test_local_shell_component_uses_windows_powershell(monkeypatch):
|
||||
calls = []
|
||||
|
||||
def fake_run(*args, **kwargs):
|
||||
calls.append((args, kwargs))
|
||||
return _FakePopen(stdout=b"")
|
||||
|
||||
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
||||
monkeypatch.setattr(local_booter.sys, "platform", "win32")
|
||||
|
||||
result = asyncio.run(LocalShellComponent().exec("Get-ChildItem"))
|
||||
|
||||
assert result["exit_code"] == 0
|
||||
assert calls[0][0][0] == [
|
||||
"powershell.exe",
|
||||
"-NoLogo",
|
||||
"-NoProfile",
|
||||
"-NonInteractive",
|
||||
"-Command",
|
||||
"Get-ChildItem",
|
||||
]
|
||||
assert calls[0][1]["shell"] is False
|
||||
|
||||
|
||||
def test_local_shell_component_keeps_platform_shell_outside_windows(monkeypatch):
|
||||
calls = []
|
||||
|
||||
def fake_run(*args, **kwargs):
|
||||
calls.append((args, kwargs))
|
||||
return _FakePopen(stdout=b"")
|
||||
|
||||
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
||||
monkeypatch.setattr(local_booter.sys, "platform", "linux")
|
||||
|
||||
result = asyncio.run(LocalShellComponent().exec("pwd"))
|
||||
|
||||
assert result["exit_code"] == 0
|
||||
assert calls[0][0][0] == "pwd"
|
||||
assert calls[0][1]["shell"] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_managed_shell_uses_windows_powershell(monkeypatch, tmp_path):
|
||||
calls = []
|
||||
|
||||
class FakeStdout:
|
||||
def __init__(self):
|
||||
self.chunks = [b"done\n", b""]
|
||||
|
||||
async def read(self, _limit):
|
||||
return self.chunks.pop(0)
|
||||
|
||||
class FakeProcess:
|
||||
def __init__(self):
|
||||
self.pid = 12345
|
||||
self.returncode = None
|
||||
self.stdout = FakeStdout()
|
||||
self.stdin = None
|
||||
|
||||
async def wait(self):
|
||||
self.returncode = 0
|
||||
return 0
|
||||
|
||||
async def fake_create_subprocess_exec(*args, **kwargs):
|
||||
calls.append((args, kwargs))
|
||||
return FakeProcess()
|
||||
|
||||
async def fail_create_subprocess_shell(*_args, **_kwargs):
|
||||
raise AssertionError("Windows managed commands must not use cmd.exe.")
|
||||
|
||||
monkeypatch.setattr(local_booter.sys, "platform", "win32")
|
||||
monkeypatch.setattr(
|
||||
local_booter.asyncio,
|
||||
"create_subprocess_exec",
|
||||
fake_create_subprocess_exec,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
local_booter.asyncio,
|
||||
"create_subprocess_shell",
|
||||
fail_create_subprocess_shell,
|
||||
)
|
||||
|
||||
result = await LocalShellComponent().exec_managed(
|
||||
"Get-ChildItem",
|
||||
owner_id="owner-a",
|
||||
cwd=str(tmp_path),
|
||||
yield_time_ms=5_000,
|
||||
)
|
||||
|
||||
assert result["status"] == "completed"
|
||||
assert result["stdout"] == "done\n"
|
||||
assert calls[0][0] == (
|
||||
"powershell.exe",
|
||||
"-NoLogo",
|
||||
"-NoProfile",
|
||||
"-NonInteractive",
|
||||
"-Command",
|
||||
"Get-ChildItem",
|
||||
)
|
||||
assert "creationflags" in calls[0][1]
|
||||
|
||||
|
||||
def test_local_shell_component_prefers_utf8_before_windows_locale(
|
||||
monkeypatch,
|
||||
):
|
||||
|
||||
@@ -169,6 +169,24 @@ def test_append_system_reminders_includes_weekday(mock_event):
|
||||
]
|
||||
|
||||
|
||||
def test_local_mode_prompt_uses_windows_powershell_51():
|
||||
with patch("astrbot.core.astr_main_agent.platform.system", return_value="Windows"):
|
||||
prompt = ama._build_local_mode_prompt()
|
||||
|
||||
assert "Windows PowerShell 5.1 (powershell.exe)" in prompt
|
||||
assert "PowerShell 7-only syntax" in prompt
|
||||
assert "cmd.exe" not in prompt
|
||||
|
||||
|
||||
def test_local_mode_prompt_keeps_posix_shell_guidance():
|
||||
with patch("astrbot.core.astr_main_agent.platform.system", return_value="Linux"):
|
||||
prompt = ama._build_local_mode_prompt()
|
||||
|
||||
assert "Unix-like" in prompt
|
||||
assert "POSIX-compatible" in prompt
|
||||
assert "PowerShell" not in prompt
|
||||
|
||||
|
||||
class TestMainAgentBuildConfig:
|
||||
"""Tests for MainAgentBuildConfig dataclass."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user