From d6616d0da3c7da8737023ad200ec13e30dd37888 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Mon, 20 Jul 2026 03:02:34 +0800 Subject: [PATCH] core/tools: timed human allow for denylisted commands Denylist hits ask the user (default deny, timeout) instead of only hard rejecting when a policy hook is installed. Independent of YOLO soft-confirm; session grants skip re-prompt for the same basename. CLI installs the hook via install_cli_limit_hooks. --- CLAUDE.md | 2 +- src/plyngent/cli/app.py | 4 ++ src/plyngent/cli/limits.py | 91 +++++++++++++++++++++++++++++- src/plyngent/tools/__init__.py | 6 ++ src/plyngent/tools/workspace.py | 81 +++++++++++++++++++++++++- tests/test_cli/test_limits.py | 12 ++++ tests/test_tools/test_workspace.py | 79 ++++++++++++++++++++++++++ 7 files changed, 268 insertions(+), 7 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 93414f9..ac44e0d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -78,7 +78,7 @@ Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init (+ lightweight SQLite ` Module-level `@tool` handlers. Call `set_workspace_root()` before use. -- **`workspace`**: path resolve under primary root **or** temporary allowlist roots; path substring denylist; command basename denylist; clearer deny messages. +- **`workspace`**: path resolve under primary root **or** temporary allowlist roots; path substring denylist; command basename denylist (CLI: timed human allow override, default deny on timeout; **not** skipped by YOLO); clearer deny messages. - **`file`**: `read_file` (`with_lineno`), `write_file`, `listdir`, `tree` (VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `grep_files` (regex, skip VCS/binary), `edit_replace` (`max_replaces`, reports remaining matches), `edit_lineno` (1-based range), `copy_path` / `move_path` / `delete_path`, `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind). - **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); `run_command_batch` (serial steps, `pipe_out` on provider, `mix_stderr`, `stop_on_error`); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (escapes) / `ask_into_pty` (human→PTY only; `secret` no-echo; answer never in tool result) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep). - PTY: backend in `pty_backend.py`; structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)` (**CSI sanitized** so host TTY is never reset); keys via `write_pty_keys` only (`\xHH`, `ctrl+x`, `key=esc`); secrets via `ask_into_pty` (not `write_pty` data); session limit/idle TTL/output budget; close terminate→kill. diff --git a/src/plyngent/cli/app.py b/src/plyngent/cli/app.py index dfdd31b..0d50f21 100644 --- a/src/plyngent/cli/app.py +++ b/src/plyngent/cli/app.py @@ -327,6 +327,10 @@ async def _run_chat( # noqa: C901, PLR0912, PLR0915 — chat orchestration from plyngent.tools.temp_workspace import cleanup_temporary_workspaces PtyManager.close_all() + from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook + + set_policy_confirm_hook(None) + clear_policy_allowed_commands() _ = cleanup_temporary_workspaces() diff --git a/src/plyngent/cli/limits.py b/src/plyngent/cli/limits.py index 795964e..b20fa63 100644 --- a/src/plyngent/cli/limits.py +++ b/src/plyngent/cli/limits.py @@ -1,6 +1,9 @@ from __future__ import annotations +import contextlib +import selectors import shutil +import sys from typing import TYPE_CHECKING, Literal from plyngent.cli.interrupt import pause_task_cancel_for_prompt @@ -19,8 +22,7 @@ from plyngent.prompting import ( from plyngent.tools.process.pty_session import PtyManager if TYPE_CHECKING: - from collections.abc import Mapping - + from collections.abc import Mapping, Sequence type WorkspaceMismatchChoice = Literal["keep", "rebind", "abort"] _BOX_MIN_WIDTH = 40 @@ -150,6 +152,86 @@ def prompt_confirm_tool(name: str, args: Mapping[str, object], reason: str) -> b return _prompt_confirm_tool_sync(name, args, reason) +def _try_readline() -> str | None: + try: + return sys.stdin.readline() + except OSError, EOFError: + return None + + +def _read_yes_no_line_with_timeout(timeout_seconds: float) -> str | None: + """Read one line from stdin with *timeout_seconds*; ``None`` on timeout/EOF. + + Uses :mod:`selectors` on the process stdin FD. On Windows without selector + support, falls back to blocking readline (no true timeout). + """ + try: + if not sys.stdin.isatty(): + return None + fd = sys.stdin.fileno() + except OSError, ValueError, AttributeError: + return _try_readline() + + try: + selector = selectors.DefaultSelector() + _ = selector.register(fd, selectors.EVENT_READ) + except OSError, ValueError, AttributeError: + return _try_readline() + + try: + events = selector.select(timeout_seconds) + if not events: + return None + return sys.stdin.readline() + except OSError, ValueError, EOFError: + return None + finally: + selector.close() + + +def prompt_policy_command_confirm( + basename: str, + argv: Sequence[str], + timeout_seconds: float, +) -> bool: + """Ask whether to allow a denylisted command for this session (timed). + + Independent of YOLO: always prompts when interactive. Default is **deny**. + Timeout, cancel, or non-interactive → False. + """ + argv_preview = " ".join(str(part) for part in argv) + reason = ( + f"policy denylist blocks basename {basename!r}\n" + f"argv: {argv_preview}\n" + f"Allow this basename for the rest of this process?\n" + f"(timeout {timeout_seconds:g}s defaults to DENY; not skipped by YOLO)" + ) + try: + with pause_task_cancel_for_prompt(): + backend = get_prompt_backend() + if not backend.is_interactive(): + return False + backend.echo() + backend.secho(format_tool_confirm_box("policy", reason), fg="yellow") + backend.echo() + prompt = f"[policy] allow {basename!r}? [y/N] (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"[policy] timed out after {timeout_seconds:g}s — denied", + fg="red", + err=True, + ) + return False + return raw.strip().lower() in {"y", "yes"} + except NonInteractiveError, KeyboardInterrupt, EOFError: + return False + + 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 @@ -268,6 +350,9 @@ async def prompt_workspace_mismatch_async( def install_cli_limit_hooks() -> None: - """Register interactive continue hooks and prompt cancel-pause for the CLI.""" + """Register interactive continue hooks, policy confirm, and prompt cancel-pause.""" configure_prompting(pause_factory=pause_task_cancel_for_prompt) PtyManager.set_limit_continue_hook(prompt_continue_limit) + from plyngent.tools.workspace import set_policy_confirm_hook + + set_policy_confirm_hook(prompt_policy_command_confirm) diff --git a/src/plyngent/tools/__init__.py b/src/plyngent/tools/__init__.py index 609436d..b31ea1b 100644 --- a/src/plyngent/tools/__init__.py +++ b/src/plyngent/tools/__init__.py @@ -43,20 +43,26 @@ from .vcs import vcs_status as vcs_status from .workspace import ( DEFAULT_COMMAND_DENYLIST as DEFAULT_COMMAND_DENYLIST, ) +from .workspace import DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS as DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS from .workspace import WorkspaceError as WorkspaceError from .workspace import add_workspace_allowlist as add_workspace_allowlist 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 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 list_workspace_allowlist as list_workspace_allowlist 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 resolve_path as resolve_path from .workspace import set_command_denylist as set_command_denylist 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 DEFAULT_TOOLS = [*FILE_TOOLS, *PROCESS_TOOLS, *VCS_TOOLS, *CHAT_TOOLS, *TODO_TOOLS] diff --git a/src/plyngent/tools/workspace.py b/src/plyngent/tools/workspace.py index 8172c71..e92b8e6 100644 --- a/src/plyngent/tools/workspace.py +++ b/src/plyngent/tools/workspace.py @@ -1,6 +1,10 @@ from __future__ import annotations from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Callable, Sequence DEFAULT_COMMAND_DENYLIST: frozenset[str] = frozenset( { @@ -29,6 +33,12 @@ DEFAULT_COMMAND_DENYLIST: frozenset[str] = frozenset( # Max concurrent temporary workspaces registered in one process. MAX_TEMPORARY_WORKSPACES = 16 +# Timed human override for command denylist (independent of YOLO soft-confirm). +DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS = 30.0 + +# Hook: (basename, argv, timeout_seconds) -> True allow for this session basename. +type PolicyConfirmHook = Callable[[str, Sequence[str], float], bool] + class WorkspaceError(ValueError): """Raised when a path or command violates workspace policy.""" @@ -42,10 +52,17 @@ class _WorkspaceState: allowlist: list[Path] # Paths created by new_temporary_workspace (subset of allowlist); cleaned on chat exit. temporary_owned: list[Path] + # Basenames the human allowed this process (denylist override, not permanent). + policy_allowed_commands: set[str] + policy_confirm_hook: PolicyConfirmHook | None + policy_confirm_timeout_seconds: float def __init__(self) -> None: self.allowlist = [] self.temporary_owned = [] + self.policy_allowed_commands = set() + self.policy_confirm_hook = None + self.policy_confirm_timeout_seconds = DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS _state = _WorkspaceState() @@ -87,12 +104,47 @@ def get_path_denylist() -> tuple[str, ...]: def set_command_denylist(names: list[str] | tuple[str, ...] | frozenset[str] | None) -> None: """Set denied command basenames (None restores defaults).""" _state.command_denylist = DEFAULT_COMMAND_DENYLIST if names is None else frozenset(names) + # Session overrides only make sense against the active denylist. + _state.policy_allowed_commands &= _state.command_denylist def get_command_denylist() -> frozenset[str]: return _state.command_denylist +def set_policy_confirm_hook(hook: PolicyConfirmHook | None) -> None: + """Register a timed human confirm for denylisted commands (CLI installs this).""" + _state.policy_confirm_hook = hook + + +def get_policy_confirm_hook() -> PolicyConfirmHook | None: + return _state.policy_confirm_hook + + +def set_policy_confirm_timeout(seconds: float) -> None: + """Timeout for policy confirm prompts (must be > 0).""" + if seconds <= 0: + msg = "policy confirm timeout must be > 0" + raise WorkspaceError(msg) + _state.policy_confirm_timeout_seconds = float(seconds) + + +def get_policy_confirm_timeout() -> float: + return _state.policy_confirm_timeout_seconds + + +def clear_policy_allowed_commands() -> None: + """Drop session-scoped denylist overrides (tests / chat exit).""" + _state.policy_allowed_commands.clear() + + +def grant_policy_command(basename: str) -> None: + """Allow *basename* for the rest of this process despite the denylist.""" + name = basename.strip() + if name: + _state.policy_allowed_commands.add(name) + + def add_workspace_allowlist(root: Path | str, *, owned: bool = False) -> Path: """Allow tool paths under *root* in addition to the primary workspace. @@ -180,11 +232,34 @@ def resolve_path(path: str | Path) -> Path: def check_command_allowed(argv: list[str]) -> None: - """Raise if argv is empty or the executable basename is denylisted.""" + """Raise if argv is empty or the executable basename is denylisted. + + Denylisted basenames are not hard-rejected when a policy confirm hook is + installed: the human is asked (with a timeout; default deny). Session grants + skip re-prompting for the same basename. Independent of YOLO soft-confirm. + """ if not argv: msg = "command argv must not be empty" raise WorkspaceError(msg) binary = Path(argv[0]).name - if binary in _state.command_denylist: - msg = f"command denied by policy (basename {binary!r} is blocked)" + if binary not in _state.command_denylist: + return + if binary in _state.policy_allowed_commands: + return + hook = _state.policy_confirm_hook + if hook is not None: + timeout = _state.policy_confirm_timeout_seconds + try: + allowed = bool(hook(binary, list(argv), timeout)) + except Exception as exc: + msg = f"command denied by policy (basename {binary!r}; confirm failed: {exc})" + raise WorkspaceError(msg) from exc + if allowed: + _state.policy_allowed_commands.add(binary) + return + msg = ( + f"command denied by policy (basename {binary!r} is blocked; user declined or timed out after {timeout:g}s)" + ) raise WorkspaceError(msg) + msg = f"command denied by policy (basename {binary!r} is blocked)" + raise WorkspaceError(msg) diff --git a/tests/test_cli/test_limits.py b/tests/test_cli/test_limits.py index aa0d522..a51ed19 100644 --- a/tests/test_cli/test_limits.py +++ b/tests/test_cli/test_limits.py @@ -41,8 +41,20 @@ def test_format_tool_confirm_box_multiline() -> None: def test_install_cli_limit_hooks() -> None: + from plyngent.tools.workspace import get_policy_confirm_hook, set_policy_confirm_hook + install_cli_limit_hooks() assert callable(getattr(PtyManager, "_limit_continue", None)) + assert get_policy_confirm_hook() is not None PtyManager.set_limit_continue_hook(None) + set_policy_confirm_hook(None) # Backend remains usable after install. assert get_prompt_backend().is_interactive() or True + + +def test_policy_confirm_noninteractive_denies() -> None: + from plyngent.cli.limits import prompt_policy_command_confirm + from plyngent.prompting import NonInteractiveBackend + + with temporary_backend(NonInteractiveBackend()): + assert prompt_policy_command_confirm("sudo", ["sudo", "id"], 1.0) is False diff --git a/tests/test_tools/test_workspace.py b/tests/test_tools/test_workspace.py index d260d33..84aa409 100644 --- a/tests/test_tools/test_workspace.py +++ b/tests/test_tools/test_workspace.py @@ -43,12 +43,91 @@ def test_path_denylist(workspace: object) -> None: def test_command_denylist(workspace: object) -> None: del workspace + from plyngent.tools.workspace import ( + clear_policy_allowed_commands, + set_policy_confirm_hook, + ) + + set_policy_confirm_hook(None) + clear_policy_allowed_commands() with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"): check_command_allowed(["rm", "-rf", "/"]) check_command_allowed(["echo", "ok"]) set_command_denylist(None) +def test_command_denylist_policy_confirm_allow(workspace: object) -> None: + del workspace + from plyngent.tools.workspace import ( + clear_policy_allowed_commands, + set_policy_confirm_hook, + ) + + calls: list[tuple[str, float]] = [] + + def hook(basename: str, argv: object, timeout: float) -> bool: + del argv + calls.append((basename, timeout)) + return basename == "sudo" + + set_policy_confirm_hook(hook) + clear_policy_allowed_commands() + try: + check_command_allowed(["sudo", "echo", "test"]) + assert calls == [("sudo", 30.0)] + # Session grant: second call does not re-prompt. + check_command_allowed(["sudo", "id"]) + assert len(calls) == 1 + finally: + set_policy_confirm_hook(None) + clear_policy_allowed_commands() + + +def test_command_denylist_policy_confirm_deny(workspace: object) -> None: + del workspace + from plyngent.tools.workspace import ( + clear_policy_allowed_commands, + set_policy_confirm_hook, + ) + + set_policy_confirm_hook(lambda *_a: False) + clear_policy_allowed_commands() + try: + with pytest.raises(WorkspaceError, match="declined or timed out"): + check_command_allowed(["sudo", "echo", "x"]) + finally: + set_policy_confirm_hook(None) + clear_policy_allowed_commands() + + +def test_command_denylist_policy_confirm_timeout_value(workspace: object) -> None: + del workspace + from plyngent.tools.workspace import ( + clear_policy_allowed_commands, + set_policy_confirm_hook, + set_policy_confirm_timeout, + ) + + seen: list[float] = [] + + def hook(basename: str, argv: object, timeout: float) -> bool: + del basename, argv + seen.append(timeout) + return False + + set_policy_confirm_timeout(5.0) + set_policy_confirm_hook(hook) + clear_policy_allowed_commands() + try: + with pytest.raises(WorkspaceError): + check_command_allowed(["rm", "x"]) + assert seen == [5.0] + finally: + set_policy_confirm_timeout(30.0) + set_policy_confirm_hook(None) + clear_policy_allowed_commands() + + def test_root_required() -> None: clear_workspace_root() with pytest.raises(WorkspaceError, match="not set"):