Files
opensquilla--opensquilla/tests/test_tools/test_filesystem_read_workspace.py
2026-07-13 13:12:33 +08:00

587 lines
20 KiB
Python

from __future__ import annotations
import json
import os
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from types import SimpleNamespace
import pytest
from opensquilla.attachment_refs import write_transcript_material
from opensquilla.engine.types import ToolCall
from opensquilla.sandbox import sensitive_paths
from opensquilla.sandbox.config import SandboxSettings
from opensquilla.sandbox.integration import configure_runtime, reset_runtime
from opensquilla.tools.builtin import filesystem as fs
from opensquilla.tools.dispatch import build_tool_handler
from opensquilla.tools.registry import get_default_registry
from opensquilla.tools.types import CallerKind, ToolContext, ToolError, current_tool_context
@contextmanager
def tool_context(
workspace: Path,
*,
strict: bool = True,
artifact_media_root: Path | None = None,
artifact_session_id: str | None = None,
run_mode: str | None = None,
) -> Iterator[None]:
token = current_tool_context.set(
ToolContext(
caller_kind=CallerKind.CLI,
channel_kind="cli",
channel_id="cli:test",
workspace_dir=str(workspace),
workspace_strict=strict,
artifact_media_root=str(artifact_media_root) if artifact_media_root else None,
artifact_session_id=artifact_session_id,
run_mode=run_mode,
)
)
try:
yield
finally:
current_tool_context.reset(token)
@pytest.mark.asyncio
async def test_read_file_offset_limit_does_not_call_read_bytes(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
target = tmp_path / "big.log"
target.write_text("".join(f"line {i}\n" for i in range(1, 1001)), encoding="utf-8")
def fail_read_bytes(self: Path) -> bytes: # pragma: no cover - must not be called
raise AssertionError("read_bytes should not be used for bounded read_file")
monkeypatch.setattr(Path, "read_bytes", fail_read_bytes)
output = await fs.read_file(str(target), offset=500, limit=2)
assert "500\tline 500" in output
assert "501\tline 501" in output
assert "499\tline 499" not in output
assert "502\tline 502" not in output
@pytest.mark.asyncio
async def test_read_file_binary_detection_samples_first_8192_bytes(tmp_path: Path) -> None:
first_sample = tmp_path / "first.txt"
first_sample.write_bytes(b"abc\x00def")
with pytest.raises(ToolError, match="NUL"):
await fs.read_file(str(first_sample), limit=1)
later_nul = tmp_path / "later.txt"
later_nul.write_bytes(("ok\n" * 4100).encode("utf-8") + b"\x00tail\n")
output = await fs.read_file(str(later_nul), offset=1, limit=1)
assert output == "1\tok\n"
@pytest.mark.asyncio
async def test_read_file_invalid_utf8_before_selected_window_errors(tmp_path: Path) -> None:
target = tmp_path / "invalid.txt"
target.write_bytes(b"ok\n\xff\nlater\n")
with pytest.raises(ToolError, match="not valid UTF-8"):
await fs.read_file(str(target), offset=3, limit=1)
@pytest.mark.asyncio
async def test_workspace_strict_allows_inside_workspace(tmp_path: Path) -> None:
text_file = tmp_path / "inside.txt"
text_file.write_text("needle\n", encoding="utf-8")
csv_file = tmp_path / "inside.csv"
csv_file.write_text("a,b\n1,2\n", encoding="utf-8")
with tool_context(tmp_path):
assert "1\tneedle" in await fs.read_file(str(text_file))
assert "inside.csv" in await fs.read_spreadsheet(str(csv_file))
assert "inside.txt" in await fs.list_dir(str(tmp_path))
assert "inside.txt" in await fs.glob_search("*.txt", path=str(tmp_path))
assert "needle" in await fs.grep_search("needle", path=str(tmp_path))
@pytest.mark.asyncio
async def test_grep_search_skips_vcs_metadata_but_keeps_github(
tmp_path: Path,
) -> None:
git_dir = tmp_path / ".git"
git_dir.mkdir()
(git_dir / "index").write_text("needle hidden metadata\n", encoding="utf-8")
github_dir = tmp_path / ".github"
github_dir.mkdir()
(github_dir / "workflow.yml").write_text("needle workflow\n", encoding="utf-8")
(tmp_path / "src.py").write_text("needle source\n", encoding="utf-8")
with tool_context(tmp_path):
grepped = await fs.grep_search("needle", path=str(tmp_path))
globbed = await fs.glob_search("**/*", path=str(tmp_path))
assert ".git/index" not in grepped
assert ".git/index" not in globbed
assert "workflow.yml:1: needle workflow" in grepped
assert "src.py:1: needle source" in grepped
@pytest.mark.asyncio
async def test_grep_search_skips_binary_and_invalid_utf8_files(tmp_path: Path) -> None:
(tmp_path / "source.txt").write_text("needle source\n", encoding="utf-8")
(tmp_path / "archive.zip").write_bytes(b"needle binary payload")
(tmp_path / "invalid.txt").write_bytes(b"needle \xff invalid utf8")
with tool_context(tmp_path):
grepped = await fs.grep_search("needle", path=str(tmp_path))
assert "source.txt:1: needle source" in grepped
assert "archive.zip" not in grepped
assert "invalid.txt" not in grepped
@pytest.mark.asyncio
async def test_grep_search_truncates_long_matching_lines(tmp_path: Path) -> None:
long_tail = "x" * 3000
(tmp_path / "source.txt").write_text(f"needle {long_tail}\n", encoding="utf-8")
with tool_context(tmp_path):
grepped = await fs.grep_search("needle", path=str(tmp_path))
assert "source.txt:1: needle " in grepped
assert "line truncated" in grepped
assert "omitted_chars=" in grepped
assert long_tail not in grepped
@pytest.mark.asyncio
async def test_grep_search_clamps_large_max_results(tmp_path: Path) -> None:
lines = [f"needle {i}" for i in range(1100)]
(tmp_path / "source.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")
with tool_context(tmp_path):
grepped = await fs.grep_search("needle", path=str(tmp_path), max_results=5000)
assert "limit: 1000" in grepped
assert "has_more: true" in grepped
result_lines = grepped.split("---\n", 1)[1].splitlines()
assert len(result_lines) == 1000
assert "needle 999" in result_lines[-1]
assert "needle 1000" not in grepped
@pytest.mark.asyncio
async def test_grep_search_supports_offset_and_has_more(tmp_path: Path) -> None:
lines = [f"needle {i}" for i in range(5)]
(tmp_path / "source.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")
with tool_context(tmp_path):
grepped = await fs.grep_search("needle", path=str(tmp_path), max_results=2, offset=2)
assert "returned: 2" in grepped
assert "offset: 2" in grepped
assert "limit: 2" in grepped
assert "has_more: true" in grepped
body = grepped.split("---\n", 1)[1]
assert "needle 2" in body
assert "needle 3" in body
assert "needle 1" not in body
assert "needle 4" not in body
@pytest.mark.asyncio
async def test_grep_search_allows_explicit_unlimited_and_hidden_line_numbers(
tmp_path: Path,
) -> None:
(tmp_path / "source.txt").write_text("needle a\nneedle b\n", encoding="utf-8")
with tool_context(tmp_path):
grepped = await fs.grep_search(
"needle",
path=str(tmp_path),
max_results=0,
include_line_numbers=False,
)
assert "limit: unlimited" in grepped
assert "has_more: false" in grepped
body = grepped.split("---\n", 1)[1]
assert "source.txt: needle a" in body
assert "source.txt:1:" not in body
@pytest.mark.asyncio
async def test_workspace_strict_allows_current_session_material_read(tmp_path: Path) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
media_root = tmp_path / "media"
sha, material_path, _wrote = write_transcript_material(
media_root=media_root,
session_id="s1",
payload=b"material body\n",
)
assert sha
with tool_context(
workspace,
artifact_media_root=media_root,
artifact_session_id="s1",
):
output = await fs.read_file(str(material_path))
assert "1\tmaterial body" in output
@pytest.mark.asyncio
async def test_workspace_strict_blocks_other_session_material_read(tmp_path: Path) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
media_root = tmp_path / "media"
_sha, other_material_path, _wrote = write_transcript_material(
media_root=media_root,
session_id="s2",
payload=b"other material\n",
)
with tool_context(
workspace,
artifact_media_root=media_root,
artifact_session_id="s1",
):
with pytest.raises(ToolError, match="outside active read roots"):
await fs.read_file(str(other_material_path))
@pytest.mark.asyncio
async def test_workspace_inside_sensitive_parent_allows_normal_files(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
monkeypatch.setattr(sensitive_paths, "_SENSITIVE_PREFIXES", (str(tmp_path),))
monkeypatch.setattr(
sensitive_paths,
"_WORKSPACE_PARENT_EXCEPTION_MARKERS",
(str(tmp_path),),
)
target = workspace / "notes" / "plan.md"
target.parent.mkdir()
target.write_text("hello\n", encoding="utf-8")
with tool_context(workspace):
write_gate = await fs._gate_out_of_workspace_write(
"write_file",
target.resolve(),
"notes/plan.md",
None,
)
read_result = await fs.read_file("notes/plan.md")
listed = await fs.list_dir("notes")
globbed = await fs.glob_search("*.md", path="notes")
grepped = await fs.grep_search("hello", path="notes")
assert write_gate is None
assert "1\thello" in read_result
assert "plan.md" in listed
assert "plan.md" in globbed
assert "plan.md:1: hello" in grepped
@pytest.mark.asyncio
async def test_workspace_inside_sensitive_parent_keeps_leaf_secret_blocks(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
monkeypatch.setattr(sensitive_paths, "_SENSITIVE_PREFIXES", (str(tmp_path),))
monkeypatch.setattr(
sensitive_paths,
"_WORKSPACE_PARENT_EXCEPTION_MARKERS",
(str(tmp_path),),
)
with tool_context(workspace):
payload = await fs._gate_out_of_workspace_write(
"write_file",
(workspace / ".env").resolve(),
".env",
None,
)
assert payload is not None
assert payload["status"] == "blocked"
assert payload["reason"] == "sensitive_path"
assert not (workspace / ".env").exists()
def test_resolve_path_rejects_foreign_posix_absolute_path_on_windows(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(fs, "os", SimpleNamespace(name="nt"), raising=False)
workspace = tmp_path / "workspace"
workspace.mkdir()
with tool_context(workspace):
with pytest.raises(ToolError) as exc_info:
fs._resolve_path("/Users/a1/Desktop/report.pptx")
message = str(exc_info.value)
assert "foreign_host_path" in message
assert "/Users/a1/Desktop/report.pptx" in message
assert "workspace-relative" in message
assert "D:\\Users" not in message
@pytest.mark.asyncio
async def test_workspace_strict_blocks_outside_base_path(tmp_path: Path) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
outside_file = outside / "outside.txt"
outside_file.write_text("secret\n", encoding="utf-8")
outside_csv = outside / "outside.csv"
outside_csv.write_text("a,b\n1,2\n", encoding="utf-8")
with tool_context(workspace):
for call in (
lambda: fs.read_file(str(outside_file)),
lambda: fs.read_spreadsheet(str(outside_csv)),
lambda: fs.list_dir(str(outside)),
lambda: fs.glob_search("*.txt", path=str(outside)),
lambda: fs.grep_search("secret", path=str(outside)),
):
with pytest.raises(ToolError, match="outside active read roots"):
await call()
@pytest.mark.asyncio
async def test_full_host_access_allows_reading_outside_active_read_roots(
tmp_path: Path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
outside_file = outside / "outside.txt"
outside_file.write_text("needle\n", encoding="utf-8")
outside_csv = outside / "outside.csv"
outside_csv.write_text("a,b\n1,2\n", encoding="utf-8")
with tool_context(workspace, run_mode="full"):
assert "1\tneedle" in await fs.read_file(str(outside_file))
assert "Workbook: outside.csv" in await fs.read_spreadsheet(str(outside_csv))
assert "outside.txt" in await fs.list_dir(str(outside))
assert str(outside_file) in await fs.glob_search("*.txt", path=str(outside))
assert "outside.txt:1: needle" in await fs.grep_search("needle", path=str(outside))
@pytest.mark.asyncio
async def test_workspace_strict_block_is_actionable_in_tool_failure_envelope(
tmp_path: Path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
handler = build_tool_handler(get_default_registry())
with tool_context(workspace):
result = await handler(
ToolCall(
tool_use_id="tc-glob-outside",
tool_name="glob_search",
arguments={"pattern": "*.txt", "path": str(outside)},
)
)
envelope = json.loads(result.content)
assert result.is_error is True
assert envelope["status"] == "error"
assert envelope["tool"] == "glob_search"
assert "outside active read roots" in envelope["user_message"]
assert "internal error" not in envelope["user_message"]
assert envelope["retry_allowed"] is False
@pytest.mark.asyncio
async def test_workspace_write_deny_block_is_actionable_in_tool_failure_envelope(
tmp_path: Path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
configure_runtime(
SandboxSettings(sandbox=False, security_grading=False, allow_legacy_mode=True),
workspace=workspace,
)
handler = build_tool_handler(get_default_registry())
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
channel_kind="cli",
channel_id="cli:test",
elevated="bypass",
workspace_dir=str(workspace),
workspace_write_deny_globs=["**/test_*.py"],
)
)
try:
result = await handler(
ToolCall(
tool_use_id="tc-write-deny",
tool_name="write_file",
arguments={
"path": str(workspace / "test_bug.py"),
"content": "print('nope')\n",
},
)
)
finally:
current_tool_context.reset(token)
reset_runtime()
envelope = json.loads(result.content)
assert result.is_error is True
assert envelope["status"] == "error"
assert envelope["tool"] == "write_file"
assert "workspace write deny policy" in envelope["user_message"]
assert "internal error" not in envelope["user_message"]
assert envelope["retry_allowed"] is False
assert not (workspace / "test_bug.py").exists()
@pytest.mark.asyncio
async def test_workspace_strict_blocks_nonexistent_outside_before_not_found(
tmp_path: Path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
outside_missing = tmp_path / "outside" / "missing.txt"
outside_missing_dir = tmp_path / "outside" / "missing-dir"
with tool_context(workspace):
for call in (
lambda: fs.read_file(str(outside_missing)),
lambda: fs.list_dir(str(outside_missing_dir)),
lambda: fs.glob_search("*.txt", path=str(outside_missing_dir)),
lambda: fs.grep_search("needle", path=str(outside_missing_dir)),
):
with pytest.raises(ToolError, match="outside active read roots"):
await call()
@pytest.mark.asyncio
async def test_workspace_strict_disabled_allows_outside_read_when_not_sensitive(
tmp_path: Path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
outside = tmp_path / "outside.txt"
outside.write_text("outside\n", encoding="utf-8")
with tool_context(workspace, strict=False):
assert "outside" in await fs.read_file(str(outside))
def _make_symlink(link: Path, target: Path) -> None:
try:
os.symlink(target, link)
except (OSError, NotImplementedError) as exc:
pytest.skip(f"symlink unsupported/unavailable: {exc}")
@pytest.mark.asyncio
async def test_workspace_strict_blocks_read_file_symlink_escape(tmp_path: Path) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
outside = tmp_path / "outside.txt"
outside.write_text("outside\n", encoding="utf-8")
link = workspace / "link.txt"
_make_symlink(link, outside)
with tool_context(workspace):
with pytest.raises(ToolError, match="outside active read roots"):
await fs.read_file(str(link))
@pytest.mark.asyncio
async def test_workspace_strict_surfaces_list_dir_symlink_escape(tmp_path: Path) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
outside = tmp_path / "outside.txt"
outside.write_text("outside\n", encoding="utf-8")
link = workspace / "link.txt"
_make_symlink(link, outside)
with tool_context(workspace):
output = await fs.list_dir(str(workspace))
assert "[blocked]" in output
assert "outside active read roots" in output
@pytest.mark.asyncio
async def test_list_dir_survives_broken_symlink(tmp_path: Path) -> None:
(tmp_path / "ok.txt").write_text("hello\n", encoding="utf-8")
_make_symlink(tmp_path / "dangling", tmp_path / "missing-target")
with tool_context(tmp_path):
output = await fs.list_dir(str(tmp_path))
assert "ok.txt" in output
assert "dangling" in output
@pytest.mark.asyncio
async def test_workspace_strict_surfaces_glob_and_grep_symlink_escape(
tmp_path: Path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
outside = tmp_path / "outside.txt"
outside.write_text("needle\n", encoding="utf-8")
link = workspace / "link.txt"
_make_symlink(link, outside)
with tool_context(workspace):
globbed = await fs.glob_search("*.txt", path=str(workspace))
grepped = await fs.grep_search("needle", path=str(workspace))
assert "[blocked]" in globbed
assert "outside active read roots" in globbed
assert "[blocked]" in grepped
assert "outside active read roots" in grepped
@pytest.mark.asyncio
async def test_sensitive_path_priority_over_workspace_strict(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
outside = tmp_path / "secret"
outside.mkdir()
outside_file = outside / "secret.txt"
outside_file.write_text("secret\n", encoding="utf-8")
monkeypatch.setattr(
"opensquilla.sandbox.sensitive_paths.is_sensitive_path",
lambda path: "/secret" if "secret" in path else None,
)
with tool_context(workspace):
file_result = json.loads(await fs.read_file(str(outside_file)))
dir_result = json.loads(await fs.list_dir(str(outside)))
assert file_result["reason"] == "sensitive_path"
assert dir_result["reason"] == "sensitive_path"
assert "workspace_strict" not in file_result.get("message", "")
assert "workspace_strict" not in dir_result.get("message", "")