From 12236cffb54127de168f3fde9e82b9bebc097fff Mon Sep 17 00:00:00 2001 From: worldmozara Date: Fri, 23 Oct 2026 10:47:36 +0800 Subject: [PATCH] core/tools: read each wrapper's options with getopt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four hand-written tables said which wrapper options take a value, and the comment above them admitted the failure mode: an entry for a flag eats the program behind it, which is how `exec -c rm -rf /` hid `rm`. Guessing was the other half — an option nobody listed was skipped bare, so the token after it became the program. Each wrapper now carries a `getopt` spec instead (`command_scan._WRAPPERS`: short options with `:`, long options with `=`), which spells flags and values apart, understands jammed forms (`-p2222`, `--with=pkg`), `--`, and unique long prefixes, and stops at the first token that is not an option — exactly the wrapped program. The remote clients and the `run` launchers are the same thing with a flag on the spec. An option the spec does not know is no longer skipped as if it were a flag: the unwrap stops at that wrapper (`unparsed`) and the review asks — `options not understood sudo (cannot read its options: option --new-sudo-flag not recognized)`, with the whole argv. The wrapper itself stays the command the denylist judges, so a new `sudo` flag cannot hide `rm` from the hard gate. --- src/plyngent/tools/command_scan.py | 551 +++++++++++++++-------------- src/plyngent/tools/danger.py | 2 + 2 files changed, 287 insertions(+), 266 deletions(-) diff --git a/src/plyngent/tools/command_scan.py b/src/plyngent/tools/command_scan.py index 4003b12..b524f70 100644 --- a/src/plyngent/tools/command_scan.py +++ b/src/plyngent/tools/command_scan.py @@ -9,6 +9,7 @@ program — further in. from __future__ import annotations +import getopt import re from dataclasses import dataclass from typing import TYPE_CHECKING @@ -28,34 +29,9 @@ __all__ = [ "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. @@ -119,99 +95,185 @@ _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. List value-taking options only: an entry for a flag -# eats the program behind it, which hid ``rm`` from ``exec -c rm -rf /`` and -# judged ``exec -c git -c … init`` as ``init``. An option with an *optional* -# value (``xargs -e``, ``watch -d``) needs no entry either — its bare form runs -# the program, and the jammed form carries the value in the same token. -_VALUE_OPTS: dict[str, frozenset[str]] = { - "doas": frozenset({"-u", "--user"}), - "env": frozenset({"-C", "-S", "-u", "--chdir", "--split-string", "--unset"}), - "exec": frozenset({"-a"}), - "ionice": frozenset({"-c", "-n", "-p", "-P", "-u"}), - "nice": frozenset({"-n", "--adjustment"}), - "pkexec": frozenset({"--user"}), - "stdbuf": frozenset({"-e", "-i", "-o"}), - "su": frozenset({"-c", "-g", "-G", "-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", - } +# Every wrapper's own options, spelled the way ``getopt`` reads them: a short +# option with ``:`` (``u:``) takes a value, a bare one is a flag, and a long +# option ending in ``=`` takes one. The flags matter as much as the values — a +# value-only list let a flag eat the program behind it (``exec -c rm -rf /`` +# resolved to the ``-c``, hiding ``rm``), and an unknown option is no longer +# guessed at (see `_parse_options`). + + +@dataclass(frozen=True) +class _Spec: + """How one wrapper is invoked, so the program behind it can be found.""" + + short: str = "" + long: tuple[str, ...] = () + #: Options whose value is a command line the wrapper runs (``env -S``). + command_opts: frozenset[str] = frozenset() + #: Takes a bare duration of its own before its command (``timeout 5 cmd``). + duration: bool = False + #: `` run ``: only the ``run`` form wraps a program. + launcher: bool = False + #: Runs its command on another machine (``ssh`` / ``mosh``). + remote: bool = False + + +_WRAPPERS: dict[str, _Spec] = { + "command": _Spec(short="pvV"), + "doas": _Spec(short="C:nsu:", long=("user=",)), + # ``env -S`` / ``su -c`` carry a *command line*: the value is code the review + # must surface (`_Spec.command_opts`), not an argument. + "env": _Spec( + short="0iC:u:S:v", + long=("chdir=", "debug", "ignore-environment", "null", "split-string=", "unset="), + command_opts=frozenset({"--split-string", "-S"}), + ), + "exec": _Spec(short="cla:"), + "ionice": _Spec(short="c:n:p:P:u:thV"), + "nice": _Spec(short="n:", long=("adjustment=",)), + "nohup": _Spec(), + "pkexec": _Spec(long=("disable-internal-agent", "help", "user=", "version")), + "setsid": _Spec(short="cfwV"), + "stdbuf": _Spec(short="e:i:o:", long=("error=", "input=", "output=")), + "su": _Spec( + short="c:g:G:lmps:u:w:", + long=( + "command=", + "group=", + "login", + "preserve-environment", + "session-command=", + "shell=", + "whitelist-environment=", + ), + command_opts=frozenset({"--command", "--session-command", "-c"}), + ), + "sudo": _Spec( + short="AbEHiKknPSsVvLlC:D:g:h:p:r:R:t:T:u:U:", + long=( + "askpass", + "background", + "chdir=", + "close-from=", + "command-timeout=", + "edit", + "group=", + "host=", + "list", + "login", + "non-interactive", + "other-user=", + "preserve-env", + "prompt=", + "remove-timestamp", + "reset-timestamp", + "role=", + "set-home", + "shell", + "stdin", + "type=", + "user=", + "validate", + "version", + ), + ), + "time": _Spec(short="af:o:pq", long=("append", "format=", "output=", "portability", "quiet", "verbose")), + "timeout": _Spec( + short="k:s:v", + long=("foreground", "kill-after=", "preserve-status", "signal="), + duration=True, + ), + "watch": _Spec( + short="bcdeghn:prtvx", + long=( + "beep", + "chgexit", + "color", + "differences", + "errexit", + "exec", + "interval=", + "no-rerun", + "no-title", + "precise", + ), + ), + "xargs": _Spec( + short="0a:d:eE:iI:L:ln:P:prs:txo", + long=( + "arg-file=", + "delimiter=", + "eof=", + "interactive", + "max-args=", + "max-chars=", + "max-lines=", + "max-procs=", + "no-run-if-empty", + "null", + "open-tty", + "replace=", + "show-limits", + "verbose", + ), ), - "time": frozenset({"-f", "-o", "--format", "--output"}), - "timeout": frozenset({"-k", "-s", "--kill-after", "--signal"}), - "watch": frozenset({"-n"}), - "xargs": frozenset({"-a", "-d", "-E", "-I", "-L", "-n", "-P", "-s"}), } -# Options whose value is a *command line* the wrapper will run, so the value is -# code rather than an argument: ``env -S "rm -rf /"`` and ``su -c "…"``. An -# interpreter's own ``-c`` is the same idea, but there the interpreter is the -# program under review; here a wrapper hides the command, so the review has to -# surface it (:func:`plyngent.tools.danger.classify_danger`). -_COMMAND_STRING_OPTS: dict[str, frozenset[str]] = { - "env": frozenset({"-S", "--split-string"}), - "su": frozenset({"-c", "--command", "--session-command"}), -} +# `` run …``: the option sets of the launchers below differ, +# so one union stands in for all of them — a value option is the only kind that +# can swallow the program, and the common flags are listed to keep them quiet. +_LAUNCHER_SPEC = _Spec( + short="C:e:F:G:m:n:p:s:w:", + long=( + "all-extras", + "all-groups", + "all-packages", + "config=", + "constraint=", + "default-index=", + "directory=", + "env=", + "env-file=", + "extra=", + "extras=", + "file=", + "frozen", + "group=", + "index=", + "locked", + "module=", + "name=", + "no-project", + "no-sync", + "offline", + "overrides=", + "package=", + "prefix=", + "project=", + "python=", + "python-version=", + "quiet", + "refresh", + "script=", + "verbose", + "with=", + "without=", + ), + launcher=True, +) -# Remote wrappers: ``ssh``/``mosh`` run their command on another machine, so the -# program under review is the one *after* the destination instead of the client. -# A remote wrapper with no command only opens a remote shell — what is typed or -# piped into that shell never reaches this scan (see the SSH note in AGENTS.md). -_REMOTE_WRAPPERS: frozenset[str] = frozenset({"mosh", "ssh"}) - -# Length of a short option (``-p``), so ``-p2222`` reads as option plus value. -_SHORT_OPTION = 2 - -# Remote-client options that consume a value (``ssh -o BatchMode=yes -p 2222``). -_REMOTE_VALUE_OPTS: dict[str, frozenset[str]] = { - "mosh": frozenset({"-p", "--client-port", "--family", "--port", "--predict", "--server", "--ssh"}), - "ssh": frozenset( - { - "-B", - "-b", - "-c", - "-D", - "-E", - "-e", - "-F", - "-I", - "-i", - "-J", - "-L", - "-l", - "-m", - "-O", - "-o", - "-P", - "-p", - "-Q", - "-R", - "-S", - "-W", - "-w", - } +# ``ssh`` / ``mosh``: the program under review is the one after the destination, +# which is the first token that is not one of their options. +_REMOTE_WRAPPERS: dict[str, _Spec] = { + "mosh": _Spec( + short="46hP:p:v", + long=("client-port=", "family=", "help", "no-ssh-pty", "port=", "predict=", "server=", "ssh="), + remote=True, + ), + "ssh": _Spec( + short="46AaCfgGKkMNnqSTtVvXxYysB:b:c:D:E:e:F:I:i:J:L:l:m:O:o:P:p:Q:R:S:W:w:", + remote=True, ), } @@ -291,6 +353,8 @@ class UnwrappedCommand: command_strings: tuple[CommandString, ...] = () #: The remote hop, when the command runs on another machine. remote: RemoteTarget | None = None + #: An option the wrapper's spec does not know: the program behind it is a guess. + unparsed: str = "" @property def command(self) -> tuple[str, ...]: @@ -342,36 +406,39 @@ 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, and an - option that carries a command line is collected in ``command_strings``. + them; their own options and ``FOO=bar`` assignments are read by ``getopt`` + (:func:`_parse_options`), and an option that carries a command line is + collected in ``command_strings``. ``ssh`` / ``mosh`` are unwrapped the same way — the program under review is the one after the destination (``ssh prod rm -rf /`` runs ``rm``), and the hop is reported in :attr:`UnwrappedCommand.remote`. A client with no command opens a remote shell (``remote.shell``); nothing behind it can be resolved here. + + An option the wrapper's spec does not know stops the unwrap at that wrapper + and is reported in :attr:`UnwrappedCommand.unparsed`: the program behind it + is unknown, so the review asks instead of guessing (the wrapper itself is + still judged, so ``sudo rm -rf /`` keeps its ``sudo`` prompt). """ parts = list(argv) wrappers: list[str] = [] command_strings: list[CommandString] = [] remote: RemoteTarget | None = None + unparsed = "" index = 0 while index < len(parts): name = basename(parts[index]) - if name in _REMOTE_WRAPPERS: - start = _skip_remote(name, parts, index + 1) - if start is not None: - remote = start.target - if start.index is None: - break # a remote shell: the client is the whole command - wrappers.append(name) - index = start.index - continue - nxt = _skip_wrapper(name, parts, index + 1, command_strings=command_strings) - if nxt is None: + step = _step(name, parts, index + 1, command_strings=command_strings) + if step.unparsed: + unparsed = f"{name} {step.unparsed}" + break + if step.remote is not None: + remote = step.remote + if step.index is None: break wrappers.append(name) - index = nxt + index = step.index base = basename(parts[index]) if index < len(parts) else "" chain = dict.fromkeys((*wrappers, base)) return UnwrappedCommand( @@ -382,155 +449,107 @@ def unwrap_command(argv: Sequence[str]) -> UnwrappedCommand: self_review=tuple(name for name in chain if name in _SELF_REVIEW), command_strings=tuple(command_strings), remote=remote, + unparsed=unparsed, ) @dataclass(frozen=True) -class _RemoteStart: - """Where a remote client's own command starts, and the hop it reaches.""" +class _Parsed: + """A wrapper's own options, where its command starts, and what it did not know.""" - #: Index of the remote command, or ``None`` when only a shell opens. - index: int | None - target: RemoteTarget + options: tuple[tuple[str, str], ...] = () + index: int = 0 + unparsed: str = "" -def _skip_remote(name: str, parts: list[str], start: int) -> _RemoteStart | None: - """Where the command *name* runs elsewhere starts; ``None`` without a hop. +@dataclass(frozen=True) +class _Step: + """One wrapper resolved: its command, the hop it reaches, or its parse error.""" - The destination is the first token that is not one of the client's own - options (``ssh -o BatchMode=yes -p 2222 [user@]host …``); the command follows - it, or — for ``mosh`` — follows an explicit ``--``. No token after the - destination means the client just opens a remote shell. - """ - value_opts = _REMOTE_VALUE_OPTS.get(name, frozenset()) + index: int | None = None + unparsed: str = "" + remote: RemoteTarget | None = None + + +def _step(name: str, parts: list[str], start: int, *, command_strings: list[CommandString]) -> _Step: + """Resolve the wrapper *name* at ``parts[start - 1]`` past its own options.""" + spec = _spec_for(name) + if spec is None: + return _Step() index = start - host = "" - batch = False - while index < len(parts): - token = parts[index] - if not token.startswith("-"): - host = token - index += 1 - break - option, value, consumed = _remote_option(token, parts, index, value_opts=value_opts) - index += consumed - if option == "-o" and value is not None and value.partition("=")[0].strip().lower() == "batchmode": - batch = value.partition("=")[2].strip().lower() in {"1", "on", "true", "yes"} - if not host: - return None + if spec.launcher: + if index >= len(parts) or parts[index] != "run": + return _Step() # ``npm install``: the launcher is the whole command + index += 1 + parsed = _parse_options(parts, index, spec) + if parsed.unparsed: + return _Step(unparsed=parsed.unparsed) + for option, value in parsed.options: + if option in spec.command_opts: + command_strings.append(CommandString(wrapper=name, option=option, text=value)) + index = _skip_prefixes(parts, parsed.index, spec) + if spec.remote: + return _remote_step(name, parts, index, parsed.options) + return _Step(index=index) if index < len(parts) else _Step() + + +def _spec_for(name: str) -> _Spec | None: + """The option spec of a wrapper, or ``None`` when *name* wraps nothing.""" + spec = _WRAPPERS.get(name) or _REMOTE_WRAPPERS.get(name) + if spec is None and name in _RUN_LAUNCHERS: + return _LAUNCHER_SPEC + return spec + + +def _skip_prefixes(parts: list[str], index: int, spec: _Spec) -> int: + """Skip the ``FOO=bar`` assignments and the bare duration a wrapper takes first.""" + while index < len(parts) and _ASSIGNMENT.match(parts[index]): + index += 1 + if spec.duration and index < len(parts) and _DURATION.fullmatch(parts[index]): + index += 1 + return index + + +def _parse_options(parts: list[str], start: int, spec: _Spec) -> _Parsed: + """Read a wrapper's own options, stopping at the first token that is not one. + + ``getopt`` (POSIX order — not ``gnu_getopt``, which permutes) consumes the + wrapper's options and leaves the rest, so the length of that remainder is + where the wrapped program starts. Jammed forms (``-p2222``, + ``-oBatchMode=yes``, ``--with=pkg``) are its business, not ours. An option + the spec does not know is *not* skipped as a flag — that is how a flag ended + up eating the program — but reported, so the review can ask. + """ + try: + options, rest = getopt.getopt(list(parts[start:]), spec.short, list(spec.long)) + except getopt.GetoptError as exc: + return _Parsed(index=start, unparsed=f"(cannot read its options: {exc})") + return _Parsed(options=tuple(options), index=len(parts) - len(rest)) + + +def _remote_step(name: str, parts: list[str], index: int, options: Sequence[tuple[str, str]]) -> _Step: + """The hop a remote client reaches, and where its command starts (if any). + + The destination is the first token that is not one of the client's options; + the command follows it, or — for ``mosh`` — follows an explicit ``--``. No + token after the destination means the client just opens a remote shell. + """ + if index >= len(parts): + return _Step() + host = parts[index] + index += 1 if name == "mosh" and index < len(parts) and parts[index] == "--": index += 1 + batch = _batch_mode(options) if index >= len(parts): - return _RemoteStart(index=None, target=RemoteTarget(wrapper=name, host=host, batch=batch, shell=True)) - return _RemoteStart(index=index, target=RemoteTarget(wrapper=name, host=host, batch=batch)) + return _Step(remote=RemoteTarget(wrapper=name, host=host, batch=batch, shell=True)) + return _Step(index=index, remote=RemoteTarget(wrapper=name, host=host, batch=batch)) -def _remote_option( - token: str, - parts: list[str], - index: int, - *, - value_opts: frozenset[str], -) -> tuple[str, str | None, int]: - """The option *token* names, its value, and how many tokens it consumed. - - Jammed forms carry the value in the token itself: ``-p2222`` and - ``-oBatchMode=yes`` are one token, ``--ssh='ssh -p 2222'`` too. - """ - if token.startswith("--"): - option, equal, attached = token.partition("=") - if equal: - return option, attached, 1 - else: - option = token[:_SHORT_OPTION] - if option in value_opts and len(token) > _SHORT_OPTION: - return option, token[_SHORT_OPTION:], 1 - if option not in value_opts: - return token, None, 1 - if index + 1 < len(parts): - return option, parts[index + 1], 2 - return option, "", 1 - - -def _skip_wrapper( - name: str, - parts: list[str], - start: int, - *, - command_strings: list[CommandString], -) -> 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, - name=name, - value_opts=_VALUE_OPTS.get(name, frozenset()), - leading_duration=name in _DURATION_WRAPPERS, - command_strings=command_strings, - ) - 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, - name=name, - value_opts=_RUN_VALUE_OPTS, - leading_duration=False, - command_strings=command_strings, - ) - return nxt if nxt < len(parts) else None - return None - - -def _record_command_string( - name: str, - option: str, - text: str, - command_strings: list[CommandString], -) -> None: - """Note *text* when *option* hands a command line to wrapper *name*.""" - if option in _COMMAND_STRING_OPTS.get(name, frozenset()): - command_strings.append(CommandString(wrapper=name, option=option, text=text)) - - -def _skip_options( - parts: list[str], - start: int, - *, - name: str, - value_opts: frozenset[str], - leading_duration: bool, - command_strings: list[CommandString], -) -> 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: - value = parts[index] if index < len(parts) else "" - _record_command_string(name, token, value, command_strings) - index += 1 - continue - # Jammed forms: ``--split-string=…`` and ``-S…`` carry the value in - # the token itself, so a command line may hide there too. - option, equal, attached = token.partition("=") - if equal: - _record_command_string(name, option, attached, command_strings) - elif token[:2] in _COMMAND_STRING_OPTS.get(name, frozenset()): - _record_command_string(name, token[:2], token[2:], command_strings) - continue - if leading_duration and not took_duration and _DURATION.fullmatch(token): - took_duration = True - index += 1 - continue - return index - return index +def _batch_mode(options: Sequence[tuple[str, str]]) -> bool: + """Whether ``-o BatchMode=yes`` was given: the client never prompts.""" + for option, value in options: + key, _, setting = value.partition("=") + if option == "-o" and key.strip().lower() == "batchmode": + return setting.strip().lower() in {"1", "on", "true", "yes"} + return False diff --git a/src/plyngent/tools/danger.py b/src/plyngent/tools/danger.py index d1b6056..3f5f17e 100644 --- a/src/plyngent/tools/danger.py +++ b/src/plyngent/tools/danger.py @@ -77,6 +77,8 @@ def _command_reason(argv: Sequence[str], *, via: str) -> DangerReason | None: findings.append(f"remote shell on {remote.host!r} ({remote.wrapper} with no command; its input is not checked)") elif remote is not None and remote.batch: findings.append(f"remote command on {remote.host!r} (BatchMode=yes, no prompt on either end)") + if scan.unparsed: + findings.append(f"options not understood {scan.unparsed} — the wrapped program is a guess") findings.extend(f"{span.label} runs a command line" for span in scan.command_strings) findings = list(dict.fromkeys(findings)) if not findings: