mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 23:47:54 +08:00
core/tools: review a wrapper option that carries a command line
`env -S "rm -rf /"` ran a command nothing looked at: the option belongs to the wrapper, so the scan skipped it and the denylist found no program to block while the soft confirm had no interpreter to review. Collect those values — `env -S`/`--split-string` and `su -c`/`--command`/`--session-command`, jammed `--opt=value` and `-Svalue` forms included — and confirm the call the way an interpreter's `-c` one-liner is confirmed, printing the text under its own `env -S` label. Recursing into the text for a hard denylist is left out for now: the review asks the human, who sees the command.
This commit is contained in:
@@ -16,7 +16,7 @@ from typing import TYPE_CHECKING
|
|||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from collections.abc import Sequence
|
from collections.abc import Sequence
|
||||||
|
|
||||||
__all__ = ["UnwrappedCommand", "basename", "is_interpreter", "unwrap_command"]
|
__all__ = ["CommandString", "UnwrappedCommand", "basename", "is_interpreter", "unwrap_command"]
|
||||||
|
|
||||||
# Wrappers that run the program named on their own command line.
|
# Wrappers that run the program named on their own command line.
|
||||||
_ENV_WRAPPERS: frozenset[str] = frozenset(
|
_ENV_WRAPPERS: frozenset[str] = frozenset(
|
||||||
@@ -151,6 +151,16 @@ _VALUE_OPTS: dict[str, frozenset[str]] = {
|
|||||||
"xargs": frozenset({"-a", "-d", "-E", "-I", "-L", "-n", "-P", "-s"}),
|
"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 options that consume a value (``uv run --with rich python``).
|
# Launcher options that consume a value (``uv run --with rich python``).
|
||||||
_RUN_VALUE_OPTS: frozenset[str] = frozenset(
|
_RUN_VALUE_OPTS: frozenset[str] = frozenset(
|
||||||
{
|
{
|
||||||
@@ -189,6 +199,20 @@ _RUN_VALUE_OPTS: frozenset[str] = frozenset(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class CommandString:
|
||||||
|
"""A command line a wrapper took as an option *value* (``env -S …``)."""
|
||||||
|
|
||||||
|
wrapper: str
|
||||||
|
option: str
|
||||||
|
text: str
|
||||||
|
|
||||||
|
@property
|
||||||
|
def label(self) -> str:
|
||||||
|
"""``env -S`` — how the option is named in review output."""
|
||||||
|
return f"{self.wrapper} {self.option}"
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class UnwrappedCommand:
|
class UnwrappedCommand:
|
||||||
"""A command plus the program it really runs, past wrappers and launchers."""
|
"""A command plus the program it really runs, past wrappers and launchers."""
|
||||||
@@ -198,6 +222,7 @@ class UnwrappedCommand:
|
|||||||
wrappers: tuple[str, ...]
|
wrappers: tuple[str, ...]
|
||||||
base: str
|
base: str
|
||||||
self_review: tuple[str, ...]
|
self_review: tuple[str, ...]
|
||||||
|
command_strings: tuple[CommandString, ...] = ()
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def command(self) -> tuple[str, ...]:
|
def command(self) -> tuple[str, ...]:
|
||||||
@@ -224,14 +249,16 @@ def unwrap_command(argv: Sequence[str]) -> UnwrappedCommand:
|
|||||||
"""Resolve *argv* to the program behind any wrappers and ``run`` launchers.
|
"""Resolve *argv* to the program behind any wrappers and ``run`` launchers.
|
||||||
|
|
||||||
Wrappers are unwrapped outer → inner as long as a program still follows
|
Wrappers are unwrapped outer → inner as long as a program still follows
|
||||||
them; their own options and ``FOO=bar`` assignments are skipped.
|
them; their own options and ``FOO=bar`` assignments are skipped, and an
|
||||||
|
option that carries a command line is collected in ``command_strings``.
|
||||||
"""
|
"""
|
||||||
parts = list(argv)
|
parts = list(argv)
|
||||||
wrappers: list[str] = []
|
wrappers: list[str] = []
|
||||||
|
command_strings: list[CommandString] = []
|
||||||
index = 0
|
index = 0
|
||||||
while index < len(parts):
|
while index < len(parts):
|
||||||
name = basename(parts[index])
|
name = basename(parts[index])
|
||||||
nxt = _skip_wrapper(name, parts, index + 1)
|
nxt = _skip_wrapper(name, parts, index + 1, command_strings=command_strings)
|
||||||
if nxt is None:
|
if nxt is None:
|
||||||
break
|
break
|
||||||
wrappers.append(name)
|
wrappers.append(name)
|
||||||
@@ -244,31 +271,60 @@ def unwrap_command(argv: Sequence[str]) -> UnwrappedCommand:
|
|||||||
wrappers=tuple(wrappers),
|
wrappers=tuple(wrappers),
|
||||||
base=base,
|
base=base,
|
||||||
self_review=tuple(name for name in chain if name in _SELF_REVIEW),
|
self_review=tuple(name for name in chain if name in _SELF_REVIEW),
|
||||||
|
command_strings=tuple(command_strings),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def _skip_wrapper(name: str, parts: list[str], start: int) -> int | None:
|
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."""
|
"""Index of the program *name* wraps, or ``None`` when it is the whole command."""
|
||||||
if name in _ENV_WRAPPERS:
|
if name in _ENV_WRAPPERS:
|
||||||
nxt = _skip_options(
|
nxt = _skip_options(
|
||||||
parts,
|
parts,
|
||||||
start,
|
start,
|
||||||
|
name=name,
|
||||||
value_opts=_VALUE_OPTS.get(name, frozenset()),
|
value_opts=_VALUE_OPTS.get(name, frozenset()),
|
||||||
leading_duration=name in _DURATION_WRAPPERS,
|
leading_duration=name in _DURATION_WRAPPERS,
|
||||||
|
command_strings=command_strings,
|
||||||
)
|
)
|
||||||
return nxt if nxt < len(parts) else None
|
return nxt if nxt < len(parts) else None
|
||||||
if name in _RUN_LAUNCHERS and start < len(parts) and parts[start] == "run":
|
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)
|
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 nxt if nxt < len(parts) else None
|
||||||
return 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(
|
def _skip_options(
|
||||||
parts: list[str],
|
parts: list[str],
|
||||||
start: int,
|
start: int,
|
||||||
*,
|
*,
|
||||||
|
name: str,
|
||||||
value_opts: frozenset[str],
|
value_opts: frozenset[str],
|
||||||
leading_duration: bool,
|
leading_duration: bool,
|
||||||
|
command_strings: list[CommandString],
|
||||||
) -> int:
|
) -> int:
|
||||||
"""Index of the first token after a wrapper's own options and assignments."""
|
"""Index of the first token after a wrapper's own options and assignments."""
|
||||||
index = start
|
index = start
|
||||||
@@ -283,7 +339,17 @@ def _skip_options(
|
|||||||
if token.startswith("-"):
|
if token.startswith("-"):
|
||||||
index += 1
|
index += 1
|
||||||
if token in value_opts:
|
if token in value_opts:
|
||||||
|
value = parts[index] if index < len(parts) else ""
|
||||||
|
_record_command_string(name, token, value, command_strings)
|
||||||
index += 1
|
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
|
continue
|
||||||
if leading_duration and not took_duration and _DURATION.fullmatch(token):
|
if leading_duration and not took_duration and _DURATION.fullmatch(token):
|
||||||
took_duration = True
|
took_duration = True
|
||||||
|
|||||||
@@ -61,13 +61,18 @@ def _command_reason(argv: Sequence[str], *, via: str) -> str | None:
|
|||||||
forms (``env FOO=1 python …``, ``nohup pdm run python …``), and an
|
forms (``env FOO=1 python …``, ``nohup pdm run python …``), and an
|
||||||
interpreter is confirmed the same way however it runs — a bare shell, a
|
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
|
script (``python x.py``), or a ``-c`` one-liner, whose code is printed below
|
||||||
``command:`` instead of inline.
|
``command:`` instead of inline. A wrapper option that takes a command line
|
||||||
|
(``env -S "…"``, ``su -c "…"``) hides code the same way, so it is confirmed
|
||||||
|
too, with the text it carries printed under its own ``<wrapper> <option>:``
|
||||||
|
label.
|
||||||
"""
|
"""
|
||||||
scan = unwrap_command(argv)
|
scan = unwrap_command(argv)
|
||||||
interpreter = is_interpreter(scan.base)
|
interpreter = is_interpreter(scan.base)
|
||||||
findings = [f"{name} (detached run)" for name in scan.self_review]
|
findings = [f"{name} (detached run)" for name in scan.self_review]
|
||||||
if interpreter:
|
if interpreter:
|
||||||
findings.append(f"interpreter {scan.base!r}")
|
findings.append(f"interpreter {scan.base!r}")
|
||||||
|
findings.extend(f"{span.label} runs a command line" for span in scan.command_strings)
|
||||||
|
findings = list(dict.fromkeys(findings))
|
||||||
if not findings:
|
if not findings:
|
||||||
return None
|
return None
|
||||||
code = _dash_c_code(scan.command) if interpreter else None
|
code = _dash_c_code(scan.command) if interpreter else None
|
||||||
@@ -76,6 +81,8 @@ def _command_reason(argv: Sequence[str], *, via: str) -> str | None:
|
|||||||
reason = f"{via}: {' + '.join(findings)} — review before allow\n argv: {display}"
|
reason = f"{via}: {' + '.join(findings)} — review before allow\n argv: {display}"
|
||||||
if code is not None:
|
if code is not None:
|
||||||
reason += f"\n command:\n{_indent_block(code[1])}"
|
reason += f"\n command:\n{_indent_block(code[1])}"
|
||||||
|
for span in scan.command_strings:
|
||||||
|
reason += f"\n {span.label}:\n{_indent_block(span.text)}"
|
||||||
return reason
|
return reason
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -92,6 +92,31 @@ def test_unwrap_value_free_flags_do_not_swallow_the_program() -> None:
|
|||||||
assert scan.command[0] == base, argv
|
assert scan.command[0] == base, argv
|
||||||
|
|
||||||
|
|
||||||
|
def test_unwrap_collects_command_line_option_values() -> None:
|
||||||
|
"""A wrapper option that takes a command line is reported, not just skipped.
|
||||||
|
|
||||||
|
``env -S "rm -rf /"`` hands over a command the review must see; the value is
|
||||||
|
code, so it must not be treated as an argument (or as the program).
|
||||||
|
"""
|
||||||
|
scan = unwrap_command(["env", "-S", "rm -rf /"])
|
||||||
|
assert scan.base == "env"
|
||||||
|
assert [(span.wrapper, span.option, span.text) for span in scan.command_strings] == [("env", "-S", "rm -rf /")]
|
||||||
|
assert scan.command_strings[0].label == "env -S"
|
||||||
|
|
||||||
|
for argv, option, text in (
|
||||||
|
(["env", "--split-string=rm -rf /"], "--split-string", "rm -rf /"),
|
||||||
|
(["env", "-Srm -rf /"], "-S", "rm -rf /"),
|
||||||
|
(["su", "-c", "rm -rf /"], "-c", "rm -rf /"),
|
||||||
|
(["su", "--session-command=rm -rf /"], "--session-command", "rm -rf /"),
|
||||||
|
):
|
||||||
|
spans = unwrap_command(argv).command_strings
|
||||||
|
assert [(span.option, span.text) for span in spans] == [(option, text)], argv
|
||||||
|
|
||||||
|
# Plain wrapper options stay arguments: no command line to review.
|
||||||
|
assert unwrap_command(["env", "-u", "FOO=1", "rm", "-rf", "/"]).command_strings == ()
|
||||||
|
assert unwrap_command(["env", "FOO=1", "git", "-c", "x=y", "log"]).command_strings == ()
|
||||||
|
|
||||||
|
|
||||||
def test_unwrap_empty_argv() -> None:
|
def test_unwrap_empty_argv() -> None:
|
||||||
scan = unwrap_command([])
|
scan = unwrap_command([])
|
||||||
assert scan.base == ""
|
assert scan.base == ""
|
||||||
|
|||||||
@@ -144,6 +144,26 @@ def test_value_free_flags_do_not_hide_the_interpreter() -> None:
|
|||||||
assert "print(1)" in reason
|
assert "print(1)" in reason
|
||||||
|
|
||||||
|
|
||||||
|
def test_wrapper_command_line_option_is_reviewed() -> None:
|
||||||
|
"""`env -S` / `su -c` carry a command line, so they are confirmed like `-c`."""
|
||||||
|
reason = classify_danger("run_argv", {"argv": ["env", "-S", "rm -rf /"]})
|
||||||
|
assert reason is not None
|
||||||
|
assert "env -S runs a command line" in reason
|
||||||
|
assert " env -S:\n rm -rf /" in reason
|
||||||
|
|
||||||
|
# Plain wrapper options are not a review trigger.
|
||||||
|
assert classify_danger("run_argv", {"argv": ["env", "-u", "FOO=1", "git", "status"]}) is None
|
||||||
|
assert classify_danger("run_argv", {"argv": ["env", "FOO=1", "git", "-c", "x=y", "log"]}) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_wrapper_command_line_value_needs_no_interpreter() -> None:
|
||||||
|
"""The wrapper's value is reviewed even when no interpreter is resolved."""
|
||||||
|
reason = classify_danger("run_argv", {"argv": ["env", "-S", "python -O script.py"]})
|
||||||
|
assert reason is not None
|
||||||
|
assert "env -S runs a command line" in reason
|
||||||
|
assert "python -O script.py" in reason
|
||||||
|
|
||||||
|
|
||||||
def test_only_the_interpreter_dash_c_becomes_a_placeholder() -> None:
|
def test_only_the_interpreter_dash_c_becomes_a_placeholder() -> None:
|
||||||
reason = classify_danger("run_argv", {"argv": ["ionice", "-c", "2", "python", "-c", "print(1)"]})
|
reason = classify_danger("run_argv", {"argv": ["ionice", "-c", "2", "python", "-c", "print(1)"]})
|
||||||
assert reason is not None
|
assert reason is not None
|
||||||
|
|||||||
Reference in New Issue
Block a user