Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion reme/steps/file_io/delete.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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()
Expand Down
62 changes: 35 additions & 27 deletions reme/steps/file_io/frontmatter_delete.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,19 @@
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

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

Expand All @@ -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)

Expand Down
11 changes: 10 additions & 1 deletion reme/steps/file_io/move.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
98 changes: 98 additions & 0 deletions tests/unit/test_path_permission.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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

Expand Down Expand Up @@ -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."""
Expand Down
Loading