Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Clarify filesystem workspace write policy

Browse Source
This commit is contained in:
chengyongru
2026-06-18 00:03:26 +08:00
committed by Xubin Ren
parent fc635377bc
commit 732992df4f
9 changed files with 371 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/agent/test_dream.py
+85 -7
View File
@@ -126,6 +126,86 @@ class TestDreamTools:
"write_file",
}
@pytest.mark.asyncio
async def test_dream_can_edit_canonical_memory_files(self, store):
tools = store.build_dream_tools()
memory_result = await tools.execute(
"apply_patch",
{
"edits": [
{
"path": "memory/MEMORY.md",
"action": "replace",
"old_text": "Project X active",
"new_text": "Project Y active",
}
]
},
)
soul_result = await tools.execute(
"edit_file",
{
"path": "SOUL.md",
"old_text": "Helpful",
"new_text": "Precise",
},
)
assert "Patch applied" in memory_result
assert "Successfully edited" in soul_result
assert "Project Y active" in store.memory_file.read_text(encoding="utf-8")
assert "Precise" in store.soul_file.read_text(encoding="utf-8")
@pytest.mark.asyncio
async def test_dream_can_write_workspace_skills(self, store):
tools = store.build_dream_tools()
target = store.workspace / "skills" / "demo" / "SKILL.md"
result = await tools.execute(
"write_file",
{
"path": "skills/demo/SKILL.md",
"content": "---\nname: demo\ndescription: Demo skill.\n---\n\nUse when needed.\n",
},
)
assert "Successfully wrote" in result
assert target.read_text(encoding="utf-8").startswith("---\nname: demo")
@pytest.mark.asyncio
async def test_dream_cannot_modify_memory_internal_files(self, store):
tools = store.build_dream_tools()
store.history_file.write_text("before\n", encoding="utf-8")
store._dream_cursor_file.write_text("1", encoding="utf-8")
history_result = await tools.execute(
"apply_patch",
{
"edits": [
{
"path": "memory/history.jsonl",
"action": "replace",
"old_text": "before",
"new_text": "after",
}
]
},
)
cursor_result = await tools.execute(
"edit_file",
{
"path": "memory/.dream_cursor",
"old_text": "1",
"new_text": "2",
},
)
assert "outside allowed directory" in history_result
assert "outside allowed directory" in cursor_result
assert store.history_file.read_text(encoding="utf-8") == "before\n"
assert store._dream_cursor_file.read_text(encoding="utf-8") == "1"
class TestEphemeralDirect:
"""Tests for the ephemeral flag that skips history.jsonl writes for Dream."""
@@ -149,7 +229,9 @@ class TestEphemeralDirect:
provider.supports_tools = True
provider.generation = MagicMock(max_tokens=4096)
provider.chat_with_retry = AsyncMock(
return_value=LLMResponse(content="done", tool_calls=[], finish_reason="stop", usage={})
return_value=MagicMock(
content="done", finish_reason="stop", tool_calls=[], usage={},
)
)
with (
@@ -181,13 +263,9 @@ class TestEphemeralDirect:
mock_archive.assert_not_called()
async def test_non_ephemeral_runs_normally(self, tmp_path, _make_loop):
"""Without ephemeral, the normal path returns the model response."""
"""Without ephemeral, the normal path is untouched — no crash."""
loop, store = _make_loop
response = await loop.process_direct("test", session_key="cli:normal")
assert response is not None
assert response.content == "done"
loop.provider.chat_with_retry.assert_awaited()
await loop.process_direct("test", session_key="cli:normal")
async def test_ephemeral_sets_ctx_flag(self, tmp_path, _make_loop):
"""Verify that ephemeral=True is forwarded to TurnContext."""
tests/tools/test_apply_patch_tool.py
+135 -31
View File
@@ -256,14 +256,56 @@ def test_apply_patch_edits_dry_run_validates_without_writing(tmp_path):
assert not (tmp_path / "added.txt").exists()
def test_apply_patch_edits_rejects_absolute_and_parent_paths(tmp_path):
tool = ApplyPatchTool(workspace=tmp_path)
def test_apply_patch_edits_allows_absolute_and_parent_paths_when_unrestricted(tmp_path):
workspace = tmp_path / "workspace"
outside = tmp_path / "outside"
workspace.mkdir()
outside.mkdir()
absolute_target = outside / "absolute.txt"
parent_target = outside / "parent.txt"
tool = ApplyPatchTool(workspace=workspace, restrict_to_workspace=False)
absolute = asyncio.run(
tool.execute(
edits=[
{
"path": "/tmp/owned.txt",
"path": str(absolute_target),
"action": "add",
"new_text": "absolute",
}
]
)
)
parent = asyncio.run(
tool.execute(
edits=[
{
"path": "../outside/parent.txt",
"action": "add",
"new_text": "parent",
}
]
)
)
assert "Patch applied" in absolute
assert "Patch applied" in parent
assert absolute_target.read_text() == "absolute\n"
assert parent_target.read_text() == "parent\n"
def test_apply_patch_edits_rejects_outside_paths_when_restricted(tmp_path):
workspace = tmp_path / "workspace"
outside = tmp_path / "outside"
workspace.mkdir()
outside.mkdir()
tool = ApplyPatchTool(workspace=workspace, allowed_dir=workspace)
absolute = asyncio.run(
tool.execute(
edits=[
{
"path": str(outside / "absolute.txt"),
"action": "add",
"new_text": "nope",
}
@@ -274,29 +316,7 @@ def test_apply_patch_edits_rejects_absolute_and_parent_paths(tmp_path):
tool.execute(
edits=[
{
"path": "../owned.txt",
"action": "add",
"new_text": "nope",
}
]
)
)
windows_absolute = asyncio.run(
tool.execute(
edits=[
{
"path": r"C:\owned.txt",
"action": "add",
"new_text": "nope",
}
]
)
)
windows_parent = asyncio.run(
tool.execute(
edits=[
{
"path": r"..\owned.txt",
"path": "../outside/parent.txt",
"action": "add",
"new_text": "nope",
}
@@ -304,11 +324,95 @@ def test_apply_patch_edits_rejects_absolute_and_parent_paths(tmp_path):
)
)
assert "must be relative" in absolute
assert "must not contain '..'" in parent
assert "must be relative" in windows_absolute
assert "must not contain '..'" in windows_parent
assert not (tmp_path.parent / "owned.txt").exists()
assert "outside allowed directory" in absolute
assert "outside allowed directory" in parent
assert not (outside / "absolute.txt").exists()
assert not (outside / "parent.txt").exists()
def test_apply_patch_legacy_extra_allowed_dirs_are_read_only(tmp_path):
workspace = tmp_path / "workspace"
skills = tmp_path / "skills"
workspace.mkdir()
skills.mkdir()
target = skills / "demo.md"
target.write_text("before\n", encoding="utf-8")
tool = ApplyPatchTool(
workspace=workspace,
allowed_dir=workspace,
extra_allowed_dirs=[skills],
)
result = asyncio.run(
tool.execute(
edits=[
{
"path": str(target),
"action": "replace",
"old_text": "before",
"new_text": "after",
}
]
)
)
assert "outside allowed directory" in result
assert target.read_text(encoding="utf-8") == "before\n"
def test_apply_patch_media_dir_is_read_only_by_default(tmp_path, monkeypatch):
workspace = tmp_path / "workspace"
media = tmp_path / "media"
workspace.mkdir()
media.mkdir()
target = media / "demo.md"
target.write_text("before\n", encoding="utf-8")
monkeypatch.setattr("nanobot.agent.tools.path_utils.get_media_dir", lambda: media)
tool = ApplyPatchTool(workspace=workspace, allowed_dir=workspace)
result = asyncio.run(
tool.execute(
edits=[
{
"path": str(target),
"action": "replace",
"old_text": "before",
"new_text": "after",
}
]
)
)
assert "outside allowed directory" in result
assert target.read_text(encoding="utf-8") == "before\n"
def test_apply_patch_allows_explicit_extra_write_allowed_dirs_when_restricted(tmp_path):
workspace = tmp_path / "workspace"
writable = tmp_path / "writable"
workspace.mkdir()
writable.mkdir()
target = writable / "allowed.txt"
tool = ApplyPatchTool(
workspace=workspace,
allowed_dir=workspace,
extra_write_allowed_dirs=[writable],
)
result = asyncio.run(
tool.execute(
edits=[
{
"path": str(target),
"action": "add",
"new_text": "allowed",
}
]
)
)
assert "Patch applied" in result
assert target.read_text(encoding="utf-8") == "allowed\n"
def test_apply_patch_edits_reports_invalid_edit_shapes(tmp_path):
tests/tools/test_filesystem_tools.py
+70 -9
View File
@@ -6,6 +6,7 @@ from nanobot.agent.tools.filesystem import (
EditFileTool,
ListDirTool,
ReadFileTool,
WriteFileTool,
_find_match,
)
@@ -283,7 +284,7 @@ class TestListDirTool:
# ---------------------------------------------------------------------------
# Workspace restriction + extra_allowed_dirs
# Workspace restriction + extra read/write allowed dirs
# ---------------------------------------------------------------------------
class TestWorkspaceRestriction:
@@ -314,7 +315,7 @@ class TestWorkspaceRestriction:
tool = ReadFileTool(
workspace=workspace, allowed_dir=workspace,
extra_allowed_dirs=[skills_dir],
extra_read_allowed_dirs=[skills_dir],
)
result = await tool.execute(path=str(skill_file))
assert "Test Skill" in result
@@ -337,18 +338,52 @@ class TestWorkspaceRestriction:
assert "Error" not in result
@pytest.mark.asyncio
async def test_extra_dirs_does_not_widen_write(self, tmp_path):
from nanobot.agent.tools.filesystem import WriteFileTool
async def test_write_blocked_in_media_dir_by_default(self, tmp_path, monkeypatch):
workspace = tmp_path / "ws"
workspace.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
media_dir = tmp_path / "media"
media_dir.mkdir()
monkeypatch.setattr("nanobot.agent.tools.path_utils.get_media_dir", lambda: media_dir)
tool = WriteFileTool(workspace=workspace, allowed_dir=workspace)
result = await tool.execute(path=str(outside / "hack.txt"), content="pwned")
result = await tool.execute(path=str(media_dir / "hack.txt"), content="pwned")
assert "Error" in result
assert "outside" in result.lower()
assert not (media_dir / "hack.txt").exists()
@pytest.mark.asyncio
async def test_legacy_extra_allowed_dirs_does_not_widen_write(self, tmp_path):
workspace = tmp_path / "ws"
workspace.mkdir()
skills_dir = tmp_path / "skills"
skills_dir.mkdir()
tool = WriteFileTool(
workspace=workspace,
allowed_dir=workspace,
extra_allowed_dirs=[skills_dir],
)
result = await tool.execute(path=str(skills_dir / "hack.txt"), content="pwned")
assert "Error" in result
assert "outside" in result.lower()
assert not (skills_dir / "hack.txt").exists()
@pytest.mark.asyncio
async def test_write_allowed_with_extra_write_dir(self, tmp_path):
workspace = tmp_path / "ws"
workspace.mkdir()
writable = tmp_path / "writable"
writable.mkdir()
tool = WriteFileTool(
workspace=workspace,
allowed_dir=workspace,
extra_write_allowed_dirs=[writable],
)
result = await tool.execute(path=str(writable / "ok.txt"), content="allowed")
assert "Successfully wrote" in result
assert (writable / "ok.txt").read_text(encoding="utf-8") == "allowed"
@pytest.mark.asyncio
async def test_read_still_blocked_for_unrelated_dir(self, tmp_path):
@@ -398,7 +433,11 @@ class TestWorkspaceRestriction:
skill_file.parent.mkdir()
skill_file.write_text("# Weather\nOriginal content.")
tool = EditFileTool(workspace=workspace, allowed_dir=workspace)
tool = EditFileTool(
workspace=workspace,
allowed_dir=workspace,
extra_allowed_dirs=[skills_dir],
)
result = await tool.execute(
path=str(skill_file),
old_text="Original content.",
@@ -407,3 +446,25 @@ class TestWorkspaceRestriction:
assert "Error" in result
assert "outside" in result.lower()
assert skill_file.read_text() == "# Weather\nOriginal content."
@pytest.mark.asyncio
async def test_edit_allowed_with_extra_write_dir(self, tmp_path):
workspace = tmp_path / "ws"
workspace.mkdir()
writable = tmp_path / "writable"
writable.mkdir()
target = writable / "note.txt"
target.write_text("before\n", encoding="utf-8")
tool = EditFileTool(
workspace=workspace,
allowed_dir=workspace,
extra_write_allowed_dirs=[writable],
)
result = await tool.execute(
path=str(target),
old_text="before",
new_text="after",
)
assert "Successfully edited" in result
assert target.read_text(encoding="utf-8") == "after\n"