From a1f106a5e221cc655fcda052fca821539e00e14f Mon Sep 17 00:00:00 2001 From: worldmozara Date: Wed, 9 Sep 2026 22:23:25 +0800 Subject: [PATCH] cli: directory access confirm, /grants, and session grants --- src/plyngent/cli/app.py | 8 ++ src/plyngent/cli/limits.py | 59 +++++++++++ src/plyngent/cli/slash.py | 66 +++++++++++++ src/plyngent/cli/state.py | 90 ++++++++++++++++- src/plyngent/tools/__init__.py | 2 + src/plyngent/tools/access.py | 65 ++++++++++++- src/plyngent/tools/context.py | 2 + tests/test_cli/test_grants_cmd.py | 156 ++++++++++++++++++++++++++++++ tests/test_cli/test_limits.py | 36 +++++++ 9 files changed, 477 insertions(+), 7 deletions(-) create mode 100644 tests/test_cli/test_grants_cmd.py diff --git a/src/plyngent/cli/app.py b/src/plyngent/cli/app.py index 9bf3fc1..c9fa97e 100644 --- a/src/plyngent/cli/app.py +++ b/src/plyngent/cli/app.py @@ -333,6 +333,10 @@ async def _run_chat( # noqa: C901, PLR0912, PLR0915 — chat orchestration prompt_policy_fetch_confirm, instance=state.instance_state, ) + from plyngent.tools.access import set_directory_access_confirm_hook + + # Installed for one-shot too: the hook itself honours --yes / non-TTY. + set_directory_access_confirm_hook(state.directory_access_confirm_hook, instance=state.instance_state) # Seed cache if we already fetched; else warm in background for Tab. if remote_ids is not None: state.seed_remote_models(remote_ids) @@ -373,6 +377,10 @@ async def _run_chat( # noqa: C901, PLR0912, PLR0915 — chat orchestration set_fetch_policy_confirm_hook(None, instance=state_obj.instance_state) clear_private_grants(instance=state_obj.instance_state) clear_ssrf_assume_public_cidrs() + from plyngent.tools.access import clear_process_access, set_directory_access_confirm_hook + + set_directory_access_confirm_hook(None, instance=state_obj.instance_state) + clear_process_access(instance=state_obj.instance_state) if state_obj.mcp_manager is not None: await state_obj.mcp_manager.aclose() await state_obj.instance_state.shutdown() diff --git a/src/plyngent/cli/limits.py b/src/plyngent/cli/limits.py index 649d2fd..6cecda8 100644 --- a/src/plyngent/cli/limits.py +++ b/src/plyngent/cli/limits.py @@ -18,10 +18,13 @@ from plyngent.prompting import ( confirm, get_prompt_backend, ) +from plyngent.tools.access import AccessDecision from plyngent.tools.process.pty_session import PtyManager +from plyngent.tools.workspace import AccessMode, parse_access_mode if TYPE_CHECKING: from collections.abc import Mapping, Sequence + from pathlib import Path type WorkspaceMismatchChoice = Literal["keep", "rebind", "abort"] _BOX_MIN_WIDTH = 40 @@ -287,6 +290,62 @@ def prompt_policy_fetch_confirm( return False +def prompt_directory_access_confirm( + path: Path, + mode: AccessMode, + reason: str, + timeout_seconds: float, +) -> AccessDecision | str | None: + """Ask whether to grant access to *path* at *mode* (timed; default deny). + + ``y`` approves the requested level; ``read`` / ``write`` / ``exec`` approve + that level instead. The host's hook decides YOLO/``--yes`` before calling + this. Timeout, cancel, or non-interactive → ``None`` (deny). + """ + lines = [ + f"request: {mode.name.lower()} access to {path}", + f"kind: {'directory' if path.is_dir() else 'file'}", + ] + if reason: + lines.append(f"why: {reason}") + lines.append(f"(timeout {timeout_seconds:g}s defaults to DENY)") + try: + with pause_task_cancel_for_prompt(): + backend = get_prompt_backend() + if not backend.is_interactive(): + return None + backend.echo() + backend.secho( + format_tool_confirm_box("request_directory_access", "\n".join(lines)), + fg="yellow", + ) + backend.echo() + prompt = ( + f"[access] allow {path} ({mode.name.lower()})? [y/N|read|write|exec] (timeout {timeout_seconds:g}s): " + ) + with contextlib.suppress(OSError): + _ = sys.stderr.write(prompt) + _ = sys.stderr.flush() + raw = _read_yes_no_line_with_timeout(timeout_seconds) + if raw is None: + with contextlib.suppress(OSError, NonInteractiveError): + backend.secho( + f"[access] timed out after {timeout_seconds:g}s — denied", + fg="red", + err=True, + ) + return None + token = raw.strip().lower() + if token in {"y", "yes"}: + return AccessDecision(mode) + level = parse_access_mode(token) + if level is not None: + return AccessDecision(level) + return None + except NonInteractiveError, KeyboardInterrupt, EOFError: + return None + + async def prompt_confirm_tool_async(name: str, args: Mapping[str, object], reason: str) -> bool | str: """Async confirm: True allow, False deny, str = deny with user comment.""" del args diff --git a/src/plyngent/cli/slash.py b/src/plyngent/cli/slash.py index 7f980a7..3ca184a 100644 --- a/src/plyngent/cli/slash.py +++ b/src/plyngent/cli/slash.py @@ -521,6 +521,7 @@ def config_cmd(state: ReplState) -> None: def status_cmd(state: ReplState) -> None: """Show session/provider/tools/rounds status.""" from plyngent.agent.budget import estimate_messages_chars + from plyngent.tools.access import access_grant_rows pending = state.agent.pending_retry_text pending_disp = "yes" if pending else "no" @@ -555,6 +556,7 @@ def status_cmd(state: ReplState) -> None: f"usage_last_turn={last_u.format_line(billed=True)} " f"rounds={last_rounds}\n" f"usage_session={session_u.format_line(billed=True)}\n" + f"grants={len(access_grant_rows(instance=state.instance_state, session=state.session_state))} " f"workspace={state.workspace}" ) @@ -1205,6 +1207,70 @@ def mcp_slash_cmd(state: ReplState, action: str | None) -> None: _print_mcp_status(state) +@slash.command("grants") +@click.argument("action", required=False, type=click.Choice(["list", "revoke"])) +@click.argument("target", required=False) +@click.pass_obj +def grants_cmd(state: ReplState, action: str | None, target: str | None) -> None: + """List or revoke directory-access grants. + + ``/grants`` or ``/grants list`` — session, process, then config grants. + ``/grants revoke `` — drop one grant or all session/process + grants. Config grants come from ``[agent].allow_paths``; edit the TOML. + """ + from plyngent.tools.access import ( + access_grant_rows, + clear_process_access, + clear_session_access, + revoke_access, + ) + + act = (action or "list").lower() + if act == "list": + rows = access_grant_rows(instance=state.instance_state, session=state.session_state) + if not rows: + click.echo("no directory-access grants") + return + click.echo(f"directory-access grants ({len(rows)}):") + for index, (path, mode, source) in enumerate(rows): + click.echo(f" {index}\t{path}\t{mode.name.lower()}\t{source}") + click.echo("revoke with /grants revoke or /grants revoke all") + return + + if target is None or not target.strip(): + msg = "/grants revoke requires an index or 'all'" + raise click.UsageError(msg) + token = target.strip().lower() + if token == "all": + clear_session_access(session=state.session_state) + clear_process_access(instance=state.instance_state) + _await(state.persist_access_grants()) + click.echo("revoked all session/process access grants (config grants unchanged)") + return + if not token.isdigit(): + msg = "/grants revoke expects an index or 'all'" + raise click.UsageError(msg) + rows = access_grant_rows(instance=state.instance_state, session=state.session_state) + index = int(token) + if index < 0 or index >= len(rows): + msg = f"no grant at index {index}" + raise click.UsageError(msg) + path, _mode, source = rows[index] + if source == "config": + click.echo(f"config grant {path} is not revocable here; edit [agent].allow_paths") + return + if revoke_access( + path, + source=source, + instance=state.instance_state, + session=state.session_state, + ): + _await(state.persist_access_grants()) + click.echo(f"revoked {path} ({source})") + else: + click.echo(f"grant not found: {path}") + + @slash.command("yolo") @click.argument("mode", required=False, type=YOLO_MODE, metavar="[on|off|once]") @click.pass_obj diff --git a/src/plyngent/cli/state.py b/src/plyngent/cli/state.py index 229ac16..d26cb79 100644 --- a/src/plyngent/cli/state.py +++ b/src/plyngent/cli/state.py @@ -26,17 +26,31 @@ from plyngent.tools import InstanceState, SessionState from plyngent.tools.view import MemoryViewStore, session_data_view if TYPE_CHECKING: - from collections.abc import Sequence + from collections.abc import Mapping, Sequence from plyngent.config.models import Provider from plyngent.config.store import ConfigStore from plyngent.memory import MemoryStore from plyngent.memory.database.schema import Session as SessionRow from plyngent.runtime.mcp_client import McpManager + from plyngent.tools.access import AccessDecision + from plyngent.tools.workspace import AccessMode type YoloMode = Literal["off", "on", "once"] +def _parsed_access_grants(raw: Mapping[str, str]) -> dict[Path, AccessMode]: + """Parse stored ``{path: mode}`` grants into resolved keys (skip bad modes).""" + from plyngent.tools.workspace import parse_access_mode + + grants: dict[Path, AccessMode] = {} + for path_str, mode_str in raw.items(): + mode = parse_access_mode(mode_str) + if mode is not None: + grants[Path(path_str).expanduser().resolve()] = mode + return grants + + @dataclass class ReplState: """Mutable REPL session state.""" @@ -82,6 +96,7 @@ class ReplState: self.workspace = Path(self.workspace).expanduser().resolve() self.instance_state.workspace_root = self.workspace self.instance_state.workspace.root = self.workspace + self._install_config_access() self.session_state = self._session_data_for_todo() self.agent = self._make_agent() self.sync_display_flags() @@ -148,11 +163,17 @@ class ReplState: self._todo_persist_tasks.add(task) task.add_done_callback(self._todo_persist_tasks.discard) - def _session_data_for_todo(self, *, grants: dict[str, bool] | None = None) -> SessionState: + def _session_data_for_todo( + self, + *, + grants: dict[str, bool] | None = None, + access_grants: dict[Path, AccessMode] | None = None, + ) -> SessionState: """Build a SessionState with live todo + MemoryViewStore seed (durable tree). *grants* defaults to empty (new session / startup). Pass the previous map only when intentionally preserving soft-confirm trust across rebinds. + *access_grants* seeds session directory grants (empty clears them). """ grant_map = dict(grants or {}) store = MemoryViewStore({"todo": self.todo_stack.to_raw(), "grants": dict(grant_map)}) @@ -161,7 +182,9 @@ class ReplState: data=session_data_view(store=store), todo=self.todo_stack, on_todo_change=self._todo_on_change, + on_access_change=self._on_access_change, grants=grant_map, + access_grants=dict(access_grants or {}), ) def _bind_todo_tools(self) -> None: @@ -203,6 +226,67 @@ class ReplState: if hasattr(self, "agent"): self.agent.todo_stack = self.todo_stack + def _on_access_change(self) -> None: + """Schedule memory persist for live directory-access grants.""" + import asyncio + + if self.session_id is None: + return + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return + task = loop.create_task(self.persist_access_grants()) + self._todo_persist_tasks.add(task) + task.add_done_callback(self._todo_persist_tasks.discard) + + async def persist_access_grants(self) -> None: + """Write live session directory grants to the active session row.""" + if self.session_id is None: + return + payload = {str(path): mode.name.lower() for path, mode in self.session_state.access_grants.items()} + _ = await self.memory.update_session_access_grants(self.session_id, payload) + + async def load_access_grants(self) -> None: + """Load durable session directory grants into the live map (replace).""" + self.session_state.access_grants.clear() + if self.session_id is None: + return + raw = await self.memory.get_session_access_grants(self.session_id) + self.session_state.access_grants.update(_parsed_access_grants(raw)) + + def directory_access_confirm_hook( + self, + path: Path, + mode: AccessMode, + reason: str, + timeout_seconds: float, + ) -> AccessDecision | str | None: + """Directory-access confirm: YOLO/``--yes`` auto-approves, else ask.""" + from plyngent.cli.limits import prompt_directory_access_confirm + from plyngent.tools.access import AccessDecision as Decision + + if self.effective_yolo() != "off": + return Decision(mode, persist=False) + return prompt_directory_access_confirm(path, mode, reason, timeout_seconds) + + def _install_config_access(self) -> None: + """Install ``[agent].allow_paths`` static grants on the policy bag.""" + import click + + from plyngent.tools.access import set_config_access + + _applied, skipped = set_config_access( + self.config.agent_config.allow_paths, + instance=self.instance_state, + ) + if skipped: + click.secho( + f"warning: ignored allow_paths entries (missing path or bad mode): {', '.join(skipped)}", + fg="yellow", + err=True, + ) + def _tool_registry(self) -> ToolRegistry | None: if not self.tools_enabled: return None @@ -645,6 +729,7 @@ class ReplState: self.agent = self._make_agent() await self.agent.load_history() await self.load_todo_stack() + await self.load_access_grants() async def resume_latest_or_new(self, name: str = "chat") -> str: """Resume most recently updated session for this workspace, or create one.""" @@ -660,6 +745,7 @@ class ReplState: self.agent = self._make_agent() await self.agent.load_history() await self.load_todo_stack() + await self.load_access_grants() _ = await self.memory.touch_session(latest.sid) return "resume" diff --git a/src/plyngent/tools/__init__.py b/src/plyngent/tools/__init__.py index 912f6b1..43fc161 100644 --- a/src/plyngent/tools/__init__.py +++ b/src/plyngent/tools/__init__.py @@ -1,6 +1,7 @@ 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 access_grant_rows as access_grant_rows 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 @@ -8,6 +9,7 @@ from .access import get_directory_access_confirm_hook as get_directory_access_co 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 revoke_access as revoke_access from .access import set_config_access as set_config_access from .access import set_directory_access_confirm_hook as set_directory_access_confirm_hook from .catalog import ToolCatalog as ToolCatalog diff --git a/src/plyngent/tools/access.py b/src/plyngent/tools/access.py index 505af0a..e67fd05 100644 --- a/src/plyngent/tools/access.py +++ b/src/plyngent/tools/access.py @@ -80,6 +80,12 @@ def set_directory_access_confirm_hook( inst.extras[_HOOK_KEY] = hook +def _notify_access_change(session: SessionState | None) -> None: + """Fire the host's persist hook after the live session map changes.""" + if session is not None and session.on_access_change is not None: + session.on_access_change() + + def grant_session_access( path: Path | str, mode: AccessMode, @@ -90,6 +96,7 @@ def grant_session_access( resolved = Path(path).expanduser().resolve() sess = session if session is not None else require_session() sess.access_grants[resolved] = mode + _notify_access_change(sess) return resolved @@ -100,9 +107,10 @@ def grant_process_access(path: Path | str, mode: AccessMode) -> Path: return resolved -def clear_process_access() -> None: +def clear_process_access(*, instance: InstanceState | None = None) -> None: """Drop process-scoped (``--yes``) grants.""" - require_bound_instance().workspace.yolo_allow.clear() + inst = instance if instance is not None else require_bound_instance() + inst.workspace.yolo_allow.clear() def clear_session_access(*, session: SessionState | None = None) -> None: @@ -110,6 +118,7 @@ def clear_session_access(*, session: SessionState | None = None) -> None: sess = session if session is not None else get_session() if sess is not None: sess.access_grants.clear() + _notify_access_change(sess) def active_grant_count() -> int: @@ -122,14 +131,19 @@ def active_grant_count() -> int: return total -def set_config_access(entries: Mapping[str, str]) -> tuple[list[Path], list[str]]: - """Install static TOML pre-allow entries on the bound instance policy. +def set_config_access( + entries: Mapping[str, str], + *, + instance: InstanceState | None = None, +) -> tuple[list[Path], list[str]]: + """Install static TOML pre-allow entries on the instance policy bag. Replaces any previous config grants. Returns ``(applied roots, skipped keys)``; entries are skipped when the mode is unknown or the path does not exist (the host warns about skipped keys). """ - policy = require_bound_instance().workspace + inst = instance if instance is not None else require_bound_instance() + policy = inst.workspace policy.config_allow.clear() applied: list[Path] = [] skipped: list[str] = [] @@ -148,6 +162,47 @@ def set_config_access(entries: Mapping[str, str]) -> tuple[list[Path], list[str] return applied, skipped +def access_grant_rows( + *, + instance: InstanceState | None = None, + session: SessionState | None = None, +) -> list[tuple[Path, AccessMode, str]]: + """``(path, mode, source)`` rows: session, then process, then config.""" + inst = instance if instance is not None else require_bound_instance() + sess = session if session is not None else get_session() + rows: list[tuple[Path, AccessMode, str]] = [] + if sess is not None: + rows.extend((path, mode, "session") for path, mode in sess.access_grants.items()) + rows.extend((path, mode, "process") for path, mode in inst.workspace.yolo_allow.items()) + rows.extend((path, mode, "config") for path, mode in inst.workspace.config_allow.items()) + return rows + + +def revoke_access( + path: Path | str, + *, + source: str, + instance: InstanceState | None = None, + session: SessionState | None = None, +) -> bool: + """Remove the exact resolved *path* from one store; false if not present. + + ``config`` grants are owned by ``[agent].allow_paths`` and are not removed + here (the caller should point the human at the config file). + """ + resolved = Path(path).expanduser().resolve() + if source == "session": + sess = session if session is not None else get_session() + if sess is None or sess.access_grants.pop(resolved, None) is None: + return False + _notify_access_change(sess) + return True + if source == "process": + inst = instance if instance is not None else require_bound_instance() + return inst.workspace.yolo_allow.pop(resolved, None) is not None + return False + + def _resolve_target(path: str) -> Path | str: """Resolve a requested path; return a model-facing error string on failure.""" try: diff --git a/src/plyngent/tools/context.py b/src/plyngent/tools/context.py index 241820b..f0f6489 100644 --- a/src/plyngent/tools/context.py +++ b/src/plyngent/tools/context.py @@ -74,6 +74,8 @@ class SessionState: todo: TodoStack | None = None # Host hook after todo tools mutate (e.g. CLI memory persist). on_todo_change: Callable[[], None] | None = None + # Host hook after directory-access grants change (e.g. CLI memory persist). + on_access_change: Callable[[], None] | None = None # Soft-confirm grants live map: tool_name → True (Phase 1 key is tool name). # Durable copy lives under data["grants"] (see plyngent.tools.grants). grants: dict[str, bool] = field(default_factory=dict) diff --git a/tests/test_cli/test_grants_cmd.py b/tests/test_cli/test_grants_cmd.py new file mode 100644 index 0000000..fe7d9a6 --- /dev/null +++ b/tests/test_cli/test_grants_cmd.py @@ -0,0 +1,156 @@ +"""CLI slash ``/grants``, session hydrate/persist, and the access confirm hook.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +import tomlkit + +from plyngent.cli.slash import handle_slash +from plyngent.cli.state import ReplState +from plyngent.config.models import DatabaseConfig, OpenAIProvider +from plyngent.config.store import ConfigStore +from plyngent.memory import MemoryStore +from plyngent.prompting import NonInteractiveBackend, temporary_backend +from plyngent.tools import AccessMode, grant_session_access +from plyngent.tools.access import AccessDecision + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + from pathlib import Path + + +async def _new_state( + tmp_path: Path, + *, + document: object | None = None, +) -> tuple[ReplState, MemoryStore]: + memory = await MemoryStore.open(DatabaseConfig()) + provider = OpenAIProvider(access_key_or_token="sk-test") + config = ConfigStore( + path=tmp_path / "plyngent.toml", + document=tomlkit.document() if document is None else document, + ) + config.providers = {"local": provider} + state = ReplState( + config=config, + memory=memory, + workspace=tmp_path, + provider_name="local", + provider=provider, + model="gpt-test", + tools_enabled=False, + ) + await state.new_session("t") + return state, memory + + +@pytest.fixture +async def state(tmp_path: Path) -> AsyncIterator[ReplState]: + st, memory = await _new_state(tmp_path) + yield st + await memory.close() + + +async def test_grants_list_empty(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: + assert await handle_slash(state, "/grants") is True + assert "no directory-access grants" in capsys.readouterr().out + + +async def test_grants_list_and_revoke_session( + state: ReplState, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + outside = (tmp_path.parent / "outside-grant").resolve() + outside.mkdir(exist_ok=True) + _ = grant_session_access(outside, AccessMode.READ, session=state.session_state) + state.instance_state.workspace.yolo_allow[outside] = AccessMode.EXEC + state.instance_state.workspace.config_allow[tmp_path] = AccessMode.READ + + assert await handle_slash(state, "/grants list") is True + out = capsys.readouterr().out + assert "directory-access grants (3)" in out + assert "session" in out + assert "process" in out + assert "config" in out + + # Rows are ordered session, process, config. + assert await handle_slash(state, "/grants revoke 0") is True + assert "revoked" in capsys.readouterr().out + assert state.session_state.access_grants == {} + assert await state.memory.get_session_access_grants(state.session_id) == {} + + +async def test_grants_revoke_all_keeps_config( + state: ReplState, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + outside = (tmp_path.parent / "outside-all").resolve() + outside.mkdir(exist_ok=True) + _ = grant_session_access(outside, AccessMode.READ, session=state.session_state) + state.instance_state.workspace.yolo_allow[outside] = AccessMode.WRITE + state.instance_state.workspace.config_allow[tmp_path] = AccessMode.READ + + assert await handle_slash(state, "/grants revoke all") is True + assert "config grants unchanged" in capsys.readouterr().out + assert state.session_state.access_grants == {} + assert state.instance_state.workspace.yolo_allow == {} + assert state.instance_state.workspace.config_allow == {tmp_path: AccessMode.READ} + + +async def test_grants_revoke_config_points_at_toml( + state: ReplState, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + state.instance_state.workspace.config_allow[tmp_path] = AccessMode.READ + assert await handle_slash(state, "/grants revoke 0") is True + assert "edit [agent].allow_paths" in capsys.readouterr().out + assert state.instance_state.workspace.config_allow == {tmp_path: AccessMode.READ} + + +async def test_grants_revoke_bad_index( + state: ReplState, + capsys: pytest.CaptureFixture[str], +) -> None: + assert await handle_slash(state, "/grants revoke 99") is True + assert "no grant at index 99" in capsys.readouterr().err + + +async def test_persist_and_load_roundtrip(state: ReplState, tmp_path: Path) -> None: + outside = (tmp_path.parent / "outside-hydrate").resolve() + outside.mkdir(exist_ok=True) + _ = grant_session_access(outside, AccessMode.WRITE, session=state.session_state) + await state.persist_access_grants() + state.session_state.access_grants.clear() + await state.load_access_grants() + assert state.session_state.access_grants == {outside: AccessMode.WRITE} + + +def test_yolo_hook_auto_approves_as_process_grant(state: ReplState, tmp_path: Path) -> None: + state.set_yolo("on") + decision = state.directory_access_confirm_hook(tmp_path, AccessMode.EXEC, "", 30.0) + assert decision == AccessDecision(AccessMode.EXEC, persist=False) + + +def test_default_hook_denies_noninteractive(state: ReplState, tmp_path: Path) -> None: + with temporary_backend(NonInteractiveBackend()): + assert state.directory_access_confirm_hook(tmp_path, AccessMode.READ, "", 1.0) is None + + +async def test_config_allow_paths_installed( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + outside = (tmp_path.parent / "outside-config").resolve() + outside.mkdir(exist_ok=True) + document = tomlkit.parse(f'[agent]\nallow_paths = {{ "{outside}" = "read", "/does/not/exist" = "read" }}\n') + state, memory = await _new_state(tmp_path, document=document) + try: + assert state.instance_state.workspace.config_allow == {outside: AccessMode.READ} + assert "ignored allow_paths entries" in capsys.readouterr().err + finally: + await memory.close() diff --git a/tests/test_cli/test_limits.py b/tests/test_cli/test_limits.py index 1125be8..61b137c 100644 --- a/tests/test_cli/test_limits.py +++ b/tests/test_cli/test_limits.py @@ -1,5 +1,7 @@ from __future__ import annotations +import pytest + from plyngent.cli.limits import ( format_tool_confirm_box, install_cli_limit_hooks, @@ -54,3 +56,37 @@ def test_policy_confirm_noninteractive_denies() -> None: with temporary_backend(NonInteractiveBackend()): assert prompt_policy_command_confirm("sudo", ["sudo", "id"], 1.0) is False + + +def test_directory_access_prompt_noninteractive_denies() -> None: + from pathlib import Path + + from plyngent.cli.limits import prompt_directory_access_confirm + from plyngent.prompting import NonInteractiveBackend + from plyngent.tools.workspace import AccessMode + + with temporary_backend(NonInteractiveBackend()): + assert prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 1.0) is None + + +def test_directory_access_prompt_yes_and_level(monkeypatch: pytest.MonkeyPatch) -> None: + from pathlib import Path + + from plyngent.cli import limits + from plyngent.tools.access import AccessDecision + from plyngent.tools.workspace import AccessMode + + target = Path("/tmp") + with temporary_backend(ScriptedBackend([])): + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: "y\n") + assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) == AccessDecision( + AccessMode.READ + ) + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: "exec\n") + assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) == AccessDecision( + AccessMode.EXEC + ) + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: "n\n") + assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) is None + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: None) + assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) is None