mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-08-28 05:25:04 +00:00
feat(Step tools): add white/black path prefix permission filtering to read, edit, write (#391)
* feat(read): add white/black path prefix permission filtering to ReadStep * feat: add PrefixCheck mixin for path-prefix permission in file I/O steps * feat: add injected_job_kwargs mechanism and refine path-prefix permission * refactor(file_io): consolidate prefix_check into _path module
This commit is contained in:
xyf2020
GitHub
parent
2f79977df0
commit
f34dcdb09b
17 changed files with 741 additions and 40 deletions
|
|
@ -156,18 +156,24 @@ class AsAgentWrapper(BaseAgentWrapper):
|
|||
self._session_cleanup_done = False
|
||||
|
||||
@classmethod
|
||||
def _make_tool(cls, job: "BaseJob", tool_context_id: str | None = None) -> FunctionTool:
|
||||
def _make_tool(
|
||||
cls,
|
||||
job: "BaseJob",
|
||||
tool_context_id: str | None = None,
|
||||
injected_job_kwargs: dict[str, Any] | None = None,
|
||||
) -> FunctionTool:
|
||||
injected = cls._resolve_injected_job_kwargs(
|
||||
{"tool_context_id": tool_context_id, "injected_job_kwargs": injected_job_kwargs},
|
||||
)
|
||||
|
||||
async def run_job(**kwargs) -> ToolChunk:
|
||||
if tool_context_id:
|
||||
assert "tool_context_id" not in kwargs, "tool_context_id is injected by agent_wrapper"
|
||||
kwargs["tool_context_id"] = tool_context_id
|
||||
response = await job(**kwargs)
|
||||
response = await job(**cls._merge_injected_job_kwargs(kwargs, injected))
|
||||
state = ToolResultState.SUCCESS if response.success else ToolResultState.ERROR
|
||||
return ToolChunk(content=[TextBlock(text=str(response.answer))], state=state)
|
||||
|
||||
tool = FunctionTool(func=run_job, name=job.name, description=job.description, is_concurrency_safe=False)
|
||||
if job.parameters:
|
||||
tool.input_schema = job.parameters
|
||||
if parameters := cls._strip_injected_parameters(job.parameters, injected):
|
||||
tool.input_schema = parameters
|
||||
return tool
|
||||
|
||||
def _builtin_tools(
|
||||
|
|
@ -308,7 +314,7 @@ class AsAgentWrapper(BaseAgentWrapper):
|
|||
builtin_tools = []
|
||||
tools: list[ToolBase] = []
|
||||
tools.extend(self._builtin_tools(builtin_tools, sequential_tool_calls=sequential_tool_calls))
|
||||
tools.extend(self._make_tool(job, tool_context_id) for job in resolved_jobs)
|
||||
tools.extend(self._make_tool(job, tool_context_id, kwargs.get("injected_job_kwargs")) for job in resolved_jobs)
|
||||
toolkit = kwargs.get("toolkit") or Toolkit(
|
||||
tools=tools,
|
||||
skills_or_loaders=skills,
|
||||
|
|
|
|||
|
|
@ -155,6 +155,46 @@ class BaseAgentWrapper(BaseComponent):
|
|||
return schema
|
||||
raise TypeError("output_schema must be a JSON schema dict or BaseModel class")
|
||||
|
||||
@staticmethod
|
||||
def _resolve_injected_job_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Collect server-owned kwargs merged into every job tool call.
|
||||
|
||||
``injected_job_kwargs`` values are enforced constraints added by the
|
||||
wrapper after receiving the model's tool arguments; the model can
|
||||
neither see nor override them. ``tool_context_id`` keeps its dedicated
|
||||
option but is carried through the same mechanism.
|
||||
"""
|
||||
injected = dict(kwargs.get("injected_job_kwargs") or {})
|
||||
if tool_context_id := kwargs.get("tool_context_id"):
|
||||
injected["tool_context_id"] = tool_context_id
|
||||
return injected
|
||||
|
||||
@staticmethod
|
||||
def _merge_injected_job_kwargs(model_kwargs: dict[str, Any], injected: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Merge server-owned kwargs over model tool arguments, rejecting conflicts.
|
||||
|
||||
Silently letting model values win would make injected constraints
|
||||
bypassable, so any overlap is an explicit error.
|
||||
"""
|
||||
if conflicts := sorted(injected.keys() & model_kwargs.keys()):
|
||||
names = ", ".join(conflicts)
|
||||
raise ValueError(f"injected tool arguments cannot be provided by the model: {names}")
|
||||
return {**model_kwargs, **injected}
|
||||
|
||||
@staticmethod
|
||||
def _strip_injected_parameters(parameters: dict | None, injected: dict[str, Any]) -> dict | None:
|
||||
"""Hide injected keys from the tool parameter schema exposed to the model."""
|
||||
if not parameters or not injected:
|
||||
return parameters
|
||||
parameters = dict(parameters)
|
||||
if "properties" in parameters:
|
||||
parameters["properties"] = {
|
||||
name: schema for name, schema in parameters["properties"].items() if name not in injected
|
||||
}
|
||||
if "required" in parameters:
|
||||
parameters["required"] = [name for name in parameters["required"] if name not in injected]
|
||||
return parameters
|
||||
|
||||
def _resolve_job_tools(self, job_tools: list[str]) -> list["BaseJob"]:
|
||||
"""Resolve job name strings to BaseJob instances via app_context."""
|
||||
if not job_tools:
|
||||
|
|
|
|||
|
|
@ -73,16 +73,20 @@ class CcAgentWrapper(BaseAgentWrapper):
|
|||
self.logger.warning(f"Failed to link Claude Code skills into {target}: {exc}")
|
||||
|
||||
@classmethod
|
||||
def _make_tool(cls, job: "BaseJob", tool_context_id: str | None = None):
|
||||
def _make_tool(
|
||||
cls,
|
||||
job: "BaseJob",
|
||||
tool_context_id: str | None = None,
|
||||
injected_job_kwargs: dict[str, Any] | None = None,
|
||||
):
|
||||
from claude_agent_sdk import SdkMcpTool
|
||||
|
||||
injected = cls._resolve_injected_job_kwargs(
|
||||
{"tool_context_id": tool_context_id, "injected_job_kwargs": injected_job_kwargs},
|
||||
)
|
||||
|
||||
async def run_job(args):
|
||||
call_args = dict(args)
|
||||
if tool_context_id:
|
||||
if "tool_context_id" in call_args:
|
||||
raise ValueError("tool_context_id is injected by agent_wrapper")
|
||||
call_args["tool_context_id"] = tool_context_id
|
||||
response = await job(**call_args)
|
||||
response = await job(**cls._merge_injected_job_kwargs(dict(args), injected))
|
||||
return {
|
||||
"content": [{"type": "text", "text": str(response.answer)}],
|
||||
"is_error": not response.success,
|
||||
|
|
@ -91,7 +95,7 @@ class CcAgentWrapper(BaseAgentWrapper):
|
|||
return SdkMcpTool(
|
||||
name=job.name,
|
||||
description=job.description,
|
||||
input_schema=job.parameters,
|
||||
input_schema=cls._strip_injected_parameters(job.parameters, injected),
|
||||
handler=run_job,
|
||||
)
|
||||
|
||||
|
|
@ -142,7 +146,7 @@ class CcAgentWrapper(BaseAgentWrapper):
|
|||
|
||||
if "setting_sources" not in kwargs and kwargs.get("skills") is None:
|
||||
kwargs["setting_sources"] = []
|
||||
skip_keys = {"job_tools", "output_schema", "api_key", "base_url", "credential"}
|
||||
skip_keys = {"job_tools", "injected_job_kwargs", "output_schema", "api_key", "base_url", "credential"}
|
||||
option_fields = {field.name for field in fields(ClaudeAgentOptions)}
|
||||
option_kwargs = {key: value for key, value in kwargs.items() if key not in skip_keys and key in option_fields}
|
||||
option_kwargs["disallowed_tools"] = list(
|
||||
|
|
@ -193,7 +197,10 @@ class CcAgentWrapper(BaseAgentWrapper):
|
|||
opts.mcp_servers = dict(opts.mcp_servers)
|
||||
if self.MCP_SERVER_NAME in opts.mcp_servers:
|
||||
raise ValueError(f"mcp_servers already contains reserved server name {self.MCP_SERVER_NAME!r}")
|
||||
sdk_tools = [self._make_tool(job, kwargs.get("tool_context_id")) for job in resolved_jobs]
|
||||
sdk_tools = [
|
||||
self._make_tool(job, kwargs.get("tool_context_id"), kwargs.get("injected_job_kwargs"))
|
||||
for job in resolved_jobs
|
||||
]
|
||||
opts.mcp_servers[self.MCP_SERVER_NAME] = create_sdk_mcp_server(
|
||||
name=self.MCP_SERVER_NAME,
|
||||
tools=sdk_tools,
|
||||
|
|
|
|||
|
|
@ -255,6 +255,8 @@ class CodexAgentWrapper(BaseAgentWrapper):
|
|||
for name in job_names:
|
||||
args.extend(["--job", name])
|
||||
args.extend(["--tool-context-id", str(kwargs.get("tool_context_id") or "")])
|
||||
if injected_job_kwargs := dict(kwargs.get("injected_job_kwargs") or {}):
|
||||
args.extend(["--injected-job-kwargs", json.dumps(injected_job_kwargs, sort_keys=True)])
|
||||
return {
|
||||
"command": sys.executable,
|
||||
"args": args,
|
||||
|
|
@ -298,10 +300,17 @@ class CodexAgentWrapper(BaseAgentWrapper):
|
|||
fork_session = bool(kwargs.get("fork_session", False))
|
||||
if fork_session and not thread_id:
|
||||
raise ValueError("fork_session=True requires resume or session_id")
|
||||
requested_tool_context = str(kwargs.get("tool_context_id") or "")
|
||||
requested_tool_context = json.dumps(
|
||||
{
|
||||
"tool_context_id": str(kwargs.get("tool_context_id") or ""),
|
||||
"injected_job_kwargs": dict(kwargs.get("injected_job_kwargs") or {}),
|
||||
},
|
||||
sort_keys=True,
|
||||
default=str,
|
||||
)
|
||||
if not fork_session and thread_id in self._thread_tool_contexts:
|
||||
if requested_tool_context != self._thread_tool_contexts[thread_id]:
|
||||
raise ValueError("tool_context_id cannot change when resuming a Codex thread")
|
||||
raise ValueError("tool_context_id and injected_job_kwargs cannot change when resuming a Codex thread")
|
||||
|
||||
common = {
|
||||
"approval_mode": self._enum(ApprovalMode, kwargs.get("approval_mode"), ApprovalMode.auto_review),
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
"""FastMCP STDIO bridge that exposes selected ReMe jobs to Codex."""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
|
@ -14,10 +15,20 @@ def _parse_args() -> argparse.Namespace:
|
|||
parser.add_argument("--workspace", required=True, help="ReMe workspace directory")
|
||||
parser.add_argument("--job", dest="jobs", action="append", required=True, help="ReMe job name; repeat as needed")
|
||||
parser.add_argument("--tool-context-id", default="", help="Context id injected into every job call")
|
||||
parser.add_argument(
|
||||
"--injected-job-kwargs",
|
||||
default="",
|
||||
help="JSON object of server-owned kwargs injected into every job call",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def _prepare_config(config: dict[str, Any], job_names: list[str], tool_context_id: str = "") -> dict[str, Any]:
|
||||
def _prepare_config(
|
||||
config: dict[str, Any],
|
||||
job_names: list[str],
|
||||
tool_context_id: str = "",
|
||||
injected_job_kwargs: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Configure the dedicated child Application to serve selected jobs over MCP STDIO."""
|
||||
selected = set(job_names)
|
||||
jobs: dict[str, dict[str, Any]] = {}
|
||||
|
|
@ -40,8 +51,11 @@ def _prepare_config(config: dict[str, Any], job_names: list[str], tool_context_i
|
|||
"jobs": job_names,
|
||||
"tool_error_on_failure": True,
|
||||
}
|
||||
injected = dict(injected_job_kwargs or {})
|
||||
if tool_context_id:
|
||||
service["injected_job_kwargs"] = {"tool_context_id": tool_context_id}
|
||||
injected["tool_context_id"] = tool_context_id
|
||||
if injected:
|
||||
service["injected_job_kwargs"] = injected
|
||||
|
||||
prepared = dict(config)
|
||||
prepared["jobs"] = jobs
|
||||
|
|
@ -60,7 +74,10 @@ def main() -> None:
|
|||
log_to_file=False,
|
||||
log_config=False,
|
||||
)
|
||||
config = _prepare_config(config, args.jobs, args.tool_context_id)
|
||||
injected_job_kwargs = json.loads(args.injected_job_kwargs) if args.injected_job_kwargs else None
|
||||
if injected_job_kwargs is not None and not isinstance(injected_job_kwargs, dict):
|
||||
raise TypeError("--injected-job-kwargs must be a JSON object")
|
||||
config = _prepare_config(config, args.jobs, args.tool_context_id, injected_job_kwargs)
|
||||
ReMe(**config).run_app()
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -117,7 +117,11 @@ def _interpolate_timestamps(items: list[dict]) -> list[dict]:
|
|||
|
||||
@R.register("beam_auto_memory_step")
|
||||
class BeamAutoMemoryStep(AutoMemoryStep):
|
||||
"""AutoMemoryStep variant that interpolates timestamps and pins daily_write to the resolved day."""
|
||||
"""AutoMemoryStep variant that interpolates timestamps for BEAM sessions.
|
||||
|
||||
Date pinning for ``daily_write`` is handled by the base class through the
|
||||
agent wrapper's server-owned ``injected_job_kwargs``.
|
||||
"""
|
||||
|
||||
def _build_messages(self, raw_messages: list) -> list[Msg]:
|
||||
# Interpolate timestamps: if any message carries created_at, fill in
|
||||
|
|
@ -127,6 +131,3 @@ class BeamAutoMemoryStep(AutoMemoryStep):
|
|||
[item if not isinstance(item, dict) else dict(item) for item in raw_messages],
|
||||
)
|
||||
return [self._to_msg(item) for item in interpolated]
|
||||
|
||||
def _reply_extra_kwargs(self, day: str) -> dict:
|
||||
return {"tool_defaults": {"daily_write": {"date": day}}}
|
||||
|
|
|
|||
|
|
@ -117,7 +117,11 @@ def _interpolate_timestamps(items: list[dict]) -> list[dict]:
|
|||
|
||||
@R.register("lme_auto_memory_step")
|
||||
class LmeAutoMemoryStep(AutoMemoryStep):
|
||||
"""AutoMemoryStep variant that interpolates timestamps and pins daily_write to the resolved day."""
|
||||
"""AutoMemoryStep variant that interpolates timestamps for LongMemEval sessions.
|
||||
|
||||
Date pinning for ``daily_write`` is handled by the base class through the
|
||||
agent wrapper's server-owned ``injected_job_kwargs``.
|
||||
"""
|
||||
|
||||
def _build_messages(self, raw_messages: list) -> list[Msg]:
|
||||
# Interpolate timestamps: if any message carries created_at, fill in
|
||||
|
|
@ -127,6 +131,3 @@ class LmeAutoMemoryStep(AutoMemoryStep):
|
|||
[item if not isinstance(item, dict) else dict(item) for item in raw_messages],
|
||||
)
|
||||
return [self._to_msg(item) for item in interpolated]
|
||||
|
||||
def _reply_extra_kwargs(self, day: str) -> dict:
|
||||
return {"tool_defaults": {"daily_write": {"date": day}}}
|
||||
|
|
|
|||
|
|
@ -312,11 +312,17 @@ class AutoMemoryStep(BaseStep):
|
|||
)
|
||||
|
||||
self.logger.info(f"[{self.name}] agent start path={note_path} template={template_key}")
|
||||
# Existing-note updates are restricted to the resolved note path. New
|
||||
# notes retain the upstream ``daily_write`` date behavior, where the
|
||||
# model supplies the date from the prompt.
|
||||
reply_kwargs = self._reply_extra_kwargs(day)
|
||||
if not created:
|
||||
reply_kwargs["injected_job_kwargs"] = {"_allowed_paths": [note_path]}
|
||||
result = await self.agent_wrapper.reply(
|
||||
user_message,
|
||||
system_prompt=self.prompt_format("system_prompt"),
|
||||
job_tools=self.create_tools if created else self.update_tools,
|
||||
**self._reply_extra_kwargs(day),
|
||||
**reply_kwargs,
|
||||
)
|
||||
self.logger.info(f"[{self.name}] agent done path={note_path} has_result={bool(result.get('result'))}")
|
||||
|
||||
|
|
|
|||
|
|
@ -82,10 +82,17 @@ def resolve_path(
|
|||
return workspace_path.resolve(), None
|
||||
return None, "`path` is required"
|
||||
s = str(raw).strip()
|
||||
# ``~`` paths refer to the user's home directory rather than the workspace.
|
||||
# They are deliberately unsupported: do not expand them (which could make a
|
||||
# home-relative path look like a workspace escape), and report them as a
|
||||
# missing target to callers.
|
||||
if s.startswith("~"):
|
||||
return None, f"file {s!r} does not exist"
|
||||
p = Path(s)
|
||||
workspace = workspace_path.resolve()
|
||||
if p.is_absolute():
|
||||
logger.info("absolute path detected, recommending relative paths")
|
||||
if Path(s).is_absolute():
|
||||
logger.info("absolute path detected, recommending relative paths")
|
||||
target = p.resolve()
|
||||
else:
|
||||
for part in p.parts:
|
||||
|
|
@ -98,6 +105,48 @@ def resolve_path(
|
|||
return target, None
|
||||
|
||||
|
||||
def _check_path_permission(workspace_path: Path, target: Path, allowed_paths) -> bool:
|
||||
"""Return whether ``target`` is covered by the optional allowed-path scope.
|
||||
|
||||
``None`` leaves access unrestricted. Existing files allow that exact path;
|
||||
existing directories allow their descendants. Missing entries are treated
|
||||
as directory-like path prefixes, so a scoped ``write`` may create the path
|
||||
itself or a descendant. Containment is path-component based, never a string
|
||||
prefix. Entries beginning with ``~`` are ignored. Invalid constraints and
|
||||
workspace-escaping entries fail closed.
|
||||
"""
|
||||
if allowed_paths is None:
|
||||
return True
|
||||
if isinstance(allowed_paths, (str, Path)):
|
||||
allowed_paths = [allowed_paths]
|
||||
if not isinstance(allowed_paths, (list, tuple)) or not allowed_paths:
|
||||
logger.warning("invalid _allowed_paths constraint; denying access (fail closed)")
|
||||
return False
|
||||
|
||||
allowed_files: list[Path] = []
|
||||
allowed_dirs: list[Path] = []
|
||||
for raw in allowed_paths:
|
||||
raw_string = str(raw).strip()
|
||||
if raw_string.startswith("~"):
|
||||
logger.warning(f"home-relative _allowed_paths entry {raw_string!r}; skipping")
|
||||
continue
|
||||
resolved, err = resolve_path(workspace_path, raw_string)
|
||||
if err or resolved is None:
|
||||
logger.warning(f"invalid _allowed_paths entry {raw_string!r} ({err}); denying access (fail closed)")
|
||||
return False
|
||||
if resolved.is_file():
|
||||
allowed_files.append(resolved)
|
||||
elif resolved.is_dir():
|
||||
allowed_dirs.append(resolved)
|
||||
else:
|
||||
allowed_dirs.append(resolved)
|
||||
|
||||
resolved_target = target.resolve()
|
||||
return resolved_target in allowed_files or any(
|
||||
is_relative_to(resolved_target, directory) for directory in allowed_dirs
|
||||
)
|
||||
|
||||
|
||||
def gate_md(target: Path) -> tuple[Path, bool]:
|
||||
"""Markdown gate with compatibility fallback.
|
||||
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import frontmatter
|
|||
import yaml
|
||||
|
||||
from ._file_io import get_path_lock, read_file_safe, write_file_safe
|
||||
from ._path import NON_MD_WARNING, gate_md, resolve_path
|
||||
from ._path import _check_path_permission, NON_MD_WARNING, gate_md, resolve_path
|
||||
from ..base_step import BaseStep
|
||||
from ...components import R
|
||||
|
||||
|
|
@ -17,6 +17,9 @@ class EditStep(BaseStep):
|
|||
re-emitted unchanged — matches that fall inside front matter are ignored,
|
||||
so a typo in `old` cannot corrupt structured metadata.
|
||||
|
||||
Permission: honors the request-scoped ``_allowed_paths`` constraint
|
||||
injected by the server into the RuntimeContext.
|
||||
|
||||
Concurrency: in-process per-path ``asyncio.Lock`` serializes the
|
||||
read-modify-write cycle against the same file (multi-worker / multi-
|
||||
process safety is out of scope)."""
|
||||
|
|
@ -50,6 +53,10 @@ class EditStep(BaseStep):
|
|||
|
||||
target, is_md = gate_md(target)
|
||||
|
||||
if not _check_path_permission(self.workspace_path, target, self.context.get("_allowed_paths")):
|
||||
self._fail("no permission to edit this file", path=str(target))
|
||||
return None
|
||||
|
||||
lock = await get_path_lock(target)
|
||||
async with lock:
|
||||
if not target.exists():
|
||||
|
|
|
|||
|
|
@ -17,14 +17,20 @@ from pathlib import Path
|
|||
import frontmatter
|
||||
|
||||
from ._file_io import get_path_lock
|
||||
from ._path import resolve_path
|
||||
from ._path import _check_path_permission, resolve_path
|
||||
from ..base_step import BaseStep
|
||||
from ...components import R
|
||||
|
||||
|
||||
@R.register("frontmatter_update_step")
|
||||
class FrontmatterUpdateStep(BaseStep):
|
||||
"""Set frontmatter keys on a markdown file from a ``metadata`` dict."""
|
||||
"""Set frontmatter keys on a markdown file from a ``metadata`` dict.
|
||||
|
||||
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.
|
||||
"""
|
||||
|
||||
async def execute(self):
|
||||
assert self.context is not None
|
||||
|
|
@ -37,6 +43,8 @@ class FrontmatterUpdateStep(BaseStep):
|
|||
target, err = resolve_path(workspace_dir, path)
|
||||
if err or target is None:
|
||||
payload: dict = {"path": path, "error": err or "invalid path"}
|
||||
elif 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:
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@
|
|||
from pathlib import Path
|
||||
|
||||
from ._file_io import read_file_lines_safe, read_file_safe, truncate_text_output
|
||||
from ._path import NON_MD_WARNING, gate_md, resolve_path
|
||||
from ._path import _check_path_permission, NON_MD_WARNING, gate_md, resolve_path
|
||||
from ..base_step import BaseStep
|
||||
from ...components import R
|
||||
from ...constants import DEFAULT_MAX_BYTES, MAX_FILE_READ_BYTES
|
||||
|
|
@ -21,6 +21,9 @@ class ReadStep(BaseStep):
|
|||
neighbors (out/in link targets) with name/description meta,
|
||||
fetched via the file_store. Same rendering as SearchStep.
|
||||
max_neighbors_per_direction (int, default 10): cap per direction.
|
||||
|
||||
Permission: honors the request-scoped ``_allowed_paths`` constraint
|
||||
injected by the server into the RuntimeContext.
|
||||
"""
|
||||
|
||||
def _fail(self, message: str, **meta) -> None:
|
||||
|
|
@ -101,6 +104,9 @@ class ReadStep(BaseStep):
|
|||
target = self._resolve_target(raw)
|
||||
if target is None:
|
||||
return None
|
||||
if not _check_path_permission(self.workspace_path, target, self.context.get("_allowed_paths")):
|
||||
self._fail("no permission to access this file", path=str(target))
|
||||
return None
|
||||
if not self._validate_line_args(start_line, end_line):
|
||||
return None
|
||||
if not self._check_file(target):
|
||||
|
|
@ -137,8 +143,8 @@ class ReadStep(BaseStep):
|
|||
requested_end,
|
||||
max_collect_bytes=DEFAULT_MAX_BYTES * 2,
|
||||
)
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
self._fail(f"read failed: {e}", path=str(target))
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
self._fail(f"read failed: {exc}", path=str(target))
|
||||
return None
|
||||
if s > total:
|
||||
self._fail(f"start_line {s} exceeds file length ({total} lines)", path=str(target), total_lines=total)
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@
|
|||
import frontmatter
|
||||
|
||||
from ._file_io import detect_file_encoding, get_path_lock, write_file_safe
|
||||
from ._path import NON_MD_WARNING, gate_md, resolve_path
|
||||
from ._path import _check_path_permission, NON_MD_WARNING, gate_md, resolve_path
|
||||
from ..base_step import BaseStep
|
||||
from ...components import R
|
||||
|
||||
|
|
@ -18,6 +18,9 @@ class WriteStep(BaseStep):
|
|||
``metadata`` are ignored — only the top-level explicit parameters are
|
||||
honored for those two reserved fields.
|
||||
|
||||
Permission: honors the request-scoped ``_allowed_paths`` constraint
|
||||
injected by the server into the RuntimeContext.
|
||||
|
||||
Concurrency: in-process per-path ``asyncio.Lock`` serializes concurrent
|
||||
writes to the same file (multi-worker / multi-process safety is out of
|
||||
scope).
|
||||
|
|
@ -44,6 +47,10 @@ class WriteStep(BaseStep):
|
|||
|
||||
target, is_md = gate_md(target)
|
||||
|
||||
if not _check_path_permission(self.workspace_path, target, self.context.get("_allowed_paths")):
|
||||
self._fail("no permission to write this file", path=str(target))
|
||||
return None
|
||||
|
||||
# Non-markdown files have no frontmatter convention: name/description
|
||||
# and metadata are silently dropped and the body is written verbatim.
|
||||
if is_md:
|
||||
|
|
|
|||
|
|
@ -1755,6 +1755,7 @@ def test_auto_memory_uses_message_day_for_historical_create():
|
|||
def write_historical_note(inputs, _kwargs):
|
||||
assert f"Today: {historical_day}" in inputs
|
||||
assert f"date={historical_day}" in inputs
|
||||
assert "injected_job_kwargs" not in _kwargs
|
||||
write_file(
|
||||
cwd / "daily" / historical_day / "memory.md",
|
||||
"---\nname: memory\nsession_id: s1\n"
|
||||
|
|
|
|||
|
|
@ -75,6 +75,38 @@ def test_mcp_config_uses_stdio_bridge_and_selected_jobs(tmp_path):
|
|||
assert config["args"][config["args"].index("--tool-context-id") + 1] == "ctx-1"
|
||||
|
||||
|
||||
def test_mcp_config_serializes_injected_job_kwargs(tmp_path):
|
||||
wrapper, _job = _wrapper(tmp_path, mcp_config="custom.yaml")
|
||||
|
||||
config = wrapper._mcp_server_config( # pylint: disable=protected-access
|
||||
{
|
||||
"job_tools": ["search"],
|
||||
"tool_context_id": "ctx-1",
|
||||
"injected_job_kwargs": {"_allowed_paths": ["daily/2025-06-01/note.md"]},
|
||||
},
|
||||
)
|
||||
|
||||
raw = config["args"][config["args"].index("--injected-job-kwargs") + 1]
|
||||
assert json.loads(raw) == {"_allowed_paths": ["daily/2025-06-01/note.md"]}
|
||||
|
||||
plain = wrapper._mcp_server_config({"job_tools": ["search"]}) # pylint: disable=protected-access
|
||||
assert "--injected-job-kwargs" not in plain["args"]
|
||||
|
||||
|
||||
def test_prepare_config_merges_injected_job_kwargs_with_tool_context():
|
||||
prepared = _prepare_config(
|
||||
{"jobs": {"selected": {"backend": "base"}}},
|
||||
["selected"],
|
||||
"ctx-1",
|
||||
{"_allowed_paths": ["daily/2025-06-01/note.md"]},
|
||||
)
|
||||
|
||||
assert prepared["service"]["injected_job_kwargs"] == {
|
||||
"_allowed_paths": ["daily/2025-06-01/note.md"],
|
||||
"tool_context_id": "ctx-1",
|
||||
}
|
||||
|
||||
|
||||
def test_thread_config_preserves_other_mcp_servers(tmp_path):
|
||||
wrapper, _job = _wrapper(tmp_path)
|
||||
config = wrapper._thread_config( # pylint: disable=protected-access
|
||||
|
|
|
|||
241
tests/unit/test_injected_job_kwargs.py
Normal file
241
tests/unit/test_injected_job_kwargs.py
Normal file
|
|
@ -0,0 +1,241 @@
|
|||
"""Tests for the server-owned ``injected_job_kwargs`` wrapper mechanism.
|
||||
|
||||
Each agent wrapper merges these kwargs into every job tool call after
|
||||
receiving the model's arguments, rejects model-supplied conflicts, and hides
|
||||
the injected keys from the tool schema exposed to the model.
|
||||
"""
|
||||
|
||||
# pylint: disable=protected-access,missing-function-docstring
|
||||
|
||||
|
||||
import pytest
|
||||
|
||||
from reme.components.agent_wrapper import AsAgentWrapper, BaseAgentWrapper, CcAgentWrapper
|
||||
from reme.components.file_store import LocalFileStore
|
||||
from reme.schema import Response
|
||||
from reme.steps.evolve.auto_memory import AutoMemoryStep
|
||||
|
||||
|
||||
class _Job:
|
||||
"""Minimal job double recording every call."""
|
||||
|
||||
def __init__(self, name="write"):
|
||||
self.name = name
|
||||
self.description = "Write a note"
|
||||
self.parameters = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"path": {"type": "string"},
|
||||
"content": {"type": "string"},
|
||||
"date": {"type": "string"},
|
||||
},
|
||||
"required": ["path", "content", "date"],
|
||||
}
|
||||
self.calls: list[dict] = []
|
||||
|
||||
async def __call__(self, **kwargs):
|
||||
self.calls.append(kwargs)
|
||||
return Response(answer="done")
|
||||
|
||||
|
||||
# -- base helpers ---------------------------------------------------------------
|
||||
|
||||
|
||||
def test_merge_rejects_model_supplied_conflicts():
|
||||
injected = {"_allowed_paths": ["daily/a.md"], "date": "2025-06-01"}
|
||||
merged = BaseAgentWrapper._merge_injected_job_kwargs({"path": "daily/a.md"}, injected)
|
||||
assert merged == {"path": "daily/a.md", "_allowed_paths": ["daily/a.md"], "date": "2025-06-01"}
|
||||
|
||||
with pytest.raises(ValueError, match="cannot be provided by the model: _allowed_paths, date"):
|
||||
BaseAgentWrapper._merge_injected_job_kwargs(
|
||||
{"_allowed_paths": ["anywhere"], "date": "1999-01-01"},
|
||||
injected,
|
||||
)
|
||||
|
||||
|
||||
def test_strip_injected_parameters_hides_keys_from_schema():
|
||||
job = _Job()
|
||||
stripped = BaseAgentWrapper._strip_injected_parameters(job.parameters, {"date": "2025-06-01"})
|
||||
assert "date" not in stripped["properties"]
|
||||
assert stripped["required"] == ["path", "content"]
|
||||
# Underscore keys never appear in schemas; stripping is a no-op then.
|
||||
untouched = BaseAgentWrapper._strip_injected_parameters(job.parameters, {"_allowed_paths": ["x"]})
|
||||
assert untouched["properties"].keys() == job.parameters["properties"].keys()
|
||||
# The original job schema is never mutated.
|
||||
assert "date" in job.parameters["properties"]
|
||||
|
||||
|
||||
# -- AgentScope wrapper -----------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_as_tool_injects_kwargs_and_rejects_conflicts():
|
||||
job = _Job()
|
||||
tool = AsAgentWrapper._make_tool(job, "ctx-1", {"_allowed_paths": ["daily/a.md"], "date": "2025-06-01"})
|
||||
|
||||
assert "date" not in tool.input_schema["properties"]
|
||||
assert tool.input_schema["required"] == ["path", "content"]
|
||||
|
||||
await tool.call(path="daily/a.md", content="hi")
|
||||
assert job.calls == [
|
||||
{
|
||||
"path": "daily/a.md",
|
||||
"content": "hi",
|
||||
"_allowed_paths": ["daily/a.md"],
|
||||
"date": "2025-06-01",
|
||||
"tool_context_id": "ctx-1",
|
||||
},
|
||||
]
|
||||
|
||||
with pytest.raises(ValueError, match="cannot be provided by the model"):
|
||||
await tool.call(path="daily/a.md", content="hi", _allowed_paths=["everything"])
|
||||
assert len(job.calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_as_tool_without_injection_keeps_original_schema():
|
||||
job = _Job()
|
||||
tool = AsAgentWrapper._make_tool(job)
|
||||
assert tool.input_schema["required"] == ["path", "content", "date"]
|
||||
|
||||
await tool.call(path="a.md", content="x", date="2025-06-01")
|
||||
assert job.calls == [{"path": "a.md", "content": "x", "date": "2025-06-01"}]
|
||||
|
||||
|
||||
# -- Claude Code wrapper ----------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cc_tool_injects_kwargs_and_rejects_conflicts():
|
||||
job = _Job()
|
||||
tool = CcAgentWrapper._make_tool(job, "ctx-1", {"_allowed_paths": ["daily/a.md"]})
|
||||
|
||||
assert "_allowed_paths" not in tool.input_schema["properties"]
|
||||
|
||||
result = await tool.handler({"path": "daily/a.md", "content": "hi", "date": "2025-06-01"})
|
||||
assert result["is_error"] is False
|
||||
assert job.calls == [
|
||||
{
|
||||
"path": "daily/a.md",
|
||||
"content": "hi",
|
||||
"date": "2025-06-01",
|
||||
"_allowed_paths": ["daily/a.md"],
|
||||
"tool_context_id": "ctx-1",
|
||||
},
|
||||
]
|
||||
|
||||
with pytest.raises(ValueError, match="cannot be provided by the model"):
|
||||
await tool.handler({"path": "daily/a.md", "content": "hi", "_allowed_paths": ["everything"]})
|
||||
assert len(job.calls) == 1
|
||||
|
||||
|
||||
# -- AutoMemoryStep ---------------------------------------------------------------
|
||||
|
||||
|
||||
class _RecordingWrapper(BaseAgentWrapper):
|
||||
"""Agent wrapper double capturing reply() kwargs."""
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self.calls: list[dict] = []
|
||||
|
||||
async def reply(self, inputs, **kwargs) -> dict:
|
||||
self.calls.append(kwargs)
|
||||
return {"session_id": "s-1", "last_message": {}, "result": "ok"}
|
||||
|
||||
|
||||
async def _auto_memory_step(wrapper) -> AutoMemoryStep:
|
||||
store = LocalFileStore(name="t_inject", embedding_store="") # workspace rooted at cwd
|
||||
await store.start()
|
||||
return AutoMemoryStep(name="auto_memory", agent_wrapper=wrapper, file_store=store)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auto_memory_create_uses_model_supplied_date_and_daily_write(tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
wrapper = _RecordingWrapper(name="fake")
|
||||
step = await _auto_memory_step(wrapper)
|
||||
|
||||
async def no_note(_day, _session_id):
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(step, "_list_session_note", no_note)
|
||||
|
||||
await step(
|
||||
session_id="sess-1",
|
||||
messages=[{"name": "user", "role": "user", "content": "hi", "created_at": "2025-06-01T10:00:00"}],
|
||||
)
|
||||
|
||||
assert step.context.response.success is True
|
||||
assert wrapper.calls[0]["job_tools"] == ["daily_write"]
|
||||
assert "injected_job_kwargs" not in wrapper.calls[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auto_memory_update_scopes_tools_to_exact_note_path(tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
wrapper = _RecordingWrapper(name="fake")
|
||||
step = await _auto_memory_step(wrapper)
|
||||
session_id = "sess-1"
|
||||
note_path = "daily/2025-06-01/existing-note.md"
|
||||
note_file = tmp_path / note_path
|
||||
note_file.parent.mkdir(parents=True)
|
||||
note_file.write_text(
|
||||
f"---\nsession_id: {session_id}\nsource_conversation: '{step._session_link(session_id)}'\n---\nbody\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
async def existing_note(_day, _session_id):
|
||||
return {"path": note_path, "session_id": session_id}
|
||||
|
||||
async def no_index(_store, _day, _daily_dir):
|
||||
return {}
|
||||
|
||||
monkeypatch.setattr(step, "_list_session_note", existing_note)
|
||||
monkeypatch.setattr("reme.steps.evolve.auto_memory.refresh_day_index", no_index)
|
||||
|
||||
await step(
|
||||
session_id=session_id,
|
||||
messages=[{"name": "user", "role": "user", "content": "hi again", "created_at": "2025-06-01T11:00:00"}],
|
||||
)
|
||||
|
||||
assert step.context.response.success is True
|
||||
assert wrapper.calls[0]["job_tools"] == ["read", "edit", "frontmatter_update", "write"]
|
||||
assert wrapper.calls[0]["injected_job_kwargs"] == {"_allowed_paths": [note_path]}
|
||||
|
||||
|
||||
def test_auto_memory_keeps_original_tool_names():
|
||||
"""BEAM/LME configs define only the original jobs; no *_daily variants exist."""
|
||||
step = AutoMemoryStep(name="auto_memory")
|
||||
assert step.create_tools == ["daily_write"]
|
||||
assert step.update_tools == ["read", "edit", "frontmatter_update", "write"]
|
||||
|
||||
|
||||
def test_auto_memory_create_prompts_match_upstream_date_arguments():
|
||||
"""Auto-memory prompts keep the upstream model-supplied date argument."""
|
||||
from pathlib import Path
|
||||
|
||||
prompt_files = (
|
||||
Path("reme/steps/evolve/auto_memory.yaml"),
|
||||
Path("reme/steps/benchmark/beam/auto_memory.yaml"),
|
||||
Path("reme/steps/benchmark/lme/auto_memory.yaml"),
|
||||
)
|
||||
for prompt_file in prompt_files:
|
||||
content = prompt_file.read_text(encoding="utf-8")
|
||||
assert "date={today}" in content or "`date`: {today}" in content or "`date`：{today}" in content
|
||||
|
||||
|
||||
def test_configs_define_original_jobs_without_daily_variants():
|
||||
from reme.config import resolve_app_config
|
||||
|
||||
for config_name in ("default", "lme", "beam"):
|
||||
config = resolve_app_config(config=config_name, log_config=False)
|
||||
jobs = config["jobs"]
|
||||
for name in ("read", "edit", "write", "frontmatter_update", "daily_write"):
|
||||
assert name in jobs, f"{config_name} missing job {name}"
|
||||
for name in ("read_daily", "edit_daily", "write_daily"):
|
||||
assert name not in jobs, f"{config_name} unexpectedly defines {name}"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
263
tests/unit/test_path_permission.py
Normal file
263
tests/unit/test_path_permission.py
Normal file
|
|
@ -0,0 +1,263 @@
|
|||
"""Tests for the request-scoped ``_allowed_paths`` permission mechanism.
|
||||
|
||||
The constraint is server-owned: AutoMemoryStep passes it to the agent wrapper
|
||||
via ``injected_job_kwargs``; the wrapper merges it into every job tool call
|
||||
(rejecting model-supplied conflicts) and BaseJob places it into the
|
||||
per-invocation RuntimeContext, where the file I/O steps read it.
|
||||
|
||||
Covers:
|
||||
- ReadStep / EditStep / WriteStep / FrontmatterUpdateStep 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
|
||||
(AgentScope, Claude Code, Codex serialization).
|
||||
- AutoMemoryStep injecting ``date`` (create) / ``_allowed_paths`` (update)
|
||||
while keeping the original read/edit/write/frontmatter_update tool names.
|
||||
"""
|
||||
|
||||
# pylint: disable=protected-access
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from reme.components.file_store import LocalFileStore
|
||||
from reme.steps.file_io._path import _check_path_permission
|
||||
from reme.steps.file_io.edit import EditStep
|
||||
from reme.steps.file_io.frontmatter_update import FrontmatterUpdateStep
|
||||
from reme.steps.file_io.read import ReadStep
|
||||
from reme.steps.file_io.write import WriteStep
|
||||
|
||||
|
||||
class temp_chdir:
|
||||
"""Context manager to temporarily chdir into a path and restore on exit."""
|
||||
|
||||
def __init__(self, path):
|
||||
self.path = path
|
||||
self.old = None
|
||||
|
||||
def __enter__(self):
|
||||
self.old = os.getcwd()
|
||||
os.chdir(self.path)
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc):
|
||||
os.chdir(self.old)
|
||||
|
||||
|
||||
def _seed(workspace: Path, rel: str, body: str = "body\n") -> Path:
|
||||
target = workspace / rel
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_text(body, encoding="utf-8")
|
||||
return target
|
||||
|
||||
|
||||
async def _make_store() -> LocalFileStore:
|
||||
store = LocalFileStore(name="t_perm", embedding_store="")
|
||||
await store.start()
|
||||
return store
|
||||
|
||||
|
||||
async def _run(step_cls, store: LocalFileStore, **kwargs):
|
||||
"""Run a file I/O step; ``_allowed_paths`` rides in like injected job kwargs."""
|
||||
step = step_cls(file_store=store)
|
||||
await step(**kwargs)
|
||||
return step.context.response
|
||||
|
||||
|
||||
NOTE = "daily/2025-06-01/podcast-habits.md"
|
||||
SIBLING = "daily/2025-06-01/other-note.md"
|
||||
OUTSIDE = "topics/roadmap.md"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_scope_allows_exact_note():
|
||||
"""All four update tools succeed on the exact note_path they are scoped to."""
|
||||
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
|
||||
_seed(Path(tmp), NOTE, "---\nname: x\n---\nold text\n")
|
||||
store = await _make_store()
|
||||
scope = {"_allowed_paths": [NOTE]}
|
||||
|
||||
resp = await _run(ReadStep, store, path=NOTE, **scope)
|
||||
assert resp.success is True
|
||||
assert "old text" in str(resp.answer)
|
||||
|
||||
resp = await _run(EditStep, store, path=NOTE, old="old text", new="new text", **scope)
|
||||
assert resp.success is True
|
||||
|
||||
resp = await _run(FrontmatterUpdateStep, store, path=NOTE, metadata={"name": "renamed"}, **scope)
|
||||
assert resp.success is True
|
||||
|
||||
resp = await _run(WriteStep, store, path=NOTE, name="n", description="d", content="rewritten", **scope)
|
||||
assert resp.success is True
|
||||
assert "rewritten" in (Path(tmp) / NOTE).read_text(encoding="utf-8")
|
||||
await store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_scope_rejects_sibling_note_in_same_daily_dir():
|
||||
"""Exact-file scope denies another note in the same daily directory (least privilege)."""
|
||||
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
|
||||
_seed(Path(tmp), NOTE)
|
||||
sibling = _seed(Path(tmp), SIBLING, "---\nname: s\n---\nkeep\n")
|
||||
store = await _make_store()
|
||||
scope = {"_allowed_paths": [NOTE]}
|
||||
|
||||
for coro in (
|
||||
_run(ReadStep, store, path=SIBLING, **scope),
|
||||
_run(EditStep, store, path=SIBLING, old="keep", new="gone", **scope),
|
||||
_run(WriteStep, store, path=SIBLING, content="overwrite", **scope),
|
||||
_run(FrontmatterUpdateStep, store, path=SIBLING, metadata={"name": "hijack"}, **scope),
|
||||
):
|
||||
resp = await coro
|
||||
assert resp.success is False
|
||||
assert "no permission" in str(resp.answer).lower()
|
||||
assert "keep" in sibling.read_text(encoding="utf-8")
|
||||
await store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_scope_rejects_paths_outside_daily_dir():
|
||||
"""Exact-file scope denies files elsewhere in the workspace."""
|
||||
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
|
||||
_seed(Path(tmp), NOTE)
|
||||
outside = _seed(Path(tmp), OUTSIDE, "roadmap\n")
|
||||
store = await _make_store()
|
||||
scope = {"_allowed_paths": [NOTE]}
|
||||
|
||||
for coro in (
|
||||
_run(ReadStep, store, path=OUTSIDE, **scope),
|
||||
_run(EditStep, store, path=OUTSIDE, old="roadmap", new="x", **scope),
|
||||
_run(WriteStep, store, path=OUTSIDE, content="x", **scope),
|
||||
_run(FrontmatterUpdateStep, store, path=OUTSIDE, metadata={"k": "v"}, **scope),
|
||||
):
|
||||
resp = await coro
|
||||
assert resp.success is False
|
||||
assert "no permission" in str(resp.answer).lower()
|
||||
assert outside.read_text(encoding="utf-8") == "roadmap\n"
|
||||
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."""
|
||||
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
|
||||
_seed(Path(tmp), NOTE, "gated\n")
|
||||
store = await _make_store()
|
||||
|
||||
resp = await _run(ReadStep, store, path=NOTE.removesuffix(".md"), _allowed_paths=[NOTE])
|
||||
assert resp.success is True
|
||||
assert "gated" in str(resp.answer)
|
||||
await store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_custom_daily_dir_scope_needs_no_config_jobs():
|
||||
"""A customized daily_dir (e.g. journal/) works because the scope is the note path itself."""
|
||||
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
|
||||
note = "journal/2025-06-01/trip.md"
|
||||
_seed(Path(tmp), note, "trip\n")
|
||||
_seed(Path(tmp), "journal/2025-06-01/other.md")
|
||||
store = await _make_store()
|
||||
scope = {"_allowed_paths": [note]}
|
||||
|
||||
resp = await _run(ReadStep, store, path=note, **scope)
|
||||
assert resp.success is True
|
||||
|
||||
resp = await _run(WriteStep, store, path="journal/2025-06-01/other.md", content="x", **scope)
|
||||
assert resp.success is False
|
||||
assert "no permission" in str(resp.answer).lower()
|
||||
await store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_directory_scope_allows_nested_paths():
|
||||
"""A directory entry acts as a prefix scope with path-component boundaries."""
|
||||
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
|
||||
_seed(Path(tmp), "daily/2025-06-01/deep/nested.md", "nested\n")
|
||||
_seed(Path(tmp), "daily-report/leak.md", "leak\n")
|
||||
store = await _make_store()
|
||||
scope = {"_allowed_paths": ["daily"]}
|
||||
|
||||
resp = await _run(ReadStep, store, path="daily/2025-06-01/deep/nested.md", **scope)
|
||||
assert resp.success is True
|
||||
|
||||
resp = await _run(ReadStep, store, path="daily-report/leak.md", **scope)
|
||||
assert resp.success is False
|
||||
assert "no permission" in str(resp.answer).lower()
|
||||
await store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_nonexistent_allowed_paths_allow_component_bounded_descendants():
|
||||
"""A missing allowed path permits itself and descendants, not string-prefix siblings."""
|
||||
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
|
||||
workspace = Path(tmp)
|
||||
_seed(workspace, "X/YZ/not-allowed.md", "keep\n")
|
||||
store = await _make_store()
|
||||
|
||||
scope = {"_allowed_paths": ["X/Y"]}
|
||||
assert _check_path_permission(workspace, workspace / "X/Y", scope["_allowed_paths"])
|
||||
assert _check_path_permission(workspace, workspace / "X/Y/Z", scope["_allowed_paths"])
|
||||
assert not _check_path_permission(workspace, workspace / "X/YZ", scope["_allowed_paths"])
|
||||
|
||||
resp = await _run(WriteStep, store, path="X/Y/nested", content="nested", **scope)
|
||||
assert resp.success is True
|
||||
assert (workspace / "X/Y/nested.md").read_text(encoding="utf-8") == "nested\n"
|
||||
|
||||
resp = await _run(ReadStep, store, path="X/YZ/not-allowed.md", **scope)
|
||||
assert resp.success is False
|
||||
assert "no permission" in str(resp.answer).lower()
|
||||
await store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_home_relative_paths_are_not_supported():
|
||||
"""Home-relative targets are rejected and home-relative scopes grant nothing."""
|
||||
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
|
||||
_seed(Path(tmp), "journal/kept.md", "kept\n")
|
||||
store = await _make_store()
|
||||
|
||||
resp = await _run(ReadStep, store, path="journal/kept.md", _allowed_paths=["~/journal/kept.md"])
|
||||
assert resp.success is False
|
||||
assert "no permission" in str(resp.answer).lower()
|
||||
|
||||
resp = await _run(ReadStep, store, path="~/journal/kept.md")
|
||||
assert resp.success is False
|
||||
assert "does not exist" in str(resp.answer).lower()
|
||||
await store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_injected_constraints_fail_closed():
|
||||
"""Empty lists, escaping entries, and non-list values all deny access."""
|
||||
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
|
||||
_seed(Path(tmp), NOTE, "target\n")
|
||||
store = await _make_store()
|
||||
|
||||
for bad_scope in ([], ["../escape.md"], ["/etc/passwd"], 42, {"path": NOTE}):
|
||||
resp = await _run(ReadStep, store, path=NOTE, _allowed_paths=bad_scope)
|
||||
assert resp.success is False, f"scope {bad_scope!r} should fail closed"
|
||||
assert "no permission" in str(resp.answer).lower()
|
||||
await store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_constraint_allows_all():
|
||||
"""Without ``_allowed_paths`` the steps impose no additional restriction."""
|
||||
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
|
||||
_seed(Path(tmp), "anywhere/x.md", "free\n")
|
||||
store = await _make_store()
|
||||
|
||||
resp = await _run(ReadStep, store, path="anywhere/x.md")
|
||||
assert resp.success is True
|
||||
|
||||
resp = await _run(WriteStep, store, path="anywhere/new.md", content="ok")
|
||||
assert resp.success is True
|
||||
await store.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Loading…
Add table
Reference in a new issue