mirror of
https://github.com/AstrBotDevs/AstrBot.git
synced 2026-10-16 15:06:04 +08:00
RuochenPan
3589ad5d99
Sandboxed processes can write to shared temp roots, so a named session log there could be replaced with a symlink to a host file that host-side reads and appends would follow. Keep managed shell output on an anonymous temporary file handle instead of a named path. Also stop _resolve_path_from_sandbox from consulting a booter under the local runtime: workspace and local-file checks already enforce caller permissions, and the fallback branch probed the host shell without them. Reject such paths with FileNotFoundError instead.
1573 lines
48 KiB
Python
1573 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
|
|
async def test_managed_shell_accepts_stdin_and_polls_incremental_output():
|
|
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=5_000,
|
|
)
|
|
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
|
|
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()
|