wehub-resource-sync
1557 lines
54 KiB
Python
1557 lines
54 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.sandbox.config import SandboxSettings
|
|
from opensquilla.sandbox.integration import configure_runtime, get_runtime, reset_runtime
|
|
from opensquilla.sandbox.operation_runtime import SandboxOperation, SandboxOperationResult
|
|
from opensquilla.sandbox.path_validation import decide_path_access
|
|
from opensquilla.sandbox.run_context import RunContext
|
|
from opensquilla.sandbox.run_mode import RunMode
|
|
from opensquilla.sandbox.types import SandboxRequest
|
|
from opensquilla.tools.builtin import filesystem as fs
|
|
from opensquilla.tools.builtin import shell
|
|
from opensquilla.tools.types import CallerKind, ToolContext, current_tool_context
|
|
|
|
|
|
class _InlineExecutorLoop:
|
|
async def run_in_executor(self, executor: object, func: object, *args: object) -> object:
|
|
return func(*args) # type: ignore[operator]
|
|
|
|
|
|
class _FilesystemBackend:
|
|
name = "filesystem_backend"
|
|
|
|
def operation_domains_supported(self) -> frozenset[str]:
|
|
return frozenset({"filesystem"})
|
|
|
|
async def run_operation(self, operation: SandboxOperation) -> SandboxOperationResult:
|
|
request = getattr(operation, "request", None)
|
|
path = getattr(request, "path", None)
|
|
if path is None:
|
|
raise AssertionError("filesystem operation missing path")
|
|
if operation.kind == "read_file":
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"File not found: {path}")
|
|
return SandboxOperationResult(message=path.read_text(encoding="utf-8"))
|
|
if operation.kind == "list_dir":
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Path not found: {path}")
|
|
entries = []
|
|
for entry in sorted(path.iterdir(), key=lambda item: item.name):
|
|
if entry.is_dir():
|
|
entries.append(f"[dir] {entry.name}/")
|
|
else:
|
|
entries.append(f"[file] {entry.name} ({entry.stat().st_size} bytes)")
|
|
return SandboxOperationResult(
|
|
message="\n".join(entries) if entries else f"{path}: (empty directory)"
|
|
)
|
|
if operation.kind == "write_text":
|
|
created = not path.exists()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(request.content, encoding="utf-8")
|
|
return SandboxOperationResult(
|
|
message=f"Written {len(request.content)} bytes to {path}",
|
|
created=created,
|
|
)
|
|
if operation.kind == "edit_text":
|
|
original = path.read_text(encoding="utf-8")
|
|
updated = original.replace(request.old_text, request.new_text, 1)
|
|
path.write_text(updated, encoding="utf-8")
|
|
return SandboxOperationResult(
|
|
message=(
|
|
f"Edited {path}: replaced {len(request.old_text)} chars "
|
|
f"with {len(request.new_text)} chars"
|
|
)
|
|
)
|
|
if operation.kind == "grep_search":
|
|
matches = []
|
|
for entry in sorted(path.rglob("*")):
|
|
if entry.is_symlink() or not entry.is_file():
|
|
continue
|
|
try:
|
|
text = entry.read_text(encoding="utf-8")
|
|
except UnicodeDecodeError:
|
|
continue
|
|
for line_no, line in enumerate(text.splitlines(), start=1):
|
|
if request.pattern in line:
|
|
matches.append(f"{entry}:{line_no}:{line}")
|
|
return SandboxOperationResult(
|
|
message="\n".join(matches) if matches else "No matches"
|
|
)
|
|
raise AssertionError(f"unsupported filesystem operation: {operation.kind}")
|
|
|
|
|
|
def _install_filesystem_read_backend() -> None:
|
|
runtime = get_runtime()
|
|
assert runtime is not None
|
|
runtime.backend = _FilesystemBackend()
|
|
|
|
|
|
@contextmanager
|
|
def tool_context(
|
|
workspace: Path,
|
|
*,
|
|
run_mode: str | None = "standard",
|
|
sandbox_mounts: list[dict[str, object]] | None = None,
|
|
workspace_strict: bool = False,
|
|
) -> Iterator[ToolContext]:
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(workspace),
|
|
workspace_strict=workspace_strict,
|
|
run_mode=run_mode,
|
|
session_key="s1",
|
|
sandbox_mounts=sandbox_mounts or [],
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
try:
|
|
yield ctx
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def sandbox_runtime(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
|
|
from opensquilla.application import approval_queue as approval_queue_mod
|
|
|
|
monkeypatch.setattr(
|
|
approval_queue_mod,
|
|
"_DEFAULT_APPROVAL_QUEUE_PATH",
|
|
tmp_path / "approval_queue.sqlite",
|
|
)
|
|
reset_approval_queue()
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
runtime = configure_runtime(
|
|
SandboxSettings(run_mode="standard", backend="noop", allow_legacy_mode=True),
|
|
workspace=workspace,
|
|
)
|
|
runtime.backend = _FilesystemBackend()
|
|
try:
|
|
yield
|
|
finally:
|
|
reset_approval_queue()
|
|
reset_runtime()
|
|
|
|
|
|
def test_normal_sibling_path_requests_ro_mount(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
sibling = tmp_path / "sibling" / "notes.txt"
|
|
|
|
decision = decide_path_access(sibling, workspace=workspace)
|
|
|
|
assert decision.status == "request"
|
|
assert decision.access == "ro"
|
|
assert decision.normalized_path == str(sibling.resolve(strict=False))
|
|
|
|
|
|
def test_sensitive_ssh_path_is_blocked(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
target = Path.home() / ".ssh" / "id_rsa"
|
|
|
|
decision = decide_path_access(target, workspace=workspace)
|
|
|
|
assert decision.status == "blocked"
|
|
assert decision.reason == "sensitive_path"
|
|
|
|
|
|
def test_workspace_child_is_allowed(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
target = workspace / "src" / "app.py"
|
|
|
|
decision = decide_path_access(target, workspace=workspace)
|
|
|
|
assert decision.status == "allowed"
|
|
assert decision.access == "ro"
|
|
|
|
|
|
def test_default_container_workspace_child_is_allowed_before_root_block() -> None:
|
|
workspace = "/root/.opensquilla/workspace"
|
|
target = "/root/.opensquilla/workspace/project/src/app.py"
|
|
|
|
decision = decide_path_access(target, workspace=workspace)
|
|
|
|
assert decision.status == "allowed"
|
|
assert decision.access == "ro"
|
|
|
|
|
|
def test_sensitive_file_inside_default_container_workspace_stays_blocked() -> None:
|
|
workspace = "/root/.opensquilla/workspace"
|
|
target = "/root/.opensquilla/workspace/project/.env.local"
|
|
|
|
decision = decide_path_access(target, workspace=workspace)
|
|
|
|
assert decision.status == "blocked"
|
|
assert decision.reason == "sensitive_path"
|
|
|
|
|
|
def test_write_request_asks_for_rw_mount(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
sibling = tmp_path / "sibling" / "notes.txt"
|
|
|
|
decision = decide_path_access(sibling, workspace=workspace, write=True)
|
|
|
|
assert decision.status == "request"
|
|
assert decision.access == "rw"
|
|
|
|
|
|
def test_request_path_builds_structured_mount_escalation_choices(tmp_path: Path) -> None:
|
|
from opensquilla.sandbox.escalation import build_path_approval_params
|
|
|
|
workspace = tmp_path / "workspace"
|
|
sibling = tmp_path / "sibling" / "notes.txt"
|
|
decision = decide_path_access(sibling, workspace=workspace, write=True)
|
|
|
|
proposal = build_path_approval_params(
|
|
decision,
|
|
session_key="agent:main:webchat:abc",
|
|
workspace=str(workspace),
|
|
)
|
|
|
|
assert proposal is not None
|
|
assert proposal["approvalKind"] == "sandbox_path"
|
|
assert proposal["path"] == str(sibling.resolve(strict=False))
|
|
assert proposal["access"] == "rw"
|
|
assert [choice["id"] for choice in proposal["choices"]] == [
|
|
"allow_once",
|
|
"allow_same_type",
|
|
"deny",
|
|
]
|
|
assert [choice["label"] for choice in proposal["choices"]] == [
|
|
"Allow once",
|
|
"Allow same type",
|
|
"Deny",
|
|
]
|
|
assert proposal["choices"][0]["style"] == "primary"
|
|
|
|
|
|
def test_blocked_path_has_no_mount_escalation_choices(tmp_path: Path) -> None:
|
|
from opensquilla.sandbox.escalation import build_path_approval_params
|
|
|
|
workspace = tmp_path / "workspace"
|
|
decision = decide_path_access("/", workspace=workspace, write=False)
|
|
|
|
assert decision.status == "blocked"
|
|
assert build_path_approval_params(
|
|
decision,
|
|
session_key="agent:main:webchat:abc",
|
|
workspace=str(workspace),
|
|
) is None
|
|
|
|
|
|
def test_most_specific_rw_mount_allows_write_under_ro_parent(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
parent = tmp_path / "parent"
|
|
child = parent / "child"
|
|
target = child / "out.txt"
|
|
|
|
decision = decide_path_access(
|
|
target,
|
|
workspace=workspace,
|
|
mounts=[
|
|
{"path": str(parent), "access": "ro"},
|
|
{"path": str(child), "access": "rw"},
|
|
],
|
|
write=True,
|
|
)
|
|
|
|
assert decision.status == "allowed"
|
|
assert decision.access == "rw"
|
|
|
|
|
|
def test_most_specific_ro_mount_requests_write_under_rw_parent(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
parent = tmp_path / "parent"
|
|
child = parent / "child"
|
|
target = child / "out.txt"
|
|
|
|
decision = decide_path_access(
|
|
target,
|
|
workspace=workspace,
|
|
mounts=[
|
|
{"path": str(parent), "access": "rw"},
|
|
{"path": str(child), "access": "ro"},
|
|
],
|
|
write=True,
|
|
)
|
|
|
|
assert decision.status == "request"
|
|
assert decision.access == "rw"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_ro_mount_allows_filesystem_read_and_list(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
mounted = tmp_path / "mounted"
|
|
mounted.mkdir()
|
|
missing_file = mounted / "missing.txt"
|
|
missing_dir = mounted / "missing-dir"
|
|
|
|
with tool_context(
|
|
workspace,
|
|
sandbox_mounts=[{"path": str(mounted), "access": "ro"}],
|
|
):
|
|
with pytest.raises(FileNotFoundError):
|
|
await fs.read_file(str(missing_file))
|
|
with pytest.raises(FileNotFoundError):
|
|
await fs.list_dir(str(missing_dir))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_ro_mount_allows_list_dir_when_workspace_strict(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
mounted = tmp_path / "mounted"
|
|
mounted.mkdir()
|
|
(mounted / "notes.txt").write_text("hello\n", encoding="utf-8")
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(
|
|
workspace,
|
|
sandbox_mounts=[{"path": str(mounted), "access": "ro"}],
|
|
workspace_strict=True,
|
|
):
|
|
result = await fs.list_dir(str(mounted))
|
|
|
|
assert "notes.txt" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_read_outside_workspace_requests_ro_mount(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside" / "notes.txt"
|
|
outside.parent.mkdir()
|
|
outside.write_text("outside body\n", encoding="utf-8")
|
|
|
|
with tool_context(workspace):
|
|
payload = json.loads(await fs.read_file(str(outside)))
|
|
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["approval_id"]
|
|
assert payload["path"] == str(outside.resolve(strict=False))
|
|
assert payload["access"] == "ro"
|
|
assert payload["approvalKind"] == "sandbox_path"
|
|
assert [choice["id"] for choice in payload["choices"]] == [
|
|
"allow_once",
|
|
"allow_same_type",
|
|
"deny",
|
|
]
|
|
assert "outside the current workspace" in payload["message"]
|
|
assert str(workspace) in payload["message"]
|
|
assert "read-only or read/write access" in payload["message"]
|
|
assert "appears as /workspace" not in payload["message"]
|
|
pending = get_approval_queue().get(payload["approval_id"])
|
|
assert pending.params["approvalKind"] == "sandbox_path"
|
|
assert pending.params["path"] == str(outside.resolve(strict=False))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_denied_sandbox_path_request_does_not_create_repeated_prompt(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from opensquilla.gateway.rpc_approvals import _handle_exec_approval_resolve
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(workspace):
|
|
first = json.loads(await fs.list_dir(str(outside)))
|
|
|
|
assert first["status"] == "approval_required"
|
|
assert first["approvalKind"] == "sandbox_path"
|
|
approval_id = first["approval_id"]
|
|
assert len(get_approval_queue().list_pending("exec")) == 1
|
|
|
|
await _handle_exec_approval_resolve(
|
|
{"id": approval_id, "approved": False, "choice": "deny"},
|
|
SimpleNamespace(session_manager=None, config=None),
|
|
)
|
|
|
|
with tool_context(workspace):
|
|
second = json.loads(await fs.list_dir(str(outside)))
|
|
|
|
assert second["status"] == "approval_denied"
|
|
assert second["approval_id"] == approval_id
|
|
assert "user denied" in second["message"].lower()
|
|
assert "do not ask" in second["message"].lower()
|
|
assert "Add the requested path" not in second["message"]
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_denied_sandbox_path_request_can_be_requested_again_next_turn(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from opensquilla.gateway.rpc_approvals import _handle_exec_approval_resolve
|
|
from opensquilla.sandbox.escalation import clear_sandbox_approval_denials
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(workspace):
|
|
first = json.loads(await fs.list_dir(str(outside)))
|
|
|
|
assert first["status"] == "approval_required"
|
|
approval_id = first["approval_id"]
|
|
|
|
await _handle_exec_approval_resolve(
|
|
{"id": approval_id, "approved": False, "choice": "deny"},
|
|
SimpleNamespace(session_manager=None, config=None),
|
|
)
|
|
|
|
with tool_context(workspace):
|
|
same_turn = json.loads(await fs.list_dir(str(outside)))
|
|
|
|
assert same_turn["status"] == "approval_denied"
|
|
assert same_turn["approval_id"] == approval_id
|
|
|
|
clear_sandbox_approval_denials("s1")
|
|
|
|
with tool_context(workspace):
|
|
next_turn = json.loads(await fs.list_dir(str(outside)))
|
|
|
|
assert next_turn["status"] == "approval_required"
|
|
assert next_turn["approval_id"] != approval_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_denied_sandbox_path_request_clears_duplicate_pending_prompts(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from opensquilla.gateway.rpc_approvals import _handle_exec_approval_resolve
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(workspace):
|
|
first = json.loads(await fs.list_dir(str(outside)))
|
|
second = json.loads(await fs.list_dir(str(outside)))
|
|
|
|
assert first["status"] == "approval_required"
|
|
assert second["status"] == "approval_pending"
|
|
assert second["approval_id"] == first["approval_id"]
|
|
assert len(get_approval_queue().list_pending("exec")) == 1
|
|
|
|
await _handle_exec_approval_resolve(
|
|
{"id": first["approval_id"], "approved": False, "choice": "deny"},
|
|
SimpleNamespace(session_manager=None, config=None),
|
|
)
|
|
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_write_outside_workspace_requests_rw_mount(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside" / "notes.txt"
|
|
|
|
with tool_context(workspace):
|
|
payload = json.loads(await fs.write_file(str(outside), "outside body\n"))
|
|
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["approval_id"]
|
|
assert payload["path"] == str(outside.resolve(strict=False))
|
|
assert payload["access"] == "rw"
|
|
assert payload["approvalKind"] == "sandbox_path"
|
|
assert [choice["id"] for choice in payload["choices"]] == [
|
|
"allow_once",
|
|
"allow_same_type",
|
|
"deny",
|
|
]
|
|
assert not outside.exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_sandbox_write_outside_workspace_auto_grants_rw_mount(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside" / "notes.txt"
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(workspace, run_mode="trusted") as ctx:
|
|
result = await fs.write_file(str(outside), "outside body\n")
|
|
|
|
assert "Written 13 bytes" in result
|
|
assert outside.read_text(encoding="utf-8") == "outside body\n"
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
assert ctx.sandbox_mounts == [
|
|
{"path": str(outside.parent.resolve(strict=False)), "access": "rw"}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_rw_mount_allows_write_file_without_legacy_approval(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
mounted = tmp_path / "mounted"
|
|
mounted.mkdir()
|
|
target = mounted / "out.txt"
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(
|
|
workspace,
|
|
sandbox_mounts=[{"path": str(mounted), "access": "rw"}],
|
|
):
|
|
result = await fs.write_file(str(target), "x")
|
|
|
|
assert "Written 1 bytes" in result
|
|
assert target.read_text(encoding="utf-8") == "x"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_rw_mount_allows_edit_file_without_legacy_approval(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
mounted = tmp_path / "mounted"
|
|
mounted.mkdir()
|
|
target = mounted / "out.txt"
|
|
target.write_text("old\n", encoding="utf-8")
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(
|
|
workspace,
|
|
sandbox_mounts=[{"path": str(mounted), "access": "rw"}],
|
|
):
|
|
result = await fs.edit_file(str(target), "old", "new")
|
|
|
|
assert "Edited" in result
|
|
assert target.read_text(encoding="utf-8") == "new\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_ro_mount_write_requests_rw_mount_not_legacy_approval(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
mounted = tmp_path / "mounted"
|
|
mounted.mkdir()
|
|
target = mounted / "out.txt"
|
|
|
|
with tool_context(
|
|
workspace,
|
|
sandbox_mounts=[{"path": str(mounted), "access": "ro"}],
|
|
):
|
|
payload = json.loads(await fs.write_file(str(target), "x"))
|
|
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["path"] == str(target.resolve(strict=False))
|
|
assert payload["access"] == "rw"
|
|
assert payload["approval_id"]
|
|
assert not target.exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_dir_retry_accepts_path_approval_id_after_mount(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
mounted = tmp_path / "mounted"
|
|
mounted.mkdir()
|
|
(mounted / "notes.txt").write_text("hello\n", encoding="utf-8")
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(
|
|
workspace,
|
|
sandbox_mounts=[{"path": str(mounted), "access": "ro"}],
|
|
):
|
|
result = await fs.list_dir(str(mounted), approval_id="approved-path")
|
|
|
|
assert "notes.txt" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grep_search_does_not_follow_workspace_symlink_to_unmounted_path(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
outside_file = outside / "secret.txt"
|
|
outside_file.write_text("needle secret-token\n", encoding="utf-8")
|
|
link = workspace / "linked-secret.txt"
|
|
try:
|
|
link.symlink_to(outside_file)
|
|
except OSError as exc:
|
|
if getattr(exc, "winerror", None) == 1314:
|
|
pytest.skip("creating symlinks requires Windows developer mode or elevation")
|
|
raise
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(workspace):
|
|
result = await fs.grep_search("needle", path=str(workspace))
|
|
|
|
assert "secret-token" not in result
|
|
assert "outside current sandbox view" in result or "No matches" in result
|
|
|
|
|
|
def test_shell_windows_null_redirection_does_not_request_write_access(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.sandbox.operation_profile import OperationProfile
|
|
|
|
monkeypatch.setattr(shell, "_windows_sandbox_backend_active", lambda: True)
|
|
profile = OperationProfile("unknown_shell")
|
|
|
|
assert shell._shell_write_access_targets("chcp 65001 >nul && echo ok", profile) == ()
|
|
assert shell._shell_write_access_targets("where winget 2>NUL || echo missing", profile) == ()
|
|
assert shell._shell_write_access_targets("echo ok > output.txt", profile) == (
|
|
"output.txt",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shell_read_only_workdir_outside_workspace_requests_ro_mount(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
backend_calls: list[object] = []
|
|
|
|
async def fail_backend(request: object, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
raise AssertionError("backend should not run before path access is granted")
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fail_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace):
|
|
payload = json.loads(await shell.exec_command("pwd", workdir=str(outside)))
|
|
|
|
assert payload["status"] == "approval_required"
|
|
approval_id = str(payload["approval_id"])
|
|
assert payload["path"] == str(outside.resolve(strict=False))
|
|
assert payload["access"] == "ro"
|
|
assert payload["approvalKind"] == "sandbox_path"
|
|
assert backend_calls == []
|
|
|
|
with tool_context(workspace):
|
|
pending = json.loads(
|
|
await shell.exec_command("pwd", workdir=str(outside), approval_id=approval_id)
|
|
)
|
|
|
|
assert pending["status"] == "approval_pending"
|
|
assert pending["approval_id"] == approval_id
|
|
assert backend_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shell_ro_workdir_mount_stays_read_only_in_policy(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
backend_calls: list[SandboxRequest] = []
|
|
|
|
async def fake_backend(request: SandboxRequest, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
return SimpleNamespace(stdout="", stderr="", returncode=0, backend_notes=[])
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fake_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(
|
|
workspace,
|
|
sandbox_mounts=[{"path": str(outside), "access": "ro"}],
|
|
):
|
|
await shell.exec_command(
|
|
"python -c 'open(\"x\", \"w\").write(\"1\")'",
|
|
workdir=str(outside),
|
|
)
|
|
|
|
assert len(backend_calls) == 1
|
|
request = backend_calls[0]
|
|
workspace_mount = next(
|
|
mount
|
|
for mount in request.policy.mounts
|
|
if str(mount.sandbox_path) == "/workspace"
|
|
)
|
|
outside_mount = next(
|
|
mount
|
|
for mount in request.policy.mounts
|
|
if mount.host_path == outside.resolve(strict=False)
|
|
)
|
|
assert request.cwd == outside.resolve(strict=False)
|
|
assert workspace_mount.host_path == workspace.resolve(strict=False)
|
|
assert workspace_mount.mode == "rw"
|
|
assert outside_mount.mode == "ro"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shell_workdir_relative_write_requests_rw_mount(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
backend_calls: list[object] = []
|
|
|
|
async def fail_backend(request: object, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
raise AssertionError("backend should not run before path access is granted")
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fail_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace):
|
|
payload = json.loads(await shell.exec_command("echo ok > out.txt", workdir=str(outside)))
|
|
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["path"] == str(outside.resolve(strict=False))
|
|
assert payload["access"] == "rw"
|
|
assert payload["approvalKind"] == "sandbox_path"
|
|
assert backend_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standard_shell_simple_read_path_outside_workspace_requests_ro_mount(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
backend_calls: list[object] = []
|
|
|
|
async def fail_backend(request: object, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
raise AssertionError("backend should not run before path access is granted")
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fail_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace, run_mode="standard"):
|
|
payload = json.loads(await shell.exec_command(f"ls {outside}"))
|
|
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["path"] == str(outside.resolve(strict=False))
|
|
assert payload["access"] == "ro"
|
|
assert payload["approvalKind"] == "sandbox_path"
|
|
assert backend_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standard_shell_powershell_read_path_outside_workspace_requests_ro_mount(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
backend_calls: list[object] = []
|
|
|
|
async def fail_backend(request: object, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
raise AssertionError("backend should not run before path access is granted")
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fail_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
command = f'powershell -NoProfile -Command "Get-ChildItem -LiteralPath \'{outside}\'"'
|
|
with tool_context(workspace, run_mode="standard"):
|
|
payload = json.loads(await shell.exec_command(command))
|
|
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["path"] == str(outside.resolve(strict=False))
|
|
assert payload["access"] == "ro"
|
|
assert payload["approvalKind"] == "sandbox_path"
|
|
assert backend_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_filesystem_read_path_outside_workspace_auto_mounts_without_prompt(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
target = outside / "notes.txt"
|
|
target.write_text("trusted read\n", encoding="utf-8")
|
|
_install_filesystem_read_backend()
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(workspace, run_mode="trusted") as ctx:
|
|
result = await fs.read_file(str(target))
|
|
|
|
assert "trusted read" in result
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
assert ctx.sandbox_mounts == [{"path": str(target.resolve(strict=False)), "access": "ro"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_run_context_read_path_outside_workspace_auto_mounts_without_prompt(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
target = outside / "notes.txt"
|
|
target.write_text("trusted context read\n", encoding="utf-8")
|
|
_install_filesystem_read_backend()
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(workspace, run_mode=None) as ctx:
|
|
ctx.sandbox_run_context = RunContext(
|
|
run_mode=RunMode.TRUSTED,
|
|
workspace=str(workspace),
|
|
)
|
|
result = await fs.read_file(str(target))
|
|
|
|
assert "trusted context read" in result
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
assert ctx.sandbox_mounts == [{"path": str(target.resolve(strict=False)), "access": "ro"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_shell_simple_read_path_outside_workspace_auto_mounts_without_prompt(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
backend_calls: list[object] = []
|
|
|
|
async def fake_backend(request: object, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
return SimpleNamespace(stdout="listed\n", stderr="", returncode=0, backend_notes=[])
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fake_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace, run_mode="trusted") as ctx:
|
|
result = await shell.exec_command(f"ls {outside}")
|
|
|
|
assert "exit_code=0" in result
|
|
assert "listed" in result
|
|
assert backend_calls
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
assert ctx.sandbox_mounts == [{"path": str(outside.resolve(strict=False)), "access": "ro"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_shell_powershell_read_path_outside_workspace_auto_mounts_without_prompt(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
backend_calls: list[object] = []
|
|
|
|
async def fake_backend(request: object, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
return SimpleNamespace(stdout="listed\n", stderr="", returncode=0, backend_notes=[])
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fake_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
command = f'powershell -NoProfile -Command "Get-ChildItem -LiteralPath \'{outside}\'"'
|
|
with tool_context(workspace, run_mode="trusted") as ctx:
|
|
result = await shell.exec_command(command)
|
|
|
|
assert "exit_code=0" in result
|
|
assert "listed" in result
|
|
assert backend_calls
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
assert ctx.sandbox_mounts == [{"path": str(outside.resolve(strict=False)), "access": "ro"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_shell_delete_existing_file_auto_mounts_file_without_prompt(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside" / "outside-sandbox-smoke.txt"
|
|
outside.parent.mkdir()
|
|
outside.write_text("hello\n", encoding="utf-8")
|
|
backend_calls: list[SandboxRequest] = []
|
|
|
|
async def fake_backend(request: SandboxRequest, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
return SimpleNamespace(stdout="", stderr="", returncode=0, backend_notes=[])
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fake_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=True, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace, run_mode="trusted") as ctx:
|
|
result = await shell.exec_command(f'del "{outside}"')
|
|
|
|
assert "exit_code=0" in result
|
|
assert backend_calls
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
assert ctx.sandbox_mounts == [{"path": str(outside.resolve(strict=False)), "access": "rw"}]
|
|
request = backend_calls[0]
|
|
assert any(mount.host_path == outside for mount in request.policy.mounts)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shell_write_to_protected_metadata_is_blocked_before_backend(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
repo = tmp_path / "repo"
|
|
(repo / ".git").mkdir(parents=True)
|
|
(repo / ".codex").mkdir()
|
|
backend_calls: list[SandboxRequest] = []
|
|
|
|
async def fake_backend(request: SandboxRequest, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
return SimpleNamespace(stdout="", stderr="", returncode=0, backend_notes=[])
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fake_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace, run_mode="trusted"):
|
|
git_result = await shell.exec_command(
|
|
f"sh -lc 'printf \"%s\\n\" blocked > {repo}/.git/_sandbox_should_not_write.txt'"
|
|
)
|
|
codex_result = await shell.exec_command(
|
|
f"sh -lc 'printf \"%s\\n\" blocked > {repo}/.codex/_sandbox_should_not_write.txt'"
|
|
)
|
|
|
|
git_payload = json.loads(git_result)
|
|
codex_payload = json.loads(codex_result)
|
|
assert git_payload["reason"] == "protected_metadata"
|
|
assert git_payload["protected_name"] == ".git"
|
|
assert codex_payload["reason"] == "protected_metadata"
|
|
assert codex_payload["protected_name"] == ".codex"
|
|
assert backend_calls == []
|
|
assert not (repo / ".git/_sandbox_should_not_write.txt").exists()
|
|
assert not (repo / ".codex/_sandbox_should_not_write.txt").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("metadata_dir", [".git", ".codex"])
|
|
async def test_full_host_access_shell_write_to_protected_metadata_uses_host(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
metadata_dir: str,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
repo = tmp_path / "repo"
|
|
(repo / metadata_dir).mkdir(parents=True)
|
|
target = repo / metadata_dir / "_full_host_write_probe.txt"
|
|
host_calls: list[str] = []
|
|
backend_calls: list[SandboxRequest] = []
|
|
|
|
async def fail_backend(request: SandboxRequest, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
raise AssertionError("full host access should not use the sandbox backend")
|
|
|
|
async def fake_host(
|
|
command: str,
|
|
*,
|
|
cwd: str | None,
|
|
env: dict[str, str],
|
|
stdin_bytes: bytes | None,
|
|
effective_timeout: float,
|
|
) -> str:
|
|
host_calls.append(command)
|
|
return "host-ran"
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fail_backend)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(shell, "_windows_sandbox_backend_active", lambda runtime=None: True)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
command = (
|
|
"powershell -NoProfile -Command "
|
|
f"\"Set-Content -LiteralPath '{target}' -Value full-host\""
|
|
)
|
|
with tool_context(workspace, run_mode="full"):
|
|
result = await shell.exec_command(command)
|
|
|
|
assert result == "host-ran"
|
|
assert host_calls == [command]
|
|
assert backend_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_shell_delete_existing_file_under_rw_mount_adds_file_mount(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
mounted = tmp_path / "outside"
|
|
mounted.mkdir()
|
|
outside = mounted / "outside-sandbox-smoke.txt"
|
|
outside.write_text("hello\n", encoding="utf-8")
|
|
backend_calls: list[SandboxRequest] = []
|
|
|
|
async def fake_backend(request: SandboxRequest, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
return SimpleNamespace(stdout="", stderr="", returncode=0, backend_notes=[])
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fake_backend)
|
|
monkeypatch.setattr(shell, "_windows_sandbox_backend_active", lambda runtime=None: True)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=True, reason=""),
|
|
)
|
|
|
|
with tool_context(
|
|
workspace,
|
|
run_mode="trusted",
|
|
sandbox_mounts=[{"path": str(mounted.resolve(strict=False)), "access": "rw"}],
|
|
) as ctx:
|
|
result = await shell.exec_command(f'del "{outside}"')
|
|
|
|
assert "exit_code=0" in result
|
|
assert backend_calls
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
assert ctx.sandbox_mounts == [
|
|
{"path": str(mounted.resolve(strict=False)), "access": "rw"},
|
|
{"path": str(outside.resolve(strict=False)), "access": "rw"},
|
|
]
|
|
request = backend_calls[0]
|
|
assert any(mount.host_path == outside for mount in request.policy.mounts)
|
|
|
|
|
|
def test_windows_shell_policy_ignores_deleted_active_file_mount(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.sandbox.types import (
|
|
MountSpec,
|
|
NetworkMode,
|
|
ResourceLimits,
|
|
SandboxPolicy,
|
|
SecurityLevel,
|
|
)
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
stale = workspace / "sandbox_probe_workspace.txt"
|
|
stale.write_text("workspace-ok", encoding="utf-8")
|
|
stale.unlink()
|
|
policy = SandboxPolicy(
|
|
level=SecurityLevel.STANDARD,
|
|
network=NetworkMode.NONE,
|
|
mounts=(
|
|
MountSpec(workspace, workspace, mode="rw"),
|
|
MountSpec(stale, stale, mode="rw", required=False),
|
|
),
|
|
workspace_rw=True,
|
|
tmp_writable=True,
|
|
limits=ResourceLimits(),
|
|
env_allowlist=(),
|
|
require_approval=False,
|
|
)
|
|
monkeypatch.setattr(shell, "_windows_sandbox_backend_active", lambda runtime=None: True)
|
|
|
|
with tool_context(
|
|
workspace,
|
|
run_mode="trusted",
|
|
sandbox_mounts=[{"path": str(stale.resolve(strict=False)), "access": "rw"}],
|
|
):
|
|
updated = shell._policy_with_active_tool_mounts(policy)
|
|
|
|
assert stale not in {mount.host_path for mount in updated.mounts}
|
|
assert workspace in {mount.host_path for mount in updated.mounts}
|
|
|
|
|
|
def test_shell_policy_preserves_workspace_rw_absolute_alias(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.sandbox.types import (
|
|
SANDBOX_WORKSPACE_PATH,
|
|
MountSpec,
|
|
NetworkMode,
|
|
ResourceLimits,
|
|
SandboxPolicy,
|
|
SecurityLevel,
|
|
)
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
policy = SandboxPolicy(
|
|
level=SecurityLevel.STANDARD,
|
|
network=NetworkMode.NONE,
|
|
mounts=(
|
|
MountSpec(
|
|
host_path=workspace,
|
|
sandbox_path=SANDBOX_WORKSPACE_PATH,
|
|
mode="rw",
|
|
required=True,
|
|
),
|
|
),
|
|
workspace_rw=True,
|
|
tmp_writable=True,
|
|
limits=ResourceLimits(),
|
|
env_allowlist=(),
|
|
require_approval=False,
|
|
)
|
|
monkeypatch.setattr(shell, "_windows_sandbox_backend_active", lambda runtime=None: False)
|
|
|
|
with tool_context(
|
|
workspace,
|
|
run_mode="trusted",
|
|
sandbox_mounts=[{"path": str(workspace), "access": "ro"}],
|
|
):
|
|
updated = shell._policy_with_active_tool_mounts(policy)
|
|
|
|
mounts_by_sandbox = {str(mount.sandbox_path): mount for mount in updated.mounts}
|
|
assert mounts_by_sandbox["/workspace"].mode == "rw"
|
|
assert mounts_by_sandbox[str(workspace)].mode == "rw"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shell_copy_from_outside_workspace_requests_ro_mount_before_backend(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
source = outside / "notes.txt"
|
|
target = workspace / "notes.txt"
|
|
backend_calls: list[object] = []
|
|
|
|
async def fail_backend(request: object, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
raise AssertionError("backend should not run before source path access is granted")
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fail_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace, run_mode="standard"):
|
|
payload = json.loads(await shell.exec_command(f"cp {source} {target}"))
|
|
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["path"] == str(source.resolve(strict=False))
|
|
assert payload["access"] == "ro"
|
|
assert payload["approvalKind"] == "sandbox_path"
|
|
assert backend_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standard_shell_copy_to_outside_workspace_requests_rw_mount_before_backend(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
source = workspace / "notes.txt"
|
|
target = tmp_path / "outside" / "notes.txt"
|
|
backend_calls: list[object] = []
|
|
|
|
async def fail_backend(request: object, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
raise AssertionError("backend should not run before destination path access is granted")
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fail_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace, run_mode="standard"):
|
|
payload = json.loads(await shell.exec_command(f"cp {source} {target}"))
|
|
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["path"] == str(target.resolve(strict=False))
|
|
assert payload["access"] == "rw"
|
|
assert payload["approvalKind"] == "sandbox_path"
|
|
assert backend_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_shell_copy_to_outside_workspace_auto_mounts_rw(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
source = workspace / "notes.txt"
|
|
source.write_text("hello\n", encoding="utf-8")
|
|
target = tmp_path / "outside" / "notes.txt"
|
|
backend_calls: list[SandboxRequest] = []
|
|
|
|
async def fake_backend(request: SandboxRequest, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
return SimpleNamespace(stdout="", stderr="", returncode=0, backend_notes=[])
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fake_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace, run_mode="trusted") as ctx:
|
|
result = await shell.exec_command(f"cp {source} {target}")
|
|
|
|
assert "exit_code=0" in result
|
|
assert backend_calls
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
assert ctx.sandbox_mounts == [
|
|
{"path": str(target.parent.resolve(strict=False)), "access": "rw"}
|
|
]
|
|
request = backend_calls[0]
|
|
assert any(mount.host_path == target.parent for mount in request.policy.mounts)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_shell_external_workdir_write_auto_mounts_rw(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
backend_calls: list[SandboxRequest] = []
|
|
|
|
async def fake_backend(request: SandboxRequest, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
return SimpleNamespace(stdout="", stderr="", returncode=0, backend_notes=[])
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fake_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace, run_mode="trusted") as ctx:
|
|
result = await shell.exec_command("echo hi > out.txt", workdir=str(outside))
|
|
|
|
assert "exit_code=0" in result
|
|
assert backend_calls
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
assert ctx.sandbox_mounts == [{"path": str(outside.resolve(strict=False)), "access": "rw"}]
|
|
request = backend_calls[0]
|
|
assert any(mount.host_path == outside for mount in request.policy.mounts)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shell_absolute_redirection_requests_rw_mount_before_backend(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
target = tmp_path / "outside" / "out.txt"
|
|
backend_calls: list[object] = []
|
|
|
|
async def fail_backend(request: object, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
raise AssertionError("backend should not run before redirection target is granted")
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fail_backend)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace, run_mode="standard"):
|
|
payload = json.loads(await shell.exec_command(f"echo hi > {target}"))
|
|
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["path"] == str(target.resolve(strict=False))
|
|
assert payload["access"] == "rw"
|
|
assert payload["approvalKind"] == "sandbox_path"
|
|
assert backend_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shell_simple_read_path_full_host_access_does_not_request_mount(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
backend_calls: list[object] = []
|
|
host_calls: list[tuple[str, str | None]] = []
|
|
|
|
async def fail_backend(request: object, *, runtime: object = None) -> object:
|
|
backend_calls.append(request)
|
|
raise AssertionError("full host access should not use the sandbox backend")
|
|
|
|
async def fake_host(
|
|
command: str,
|
|
*,
|
|
cwd: str | None,
|
|
env: dict[str, str],
|
|
stdin_bytes: bytes | None,
|
|
effective_timeout: float,
|
|
) -> str:
|
|
host_calls.append((command, cwd))
|
|
return "host-ran"
|
|
|
|
monkeypatch.setattr(shell, "run_under_backend", fail_backend)
|
|
monkeypatch.setattr(shell, "_run_host_shell_command", fake_host)
|
|
monkeypatch.setattr(
|
|
shell,
|
|
"check_safe_bin",
|
|
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
|
|
)
|
|
|
|
with tool_context(workspace, run_mode="full"):
|
|
result = await shell.exec_command(f"ls {outside}")
|
|
|
|
assert result == "host-ran"
|
|
assert host_calls == [(f"ls {outside}", str(workspace.resolve()))]
|
|
assert backend_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_path_request_approval_does_not_mutate_until_choice_is_resolved(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from opensquilla.gateway.rpc import RpcContext, get_dispatcher
|
|
from opensquilla.sandbox.run_context import get_run_context
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside" / "notes.txt"
|
|
manager = SimpleNamespace()
|
|
manager.node = SimpleNamespace(session_key="s1", agent_id="main", origin=None)
|
|
|
|
async def _get_session(session_key: str):
|
|
return manager.node if session_key == manager.node.session_key else None
|
|
|
|
async def _update(session_key: str, **fields):
|
|
for key, value in fields.items():
|
|
setattr(manager.node, key, value)
|
|
return manager.node
|
|
|
|
manager.get_session = _get_session
|
|
manager.update = _update
|
|
config = SimpleNamespace(
|
|
sandbox=SimpleNamespace(run_mode="standard", sandbox=True, security_grading=True),
|
|
permissions=SimpleNamespace(default_mode="off"),
|
|
)
|
|
|
|
with tool_context(workspace):
|
|
payload = json.loads(await fs.write_file(str(outside), "outside body\n"))
|
|
|
|
approval_id = str(payload["approval_id"])
|
|
pending = get_approval_queue().get(approval_id)
|
|
assert pending.resolved is False
|
|
assert outside.exists() is False
|
|
saved_before = await get_run_context(manager, "s1", config=config, workspace=str(workspace))
|
|
assert saved_before.mounts == ()
|
|
|
|
result = await get_dispatcher().dispatch(
|
|
"r1",
|
|
"exec.approval.resolve",
|
|
{"id": approval_id, "approved": True, "choice": "allow_same_type"},
|
|
RpcContext(conn_id="test", session_manager=manager, config=config),
|
|
)
|
|
assert result.error is None, result.error
|
|
|
|
saved_after = await get_run_context(manager, "s1", config=config, workspace=str(workspace))
|
|
assert [(mount.path, mount.access, mount.scope) for mount in saved_after.mounts] == [
|
|
(str(outside.resolve(strict=False)), "rw", "chat")
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_retry_uses_resolved_rw_mount_even_with_stale_tool_context(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.gateway.rpc import RpcContext, get_dispatcher
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside" / "notes.txt"
|
|
manager = SimpleNamespace()
|
|
manager.node = SimpleNamespace(session_key="s1", agent_id="main", origin=None)
|
|
|
|
async def _get_session(session_key: str):
|
|
return manager.node if session_key == manager.node.session_key else None
|
|
|
|
async def _update(session_key: str, **fields):
|
|
for key, value in fields.items():
|
|
setattr(manager.node, key, value)
|
|
return manager.node
|
|
|
|
manager.get_session = _get_session
|
|
manager.update = _update
|
|
config = SimpleNamespace(
|
|
sandbox=SimpleNamespace(run_mode="standard", sandbox=True, security_grading=True),
|
|
permissions=SimpleNamespace(default_mode="off"),
|
|
)
|
|
monkeypatch.setattr(fs.asyncio, "get_event_loop", lambda: _InlineExecutorLoop())
|
|
|
|
with tool_context(workspace):
|
|
payload = json.loads(await fs.write_file(str(outside), "outside body\n"))
|
|
approval_id = str(payload["approval_id"])
|
|
|
|
result = await get_dispatcher().dispatch(
|
|
"r1",
|
|
"exec.approval.resolve",
|
|
{"id": approval_id, "approved": True, "choice": "allow_same_type"},
|
|
RpcContext(conn_id="test", session_manager=manager, config=config),
|
|
)
|
|
assert result.error is None, result.error
|
|
|
|
retried = await fs.write_file(str(outside), "outside body\n", approval_id=approval_id)
|
|
|
|
assert "Written 13 bytes" in retried
|
|
assert outside.read_text(encoding="utf-8") == "outside body\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_request_rejects_removed_mount_choice(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from opensquilla.gateway.rpc import RpcContext, get_dispatcher
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
outside = tmp_path / "outside" / "notes.txt"
|
|
manager = SimpleNamespace()
|
|
manager.node = SimpleNamespace(session_key="s1", agent_id="main", origin=None)
|
|
|
|
async def _get_session(session_key: str):
|
|
return manager.node if session_key == manager.node.session_key else None
|
|
|
|
async def _update(session_key: str, **fields):
|
|
for key, value in fields.items():
|
|
setattr(manager.node, key, value)
|
|
return manager.node
|
|
|
|
manager.get_session = _get_session
|
|
manager.update = _update
|
|
config = SimpleNamespace(
|
|
sandbox=SimpleNamespace(run_mode="standard", sandbox=True, security_grading=True),
|
|
permissions=SimpleNamespace(default_mode="off"),
|
|
)
|
|
|
|
with tool_context(workspace):
|
|
payload = json.loads(await fs.write_file(str(outside), "outside body\n"))
|
|
approval_id = str(payload["approval_id"])
|
|
|
|
result = await get_dispatcher().dispatch(
|
|
"r1",
|
|
"exec.approval.resolve",
|
|
{"id": approval_id, "approved": True, "choice": "mount_ro_chat"},
|
|
RpcContext(conn_id="test", session_manager=manager, config=config),
|
|
)
|
|
assert result.error is not None
|
|
|
|
assert outside.exists() is False
|