From 49d1e08ebe931ac56994bd1babc8262dea36ad6b Mon Sep 17 00:00:00 2001 From: worldmozara Date: Wed, 9 Sep 2026 22:16:22 +0800 Subject: [PATCH] tools: request_directory_access and grant stores --- src/plyngent/config/models.py | 2 + src/plyngent/tools/__init__.py | 14 ++ src/plyngent/tools/access.py | 205 +++++++++++++++++++++++++++ src/plyngent/tools/catalog.py | 2 + src/plyngent/tools/workspace.py | 62 +++++--- tests/test_tools/test_access.py | 236 +++++++++++++++++++++++++++++++ tests/test_tools/test_catalog.py | 3 +- 7 files changed, 501 insertions(+), 23 deletions(-) create mode 100644 src/plyngent/tools/access.py create mode 100644 tests/test_tools/test_access.py diff --git a/src/plyngent/config/models.py b/src/plyngent/config/models.py index ffb8d83..a5dc617 100644 --- a/src/plyngent/config/models.py +++ b/src/plyngent/config/models.py @@ -23,6 +23,8 @@ DEFAULT_TOOL_DIRECTIVES = """\ ### Workspace - Explore with `tree`/`listdir`/`glob_paths`/`grep_files`/`read_file` when unsure. - Stay under the workspace (or a path from `new_temporary_workspace`). Respect path denylists. +- Need a path outside the workspace? Call `request_directory_access` \ +(human approves; denylists apply) instead of retrying an escaped path. ### Files - Prefer file tools over shell. `edit_replace` fails usually means a bad match — \ diff --git a/src/plyngent/tools/__init__.py b/src/plyngent/tools/__init__.py index 37cae6d..f87a669 100644 --- a/src/plyngent/tools/__init__.py +++ b/src/plyngent/tools/__init__.py @@ -1,3 +1,14 @@ +from .access import ACCESS_TOOLS as ACCESS_TOOLS +from .access import MAX_ACCESS_GRANTS as MAX_ACCESS_GRANTS +from .access import AccessDecision as AccessDecision +from .access import active_grant_count as active_grant_count +from .access import clear_process_access as clear_process_access +from .access import clear_session_access as clear_session_access +from .access import get_directory_access_confirm_hook as get_directory_access_confirm_hook +from .access import grant_process_access as grant_process_access +from .access import grant_session_access as grant_session_access +from .access import request_directory_access as request_directory_access +from .access import set_directory_access_confirm_hook as set_directory_access_confirm_hook from .catalog import ToolCatalog as ToolCatalog from .catalog import ToolSource as ToolSource from .catalog import catalog_scope as catalog_scope @@ -75,11 +86,13 @@ from .workspace import check_command_allowed as check_command_allowed from .workspace import clear_policy_allowed_commands as clear_policy_allowed_commands from .workspace import clear_workspace_allowlist as clear_workspace_allowlist from .workspace import clear_workspace_root as clear_workspace_root +from .workspace import denylist_match as denylist_match from .workspace import get_command_denylist as get_command_denylist from .workspace import get_path_denylist as get_path_denylist 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 grant_policy_command as grant_policy_command +from .workspace import granted_access_mode as granted_access_mode 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 @@ -91,3 +104,4 @@ from .workspace import set_path_denylist as set_path_denylist from .workspace import set_policy_confirm_hook as set_policy_confirm_hook from .workspace import set_policy_confirm_timeout as set_policy_confirm_timeout from .workspace import set_workspace_root as set_workspace_root +from .workspace import within_workspace_roots as within_workspace_roots diff --git a/src/plyngent/tools/access.py b/src/plyngent/tools/access.py new file mode 100644 index 0000000..5cd2023 --- /dev/null +++ b/src/plyngent/tools/access.py @@ -0,0 +1,205 @@ +"""Directory-access grants: request tool plus config / session / process stores. + +A grant is one ``path → mode`` pair; its *source* is structural: + +- :attr:`WorkspacePolicy.config_allow` — static TOML pre-allow (never prompts), +- :attr:`SessionState.access_grants` — human-approved for this session, +- :attr:`WorkspacePolicy.yolo_allow` — ``--yes`` / YOLO auto-approval, process only. + +``resolve_path`` consults all three. Grants are a policy gate for path-resolving +tools, **not** a sandbox: an ``exec`` grant (or any command) can reach the rest +of the filesystem. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING, NamedTuple + +from plyngent.agent import ToolTag, tool +from plyngent.tools.context import get_session, require_session +from plyngent.tools.workspace import ( + AccessMode, + denylist_match, + get_policy_confirm_timeout, + granted_access_mode, + mode_covers, + parse_access_mode, + require_bound_instance, + within_workspace_roots, +) + +if TYPE_CHECKING: + from collections.abc import Callable + + from plyngent.tools.context import InstanceState, SessionState + +# Max active grants across all three stores (bounds prompt/listing noise). +MAX_ACCESS_GRANTS = 32 + +_HOOK_KEY = "directory_access_confirm_hook" + + +class AccessDecision(NamedTuple): + """Hook approval for one directory-access request. + + ``mode`` is the granted level (may differ from the requested one); + ``persist`` is true for a session grant, false for a process-only grant. + """ + + mode: AccessMode + persist: bool = True + + +type DirectoryAccessConfirmHook = Callable[[Path, AccessMode, str, float], AccessDecision | str | None] + + +def get_directory_access_confirm_hook(instance: InstanceState | None = None) -> DirectoryAccessConfirmHook | None: + """Return the host-installed confirm hook for this instance, if any.""" + inst = instance if instance is not None else require_bound_instance() + hook = inst.extras.get(_HOOK_KEY) + if hook is None or not callable(hook): + return None + + def _as_hook(path: Path, mode: AccessMode, reason: str, timeout: float) -> AccessDecision | str | None: + result: object = hook(path, mode, reason, timeout) + if isinstance(result, AccessDecision | str): + return result + return None + + return _as_hook + + +def set_directory_access_confirm_hook( + hook: DirectoryAccessConfirmHook | None, + *, + instance: InstanceState | None = None, +) -> None: + """Install (or clear) the confirm hook the CLI uses to ask the human.""" + inst = instance if instance is not None else require_bound_instance() + inst.extras[_HOOK_KEY] = hook + + +def grant_session_access( + path: Path | str, + mode: AccessMode, + *, + session: SessionState | None = None, +) -> Path: + """Grant *mode* on *path* for the bound session; returns the resolved path.""" + resolved = Path(path).expanduser().resolve() + sess = session if session is not None else require_session() + sess.access_grants[resolved] = mode + return resolved + + +def grant_process_access(path: Path | str, mode: AccessMode) -> Path: + """Grant *mode* on *path* for this process only (``--yes``); never persisted.""" + resolved = Path(path).expanduser().resolve() + require_bound_instance().workspace.yolo_allow[resolved] = mode + return resolved + + +def clear_process_access() -> None: + """Drop process-scoped (``--yes``) grants.""" + require_bound_instance().workspace.yolo_allow.clear() + + +def clear_session_access(*, session: SessionState | None = None) -> None: + """Drop live session grants (durable copy is cleared by the host).""" + sess = session if session is not None else get_session() + if sess is not None: + sess.access_grants.clear() + + +def active_grant_count() -> int: + """Total live grants across config, session, and process stores.""" + policy = require_bound_instance().workspace + total = len(policy.config_allow) + len(policy.yolo_allow) + session = get_session() + if session is not None: + total += len(session.access_grants) + return total + + +def _resolve_target(path: str) -> Path | str: + """Resolve a requested path; return a model-facing error string on failure.""" + try: + target = Path(path).expanduser().resolve() + except OSError as exc: + return f"error: cannot resolve path {path!r}: {exc}" + if not target.exists(): + return f"error: path does not exist: {target}" + return target + + +def _precheck_target(target: Path, requested: AccessMode) -> str | None: + """Return a short-circuit message when no prompt is needed, else ``None``.""" + if requested is AccessMode.EXEC and not target.is_dir(): + return f"error: exec access requires a directory: {target}" + matched = denylist_match(target, directory=target.is_dir()) + if matched is not None: + return f"error: path denied by policy (matched {matched!r}); it can never be granted: {target}" + if within_workspace_roots(target): + return f"already accessible: {target} is inside the workspace" + granted = granted_access_mode(target) + if granted is not None and mode_covers(granted, requested): + return f"already granted: {target} ({granted.name.lower()})" + if active_grant_count() >= MAX_ACCESS_GRANTS: + return f"error: too many active access grants (max {MAX_ACCESS_GRANTS}); revoke some first" + return None + + +def _apply_decision(target: Path, decision: AccessDecision | str | None, timeout: float) -> str: + """Record an approved grant or format the denial for the model.""" + if isinstance(decision, AccessDecision): + if decision.persist and get_session() is not None: + _ = grant_session_access(target, decision.mode) + scope = "session" + else: + _ = grant_process_access(target, decision.mode) + scope = "process" + return ( + f"access granted: {target} ({decision.mode.name.lower()}, {scope})\n" + "note: the path denylist still applies; the human can revoke with /grants" + ) + if isinstance(decision, str) and decision.strip(): + return f"error: access to {target} denied by user; user comment: {decision.strip()}" + return f"error: access to {target} denied (declined or timed out after {timeout:g}s)" + + +@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.SESSION_STATE) +async def request_directory_access(path: str, reason: str = "", mode: str = "read") -> str: + """Request human approval to use a directory (or exact file) outside the workspace. + + Call this instead of retrying a path that failed with ``path escapes + workspace root``. ``mode`` is ``read`` (inspect files), ``write`` (also + modify files) or ``exec`` (also use it as ``run_argv`` / PTY cwd); the human + may approve a different level. Approval lasts for the session (or for this + process when ``--yes`` auto-approves). Paths matching the configured + denylist can never be granted. + """ + requested = parse_access_mode(mode) + if requested is None: + return f"error: invalid mode {mode!r} (use read, write, or exec)" + resolved = _resolve_target(path) + if isinstance(resolved, str): + return resolved + target = resolved + precheck = _precheck_target(target, requested) + if precheck is not None: + return precheck + hook = get_directory_access_confirm_hook() + if hook is None: + return f"error: access to {target} requires a human confirm (no confirm hook installed; denied)" + timeout = get_policy_confirm_timeout() + try: + decision = hook(target, requested, reason, timeout) + except Exception as exc: # noqa: BLE001 — surface confirm failures to the model + return f"error: access confirm for {target} failed: {exc}" + return _apply_decision(target, decision, timeout) + + +ACCESS_TOOLS = [ + request_directory_access, +] diff --git a/src/plyngent/tools/catalog.py b/src/plyngent/tools/catalog.py index 01d37dd..cce2bd4 100644 --- a/src/plyngent/tools/catalog.py +++ b/src/plyngent/tools/catalog.py @@ -218,6 +218,7 @@ def _ensure_builtin_definitions_registered(catalog: ToolCatalog) -> None: import time (usually the process catalog). Test ``catalog_scope(empty=True)`` overrides must still see builtins for ``default_tool_definitions``. """ + from plyngent.tools.access import ACCESS_TOOLS from plyngent.tools.chat import CHAT_TOOLS from plyngent.tools.file import FILE_TOOLS from plyngent.tools.net import NET_TOOLS @@ -226,6 +227,7 @@ def _ensure_builtin_definitions_registered(catalog: ToolCatalog) -> None: from plyngent.tools.vcs import VCS_TOOLS for definition in ( + *ACCESS_TOOLS, *FILE_TOOLS, *PROCESS_TOOLS, *VCS_TOOLS, diff --git a/src/plyngent/tools/workspace.py b/src/plyngent/tools/workspace.py index 19c1244..7e58dcb 100644 --- a/src/plyngent/tools/workspace.py +++ b/src/plyngent/tools/workspace.py @@ -284,27 +284,46 @@ def _under_any_root(resolved: Path, instance: InstanceState, policy: WorkspacePo return False -def _grants_cover(grants: Mapping[Path, AccessMode], resolved: Path, required: AccessMode) -> bool: - """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 granted_access_mode(resolved: Path) -> AccessMode | None: + """Highest config / session / process mode covering *resolved*, if any.""" + policy = active_workspace_policy() + session = _bound_session() + buckets: list[Mapping[Path, AccessMode]] = [policy.config_allow, policy.yolo_allow] + if session is not None: + buckets.append(session.access_grants) + best: AccessMode | None = None + for grants in buckets: + for root, granted in grants.items(): + if (resolved == root or resolved.is_relative_to(root)) and (best is None or granted > best): + best = granted + return best 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) + granted = granted_access_mode(resolved) + return granted is not None and mode_covers(granted, required) + + +def within_workspace_roots(resolved: Path) -> bool: + """Whether *resolved* is under the primary root(s) or a temp allowlist root.""" + instance = require_bound_instance() + return _under_any_root(resolved, instance, instance.workspace) + + +def denylist_match(resolved: Path, *, directory: bool = False) -> str | None: + """Return the path denylist pattern matching *resolved*, or ``None``. + + *directory* also matches a trailing-separator pattern (``/secret/``) against + the directory path itself, not just its children. + """ + resolved_str = str(resolved).replace("\\", "/") + if directory and not resolved_str.endswith("/"): + resolved_str += "/" + for pattern in active_workspace_policy().path_denylist: + if pattern and pattern.replace("\\", "/") in resolved_str: + return pattern + return None def resolve_path(path: str | Path, *, required: AccessMode = AccessMode.WRITE) -> Path: @@ -329,11 +348,10 @@ def resolve_path(path: str | Path, *, required: AccessMode = AccessMode.WRITE) - ) raise WorkspaceError(msg) # Normalize separators so denylist entries like ``/secrets/`` match on Windows. - resolved_str = str(resolved).replace("\\", "/") - for pattern in policy.path_denylist: - if pattern and pattern.replace("\\", "/") in resolved_str: - msg = f"path denied by policy (matched {pattern!r}): {path}" - raise WorkspaceError(msg) + matched = denylist_match(resolved) + if matched is not None: + msg = f"path denied by policy (matched {matched!r}): {path}" + raise WorkspaceError(msg) return resolved diff --git a/tests/test_tools/test_access.py b/tests/test_tools/test_access.py new file mode 100644 index 0000000..f9ca8d8 --- /dev/null +++ b/tests/test_tools/test_access.py @@ -0,0 +1,236 @@ +"""``request_directory_access`` tool and the grant stores.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from plyngent.agent import ToolTag +from plyngent.tools import ( + MAX_ACCESS_GRANTS, + AccessDecision, + AccessMode, + SessionState, + active_workspace_policy, + get_directory_access_confirm_hook, + grant_session_access, + request_directory_access, + resolve_path, + set_directory_access_confirm_hook, + set_path_denylist, +) +from plyngent.tools.context import bind_session +from tests.test_tools.helpers import call_async + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + + +@pytest.fixture +def session() -> Iterator[SessionState]: + state = SessionState() + with bind_session(state): + yield state + + +def _never_called(*_args: object) -> AccessDecision | str | None: + msg = "confirm hook should not be called" + raise AssertionError(msg) + + +def test_tool_tags_and_registry() -> None: + tags = request_directory_access.tags + assert tags & ToolTag.LOCAL + assert tags & ToolTag.INSTANCE_STATE + assert tags & ToolTag.SESSION_STATE + assert not (tags & ToolTag.READ_ONLY) + from plyngent.tools.catalog import default_tool_definitions + + assert "request_directory_access" in {definition.name for definition in default_tool_definitions()} + + +def test_hook_roundtrip(workspace: object, session: SessionState) -> None: + del workspace, session + assert get_directory_access_confirm_hook() is None + set_directory_access_confirm_hook(_never_called) + assert get_directory_access_confirm_hook() is not None + set_directory_access_confirm_hook(None) + assert get_directory_access_confirm_hook() is None + + +async def test_approve_grants_session_access( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace + outside = tmp_path_factory.mktemp("approved").resolve() + calls: list[tuple[Path, AccessMode, str, float]] = [] + + def hook(path: Path, mode: AccessMode, reason: str, timeout: float) -> AccessDecision: + calls.append((path, mode, reason, timeout)) + return AccessDecision(AccessMode.WRITE) + + set_directory_access_confirm_hook(hook) + out = await call_async(request_directory_access, str(outside), "need the dataset", "read") + assert "access granted" in out + assert "write, session" in out + assert calls == [(outside, AccessMode.READ, "need the dataset", 30.0)] + assert session.access_grants[outside] == AccessMode.WRITE + target = outside / "data.txt" + _ = target.write_text("x", encoding="utf-8") + assert resolve_path(str(target), required=AccessMode.WRITE) == target + + +async def test_process_grant_when_not_persisting( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace + outside = tmp_path_factory.mktemp("process").resolve() + set_directory_access_confirm_hook(lambda *_a: AccessDecision(AccessMode.READ, persist=False)) + out = await call_async(request_directory_access, str(outside), mode="read") + assert "read, process" in out + assert session.access_grants == {} + assert active_workspace_policy().yolo_allow[outside] == AccessMode.READ + + +async def test_deny_with_comment( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace + outside = tmp_path_factory.mktemp("denied").resolve() + set_directory_access_confirm_hook(lambda *_a: "no, that is my tax data") + out = await call_async(request_directory_access, str(outside)) + assert out.startswith("error: access to") + assert "user comment: no, that is my tax data" in out + assert session.access_grants == {} + + +async def test_deny_none( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace + outside = tmp_path_factory.mktemp("timeout").resolve() + set_directory_access_confirm_hook(lambda *_a: None) + out = await call_async(request_directory_access, str(outside)) + assert "denied (declined or timed out" in out + assert session.access_grants == {} + + +async def test_no_hook_denies( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace + outside = tmp_path_factory.mktemp("nohook").resolve() + set_directory_access_confirm_hook(None) + out = await call_async(request_directory_access, str(outside)) + assert "no confirm hook installed" in out + + +async def test_denylist_can_never_be_granted( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace + outside = tmp_path_factory.mktemp("secret-zone").resolve() + secret = outside / "secret" + secret.mkdir() + set_path_denylist(["/secret/"]) + set_directory_access_confirm_hook(_never_called) + try: + out = await call_async(request_directory_access, str(secret), mode="read") + assert "can never be granted" in out + assert session.access_grants == {} + finally: + set_path_denylist(None) + + +async def test_already_accessible_inside_workspace( + workspace: object, + session: SessionState, +) -> None: + from pathlib import Path + + assert isinstance(workspace, Path) + set_directory_access_confirm_hook(_never_called) + out = await call_async(request_directory_access, str(workspace), mode="read") + assert "already accessible" in out + assert session.access_grants == {} + + +async def test_already_granted_skips_prompt( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace + outside = tmp_path_factory.mktemp("granted").resolve() + _ = grant_session_access(outside, AccessMode.READ) + set_directory_access_confirm_hook(_never_called) + out = await call_async(request_directory_access, str(outside), mode="read") + assert "already granted" in out + + +async def test_invalid_mode( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace, session + outside = tmp_path_factory.mktemp("mode").resolve() + out = await call_async(request_directory_access, str(outside), mode="sudo") + assert "invalid mode" in out + + +async def test_exec_requires_directory( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace, session + outside = tmp_path_factory.mktemp("exec-file").resolve() + target = outside / "f.txt" + _ = target.write_text("x", encoding="utf-8") + out = await call_async(request_directory_access, str(target), mode="exec") + assert "exec access requires a directory" in out + + +async def test_missing_path( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace, session + missing = tmp_path_factory.mktemp("missing").resolve() / "nope" + out = await call_async(request_directory_access, str(missing)) + assert "does not exist" in out + + +async def test_grant_cap( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + del workspace + outside = tmp_path_factory.mktemp("cap").resolve() + policy = active_workspace_policy() + for index in range(MAX_ACCESS_GRANTS): + policy.config_allow[outside / f"d{index}"] = AccessMode.READ + set_directory_access_confirm_hook(_never_called) + try: + out = await call_async(request_directory_access, str(outside)) + assert "too many active access grants" in out + assert session.access_grants == {} + finally: + policy.config_allow.clear() diff --git a/tests/test_tools/test_catalog.py b/tests/test_tools/test_catalog.py index 939327b..b9ae5ae 100644 --- a/tests/test_tools/test_catalog.py +++ b/tests/test_tools/test_catalog.py @@ -6,6 +6,7 @@ import inspect from plyngent.agent import ToolRegistry, ToolTag, tool from plyngent.tools import default_tool_definitions, register_builtin_tools +from plyngent.tools.access import ACCESS_TOOLS from plyngent.tools.catalog import ToolCatalog, ToolSource, catalog_scope, get_catalog, registration_source from plyngent.tools.chat import CHAT_TOOLS from plyngent.tools.file import FILE_TOOLS @@ -18,7 +19,7 @@ from plyngent.tools.vcs import VCS_TOOLS def test_default_tool_names_match_group_lists() -> None: register_builtin_tools() selected = default_tool_definitions(surface="local") - groups = [*FILE_TOOLS, *PROCESS_TOOLS, *VCS_TOOLS, *CHAT_TOOLS, *TODO_TOOLS, *NET_TOOLS] + groups = [*ACCESS_TOOLS, *FILE_TOOLS, *PROCESS_TOOLS, *VCS_TOOLS, *CHAT_TOOLS, *TODO_TOOLS, *NET_TOOLS] assert sorted(t.name for t in selected) == sorted(t.name for t in groups) assert len(selected) == len(groups)