From 15f218e918632cbf262fd3e51349c6d472e62293 Mon Sep 17 00:00:00 2001 From: chengyongru Date: Wed, 10 Jun 2026 16:41:35 +0800 Subject: [PATCH] fix: enforce exact Dream memory file writes maintainer edit: Dream write tools used file paths as directory roots, so a missing canonical memory file could be treated as a parent directory. Add exact-file allowlist support and keep skills/ as the only Dream write directory. --- .agent/security.md | 2 +- nanobot/agent/memory.py | 10 +++--- nanobot/agent/tools/filesystem.py | 8 +++++ nanobot/agent/tools/path_utils.py | 2 ++ nanobot/security/workspace_policy.py | 30 ++++++++++++++--- tests/agent/test_dream.py | 44 ++++++++++++++++++++++--- tests/security/test_workspace_policy.py | 24 ++++++++++++++ tests/tools/test_filesystem_tools.py | 24 ++++++++++++++ 8 files changed, 129 insertions(+), 15 deletions(-) diff --git a/.agent/security.md b/.agent/security.md index 3276c199..ca961266 100644 --- a/.agent/security.md +++ b/.agent/security.md @@ -6,7 +6,7 @@ The agent operates with significant power (file system, shell, web). The followi Filesystem tools (`read_file`, `write_file`, `edit_file`, `list_dir`, `apply_patch`) resolve paths through the workspace path resolver (`agent/tools/filesystem.py` / `agent/tools/path_utils.py`), which enforces that the resolved path must lie under the active workspace when workspace restriction is enabled. The media upload directory is always an internal extra read root while restricted. -Additional filesystem roots must be capability-specific. `extra_allowed_dirs` is a legacy read-only alias. Use `extra_read_allowed_dirs` for read-only roots and `extra_write_allowed_dirs` only when a write-capable tool is intentionally allowed to modify an extra root. +Additional filesystem roots must be capability-specific. `extra_allowed_dirs` is a legacy read-only alias. Use `extra_read_allowed_dirs` for read-only roots, `extra_write_allowed_dirs` only when a write-capable tool is intentionally allowed to modify an extra directory, and exact file allowlists when a tool may modify only specific files. Shell execution (`ExecTool`, `agent/tools/shell.py`) also respects `restrict_to_workspace` as an application-level guard: if enabled and `working_dir` is outside the workspace, the command is rejected before execution, and command text is checked for obvious workspace escapes. This is not process-level isolation; use an exec sandbox backend for that. diff --git a/nanobot/agent/memory.py b/nanobot/agent/memory.py index 3bcefa0c..e0d1f02a 100644 --- a/nanobot/agent/memory.py +++ b/nanobot/agent/memory.py @@ -504,7 +504,7 @@ class MemoryStore: skills_dir.mkdir(parents=True, exist_ok=True) extra_read = [BUILTIN_SKILLS_DIR] if BUILTIN_SKILLS_DIR.exists() else None - editable_roots = [self.soul_file, self.user_file, skills_dir] + editable_files = [self.memory_file, self.soul_file, self.user_file] tools.register(ReadFileTool( workspace=workspace, @@ -514,14 +514,14 @@ class MemoryStore: )) tools.register(EditFileTool( workspace=workspace, - allowed_dir=self.memory_file, - extra_write_allowed_dirs=editable_roots, + allowed_dir=skills_dir, + extra_write_allowed_files=editable_files, file_states=file_states, )) tools.register(ApplyPatchTool( workspace=workspace, - allowed_dir=self.memory_file, - extra_write_allowed_dirs=editable_roots, + allowed_dir=skills_dir, + extra_write_allowed_files=editable_files, file_states=file_states, )) tools.register(WriteFileTool( diff --git a/nanobot/agent/tools/filesystem.py b/nanobot/agent/tools/filesystem.py index 3c0df38f..b6105c08 100644 --- a/nanobot/agent/tools/filesystem.py +++ b/nanobot/agent/tools/filesystem.py @@ -47,6 +47,8 @@ class _FsTool(Tool): extra_allowed_dirs: list[Path] | None = None, extra_read_allowed_dirs: list[Path] | None = None, extra_write_allowed_dirs: list[Path] | None = None, + extra_read_allowed_files: list[Path] | None = None, + extra_write_allowed_files: list[Path] | None = None, file_states: FileStates | None = None, restrict_to_workspace: bool | None = None, sandbox_restricts_workspace: bool = False, @@ -60,6 +62,8 @@ class _FsTool(Tool): *(extra_read_allowed_dirs or []), ] self._extra_write_allowed_dirs = list(extra_write_allowed_dirs or []) + self._extra_read_allowed_files = list(extra_read_allowed_files or []) + self._extra_write_allowed_files = list(extra_write_allowed_files or []) self._extra_allowed_dirs = self._extra_read_allowed_dirs self._restrict_to_workspace = ( bool(restrict_to_workspace) @@ -117,6 +121,7 @@ class _FsTool(Tool): self, path: str, extra_allowed_dirs: list[Path] | None, + extra_allowed_files: list[Path] | None, *, include_media_dir: bool, ) -> Path: @@ -130,6 +135,7 @@ class _FsTool(Tool): access.project_path, self._effective_allowed_root(access.allowed_root), extra_allowed_dirs, + extra_allowed_files, include_media_dir=include_media_dir, ) @@ -137,6 +143,7 @@ class _FsTool(Tool): return self._resolve_with_extra( path, self._extra_read_allowed_dirs, + self._extra_read_allowed_files, include_media_dir=True, ) @@ -144,6 +151,7 @@ class _FsTool(Tool): return self._resolve_with_extra( path, self._extra_write_allowed_dirs, + self._extra_write_allowed_files, include_media_dir=False, ) diff --git a/nanobot/agent/tools/path_utils.py b/nanobot/agent/tools/path_utils.py index 757ae0b9..ca3f10e7 100644 --- a/nanobot/agent/tools/path_utils.py +++ b/nanobot/agent/tools/path_utils.py @@ -19,6 +19,7 @@ def resolve_workspace_path( workspace: Path | None = None, allowed_dir: Path | None = None, extra_allowed_dirs: list[Path] | None = None, + extra_allowed_files: list[Path] | None = None, include_media_dir: bool = True, ) -> Path: """Resolve path against workspace and enforce allowed directory containment.""" @@ -29,4 +30,5 @@ def resolve_workspace_path( workspace=workspace, allowed_root=allowed_dir, extra_allowed_roots=extra_roots, + extra_allowed_files=extra_allowed_files, ) diff --git a/nanobot/security/workspace_policy.py b/nanobot/security/workspace_policy.py index 31ebde80..e9706294 100644 --- a/nanobot/security/workspace_policy.py +++ b/nanobot/security/workspace_policy.py @@ -44,6 +44,22 @@ def is_path_allowed(path: str | Path, roots: Iterable[str | Path]) -> bool: return any(is_path_within(path, root) for root in roots) +def is_path_exactly_allowed(path: str | Path, files: Iterable[str | Path]) -> bool: + """Return True when *path* resolves exactly to one of the allowed files.""" + try: + resolved_path = Path(path).expanduser().resolve(strict=False) + except (OSError, RuntimeError, TypeError, ValueError): + return False + for file in files: + try: + resolved_file = Path(file).expanduser().resolve(strict=False) + except (OSError, RuntimeError, TypeError, ValueError): + continue + if resolved_path == resolved_file: + return True + return False + + def require_path_within( path: str | Path, root: str | Path, @@ -67,17 +83,23 @@ def resolve_allowed_path( workspace: str | Path | None = None, allowed_root: str | Path | None = None, extra_allowed_roots: Iterable[str | Path] | None = None, + extra_allowed_files: Iterable[str | Path] | None = None, strict: bool = False, ) -> Path: """Resolve a path and enforce containment in allowed roots when configured.""" resolved = resolve_path(path, workspace, strict=False) - if allowed_root is None: + files = list(extra_allowed_files or []) + if allowed_root is None and not files: return resolve_path(path, workspace, strict=strict) if strict else resolved - roots = [allowed_root, *(extra_allowed_roots or [])] - if not is_path_allowed(resolved, roots): + roots = [] + if allowed_root is not None: + roots.append(allowed_root) + roots.extend(extra_allowed_roots or []) + if not is_path_allowed(resolved, roots) and not is_path_exactly_allowed(resolved, files): + boundary = Path(allowed_root).expanduser() if allowed_root is not None else "allowed files" raise WorkspaceBoundaryError( - f"Path {path} is outside allowed directory {Path(allowed_root).expanduser()}" + f"Path {path} is outside allowed directory {boundary}" + WORKSPACE_BOUNDARY_NOTE ) if strict: diff --git a/tests/agent/test_dream.py b/tests/agent/test_dream.py index a59d6307..0f0b0c8a 100644 --- a/tests/agent/test_dream.py +++ b/tests/agent/test_dream.py @@ -206,6 +206,38 @@ class TestDreamTools: assert store.history_file.read_text(encoding="utf-8") == "before\n" assert store._dream_cursor_file.read_text(encoding="utf-8") == "1" + @pytest.mark.asyncio + async def test_dream_cannot_create_children_under_canonical_files(self, store): + tools = store.build_dream_tools() + + memory_child = store.memory_file / "evil.txt" + user_child = store.user_file / "evil.txt" + memory_result = await tools.execute( + "apply_patch", + { + "edits": [ + { + "path": "memory/MEMORY.md/evil.txt", + "action": "add", + "new_text": "owned", + } + ] + }, + ) + user_result = await tools.execute( + "edit_file", + { + "path": "USER.md/evil.txt", + "old_text": "", + "new_text": "owned", + }, + ) + + assert "outside allowed directory" in memory_result + assert "outside allowed directory" in user_result + assert not memory_child.exists() + assert not user_child.exists() + class TestEphemeralDirect: """Tests for the ephemeral flag that skips history.jsonl writes for Dream.""" @@ -229,9 +261,7 @@ class TestEphemeralDirect: provider.supports_tools = True provider.generation = MagicMock(max_tokens=4096) provider.chat_with_retry = AsyncMock( - return_value=MagicMock( - content="done", finish_reason="stop", tool_calls=[], usage={}, - ) + return_value=LLMResponse(content="done", tool_calls=[], finish_reason="stop", usage={}) ) with ( @@ -263,9 +293,13 @@ class TestEphemeralDirect: mock_archive.assert_not_called() async def test_non_ephemeral_runs_normally(self, tmp_path, _make_loop): - """Without ephemeral, the normal path is untouched — no crash.""" + """Without ephemeral, the normal path returns the model response.""" loop, store = _make_loop - await loop.process_direct("test", session_key="cli:normal") + 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() async def test_ephemeral_sets_ctx_flag(self, tmp_path, _make_loop): """Verify that ephemeral=True is forwarded to TurnContext.""" diff --git a/tests/security/test_workspace_policy.py b/tests/security/test_workspace_policy.py index 0ed89dcc..ddbfe0ea 100644 --- a/tests/security/test_workspace_policy.py +++ b/tests/security/test_workspace_policy.py @@ -67,3 +67,27 @@ def test_resolve_allowed_path_allows_extra_root(tmp_path: Path) -> None: ) assert resolved == image.resolve() + + +def test_resolve_allowed_path_allows_extra_file_only_exactly(tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + allowed = outside / "allowed.txt" + + resolved = resolve_allowed_path( + allowed, + workspace=workspace, + allowed_root=workspace, + extra_allowed_files=[allowed], + ) + + assert resolved == allowed.resolve() + with pytest.raises(WorkspaceBoundaryError, match="outside allowed directory"): + resolve_allowed_path( + allowed / "child.txt", + workspace=workspace, + allowed_root=workspace, + extra_allowed_files=[allowed], + ) diff --git a/tests/tools/test_filesystem_tools.py b/tests/tools/test_filesystem_tools.py index 3ca06c35..2df98776 100644 --- a/tests/tools/test_filesystem_tools.py +++ b/tests/tools/test_filesystem_tools.py @@ -385,6 +385,30 @@ class TestWorkspaceRestriction: assert "Successfully wrote" in result assert (writable / "ok.txt").read_text(encoding="utf-8") == "allowed" + @pytest.mark.asyncio + async def test_extra_write_allowed_files_allow_only_exact_file(self, tmp_path): + workspace = tmp_path / "ws" + workspace.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + allowed_file = outside / "allowed.txt" + child_path = allowed_file / "child.txt" + + tool = WriteFileTool( + workspace=workspace, + allowed_dir=workspace, + extra_write_allowed_files=[allowed_file], + ) + + exact = await tool.execute(path=str(allowed_file), content="allowed") + child = await tool.execute(path=str(child_path), content="blocked") + + assert "Successfully wrote" in exact + assert allowed_file.read_text(encoding="utf-8") == "allowed" + assert "Error" in child + assert "outside" in child.lower() + assert not child_path.exists() + @pytest.mark.asyncio async def test_read_still_blocked_for_unrelated_dir(self, tmp_path): workspace = tmp_path / "ws"