From 7862a555bce376a23154ade7d7e63825790b6f09 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Tue, 6 Oct 2026 19:44:29 +0800 Subject: [PATCH] core/tools: review interpreters and detached runners past wrappers --- src/plyngent/tools/danger.py | 136 ++++++++++---------------------- tests/test_tools/test_danger.py | 61 +++++++++++++- 2 files changed, 101 insertions(+), 96 deletions(-) diff --git a/src/plyngent/tools/danger.py b/src/plyngent/tools/danger.py index 76789f3..2d4f6ab 100644 --- a/src/plyngent/tools/danger.py +++ b/src/plyngent/tools/danger.py @@ -4,61 +4,13 @@ import json import shlex from typing import TYPE_CHECKING, cast +from plyngent.tools.command_scan import is_interpreter, unwrap_command from plyngent.tools.workspace import AccessMode, WorkspaceError, resolve_path if TYPE_CHECKING: from collections.abc import Mapping, Sequence -_SHELL_BASENAMES: frozenset[str] = frozenset( - { - "bash", - "sh", - "zsh", - "fish", - "dash", - "ksh", - "csh", - "tcsh", - "powershell", - "pwsh", - "cmd", - "cmd.exe", - "python", - "python3", - "python2", - "ipython", - "ipython3", - "node", - "nodejs", - "deno", - "bun", - "ruby", - "perl", - "php", - "lua", - "r", - "julia", - "irb", - "pry", - "ghci", - "scala", - "jshell", - "sqlite3", - "psql", - "mysql", - "mongo", - "redis-cli", - } -) - - -def _basename(argv0: str) -> str: - name = argv0.replace("\\", "/").rsplit("/", 1)[-1] - name = name.removesuffix(".exe") - return name.lower() - - def _as_argv(args: Mapping[str, object], *, key: str = "command") -> list[str] | None: command = args.get(key) if not isinstance(command, list) or not command: @@ -71,67 +23,60 @@ def _as_argv(args: Mapping[str, object], *, key: str = "command") -> list[str] | return out -def _find_dash_c_code(argv: Sequence[str]) -> str | None: - """Return the argument after ``-c`` / ``-c…`` if present (python/bash/node style).""" - for index, part in enumerate(argv[1:], start=1): +def _dash_c_code(command: Sequence[str]) -> tuple[int, str] | None: + """Index and text of the ``-c`` code in an interpreter argv, if present. + + Only interpreters hand ``-c`` a block of code; for anything else it is a + plain flag (``grep -c``, ``od -c``) with nothing to review. + """ + for index, part in enumerate(command[1:], start=1): if part == "-c": - return argv[index + 1] if index + 1 < len(argv) else "" + return index, command[index + 1] if index + 1 < len(command) else "" # Combined short forms are uncommon; only exact -c is supported. return None def _indent_block(text: str, *, prefix: str = " ") -> str: - """Indent every line of *text* (for multi-line -c bodies in confirm prompts).""" + """Indent every line of *text* (for multi-line -c code in confirm prompts).""" if not text: return prefix return "\n".join(prefix + line if line else prefix.rstrip() for line in text.splitlines()) -def _format_argv_for_confirm(argv: Sequence[str], *, code: str | None) -> str: - """One-line argv summary; replace -c payload with ``$(command)`` when present.""" - if code is None: - return shlex.join(list(argv)) - parts: list[str] = [] - skip_next = False - for part in argv: - if skip_next: - skip_next = False - continue - if part == "-c": - parts.append("-c") - parts.append("$(command)") - skip_next = True - continue - parts.append(part) +def _format_argv_for_confirm(argv: Sequence[str], *, code_index: int | None) -> str: + """One-line argv summary; the ``-c`` code token becomes ``$(command)``.""" + parts = list(argv) + if code_index is not None and code_index + 1 < len(parts): + parts[code_index + 1] = "$(command)" return shlex.join(parts) -def _shell_or_dash_c_reason(argv: Sequence[str], *, via: str) -> str | None: - """Confirm interactive shells and ``*-c`` one-liners so the user can inspect argv. +def _command_reason(argv: Sequence[str], *, via: str) -> str | None: + """Confirm a command that runs code or detaches, whatever wraps it. Multi-line reason (shown inside the CLI confirm box). ``via`` is a short label such as ``run_argv`` or ``open_pty``. - For ``-c`` scripts, argv shows ``$(command)`` instead of inlining the body; - the full script is printed below untruncated, with every line indented. + The program is resolved past environment wrappers and `` run`` + forms (``env FOO=1 python …``, ``nohup pdm run python …``), and an + interpreter is confirmed the same way however it runs — a bare shell, a + script (``python x.py``), or a ``-c`` one-liner, whose code is printed below + ``command:`` instead of inline. """ - if not argv: + scan = unwrap_command(argv) + interpreter = is_interpreter(scan.base) + findings = [f"{name} (detached run)" for name in scan.self_review] + if interpreter: + findings.append(f"interpreter {scan.base!r}") + if not findings: return None - base = _basename(argv[0]) - code = _find_dash_c_code(argv) - display = _format_argv_for_confirm(argv, code=code) - + code = _dash_c_code(scan.command) if interpreter else None + code_index = scan.offset + code[0] if code is not None else None + display = _format_argv_for_confirm(argv, code_index=code_index) + reason = f"{via}: {' + '.join(findings)} — review before allow\n argv: {display}" if code is not None: - body = _indent_block(code, prefix=" ") - return f"{via}: {base} -c (review code before allow)\n argv: {display}\n command:\n{body}" - - if base in _SHELL_BASENAMES and len(argv) == 1: - return f"{via}: interactive {base!r} (review before allow)\n argv: {display}" - - if base in _SHELL_BASENAMES and "-c" not in argv[1:]: - return f"{via}: shell/runtime {base!r} without -c (review before allow)\n argv: {display}" - - return None + reason += f"\n command:\n{_indent_block(code[1])}" + return reason def _write_file_reason(args: Mapping[str, object]) -> str | None: @@ -185,7 +130,7 @@ def _run_argv_reason(args: Mapping[str, object]) -> str | None: argv = _as_argv(args, key="argv") if argv is None: return None - return _shell_or_dash_c_reason(argv, via="run_argv") + return _command_reason(argv, via="run_argv") def _batch_step_argv(item: object) -> list[str] | None: @@ -204,7 +149,7 @@ def _batch_step_argv(item: object) -> list[str] | None: def _run_argv_batch_reason(args: Mapping[str, object]) -> str | None: - """One confirm for the whole batch if any step is shell/REPL/-c.""" + """One confirm for the whole batch if any step runs code or detaches.""" raw = args.get("steps") if isinstance(raw, str): try: @@ -218,7 +163,7 @@ def _run_argv_batch_reason(args: Mapping[str, object]) -> str | None: reason for index, item in enumerate(cast("list[object]", raw)) if (argv := _batch_step_argv(item)) is not None - and (reason := _shell_or_dash_c_reason(argv, via=f"run_argv_batch[{index}]")) is not None + and (reason := _command_reason(argv, via=f"run_argv_batch[{index}]")) is not None ] if not risky: return None @@ -230,7 +175,7 @@ def _open_pty_reason(args: Mapping[str, object]) -> str | None: argv = _as_argv(args) if argv is None: return None - return _shell_or_dash_c_reason(argv, via="open_pty") + return _command_reason(argv, via="open_pty") def _fetch_reason(args: Mapping[str, object]) -> str | None: @@ -250,8 +195,9 @@ def classify_danger(name: str, args: Mapping[str, object]) -> str | None: # noq """Return a short reason if ``name``/``args`` need user confirm, else ``None``. Hard denylists (paths/commands) still raise independently. This only covers - soft confirms for mutating tools and risky shell/REPL launches - (interactive shells and ``python -c`` / ``bash -c`` one-liners). + soft confirms for mutating tools and risky command launches — interpreters + (however invoked) and detached runners such as ``nohup``, resolved past + environment wrappers and `` run``. Private/loopback fetch targets use a separate policy grant (not YOLO). """ if name == "delete_path": diff --git a/tests/test_tools/test_danger.py b/tests/test_tools/test_danger.py index cf0beb7..aae3cc5 100644 --- a/tests/test_tools/test_danger.py +++ b/tests/test_tools/test_danger.py @@ -96,6 +96,65 @@ def test_classify_shell_and_dash_c() -> None: assert r5 is not None and "-c" in r5 +def test_dash_c_means_code_only_for_interpreters() -> None: + # ``-c`` is a plain flag elsewhere: no code to review, no confirm. + assert classify_danger("run_argv", {"argv": ["grep", "-c", "needle", "f.txt"]}) is None + assert classify_danger("run_argv", {"argv": ["od", "-c", "f.bin"]}) is None + assert classify_danger("run_argv", {"argv": ["sort", "-c", "f.txt"]}) is None + + +def test_interpreter_review_does_not_split_on_dash_c() -> None: + script = classify_danger("run_argv", {"argv": ["python", "script.py"]}) + one_liner = classify_danger("run_argv", {"argv": ["python", "-c", "print(1)"]}) + bare_shell = classify_danger("open_pty", {"command": ["bash"]}) + assert script is not None and "interpreter 'python'" in script + assert one_liner is not None and "interpreter 'python'" in one_liner + assert bare_shell is not None and "interpreter 'bash'" in bare_shell + # Only the ``-c`` form has code to print below the argv line. + assert "command:" not in script + assert "command:" in one_liner + assert "print(1)" in one_liner + + +def test_versioned_interpreter_is_still_reviewed() -> None: + reason = classify_danger("run_argv", {"argv": ["python3.12", "-c", "import os"]}) + assert reason is not None and "interpreter 'python3.12'" in reason + assert "import os" in reason + + +def test_wrappers_reach_the_interpreter() -> None: + for argv in ( + ["env", "FOO=1", "python", "-c", "print(1)"], + ["sudo", "-u", "root", "python", "-c", "print(1)"], + ["timeout", "5", "bash", "-c", "rm -rf /"], + ["pdm", "run", "python", "-c", "print(1)"], + ["uv", "run", "--with", "rich", "python", "-c", "print(1)"], + ): + reason = classify_danger("run_argv", {"argv": argv}) + assert reason is not None, argv + assert "interpreter 'python'" in reason or "interpreter 'bash'" in reason + assert "$(command)" in reason + + +def test_only_the_interpreter_dash_c_becomes_a_placeholder() -> None: + reason = classify_danger("run_argv", {"argv": ["ionice", "-c", "2", "python", "-c", "print(1)"]}) + assert reason is not None + argv_line = next(line for line in reason.splitlines() if line.startswith(" argv:")) + assert "ionice -c 2" in argv_line + assert argv_line.count("$(command)") == 1 + assert "print(1)" in reason + + +def test_detached_runner_is_always_reviewed() -> None: + assert classify_danger("run_argv", {"argv": ["sleep", "5"]}) is None + reason = classify_danger("run_argv", {"argv": ["nohup", "sleep", "5"]}) + assert reason is not None and "nohup (detached run)" in reason + nested = classify_danger("run_argv", {"argv": ["nohup", "python", "-c", "print(1)"]}) + assert nested is not None + assert "nohup (detached run)" in nested and "interpreter 'python'" in nested + assert "print(1)" in nested + + async def test_confirm_deny_with_comment() -> None: from plyngent.agent.tools import ToolRegistry, tool from plyngent.tools.danger import classify_danger as danger @@ -124,7 +183,7 @@ def test_shell_confirm_formats_command_placeholder() -> None: assert reason is not None assert "$(command)" in reason assert "line1" in reason and "line2" in reason and "line3" in reason - argv_line = next(ln for ln in reason.splitlines() if "argv:" in ln) + argv_line = next(ln for ln in reason.splitlines() if ln.startswith(" argv:")) assert "line1" not in argv_line lines = reason.splitlines() idx = lines.index(" command:")