From 69a055786a934ea24adfdaa7259c90d2f1b927e3 Mon Sep 17 00:00:00 2001 From: RerankerGuo Date: Sun, 8 Nov 2026 16:08:06 +0800 Subject: [PATCH] fix(file-io): scope delete, move and frontmatter_delete to allowed paths DeleteStep, MoveStep and FrontmatterDeleteStep mutated any workspace path without consulting the request-scoped _allowed_paths constraint, so a scope-limited agent session could still unlink or rmtree paths outside its scope, move files in or out of it, or drop frontmatter keys from other notes. Enforce the same fail-closed check the other file I/O steps already apply: delete checks the target, move checks both the source and the destination, and frontmatter_delete mirrors frontmatter_update. Without an injected constraint the steps keep their unrestricted behavior. --- reme/steps/file_io/delete.py | 9 ++- reme/steps/file_io/frontmatter_delete.py | 62 ++++++++------- reme/steps/file_io/move.py | 11 ++- tests/unit/test_path_permission.py | 98 ++++++++++++++++++++++++ 4 files changed, 151 insertions(+), 29 deletions(-) diff --git a/reme/steps/file_io/delete.py b/reme/steps/file_io/delete.py index c4eec0883..1edaea8cc 100644 --- a/reme/steps/file_io/delete.py +++ b/reme/steps/file_io/delete.py @@ -19,12 +19,17 @@ canonical "new target" — the agent decides per-reference whether to rewrite (via ``file_move`` to a merge target), edit the citing prose, or accept dangling. + +Permission: honors the request-scoped ``_allowed_paths`` constraint +injected by the server into the RuntimeContext; without it, restricting +read/edit/write alone would still let a scoped request delete any path +in the workspace, including whole folders. """ import shutil from pathlib import Path -from ._path import resolve_path +from ._path import _check_path_permission, resolve_path from ..base_step import BaseStep from ...components import R from ...utils.wikilink_handler import WikilinkHandler @@ -74,6 +79,8 @@ async def _delete(self, path: str) -> dict: target, err = resolve_path(workspace_dir, path) if err or target is None: return {"path": path, "error": err or "invalid path"} + if not _check_path_permission(workspace_dir, target, self.context.get("_allowed_paths")): + return {"path": path, "error": "no permission to delete this path"} if target.is_file(): inbound = await WikilinkHandler.find_inbound(self.file_store, target=path) target.unlink() diff --git a/reme/steps/file_io/frontmatter_delete.py b/reme/steps/file_io/frontmatter_delete.py index 5a1f1876c..3d0fae600 100644 --- a/reme/steps/file_io/frontmatter_delete.py +++ b/reme/steps/file_io/frontmatter_delete.py @@ -7,6 +7,11 @@ zero-side-effect read. ``path`` is a path relative to the workspace. + +Permission: honors the request-scoped ``_allowed_paths`` constraint +injected by the server into the RuntimeContext; without it, restricting +read/edit/write alone would still leave frontmatter of arbitrary workspace +Markdown files mutable. """ from pathlib import Path @@ -14,7 +19,7 @@ import frontmatter from ._file_io import get_path_lock -from ._path import display_path, gate_md, resolve_path +from ._path import _check_path_permission, display_path, gate_md, resolve_path from ..base_step import BaseStep from ...components import R @@ -39,32 +44,35 @@ async def execute(self): else: original_target = target target, is_md = gate_md(target) - lock = await get_path_lock(target) - async with lock: - if not target.is_file(): - payload = {"path": path, "error": f"{display_path(workspace_dir, target)} not found"} - elif not is_md: - payload = {"path": path, "error": "not markdown"} - elif not keys: - payload = {"path": path, "error": "keys is empty"} - else: - post = frontmatter.loads(target.read_text(encoding="utf-8")) - deleted: list[str] = [] - missing: list[str] = [] - for k in keys: - if k in post.metadata: - del post.metadata[k] - deleted.append(k) - else: - missing.append(k) - if deleted: - target.write_text(frontmatter.dumps(post), encoding="utf-8") - payload = { - "path": path, - "deleted": deleted, - "missing": missing, - "frontmatter": dict(post.metadata), - } + if not _check_path_permission(workspace_dir, target, self.context.get("_allowed_paths")): + payload = {"path": path, "error": "no permission to update this file"} + else: + lock = await get_path_lock(target) + async with lock: + if not target.is_file(): + payload = {"path": path, "error": f"{display_path(workspace_dir, target)} not found"} + elif not is_md: + payload = {"path": path, "error": "not markdown"} + elif not keys: + payload = {"path": path, "error": "keys is empty"} + else: + post = frontmatter.loads(target.read_text(encoding="utf-8")) + deleted: list[str] = [] + missing: list[str] = [] + for k in keys: + if k in post.metadata: + del post.metadata[k] + deleted.append(k) + else: + missing.append(k) + if deleted: + target.write_text(frontmatter.dumps(post), encoding="utf-8") + payload = { + "path": path, + "deleted": deleted, + "missing": missing, + "frontmatter": dict(post.metadata), + } if target != original_target: payload["resolved_path"] = display_path(workspace_dir, target) diff --git a/reme/steps/file_io/move.py b/reme/steps/file_io/move.py index e1bad6ba4..a0a0e119c 100644 --- a/reme/steps/file_io/move.py +++ b/reme/steps/file_io/move.py @@ -27,12 +27,16 @@ Opt out via ``retarget=False`` for the rare case where you intentionally want to leave references stale (e.g. moving aside before delete) — the original is still removed in that case (move semantics, not copy). + +Permission: honors the request-scoped ``_allowed_paths`` constraint +injected by the server into the RuntimeContext; both the source (which is +unlinked) and the destination (which is written) must be inside the scope. """ import shutil from pathlib import Path -from ._path import resolve_path +from ._path import _check_path_permission, resolve_path from ..base_step import BaseStep from ...components import R from ...utils.wikilink_handler import WikilinkHandler @@ -78,6 +82,11 @@ async def _move(self, src_path: str, dst_path: str, overwrite: bool, retarget: b if precheck_error: return precheck_error assert src_abs is not None and dst_abs is not None # narrowed by precheck + allowed_paths = self.context.get("_allowed_paths") + if not _check_path_permission(workspace_dir, src_abs, allowed_paths): + return {"src_path": src_path, "dst_path": dst_path, "error": "no permission to move this file"} + if not _check_path_permission(workspace_dir, dst_abs, allowed_paths): + return {"src_path": src_path, "dst_path": dst_path, "error": "no permission to move this file"} dst_abs.parent.mkdir(parents=True, exist_ok=True) # Step 1 — copy. Wikilinks use workspace-relative paths, so outgoing diff --git a/tests/unit/test_path_permission.py b/tests/unit/test_path_permission.py index 4e22823a5..1b4a05862 100644 --- a/tests/unit/test_path_permission.py +++ b/tests/unit/test_path_permission.py @@ -7,6 +7,7 @@ Covers: - ReadStep / EditStep / WriteStep / FrontmatterUpdateStep honoring the scope. +- DeleteStep / MoveStep / FrontmatterDeleteStep honoring the scope. - Exact-file scope (least privilege) vs directory scope, custom daily dirs. - Fail-closed behavior for invalid injected constraints. - Wrapper-level injection: merge, conflict rejection, schema hiding @@ -25,8 +26,11 @@ from reme.components.file_store import LocalFileStore from reme.steps.file_io._path import _check_path_permission +from reme.steps.file_io.delete import DeleteStep from reme.steps.file_io.edit import EditStep +from reme.steps.file_io.frontmatter_delete import FrontmatterDeleteStep from reme.steps.file_io.frontmatter_update import FrontmatterUpdateStep +from reme.steps.file_io.move import MoveStep from reme.steps.file_io.read import ReadStep from reme.steps.file_io.write import WriteStep @@ -124,6 +128,100 @@ async def test_update_scope_rejects_paths_outside_daily_dir(): await store.close() +@pytest.mark.asyncio +async def test_delete_scope_allows_exact_note_and_rejects_outside(): + """DeleteStep honors the scope: the exact note is removable, everything else is not.""" + with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): + _seed(Path(tmp), NOTE, "target\n") + outside = _seed(Path(tmp), OUTSIDE, "roadmap\n") + store = await _make_store() + scope = {"_allowed_paths": [NOTE]} + + resp = await _run(DeleteStep, store, path=OUTSIDE, **scope) + assert resp.success is False + assert "no permission" in str(resp.answer).lower() + assert outside.read_text(encoding="utf-8") == "roadmap\n" + + resp = await _run(DeleteStep, store, path=NOTE, **scope) + assert resp.success is True + assert not (Path(tmp) / NOTE).exists() + await store.close() + + +@pytest.mark.asyncio +async def test_delete_directory_scope_rejects_ancestor_and_sibling_folders(): + """Directory scope deletes the scoped folder itself but not ancestors or siblings.""" + with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): + workspace = Path(tmp) + _seed(workspace, "daily/2025-06-01/note.md", "n\n") + _seed(workspace, "topics/x.md", "x\n") + store = await _make_store() + scope = {"_allowed_paths": ["daily/2025-06-01"]} + + resp = await _run(DeleteStep, store, path="daily", **scope) + assert resp.success is False + assert "no permission" in str(resp.answer).lower() + assert (workspace / "daily/2025-06-01/note.md").exists() + + resp = await _run(DeleteStep, store, path="topics", **scope) + assert resp.success is False + assert (workspace / "topics/x.md").exists() + + resp = await _run(DeleteStep, store, path="daily/2025-06-01", **scope) + assert resp.success is True + assert not (workspace / "daily/2025-06-01").exists() + await store.close() + + +@pytest.mark.asyncio +async def test_move_scope_requires_both_endpoints_in_scope(): + """MoveStep denies any move whose source or destination is outside the scope.""" + with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): + workspace = Path(tmp) + _seed(workspace, NOTE, "target\n") + _seed(workspace, SIBLING, "sibling\n") + outside = _seed(workspace, OUTSIDE, "roadmap\n") + store = await _make_store() + scope = {"_allowed_paths": ["daily/2025-06-01"]} + + resp = await _run(MoveStep, store, src_path=NOTE, dst_path=SIBLING, overwrite=True, **scope) + assert resp.success is True + assert not (workspace / NOTE).exists() + assert (workspace / SIBLING).exists() + + resp = await _run(MoveStep, store, src_path=SIBLING, dst_path=OUTSIDE, overwrite=True, **scope) + assert resp.success is False + assert "no permission" in str(resp.answer).lower() + assert outside.read_text(encoding="utf-8") == "roadmap\n" + assert (workspace / SIBLING).exists() + + resp = await _run(MoveStep, store, src_path=OUTSIDE, dst_path=SIBLING, overwrite=True, **scope) + assert resp.success is False + assert "no permission" in str(resp.answer).lower() + assert outside.exists() + assert (workspace / SIBLING).read_text(encoding="utf-8") == "target\n" + await store.close() + + +@pytest.mark.asyncio +async def test_frontmatter_delete_scope_rejects_outside_note(): + """Exact-file scope denies removing frontmatter keys from another note.""" + with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): + _seed(Path(tmp), NOTE) + sibling = _seed(Path(tmp), SIBLING, "---\nname: s\nstatus: draft\n---\nkeep\n") + store = await _make_store() + scope = {"_allowed_paths": [NOTE]} + + resp = await _run(FrontmatterDeleteStep, store, path=SIBLING, keys=["status"], **scope) + assert resp.success is False + assert "no permission" in str(resp.answer).lower() + assert "status: draft" in sibling.read_text(encoding="utf-8") + + resp = await _run(FrontmatterDeleteStep, store, path=NOTE, keys=["status"], **scope) + assert resp.success is True + await store.close() + + @pytest.mark.asyncio async def test_scope_follows_markdown_suffix_gating(): """A model path without ``.md`` gates to the same file and stays in scope."""