cli: directory access confirm, /grants, and session grants

This commit is contained in:
2026-09-09 22:23:25 +08:00
parent 22f52a0bd2
commit a1f106a5e2
9 changed files with 477 additions and 7 deletions
+8
View File
@@ -333,6 +333,10 @@ async def _run_chat( # noqa: C901, PLR0912, PLR0915 — chat orchestration
prompt_policy_fetch_confirm, prompt_policy_fetch_confirm,
instance=state.instance_state, 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. # Seed cache if we already fetched; else warm in background for Tab.
if remote_ids is not None: if remote_ids is not None:
state.seed_remote_models(remote_ids) 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) set_fetch_policy_confirm_hook(None, instance=state_obj.instance_state)
clear_private_grants(instance=state_obj.instance_state) clear_private_grants(instance=state_obj.instance_state)
clear_ssrf_assume_public_cidrs() 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: if state_obj.mcp_manager is not None:
await state_obj.mcp_manager.aclose() await state_obj.mcp_manager.aclose()
await state_obj.instance_state.shutdown() await state_obj.instance_state.shutdown()
+59
View File
@@ -18,10 +18,13 @@ from plyngent.prompting import (
confirm, confirm,
get_prompt_backend, get_prompt_backend,
) )
from plyngent.tools.access import AccessDecision
from plyngent.tools.process.pty_session import PtyManager from plyngent.tools.process.pty_session import PtyManager
from plyngent.tools.workspace import AccessMode, parse_access_mode
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Mapping, Sequence from collections.abc import Mapping, Sequence
from pathlib import Path
type WorkspaceMismatchChoice = Literal["keep", "rebind", "abort"] type WorkspaceMismatchChoice = Literal["keep", "rebind", "abort"]
_BOX_MIN_WIDTH = 40 _BOX_MIN_WIDTH = 40
@@ -287,6 +290,62 @@ def prompt_policy_fetch_confirm(
return False 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 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.""" """Async confirm: True allow, False deny, str = deny with user comment."""
del args del args
+66
View File
@@ -521,6 +521,7 @@ def config_cmd(state: ReplState) -> None:
def status_cmd(state: ReplState) -> None: def status_cmd(state: ReplState) -> None:
"""Show session/provider/tools/rounds status.""" """Show session/provider/tools/rounds status."""
from plyngent.agent.budget import estimate_messages_chars from plyngent.agent.budget import estimate_messages_chars
from plyngent.tools.access import access_grant_rows
pending = state.agent.pending_retry_text pending = state.agent.pending_retry_text
pending_disp = "yes" if pending else "no" 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"usage_last_turn={last_u.format_line(billed=True)} "
f"rounds={last_rounds}\n" f"rounds={last_rounds}\n"
f"usage_session={session_u.format_line(billed=True)}\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}" f"workspace={state.workspace}"
) )
@@ -1205,6 +1207,70 @@ def mcp_slash_cmd(state: ReplState, action: str | None) -> None:
_print_mcp_status(state) _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 <index|all>`` — 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 <index> 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") @slash.command("yolo")
@click.argument("mode", required=False, type=YOLO_MODE, metavar="[on|off|once]") @click.argument("mode", required=False, type=YOLO_MODE, metavar="[on|off|once]")
@click.pass_obj @click.pass_obj
+88 -2
View File
@@ -26,17 +26,31 @@ from plyngent.tools import InstanceState, SessionState
from plyngent.tools.view import MemoryViewStore, session_data_view from plyngent.tools.view import MemoryViewStore, session_data_view
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Sequence from collections.abc import Mapping, Sequence
from plyngent.config.models import Provider from plyngent.config.models import Provider
from plyngent.config.store import ConfigStore from plyngent.config.store import ConfigStore
from plyngent.memory import MemoryStore from plyngent.memory import MemoryStore
from plyngent.memory.database.schema import Session as SessionRow from plyngent.memory.database.schema import Session as SessionRow
from plyngent.runtime.mcp_client import McpManager 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"] 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 @dataclass
class ReplState: class ReplState:
"""Mutable REPL session state.""" """Mutable REPL session state."""
@@ -82,6 +96,7 @@ class ReplState:
self.workspace = Path(self.workspace).expanduser().resolve() self.workspace = Path(self.workspace).expanduser().resolve()
self.instance_state.workspace_root = self.workspace self.instance_state.workspace_root = self.workspace
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.session_state = self._session_data_for_todo()
self.agent = self._make_agent() self.agent = self._make_agent()
self.sync_display_flags() self.sync_display_flags()
@@ -148,11 +163,17 @@ class ReplState:
self._todo_persist_tasks.add(task) self._todo_persist_tasks.add(task)
task.add_done_callback(self._todo_persist_tasks.discard) 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). """Build a SessionState with live todo + MemoryViewStore seed (durable tree).
*grants* defaults to empty (new session / startup). Pass the previous map *grants* defaults to empty (new session / startup). Pass the previous map
only when intentionally preserving soft-confirm trust across rebinds. only when intentionally preserving soft-confirm trust across rebinds.
*access_grants* seeds session directory grants (empty clears them).
""" """
grant_map = dict(grants or {}) grant_map = dict(grants or {})
store = MemoryViewStore({"todo": self.todo_stack.to_raw(), "grants": dict(grant_map)}) store = MemoryViewStore({"todo": self.todo_stack.to_raw(), "grants": dict(grant_map)})
@@ -161,7 +182,9 @@ class ReplState:
data=session_data_view(store=store), data=session_data_view(store=store),
todo=self.todo_stack, todo=self.todo_stack,
on_todo_change=self._todo_on_change, on_todo_change=self._todo_on_change,
on_access_change=self._on_access_change,
grants=grant_map, grants=grant_map,
access_grants=dict(access_grants or {}),
) )
def _bind_todo_tools(self) -> None: def _bind_todo_tools(self) -> None:
@@ -203,6 +226,67 @@ class ReplState:
if hasattr(self, "agent"): if hasattr(self, "agent"):
self.agent.todo_stack = self.todo_stack 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: def _tool_registry(self) -> ToolRegistry | None:
if not self.tools_enabled: if not self.tools_enabled:
return None return None
@@ -645,6 +729,7 @@ class ReplState:
self.agent = self._make_agent() self.agent = self._make_agent()
await self.agent.load_history() await self.agent.load_history()
await self.load_todo_stack() await self.load_todo_stack()
await self.load_access_grants()
async def resume_latest_or_new(self, name: str = "chat") -> str: async def resume_latest_or_new(self, name: str = "chat") -> str:
"""Resume most recently updated session for this workspace, or create one.""" """Resume most recently updated session for this workspace, or create one."""
@@ -660,6 +745,7 @@ class ReplState:
self.agent = self._make_agent() self.agent = self._make_agent()
await self.agent.load_history() await self.agent.load_history()
await self.load_todo_stack() await self.load_todo_stack()
await self.load_access_grants()
_ = await self.memory.touch_session(latest.sid) _ = await self.memory.touch_session(latest.sid)
return "resume" return "resume"
+2
View File
@@ -1,6 +1,7 @@
from .access import ACCESS_TOOLS as ACCESS_TOOLS from .access import ACCESS_TOOLS as ACCESS_TOOLS
from .access import MAX_ACCESS_GRANTS as MAX_ACCESS_GRANTS from .access import MAX_ACCESS_GRANTS as MAX_ACCESS_GRANTS
from .access import AccessDecision as AccessDecision 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 active_grant_count as active_grant_count
from .access import clear_process_access as clear_process_access from .access import clear_process_access as clear_process_access
from .access import clear_session_access as clear_session_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_process_access as grant_process_access
from .access import grant_session_access as grant_session_access from .access import grant_session_access as grant_session_access
from .access import request_directory_access as request_directory_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_config_access as set_config_access
from .access import set_directory_access_confirm_hook as set_directory_access_confirm_hook from .access import set_directory_access_confirm_hook as set_directory_access_confirm_hook
from .catalog import ToolCatalog as ToolCatalog from .catalog import ToolCatalog as ToolCatalog
+60 -5
View File
@@ -80,6 +80,12 @@ def set_directory_access_confirm_hook(
inst.extras[_HOOK_KEY] = 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( def grant_session_access(
path: Path | str, path: Path | str,
mode: AccessMode, mode: AccessMode,
@@ -90,6 +96,7 @@ def grant_session_access(
resolved = Path(path).expanduser().resolve() resolved = Path(path).expanduser().resolve()
sess = session if session is not None else require_session() sess = session if session is not None else require_session()
sess.access_grants[resolved] = mode sess.access_grants[resolved] = mode
_notify_access_change(sess)
return resolved return resolved
@@ -100,9 +107,10 @@ def grant_process_access(path: Path | str, mode: AccessMode) -> Path:
return resolved return resolved
def clear_process_access() -> None: def clear_process_access(*, instance: InstanceState | None = None) -> None:
"""Drop process-scoped (``--yes``) grants.""" """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: 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() sess = session if session is not None else get_session()
if sess is not None: if sess is not None:
sess.access_grants.clear() sess.access_grants.clear()
_notify_access_change(sess)
def active_grant_count() -> int: def active_grant_count() -> int:
@@ -122,14 +131,19 @@ def active_grant_count() -> int:
return total return total
def set_config_access(entries: Mapping[str, str]) -> tuple[list[Path], list[str]]: def set_config_access(
"""Install static TOML pre-allow entries on the bound instance policy. 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 Replaces any previous config grants. Returns ``(applied roots, skipped
keys)``; entries are skipped when the mode is unknown or the path does not keys)``; entries are skipped when the mode is unknown or the path does not
exist (the host warns about skipped keys). 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() policy.config_allow.clear()
applied: list[Path] = [] applied: list[Path] = []
skipped: list[str] = [] skipped: list[str] = []
@@ -148,6 +162,47 @@ def set_config_access(entries: Mapping[str, str]) -> tuple[list[Path], list[str]
return applied, skipped 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: def _resolve_target(path: str) -> Path | str:
"""Resolve a requested path; return a model-facing error string on failure.""" """Resolve a requested path; return a model-facing error string on failure."""
try: try:
+2
View File
@@ -74,6 +74,8 @@ class SessionState:
todo: TodoStack | None = None todo: TodoStack | None = None
# Host hook after todo tools mutate (e.g. CLI memory persist). # Host hook after todo tools mutate (e.g. CLI memory persist).
on_todo_change: Callable[[], None] | None = None 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). # Soft-confirm grants live map: tool_name → True (Phase 1 key is tool name).
# 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)
+156
View File
@@ -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()
+36
View File
@@ -1,5 +1,7 @@
from __future__ import annotations from __future__ import annotations
import pytest
from plyngent.cli.limits import ( from plyngent.cli.limits import (
format_tool_confirm_box, format_tool_confirm_box,
install_cli_limit_hooks, install_cli_limit_hooks,
@@ -54,3 +56,37 @@ def test_policy_confirm_noninteractive_denies() -> None:
with temporary_backend(NonInteractiveBackend()): with temporary_backend(NonInteractiveBackend()):
assert prompt_policy_command_confirm("sudo", ["sudo", "id"], 1.0) is False 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