diff --git a/src/plyngent/tools/command_scan.py b/src/plyngent/tools/command_scan.py new file mode 100644 index 0000000..58d9c70 --- /dev/null +++ b/src/plyngent/tools/command_scan.py @@ -0,0 +1,290 @@ +"""Resolve the program a command really runs, past wrappers and launchers. + +Confirming a risky command (``danger``) and denying a policy-blocked one +(``workspace.check_command_allowed``) must judge the program that will really +run, not the outermost token: ``env FOO=1 python -c …``, ``nohup pdm run python +…`` and ``timeout 5 bash -c …`` all reach the interpreter — or the denylisted +program — further in. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Sequence + +__all__ = ["UnwrappedCommand", "basename", "is_interpreter", "unwrap_command"] + +# Wrappers that run the program named on their own command line. +_ENV_WRAPPERS: frozenset[str] = frozenset( + { + "command", + "doas", + "env", + "exec", + "ionice", + "nice", + "nohup", + "pkexec", + "setsid", + "stdbuf", + "su", + "sudo", + "time", + "timeout", + "watch", + "xargs", + } +) + +# Detached runners: review-worthy whatever program they wrap. +_SELF_REVIEW: frozenset[str] = frozenset({"nohup"}) + +# Wrappers that take a bare duration of their own (``timeout 5 cmd``). +_DURATION_WRAPPERS: frozenset[str] = frozenset({"timeout"}) + +# `` run …``: task runners and package managers that launch +# a program in their environment. A package-manager script (``npm run build``) +# names a script rather than a program; the launcher basename is still checked. +_RUN_LAUNCHERS: frozenset[str] = frozenset( + {"conda", "hatch", "mise", "npm", "pdm", "pipenv", "pnpm", "poetry", "uv", "yarn"} +) + +# Programs that run code handed to them as text: for these, ``-c`` carries the +# code (``grep -c`` / ``od -c`` is a plain flag, never a block of code). +_INTERPRETERS: frozenset[str] = frozenset( + { + "bash", + "bun", + "cmd", + "csh", + "dash", + "deno", + "fish", + "ghci", + "ipython", + "ipython3", + "irb", + "jshell", + "julia", + "ksh", + "lua", + "mongo", + "mysql", + "node", + "nodejs", + "perl", + "php", + "powershell", + "pry", + "psql", + "pwsh", + "python", + "python2", + "python3", + "r", + "redis-cli", + "ruby", + "scala", + "sh", + "sqlite3", + "tcsh", + "zsh", + } +) + +# Trailing version of an interpreter: ``python3.12`` / ``node20`` / ``php8.2``. +_VERSION_SUFFIX = re.compile(r"[0-9][0-9.]*$") + +# ``FOO=bar`` in front of a program (env/sudo style). +_ASSIGNMENT = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=") + +# A bare duration, as ``timeout`` takes: ``5``, ``1.5``, ``30s``. +_DURATION = re.compile(r"[0-9]+(?:\.[0-9]+)?[smhd]?") + +# Options that consume the next token, so ``sudo -u root python`` still finds +# ``python``; jammed forms (``-n10``, ``--with=pkg``) need no entry, and unknown +# options are skipped bare. +_VALUE_OPTS: dict[str, frozenset[str]] = { + "command": frozenset({"-p", "-v", "-V"}), + "doas": frozenset({"-u", "--user"}), + "env": frozenset({"-C", "-S", "-u", "--chdir", "--split-string", "--unset"}), + "exec": frozenset({"-a", "-c", "-l"}), + "ionice": frozenset({"-c", "-n", "-p", "-P", "-u"}), + "nice": frozenset({"-n", "--adjustment"}), + "pkexec": frozenset({"--disable-internal-agent", "--user"}), + "stdbuf": frozenset({"-e", "-i", "-o"}), + "su": frozenset({"-c", "-g", "-G", "-p", "-s", "-u", "-w", "--command", "--group", "--session-command", "--shell"}), + "sudo": frozenset( + { + "-C", + "-D", + "-g", + "-h", + "-p", + "-r", + "-R", + "-t", + "-T", + "-u", + "-U", + "--close-from", + "--command-timeout", + "--group", + "--host", + "--other-user", + "--prompt", + "--role", + "--type", + "--user", + } + ), + "time": frozenset({"-f", "-o", "--format", "--output"}), + "timeout": frozenset({"-k", "-s", "--kill-after", "--signal"}), + "watch": frozenset({"-n"}), + "xargs": frozenset({"-a", "-d", "-E", "-e", "-I", "-L", "-n", "-P", "-s"}), +} + +# Launcher options that consume a value (``uv run --with rich python``). +_RUN_VALUE_OPTS: frozenset[str] = frozenset( + { + "-C", + "-e", + "-F", + "-G", + "-m", + "-n", + "-p", + "-s", + "-w", + "--config", + "--constraint", + "--default-index", + "--directory", + "--env", + "--env-file", + "--extra", + "--extras", + "--file", + "--group", + "--index", + "--module", + "--name", + "--overrides", + "--package", + "--prefix", + "--project", + "--python", + "--python-version", + "--script", + "--with", + "--without", + } +) + + +@dataclass(frozen=True) +class UnwrappedCommand: + """A command plus the program it really runs, past wrappers and launchers.""" + + argv: tuple[str, ...] + offset: int + wrappers: tuple[str, ...] + base: str + self_review: tuple[str, ...] + + @property + def command(self) -> tuple[str, ...]: + """The program and its arguments (the whole argv when nothing unwrapped).""" + return self.argv[self.offset :] + + +def basename(argv0: str) -> str: + """Lower-cased basename of an executable token (``\\`` separators, ``.exe``).""" + name = argv0.replace("\\", "/").rsplit("/", 1)[-1].lower() + return name.removesuffix(".exe") + + +def is_interpreter(base: str) -> bool: + """Whether *base* runs code from its own command line. + + Version suffixes are ignored, so ``python3.12`` / ``node20`` / ``php8.2`` + count as interpreters too. + """ + return base in _INTERPRETERS or _VERSION_SUFFIX.sub("", base) in _INTERPRETERS + + +def unwrap_command(argv: Sequence[str]) -> UnwrappedCommand: + """Resolve *argv* to the program behind any wrappers and ``run`` launchers. + + Wrappers are unwrapped outer → inner as long as a program still follows + them; their own options and ``FOO=bar`` assignments are skipped. + """ + parts = list(argv) + wrappers: list[str] = [] + index = 0 + while index < len(parts): + name = basename(parts[index]) + nxt = _skip_wrapper(name, parts, index + 1) + if nxt is None: + break + wrappers.append(name) + index = nxt + base = basename(parts[index]) if index < len(parts) else "" + chain = dict.fromkeys((*wrappers, base)) + return UnwrappedCommand( + argv=tuple(parts), + offset=index, + wrappers=tuple(wrappers), + base=base, + self_review=tuple(name for name in chain if name in _SELF_REVIEW), + ) + + +def _skip_wrapper(name: str, parts: list[str], start: int) -> int | None: + """Index of the program *name* wraps, or ``None`` when it is the whole command.""" + if name in _ENV_WRAPPERS: + nxt = _skip_options( + parts, + start, + value_opts=_VALUE_OPTS.get(name, frozenset()), + leading_duration=name in _DURATION_WRAPPERS, + ) + return nxt if nxt < len(parts) else None + if name in _RUN_LAUNCHERS and start < len(parts) and parts[start] == "run": + nxt = _skip_options(parts, start + 1, value_opts=_RUN_VALUE_OPTS, leading_duration=False) + return nxt if nxt < len(parts) else None + return None + + +def _skip_options( + parts: list[str], + start: int, + *, + value_opts: frozenset[str], + leading_duration: bool, +) -> int: + """Index of the first token after a wrapper's own options and assignments.""" + index = start + took_duration = False + while index < len(parts): + token = parts[index] + if token == "--": + return index + 1 + if _ASSIGNMENT.match(token): + index += 1 + continue + if token.startswith("-"): + index += 1 + if token in value_opts: + index += 1 + continue + if leading_duration and not took_duration and _DURATION.fullmatch(token): + took_duration = True + index += 1 + continue + return index + return index diff --git a/tests/test_tools/test_command_scan.py b/tests/test_tools/test_command_scan.py new file mode 100644 index 0000000..8c65843 --- /dev/null +++ b/tests/test_tools/test_command_scan.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +from plyngent.tools.command_scan import basename, is_interpreter, unwrap_command + + +def test_basename() -> None: + assert basename("/usr/bin/python3.12") == "python3.12" + assert basename("C:\\Tools\\RM.EXE") == "rm" + assert basename("pdm") == "pdm" + + +def test_unwrap_plain_command() -> None: + scan = unwrap_command(["grep", "-c", "needle", "f.txt"]) + assert scan.wrappers == () + assert scan.offset == 0 + assert scan.base == "grep" + assert scan.command == ("grep", "-c", "needle", "f.txt") + assert scan.self_review == () + + +def test_unwrap_env_assignment_and_options() -> None: + scan = unwrap_command(["env", "FOO=1", "nice", "-n", "10", "python", "-c", "print(1)"]) + assert scan.wrappers == ("env", "nice") + assert scan.offset == 5 + assert scan.base == "python" + assert scan.command == ("python", "-c", "print(1)") + + +def test_unwrap_privileged_wrapper() -> None: + scan = unwrap_command(["sudo", "-u", "root", "--", "python", "script.py"]) + assert scan.wrappers == ("sudo",) + assert scan.base == "python" + assert scan.command == ("python", "script.py") + + +def test_unwrap_wrapper_duration_of_its_own() -> None: + scan = unwrap_command(["timeout", "30s", "bash", "-c", "ls"]) + assert scan.wrappers == ("timeout",) + assert scan.base == "bash" + scan = unwrap_command(["timeout", "5"]) + assert scan.wrappers == () + assert scan.base == "timeout" + + +def test_unwrap_run_launcher_skips_its_options() -> None: + scan = unwrap_command(["uv", "run", "--with", "rich", "python", "-c", "print(1)"]) + assert scan.wrappers == ("uv",) + assert scan.base == "python" + assert scan.command == ("python", "-c", "print(1)") + + +def test_unwrap_nested_wrapper_and_launcher() -> None: + scan = unwrap_command(["nohup", "pdm", "run", "python", "script.py"]) + assert scan.wrappers == ("nohup", "pdm") + assert scan.offset == 3 + assert scan.base == "python" + assert scan.self_review == ("nohup",) + + +def test_unwrap_stops_when_no_program_follows() -> None: + for argv in (["env"], ["timeout", "5"], ["pdm", "run"], ["nohup"]): + scan = unwrap_command(argv) + assert scan.wrappers == (), argv + assert scan.offset == 0, argv + assert scan.base == argv[0], argv + + +def test_unwrap_empty_argv() -> None: + scan = unwrap_command([]) + assert scan.base == "" + assert scan.command == () + + +def test_is_interpreter_ignores_version_suffix() -> None: + assert is_interpreter("python") + assert is_interpreter("python3.12") + assert is_interpreter("node20") + assert is_interpreter("php8.2") + assert is_interpreter("sqlite3") + assert is_interpreter("bash") + assert not is_interpreter("grep") + assert not is_interpreter("od") + assert not is_interpreter("phpmyadmin") + assert not is_interpreter("")