tools/workspace: mode-aware path grants and escape hint

This commit is contained in:
2026-09-09 22:11:50 +08:00
parent 66c58cf156
commit ba021cb138
16 changed files with 214 additions and 40 deletions
+3
View File
@@ -66,6 +66,7 @@ from .workspace import (
DEFAULT_COMMAND_DENYLIST as DEFAULT_COMMAND_DENYLIST, DEFAULT_COMMAND_DENYLIST as DEFAULT_COMMAND_DENYLIST,
) )
from .workspace import DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS as DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS from .workspace import DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS as DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS
from .workspace import AccessMode as AccessMode
from .workspace import WorkspaceError as WorkspaceError from .workspace import WorkspaceError as WorkspaceError
from .workspace import WorkspacePolicy as WorkspacePolicy from .workspace import WorkspacePolicy as WorkspacePolicy
from .workspace import active_workspace_policy as active_workspace_policy from .workspace import active_workspace_policy as active_workspace_policy
@@ -80,6 +81,8 @@ from .workspace import get_policy_confirm_timeout as get_policy_confirm_timeout
from .workspace import get_workspace_root as get_workspace_root from .workspace import get_workspace_root as get_workspace_root
from .workspace import grant_policy_command as grant_policy_command from .workspace import grant_policy_command as grant_policy_command
from .workspace import list_workspace_allowlist as list_workspace_allowlist from .workspace import list_workspace_allowlist as list_workspace_allowlist
from .workspace import mode_covers as mode_covers
from .workspace import parse_access_mode as parse_access_mode
from .workspace import pop_owned_temporary_workspaces as pop_owned_temporary_workspaces from .workspace import pop_owned_temporary_workspaces as pop_owned_temporary_workspaces
from .workspace import remove_workspace_allowlist as remove_workspace_allowlist from .workspace import remove_workspace_allowlist as remove_workspace_allowlist
from .workspace import resolve_path as resolve_path from .workspace import resolve_path as resolve_path
+4 -1
View File
@@ -14,7 +14,7 @@ if TYPE_CHECKING:
from pathlib import Path from pathlib import Path
from plyngent.agent.todo_stack import TodoStack from plyngent.agent.todo_stack import TodoStack
from plyngent.tools.workspace import WorkspacePolicy from plyngent.tools.workspace import AccessMode, WorkspacePolicy
def _default_instance_data() -> PersistentDataView[Any]: def _default_instance_data() -> PersistentDataView[Any]:
@@ -78,6 +78,9 @@ class SessionState:
# Durable copy lives under data["grants"] (see plyngent.tools.grants). # Durable copy lives under data["grants"] (see plyngent.tools.grants).
grants: dict[str, bool] = field(default_factory=dict) grants: dict[str, bool] = field(default_factory=dict)
extras: dict[str, Any] = field(default_factory=dict) extras: dict[str, Any] = field(default_factory=dict)
# Directory-access grants for this session (resolved path → mode). Config /
# process-scoped grants live on WorkspacePolicy; see plyngent.tools.access.
access_grants: dict[Path, AccessMode] = field(default_factory=dict)
def has_grant(self, tool_name: str) -> bool: def has_grant(self, tool_name: str) -> bool:
return bool(self.grants.get(tool_name)) return bool(self.grants.get(tool_name))
+3 -3
View File
@@ -4,7 +4,7 @@ import json
import shlex import shlex
from typing import TYPE_CHECKING, cast from typing import TYPE_CHECKING, cast
from plyngent.tools.workspace import WorkspaceError, resolve_path from plyngent.tools.workspace import AccessMode, WorkspaceError, resolve_path
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Mapping, Sequence from collections.abc import Mapping, Sequence
@@ -140,7 +140,7 @@ def _write_file_reason(args: Mapping[str, object]) -> str | None:
if not isinstance(path, str) or not path: if not isinstance(path, str) or not path:
return None return None
try: try:
target = resolve_path(path) target = resolve_path(path, required=AccessMode.WRITE)
except WorkspaceError: except WorkspaceError:
# Path policy will fail later; no soft-confirm without a resolved target. # Path policy will fail later; no soft-confirm without a resolved target.
return None return None
@@ -160,7 +160,7 @@ def _copy_path_reason(args: Mapping[str, object]) -> str | None:
if not isinstance(dst, str) or not dst: if not isinstance(dst, str) or not dst:
return f"copy {src!r} → {dst!r} (overwrite)" return f"copy {src!r} → {dst!r} (overwrite)"
try: try:
target = resolve_path(dst) target = resolve_path(dst, required=AccessMode.WRITE)
except WorkspaceError: except WorkspaceError:
return None return None
if target.exists(): if target.exists():
+2 -2
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from plyngent.agent import ToolTag, lineno_read_lines, lineno_read_mtime, tool from plyngent.agent import ToolTag, lineno_read_lines, lineno_read_mtime, tool
from plyngent.tools.workspace import resolve_path from plyngent.tools.workspace import AccessMode, resolve_path
if TYPE_CHECKING: if TYPE_CHECKING:
from pathlib import Path from pathlib import Path
@@ -114,7 +114,7 @@ async def edit_lineno(path: str, start_line: int, end_line: int, new_content: st
``new_content`` may be multi-line. Use an empty string to delete the range. ``new_content`` may be multi-line. Use an empty string to delete the range.
""" """
target = resolve_path(path) target = resolve_path(path, required=AccessMode.WRITE)
if not target.is_file(): if not target.is_file():
return f"error: not a file: {path}" return f"error: not a file: {path}"
+2 -2
View File
@@ -1,7 +1,7 @@
from __future__ import annotations from __future__ import annotations
from plyngent.agent import ToolTag, tool from plyngent.agent import ToolTag, tool
from plyngent.tools.workspace import resolve_path from plyngent.tools.workspace import AccessMode, resolve_path
def _count_non_overlapping(text: str, needle: str) -> int: def _count_non_overlapping(text: str, needle: str) -> int:
@@ -60,7 +60,7 @@ async def edit_replace(path: str, old_string: str, new_string: str, max_replaces
return "error: old_string must not be empty" return "error: old_string must not be empty"
if max_replaces < 1: if max_replaces < 1:
return "error: max_replaces must be >= 1" return "error: max_replaces must be >= 1"
target = resolve_path(path) target = resolve_path(path, required=AccessMode.WRITE)
if not target.is_file(): if not target.is_file():
return f"error: not a file: {path}" return f"error: not a file: {path}"
text = target.read_text(encoding="utf-8", errors="replace") text = target.read_text(encoding="utf-8", errors="replace")
+19 -9
View File
@@ -4,7 +4,7 @@ import shutil
from pathlib import Path from pathlib import Path
from plyngent.agent import ToolTag, invalidate_lineno_read, tool from plyngent.agent import ToolTag, invalidate_lineno_read, tool
from plyngent.tools.workspace import WorkspaceError, get_workspace_root, resolve_path from plyngent.tools.workspace import AccessMode, WorkspaceError, get_workspace_root, resolve_path
def _kind(path: Path) -> str: def _kind(path: Path) -> str:
@@ -26,18 +26,28 @@ def _remove_existing(path: Path) -> None:
def _resolve_pair(src: str, dst: str) -> tuple[Path, Path, Path] | str: def _resolve_pair(src: str, dst: str) -> tuple[Path, Path, Path] | str:
try: try:
source = resolve_path(src) source = resolve_path(src, required=AccessMode.READ)
dest = resolve_path(dst) dest = resolve_path(dst, required=AccessMode.WRITE)
root = get_workspace_root() root = get_workspace_root()
except WorkspaceError as exc: except WorkspaceError as exc:
return f"error: {exc}" return f"error: {exc}"
return source, dest, root return source, dest, root
def _prepare_dest(source: Path, dest: Path, root: Path, *, overwrite: bool, dst_label: str) -> Path | str: def _display(path: Path, root: Path) -> str:
"""Workspace-relative display path, falling back to absolute outside the root."""
try:
return str(path.relative_to(root))
except ValueError:
return str(path)
def _prepare_dest(source: Path, dest: Path, *, overwrite: bool, dst_label: str) -> Path | str:
"""Resolve copy/move destination; return error string on conflict.""" """Resolve copy/move destination; return error string on conflict."""
if source.is_file() and dest.is_dir(): if source.is_file() and dest.is_dir():
dest = resolve_path(str((dest / source.name).relative_to(root))) # ``dest`` may be outside the primary root (temp workspace / grant), so
# resolve the child directly instead of relativizing against the root.
dest = resolve_path(str(dest / source.name), required=AccessMode.WRITE)
if dest.exists() or dest.is_symlink(): if dest.exists() or dest.is_symlink():
if not overwrite: if not overwrite:
return f"error: destination exists: {dst_label} (set overwrite=true)" return f"error: destination exists: {dst_label} (set overwrite=true)"
@@ -56,7 +66,7 @@ def _copy_or_move_validated(
overwrite: bool, overwrite: bool,
action: str, action: str,
) -> str: ) -> str:
prepared = _prepare_dest(source, dest, root, overwrite=overwrite, dst_label=dst) prepared = _prepare_dest(source, dest, overwrite=overwrite, dst_label=dst)
if isinstance(prepared, str): if isinstance(prepared, str):
return prepared return prepared
dest = prepared dest = prepared
@@ -72,12 +82,12 @@ def _copy_or_move_validated(
# new. copy2/copytree preserve the source mtime, so the freshness # new. copy2/copytree preserve the source mtime, so the freshness
# check alone could miss the change — invalidate explicitly. # check alone could miss the change — invalidate explicitly.
invalidate_lineno_read(str(dest)) invalidate_lineno_read(str(dest))
return f"copied {label} {src} -> {dest.relative_to(root)}" return f"copied {label} {src} -> {_display(dest, root)}"
moved = Path(shutil.move(str(source), str(dest))).resolve() moved = Path(shutil.move(str(source), str(dest))).resolve()
# shutil.move preserves mtime too; drop read state for both paths. # shutil.move preserves mtime too; drop read state for both paths.
invalidate_lineno_read(str(source)) invalidate_lineno_read(str(source))
invalidate_lineno_read(str(moved)) invalidate_lineno_read(str(moved))
return f"moved {_kind(moved)} {src} -> {moved.relative_to(root)}" return f"moved {_kind(moved)} {src} -> {_display(moved, root)}"
except WorkspaceError as exc: except WorkspaceError as exc:
return f"error: {exc}" return f"error: {exc}"
except OSError as exc: except OSError as exc:
@@ -147,7 +157,7 @@ async def delete_path(path: str, *, recursive: bool = False) -> str:
``recursive=true`` (uses ``shutil.rmtree``). The workspace root cannot be deleted. ``recursive=true`` (uses ``shutil.rmtree``). The workspace root cannot be deleted.
""" """
try: try:
target = resolve_path(path) target = resolve_path(path, required=AccessMode.WRITE)
root = get_workspace_root() root = get_workspace_root()
except WorkspaceError as exc: except WorkspaceError as exc:
return f"error: {exc}" return f"error: {exc}"
+2 -2
View File
@@ -4,7 +4,7 @@ from typing import TYPE_CHECKING
from plyngent.agent import ToolTag, tool from plyngent.agent import ToolTag, tool
from plyngent.tools.file.tree import VCS_DIR_NAMES from plyngent.tools.file.tree import VCS_DIR_NAMES
from plyngent.tools.workspace import WorkspaceError, get_workspace_root, resolve_path from plyngent.tools.workspace import AccessMode, WorkspaceError, get_workspace_root, resolve_path
if TYPE_CHECKING: if TYPE_CHECKING:
from pathlib import Path from pathlib import Path
@@ -57,7 +57,7 @@ def _collect_glob(
def _resolve_glob_base(path: str) -> tuple[Path, Path] | str: def _resolve_glob_base(path: str) -> tuple[Path, Path] | str:
try: try:
root = get_workspace_root() root = get_workspace_root()
base = resolve_path(path) base = resolve_path(path, required=AccessMode.READ)
except WorkspaceError as exc: except WorkspaceError as exc:
return f"error: {exc}" return f"error: {exc}"
if not base.is_dir(): if not base.is_dir():
+2 -2
View File
@@ -5,7 +5,7 @@ from typing import TYPE_CHECKING
from plyngent.agent import ToolTag, tool from plyngent.agent import ToolTag, tool
from plyngent.tools.file.tree import VCS_DIR_NAMES from plyngent.tools.file.tree import VCS_DIR_NAMES
from plyngent.tools.workspace import WorkspaceError, get_workspace_root, resolve_path from plyngent.tools.workspace import AccessMode, WorkspaceError, get_workspace_root, resolve_path
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Iterator from collections.abc import Iterator
@@ -65,7 +65,7 @@ def _iter_files(base: Path, *, skip_hidden_dirs: bool) -> list[Path]:
def _resolve_files(path: str, *, skip_hidden_dirs: bool) -> list[Path] | str: def _resolve_files(path: str, *, skip_hidden_dirs: bool) -> list[Path] | str:
try: try:
base = resolve_path(path) base = resolve_path(path, required=AccessMode.READ)
except WorkspaceError as exc: except WorkspaceError as exc:
return f"error: {exc}" return f"error: {exc}"
if not base.exists(): if not base.exists():
+2 -2
View File
@@ -1,13 +1,13 @@
from __future__ import annotations from __future__ import annotations
from plyngent.agent import ToolTag, tool from plyngent.agent import ToolTag, tool
from plyngent.tools.workspace import resolve_path from plyngent.tools.workspace import AccessMode, resolve_path
@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY) @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
async def listdir(path: str = ".") -> str: async def listdir(path: str = ".") -> str:
"""List entries in a directory under the workspace (name and type).""" """List entries in a directory under the workspace (name and type)."""
target = resolve_path(path) target = resolve_path(path, required=AccessMode.READ)
if not target.is_dir(): if not target.is_dir():
return f"error: not a directory: {path}" return f"error: not a directory: {path}"
lines: list[str] = [] lines: list[str] = []
+3 -3
View File
@@ -5,7 +5,7 @@ from typing import cast
from plyngent.agent import ToolTag, mark_lineno_read, tool from plyngent.agent import ToolTag, mark_lineno_read, tool
from plyngent.agent.budget import DEFAULT_TOOL_RESULT_MAX_CHARS from plyngent.agent.budget import DEFAULT_TOOL_RESULT_MAX_CHARS
from plyngent.tools.truncate_token import TruncateToken, truncate_with_token, truncation_marker from plyngent.tools.truncate_token import TruncateToken, truncate_with_token, truncation_marker
from plyngent.tools.workspace import resolve_path from plyngent.tools.workspace import AccessMode, resolve_path
_LINENO_WIDTH = 6 _LINENO_WIDTH = 6
@@ -41,7 +41,7 @@ def read_raw_text(path: str) -> tuple[str | None, str]:
always refer to the raw file text (never the ``L{begin}-{end}`` header). always refer to the raw file text (never the ``L{begin}-{end}`` header).
Missing paths and directories are distinguished for the caller. Missing paths and directories are distinguished for the caller.
""" """
target = resolve_path(path) target = resolve_path(path, required=AccessMode.READ)
if not target.exists(): if not target.exists():
return None, f"error: file not found: {path}" return None, f"error: file not found: {path}"
if not target.is_file(): if not target.is_file():
@@ -72,7 +72,7 @@ async def read_file(
so ``get_truncated`` can continue reading the rest without a new request. so ``get_truncated`` can continue reading the rest without a new request.
Pass ``max_chars < 1`` to read unbounded. Pass ``max_chars < 1`` to read unbounded.
""" """
target = resolve_path(path) target = resolve_path(path, required=AccessMode.READ)
text, err = read_raw_text(path) text, err = read_raw_text(path)
if err: if err:
return err return err
+2 -2
View File
@@ -4,7 +4,7 @@ from dataclasses import dataclass
from typing import TYPE_CHECKING, Literal from typing import TYPE_CHECKING, Literal
from plyngent.agent import ToolTag, tool from plyngent.agent import ToolTag, tool
from plyngent.tools.workspace import WorkspaceError, get_path_denylist, resolve_path from plyngent.tools.workspace import AccessMode, WorkspaceError, get_path_denylist, resolve_path
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Sequence from collections.abc import Sequence
@@ -270,7 +270,7 @@ async def tree(
return "error: max_entries must be >= 1" return "error: max_entries must be >= 1"
try: try:
origin = resolve_path(path) origin = resolve_path(path, required=AccessMode.READ)
except WorkspaceError as exc: except WorkspaceError as exc:
return f"error: {exc}" return f"error: {exc}"
if not origin.is_dir(): if not origin.is_dir():
+2 -2
View File
@@ -1,13 +1,13 @@
from __future__ import annotations from __future__ import annotations
from plyngent.agent import ToolTag, tool from plyngent.agent import ToolTag, tool
from plyngent.tools.workspace import resolve_path from plyngent.tools.workspace import AccessMode, resolve_path
@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO) @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO)
async def write_file(path: str, content: str) -> str: async def write_file(path: str, content: str) -> str:
"""Write text content to a file under the workspace (creates parents).""" """Write text content to a file under the workspace (creates parents)."""
target = resolve_path(path) target = resolve_path(path, required=AccessMode.WRITE)
target.parent.mkdir(parents=True, exist_ok=True) target.parent.mkdir(parents=True, exist_ok=True)
_ = target.write_text(content, encoding="utf-8") _ = target.write_text(content, encoding="utf-8")
# The write changes the file's mtime, so edit_lineno's freshness check # The write changes the file's mtime, so edit_lineno's freshness check
+2 -1
View File
@@ -10,6 +10,7 @@ if TYPE_CHECKING:
from pathlib import Path from pathlib import Path
from plyngent.tools.workspace import ( from plyngent.tools.workspace import (
AccessMode,
WorkspaceError, WorkspaceError,
check_command_allowed, check_command_allowed,
get_workspace_root, get_workspace_root,
@@ -61,7 +62,7 @@ def format_command_result(
def resolve_workdir(cwd: str) -> tuple["Path", str]: def resolve_workdir(cwd: str) -> tuple["Path", str]:
"""Return (absolute workdir, display path relative to workspace when possible).""" """Return (absolute workdir, display path relative to workspace when possible)."""
workdir = resolve_path(cwd) workdir = resolve_path(cwd, required=AccessMode.EXEC)
if not workdir.is_dir(): if not workdir.is_dir():
msg = f"not a directory: {cwd}" msg = f"not a directory: {cwd}"
raise WorkspaceError(msg) raise WorkspaceError(msg)
+2 -2
View File
@@ -9,7 +9,7 @@ from typing import TYPE_CHECKING, ClassVar, cast
from plyngent.tools.process.pty_backend import PtyHandle, pty_available, spawn_pty from plyngent.tools.process.pty_backend import PtyHandle, pty_available, spawn_pty
from plyngent.tools.process.pty_terminal import sanitize_pty_output_for_tool from plyngent.tools.process.pty_terminal import sanitize_pty_output_for_tool
from plyngent.tools.workspace import WorkspaceError, check_command_allowed, resolve_path from plyngent.tools.workspace import AccessMode, WorkspaceError, check_command_allowed, resolve_path
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Callable from collections.abc import Callable
@@ -122,7 +122,7 @@ class PtyManager:
msg = "PTY is not available on this platform (Windows needs pywinpty)" msg = "PTY is not available on this platform (Windows needs pywinpty)"
raise WorkspaceError(msg) raise WorkspaceError(msg)
check_command_allowed(command) check_command_allowed(command)
workdir = resolve_path(cwd) workdir = resolve_path(cwd, required=AccessMode.EXEC)
if not workdir.is_dir(): if not workdir.is_dir():
msg = f"cwd is not a directory: {cwd}" msg = f"cwd is not a directory: {cwd}"
raise WorkspaceError(msg) raise WorkspaceError(msg)
+76 -7
View File
@@ -1,13 +1,14 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass, field from dataclasses import dataclass, field
from enum import IntEnum
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING, cast from typing import TYPE_CHECKING, cast
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Callable, Sequence from collections.abc import Callable, Mapping, Sequence
from plyngent.tools.context import InstanceState from plyngent.tools.context import InstanceState, SessionState
DEFAULT_COMMAND_DENYLIST: frozenset[str] = frozenset( DEFAULT_COMMAND_DENYLIST: frozenset[str] = frozenset(
{ {
@@ -39,6 +40,36 @@ MAX_TEMPORARY_WORKSPACES = 16
# Timed human override for command denylist (independent of YOLO soft-confirm). # Timed human override for command denylist (independent of YOLO soft-confirm).
DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS = 30.0 DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS = 30.0
class AccessMode(IntEnum):
"""Ordered access level for path grants outside the workspace root.
A grant satisfies a requirement when its value is ``>=`` the requirement
(``read`` < ``write`` < ``exec``); ``exec`` also implies file writes.
"""
READ = 1
WRITE = 2
EXEC = 3
_ACCESS_MODE_NAMES: dict[str, AccessMode] = {
"read": AccessMode.READ,
"write": AccessMode.WRITE,
"exec": AccessMode.EXEC,
}
def parse_access_mode(value: str) -> AccessMode | None:
"""Parse a case-insensitive mode token (``read`` / ``write`` / ``exec``)."""
return _ACCESS_MODE_NAMES.get(value.strip().lower())
def mode_covers(granted: AccessMode, required: AccessMode) -> bool:
"""Whether a grant at *granted* satisfies a tool that needs *required*."""
return granted >= required
# Hook: (basename, argv, timeout_seconds) -> True allow for this session basename. # Hook: (basename, argv, timeout_seconds) -> True allow for this session basename.
type PolicyConfirmHook = Callable[[str, Sequence[str], float], bool] type PolicyConfirmHook = Callable[[str, Sequence[str], float], bool]
@@ -63,6 +94,10 @@ class WorkspacePolicy:
policy_allowed_commands: set[str] = field(default_factory=set) policy_allowed_commands: set[str] = field(default_factory=set)
policy_confirm_hook: PolicyConfirmHook | None = None policy_confirm_hook: PolicyConfirmHook | None = None
policy_confirm_timeout_seconds: float = DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS policy_confirm_timeout_seconds: float = DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS
# Static TOML pre-allow (resolved path → mode); never prompts.
config_allow: dict[Path, AccessMode] = field(default_factory=dict)
# Process-scoped YOLO / --yes grants (resolved path → mode); never persisted.
yolo_allow: dict[Path, AccessMode] = field(default_factory=dict)
def _bound_instance() -> InstanceState | None: def _bound_instance() -> InstanceState | None:
@@ -71,6 +106,12 @@ def _bound_instance() -> InstanceState | None:
return get_instance() return get_instance()
def _bound_session() -> SessionState | None:
from plyngent.tools.context import get_session
return get_session()
def require_bound_instance() -> InstanceState: def require_bound_instance() -> InstanceState:
"""Return the bound instance or raise :class:`WorkspaceError`.""" """Return the bound instance or raise :class:`WorkspaceError`."""
instance = _bound_instance() instance = _bound_instance()
@@ -243,11 +284,36 @@ def _under_any_root(resolved: Path, instance: InstanceState, policy: WorkspacePo
return False return False
def resolve_path(path: str | Path) -> Path: def _grants_cover(grants: Mapping[Path, AccessMode], resolved: Path, required: AccessMode) -> bool:
"""Resolve ``path`` under the workspace root or an allowlisted temp root. """Whether any granted root covers *resolved* at *required* mode.
A grant path covers itself and its subtree; a file grant therefore only
matches that exact file (a file has no children).
"""
for root, granted in grants.items():
if mode_covers(granted, required) and (resolved == root or resolved.is_relative_to(root)):
return True
return False
def _path_grant_covers(resolved: Path, required: AccessMode) -> bool:
"""Whether a config / session / process grant covers *resolved*."""
policy = active_workspace_policy()
if _grants_cover(policy.config_allow, resolved, required):
return True
if _grants_cover(policy.yolo_allow, resolved, required):
return True
session = _bound_session()
return session is not None and _grants_cover(session.access_grants, resolved, required)
def resolve_path(path: str | Path, *, required: AccessMode = AccessMode.WRITE) -> Path:
"""Resolve ``path`` under the workspace root, an allowlist root, or a grant.
Relative paths resolve against the **primary** workspace root. Absolute Relative paths resolve against the **primary** workspace root. Absolute
paths may also land under a temporary workspace allowlist entry. paths may also land under a temporary workspace allowlist entry or a
directory-access grant; ``required`` is the access mode the caller needs
(default :attr:`AccessMode.WRITE`, the conservative choice).
""" """
instance = require_bound_instance() instance = require_bound_instance()
policy = instance.workspace policy = instance.workspace
@@ -256,8 +322,11 @@ def resolve_path(path: str | Path) -> Path:
if not candidate.is_absolute(): if not candidate.is_absolute():
candidate = root / candidate candidate = root / candidate
resolved = candidate.expanduser().resolve() resolved = candidate.expanduser().resolve()
if not _under_any_root(resolved, instance, policy): if not _under_any_root(resolved, instance, policy) and not _path_grant_covers(resolved, required):
msg = f"path escapes workspace root ({root}): {path}" msg = (
f"path escapes workspace root ({root}): {path}; "
"call request_directory_access to request access to a directory outside the workspace"
)
raise WorkspaceError(msg) raise WorkspaceError(msg)
# Normalize separators so denylist entries like ``/secrets/`` match on Windows. # Normalize separators so denylist entries like ``/secrets/`` match on Windows.
resolved_str = str(resolved).replace("\\", "/") resolved_str = str(resolved).replace("\\", "/")
+88
View File
@@ -3,13 +3,18 @@ from __future__ import annotations
import pytest import pytest
from plyngent.tools import ( from plyngent.tools import (
AccessMode,
WorkspaceError, WorkspaceError,
active_workspace_policy,
check_command_allowed, check_command_allowed,
get_workspace_root, get_workspace_root,
mode_covers,
parse_access_mode,
resolve_path, resolve_path,
set_command_denylist, set_command_denylist,
set_path_denylist, set_path_denylist,
) )
from plyngent.tools.context import SessionState, bind_session
def test_resolve_relative_and_absolute(workspace: object) -> None: def test_resolve_relative_and_absolute(workspace: object) -> None:
@@ -27,6 +32,89 @@ def test_escape_rejected(workspace: object) -> None:
_ = resolve_path("../outside") _ = resolve_path("../outside")
def test_escape_error_hints_request_tool(workspace: object) -> None:
del workspace
with pytest.raises(WorkspaceError, match="request_directory_access"):
_ = resolve_path("../outside")
def test_access_mode_parse_and_covers() -> None:
assert parse_access_mode(" Read ") is AccessMode.READ
assert parse_access_mode("EXEC") is AccessMode.EXEC
assert parse_access_mode("bogus") is None
assert mode_covers(AccessMode.EXEC, AccessMode.WRITE)
assert mode_covers(AccessMode.WRITE, AccessMode.WRITE)
assert not mode_covers(AccessMode.READ, AccessMode.WRITE)
def test_config_grant_modes(workspace: object, tmp_path_factory: pytest.TempPathFactory) -> None:
del workspace
outside = tmp_path_factory.mktemp("granted")
target = outside / "data.txt"
_ = target.write_text("x", encoding="utf-8")
policy = active_workspace_policy()
policy.config_allow[outside] = AccessMode.READ
try:
assert resolve_path(str(target), required=AccessMode.READ) == target
with pytest.raises(WorkspaceError, match="escapes"):
_ = resolve_path(str(target), required=AccessMode.WRITE)
policy.config_allow[outside] = AccessMode.WRITE
assert resolve_path(str(target), required=AccessMode.WRITE) == target
with pytest.raises(WorkspaceError, match="escapes"):
_ = resolve_path(str(target), required=AccessMode.EXEC)
finally:
policy.config_allow.clear()
def test_session_grant_covers_subtree(workspace: object, tmp_path_factory: pytest.TempPathFactory) -> None:
del workspace
outside = tmp_path_factory.mktemp("session")
nested = outside / "a" / "b"
nested.mkdir(parents=True)
with bind_session(SessionState(access_grants={outside: AccessMode.EXEC})):
assert resolve_path(str(nested), required=AccessMode.EXEC) == nested
with pytest.raises(WorkspaceError, match="escapes"):
_ = resolve_path(str(nested), required=AccessMode.READ)
def test_file_grant_is_exact_path(workspace: object, tmp_path_factory: pytest.TempPathFactory) -> None:
del workspace
outside = tmp_path_factory.mktemp("file-grant")
granted = outside / "one.txt"
sibling = outside / "two.txt"
_ = granted.write_text("1", encoding="utf-8")
_ = sibling.write_text("2", encoding="utf-8")
policy = active_workspace_policy()
policy.config_allow[granted] = AccessMode.READ
try:
assert resolve_path(str(granted), required=AccessMode.READ) == granted
with pytest.raises(WorkspaceError, match="escapes"):
_ = resolve_path(str(sibling), required=AccessMode.READ)
# A file grant does not cover its parent directory.
with pytest.raises(WorkspaceError, match="escapes"):
_ = resolve_path(str(outside), required=AccessMode.READ)
finally:
policy.config_allow.clear()
def test_denylist_wins_over_grant(workspace: object, tmp_path_factory: pytest.TempPathFactory) -> None:
del workspace
outside = tmp_path_factory.mktemp("denied")
secret = outside / "secret"
secret.mkdir()
key = secret / "key"
_ = key.write_text("k", encoding="utf-8")
policy = active_workspace_policy()
policy.config_allow[outside] = AccessMode.WRITE
set_path_denylist(["/secret/"])
try:
with pytest.raises(WorkspaceError, match="matched '/secret/'"):
_ = resolve_path(str(key), required=AccessMode.READ)
finally:
set_path_denylist(None)
policy.config_allow.clear()
def test_path_denylist(workspace: object) -> None: def test_path_denylist(workspace: object) -> None:
from pathlib import Path from pathlib import Path