NCBM/plyngent
1
0
Fork 0
mirror of https://github.com/NCBM/plyngent.git synced 2026-11-07 15:37:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

core/tools: review interpreters and detached runners past wrappers

Browse Source
This commit is contained in:
worldmozara
2026-10-06 19:44:29 +08:00
parent 9501efb3d7
commit 7862a555bc
2 changed files with 101 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/plyngent/tools/danger.py
+41 -95
View File
@@ -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 ``<launcher> 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 ``<launcher> run``.
Private/loopback fetch targets use a separate policy grant (not YOLO).
"""
if name == "delete_path":
tests/test_tools/test_danger.py
+60 -1
View File
@@ -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:")