wehub-resource-sync
1556 lines
49 KiB
Python
1556 lines
49 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from opensquilla.tools.types import CallerKind, ToolContext, current_tool_context
|
|
|
|
|
|
def _windows_runtime() -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
effective=SimpleNamespace(sandbox_enabled=True),
|
|
backend=SimpleNamespace(name="windows_default"),
|
|
)
|
|
|
|
|
|
def _configure_approval_queue(
|
|
monkeypatch,
|
|
tmp_path,
|
|
mode: str,
|
|
*,
|
|
allow_patterns: list[str] | None = None,
|
|
deny_patterns: list[str] | None = None,
|
|
) -> None:
|
|
from opensquilla.application import approval_queue as approval_queue_mod
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
|
|
monkeypatch.setattr(
|
|
approval_queue_mod,
|
|
"_DEFAULT_APPROVAL_QUEUE_PATH",
|
|
tmp_path / "approval_queue.sqlite",
|
|
)
|
|
reset_approval_queue()
|
|
get_approval_queue().set_settings(
|
|
mode,
|
|
allow_patterns=allow_patterns,
|
|
deny_patterns=deny_patterns,
|
|
)
|
|
|
|
|
|
def test_windows_exec_command_uses_shell_host_wrapper(monkeypatch, tmp_path) -> None:
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime = _windows_runtime()
|
|
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"_trusted_windows_powershell_path",
|
|
lambda: r"C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe",
|
|
)
|
|
|
|
argv = shell._sandbox_shell_backend_argv("Write-Output ok", runtime, cwd=tmp_path)
|
|
|
|
assert argv[:3] == (
|
|
sys.executable,
|
|
"-c",
|
|
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
|
|
)
|
|
assert argv[3] == r"C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe"
|
|
assert argv[4] == "Write-Output ok"
|
|
assert argv[5] == str(tmp_path)
|
|
assert argv[6] == str(tmp_path / ".opensquilla-cache" / "shell-host")
|
|
|
|
|
|
def test_windows_exec_command_unwraps_nested_powershell_command(monkeypatch, tmp_path) -> None:
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime = _windows_runtime()
|
|
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"_trusted_windows_powershell_path",
|
|
lambda: r"C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe",
|
|
)
|
|
|
|
argv = shell._sandbox_shell_backend_argv(
|
|
(
|
|
"& 'C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe' "
|
|
'-NoProfile -Command "Write-Output child-ok"'
|
|
),
|
|
runtime,
|
|
cwd=tmp_path,
|
|
)
|
|
|
|
assert argv[:3] == (
|
|
sys.executable,
|
|
"-c",
|
|
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
|
|
)
|
|
assert argv[3] == r"C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe"
|
|
assert argv[4] == "Write-Output child-ok"
|
|
|
|
direct_argv = shell._sandbox_shell_backend_argv(
|
|
'powershell.exe -NoProfile -Command "Write-Output child-ok"',
|
|
runtime,
|
|
cwd=tmp_path,
|
|
)
|
|
|
|
assert direct_argv[4] == "Write-Output child-ok"
|
|
|
|
|
|
def test_windows_exec_command_prefers_cmd_package_manager_shims(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime = _windows_runtime()
|
|
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"_trusted_windows_powershell_path",
|
|
lambda: r"C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe",
|
|
)
|
|
|
|
argv = shell._sandbox_shell_backend_argv("npm view lodash version", runtime, cwd=tmp_path)
|
|
|
|
assert argv[4] == "& 'npm.cmd' 'view' 'lodash' 'version'"
|
|
|
|
|
|
def test_windows_shell_host_skips_windowsapps_git_alias(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
if os.name != "nt":
|
|
pytest.skip("Windows command shims are Windows-only")
|
|
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
alias_dir = tmp_path / "Microsoft" / "WindowsApps"
|
|
real_dir = tmp_path / "Git" / "cmd"
|
|
alias_dir.mkdir(parents=True)
|
|
real_dir.mkdir(parents=True)
|
|
(alias_dir / "git.cmd").write_text("@echo off\r\necho alias-git\r\n", encoding="utf-8")
|
|
(real_dir / "git.cmd").write_text("@echo off\r\necho real-git %1\r\n", encoding="utf-8")
|
|
monkeypatch.setenv("PATH", os.pathsep.join((str(alias_dir), str(real_dir))))
|
|
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
|
|
"powershell.exe",
|
|
"git --version",
|
|
str(tmp_path),
|
|
str(tmp_path / ".opensquilla-cache" / "shell-host"),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert completed.returncode == 0
|
|
assert "real-git --version" in completed.stdout
|
|
assert "alias-git" not in completed.stdout
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_approve_owner_host_effect_command_runs_on_host_once(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, "auto-approve")
|
|
runtime = _windows_runtime()
|
|
host_calls: list[dict[str, object]] = []
|
|
|
|
async def fail_gate_action(**kwargs):
|
|
raise AssertionError("auto host escalation should not enter sandbox gate")
|
|
|
|
async def fake_host(command, **kwargs):
|
|
host_calls.append({"command": command, **kwargs})
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", fail_gate_action)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command("winget install Tencent.QQ.NT", workdir=str(tmp_path))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
assert "host-ok" in result
|
|
assert [call["command"] for call in host_calls] == ["winget install Tencent.QQ.NT"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_owner_host_effect_command_runs_on_host_without_legacy_auto_approve(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, "prompt")
|
|
runtime = _windows_runtime()
|
|
host_calls: list[dict[str, object]] = []
|
|
|
|
async def fail_gate_action(**kwargs):
|
|
raise AssertionError("trusted host-effect command should not enter sandbox gate")
|
|
|
|
async def fake_host(command, **kwargs):
|
|
host_calls.append({"command": command, **kwargs})
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", fail_gate_action)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command("winget install Tencent.QQ.NT", workdir=str(tmp_path))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
assert "host-ok" in result
|
|
assert [call["command"] for call in host_calls] == ["winget install Tencent.QQ.NT"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_powershell_file_with_installer_actions_runs_on_host(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
script_path = tmp_path / "install_dingtalk.ps1"
|
|
script_path.write_text(
|
|
"\n".join(
|
|
[
|
|
"$installer = Join-Path $PSScriptRoot 'DingTalkSetup.exe'",
|
|
(
|
|
"Invoke-WebRequest "
|
|
"https://dtapp-pub.dingtalk.com/desktop/Win/Release/DingTalkSetup.exe "
|
|
"-OutFile $installer"
|
|
),
|
|
"Start-Process $installer -Wait",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
_configure_approval_queue(monkeypatch, tmp_path, "prompt")
|
|
runtime = _windows_runtime()
|
|
host_calls: list[dict[str, object]] = []
|
|
|
|
async def fail_gate_action(**kwargs):
|
|
raise AssertionError("trusted installer script should not enter sandbox gate")
|
|
|
|
async def fake_host(command, **kwargs):
|
|
host_calls.append({"command": command, **kwargs})
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", fail_gate_action)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
command = f'powershell -NoProfile -ExecutionPolicy Bypass -File "{script_path}"'
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command(command, workdir=str(tmp_path))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
assert "host-ok" in result
|
|
assert [call["command"] for call in host_calls] == [command]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"command",
|
|
[
|
|
"Get-WmiObject Win32_Product | Where-Object { $_.Name -like '*DingTalk*' }",
|
|
(
|
|
"Get-ItemProperty HKCU:\\Software\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\* "
|
|
"| Where-Object { $_.DisplayName -like '*DingTalk*' }"
|
|
),
|
|
'Start-Process "C:\\Program Files\\DingTalk\\uninst.exe" -Wait',
|
|
"Stop-Process -Name DingTalk -Force",
|
|
],
|
|
)
|
|
async def test_trusted_windows_uninstall_flow_commands_run_on_host(
|
|
monkeypatch,
|
|
tmp_path,
|
|
command,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, "prompt")
|
|
runtime = _windows_runtime()
|
|
host_calls: list[str] = []
|
|
|
|
async def fail_gate_action(**kwargs):
|
|
raise AssertionError("trusted uninstall flow command should not enter sandbox gate")
|
|
|
|
async def fake_host(command, **kwargs):
|
|
host_calls.append(command)
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", fail_gate_action)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command(command, workdir=str(tmp_path))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
assert "host-ok" in result
|
|
assert host_calls == [command]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_host_shell_folds_windows_env_key_duplicates(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
if os.name != "nt":
|
|
pytest.skip("Windows environment variables are case-insensitive")
|
|
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, "prompt")
|
|
runtime = _windows_runtime()
|
|
host_envs: list[dict[str, str]] = []
|
|
|
|
async def fake_host(command, **kwargs):
|
|
host_envs.append(dict(kwargs["env"]))
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setenv("SystemRoot", r"C:\Windows")
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command(
|
|
"winget uninstall DingTalk",
|
|
workdir=str(tmp_path),
|
|
env={"SYSTEMROOT": r"C:\Windows"},
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
assert "host-ok" in result
|
|
assert host_envs
|
|
systemroot_keys = [key for key in host_envs[0] if key.upper() == "SYSTEMROOT"]
|
|
assert systemroot_keys == ["SystemRoot"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_host_escalation_adds_user_windowsapps_to_host_path(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, "auto-approve")
|
|
runtime = _windows_runtime()
|
|
local_appdata = tmp_path / "Users" / "lrk" / "AppData" / "Local"
|
|
windows_apps = local_appdata / "Microsoft" / "WindowsApps"
|
|
windows_apps.mkdir(parents=True)
|
|
(windows_apps / "winget.exe").write_text("", encoding="utf-8")
|
|
bin_dir = tmp_path / "bin"
|
|
bin_dir.mkdir()
|
|
host_calls: list[dict[str, object]] = []
|
|
|
|
async def fail_gate_action(**kwargs):
|
|
raise AssertionError("auto host escalation should not enter sandbox gate")
|
|
|
|
async def fake_host(command, **kwargs):
|
|
host_calls.append({"command": command, **kwargs})
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setenv("LOCALAPPDATA", str(local_appdata))
|
|
monkeypatch.setenv("PATH", str(bin_dir))
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", fail_gate_action)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command("winget install Tencent.QQ.NT", workdir=str(tmp_path))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
assert "host-ok" in result
|
|
assert host_calls
|
|
path_entries = host_calls[0]["env"]["PATH"].split(";")
|
|
assert path_entries[-1] == str(windows_apps)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"command",
|
|
[
|
|
"winget install Tencent.QQ.NT",
|
|
'powershell -NoProfile -Command "winget install Tencent.QQ.NT"',
|
|
],
|
|
)
|
|
async def test_auto_host_escalation_honors_deny_patterns(
|
|
monkeypatch,
|
|
tmp_path,
|
|
command,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
_configure_approval_queue(
|
|
monkeypatch,
|
|
tmp_path,
|
|
"auto-approve",
|
|
deny_patterns=["winget *"],
|
|
)
|
|
runtime = _windows_runtime()
|
|
|
|
async def fail_gate_action(**kwargs):
|
|
raise AssertionError("denied host-effect command should not enter sandbox gate")
|
|
|
|
async def fail_host(command, **kwargs):
|
|
raise AssertionError("denied host-effect command should not run on host")
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", fail_gate_action)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fail_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command(command, workdir=str(tmp_path))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
payload = json.loads(result)
|
|
assert payload["status"] == "approval_denied"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_host_escalation_background_process_runs_on_host_once(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
class FakeStdout:
|
|
async def read(self, _size):
|
|
return b""
|
|
|
|
class FakeProcess:
|
|
stdout = FakeStdout()
|
|
returncode = 0
|
|
|
|
async def wait(self):
|
|
return 0
|
|
|
|
def terminate(self):
|
|
self.returncode = -15
|
|
|
|
def kill(self):
|
|
self.returncode = -9
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, "auto-approve")
|
|
runtime = _windows_runtime()
|
|
spawn_calls: list[dict[str, object]] = []
|
|
|
|
async def fail_gate_action(**kwargs):
|
|
raise AssertionError("auto host escalation should not enter sandbox gate")
|
|
|
|
async def fake_create_subprocess_shell(command, **kwargs):
|
|
spawn_calls.append({"command": command, **kwargs})
|
|
return FakeProcess()
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", fail_gate_action)
|
|
monkeypatch.setattr(shell.asyncio, "create_subprocess_shell", fake_create_subprocess_shell)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.background_process("msiexec /i app.msi", workdir=str(tmp_path))
|
|
session_id = result.splitlines()[0].split("=", 1)[1]
|
|
session = shell._bg_sessions[session_id]
|
|
if session.collector_task is not None:
|
|
await session.collector_task
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
shell._bg_sessions.clear()
|
|
reset_approval_queue()
|
|
|
|
assert [call["command"] for call in spawn_calls] == ["msiexec /i app.msi"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_host_background_batch_requests_write_path_access_before_host(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir()
|
|
outside = tmp_path / "outside-background.txt"
|
|
command = f"msiexec /i app.msi; Set-Content -Path {outside} -Value hi"
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, "prompt")
|
|
runtime = _windows_runtime()
|
|
|
|
async def fail_create_subprocess_shell(*args, **kwargs):
|
|
raise AssertionError("background auto-host batch should not spawn before path approval")
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell.asyncio, "create_subprocess_shell", fail_create_subprocess_shell)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(workspace),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.background_process(command, workdir=str(workspace))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
payload = json.loads(result)
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["access"] == "rw"
|
|
assert payload["path"] == str(outside.resolve(strict=False))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("mode", "is_owner"),
|
|
[
|
|
("prompt", False),
|
|
("auto-approve", False),
|
|
],
|
|
)
|
|
async def test_host_effect_command_stays_sandboxed_without_owner(
|
|
monkeypatch,
|
|
tmp_path,
|
|
mode,
|
|
is_owner,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.sandbox.config import SandboxSettings
|
|
from opensquilla.sandbox.policy import build_policy
|
|
from opensquilla.sandbox.types import SecurityLevel
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, mode)
|
|
runtime = _windows_runtime()
|
|
policy = build_policy(
|
|
SecurityLevel.STANDARD,
|
|
"shell.exec",
|
|
tmp_path,
|
|
SandboxSettings(
|
|
sandbox=True,
|
|
security_grading=True,
|
|
backend="windows_default",
|
|
network_default="none",
|
|
),
|
|
trusted=True,
|
|
)
|
|
request = SimpleNamespace(
|
|
cwd=tmp_path,
|
|
action_kind="shell.exec",
|
|
policy=policy,
|
|
reason="",
|
|
session_id="s1",
|
|
run_mode="trusted",
|
|
)
|
|
backend_calls: list[object] = []
|
|
host_calls: list[str] = []
|
|
|
|
async def fake_gate_action(**kwargs):
|
|
return object(), policy, request
|
|
|
|
async def fake_preflight(*args, **kwargs):
|
|
return None
|
|
|
|
async def fake_backend(request, *, runtime=None):
|
|
backend_calls.append(request)
|
|
return SimpleNamespace(returncode=0, stdout="sandbox-ok\n", stderr="", backend_notes=())
|
|
|
|
async def fake_host(command, **kwargs):
|
|
host_calls.append(command)
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", fake_gate_action)
|
|
monkeypatch.setattr(shell, "preflight_subprocess_managed_network", fake_preflight)
|
|
monkeypatch.setattr(shell, "_run_backend_with_managed_network", fake_backend)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=is_owner,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command("winget list", workdir=str(tmp_path))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
assert "sandbox-ok" in result
|
|
assert backend_calls
|
|
assert host_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standard_auto_approve_owner_host_effect_command_stays_sandboxed(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.sandbox.config import SandboxSettings
|
|
from opensquilla.sandbox.policy import build_policy
|
|
from opensquilla.sandbox.types import SecurityLevel
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, "auto-approve")
|
|
runtime = _windows_runtime()
|
|
policy = build_policy(
|
|
SecurityLevel.STANDARD,
|
|
"shell.exec",
|
|
tmp_path,
|
|
SandboxSettings(
|
|
sandbox=True,
|
|
security_grading=True,
|
|
backend="windows_default",
|
|
network_default="none",
|
|
),
|
|
trusted=False,
|
|
)
|
|
request = SimpleNamespace(
|
|
cwd=tmp_path,
|
|
action_kind="shell.exec",
|
|
policy=policy,
|
|
reason="",
|
|
session_id="s1",
|
|
run_mode="standard",
|
|
)
|
|
backend_calls: list[object] = []
|
|
host_calls: list[str] = []
|
|
|
|
async def fake_gate_action(**kwargs):
|
|
return object(), policy, request
|
|
|
|
async def fake_preflight(*args, **kwargs):
|
|
return None
|
|
|
|
async def fake_backend(request, *, runtime=None):
|
|
backend_calls.append(request)
|
|
return SimpleNamespace(returncode=0, stdout="sandbox-ok\n", stderr="", backend_notes=())
|
|
|
|
async def fake_host(command, **kwargs):
|
|
host_calls.append(command)
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", fake_gate_action)
|
|
monkeypatch.setattr(shell, "preflight_subprocess_managed_network", fake_preflight)
|
|
monkeypatch.setattr(shell, "_run_backend_with_managed_network", fake_backend)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="standard",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command("winget list", workdir=str(tmp_path))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
assert "sandbox-ok" in result
|
|
assert backend_calls
|
|
assert host_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_windows_host_probe_adds_user_windowsapps_to_host_env(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, "prompt")
|
|
runtime = _windows_runtime()
|
|
local_appdata = tmp_path / "Users" / "lrk" / "AppData" / "Local"
|
|
windows_apps = local_appdata / "Microsoft" / "WindowsApps"
|
|
windows_apps.mkdir(parents=True)
|
|
(windows_apps / "winget.exe").write_text("", encoding="utf-8")
|
|
bin_dir = tmp_path / "bin"
|
|
bin_dir.mkdir()
|
|
host_calls: list[dict[str, object]] = []
|
|
|
|
async def fake_gate_action(**kwargs):
|
|
raise AssertionError("trusted host probe should not enter sandbox gate")
|
|
|
|
async def fake_host(command, **kwargs):
|
|
host_calls.append({"command": command, **kwargs})
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setenv("LOCALAPPDATA", str(local_appdata))
|
|
monkeypatch.setenv("PATH", str(bin_dir))
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", fake_gate_action)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command("where winget", workdir=str(tmp_path))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
assert "host-ok" in result
|
|
assert host_calls
|
|
assert host_calls[0]["command"] == "where winget"
|
|
assert host_calls[0]["env"]["PATH"].split(";")[-1] == str(windows_apps)
|
|
|
|
|
|
def test_windows_shell_host_blocks_icmp_diagnostics_when_proxy_allowlist(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
if os.name != "nt":
|
|
pytest.skip("Windows command shims are Windows-only")
|
|
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
env = os.environ.copy()
|
|
env["OPENSQUILLA_SANDBOX_NETWORK"] = "proxy_allowlist"
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
|
|
"powershell.exe",
|
|
"ping",
|
|
str(tmp_path),
|
|
str(tmp_path / ".opensquilla-cache" / "shell-host"),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
check=False,
|
|
)
|
|
|
|
assert completed.returncode != 0
|
|
assert "blocks ICMP" in completed.stderr
|
|
|
|
|
|
def test_windows_shell_host_fallback_sets_powershell_proxy_defaults(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
if os.name != "nt":
|
|
pytest.skip("PowerShell proxy defaults are Windows-only")
|
|
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
proxy_url = "http://127.0.0.1:48123"
|
|
env = os.environ.copy()
|
|
env["HTTP_PROXY"] = proxy_url
|
|
env["HTTPS_PROXY"] = proxy_url
|
|
command = (
|
|
"try { "
|
|
"if ([System.Net.WebRequest]::DefaultWebProxy -and "
|
|
"[System.Net.WebRequest]::DefaultWebProxy.Address) { "
|
|
"Write-Output ([System.Net.WebRequest]::DefaultWebProxy.Address.AbsoluteUri) "
|
|
"} else { Write-Output 'NO_PROXY' } "
|
|
"} catch { Write-Output ('ERR:' + $_.Exception.Message) }"
|
|
)
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
|
|
"powershell.exe",
|
|
command,
|
|
str(tmp_path),
|
|
str(tmp_path / ".opensquilla-cache" / "shell-host"),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
check=False,
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert proxy_url in completed.stdout
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_windows_exec_command_does_not_mount_program_files_tools_per_command(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.sandbox.config import SandboxSettings
|
|
from opensquilla.sandbox.policy import build_policy
|
|
from opensquilla.sandbox.types import SecurityLevel
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime = _windows_runtime()
|
|
policy = build_policy(
|
|
SecurityLevel.STANDARD,
|
|
"shell.exec",
|
|
tmp_path,
|
|
SandboxSettings(
|
|
sandbox=True,
|
|
security_grading=True,
|
|
backend="windows_default",
|
|
network_default="none",
|
|
),
|
|
trusted=True,
|
|
)
|
|
node_root = tmp_path / "Program Files" / "nodejs"
|
|
git_root = tmp_path / "Program Files" / "Git"
|
|
node_root.mkdir(parents=True)
|
|
(git_root / "cmd").mkdir(parents=True)
|
|
|
|
(node_root / "npm.cmd").write_text("@echo off\r\n", encoding="utf-8")
|
|
(git_root / "cmd" / "git.exe").write_text("", encoding="utf-8")
|
|
|
|
request = SimpleNamespace(
|
|
cwd=tmp_path,
|
|
action_kind="shell.exec",
|
|
policy=policy,
|
|
reason="",
|
|
session_id="s1",
|
|
run_mode="trusted",
|
|
)
|
|
|
|
async def _fake_gate_action(**kwargs):
|
|
return object(), policy, request
|
|
|
|
async def _fake_preflight(*args, **kwargs):
|
|
return None
|
|
|
|
backend_requests = []
|
|
|
|
async def _fake_run_backend(request, *, runtime=None):
|
|
backend_requests.append(request)
|
|
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="", backend_notes=())
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", _fake_gate_action)
|
|
monkeypatch.setattr(shell, "preflight_subprocess_managed_network", _fake_preflight)
|
|
monkeypatch.setattr(shell, "_run_backend_with_managed_network", _fake_run_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command(
|
|
(
|
|
"npm view lodash version && "
|
|
"git ls-remote https://github.com/opensquilla/opensquilla.git HEAD"
|
|
),
|
|
workdir=str(tmp_path),
|
|
env={"PATH": f"{node_root}{os.pathsep}{git_root / 'cmd'}"},
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "ok" in result
|
|
assert backend_requests
|
|
mount_paths = {mount.host_path for mount in backend_requests[0].policy.mounts}
|
|
assert node_root not in mount_paths
|
|
assert git_root not in mount_paths
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_windows_exec_command_uses_shared_path_envelopes(monkeypatch, tmp_path) -> None:
|
|
from opensquilla.sandbox.config import SandboxSettings
|
|
from opensquilla.sandbox.policy import build_policy
|
|
from opensquilla.sandbox.types import SecurityLevel
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime = _windows_runtime()
|
|
policy = build_policy(
|
|
SecurityLevel.STANDARD,
|
|
"shell.exec",
|
|
tmp_path,
|
|
SandboxSettings(
|
|
sandbox=True,
|
|
security_grading=True,
|
|
backend="windows_default",
|
|
network_default="none",
|
|
),
|
|
trusted=True,
|
|
)
|
|
request = SimpleNamespace(
|
|
cwd=tmp_path,
|
|
action_kind="shell.exec",
|
|
policy=policy,
|
|
reason="",
|
|
session_id="s1",
|
|
run_mode="trusted",
|
|
)
|
|
|
|
async def _fake_gate_action(**kwargs):
|
|
return object(), policy, request
|
|
|
|
async def _fake_preflight(*args, **kwargs):
|
|
return None
|
|
|
|
backend_called = False
|
|
|
|
async def _fake_run_backend(request, *, runtime=None):
|
|
nonlocal backend_called
|
|
backend_called = True
|
|
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="", backend_notes=())
|
|
|
|
def _blocked_write_envelope(*args, **kwargs):
|
|
return {
|
|
"status": "blocked",
|
|
"reason": "sandbox_path",
|
|
"message": "shared path envelope blocked this command",
|
|
}
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", _fake_gate_action)
|
|
monkeypatch.setattr(shell, "preflight_subprocess_managed_network", _fake_preflight)
|
|
monkeypatch.setattr(shell, "_run_backend_with_managed_network", _fake_run_backend)
|
|
monkeypatch.setattr(shell, "_sandbox_write_path_access_envelope", _blocked_write_envelope)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command('where opensquilla 2>nul || echo "missing"')
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "shared path envelope blocked this command" in result
|
|
assert backend_called is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_host_exec_batch_requests_write_path_access_before_host(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir()
|
|
outside = tmp_path / "outside.txt"
|
|
command = f"where winget; Set-Content -Path {outside} -Value hi"
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, "prompt")
|
|
runtime = _windows_runtime()
|
|
host_calls: list[str] = []
|
|
|
|
async def fake_host(command, **kwargs):
|
|
host_calls.append(command)
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(workspace),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command(command, workdir=str(workspace))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
payload = json.loads(result)
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["access"] == "rw"
|
|
assert payload["path"] == str(outside.resolve(strict=False))
|
|
assert host_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_host_exec_batch_blocks_protected_metadata_write_before_host(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import reset_approval_queue
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
workspace = tmp_path / "workspace"
|
|
protected = workspace / ".git"
|
|
protected.mkdir(parents=True)
|
|
target = protected / "config"
|
|
command = f"where winget; Set-Content -Path {target} -Value hi"
|
|
|
|
_configure_approval_queue(monkeypatch, tmp_path, "prompt")
|
|
runtime = _windows_runtime()
|
|
host_calls: list[str] = []
|
|
|
|
async def fake_host(command, **kwargs):
|
|
host_calls.append(command)
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(workspace),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command(command, workdir=str(workspace))
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
reset_approval_queue()
|
|
|
|
payload = json.loads(result)
|
|
assert payload["status"] == "blocked"
|
|
assert payload["reason"] == "protected_metadata"
|
|
assert payload["resolved_path"] == str(target.resolve(strict=False))
|
|
assert host_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_windows_exec_command_merges_shell_active_mounts(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.sandbox.config import SandboxSettings
|
|
from opensquilla.sandbox.policy import build_policy
|
|
from opensquilla.sandbox.types import SecurityLevel
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime = _windows_runtime()
|
|
policy = build_policy(
|
|
SecurityLevel.STANDARD,
|
|
"shell.exec",
|
|
tmp_path,
|
|
SandboxSettings(
|
|
sandbox=True,
|
|
security_grading=True,
|
|
backend="windows_default",
|
|
network_default="none",
|
|
),
|
|
trusted=True,
|
|
)
|
|
request = SimpleNamespace(
|
|
cwd=tmp_path,
|
|
action_kind="shell.exec",
|
|
policy=policy,
|
|
reason="",
|
|
session_id="s1",
|
|
run_mode="trusted",
|
|
)
|
|
|
|
async def _fake_gate_action(**kwargs):
|
|
return object(), policy, request
|
|
|
|
async def _fake_preflight(*args, **kwargs):
|
|
return None
|
|
|
|
backend_requests = []
|
|
|
|
async def _fake_run_backend(request, *, runtime=None):
|
|
backend_requests.append(request)
|
|
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="", backend_notes=())
|
|
|
|
mounted = tmp_path / "external"
|
|
mounted.mkdir()
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", _fake_gate_action)
|
|
monkeypatch.setattr(shell, "preflight_subprocess_managed_network", _fake_preflight)
|
|
monkeypatch.setattr(shell, "_run_backend_with_managed_network", _fake_run_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"_active_sandbox_mounts",
|
|
lambda: [{"path": str(mounted), "access": "rw"}],
|
|
)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command("Write-Output ok")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "ok" in result
|
|
assert backend_requests
|
|
assert any(mount.host_path == mounted for mount in backend_requests[0].policy.mounts)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_windows_exec_command_blocks_runtime_readonly_write_target(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.sandbox.config import SandboxSettings
|
|
from opensquilla.sandbox.policy import build_policy
|
|
from opensquilla.sandbox.types import SecurityLevel
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime = _windows_runtime()
|
|
policy = build_policy(
|
|
SecurityLevel.STANDARD,
|
|
"shell.exec",
|
|
tmp_path,
|
|
SandboxSettings(
|
|
sandbox=True,
|
|
security_grading=True,
|
|
backend="windows_default",
|
|
network_default="none",
|
|
),
|
|
trusted=True,
|
|
)
|
|
request = SimpleNamespace(
|
|
cwd=tmp_path,
|
|
action_kind="shell.exec",
|
|
policy=policy,
|
|
reason="",
|
|
session_id="s1",
|
|
run_mode="trusted",
|
|
)
|
|
runtime_root = tmp_path / "runtime-src"
|
|
runtime_root.mkdir()
|
|
target = runtime_root / "__write_should_not_happen__.txt"
|
|
|
|
async def _fake_gate_action(**kwargs):
|
|
return object(), policy, request
|
|
|
|
async def _fake_preflight(*args, **kwargs):
|
|
return None
|
|
|
|
async def _fake_run_backend(request, *, runtime=None):
|
|
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="", backend_notes=())
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "gate_action", _fake_gate_action)
|
|
monkeypatch.setattr(shell, "preflight_subprocess_managed_network", _fake_preflight)
|
|
monkeypatch.setattr(shell, "_run_backend_with_managed_network", _fake_run_backend)
|
|
monkeypatch.setattr(shell, "_windows_runtime_readonly_roots", lambda: (runtime_root,))
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command(f'echo "test" > "{target}"')
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
payload = json.loads(result)
|
|
assert payload["reason"] == "runtime_readonly"
|
|
assert payload["resolved_path"] == str(target)
|
|
|
|
|
|
def test_shell_blocks_runtime_python_environment_bootstrap(monkeypatch, tmp_path) -> None:
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime_root = tmp_path / "runtime-venv"
|
|
runtime_root.mkdir()
|
|
|
|
monkeypatch.setattr(shell, "_runtime_readonly_roots", lambda runtime=None: (runtime_root,))
|
|
monkeypatch.setattr(shell, "full_host_access_active", lambda: False)
|
|
|
|
payload = shell._runtime_readonly_shell_block(
|
|
"exec_command",
|
|
"python -m ensurepip --upgrade",
|
|
str(tmp_path),
|
|
runtime=SimpleNamespace(backend=SimpleNamespace(name="bubblewrap")),
|
|
)
|
|
|
|
assert payload is not None
|
|
assert payload["reason"] == "runtime_readonly"
|
|
assert payload["runtime_operation"] == "python -m ensurepip"
|
|
assert payload["readonly_root"] == str(runtime_root)
|
|
|
|
|
|
def test_shell_blocks_runtime_python_package_install(monkeypatch, tmp_path) -> None:
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime_root = tmp_path / "runtime-venv"
|
|
runtime_python = runtime_root / "bin" / "python"
|
|
runtime_python.parent.mkdir(parents=True)
|
|
runtime_python.write_text("", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(shell, "_runtime_readonly_roots", lambda runtime=None: (runtime_root,))
|
|
monkeypatch.setattr(shell, "full_host_access_active", lambda: False)
|
|
|
|
payload = shell._runtime_readonly_shell_block(
|
|
"exec_command",
|
|
f"{runtime_python} -m pip install requests",
|
|
str(tmp_path),
|
|
runtime=SimpleNamespace(backend=SimpleNamespace(name="bubblewrap")),
|
|
)
|
|
|
|
assert payload is not None
|
|
assert payload["reason"] == "runtime_readonly"
|
|
assert payload["runtime_operation"] == "python -m pip install"
|
|
|
|
|
|
def test_shell_allows_explicit_project_venv_package_install(monkeypatch, tmp_path) -> None:
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime_root = tmp_path / "runtime-venv"
|
|
project_python = tmp_path / "project" / ".venv" / "bin" / "python"
|
|
project_python.parent.mkdir(parents=True)
|
|
project_python.write_text("", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(shell, "_runtime_readonly_roots", lambda runtime=None: (runtime_root,))
|
|
monkeypatch.setattr(shell, "full_host_access_active", lambda: False)
|
|
|
|
payload = shell._runtime_readonly_shell_block(
|
|
"exec_command",
|
|
f"{project_python} -m pip install requests",
|
|
str(tmp_path),
|
|
runtime=SimpleNamespace(backend=SimpleNamespace(name="bubblewrap")),
|
|
)
|
|
|
|
assert payload is None
|
|
|
|
|
|
def test_shell_allows_explicit_project_venv_ensurepip(monkeypatch, tmp_path) -> None:
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime_root = tmp_path / "runtime-venv"
|
|
project_python = tmp_path / "project" / ".venv" / "bin" / "python"
|
|
project_python.parent.mkdir(parents=True)
|
|
project_python.write_text("", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(shell, "_runtime_readonly_roots", lambda runtime=None: (runtime_root,))
|
|
monkeypatch.setattr(shell, "full_host_access_active", lambda: False)
|
|
|
|
payload = shell._runtime_readonly_shell_block(
|
|
"exec_command",
|
|
f"{project_python} -m ensurepip --upgrade",
|
|
str(tmp_path),
|
|
runtime=SimpleNamespace(backend=SimpleNamespace(name="bubblewrap")),
|
|
)
|
|
|
|
assert payload is None
|
|
|
|
|
|
def test_shell_blocks_windows_runtime_python_environment_bootstrap(monkeypatch) -> None:
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime_root = Path(r"X:\workspace\.venv")
|
|
|
|
monkeypatch.setattr(shell, "_runtime_readonly_roots", lambda runtime=None: (runtime_root,))
|
|
monkeypatch.setattr(shell, "full_host_access_active", lambda: False)
|
|
|
|
payload = shell._runtime_readonly_shell_block(
|
|
"exec_command",
|
|
r"X:\workspace\.venv\Scripts\python.exe -m ensurepip --upgrade",
|
|
r"X:\workspace",
|
|
runtime=SimpleNamespace(backend=SimpleNamespace(name="windows_default")),
|
|
)
|
|
|
|
assert payload is not None
|
|
assert payload["reason"] == "runtime_readonly"
|
|
assert payload["runtime_operation"] == "python -m ensurepip"
|
|
assert payload["readonly_root"] == str(runtime_root)
|
|
|
|
|
|
def test_shell_allows_windows_project_venv_ensurepip(monkeypatch) -> None:
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime_root = Path(r"X:\workspace\.venv")
|
|
|
|
monkeypatch.setattr(shell, "_runtime_readonly_roots", lambda runtime=None: (runtime_root,))
|
|
monkeypatch.setattr(shell, "full_host_access_active", lambda: False)
|
|
|
|
payload = shell._runtime_readonly_shell_block(
|
|
"exec_command",
|
|
r"X:\workspace\.tmp\proj\.venv\Scripts\python.exe -m ensurepip --upgrade",
|
|
r"X:\workspace",
|
|
runtime=SimpleNamespace(backend=SimpleNamespace(name="windows_default")),
|
|
)
|
|
|
|
assert payload is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_windows_exec_command_full_host_access_skips_runtime_readonly_block(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.tools.builtin import shell
|
|
|
|
runtime = _windows_runtime()
|
|
runtime_root = tmp_path / "runtime-src"
|
|
runtime_root.mkdir()
|
|
target = runtime_root / "__full_host_should_reach_host__.txt"
|
|
host_calls = []
|
|
|
|
async def _fake_run_host_shell_command(*args, **kwargs):
|
|
host_calls.append((args, kwargs))
|
|
return "exit_code=0\nhost-ok\n"
|
|
|
|
monkeypatch.setattr(shell, "get_runtime", lambda: runtime)
|
|
monkeypatch.setattr(shell, "_windows_runtime_readonly_roots", lambda: (runtime_root,))
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", _fake_run_host_shell_command)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="full",
|
|
)
|
|
)
|
|
try:
|
|
result = await shell.exec_command(f'echo "test" > "{target}"')
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "host-ok" in result
|
|
assert host_calls
|