mirror of
https://github.com/AstrBotDevs/AstrBot.git
synced 2026-10-16 15:06:04 +08:00
1587 lines
48 KiB
Python
1587 lines
48 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from astrbot.core.computer import process_sandbox
|
|
from astrbot.core.computer.booters import local as local_booter
|
|
from astrbot.core.computer.booters.local import (
|
|
LocalFileSystemComponent,
|
|
LocalPythonComponent,
|
|
LocalShellComponent,
|
|
)
|
|
from astrbot.core.computer.process_sandbox.unix import UnixProcessSandbox
|
|
|
|
|
|
class _FakePopen:
|
|
def __init__(self, stdout: bytes, stderr: bytes = b"", returncode: int = 0):
|
|
self._stdout = stdout
|
|
self._stderr = stderr
|
|
self.returncode = returncode
|
|
self.pid = 12345
|
|
|
|
def communicate(self, timeout=None):
|
|
return self._stdout, self._stderr
|
|
|
|
def wait(self, timeout=None):
|
|
pass
|
|
|
|
|
|
class _FakeTaskkillResult:
|
|
def __init__(self, returncode: int):
|
|
self.returncode = returncode
|
|
|
|
|
|
def _python_command(code: str) -> str:
|
|
"""Build a shell-safe Python command for the current operating system."""
|
|
if os.name == "nt":
|
|
# PowerShell re-parses the whole command text: invoke the executable
|
|
# through the call operator (&) so quoted paths with spaces survive.
|
|
return f"& '{sys.executable}' -u -c \"{code}\""
|
|
return shlex.join([sys.executable, "-u", "-c", code])
|
|
|
|
|
|
def _build_sandbox_command(
|
|
argv: list[str],
|
|
*,
|
|
workspace: Path,
|
|
env: dict[str, str] | None = None,
|
|
workspace_writable: bool = True,
|
|
allow_network: bool = False,
|
|
filesystem_scope: str = "workspace",
|
|
limits: process_sandbox.SandboxLimits | None = None,
|
|
) -> list[str]:
|
|
"""Build the current platform sandbox command for tests.
|
|
|
|
Args:
|
|
argv: Command and arguments to run inside the sandbox.
|
|
workspace: Directory exposed to the sandbox.
|
|
env: Extra environment variables exposed to the process.
|
|
workspace_writable: Whether the workspace may be modified.
|
|
allow_network: Whether the host network is exposed.
|
|
filesystem_scope: Filesystem scope exposed to the process.
|
|
limits: Optional resource ceilings applied to the command.
|
|
|
|
Returns:
|
|
Platform sandbox command and arguments.
|
|
"""
|
|
sandbox = process_sandbox.create_process_sandbox()
|
|
assert isinstance(sandbox, UnixProcessSandbox)
|
|
return sandbox.build_command(
|
|
argv,
|
|
process_sandbox.SandboxSpec(
|
|
workspace=workspace,
|
|
workspace_writable=workspace_writable,
|
|
allow_network=allow_network,
|
|
filesystem_scope=filesystem_scope,
|
|
limits=limits or process_sandbox.SandboxLimits(),
|
|
),
|
|
env=env,
|
|
)
|
|
|
|
|
|
def test_local_shell_component_decodes_utf8_output(monkeypatch):
|
|
def fake_run(*args, **kwargs):
|
|
_ = args, kwargs
|
|
return _FakePopen(stdout="技能内容".encode())
|
|
|
|
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
|
|
|
result = asyncio.run(LocalShellComponent().exec("dummy"))
|
|
|
|
assert result["stdout"] == "技能内容"
|
|
assert result["stderr"] == ""
|
|
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")
|
|
monkeypatch.setattr(local_booter.shutil, "which", lambda _cmd: None)
|
|
|
|
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_prefers_pwsh_when_available(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")
|
|
monkeypatch.setattr(
|
|
local_booter.shutil,
|
|
"which",
|
|
lambda cmd: "/opt/pwsh" if cmd == "pwsh" else None,
|
|
)
|
|
|
|
result = asyncio.run(LocalShellComponent().exec("Get-ChildItem"))
|
|
|
|
assert result["exit_code"] == 0
|
|
assert calls[0][0][0] == [
|
|
"pwsh.exe",
|
|
"-NoLogo",
|
|
"-NoProfile",
|
|
"-NonInteractive",
|
|
"-Command",
|
|
"Get-ChildItem",
|
|
]
|
|
assert calls[0][1]["shell"] is False
|
|
|
|
|
|
def test_exec_falls_back_to_powershell_when_pwsh_missing(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")
|
|
monkeypatch.setattr(local_booter.shutil, "which", lambda _cmd: None)
|
|
|
|
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.shutil, "which", lambda _cmd: None)
|
|
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",
|
|
creator_id="user-a",
|
|
creator_is_admin=False,
|
|
sandboxed=False,
|
|
permission_check=lambda: True,
|
|
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]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("platform_name", "implementation_name"),
|
|
[
|
|
("linux", "BubblewrapProcessSandbox"),
|
|
("darwin", "SeatbeltProcessSandbox"),
|
|
],
|
|
)
|
|
def test_process_sandbox_selects_system_implementation(
|
|
monkeypatch,
|
|
platform_name,
|
|
implementation_name,
|
|
):
|
|
monkeypatch.setattr(process_sandbox.sys, "platform", platform_name)
|
|
|
|
sandbox = process_sandbox.create_process_sandbox()
|
|
|
|
assert type(sandbox).__name__ == implementation_name
|
|
|
|
|
|
def test_process_sandbox_fails_closed_on_unsupported_system(monkeypatch):
|
|
monkeypatch.setattr(process_sandbox.sys, "platform", "win32")
|
|
|
|
with pytest.raises(RuntimeError, match="No Local process sandbox backend"):
|
|
process_sandbox.create_process_sandbox()
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="Unix sandbox requires POSIX.")
|
|
def test_linux_bwrap_command_is_workspace_only_and_clears_environment(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
monkeypatch.setattr(local_booter.sys, "platform", "linux")
|
|
monkeypatch.setattr(local_booter.shutil, "which", lambda name: f"/usr/bin/{name}")
|
|
|
|
command = _build_sandbox_command(
|
|
["/bin/sh", "-c", "pwd"],
|
|
workspace=tmp_path,
|
|
env={"CUSTOM_VALUE": "visible"},
|
|
)
|
|
|
|
workspace = str(tmp_path.resolve())
|
|
assert command[0] == "/usr/bin/bwrap"
|
|
assert "--unshare-all" in command
|
|
assert "--new-session" in command
|
|
assert "--clearenv" in command
|
|
assert "--share-net" not in command
|
|
assert command[command.index("--size") + 1] == str(256 * 1024 * 1024)
|
|
assert ["--bind", workspace, workspace] == command[
|
|
command.index("--bind") : command.index("--bind") + 3
|
|
]
|
|
assert ["--setenv", "CUSTOM_VALUE", "visible"] == command[
|
|
command.index("CUSTOM_VALUE") - 1 : command.index("CUSTOM_VALUE") + 2
|
|
]
|
|
assert command[-3:] == ["/bin/sh", "-c", "pwd"]
|
|
bind_sources = [
|
|
command[index + 1]
|
|
for index, value in enumerate(command)
|
|
if value in {"--bind", "--ro-bind"}
|
|
]
|
|
assert str(Path.home()) not in bind_sources
|
|
|
|
read_only_command = _build_sandbox_command(
|
|
["/usr/bin/rg", "needle", "."],
|
|
workspace=tmp_path,
|
|
workspace_writable=False,
|
|
)
|
|
assert ["--ro-bind", workspace, workspace] == read_only_command[
|
|
read_only_command.index(workspace) - 1 : read_only_command.index(workspace) + 2
|
|
]
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="Unix sandbox requires POSIX.")
|
|
def test_linux_bwrap_command_applies_network_and_filesystem_permissions(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
monkeypatch.setattr(local_booter.sys, "platform", "linux")
|
|
monkeypatch.setattr(local_booter.shutil, "which", lambda name: f"/usr/bin/{name}")
|
|
|
|
networked_workspace = _build_sandbox_command(
|
|
["/bin/sh", "-c", "pwd"],
|
|
workspace=tmp_path,
|
|
allow_network=True,
|
|
filesystem_scope="workspace",
|
|
)
|
|
host_without_network = _build_sandbox_command(
|
|
["/bin/sh", "-c", "pwd"],
|
|
workspace=tmp_path,
|
|
allow_network=False,
|
|
filesystem_scope="host",
|
|
)
|
|
|
|
assert "--share-net" in networked_workspace
|
|
assert ["--bind", "/", "/"] not in [
|
|
networked_workspace[index : index + 3]
|
|
for index in range(len(networked_workspace) - 2)
|
|
]
|
|
assert "--share-net" not in host_without_network
|
|
root_bind = host_without_network.index("--bind")
|
|
assert host_without_network[root_bind : root_bind + 3] == ["--bind", "/", "/"]
|
|
assert "--tmpfs" not in host_without_network
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="Unix sandbox requires POSIX.")
|
|
def test_process_sandbox_exposes_resource_limits(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(local_booter.sys, "platform", "linux")
|
|
monkeypatch.setattr(local_booter.shutil, "which", lambda name: f"/usr/bin/{name}")
|
|
limits = process_sandbox.SandboxLimits(
|
|
cpu_seconds=17,
|
|
file_size_bytes=18,
|
|
memory_bytes=19,
|
|
open_files=20,
|
|
processes=21,
|
|
)
|
|
|
|
command = _build_sandbox_command(
|
|
["/bin/sh", "-c", "pwd"],
|
|
workspace=tmp_path,
|
|
limits=limits,
|
|
)
|
|
wrapper_code = command[command.index("-I") + 3]
|
|
|
|
assert "(resource.RLIMIT_CPU, 17)" in wrapper_code
|
|
assert "(resource.RLIMIT_FSIZE, 18)" in wrapper_code
|
|
assert "(resource.RLIMIT_AS, 19)" in wrapper_code
|
|
assert "(resource.RLIMIT_NOFILE, 20)" in wrapper_code
|
|
assert "(resource.RLIMIT_NPROC, 21)" in wrapper_code
|
|
|
|
|
|
def test_sandbox_limits_reject_nonpositive_values():
|
|
with pytest.raises(ValueError, match="cpu_seconds"):
|
|
process_sandbox.SandboxLimits(cpu_seconds=0)
|
|
|
|
|
|
def test_linux_bwrap_command_fails_closed_when_bwrap_is_missing(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
monkeypatch.setattr(local_booter.sys, "platform", "linux")
|
|
monkeypatch.setattr(local_booter.shutil, "which", lambda _name: None)
|
|
|
|
with pytest.raises(RuntimeError, match="bubblewrap"):
|
|
_build_sandbox_command(
|
|
["/bin/sh", "-c", "pwd"],
|
|
workspace=tmp_path,
|
|
)
|
|
|
|
|
|
def test_macos_seatbelt_command_restricts_profile_and_environment(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
monkeypatch.setattr(local_booter.sys, "platform", "darwin")
|
|
monkeypatch.setattr(
|
|
local_booter.shutil,
|
|
"which",
|
|
lambda name, **_kwargs: f"/usr/bin/{name}",
|
|
)
|
|
|
|
command = _build_sandbox_command(
|
|
["/bin/sh", "-c", "pwd"],
|
|
workspace=tmp_path,
|
|
env={"CUSTOM_VALUE": "visible", "PATH": "untrusted"},
|
|
)
|
|
|
|
profile = command[command.index("-p") + 1]
|
|
environment_start = command.index("-i") + 1
|
|
launcher_start = command.index(
|
|
str(Path(sys.executable).resolve()),
|
|
environment_start,
|
|
)
|
|
environment = command[environment_start:launcher_start]
|
|
workspace = str(tmp_path.resolve())
|
|
|
|
assert command[0] == "/usr/bin/sandbox-exec"
|
|
assert f"WORKSPACE={workspace}" in command
|
|
assert '(import "system.sb")' in profile
|
|
assert "(deny default)" in profile
|
|
assert "(deny network*)" in profile
|
|
assert '(literal "/usr/bin/open")' in profile
|
|
assert "(deny appleevent-send)" in profile
|
|
assert '(global-name "com.apple.coreservices.launchservicesd")' in profile
|
|
assert "(allow file-write*" in profile
|
|
assert command[command.index("-p") + 2 : command.index("-p") + 4] == [
|
|
"/usr/bin/env",
|
|
"-i",
|
|
]
|
|
assert "CUSTOM_VALUE=visible" in environment
|
|
assert environment[-4:] == [
|
|
f"PATH={Path(sys.executable).parent}:/usr/bin:/bin",
|
|
f"HOME={workspace}",
|
|
f"TMPDIR={workspace}",
|
|
"LANG=C.UTF-8",
|
|
]
|
|
assert command[-3:] == ["/bin/sh", "-c", "pwd"]
|
|
|
|
read_only_command = _build_sandbox_command(
|
|
["/usr/bin/rg", "needle", "."],
|
|
workspace=tmp_path,
|
|
workspace_writable=False,
|
|
)
|
|
read_only_profile = read_only_command[read_only_command.index("-p") + 1]
|
|
assert "(deny file-write*)" in read_only_profile
|
|
assert "(allow file-write*" not in read_only_profile
|
|
|
|
|
|
def test_macos_seatbelt_profile_applies_network_and_filesystem_permissions(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
monkeypatch.setattr(local_booter.sys, "platform", "darwin")
|
|
monkeypatch.setattr(
|
|
local_booter.shutil,
|
|
"which",
|
|
lambda name, **_kwargs: f"/usr/bin/{name}",
|
|
)
|
|
|
|
networked_workspace = _build_sandbox_command(
|
|
["/bin/sh", "-c", "pwd"],
|
|
workspace=tmp_path,
|
|
allow_network=True,
|
|
filesystem_scope="workspace",
|
|
)
|
|
host_without_network = _build_sandbox_command(
|
|
["/bin/sh", "-c", "pwd"],
|
|
workspace=tmp_path,
|
|
allow_network=False,
|
|
filesystem_scope="host",
|
|
)
|
|
|
|
network_profile = networked_workspace[networked_workspace.index("-p") + 1]
|
|
host_profile = host_without_network[host_without_network.index("-p") + 1]
|
|
assert "(allow network*)" in network_profile
|
|
assert "(deny network*)" not in network_profile
|
|
assert "(deny network*)" in host_profile
|
|
assert "(allow file-read* file-write*" in host_profile
|
|
assert '(literal "/private/etc/passwd")' not in host_profile
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform != "darwin", reason="Seatbelt is macOS-only")
|
|
def test_macos_seatbelt_enforces_independent_network_and_filesystem_permissions(
|
|
tmp_path,
|
|
):
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir()
|
|
outside_file = tmp_path / "outside.txt"
|
|
outside_file.write_text("host-visible", encoding="utf-8")
|
|
|
|
host_without_network = _build_sandbox_command(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
(
|
|
"import pathlib, socket; "
|
|
f"assert pathlib.Path({str(outside_file)!r}).read_text() == "
|
|
"'host-visible'; "
|
|
"sock = socket.socket(); "
|
|
"\ntry: sock.bind(('127.0.0.1', 0))\n"
|
|
"except OSError: pass\n"
|
|
"else: raise SystemExit('network unexpectedly available')"
|
|
),
|
|
],
|
|
workspace=workspace,
|
|
allow_network=False,
|
|
filesystem_scope="host",
|
|
)
|
|
host_result = subprocess.run(
|
|
host_without_network,
|
|
cwd=workspace,
|
|
env={"PATH": os.defpath},
|
|
capture_output=True,
|
|
timeout=10,
|
|
)
|
|
|
|
workspace_with_network = _build_sandbox_command(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
(
|
|
"import pathlib, socket; "
|
|
f"outside = pathlib.Path({str(outside_file)!r}); "
|
|
"\ntry: outside.read_text()\n"
|
|
"except OSError: pass\n"
|
|
"else: raise SystemExit('host file unexpectedly available')\n"
|
|
"sock = socket.socket(); sock.bind(('127.0.0.1', 0))"
|
|
),
|
|
],
|
|
workspace=workspace,
|
|
allow_network=True,
|
|
filesystem_scope="workspace",
|
|
)
|
|
network_result = subprocess.run(
|
|
workspace_with_network,
|
|
cwd=workspace,
|
|
env={"PATH": os.defpath},
|
|
capture_output=True,
|
|
timeout=10,
|
|
)
|
|
|
|
assert host_result.returncode == 0, local_booter._decode_shell_output(
|
|
host_result.stderr
|
|
)
|
|
assert network_result.returncode == 0, local_booter._decode_shell_output(
|
|
network_result.stderr
|
|
)
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform != "darwin", reason="Seatbelt is macOS-only")
|
|
def test_macos_seatbelt_preserves_process_limit_and_allows_shell_fork(tmp_path):
|
|
import resource
|
|
|
|
python_path = str(Path(sys.executable).resolve())
|
|
print_process_limit = shlex.join(
|
|
[
|
|
python_path,
|
|
"-I",
|
|
"-S",
|
|
"-c",
|
|
"import resource; print(resource.getrlimit(resource.RLIMIT_NPROC)[0])",
|
|
]
|
|
)
|
|
command = _build_sandbox_command(
|
|
["/bin/sh", "-c", f"{print_process_limit}; /bin/date >/dev/null"],
|
|
workspace=tmp_path,
|
|
)
|
|
|
|
result = subprocess.run(
|
|
command,
|
|
cwd=tmp_path,
|
|
env={"PATH": os.defpath},
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
assert int(result.stdout.strip()) == resource.getrlimit(resource.RLIMIT_NPROC)[0]
|
|
|
|
|
|
def test_macos_seatbelt_command_fails_closed_when_tool_is_missing(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
monkeypatch.setattr(local_booter.sys, "platform", "darwin")
|
|
monkeypatch.setattr(
|
|
local_booter.shutil,
|
|
"which",
|
|
lambda _name, **_kwargs: None,
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="Seatbelt"):
|
|
_build_sandbox_command(
|
|
["/bin/sh", "-c", "pwd"],
|
|
workspace=tmp_path,
|
|
)
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform != "darwin", reason="Seatbelt is macOS-only")
|
|
def test_macos_seatbelt_enforces_workspace_network_and_environment(tmp_path):
|
|
outside_file = tmp_path.parent / "seatbelt-outside.txt"
|
|
outside_file.write_text("host secret", encoding="utf-8")
|
|
(tmp_path / "input.txt").write_text("allowed", encoding="utf-8")
|
|
code = f"""
|
|
import os
|
|
import pathlib
|
|
import socket
|
|
import subprocess
|
|
|
|
workspace = pathlib.Path.cwd()
|
|
assert (workspace / "input.txt").read_text() == "allowed"
|
|
(workspace / "output.txt").write_text("written")
|
|
try:
|
|
pathlib.Path({str(outside_file)!r}).read_text()
|
|
except OSError:
|
|
pass
|
|
else:
|
|
raise SystemExit("host read unexpectedly allowed")
|
|
try:
|
|
os.kill(int(os.environ["HOST_PID"]), 0)
|
|
except OSError:
|
|
pass
|
|
else:
|
|
raise SystemExit("host process unexpectedly visible")
|
|
try:
|
|
opened = subprocess.run(
|
|
["/usr/bin/open", "https://example.com"],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
).returncode == 0
|
|
except OSError:
|
|
opened = False
|
|
if opened:
|
|
raise SystemExit("URL broker unexpectedly available")
|
|
try:
|
|
sock = socket.socket()
|
|
sock.settimeout(1)
|
|
connected = sock.connect_ex(("1.1.1.1", 53)) == 0
|
|
except OSError:
|
|
connected = False
|
|
if connected:
|
|
raise SystemExit("network unexpectedly available")
|
|
assert os.environ.get("HOST_SECRET") is None
|
|
"""
|
|
command = _build_sandbox_command(
|
|
[sys.executable, "-c", code],
|
|
workspace=tmp_path,
|
|
env={"HOST_PID": str(os.getpid())},
|
|
)
|
|
|
|
result = subprocess.run(
|
|
command,
|
|
cwd=tmp_path,
|
|
env={"PATH": os.defpath, "HOST_SECRET": "must-not-leak"},
|
|
capture_output=True,
|
|
timeout=10,
|
|
)
|
|
|
|
assert result.returncode == 0, local_booter._decode_shell_output(result.stderr)
|
|
assert (tmp_path / "output.txt").read_text(encoding="utf-8") == "written"
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform != "darwin", reason="Seatbelt is macOS-only")
|
|
def test_macos_read_only_seatbelt_grep_cannot_follow_outside_ancestor_symlink(
|
|
tmp_path,
|
|
):
|
|
rg_path = shutil.which("rg")
|
|
if not rg_path:
|
|
pytest.skip("ripgrep is unavailable")
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir()
|
|
inside_file = workspace / "inside.txt"
|
|
inside_file.write_text("visible-needle\n", encoding="utf-8")
|
|
outside_dir = tmp_path / "outside"
|
|
outside_dir.mkdir()
|
|
outside_file = outside_dir / "secret.txt"
|
|
outside_file.write_text("host-secret-needle\n", encoding="utf-8")
|
|
(workspace / "redirect").symlink_to(outside_dir, target_is_directory=True)
|
|
|
|
positive_command = _build_sandbox_command(
|
|
[str(Path(rg_path).resolve()), "visible-needle", "--", str(inside_file)],
|
|
workspace=workspace,
|
|
workspace_writable=False,
|
|
)
|
|
positive = subprocess.run(
|
|
positive_command,
|
|
cwd=workspace,
|
|
env={"PATH": os.defpath},
|
|
capture_output=True,
|
|
timeout=10,
|
|
)
|
|
assert positive.returncode == 0
|
|
assert b"visible-needle" in positive.stdout
|
|
|
|
escaped_path = workspace / "redirect" / "secret.txt"
|
|
escaped_command = _build_sandbox_command(
|
|
[str(Path(rg_path).resolve()), "host-secret-needle", "--", str(escaped_path)],
|
|
workspace=workspace,
|
|
workspace_writable=False,
|
|
)
|
|
escaped = subprocess.run(
|
|
escaped_command,
|
|
cwd=workspace,
|
|
env={"PATH": os.defpath},
|
|
capture_output=True,
|
|
timeout=10,
|
|
)
|
|
assert b"host-secret-needle" not in escaped.stdout
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
not sys.platform.startswith("linux"),
|
|
reason="bubblewrap is Linux-only",
|
|
)
|
|
def test_linux_read_only_bwrap_grep_cannot_follow_outside_ancestor_symlink(
|
|
tmp_path,
|
|
):
|
|
if not shutil.which("bwrap"):
|
|
pytest.skip("bubblewrap is unavailable")
|
|
if sys.version_info >= (3, 14) and not shutil.which("rg"):
|
|
pytest.skip("ripgrep is unavailable")
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir()
|
|
inside_file = workspace / "inside.txt"
|
|
inside_file.write_text("visible-needle\n", encoding="utf-8")
|
|
outside_dir = tmp_path / "outside"
|
|
outside_dir.mkdir()
|
|
outside_file = outside_dir / "secret.txt"
|
|
outside_file.write_text("host-secret-needle\n", encoding="utf-8")
|
|
(workspace / "redirect").symlink_to(outside_dir, target_is_directory=True)
|
|
|
|
positive = asyncio.run(
|
|
LocalFileSystemComponent().search_files(
|
|
"visible-needle",
|
|
path=str(inside_file),
|
|
sandboxed=True,
|
|
sandbox_root=str(workspace),
|
|
)
|
|
)
|
|
assert positive["success"] is True
|
|
assert "visible-needle" in positive["content"]
|
|
|
|
escaped_path = workspace / "redirect" / "secret.txt"
|
|
escaped = asyncio.run(
|
|
LocalFileSystemComponent().search_files(
|
|
"host-secret-needle",
|
|
path=str(escaped_path),
|
|
sandboxed=True,
|
|
sandbox_root=str(workspace),
|
|
)
|
|
)
|
|
assert "host-secret-needle" not in escaped.get("content", "")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("platform_name", "sandbox_executable"),
|
|
[
|
|
("linux", "/usr/bin/bwrap"),
|
|
("darwin", "/usr/bin/sandbox-exec"),
|
|
],
|
|
)
|
|
@pytest.mark.skipif(os.name == "nt", reason="Unix sandbox requires POSIX.")
|
|
async def test_managed_shell_uses_platform_sandbox(
|
|
monkeypatch,
|
|
tmp_path,
|
|
platform_name,
|
|
sandbox_executable,
|
|
):
|
|
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("Sandboxed commands must use an argument vector.")
|
|
|
|
monkeypatch.setattr(local_booter.sys, "platform", platform_name)
|
|
monkeypatch.setattr(
|
|
local_booter.shutil,
|
|
"which",
|
|
lambda _name, **_kwargs: sandbox_executable,
|
|
)
|
|
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(
|
|
"pwd",
|
|
owner_id="owner-a",
|
|
creator_id="member-a",
|
|
creator_is_admin=False,
|
|
cwd=str(tmp_path),
|
|
yield_time_ms=5_000,
|
|
sandboxed=True,
|
|
permission_check=lambda: True,
|
|
)
|
|
|
|
assert result["status"] == "completed"
|
|
assert result["stdout"] == "done\n"
|
|
assert calls[0][0][0] == sandbox_executable
|
|
assert calls[0][0][-3:] == ("/bin/sh", "-c", "pwd")
|
|
assert calls[0][1]["env"] == {"PATH": os.defpath}
|
|
assert calls[0][1]["start_new_session"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sandboxed_managed_shell_delegates_launch(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
|
|
|
|
class FakeSandbox:
|
|
async def spawn_shell(self, command, spec, **kwargs):
|
|
calls.append((command, spec, kwargs))
|
|
return FakeProcess()
|
|
|
|
monkeypatch.setattr(local_booter.sys, "platform", "win32")
|
|
monkeypatch.setattr(
|
|
local_booter,
|
|
"create_process_sandbox",
|
|
lambda: FakeSandbox(),
|
|
)
|
|
|
|
result = await LocalShellComponent().exec_managed(
|
|
"Get-ChildItem",
|
|
owner_id="owner-a",
|
|
creator_id="member-a",
|
|
creator_is_admin=False,
|
|
cwd=str(tmp_path),
|
|
yield_time_ms=5_000,
|
|
sandboxed=True,
|
|
permission_check=lambda: True,
|
|
)
|
|
|
|
assert result["status"] == "completed"
|
|
assert calls[0][0] == "Get-ChildItem"
|
|
assert calls[0][1].workspace == tmp_path.resolve()
|
|
assert calls[0][2] == {"env": {}}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(os.name == "nt", reason="Unix sandbox requires POSIX.")
|
|
async def test_sandboxed_managed_shell_stops_at_output_limit(monkeypatch, tmp_path):
|
|
process_stopped = asyncio.Event()
|
|
|
|
class FakeStdout:
|
|
def __init__(self):
|
|
self.chunks = [b"123456", 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):
|
|
await process_stopped.wait()
|
|
self.returncode = -15
|
|
return self.returncode
|
|
|
|
async def fake_create_subprocess_exec(*_args, **_kwargs):
|
|
return FakeProcess()
|
|
|
|
def fake_killpg(pid, sig):
|
|
assert pid == 12345
|
|
assert sig == local_booter.signal.SIGTERM
|
|
process_stopped.set()
|
|
|
|
monkeypatch.setattr(local_booter.sys, "platform", "linux")
|
|
monkeypatch.setattr(local_booter.shutil, "which", lambda _name: "/usr/bin/bwrap")
|
|
monkeypatch.setattr(local_booter, "_LOCAL_SANDBOX_MAX_OUTPUT_BYTES", 5)
|
|
monkeypatch.setattr(
|
|
local_booter.asyncio,
|
|
"create_subprocess_exec",
|
|
fake_create_subprocess_exec,
|
|
)
|
|
monkeypatch.setattr(local_booter.os, "killpg", fake_killpg)
|
|
|
|
result = await LocalShellComponent().exec_managed(
|
|
"yes",
|
|
owner_id="owner-a",
|
|
creator_id="member-a",
|
|
creator_is_admin=False,
|
|
cwd=str(tmp_path),
|
|
yield_time_ms=5_000,
|
|
sandboxed=True,
|
|
permission_check=lambda: True,
|
|
)
|
|
|
|
assert result["status"] == "output_limited"
|
|
assert result["stdout"] == "12345"
|
|
assert result["session_closed"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("platform_name", "sandbox_executable"),
|
|
[
|
|
("linux", "/usr/bin/bwrap"),
|
|
("darwin", "/usr/bin/sandbox-exec"),
|
|
],
|
|
)
|
|
@pytest.mark.skipif(os.name == "nt", reason="Unix sandbox requires POSIX.")
|
|
async def test_local_python_uses_platform_sandbox(
|
|
monkeypatch,
|
|
tmp_path,
|
|
platform_name,
|
|
sandbox_executable,
|
|
):
|
|
calls = []
|
|
|
|
def fake_run(*args, **kwargs):
|
|
calls.append((args, kwargs))
|
|
kwargs["stdout"].write(b"done\n")
|
|
return subprocess.CompletedProcess(args[0], 0)
|
|
|
|
monkeypatch.setattr(local_booter.sys, "platform", platform_name)
|
|
monkeypatch.setattr(
|
|
local_booter.shutil,
|
|
"which",
|
|
lambda _name, **_kwargs: sandbox_executable,
|
|
)
|
|
monkeypatch.setattr(local_booter.subprocess, "run", fake_run)
|
|
|
|
result = await LocalPythonComponent().exec(
|
|
"print('done')",
|
|
cwd=str(tmp_path),
|
|
sandboxed=True,
|
|
)
|
|
|
|
assert result["data"]["output"]["text"] == "done\n"
|
|
assert calls[0][0][0][0] == sandbox_executable
|
|
assert calls[0][0][0][-3:] == [
|
|
sys.executable,
|
|
"-c",
|
|
"print('done')",
|
|
]
|
|
assert calls[0][1]["env"] == {"PATH": os.defpath}
|
|
assert "capture_output" not in calls[0][1]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(os.name == "nt", reason="Unix sandbox requires POSIX.")
|
|
async def test_sandboxed_local_python_caps_returned_output(monkeypatch, tmp_path):
|
|
def fake_run(*args, **kwargs):
|
|
kwargs["stdout"].write(b"123456")
|
|
return subprocess.CompletedProcess(args[0], 0)
|
|
|
|
monkeypatch.setattr(local_booter.sys, "platform", "linux")
|
|
monkeypatch.setattr(local_booter.shutil, "which", lambda _name: "/usr/bin/bwrap")
|
|
monkeypatch.setattr(local_booter, "_LOCAL_SANDBOX_MAX_OUTPUT_BYTES", 5)
|
|
monkeypatch.setattr(local_booter.subprocess, "run", fake_run)
|
|
|
|
result = await LocalPythonComponent().exec(
|
|
"print('large output')",
|
|
cwd=str(tmp_path),
|
|
sandboxed=True,
|
|
)
|
|
|
|
assert result["data"]["output"]["text"] == "12345"
|
|
assert result["data"]["error"] == "Execution output exceeded 5 bytes."
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sandboxed_local_python_handles_backend_timeout(monkeypatch, tmp_path):
|
|
class FakeSandbox:
|
|
def run(self, *_args, **_kwargs):
|
|
raise process_sandbox.SandboxTimeoutError
|
|
|
|
monkeypatch.setattr(
|
|
local_booter,
|
|
"create_process_sandbox",
|
|
lambda: FakeSandbox(),
|
|
)
|
|
|
|
result = await LocalPythonComponent().exec(
|
|
"while True: pass",
|
|
cwd=str(tmp_path),
|
|
sandboxed=True,
|
|
)
|
|
|
|
assert result["data"]["error"] == "Execution timed out."
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_managed_shell_prefers_pwsh_when_available(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.shutil,
|
|
"which",
|
|
lambda cmd: "/opt/pwsh" if cmd == "pwsh" else None,
|
|
)
|
|
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",
|
|
creator_id="user-a",
|
|
creator_is_admin=False,
|
|
sandboxed=False,
|
|
permission_check=lambda: True,
|
|
cwd=str(tmp_path),
|
|
yield_time_ms=5_000,
|
|
)
|
|
|
|
assert result["status"] == "completed"
|
|
assert result["stdout"] == "done\n"
|
|
assert calls[0][0] == (
|
|
"pwsh.exe",
|
|
"-NoLogo",
|
|
"-NoProfile",
|
|
"-NonInteractive",
|
|
"-Command",
|
|
"Get-ChildItem",
|
|
)
|
|
assert "creationflags" in calls[0][1]
|
|
|
|
|
|
def test_local_shell_component_prefers_utf8_before_windows_locale(
|
|
monkeypatch,
|
|
):
|
|
def fake_run(*args, **kwargs):
|
|
_ = args, kwargs
|
|
return _FakePopen(stdout="技能内容".encode())
|
|
|
|
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
|
monkeypatch.setattr(local_booter.os, "name", "nt", raising=False)
|
|
monkeypatch.setattr(
|
|
local_booter.locale,
|
|
"getpreferredencoding",
|
|
lambda _do_setlocale=False: "cp936",
|
|
)
|
|
|
|
result = asyncio.run(LocalShellComponent().exec("dummy"))
|
|
|
|
assert result["stdout"] == "技能内容"
|
|
assert result["stderr"] == ""
|
|
assert result["exit_code"] == 0
|
|
|
|
|
|
def test_local_shell_component_uses_windows_locale_for_gbk(monkeypatch):
|
|
def fake_run(*args, **kwargs):
|
|
_ = args, kwargs
|
|
return _FakePopen(stdout="微博热搜".encode("gbk"))
|
|
|
|
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
|
monkeypatch.setattr(local_booter.os, "name", "nt", raising=False)
|
|
monkeypatch.setattr(
|
|
local_booter.locale,
|
|
"getpreferredencoding",
|
|
lambda _do_setlocale=False: "cp936",
|
|
)
|
|
|
|
result = asyncio.run(LocalShellComponent().exec("dummy"))
|
|
|
|
assert result["stdout"] == "微博热搜"
|
|
assert result["stderr"] == ""
|
|
assert result["exit_code"] == 0
|
|
|
|
|
|
def test_local_shell_component_preserves_western_windows_output(monkeypatch):
|
|
def fake_run(*args, **kwargs):
|
|
_ = args, kwargs
|
|
return _FakePopen(stdout="caféA".encode("cp1252"))
|
|
|
|
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
|
monkeypatch.setattr(local_booter.os, "name", "nt", raising=False)
|
|
monkeypatch.setattr(
|
|
local_booter.locale,
|
|
"getpreferredencoding",
|
|
lambda _do_setlocale=False: "cp1252",
|
|
)
|
|
|
|
result = asyncio.run(LocalShellComponent().exec("dummy"))
|
|
|
|
assert result["stdout"] == "caféA"
|
|
assert result["stderr"] == ""
|
|
assert result["exit_code"] == 0
|
|
|
|
|
|
def test_local_shell_component_falls_back_to_utf8_replace(monkeypatch):
|
|
def fake_run(*args, **kwargs):
|
|
_ = args, kwargs
|
|
return _FakePopen(stdout=b"\xffabc")
|
|
|
|
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
|
monkeypatch.setattr(local_booter.os, "name", "posix", raising=False)
|
|
monkeypatch.setattr(
|
|
local_booter.locale,
|
|
"getpreferredencoding",
|
|
lambda _do_setlocale=False: "utf-8",
|
|
)
|
|
|
|
result = asyncio.run(LocalShellComponent().exec("dummy"))
|
|
|
|
assert result["stdout"] == "\ufffdabc"
|
|
|
|
|
|
def test_local_shell_component_falls_back_when_windows_taskkill_fails(monkeypatch):
|
|
class TimeoutPopen:
|
|
pid = 12345
|
|
|
|
def __init__(self):
|
|
self.killed = False
|
|
self.wait_timeout = None
|
|
|
|
def communicate(self, timeout=None):
|
|
raise subprocess.TimeoutExpired(cmd="dummy", timeout=timeout)
|
|
|
|
def kill(self):
|
|
self.killed = True
|
|
|
|
def wait(self, timeout=None):
|
|
self.wait_timeout = timeout
|
|
|
|
proc = TimeoutPopen()
|
|
|
|
monkeypatch.setattr(subprocess, "Popen", lambda *_args, **_kwargs: proc)
|
|
monkeypatch.setattr(
|
|
subprocess,
|
|
"run",
|
|
lambda *_args, **_kwargs: _FakeTaskkillResult(returncode=1),
|
|
)
|
|
monkeypatch.setattr(local_booter.sys, "platform", "win32")
|
|
monkeypatch.setattr(local_booter.shutil, "which", lambda _cmd: None)
|
|
|
|
with pytest.raises(subprocess.TimeoutExpired):
|
|
asyncio.run(LocalShellComponent().exec("dummy", timeout=1))
|
|
|
|
assert proc.killed
|
|
assert proc.wait_timeout == 5
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(os.name == "nt", reason="Unix process-group cleanup.")
|
|
async def test_managed_shell_cleans_children_after_leader_exits():
|
|
shell = LocalShellComponent()
|
|
result = await shell.exec_managed(
|
|
"(trap '' TERM; printf 'ready\\n'; sleep 30) & read line",
|
|
owner_id="owner",
|
|
creator_id="user",
|
|
creator_is_admin=False,
|
|
sandboxed=False,
|
|
permission_check=lambda: True,
|
|
yield_time_ms=0,
|
|
)
|
|
session = shell._sessions[result["session_id"]]
|
|
try:
|
|
await asyncio.wait_for(session.output_event.wait(), 5)
|
|
session.process.stdin.write(b"\n")
|
|
await session.process.stdin.drain()
|
|
for _ in range(500):
|
|
if session.process.returncode is not None:
|
|
break
|
|
await asyncio.sleep(0.01)
|
|
assert session.process.returncode is not None
|
|
# The child keeps stdout open and ignores SIGTERM after its parent exits.
|
|
assert not session.reader_task.done()
|
|
session.permission_check = None
|
|
await asyncio.wait_for(shell.shutdown_sessions(invalid_only=True), 8)
|
|
assert session.reader_task.done()
|
|
assert session.output_file.closed
|
|
assert not shell._sessions
|
|
finally:
|
|
await shell.shutdown_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_managed_shell_returns_completed_output_without_open_session():
|
|
shell = LocalShellComponent()
|
|
|
|
result = await shell.exec_managed(
|
|
_python_command("print('hello')"),
|
|
owner_id="owner-a",
|
|
creator_id="user-a",
|
|
creator_is_admin=False,
|
|
sandboxed=False,
|
|
permission_check=lambda: True,
|
|
yield_time_ms=5_000,
|
|
)
|
|
|
|
assert result["status"] == "completed"
|
|
assert result["stdout"].splitlines() == ["hello"]
|
|
assert result["exit_code"] == 0
|
|
assert result["session_closed"] is True
|
|
assert await shell.list_sessions(
|
|
owner_id="owner-a",
|
|
requester_id="user-a",
|
|
requester_is_admin=False,
|
|
) == {"sessions": []}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_managed_shell_allows_creator_and_conversation_admin():
|
|
shell = LocalShellComponent()
|
|
result = await shell.exec_managed(
|
|
_python_command("import time; print('ready', flush=True); time.sleep(30)"),
|
|
owner_id="owner-a",
|
|
creator_id="user-a",
|
|
creator_is_admin=False,
|
|
sandboxed=False,
|
|
permission_check=lambda: True,
|
|
# Cold-starting PowerShell + Python on CI runners exceeds 200ms.
|
|
yield_time_ms=5_000,
|
|
)
|
|
|
|
try:
|
|
assert result["status"] == "running"
|
|
assert result["stdout"].splitlines() == ["ready"]
|
|
session_id = result["session_id"]
|
|
assert (
|
|
await shell.list_sessions(
|
|
owner_id="owner-b",
|
|
requester_id="user-a",
|
|
requester_is_admin=False,
|
|
)
|
|
)["sessions"] == []
|
|
sessions = (
|
|
await shell.list_sessions(
|
|
owner_id="owner-a",
|
|
requester_id="user-a",
|
|
requester_is_admin=False,
|
|
)
|
|
)["sessions"]
|
|
assert [item["session_id"] for item in sessions] == [session_id]
|
|
assert sessions[0]["sandboxed"] is False
|
|
|
|
admin_sessions = (
|
|
await shell.list_sessions(
|
|
owner_id="owner-a",
|
|
requester_id="admin-user",
|
|
requester_is_admin=True,
|
|
)
|
|
)["sessions"]
|
|
assert [item["session_id"] for item in admin_sessions] == [session_id]
|
|
stopped = await shell.terminate_session(
|
|
owner_id="owner-a",
|
|
requester_id="admin-user",
|
|
requester_is_admin=True,
|
|
session_id=session_id,
|
|
)
|
|
|
|
assert stopped["status"] == "terminated"
|
|
assert stopped["exit_code"] is not None
|
|
assert stopped["session_closed"] is True
|
|
assert await shell.list_sessions(
|
|
owner_id="owner-a",
|
|
requester_id="user-a",
|
|
requester_is_admin=False,
|
|
) == {"sessions": []}
|
|
finally:
|
|
await shell.shutdown_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_managed_shell_rejects_cross_user_session_access():
|
|
shell = LocalShellComponent()
|
|
result = await shell.exec_managed(
|
|
_python_command("import time; input(); time.sleep(30)"),
|
|
owner_id="group-umo",
|
|
creator_id="admin-user",
|
|
creator_is_admin=True,
|
|
sandboxed=False,
|
|
permission_check=lambda: True,
|
|
yield_time_ms=100,
|
|
)
|
|
|
|
try:
|
|
session_id = result["session_id"]
|
|
member_access = {
|
|
"owner_id": "group-umo",
|
|
"requester_id": "member-user",
|
|
"requester_is_admin": False,
|
|
}
|
|
demoted_creator_access = {
|
|
"owner_id": "group-umo",
|
|
"requester_id": "admin-user",
|
|
"requester_is_admin": False,
|
|
}
|
|
other_conversation_admin_access = {
|
|
"owner_id": "other-group-umo",
|
|
"requester_id": "other-admin",
|
|
"requester_is_admin": True,
|
|
}
|
|
|
|
assert await shell.list_sessions(**member_access) == {"sessions": []}
|
|
assert await shell.list_sessions(**demoted_creator_access) == {"sessions": []}
|
|
assert await shell.list_sessions(**other_conversation_admin_access) == {
|
|
"sessions": []
|
|
}
|
|
with pytest.raises(ValueError, match="was not found"):
|
|
await shell.poll_session(
|
|
**member_access,
|
|
session_id=session_id,
|
|
cursor=0,
|
|
)
|
|
with pytest.raises(ValueError, match="was not found"):
|
|
await shell.write_session(
|
|
**member_access,
|
|
session_id=session_id,
|
|
chars="attacker-input\n",
|
|
)
|
|
with pytest.raises(ValueError, match="was not found"):
|
|
await shell.interrupt_session(
|
|
**member_access,
|
|
session_id=session_id,
|
|
)
|
|
with pytest.raises(ValueError, match="was not found"):
|
|
await shell.poll_session(
|
|
**demoted_creator_access,
|
|
session_id=session_id,
|
|
cursor=0,
|
|
)
|
|
with pytest.raises(ValueError, match="was not found"):
|
|
await shell.terminate_session(
|
|
**other_conversation_admin_access,
|
|
session_id=session_id,
|
|
)
|
|
with pytest.raises(ValueError, match="was not found"):
|
|
await shell.terminate_session(
|
|
**member_access,
|
|
session_id=session_id,
|
|
)
|
|
|
|
assert shell._sessions[session_id].process.returncode is None
|
|
admin_sessions = await shell.list_sessions(
|
|
owner_id="group-umo",
|
|
requester_id="admin-user",
|
|
requester_is_admin=True,
|
|
)
|
|
assert [item["session_id"] for item in admin_sessions["sessions"]] == [
|
|
session_id
|
|
]
|
|
stopped = await shell.terminate_session(
|
|
owner_id="group-umo",
|
|
requester_id="admin-user",
|
|
requester_is_admin=True,
|
|
session_id=session_id,
|
|
)
|
|
assert stopped["status"] == "terminated"
|
|
finally:
|
|
await shell.shutdown_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("yield_time_ms", [5_000, 300_000])
|
|
async def test_managed_shell_accepts_stdin_and_polls_incremental_output(yield_time_ms):
|
|
shell = LocalShellComponent()
|
|
result = await shell.exec_managed(
|
|
_python_command("value = input(); print(f'got:{value}', flush=True)"),
|
|
owner_id="owner-a",
|
|
creator_id="user-a",
|
|
creator_is_admin=False,
|
|
sandboxed=False,
|
|
permission_check=lambda: True,
|
|
yield_time_ms=100,
|
|
)
|
|
|
|
try:
|
|
assert result["status"] == "running"
|
|
await shell.write_session(
|
|
owner_id="owner-a",
|
|
requester_id="user-a",
|
|
requester_is_admin=False,
|
|
session_id=result["session_id"],
|
|
chars="hello\n",
|
|
)
|
|
completed = await shell.poll_session(
|
|
owner_id="owner-a",
|
|
requester_id="user-a",
|
|
requester_is_admin=False,
|
|
session_id=result["session_id"],
|
|
yield_time_ms=yield_time_ms,
|
|
)
|
|
output = completed["stdout"]
|
|
if completed["status"] == "running":
|
|
completed = await shell.poll_session(
|
|
owner_id="owner-a",
|
|
requester_id="user-a",
|
|
requester_is_admin=False,
|
|
session_id=result["session_id"],
|
|
yield_time_ms=5_000,
|
|
)
|
|
output += completed["stdout"]
|
|
|
|
assert completed["status"] == "completed"
|
|
assert output.splitlines() == ["got:hello"]
|
|
assert completed["session_closed"] is True
|
|
finally:
|
|
await shell.shutdown_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("yield_time_ms", [-1, 300_001])
|
|
async def test_managed_shell_rejects_invalid_poll_wait(yield_time_ms):
|
|
with pytest.raises(ValueError, match="must be between 0 and 300000"):
|
|
await LocalShellComponent().poll_session(
|
|
owner_id="owner-a",
|
|
requester_id="user-a",
|
|
requester_is_admin=False,
|
|
session_id="missing",
|
|
yield_time_ms=yield_time_ms,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_managed_shell_hard_timeout_terminates_session():
|
|
shell = LocalShellComponent()
|
|
result = await shell.exec_managed(
|
|
_python_command("import time; time.sleep(30)"),
|
|
owner_id="owner-a",
|
|
creator_id="user-a",
|
|
creator_is_admin=False,
|
|
sandboxed=False,
|
|
permission_check=lambda: True,
|
|
timeout=1,
|
|
yield_time_ms=0,
|
|
)
|
|
session = shell._sessions[result["session_id"]]
|
|
|
|
try:
|
|
timed_out = await shell.poll_session(
|
|
owner_id="owner-a",
|
|
requester_id="user-a",
|
|
requester_is_admin=False,
|
|
session_id=result["session_id"],
|
|
yield_time_ms=3_000,
|
|
)
|
|
|
|
assert timed_out["status"] == "timed_out"
|
|
assert timed_out["exit_code"] is not None
|
|
assert timed_out["session_closed"] is True
|
|
assert session.output_file.closed
|
|
finally:
|
|
await shell.shutdown_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_managed_shell_keeps_completed_session_until_output_is_drained():
|
|
shell = LocalShellComponent()
|
|
result = await shell.exec_managed(
|
|
_python_command("print('x' * 25000)"),
|
|
owner_id="owner-a",
|
|
creator_id="user-a",
|
|
creator_is_admin=False,
|
|
sandboxed=False,
|
|
permission_check=lambda: True,
|
|
yield_time_ms=5_000,
|
|
max_output_chars=10_000,
|
|
)
|
|
session = shell._sessions[result["session_id"]]
|
|
|
|
try:
|
|
assert result["status"] == "completed"
|
|
assert result["has_more"] is True
|
|
assert not session.output_file.closed
|
|
output = result["stdout"]
|
|
while result["has_more"]:
|
|
result = await shell.poll_session(
|
|
owner_id="owner-a",
|
|
requester_id="user-a",
|
|
requester_is_admin=False,
|
|
session_id=result["session_id"],
|
|
max_output_chars=10_000,
|
|
)
|
|
output += result["stdout"]
|
|
|
|
assert output.splitlines() == ["x" * 25000]
|
|
assert result["session_closed"] is True
|
|
assert session.output_file.closed
|
|
assert await shell.list_sessions(
|
|
owner_id="owner-a",
|
|
requester_id="user-a",
|
|
requester_is_admin=False,
|
|
) == {"sessions": []}
|
|
finally:
|
|
await shell.shutdown_sessions()
|