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

core/tools: read each wrapper's options with getopt

Browse Source 
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.
This commit is contained in:
worldmozara
2026-10-23 10:47:36 +08:00
parent 3cb4d490ed
commit 12236cffb5
2 changed files with 287 additions and 266 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/plyngent/tools/command_scan.py
+285 -266
View File
@@ -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"})
# ``<launcher> run <program> …``: 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
#: ``<launcher> run <program>``: 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"}),
}
# ``<launcher> run <program> …``: 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 <new flag> 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
src/plyngent/tools/danger.py
+2
View File
@@ -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: