From 3a60f1effed92f015d5cd384dccb9d3d12cd1d01 Mon Sep 17 00:00:00 2001 From: Hugo Date: Mon, 5 Oct 2026 11:41:01 +0200 Subject: [PATCH 1/8] fix(command-injection): judge process commands by what the input controls Every tainted argument of subprocess calls was a high command injection, whether or not a shell runs it. The rule now reads the command a flow reaches: a shell string, a string naming the program or the program itself stays high; an element of an argument list run without a shell is an option, medium unless a documented option of the program runs a command it chooses; a constant prefix that keeps the input from being an option, or an operand after -- for a program that honours it, is refuted; a value after an option the model does not describe stays a hotspot to review. TaintDetector gains an assess hook returning the verdict and the severity, since a hotspot verdict only lowers the confidence. List and tuple displays keep the place of their unpacked elements in the IR, so that the order of a command survives *(...) parts. Closes #208 --- docs/plugins.md | 8 + docs/usage.md | 2 +- .../command_injection/command_injection.py | 357 +++++++++++++++++- .../security/command_injection/plugin.toml | 2 +- src/coretrace_python/ir/lowering.py | 5 +- src/coretrace_python/ir/model.py | 8 + src/coretrace_python/plugins/__init__.py | 3 +- src/coretrace_python/plugins/detectors.py | 26 +- src/coretrace_python/taint/commands.py | 45 +++ tests/test_command_arguments.py | 110 ++++++ tests/test_detectors.py | 10 +- 11 files changed, 565 insertions(+), 11 deletions(-) create mode 100644 src/coretrace_python/taint/commands.py create mode 100644 tests/test_command_arguments.py diff --git a/docs/plugins.md b/docs/plugins.md index 5d34663..c843e77 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -127,6 +127,14 @@ judges by its own criteria. The refutation's verdict, which other consumers of the flow such as the advisory correlation read, does not change; a rule reading more analyses declares them in its `requires`. +A rule that tells cases of its sinks apart by severity too overrides `assess(ctx, +function, flow, verdict)`, which returns an `Assessment` of the verdict and the +severity; by default it pairs `judge`'s verdict with the rule's `severity`. A `hotspot` +verdict lowers the confidence of the finding, never its severity. `command-injection` +reads there the command a flow reaches (`coretrace_python.taint.commands`): a shell +string or the program is `high`; an element of an argument list run without a shell is +an option, `medium` unless a documented option of the program runs a command it chooses. + ### Security models: `ModelPlugin` Declares sources, sinks and sanitizers for a library or a framework. A model plugin diff --git a/docs/usage.md b/docs/usage.md index d9e01e6..485537e 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -115,7 +115,7 @@ variable is not treated that way yet. | Rule | Reached sink | |---|---| -| `command-injection` | Shell or process execution (`os.system`, `subprocess` with a string, …). | +| `command-injection` | Shell or process execution (`os.system`, `subprocess` with a string, …), judged on what the input controls in the command. A string a shell runs, a string naming the program or the program itself is `high`. An element of an argument list run without a shell can only be read as an option: `medium`, or `high` when a documented option of the program runs a command it chooses (`git push --receive-pack=…`, `ssh -o ProxyCommand=…`, `sh -c`). A constant prefix that keeps the input from being an option (`f"v{version}"`), or an operand after `--` for a program that honours it, is refuted. A value after an option that may take one, where the program's options are not known, stays a hotspot to review. | | `code-injection` | The code `eval` or `exec` runs; the globals and locals passed with it are data. `dangerous-eval` reports every such call, this rule the ones attacker input reaches. | | `sql-injection` | A database statement (`cursor.execute`, SQLAlchemy `text`, Django `raw`, …). Parameters of a parameterised query are not statements. | | `nosql-injection` | A NoSQL query, such as a MongoDB filter, given attacker input that may hold a structure: a JSON body giving `{"$ne": null}` matches any value. The input that may hold a structure is JSON, the raw bodies a `json.loads` decodes, files, WebSocket messages and standard input. Text cannot hold an operator: form fields, query parameters, headers, cookies, URL parameters, the command line, the environment, and an entry-point parameter annotated with scalars (`str`, `int`, `Optional[str]`, `Annotated[str, Query()]`, `list[str]`, …), which FastAPI validates. A Django, REST framework, aiohttp or tornado view reads a request object: its text attributes (`GET`, `POST`, `headers`, `query_params`, `query`, `match_info`, `arguments`, …) and the URL parameters after it are text, its body, files and `data` are not. Limits: a helper function the request is passed to reads it as a whole; the keys of a dict are not followed, so a string concatenated into `$where` is not reported. The analyzer ships no NoSQL sink: model plugins declare them. | diff --git a/src/coretrace_python/bundled/security/command_injection/command_injection.py b/src/coretrace_python/bundled/security/command_injection/command_injection.py index f43942b..da9e61b 100644 --- a/src/coretrace_python/bundled/security/command_injection/command_injection.py +++ b/src/coretrace_python/bundled/security/command_injection/command_injection.py @@ -1,12 +1,132 @@ -"""Command injection: attacker-controlled input reaching a COMMAND sink.""" +"""Command injection: attacker-controlled input reaching a COMMAND sink. + +What decides the result is what the attacker still controls in the command a process is +started with (issue #208). A string that a shell interprets, a string naming the program +or the program itself is a command injection, ``high``. An element of an argument list +run without a shell injects at most an option: ``medium``, or ``high`` when ``OPTIONS`` +establishes that the program runs a command chosen through that option. A constant +prefix is judged by what it leaves to the attacker: ``f"v{x}"`` is no option, the value +of ``f"--output={x}"`` belongs to a fixed option, ``f"--{x}"`` leaves the option open. +A value after an option belongs to that option only when ``OPTIONS`` establishes that the +option takes it, and is a free element when ``OPTIONS`` establishes that the options before +it take none (``rm -rf``, read as getopt reads grouped short options); otherwise the +finding is kept for review. A ``hotspot`` verdict lowers +the confidence, never the severity, which each case sets. +""" from __future__ import annotations +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import PurePosixPath from typing import ClassVar +from coretrace_python.abstract.strings import ModuleStringsAnalysis, leading_text +from coretrace_python.analysis import AnyAnalysis from coretrace_python.findings import Severity -from coretrace_python.plugins import TaintDetector -from coretrace_python.taint import TaintKind +from coretrace_python.findings.refutation import Status, Verdict +from coretrace_python.hir import nodes +from coretrace_python.interprocedural import Arguments +from coretrace_python.ir.defuse import DefUseAnalysis +from coretrace_python.ir.model import Instruction, Value +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.plugins import Assessment, PluginContext, TaintDetector +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import TaintAnalysis, TaintFlow, TaintKind +from coretrace_python.taint.commands import CommandPart, command_parts + + +@dataclass(frozen=True) +class CommandOption: + """An option of ``program``, after one of ``subcommands`` when any are given. An + option that ``takes_value`` reads it from the next element, or after ``=`` or the + short option in the same element; with ``runs``, the program runs a command that + value chooses (``runs`` says how), and ``single`` says one element can carry the + option and its value. An option that takes no value leaves the next element free.""" + + program: str + option: str + runs: str | None = None + subcommands: tuple[str, ...] = () + single: bool = True + takes_value: bool = True + + +def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: + return tuple(CommandOption(program, option, takes_value=False) for option in options) + + +# Documented behaviours only: an option missing here is neither established to be +# dangerous nor to take no value. +OPTIONS: tuple[CommandOption, ...] = ( + CommandOption( + "git", "-c", "sets any configuration variable, core.sshCommand included", single=False + ), + *( + CommandOption("git", option, "runs the program it names", subcommands) + for option, subcommands in ( + ("--upload-pack", ("clone", "fetch", "pull", "ls-remote")), + ("--receive-pack", ("push",)), + ("--exec", ("push",)), + ) + ), + CommandOption("ssh", "-o", "sets options such as ProxyCommand, which runs a command"), + *_flags("ssh", *(f"-{letter}" for letter in "46AaCfGgKkMNnqsTtVvXxYy")), + *_flags( + "rm", + "-d", + "-f", + "-I", + "-i", + "-R", + "-r", + "-v", + "--dir", + "--force", + "--recursive", + "--verbose", + ), + CommandOption( + "tar", "--to-command", "pipes each extracted file to the command it names (GNU tar)" + ), + CommandOption("tar", "--checkpoint-action", "runs the command of an exec= action (GNU tar)"), + CommandOption("tar", "--use-compress-program", "runs the compressor it names"), + CommandOption("rsync", "-e", "runs the remote shell it names"), + CommandOption("rsync", "--rsh", "runs the remote shell it names"), + *( + CommandOption("find", option, "runs the command that follows", single=False) + for option in ("-exec", "-execdir", "-ok", "-okdir") + ), + *( + CommandOption(shell, "-c", "runs its value as a shell command", single=False) + for shell in ("sh", "bash", "dash", "zsh", "ksh") + ), + *( + CommandOption(python, "-c", "runs its value as Python code") + for python in ("python", "python3") + ), +) +# Programs documented to read ``--`` as the end of their options. +END_OF_OPTIONS = frozenset({"git", "ssh", "tar", "rsync"}) + +_PROCESS_STARTERS = frozenset( + SymbolId(f"python.subprocess.{name}") + for name in ("run", "call", "check_call", "check_output", "Popen") +) +_COMMAND_ARGUMENT = frozenset({(0, None), (None, "args")}) +_SHELL, _NO_SHELL = frozenset({"True", "1"}), frozenset({"False", "None", "0"}) + + +@dataclass(frozen=True) +class _Reading: + """What one tainted element of a command lets the attacker do: run a ``command`` + (the program, ``high``), give an option at ``severity``, possibly only ``to_review``, + or nothing, with ``severity`` None.""" + + reason: str + severity: Severity | None + command: bool = False + to_review: bool = False class CommandInjectionPlugin(TaintDetector): @@ -15,3 +135,234 @@ class CommandInjectionPlugin(TaintDetector): kind: ClassVar[TaintKind] = TaintKind.COMMAND severity: ClassVar[Severity] = Severity.HIGH title: ClassVar[str] = "Command injection" + requires: ClassVar[frozenset[AnyAnalysis]] = TaintDetector.requires | { + SSAAnalysis, + DefUseAnalysis, + ModuleStringsAnalysis, + } + + def assess( + self, ctx: PluginContext, function: nodes.Function, flow: TaintFlow, verdict: Verdict + ) -> Assessment: + default = super().assess(ctx, function, flow, verdict) + if flow.sink.symbol not in _PROCESS_STARTERS or flow.through is not None: + return default + if not flow.passed_as & _COMMAND_ARGUMENT or flow.sink_arguments is None: + return default + shell = _shell(flow.sink_arguments) + if shell is True: + return default + if shell is None: + return Assessment( + _to_review( + default.verdict, "an expression chooses whether a shell runs the command" + ), + Severity.HIGH, + ) + ssa = ctx.get(SSAAnalysis, function) + defs = { + i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None + } + parts = command_parts(flow.argument, defs, ctx.get(DefUseAnalysis, function)) + if parts is None: + return default + taint = ctx.get(TaintAnalysis, function) + command = _Command(parts, defs, ctx.get(ModuleStringsAnalysis)) + readings = [ + command.read(index) + for index, part in enumerate(parts) + if flow.source in taint.taint(part.value).sources + and taint.taint(part.value).kinds & TaintKind.COMMAND + ] + if not readings or any(reading.command for reading in readings): + return default + injected = [reading for reading in readings if reading.severity is not None] + if not injected: + return Assessment( + Verdict(flow, Status.REFUTED, "; ".join(r.reason for r in readings)), self.severity + ) + severity = max( + (reading.severity for reading in injected if reading.severity is not None), + key=lambda found: found.rank, + ) + decisive = [r for r in injected if r.severity is severity] + established = [reading for reading in decisive if not reading.to_review] + if not established: + return Assessment(_to_review(default.verdict, decisive[0].reason), severity) + return Assessment(_explained(default.verdict, established[0].reason), severity) + + +class _Command: + """The command of one call, read element by element.""" + + def __init__( + self, + parts: tuple[CommandPart, ...], + defs: Mapping[Value, Instruction], + strings: Mapping[str, str], + ) -> None: + self.parts = parts + self.defs = defs + self.strings = strings + first = self.text(0) + self.program = PurePosixPath(first).name if first is not None else None + + def text(self, index: int) -> str | None: + """The constant text of the element at ``index``, when all of it is constant.""" + + part = self.parts[index] + if not part.known: + return None + text, whole = leading_text(part.value, self.defs, self.strings) + return text if whole else None + + def read(self, index: int) -> _Reading: + part = self.parts[index] + if all(not earlier.known for earlier in self.parts[:index]): + return _Reading("the input may name the program", Severity.HIGH, command=True) + if not part.known: + return self.free(index, "the input gives whole elements") + prefix, _ = leading_text(part.value, self.defs, self.strings) + if prefix.startswith("--") and "=" in prefix: + return self.option_value(index, prefix[: prefix.index("=")]) + if prefix.startswith("-") and len(prefix) >= 2 and prefix[1] != "-": + return self.option_value(index, prefix[:2]) + if prefix.startswith("-"): + return self.free(index, f"the input completes the option {prefix!r}") + if prefix: + return _Reading( + f"the constant prefix {prefix!r} keeps the input from being an option", None + ) + before = self.text(index - 1) + if before == "--": + if self.program in END_OF_OPTIONS: + return _Reading(f"{self.program} reads the input after '--' as an operand", None) + return self.free( + index, "the input follows '--', which the program is not known to honour" + ) + if before is not None and before.startswith("-") and before != "-" and "=" not in before: + taker = self.taker(before, index) + if isinstance(taker, CommandOption): + return self.value_of(taker, before) + if taker is False: + return self.free( + index, f"the input is a whole element after {before}, which takes no value" + ) + free = self.free(index, "the input is a whole element") + return _Reading( + f"cannot tell whether {before} takes the input as its value; if not, {free.reason}", + free.severity, + to_review=True, + ) + return self.free(index, "the input is a whole element") + + def free(self, index: int, how: str) -> _Reading: + """An element the attacker may give as an option: ``high`` when an established + option of the program fits in one element there.""" + + dangerous = [ + option for option in self.applicable(index) if option.runs is not None and option.single + ] + if dangerous: + option = dangerous[0] + return _Reading( + f"{how}, and {self.program} {option.option}=... {option.runs}", Severity.HIGH + ) + return _Reading(f"{how}, so it may be read as an option", Severity.MEDIUM) + + def option_value(self, index: int, name: str) -> _Reading: + """The element starts with the option ``name`` and goes on with the input.""" + + option = self.option(name, index) + if option is None: + return _Reading(f"the input is the value of the option {name}", Severity.MEDIUM) + if not option.takes_value: + return self.free( + index, f"the input follows {name}, which takes no value, as more options" + ) + return self.value_of(option, name) + + def value_of(self, option: CommandOption, given: str) -> _Reading: + if option.runs is not None: + return _Reading( + f"the input is the value of {self.program} {given}, which {option.runs}", + Severity.HIGH, + ) + return _Reading(f"the input is the value of {self.program} {given}", Severity.MEDIUM) + + def taker(self, before: str, index: int) -> CommandOption | bool | None: + """The option ``before`` whose value is the element at ``index``; False when the + model establishes that ``before`` takes no value there, None when it cannot tell. + Short options grouped in one element (``-rf``) are read as getopt reads them: the + first that takes a value takes the rest of the element, or the next element.""" + + option = self.option(before, index) + if option is not None: + return option if option.takes_value else False + if before.startswith("--") or len(before) <= 2: + return None + letters = before[1:] + for position, letter in enumerate(letters): + option = self.option(f"-{letter}", index) + if option is None: + return None + if option.takes_value: + return option if position == len(letters) - 1 else False + return False + + def option(self, name: str, index: int) -> CommandOption | None: + return next((option for option in self.applicable(index) if option.option == name), None) + + def applicable(self, index: int) -> list[CommandOption]: + """The options of ``OPTIONS`` the element at ``index`` may give: those of the + program, after their subcommand when the command shows it before the element, or + whatever the subcommand when the command does not tell.""" + + if self.program is None: + return [] + subcommand = self.subcommand(index) + return [ + option + for option in OPTIONS + if option.program == self.program + and (not option.subcommands or subcommand is None or subcommand in option.subcommands) + ] + + def subcommand(self, index: int) -> str | None: + """The first operand before ``index``, ``""`` when there is none, or None when an + element of unknown value, or an option that may take one, comes first.""" + + for position in range(1, index): + text = self.text(position) + if text is None or text.startswith("-"): + return None + return text + return "" + + +def _shell(arguments: Arguments) -> bool | None: + """Whether a shell runs the command: True or False when the call says so with a + constant, None when an expression or unpacked keywords decide.""" + + given = arguments.given("shell") + if given is None: + return None + present, value = given + if not present: + return False + if value in _SHELL: + return True + return False if value in _NO_SHELL else None + + +def _explained(verdict: Verdict, reason: str) -> Verdict: + """``verdict`` with what the command lets the attacker do added to its evidence.""" + + return Verdict(verdict.flow, verdict.status, f"{verdict.evidence}; {reason}") + + +def _to_review(verdict: Verdict, reason: str) -> Verdict: + """A finding kept for review: a hotspot whose evidence says why, at most.""" + + status = Status.HOTSPOT if verdict.status is Status.VULNERABILITY else verdict.status + return Verdict(verdict.flow, status, f"{verdict.evidence}; {reason}") diff --git a/src/coretrace_python/bundled/security/command_injection/plugin.toml b/src/coretrace_python/bundled/security/command_injection/plugin.toml index 0c89205..e02f322 100644 --- a/src/coretrace_python/bundled/security/command_injection/plugin.toml +++ b/src/coretrace_python/bundled/security/command_injection/plugin.toml @@ -1,7 +1,7 @@ name = "command-injection" version = "1.0.0" plugin_api = ">=1,<2" -requires = ["taint.flows", "findings.refutation"] +requires = ["taint.flows", "findings.refutation", "ir.ssa", "ir.defuse", "abstract.module_strings"] provides = ["vulnerability.command-injection"] [entrypoint] diff --git a/src/coretrace_python/ir/lowering.py b/src/coretrace_python/ir/lowering.py index 0d6dde0..78d495e 100644 --- a/src/coretrace_python/ir/lowering.py +++ b/src/coretrace_python/ir/lowering.py @@ -189,7 +189,10 @@ def expression(self, node: nodes.Expression) -> Value: if isinstance(node, nodes.List | nodes.Tuple): elements, unpacked = self.spread(node.elements) builder = BuildList if isinstance(node, nodes.List) else BuildTuple - return self.emit(builder(self.new_value(), node.span, elements, unpacked)) + at = tuple( + i for i, element in enumerate(node.elements) if isinstance(element, nodes.Starred) + ) + return self.emit(builder(self.new_value(), node.span, elements, unpacked, at)) if isinstance(node, nodes.Dict): items = tuple((self.expression(k), self.expression(v)) for k, v in node.items if k is not None) unpacked = tuple(self.expression(v) for k, v in node.items if k is None) diff --git a/src/coretrace_python/ir/model.py b/src/coretrace_python/ir/model.py index 7dddd02..d6a67b0 100644 --- a/src/coretrace_python/ir/model.py +++ b/src/coretrace_python/ir/model.py @@ -134,18 +134,26 @@ class BoolOp(Operands): @dataclass(frozen=True) class BuildList(Operands): + """``unpacked_at`` gives the place of each ``*iterable`` of ``unpacked`` among all the + items of the display, so that the order of ``[1, *a, 2]`` is kept.""" + result: Value location: SourceSpan elements: tuple[Value, ...] unpacked: tuple[Value, ...] = () + unpacked_at: tuple[int, ...] = () @dataclass(frozen=True) class BuildTuple(Operands): + """``unpacked_at`` gives the place of each ``*iterable`` of ``unpacked`` among all the + items of the display, so that the order of ``[1, *a, 2]`` is kept.""" + result: Value location: SourceSpan elements: tuple[Value, ...] unpacked: tuple[Value, ...] = () + unpacked_at: tuple[int, ...] = () @dataclass(frozen=True) diff --git a/src/coretrace_python/plugins/__init__.py b/src/coretrace_python/plugins/__init__.py index 40ae2db..648f9e0 100644 --- a/src/coretrace_python/plugins/__init__.py +++ b/src/coretrace_python/plugins/__init__.py @@ -11,7 +11,7 @@ apply_refinement, run_plugins, ) -from coretrace_python.plugins.detectors import SymbolCallDetector, TaintDetector +from coretrace_python.plugins.detectors import Assessment, SymbolCallDetector, TaintDetector from coretrace_python.plugins.loader import ( IncompatiblePluginError, LoadedPlugin, @@ -35,6 +35,7 @@ __all__ = [ "PLUGIN_API_VERSION", + "Assessment", "Entrypoint", "IncompatiblePluginError", "LoadedPlugin", diff --git a/src/coretrace_python/plugins/detectors.py b/src/coretrace_python/plugins/detectors.py index aa42088..c39534d 100644 --- a/src/coretrace_python/plugins/detectors.py +++ b/src/coretrace_python/plugins/detectors.py @@ -1,7 +1,8 @@ """Reusable detector bases (architecture §15). ``TaintDetector`` turns the flows of one taint kind into findings; every security -detector is a few class attributes on top of it. ``SymbolCallDetector`` reports calls +detector is a few class attributes on top of it, and a rule that knows more about its +sinks refines the verdict (``judge``) or the severity too (``assess``). ``SymbolCallDetector`` reports calls whose callee resolves to a canonical symbol, through imports and local aliases alike, by reading the SSA form. """ @@ -9,6 +10,7 @@ from __future__ import annotations from collections.abc import Mapping, Sequence +from dataclasses import dataclass from typing import ClassVar from coretrace_python.analysis import AnyAnalysis @@ -23,6 +25,14 @@ from coretrace_python.taint import TaintAnalysis, TaintFlow, TaintKind +@dataclass(frozen=True) +class Assessment: + """A rule's verdict on one flow and the severity it reports the flow at.""" + + verdict: Verdict + severity: Severity + + class TaintDetector(Plugin): """Report every taint flow carrying ``kind`` into a sink.""" @@ -43,7 +53,8 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: verdict = verdicts.verdict(flow) if verdict.status is Status.REFUTED: continue - verdict = self.judge(ctx, function, flow, verdict) + assessment = self.assess(ctx, function, flow, verdict) + verdict = assessment.verdict if verdict.status is Status.REFUTED: continue message = f"{self.title}: {flow.source.label} input reaches {flow.sink.symbol}" @@ -65,7 +76,7 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: Finding( rule_id=self.rule_id, message=message, - severity=self.severity, + severity=assessment.severity, confidence=confidence, span=flow.location, function=function.name, @@ -81,6 +92,15 @@ def judge(self, ctx: PluginContext, function: nodes.Function, flow: TaintFlow, v return verdict + def assess( + self, ctx: PluginContext, function: nodes.Function, flow: TaintFlow, verdict: Verdict + ) -> Assessment: + """This rule's verdict on ``flow`` (``judge``'s) and its severity: the rule's own, + unless the rule tells cases of its sinks apart. A ``hotspot`` verdict lowers the + confidence of the finding, never its severity, which is set here.""" + + return Assessment(self.judge(ctx, function, flow, verdict), self.severity) + def _lower(confidence: Confidence) -> Confidence: return {Confidence.HIGH: Confidence.MEDIUM}.get(confidence, Confidence.LOW) diff --git a/src/coretrace_python/taint/commands.py b/src/coretrace_python/taint/commands.py new file mode 100644 index 0000000..48c9c5d --- /dev/null +++ b/src/coretrace_python/taint/commands.py @@ -0,0 +1,45 @@ +"""The command a process is started with, as a list or tuple display spells it: one +reading that the rules about processes share. Each element keeps its place, unpacked +literals included; an unpacked iterable of unknown contents stands for any number of +elements, perhaps none. A display that anything else may change before the call (any +other use of it) spells no command.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass + +from coretrace_python.ir.defuse import DefUse +from coretrace_python.ir.model import BuildList, BuildTuple, Instruction, Value + + +@dataclass(frozen=True) +class CommandPart: + """One element of a command, or, when ``known`` is false, an unpacked iterable of + unknown contents standing for any number of elements.""" + + value: Value + known: bool = True + + +def command_parts( + value: Value, defs: Mapping[Value, Instruction], uses: DefUse +) -> tuple[CommandPart, ...] | None: + """The elements of the command ``value`` spells, in order, or None when it is not a + list or tuple display used only there.""" + + made = defs.get(value) + if not isinstance(made, BuildList | BuildTuple) or len(uses.uses(value)) != 1: + return None + if len(made.unpacked_at) != len(made.unpacked): + return None + plain, unpacked = iter(made.elements), iter(made.unpacked) + parts: list[CommandPart] = [] + for index in range(len(made.elements) + len(made.unpacked)): + if index not in made.unpacked_at: + parts.append(CommandPart(next(plain))) + continue + iterable = next(unpacked) + nested = command_parts(iterable, defs, uses) + parts.extend(nested if nested is not None else (CommandPart(iterable, known=False),)) + return tuple(parts) diff --git a/tests/test_command_arguments.py b/tests/test_command_arguments.py new file mode 100644 index 0000000..b818a74 --- /dev/null +++ b/tests/test_command_arguments.py @@ -0,0 +1,110 @@ +"""Acceptance tests for issue #208: what the attacker controls in a process's command. + +A string that a shell interprets, a string naming the program, or the program itself is +a command injection, ``high``. An element of an argument list run without a shell +cannot inject a command, only an option: ``medium`` by default when the executable is +fixed, ``high`` when the option's dangerous behaviour is established by model data +(``git push --receive-pack=``, ``ssh -o ProxyCommand=...``, ``sh -c``). A constant +prefix is judged by what the attacker still controls: ``f"v{x}"`` is no option at all, +``f"--output={x}"`` is the value of a fixed option, ``f"--{x}"`` leaves the option open. +A value after an option is that option's value only when the model establishes that the +option consumes it, and a free element when the model establishes that the options before +it take no value (``rm -rf``); otherwise the finding stays, for review. Severity is set per case: +the ``hotspot`` verdict lowers the confidence, never the severity. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.frontend import build_hir +from coretrace_python.ir.lowering import lower_module +from coretrace_python.ir.model import BuildList, BuildTuple +from coretrace_python.source import SourceManager + +PLUGINS = Path(__file__).resolve().parent.parent / "src" / "coretrace_python" / "bundled" + +HIGH = ("high", "vulnerability", "high") +MEDIUM = ("medium", "vulnerability", "high") +HIGH_TO_REVIEW = ("high", "hotspot", "medium") +MEDIUM_TO_REVIEW = ("medium", "hotspot", "medium") + + +def judged(call: str) -> list[tuple[str, str, str]]: + """Severity, verdict and confidence of each command-injection finding on ``call``, + made with ``x`` read from the command line and ``extra`` of unknown value.""" + + text = f"import os\nimport subprocess\nimport sys\n\n\ndef run(extra):\n x = sys.argv[1]\n {call}\n" + findings = engine.check(SourceManager().add_source("app/run.py", text), [PLUGINS]) + return [ + (f.severity.value, f.metadata["verdict"], f.confidence.value) + for f in findings + if f.rule_id == "command-injection" + ] + + +@pytest.mark.parametrize( + ("call", "expected"), + [ + # 1, 2, 3, 8: the attacker reaches a shell or chooses the program. + ('subprocess.run(f"git tag {x}", shell=True)', HIGH), + ('os.system("ls " + x)', HIGH), + ("subprocess.run(x)", HIGH), + ('subprocess.run([x, "-l"])', HIGH), + ('subprocess.run(["ls"], executable=x)', HIGH), + ("subprocess.run(x, shell=extra)", HIGH_TO_REVIEW), + # 4: a free element is an option the attacker chooses. + ('subprocess.run(["ls", x])', MEDIUM), + ('subprocess.run(["git", "status", x])', MEDIUM), + ('subprocess.run(["git", "push", "origin", x])', HIGH), + ('subprocess.run(["/usr/bin/git", "push", "origin", x])', HIGH), + # 5, 5b, 5c: a constant prefix. + ('subprocess.run(["git", "tag", f"v{x}"])', None), + ('subprocess.run(["git", "commit", "-m", f"release {x}"])', None), + ('subprocess.run(["curl", f"--output={x}"])', MEDIUM), + ('subprocess.run(["git", "fetch", f"--upload-pack={x}"])', HIGH), + ('subprocess.run(["ssh", f"-o{x}", "host"])', HIGH), + ('subprocess.run(["ls", f"--{x}"])', MEDIUM), + # 6: after the end of options, for a program known to honour it. + ('subprocess.run(["git", "log", "--", x])', None), + ('subprocess.run(["ls", "--", x])', MEDIUM), + # 7: after an option that may take a value. + ('subprocess.run(["ssh", "-o", x, "host"])', HIGH), + ('subprocess.run(["sh", "-c", x])', HIGH), + ('subprocess.run(["find", ".", "-exec", x, ";"])', HIGH), + ('subprocess.run(["xcrun", "notarytool", "--keychain", x])', MEDIUM_TO_REVIEW), + ('subprocess.run(["git", "push", "--force", x])', HIGH_TO_REVIEW), + # Short options grouped in one element, read as getopt does. + ('subprocess.run(["rm", "-rf", x])', MEDIUM), + ('subprocess.run(["rm", "-r", "-f", x])', MEDIUM), + ('subprocess.run(["rm", "-rz", x])', MEDIUM_TO_REVIEW), + ('subprocess.run(["ssh", "-vo", x, "host"])', HIGH), + ('subprocess.run(["ssh", "-ov", x, "host"])', HIGH), + # Unpacked parts keep their place in the command. + ('subprocess.run(["xcrun", "notarytool", *(["--keychain", x])])', MEDIUM_TO_REVIEW), + ("subprocess.run([*extra, x])", HIGH), + ('subprocess.run(["ls", *extra, x])', MEDIUM), + ('subprocess.run(["ls", *x.split()])', MEDIUM), + ], +) +def test_the_result_follows_what_the_attacker_controls( + call: str, expected: tuple[str, str, str] | None +) -> None: + assert judged(call) == ([] if expected is None else [expected]) + + +def test_an_unpacked_element_keeps_its_place_in_a_list_or_tuple_display() -> None: + module = lower_module( + build_hir( + SourceManager().add_source("m.py", "def f(a, b):\n use([1, *a, 2, *b], (*a, 1))\n") + ) + ) + instructions = [i for block in module.functions[0].blocks for i in block.instructions] + (built_list,) = [i for i in instructions if isinstance(i, BuildList)] + (built_tuple,) = [i for i in instructions if isinstance(i, BuildTuple)] + + assert built_list.unpacked_at == (1, 3) + assert built_tuple.unpacked_at == (0,) diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 6e6c65d..59cb396 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -202,10 +202,18 @@ def test_shipped_plugins_load_with_their_manifests() -> None: } for rule in ("sql-injection", "command-injection", "code-injection", "nosql-injection", "path-traversal", "ssrf", "xss"): assert by_name[rule].provides == (f"vulnerability.{rule}",) - for rule in ("sql-injection", "command-injection", "code-injection", "nosql-injection", "path-traversal", "xss"): + for rule in ("sql-injection", "code-injection", "nosql-injection", "path-traversal", "xss"): assert by_name[rule].requires == ("taint.flows", "findings.refutation") # ``ssrf`` also reads how the URL of each flow is built. assert by_name["ssrf"].requires == ("taint.flows", "findings.refutation", "ir.ssa", "abstract.module_strings") + # ``command-injection`` also reads the command each flow reaches, element by element. + assert by_name["command-injection"].requires == ( + "taint.flows", + "findings.refutation", + "ir.ssa", + "ir.defuse", + "abstract.module_strings", + ) assert by_name["python-stdlib-models"].provides == ("model.python-stdlib",) assert by_name["weak-crypto"].provides == ("vulnerability.weak-crypto",) From 0754f5da2a761b5e5b739d6ce7bd1acfe22cde5f Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 6 Oct 2026 07:03:15 +0200 Subject: [PATCH 2/8] fix(command-injection): judge option values by what the option does Review of the first version: a constant prefix refuted the value of an option that runs it (sh -c "echo $x", ssh -o ProxyCommand=..., find -exec), and ssh operands after the host, a remote shell command, were read as plain operands. A value an established option takes is now judged by what the option does with it, whatever its prefix: run it (high), keep it as data (git commit -m, gh --title: refuted), or an effect not established (medium); after an option the model does not describe, the finding stays for review. Operands a program runs as a command (ssh after the host, the program of sudo, env, xargs, timeout..., arguments of a batch file) are high. The option model also covers cmd /c and powershell -Command, shell flags grouped with -c, tar -I, the program given by a constant executable=, sys.executable, git.exe and python3.12; shell given at its position is read; -- counts anywhere before the input, and ssh is no longer taken to end its options there. Refs #208 --- docs/usage.md | 2 +- .../command_injection/command_injection.py | 292 ++++++++++++++---- tests/test_command_arguments.py | 52 +++- 3 files changed, 274 insertions(+), 72 deletions(-) diff --git a/docs/usage.md b/docs/usage.md index 485537e..e0394c8 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -115,7 +115,7 @@ variable is not treated that way yet. | Rule | Reached sink | |---|---| -| `command-injection` | Shell or process execution (`os.system`, `subprocess` with a string, …), judged on what the input controls in the command. A string a shell runs, a string naming the program or the program itself is `high`. An element of an argument list run without a shell can only be read as an option: `medium`, or `high` when a documented option of the program runs a command it chooses (`git push --receive-pack=…`, `ssh -o ProxyCommand=…`, `sh -c`). A constant prefix that keeps the input from being an option (`f"v{version}"`), or an operand after `--` for a program that honours it, is refuted. A value after an option that may take one, where the program's options are not known, stays a hotspot to review. | +| `command-injection` | Shell or process execution (`os.system`, `subprocess` with a string, …), judged on what the input controls in the command. A string a shell runs, a string naming the program, the program itself, or an operand the program runs as a command (after the host for `ssh`, the program of `sudo`, `env`, `xargs`, `timeout`…, any argument of a batch file) is `high`. Otherwise an element of an argument list run without a shell can only be read as an option: `medium`, or `high` when a documented option of the program runs a command it chooses (`git push --receive-pack=…`, `ssh -o ProxyCommand=…`). A value a documented option takes is judged by what the option does with it, whatever its prefix: run it (`sh -c`, `cmd /c`, `python -c`: `high`), keep it as data (`git commit -m`: refuted). After an option the rule does not know, the finding stays a hotspot to review. A constant prefix that keeps a standalone input from being an option (`f"v{version}"`), or an operand after `--` for a program that honours it, is refuted. | | `code-injection` | The code `eval` or `exec` runs; the globals and locals passed with it are data. `dangerous-eval` reports every such call, this rule the ones attacker input reaches. | | `sql-injection` | A database statement (`cursor.execute`, SQLAlchemy `text`, Django `raw`, …). Parameters of a parameterised query are not statements. | | `nosql-injection` | A NoSQL query, such as a MongoDB filter, given attacker input that may hold a structure: a JSON body giving `{"$ne": null}` matches any value. The input that may hold a structure is JSON, the raw bodies a `json.loads` decodes, files, WebSocket messages and standard input. Text cannot hold an operator: form fields, query parameters, headers, cookies, URL parameters, the command line, the environment, and an entry-point parameter annotated with scalars (`str`, `int`, `Optional[str]`, `Annotated[str, Query()]`, `list[str]`, …), which FastAPI validates. A Django, REST framework, aiohttp or tornado view reads a request object: its text attributes (`GET`, `POST`, `headers`, `query_params`, `query`, `match_info`, `arguments`, …) and the URL parameters after it are text, its body, files and `data` are not. Limits: a helper function the request is passed to reads it as a whole; the keys of a dict are not followed, so a string concatenated into `$where` is not reported. The analyzer ships no NoSQL sink: model plugins declare them. | diff --git a/src/coretrace_python/bundled/security/command_injection/command_injection.py b/src/coretrace_python/bundled/security/command_injection/command_injection.py index da9e61b..0452e55 100644 --- a/src/coretrace_python/bundled/security/command_injection/command_injection.py +++ b/src/coretrace_python/bundled/security/command_injection/command_injection.py @@ -1,24 +1,29 @@ """Command injection: attacker-controlled input reaching a COMMAND sink. What decides the result is what the attacker still controls in the command a process is -started with (issue #208). A string that a shell interprets, a string naming the program -or the program itself is a command injection, ``high``. An element of an argument list -run without a shell injects at most an option: ``medium``, or ``high`` when ``OPTIONS`` -establishes that the program runs a command chosen through that option. A constant -prefix is judged by what it leaves to the attacker: ``f"v{x}"`` is no option, the value -of ``f"--output={x}"`` belongs to a fixed option, ``f"--{x}"`` leaves the option open. -A value after an option belongs to that option only when ``OPTIONS`` establishes that the -option takes it, and is a free element when ``OPTIONS`` establishes that the options before -it take none (``rm -rf``, read as getopt reads grouped short options); otherwise the -finding is kept for review. A ``hotspot`` verdict lowers -the confidence, never the severity, which each case sets. +started with (issue #208). A string that a shell interprets, a string naming the program, +the program itself, or an operand the program runs as a command (after the host for +``ssh``, the program of a wrapper such as ``sudo``, any argument of a batch file) is a +command injection, ``high``. An element of an argument list run without a shell +otherwise injects at most an option: ``medium``, or ``high`` when ``OPTIONS`` establishes +that the program runs a command chosen through an option that fits there. + +A value an option takes is judged by what ``OPTIONS`` establishes the option does with +it, whatever its prefix: run it (``sh -c``, ``cmd /c``, ``python -c``: ``high``), keep it +as data (``git commit -m``: refuted), or something not established (``medium``). After an +option ``OPTIONS`` does not describe, the finding is kept for review at the severity of a +free element; after options established to take no value (``rm -rf``, grouped short +options read as getopt reads them), the element stands alone. An element standing alone +is judged by its constant prefix: ``f"v{x}"`` is no option, the value of ``f"--output={x}"`` +belongs to a fixed option, ``f"--{x}"`` leaves the option open. A ``hotspot`` verdict +lowers the confidence, never the severity, which each case sets. """ from __future__ import annotations +import re from collections.abc import Mapping from dataclasses import dataclass -from pathlib import PurePosixPath from typing import ClassVar from coretrace_python.abstract.strings import ModuleStringsAnalysis, leading_text @@ -28,7 +33,7 @@ from coretrace_python.hir import nodes from coretrace_python.interprocedural import Arguments from coretrace_python.ir.defuse import DefUseAnalysis -from coretrace_python.ir.model import Instruction, Value +from coretrace_python.ir.model import Instruction, Symbol, Value from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.plugins import Assessment, PluginContext, TaintDetector from coretrace_python.semantic.symbols import SymbolId @@ -40,9 +45,11 @@ class CommandOption: """An option of ``program``, after one of ``subcommands`` when any are given. An option that ``takes_value`` reads it from the next element, or after ``=`` or the - short option in the same element; with ``runs``, the program runs a command that - value chooses (``runs`` says how), and ``single`` says one element can carry the - option and its value. An option that takes no value leaves the next element free.""" + short option in the same element, and ``single`` says one element can carry the option + and its value; with ``rest``, every element after it is its value. What the program + does with the value decides: with ``runs``, it runs a command the value chooses; + with ``inert``, it only keeps it as data (a commit message); with neither, the effect + is not established. An option that takes no value leaves the next element free.""" program: str option: str @@ -50,14 +57,18 @@ class CommandOption: subcommands: tuple[str, ...] = () single: bool = True takes_value: bool = True + inert: str | None = None + rest: bool = False def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: return tuple(CommandOption(program, option, takes_value=False) for option in options) +_SHELLS = ("sh", "bash", "dash", "zsh", "ksh") + # Documented behaviours only: an option missing here is neither established to be -# dangerous nor to take no value. +# dangerous, nor inert, nor to take no value. OPTIONS: tuple[CommandOption, ...] = ( CommandOption( "git", "-c", "sets any configuration variable, core.sshCommand included", single=False @@ -70,27 +81,29 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: ("--exec", ("push",)), ) ), + *( + CommandOption( + "git", option, subcommands=(subcommand,), inert=f"is the {subcommand} message" + ) + for subcommand in ("commit", "tag") + for option in ("-m", "--message") + ), + *( + CommandOption("gh", option, subcommands=(subcommand,), inert=inert) + for subcommand, options, inert in ( + ("release", ("--title", "-t", "--notes", "-n"), "is text of the release"), + ("workflow", ("--ref", "-r"), "names the branch or tag of the workflow"), + ) + for option in options + ), CommandOption("ssh", "-o", "sets options such as ProxyCommand, which runs a command"), *_flags("ssh", *(f"-{letter}" for letter in "46AaCfGgKkMNnqsTtVvXxYy")), - *_flags( - "rm", - "-d", - "-f", - "-I", - "-i", - "-R", - "-r", - "-v", - "--dir", - "--force", - "--recursive", - "--verbose", - ), CommandOption( "tar", "--to-command", "pipes each extracted file to the command it names (GNU tar)" ), CommandOption("tar", "--checkpoint-action", "runs the command of an exec= action (GNU tar)"), CommandOption("tar", "--use-compress-program", "runs the compressor it names"), + CommandOption("tar", "-I", "runs the compressor it names (GNU tar)"), CommandOption("rsync", "-e", "runs the remote shell it names"), CommandOption("rsync", "--rsh", "runs the remote shell it names"), *( @@ -99,22 +112,73 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: ), *( CommandOption(shell, "-c", "runs its value as a shell command", single=False) - for shell in ("sh", "bash", "dash", "zsh", "ksh") + for shell in _SHELLS ), + *(CommandOption(shell, "-o", single=False) for shell in _SHELLS), *( - CommandOption(python, "-c", "runs its value as Python code") - for python in ("python", "python3") + option + for shell in _SHELLS + for option in _flags(shell, *(f"-{letter}" for letter in "abCefhimnsuvx")) + ), + *(option for shell in ("bash", "zsh", "ksh") for option in _flags(shell, "-l")), + CommandOption("python", "-c", "runs its value as Python code"), + *( + CommandOption( + "cmd", option, "runs the rest of the command line as a command", single=False, rest=True + ) + for option in ("/c", "/C", "/k", "/K") + ), + *( + CommandOption( + powershell, + option, + "runs the rest of the command line as a command", + single=False, + rest=True, + ) + for powershell in ("powershell", "pwsh") + for option in ("-Command", "-command", "-c", "-EncodedCommand", "-e") + ), + *_flags( + "rm", + "-d", + "-f", + "-I", + "-i", + "-R", + "-r", + "-v", + "--dir", + "--force", + "--recursive", + "--verbose", ), ) # Programs documented to read ``--`` as the end of their options. -END_OF_OPTIONS = frozenset({"git", "ssh", "tar", "rsync"}) +END_OF_OPTIONS = frozenset({"git", "tar", "rsync", "rm"}) +# Programs whose operands, from the given one on, are a command they run, and how. +COMMAND_OPERANDS: Mapping[str, tuple[int, str]] = { + "ssh": (1, "is part of the command the remote shell runs"), + "timeout": (1, "may name the program timeout runs"), + **{ + wrapper: (0, f"may name the program {wrapper} runs") + for wrapper in ("sudo", "doas", "env", "xargs", "nohup", "nice", "time") + }, +} +# Windows runs a batch file through cmd.exe, which parses its arguments again. +_BATCH = (".bat", ".cmd") _PROCESS_STARTERS = frozenset( SymbolId(f"python.subprocess.{name}") for name in ("run", "call", "check_call", "check_output", "Popen") ) +# ``Popen(args, bufsize, executable, ..., shell, ...)``: the other starters pass their +# positional arguments to it. _COMMAND_ARGUMENT = frozenset({(0, None), (None, "args")}) +_EXECUTABLE, _SHELL_POSITION = 2, 8 _SHELL, _NO_SHELL = frozenset({"True", "1"}), frozenset({"False", "None", "0"}) +_PYTHON = re.compile(r"python[0-9.]*") +_PYTHON_EXECUTABLE = SymbolId("python.sys.executable") @dataclass(frozen=True) @@ -147,18 +211,15 @@ def assess( default = super().assess(ctx, function, flow, verdict) if flow.sink.symbol not in _PROCESS_STARTERS or flow.through is not None: return default - if not flow.passed_as & _COMMAND_ARGUMENT or flow.sink_arguments is None: + arguments = flow.sink_arguments + if arguments is None or not flow.passed_as or not flow.passed_as <= _COMMAND_ARGUMENT: return default - shell = _shell(flow.sink_arguments) + shell = _shell(arguments) if shell is True: return default if shell is None: - return Assessment( - _to_review( - default.verdict, "an expression chooses whether a shell runs the command" - ), - Severity.HIGH, - ) + reason = "an expression chooses whether a shell runs the command" + return Assessment(_to_review(default.verdict, reason), Severity.HIGH) ssa = ctx.get(SSAAnalysis, function) defs = { i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None @@ -167,7 +228,7 @@ def assess( if parts is None: return default taint = ctx.get(TaintAnalysis, function) - command = _Command(parts, defs, ctx.get(ModuleStringsAnalysis)) + command = _Command(parts, defs, ctx.get(ModuleStringsAnalysis), _executable(arguments)) readings = [ command.read(index) for index, part in enumerate(parts) @@ -200,12 +261,20 @@ def __init__( parts: tuple[CommandPart, ...], defs: Mapping[Value, Instruction], strings: Mapping[str, str], + executable: str | None, ) -> None: self.parts = parts self.defs = defs self.strings = strings - first = self.text(0) - self.program = PurePosixPath(first).name if first is not None else None + first = executable if executable is not None else self.text(0) + made = defs.get(parts[0].value) if parts and parts[0].known else None + if first is None and isinstance(made, Symbol) and made.symbol_id == _PYTHON_EXECUTABLE: + first = "python" + name = re.split(r"[\\/]", first)[-1].lower() if first is not None else None + self.batch = name is not None and name.endswith(_BATCH) + if name is not None and name.endswith(".exe"): + name = name.removesuffix(".exe") + self.program = "python" if name is not None and _PYTHON.fullmatch(name) else name def text(self, index: int) -> str | None: """The constant text of the element at ``index``, when all of it is constant.""" @@ -217,12 +286,44 @@ def text(self, index: int) -> str | None: return text if whole else None def read(self, index: int) -> _Reading: - part = self.parts[index] + """What the input at ``index`` lets the attacker do. An element an established + option takes is judged by what the option does with it, whatever its prefix; one + that stands alone by what it can still be: the program, a command operand, an + option, or an operand only.""" + if all(not earlier.known for earlier in self.parts[:index]): return _Reading("the input may name the program", Severity.HIGH, command=True) + if self.batch: + return _Reading( + "cmd.exe parses the arguments of a batch file again", Severity.HIGH, command=True + ) + for position in range(1, index): + option = self.option(self.text(position) or "", position) + if option is not None and option.rest: + return _Reading( + f"the input follows {self.program} {option.option}, which {option.runs}", + Severity.HIGH, + ) + before = self.text(index - 1) + if before is not None and before not in ("-", "--") and self.option_like(before): + taker = self.taker(before, index) + if isinstance(taker, CommandOption): + return self.value_of(taker, before) + if taker is None: + if self.command_operand(index) is not False: + return self.operand_command() + free = self.free(index, "the input is a whole element") + return _Reading( + f"cannot tell whether {before} takes the input as its value; if not, {free.reason}", + free.severity, + to_review=True, + ) + part = self.parts[index] + prefix = leading_text(part.value, self.defs, self.strings)[0] if part.known else "" + if not prefix.startswith("-") and self.command_operand(index) is not False: + return self.operand_command() if not part.known: return self.free(index, "the input gives whole elements") - prefix, _ = leading_text(part.value, self.defs, self.strings) if prefix.startswith("--") and "=" in prefix: return self.option_value(index, prefix[: prefix.index("=")]) if prefix.startswith("-") and len(prefix) >= 2 and prefix[1] != "-": @@ -233,29 +334,19 @@ def read(self, index: int) -> _Reading: return _Reading( f"the constant prefix {prefix!r} keeps the input from being an option", None ) - before = self.text(index - 1) - if before == "--": + if self.after_end_of_options(index): if self.program in END_OF_OPTIONS: return _Reading(f"{self.program} reads the input after '--' as an operand", None) return self.free( index, "the input follows '--', which the program is not known to honour" ) - if before is not None and before.startswith("-") and before != "-" and "=" not in before: - taker = self.taker(before, index) - if isinstance(taker, CommandOption): - return self.value_of(taker, before) - if taker is False: - return self.free( - index, f"the input is a whole element after {before}, which takes no value" - ) - free = self.free(index, "the input is a whole element") - return _Reading( - f"cannot tell whether {before} takes the input as its value; if not, {free.reason}", - free.severity, - to_review=True, - ) return self.free(index, "the input is a whole element") + def option_like(self, text: str) -> bool: + return text.startswith("-") or any( + option.option == text for option in OPTIONS if option.program == self.program + ) + def free(self, index: int, how: str) -> _Reading: """An element the attacker may give as an option: ``high`` when an established option of the program fits in one element there.""" @@ -266,7 +357,8 @@ def free(self, index: int, how: str) -> _Reading: if dangerous: option = dangerous[0] return _Reading( - f"{how}, and {self.program} {option.option}=... {option.runs}", Severity.HIGH + f"{how}, and an option such as {self.program} {_spelled(option.option)} {option.runs}", + Severity.HIGH, ) return _Reading(f"{how}, so it may be read as an option", Severity.MEDIUM) @@ -288,8 +380,54 @@ def value_of(self, option: CommandOption, given: str) -> _Reading: f"the input is the value of {self.program} {given}, which {option.runs}", Severity.HIGH, ) + if option.inert is not None: + return _Reading( + f"the input is the value of {self.program} {given}, which {option.inert}", None + ) return _Reading(f"the input is the value of {self.program} {given}", Severity.MEDIUM) + def operand_command(self) -> _Reading: + assert self.program is not None + return _Reading( + f"the input {COMMAND_OPERANDS[self.program][1]}", Severity.HIGH, command=True + ) + + def command_operand(self, index: int) -> bool | None: + """Whether the element at ``index`` is one of the operands the program runs as a + command: False when the program has none or the element comes before them, None + when the options before it do not tell.""" + + if self.program not in COMMAND_OPERANDS: + return False + operands = self.operands_before(index) + return None if operands is None else operands >= COMMAND_OPERANDS[self.program][0] + + def operands_before(self, index: int) -> int | None: + """The number of operands before ``index``, or None when an element of unknown + value, or an option that may take one, comes first.""" + + count, position = 0, 1 + while position < index: + text = self.text(position) + if text is None: + return None + if text == "--": + return count + index - position - 1 + if text.startswith("-") and text != "-": + if not (text.startswith("--") and "=" in text): + taker = self.taker(text, position + 1) + if taker is None: + return None + if isinstance(taker, CommandOption): + position += 1 + else: + count += 1 + position += 1 + return count + + def after_end_of_options(self, index: int) -> bool: + return any(self.text(position) == "--" for position in range(1, index)) + def taker(self, before: str, index: int) -> CommandOption | bool | None: """The option ``before`` whose value is the element at ``index``; False when the model establishes that ``before`` takes no value there, None when it cannot tell. @@ -299,7 +437,7 @@ def taker(self, before: str, index: int) -> CommandOption | bool | None: option = self.option(before, index) if option is not None: return option if option.takes_value else False - if before.startswith("--") or len(before) <= 2: + if not before.startswith("-") or before.startswith("--") or len(before) <= 2: return None letters = before[1:] for position, letter in enumerate(letters): @@ -340,11 +478,29 @@ def subcommand(self, index: int) -> str | None: return "" +def _spelled(option: str) -> str: + """An option with its value in the same element: ``--name=...`` or ``-o...``.""" + + return f"{option}=..." if option.startswith("--") else f"{option}..." + + +def _executable(arguments: Arguments) -> str | None: + """The program a constant ``executable`` names, which the command's first element + then only labels.""" + + given = arguments.given("executable", _EXECUTABLE) + written = given[1] if given is not None else None + if written is None or len(written) < 2 or written[0] not in "'\"" or written[-1] != written[0]: + return None + return written[1:-1].replace("\\\\", "\\") + + def _shell(arguments: Arguments) -> bool | None: """Whether a shell runs the command: True or False when the call says so with a - constant, None when an expression or unpacked keywords decide.""" + constant, by keyword or at its place, None when an expression or unpacked arguments + decide.""" - given = arguments.given("shell") + given = arguments.given("shell", _SHELL_POSITION) if given is None: return None present, value = given diff --git a/tests/test_command_arguments.py b/tests/test_command_arguments.py index b818a74..a5b5584 100644 --- a/tests/test_command_arguments.py +++ b/tests/test_command_arguments.py @@ -7,9 +7,13 @@ (``git push --receive-pack=``, ``ssh -o ProxyCommand=...``, ``sh -c``). A constant prefix is judged by what the attacker still controls: ``f"v{x}"`` is no option at all, ``f"--output={x}"`` is the value of a fixed option, ``f"--{x}"`` leaves the option open. -A value after an option is that option's value only when the model establishes that the -option consumes it, and a free element when the model establishes that the options before -it take no value (``rm -rf``); otherwise the finding stays, for review. Severity is set per case: +A value an option takes is judged by what the model establishes the option does with it, +whatever its prefix: run it (``sh -c``, ``cmd /c``, ``python -c``: ``high``), keep it as +data (``git commit -m``: refuted), or something not established (``medium``). After an +option the model does not describe, the finding stays for review; after options the model +establishes to take no value (``rm -rf``), the element stands alone. Operands that a +program runs as a command (after the host for ``ssh``, a wrapper's program such as +``sudo``'s, any argument of a batch file) are ``high``. Severity is set per case: the ``hotspot`` verdict lowers the confidence, never the severity. """ @@ -83,6 +87,48 @@ def judged(call: str) -> list[tuple[str, str, str]]: ('subprocess.run(["rm", "-rz", x])', MEDIUM_TO_REVIEW), ('subprocess.run(["ssh", "-vo", x, "host"])', HIGH), ('subprocess.run(["ssh", "-ov", x, "host"])', HIGH), + # An established option judges the value it takes, whatever its prefix. + ('subprocess.run(["sh", "-c", f"echo {x}"])', HIGH), + ('subprocess.run(["bash", "-c", "ls " + x])', HIGH), + ('subprocess.run(["ssh", "-o", f"ProxyCommand={x}", "host"])', HIGH), + ('subprocess.run(["git", "-c", f"core.sshCommand={x}", "fetch"])', HIGH), + ('subprocess.run(["find", ".", "-exec", f"sh -c {x}", ";"])', HIGH), + ('subprocess.run(["python3", "-c", f"print({x})"])', HIGH), + ('subprocess.run([sys.executable, "-c", f"print({x})"])', HIGH), + ('subprocess.run(["python3.12", "-c", x])', HIGH), + ('subprocess.run(["cmd", "/c", f"dir {x}"])', HIGH), + ('subprocess.run(["cmd", "/c", "dir", x])', HIGH), + ('subprocess.run(["powershell", "-Command", x])', HIGH), + ('subprocess.run(["git", "commit", "-m", x])', None), + ('subprocess.run(["gh", "release", "create", "v1", "--title", f"release {x}"])', None), + ('subprocess.run(["gh", "workflow", "run", "main.yml", "--ref", x])', None), + ('subprocess.run(["notmodelled", "-c", f"echo {x}"])', MEDIUM_TO_REVIEW), + ('subprocess.run(["bash", "-lc", x])', HIGH), + ('subprocess.run(["sh", "-ec", f"echo {x}"])', HIGH), + ('subprocess.run(["tar", f"-I{x}", "-xf", "a.tar"])', HIGH), + ('subprocess.run(["safe", "-c", x], executable="/bin/sh")', HIGH), + # Operands that are a command: a remote shell, a wrapper, a batch file. + ('subprocess.run(["ssh", "user@host", "--", x])', HIGH), + ('subprocess.run(["ssh", "host", f"cat {x}"])', HIGH), + ('subprocess.run(["sudo", x])', HIGH), + ('subprocess.run(["env", x])', HIGH), + ('subprocess.run(["timeout", "5", x])', HIGH), + ('subprocess.run(["build.bat", x])', HIGH), + ('subprocess.run(["git.exe", "push", "origin", x])', HIGH), + (r'subprocess.run(["C:\\Git\\bin\\git.exe", "push", "origin", x])', HIGH), + # An operand after the end of options, anywhere before it. + ('subprocess.run(["git", "log", "--", "a", x])', None), + ('subprocess.run(["rm", "-rf", "--", x])', None), + # Where shell comes from, and lists the module cannot read. + ('subprocess.Popen(["ls", x], -1, None, None, None, None, None, True, True)', HIGH), + ('subprocess.run(["ls", x], **extra)', HIGH_TO_REVIEW), + ('subprocess.Popen(args=["ls", x])', MEDIUM), + ('subprocess.run(("ls", x))', MEDIUM), + ('cmd = ["ls", x]\n subprocess.run(cmd)', MEDIUM), + ('cmd = ["ls"]\n cmd.append(x)\n subprocess.run(cmd)', HIGH), + ('cmd = ["ls", "-l"]\n cmd.insert(0, x)\n subprocess.run(cmd)', HIGH), + ('cmd = ["ls"]\n cmd += [x]\n subprocess.run(cmd)', HIGH), + ('cmd = ["ls", "-l"]\n cmd[0] = x\n subprocess.run(cmd)', HIGH), # Unpacked parts keep their place in the command. ('subprocess.run(["xcrun", "notarytool", *(["--keychain", x])])', MEDIUM_TO_REVIEW), ("subprocess.run([*extra, x])", HIGH), From 4dfe490672f9254994ef7bcd8c4df43911b964c7 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 6 Oct 2026 07:30:01 +0200 Subject: [PATCH 3/8] fix(command-injection): read operands a program runs, and nested commands Second review: a shell's -c is a flag whose first operand is the command, so grouped and separate forms (sh -ce, bash -c -e, sh -c --) were refuted; ssh operands after the first remote word are shell text even when they start with -; find -exec sh -c x and similar commands inside a command were judged as free elements; an element of unknown value before the input led to a refutation. Programs now declare the operands they run: shell text (ssh after the host, powershell, a shell's first operand after -c) or a command of its own (the program of sudo, env, xargs, timeout..., what follows find -exec), read again with that program's model. An element of unknown value before the input keeps the finding for review. An inert reading needs the subcommand established and covers one element only; git clone -u, case-insensitive Windows options, a glued cmd /c, and an executable that cannot be read are handled. Refs #208 --- .../command_injection/command_injection.py | 405 ++++++++++++------ tests/test_command_arguments.py | 35 +- 2 files changed, 317 insertions(+), 123 deletions(-) diff --git a/src/coretrace_python/bundled/security/command_injection/command_injection.py b/src/coretrace_python/bundled/security/command_injection/command_injection.py index 0452e55..6822c18 100644 --- a/src/coretrace_python/bundled/security/command_injection/command_injection.py +++ b/src/coretrace_python/bundled/security/command_injection/command_injection.py @@ -2,21 +2,24 @@ What decides the result is what the attacker still controls in the command a process is started with (issue #208). A string that a shell interprets, a string naming the program, -the program itself, or an operand the program runs as a command (after the host for -``ssh``, the program of a wrapper such as ``sudo``, any argument of a batch file) is a -command injection, ``high``. An element of an argument list run without a shell -otherwise injects at most an option: ``medium``, or ``high`` when ``OPTIONS`` establishes -that the program runs a command chosen through an option that fits there. +the program itself, or an operand that the program runs as shell text (``ssh`` after the +host, a shell's first operand after ``-c``, any argument of a batch file) is a command +injection, ``high``. Operands that are a command of their own (the program a wrapper such +as ``sudo`` runs, what follows ``find -exec``) are read again as that command. An element +of an argument list run without a shell otherwise injects at most an option: ``medium``, +or ``high`` when ``OPTIONS`` establishes that the program runs a command chosen through an +option that fits there. A value an option takes is judged by what ``OPTIONS`` establishes the option does with -it, whatever its prefix: run it (``sh -c``, ``cmd /c``, ``python -c``: ``high``), keep it -as data (``git commit -m``: refuted), or something not established (``medium``). After an -option ``OPTIONS`` does not describe, the finding is kept for review at the severity of a -free element; after options established to take no value (``rm -rf``, grouped short -options read as getopt reads them), the element stands alone. An element standing alone -is judged by its constant prefix: ``f"v{x}"`` is no option, the value of ``f"--output={x}"`` -belongs to a fixed option, ``f"--{x}"`` leaves the option open. A ``hotspot`` verdict -lowers the confidence, never the severity, which each case sets. +it, whatever its prefix: run it (``cmd /c``, ``python -c``: ``high``), keep it as data +(``git commit -m``: refuted), or something not established (``medium``). After an option +``OPTIONS`` does not describe, or an element of unknown value, the finding is kept for +review at the severity of a free element; after options established to take no value +(``rm -rf``, grouped short options read as getopt reads them), the element stands alone. +An element standing alone is judged by its constant prefix: ``f"v{x}"`` is no option, the +value of ``f"--output={x}"`` belongs to a fixed option, ``f"--{x}"`` leaves the option +open. A ``hotspot`` verdict lowers the confidence, never the severity, which each case +sets. """ from __future__ import annotations @@ -46,10 +49,12 @@ class CommandOption: """An option of ``program``, after one of ``subcommands`` when any are given. An option that ``takes_value`` reads it from the next element, or after ``=`` or the short option in the same element, and ``single`` says one element can carry the option - and its value; with ``rest``, every element after it is its value. What the program - does with the value decides: with ``runs``, it runs a command the value chooses; - with ``inert``, it only keeps it as data (a commit message); with neither, the effect - is not established. An option that takes no value leaves the next element free.""" + and its value; with ``rest``, every element after it is its value, and with ``opens`` + the elements after it, up to ``;`` or ``+``, are a command of their own. What the + program does with the value decides: with ``runs``, it runs a command the value + chooses; with ``inert``, it only keeps it as data (a commit message); with neither, + the effect is not established. An option that takes no value leaves the next element + free.""" program: str option: str @@ -59,6 +64,20 @@ class CommandOption: takes_value: bool = True inert: str | None = None rest: bool = False + opens: bool = False + + +@dataclass(frozen=True) +class CommandOperands: + """Operands a program runs, counted from 0 after its options: from operand ``first`` + on, either shell ``text`` the program has a shell run, or a program and its arguments, + a command of its own. With ``after``, only when that letter is among the program's + short options, and then operand ``first`` only (``sh -c 'cmd' name arguments``).""" + + first: int + text: bool + how: str + after: str | None = None def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: @@ -66,6 +85,7 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: _SHELLS = ("sh", "bash", "dash", "zsh", "ksh") +_WRAPPERS = ("sudo", "doas", "env", "xargs", "nohup", "nice", "time") # Documented behaviours only: an option missing here is neither established to be # dangerous, nor inert, nor to take no value. @@ -77,6 +97,7 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: CommandOption("git", option, "runs the program it names", subcommands) for option, subcommands in ( ("--upload-pack", ("clone", "fetch", "pull", "ls-remote")), + ("-u", ("clone", "ls-remote")), ("--receive-pack", ("push",)), ("--exec", ("push",)), ) @@ -107,26 +128,21 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: CommandOption("rsync", "-e", "runs the remote shell it names"), CommandOption("rsync", "--rsh", "runs the remote shell it names"), *( - CommandOption("find", option, "runs the command that follows", single=False) + CommandOption("find", option, "runs the command that follows", single=False, opens=True) for option in ("-exec", "-execdir", "-ok", "-okdir") ), - *( - CommandOption(shell, "-c", "runs its value as a shell command", single=False) - for shell in _SHELLS - ), *(CommandOption(shell, "-o", single=False) for shell in _SHELLS), *( option for shell in _SHELLS - for option in _flags(shell, *(f"-{letter}" for letter in "abCefhimnsuvx")) + for option in _flags(shell, *(f"-{letter}" for letter in "abCcefhilmnsuvx")) ), - *(option for shell in ("bash", "zsh", "ksh") for option in _flags(shell, "-l")), CommandOption("python", "-c", "runs its value as Python code"), *( CommandOption( "cmd", option, "runs the rest of the command line as a command", single=False, rest=True ) - for option in ("/c", "/C", "/k", "/K") + for option in ("/c", "/k") ), *( CommandOption( @@ -137,7 +153,7 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: rest=True, ) for powershell in ("powershell", "pwsh") - for option in ("-Command", "-command", "-c", "-EncodedCommand", "-e") + for option in ("-command", "-c", "-encodedcommand", "-e", "-ec") ), *_flags( "rm", @@ -156,17 +172,24 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: ) # Programs documented to read ``--`` as the end of their options. END_OF_OPTIONS = frozenset({"git", "tar", "rsync", "rm"}) -# Programs whose operands, from the given one on, are a command they run, and how. -COMMAND_OPERANDS: Mapping[str, tuple[int, str]] = { - "ssh": (1, "is part of the command the remote shell runs"), - "timeout": (1, "may name the program timeout runs"), +OPERANDS: Mapping[str, CommandOperands] = { + "ssh": CommandOperands(1, True, "is part of the command the remote shell runs"), + "powershell": CommandOperands(0, True, "is part of the command PowerShell runs"), **{ - wrapper: (0, f"may name the program {wrapper} runs") - for wrapper in ("sudo", "doas", "env", "xargs", "nohup", "nice", "time") + shell: CommandOperands(0, True, "is the command the shell runs after -c", after="c") + for shell in _SHELLS + }, + "timeout": CommandOperands(1, False, "names the program timeout runs"), + **{ + wrapper: CommandOperands(0, False, f"names the program {wrapper} runs") + for wrapper in _WRAPPERS }, } -# Windows runs a batch file through cmd.exe, which parses its arguments again. +# Windows runs a batch file through cmd.exe, which parses its arguments again; its own +# programs read their options whatever the case. _BATCH = (".bat", ".cmd") +_ANY_CASE = frozenset({"cmd", "powershell", "pwsh"}) +_TERMINATORS = (";", "+") _PROCESS_STARTERS = frozenset( SymbolId(f"python.subprocess.{name}") @@ -184,8 +207,8 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: @dataclass(frozen=True) class _Reading: """What one tainted element of a command lets the attacker do: run a ``command`` - (the program, ``high``), give an option at ``severity``, possibly only ``to_review``, - or nothing, with ``severity`` None.""" + (``high``), give an option at ``severity``, possibly only ``to_review``, or nothing, + with ``severity`` None.""" reason: str severity: Severity | None @@ -193,6 +216,16 @@ class _Reading: to_review: bool = False +@dataclass(frozen=True) +class _Operands: + """The operands before an element, by position; whether the option before it takes + the element as its value (``consumed``); whether a ``--`` ended the options.""" + + positions: tuple[int, ...] + consumed: bool = False + ended: bool = False + + class CommandInjectionPlugin(TaintDetector): name: ClassVar[str] = "command-injection" rule_id: ClassVar[str] = "command-injection" @@ -225,10 +258,12 @@ def assess( i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None } parts = command_parts(flow.argument, defs, ctx.get(DefUseAnalysis, function)) - if parts is None: + if not parts: return default taint = ctx.get(TaintAnalysis, function) - command = _Command(parts, defs, ctx.get(ModuleStringsAnalysis), _executable(arguments)) + command = _Command( + parts, defs, ctx.get(ModuleStringsAnalysis), _program(arguments, parts[0], defs) + ) readings = [ command.read(index) for index, part in enumerate(parts) @@ -254,27 +289,30 @@ def assess( class _Command: - """The command of one call, read element by element.""" + """A command, read element by element; ``program`` is the name of the program it + runs, None when unknown.""" def __init__( self, parts: tuple[CommandPart, ...], defs: Mapping[Value, Instruction], strings: Mapping[str, str], - executable: str | None, + program: str | None, ) -> None: self.parts = parts self.defs = defs self.strings = strings - first = executable if executable is not None else self.text(0) - made = defs.get(parts[0].value) if parts and parts[0].known else None - if first is None and isinstance(made, Symbol) and made.symbol_id == _PYTHON_EXECUTABLE: - first = "python" - name = re.split(r"[\\/]", first)[-1].lower() if first is not None else None + name = _name(program) if program is not None else None self.batch = name is not None and name.endswith(_BATCH) - if name is not None and name.endswith(".exe"): - name = name.removesuffix(".exe") - self.program = "python" if name is not None and _PYTHON.fullmatch(name) else name + self.program = name + + def nested(self, start: int, end: int | None = None) -> _Command: + """The command that the elements from ``start`` (to ``end``) spell.""" + + parts = self.parts[start:end] + return _Command( + parts, self.defs, self.strings, _first_program(parts[0], self.defs, self.strings) + ) def text(self, index: int) -> str | None: """The constant text of the element at ``index``, when all of it is constant.""" @@ -285,11 +323,15 @@ def text(self, index: int) -> str | None: text, whole = leading_text(part.value, self.defs, self.strings) return text if whole else None + def prefix(self, index: int) -> str: + part = self.parts[index] + return leading_text(part.value, self.defs, self.strings)[0] if part.known else "" + def read(self, index: int) -> _Reading: - """What the input at ``index`` lets the attacker do. An element an established - option takes is judged by what the option does with it, whatever its prefix; one - that stands alone by what it can still be: the program, a command operand, an - option, or an operand only.""" + """What the input at ``index`` lets the attacker do. Operands the program runs + come first; then an element an established option takes is judged by what the + option does with it, whatever its prefix; one standing alone, by what it can + still be: an option, or an operand only.""" if all(not earlier.known for earlier in self.parts[:index]): return _Reading("the input may name the program", Severity.HIGH, command=True) @@ -297,33 +339,56 @@ def read(self, index: int) -> _Reading: return _Reading( "cmd.exe parses the arguments of a batch file again", Severity.HIGH, command=True ) + operand = self.operand(index) + if operand is not None: + return operand for position in range(1, index): option = self.option(self.text(position) or "", position) + if option is not None and option.opens: + end = next( + ( + later + for later in range(position + 1, len(self.parts)) + if self.text(later) in _TERMINATORS + ), + len(self.parts), + ) + if position < index < end: + return self.inside(position + 1, end, index) if option is not None and option.rest: return _Reading( f"the input follows {self.program} {option.option}, which {option.runs}", Severity.HIGH, ) - before = self.text(index - 1) - if before is not None and before not in ("-", "--") and self.option_like(before): - taker = self.taker(before, index) - if isinstance(taker, CommandOption): - return self.value_of(taker, before) - if taker is None: - if self.command_operand(index) is not False: - return self.operand_command() + if index >= 2: + before = self.text(index - 1) + if before is None: free = self.free(index, "the input is a whole element") - return _Reading( - f"cannot tell whether {before} takes the input as its value; if not, {free.reason}", - free.severity, - to_review=True, + return self.review( + f"cannot tell what the element before the input is; if not an option, {free.reason}", + free, ) - part = self.parts[index] - prefix = leading_text(part.value, self.defs, self.strings)[0] if part.known else "" - if not prefix.startswith("-") and self.command_operand(index) is not False: - return self.operand_command() - if not part.known: + if before not in ("-", "--") and self.option_like(before): + taker = self.taker(before, index) + if isinstance(taker, CommandOption): + if not self.parts[index].known and taker.runs is None: + return self.free(index, f"the input gives whole elements after {before}") + return self.value_of(taker, before) + if taker is None: + free = self.free(index, "the input is a whole element") + return self.review( + f"cannot tell whether {before} takes the input as its value; if not, {free.reason}", + free, + ) + if not self.parts[index].known: return self.free(index, "the input gives whole elements") + prefix = self.prefix(index) + for option in self.applicable(index): + if option.rest and self.same(prefix[: len(option.option)], option.option): + return _Reading( + f"the input follows {self.program} {option.option}, which {option.runs}", + Severity.HIGH, + ) if prefix.startswith("--") and "=" in prefix: return self.option_value(index, prefix[: prefix.index("=")]) if prefix.startswith("-") and len(prefix) >= 2 and prefix[1] != "-": @@ -334,7 +399,7 @@ def read(self, index: int) -> _Reading: return _Reading( f"the constant prefix {prefix!r} keeps the input from being an option", None ) - if self.after_end_of_options(index): + if any(self.text(position) == "--" for position in range(1, index)): if self.program in END_OF_OPTIONS: return _Reading(f"{self.program} reads the input after '--' as an operand", None) return self.free( @@ -342,9 +407,110 @@ def read(self, index: int) -> _Reading: ) return self.free(index, "the input is a whole element") + def operand(self, index: int) -> _Reading | None: + """The reading of an element among the operands the program runs, or None when + the element is not one of them.""" + + if self.program not in OPERANDS: + return None + spec = OPERANDS[self.program] + unclear = _Reading( + f"the input may be an operand that {spec.how}", Severity.HIGH, command=True + ) + if spec.after is not None: + given = self.letter_given(spec.after, index) + if given is None: + return unclear + if not given: + return None + operands = self.operands(index, spec.first + 1) + if operands is None: + return unclear + if operands.consumed: + return None + count = len(operands.positions) + operand_like = operands.ended or not self.prefix(index).startswith("-") + if count > spec.first: + if not spec.text: + return self.inside(operands.positions[spec.first], None, index) + return ( + None + if spec.after is not None + else _Reading(f"the input {spec.how}", Severity.HIGH, command=True) + ) + if count == spec.first and operand_like: + return _Reading(f"the input {spec.how}", Severity.HIGH, command=True) + return None + + def inside(self, start: int, end: int | None, index: int) -> _Reading: + """The reading of the element at ``index`` in the command the elements from + ``start`` (to ``end``) spell, which the program runs.""" + + if index == start: + return _Reading( + f"the input names the program {self.program} runs", Severity.HIGH, command=True + ) + reading = self.nested(start, end).read(index - start) + return _Reading( + f"in the command {self.program} runs, {reading.reason}", + reading.severity, + reading.command, + reading.to_review, + ) + + def operands(self, index: int, enough: int) -> _Operands | None: + """The operands before ``index``, up to ``enough`` of them; None when an element of + unknown value, or an option that may take one, comes first.""" + + positions: list[int] = [] + position = 1 + while position < index and len(positions) < enough: + text = self.text(position) + if text is None: + return None + if text == "--": + return _Operands((*positions, *range(position + 1, index))[:enough], ended=True) + if self.option_like(text) and text != "-": + if not (text.startswith("--") and "=" in text): + taker = self.taker(text, position + 1) + if taker is None: + return None + if isinstance(taker, CommandOption): + position += 1 + if position == index: + return _Operands(tuple(positions), consumed=True) + else: + positions.append(position) + position += 1 + return _Operands(tuple(positions)) + + def letter_given(self, letter: str, index: int) -> bool | None: + """Whether a short option ``letter`` comes among the options before ``index``, None + when an element of unknown value comes first.""" + + position = 1 + while position < index: + text = self.text(position) + if text is None: + return None + if text == "--" or not text.startswith("-") or text == "-": + return False + if not text.startswith("--") and letter in text[1:]: + return True + if isinstance(self.taker(text, position + 1), CommandOption): + position += 1 + position += 1 + return False + + def review(self, reason: str, free: _Reading) -> _Reading: + return _Reading(reason, free.severity, to_review=True) + + def same(self, written: str, option: str) -> bool: + return written.lower() == option.lower() if self.program in _ANY_CASE else written == option + def option_like(self, text: str) -> bool: return text.startswith("-") or any( - option.option == text for option in OPTIONS if option.program == self.program + self.same(text, option.option) for option in OPTIONS if option.program == self.program ) def free(self, index: int, how: str) -> _Reading: @@ -386,48 +552,6 @@ def value_of(self, option: CommandOption, given: str) -> _Reading: ) return _Reading(f"the input is the value of {self.program} {given}", Severity.MEDIUM) - def operand_command(self) -> _Reading: - assert self.program is not None - return _Reading( - f"the input {COMMAND_OPERANDS[self.program][1]}", Severity.HIGH, command=True - ) - - def command_operand(self, index: int) -> bool | None: - """Whether the element at ``index`` is one of the operands the program runs as a - command: False when the program has none or the element comes before them, None - when the options before it do not tell.""" - - if self.program not in COMMAND_OPERANDS: - return False - operands = self.operands_before(index) - return None if operands is None else operands >= COMMAND_OPERANDS[self.program][0] - - def operands_before(self, index: int) -> int | None: - """The number of operands before ``index``, or None when an element of unknown - value, or an option that may take one, comes first.""" - - count, position = 0, 1 - while position < index: - text = self.text(position) - if text is None: - return None - if text == "--": - return count + index - position - 1 - if text.startswith("-") and text != "-": - if not (text.startswith("--") and "=" in text): - taker = self.taker(text, position + 1) - if taker is None: - return None - if isinstance(taker, CommandOption): - position += 1 - else: - count += 1 - position += 1 - return count - - def after_end_of_options(self, index: int) -> bool: - return any(self.text(position) == "--" for position in range(1, index)) - def taker(self, before: str, index: int) -> CommandOption | bool | None: """The option ``before`` whose value is the element at ``index``; False when the model establishes that ``before`` takes no value there, None when it cannot tell. @@ -449,12 +573,15 @@ def taker(self, before: str, index: int) -> CommandOption | bool | None: return False def option(self, name: str, index: int) -> CommandOption | None: - return next((option for option in self.applicable(index) if option.option == name), None) + return next( + (option for option in self.applicable(index) if self.same(name, option.option)), None + ) def applicable(self, index: int) -> list[CommandOption]: """The options of ``OPTIONS`` the element at ``index`` may give: those of the - program, after their subcommand when the command shows it before the element, or - whatever the subcommand when the command does not tell.""" + program, after their subcommand when the command shows it before the element. When + the command does not tell, options that run a command still apply, whatever their + subcommand, but not an inert reading, which needs the subcommand established.""" if self.program is None: return [] @@ -463,7 +590,11 @@ def applicable(self, index: int) -> list[CommandOption]: option for option in OPTIONS if option.program == self.program - and (not option.subcommands or subcommand is None or subcommand in option.subcommands) + and ( + not option.subcommands + or (subcommand is None and option.inert is None) + or subcommand in option.subcommands + ) ] def subcommand(self, index: int) -> str | None: @@ -478,23 +609,53 @@ def subcommand(self, index: int) -> str | None: return "" -def _spelled(option: str) -> str: - """An option with its value in the same element: ``--name=...`` or ``-o...``.""" +def _name(program: str) -> str: + """The name a program is known by in ``OPTIONS``: without its directory (POSIX or + Windows) or ``.exe``, in lower case, any ``python3.x`` as ``python``.""" - return f"{option}=..." if option.startswith("--") else f"{option}..." + name = re.split(r"[\\/]", program)[-1].lower().removesuffix(".exe") + return "python" if _PYTHON.fullmatch(name) else name + + +def _first_program( + part: CommandPart, defs: Mapping[Value, Instruction], strings: Mapping[str, str] +) -> str | None: + """The program the first element of a command names: its constant text, or Python + for ``sys.executable``; None when unknown.""" + + if not part.known: + return None + text, whole = leading_text(part.value, defs, strings) + if whole: + return text + made = defs.get(part.value) + return "python" if isinstance(made, Symbol) and made.symbol_id == _PYTHON_EXECUTABLE else None -def _executable(arguments: Arguments) -> str | None: - """The program a constant ``executable`` names, which the command's first element - then only labels.""" +def _program( + arguments: Arguments, first: CommandPart, defs: Mapping[Value, Instruction] +) -> str | None: + """The program a call runs: the one a constant ``executable`` names, which the + command's first element then only labels, or the one that element names. An + ``executable`` that is given but cannot be read leaves the program unknown.""" given = arguments.given("executable", _EXECUTABLE) + if given == (False, None): + return _first_program(first, defs, {}) written = given[1] if given is not None else None + if written == str(_PYTHON_EXECUTABLE): + return "python" if written is None or len(written) < 2 or written[0] not in "'\"" or written[-1] != written[0]: return None return written[1:-1].replace("\\\\", "\\") +def _spelled(option: str) -> str: + """An option with its value in the same element: ``--name=...`` or ``-o...``.""" + + return f"{option}=..." if option.startswith("--") else f"{option}..." + + def _shell(arguments: Arguments) -> bool | None: """Whether a shell runs the command: True or False when the call says so with a constant, by keyword or at its place, None when an expression or unpacked arguments diff --git a/tests/test_command_arguments.py b/tests/test_command_arguments.py index a5b5584..1a15348 100644 --- a/tests/test_command_arguments.py +++ b/tests/test_command_arguments.py @@ -129,10 +129,43 @@ def judged(call: str) -> list[tuple[str, str, str]]: ('cmd = ["ls", "-l"]\n cmd.insert(0, x)\n subprocess.run(cmd)', HIGH), ('cmd = ["ls"]\n cmd += [x]\n subprocess.run(cmd)', HIGH), ('cmd = ["ls", "-l"]\n cmd[0] = x\n subprocess.run(cmd)', HIGH), + # Second review: an element of unknown value before the input leaves it to review. + ('subprocess.run(["tool", extra, f"echo {x}"])', MEDIUM_TO_REVIEW), + ('subprocess.run(["sh", extra, f"echo {x}"])', HIGH), + # A shell's -c is a flag: its first operand is the command, later ones are $1, $2... + ('subprocess.run(["sh", "-ce", f"echo {x}"])', HIGH), + ('subprocess.run(["bash", "-c", "-e", f"echo {x}"])', HIGH), + ('subprocess.run(["sh", "-c", "--", f"echo {x}"])', HIGH), + ('subprocess.run(["sh", "-lc", x])', HIGH), + ('subprocess.run(["bash", "-o", "pipefail", "-c", x])', HIGH), + ('subprocess.run(["sh", "-c", "echo $1", "sh", f"v{x}"])', None), + # After the first word of a remote command, ssh operands are shell text. + ('subprocess.run(["ssh", "host", "ls", f"-l{x}"])', HIGH), + ('subprocess.run(["ssh", "host", "ls", f"--color={x}"])', HIGH), + # A command inside the command: find -exec, a wrapper's program. + ('subprocess.run(["find", ".", "-exec", "sh", "-c", x, ";"])', HIGH), + ('subprocess.run(["find", ".", "-exec", "rm", x, ";"])', MEDIUM), + ('subprocess.run(["sudo", "ls", x])', MEDIUM), + ('subprocess.run(["sudo", "sh", "-c", f"echo {x}"])', HIGH), + # Inert values only where the subcommand is established, and one element only. + ('subprocess.run(["git", "commit", "-m", *x.split()])', MEDIUM), + ('subprocess.run(["git", "-C", "repo", "branch", "-m", x])', HIGH_TO_REVIEW), + ('subprocess.run(["gh", "--title", x])', MEDIUM_TO_REVIEW), + ('subprocess.run(["git", "clone", f"-u{x}", "url"])', HIGH), + ('subprocess.run(["git", "clone", "-u", x, "url"])', HIGH), + # Windows: PowerShell runs its operands, parameters ignore case, /c may be glued. + ('subprocess.run(["powershell", f"Get-Item {x}"])', HIGH), + ('subprocess.run(["powershell", "-NoProfile", f"Get-Item {x}"])', HIGH), + ('subprocess.run(["powershell", "-COMMAND", x])', HIGH), + ('subprocess.run(["cmd", f"/c{x}"])', HIGH), + ('subprocess.run(["CMD.EXE", "/C", x])', HIGH), + # An executable that cannot be read leaves the program unknown. + ('subprocess.run(["git", "commit", "-m", x], executable=extra)', MEDIUM_TO_REVIEW), + ('subprocess.run(["x", "-c", f"echo {x}"], executable=sys.executable)', HIGH), # Unpacked parts keep their place in the command. ('subprocess.run(["xcrun", "notarytool", *(["--keychain", x])])', MEDIUM_TO_REVIEW), ("subprocess.run([*extra, x])", HIGH), - ('subprocess.run(["ls", *extra, x])', MEDIUM), + ('subprocess.run(["ls", *extra, x])', MEDIUM_TO_REVIEW), ('subprocess.run(["ls", *x.split()])', MEDIUM), ], ) From 0c5aba48715f7cc02a6aba202d9c7428ead3d4af Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 6 Oct 2026 07:57:07 +0200 Subject: [PATCH 4/8] fix(command-injection): read shell options, environment assignments, unknown programs Third review: a shell option the model does not describe (bash -O extglob, --rcfile f, sh +e) hid -c, so sh -c injections were refuted; NAME=VALUE before the program of env or sudo was taken for the program; with the program unknown, a constant prefix refuted an operand it may run; a deep chain of wrappers raised RecursionError. Reading the options before a shell's operands now stops at an option it cannot read, the shells' + options and bash long options are modelled, env and sudo skip their assignments, an unknown program keeps the finding for review, and shutil.which(name) names the program. Commands inside commands are read to a bounded depth, and element texts and options are read once per command. Refs #208 --- .../command_injection/command_injection.py | 142 +++++++++++++----- tests/test_command_arguments.py | 21 ++- 2 files changed, 123 insertions(+), 40 deletions(-) diff --git a/src/coretrace_python/bundled/security/command_injection/command_injection.py b/src/coretrace_python/bundled/security/command_injection/command_injection.py index 6822c18..2fdd1d5 100644 --- a/src/coretrace_python/bundled/security/command_injection/command_injection.py +++ b/src/coretrace_python/bundled/security/command_injection/command_injection.py @@ -36,7 +36,7 @@ from coretrace_python.hir import nodes from coretrace_python.interprocedural import Arguments from coretrace_python.ir.defuse import DefUseAnalysis -from coretrace_python.ir.model import Instruction, Symbol, Value +from coretrace_python.ir.model import Call, Instruction, Symbol, Value from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.plugins import Assessment, PluginContext, TaintDetector from coretrace_python.semantic.symbols import SymbolId @@ -70,14 +70,17 @@ class CommandOption: @dataclass(frozen=True) class CommandOperands: """Operands a program runs, counted from 0 after its options: from operand ``first`` - on, either shell ``text`` the program has a shell run, or a program and its arguments, - a command of its own. With ``after``, only when that letter is among the program's - short options, and then operand ``first`` only (``sh -c 'cmd' name arguments``).""" + on, either shell ``text`` the program has a shell run, or a program and its arguments, + a command of its own. With ``after``, only when that letter is among the program's + short options, and then operand ``first`` only (``sh -c 'cmd' name arguments``). With + ``assignments``, ``NAME=VALUE`` operands before the program set its environment and do + not count (``env A=1 cmd``).""" first: int text: bool how: str after: str | None = None + assignments: bool = False def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: @@ -131,7 +134,23 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: CommandOption("find", option, "runs the command that follows", single=False, opens=True) for option in ("-exec", "-execdir", "-ok", "-okdir") ), - *(CommandOption(shell, "-o", single=False) for shell in _SHELLS), + *(CommandOption(shell, sign + "o", single=False) for shell in _SHELLS for sign in "-+"), + *( + option + for shell in _SHELLS + for option in _flags(shell, *(f"+{letter}" for letter in "abCefhmnuvx")) + ), + *_flags( + "bash", + "--login", + "--noediting", + "--noprofile", + "--norc", + "--posix", + "--restricted", + "--verbose", + ), + *(CommandOption("bash", option, single=False) for option in ("--rcfile", "--init-file")), *( option for shell in _SHELLS @@ -181,7 +200,9 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: }, "timeout": CommandOperands(1, False, "names the program timeout runs"), **{ - wrapper: CommandOperands(0, False, f"names the program {wrapper} runs") + wrapper: CommandOperands( + 0, False, f"names the program {wrapper} runs", assignments=wrapper in ("env", "sudo") + ) for wrapper in _WRAPPERS }, } @@ -190,6 +211,11 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: _BATCH = (".bat", ".cmd") _ANY_CASE = frozenset({"cmd", "powershell", "pwsh"}) _TERMINATORS = (";", "+") +_ASSIGNMENT = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=") +_WHICH = SymbolId("python.shutil.which") +# Commands inside commands are read to this depth; deeper, an element is taken to be an +# operand the program runs. +_MAX_DEPTH = 32 _PROCESS_STARTERS = frozenset( SymbolId(f"python.subprocess.{name}") @@ -298,30 +324,37 @@ def __init__( defs: Mapping[Value, Instruction], strings: Mapping[str, str], program: str | None, + depth: int = 0, ) -> None: self.parts = parts self.defs = defs self.strings = strings + self.depth = depth name = _name(program) if program is not None else None self.batch = name is not None and name.endswith(_BATCH) self.program = name + # Each element's text, and the options by subcommand, are read once: ``read`` + # looks back over the command for every tainted element. + self._texts: dict[int, str | None] = {} + self._options: dict[str | None, list[CommandOption]] = {} def nested(self, start: int, end: int | None = None) -> _Command: """The command that the elements from ``start`` (to ``end``) spell.""" parts = self.parts[start:end] - return _Command( - parts, self.defs, self.strings, _first_program(parts[0], self.defs, self.strings) - ) + program = _first_program(parts[0], self.defs, self.strings) + return _Command(parts, self.defs, self.strings, program, self.depth + 1) def text(self, index: int) -> str | None: """The constant text of the element at ``index``, when all of it is constant.""" - part = self.parts[index] - if not part.known: - return None - text, whole = leading_text(part.value, self.defs, self.strings) - return text if whole else None + if index not in self._texts: + part = self.parts[index] + text, whole = ( + leading_text(part.value, self.defs, self.strings) if part.known else ("", False) + ) + self._texts[index] = text if whole else None + return self._texts[index] def prefix(self, index: int) -> str: part = self.parts[index] @@ -342,6 +375,11 @@ def read(self, index: int) -> _Reading: operand = self.operand(index) if operand is not None: return operand + if self.program is None: + free = self.free(index, "the input is an element") + return self.review( + f"the program is unknown and may run its operands; {free.reason}", free + ) for position in range(1, index): option = self.option(self.text(position) or "", position) if option is not None and option.opens: @@ -423,7 +461,7 @@ def operand(self, index: int) -> _Reading | None: return unclear if not given: return None - operands = self.operands(index, spec.first + 1) + operands = self.operands(index, spec.first + 1, spec.assignments) if operands is None: return unclear if operands.consumed: @@ -450,6 +488,10 @@ def inside(self, start: int, end: int | None, index: int) -> _Reading: return _Reading( f"the input names the program {self.program} runs", Severity.HIGH, command=True ) + if self.depth >= _MAX_DEPTH: + return _Reading( + f"the input is in a command {self.program} runs", Severity.HIGH, command=True + ) reading = self.nested(start, end).read(index - start) return _Reading( f"in the command {self.program} runs, {reading.reason}", @@ -458,9 +500,10 @@ def inside(self, start: int, end: int | None, index: int) -> _Reading: reading.to_review, ) - def operands(self, index: int, enough: int) -> _Operands | None: - """The operands before ``index``, up to ``enough`` of them; None when an element of - unknown value, or an option that may take one, comes first.""" + def operands(self, index: int, enough: int, assignments: bool = False) -> _Operands | None: + """The operands before ``index``, up to ``enough`` of them, without ``NAME=VALUE`` + assignments when the program reads them; None when an element of unknown value, or + an option that may take one, comes first.""" positions: list[int] = [] position = 1 @@ -479,7 +522,7 @@ def operands(self, index: int, enough: int) -> _Operands | None: position += 1 if position == index: return _Operands(tuple(positions), consumed=True) - else: + elif not (assignments and _ASSIGNMENT.match(text)): positions.append(position) position += 1 return _Operands(tuple(positions)) @@ -493,12 +536,16 @@ def letter_given(self, letter: str, index: int) -> bool | None: text = self.text(position) if text is None: return None - if text == "--" or not text.startswith("-") or text == "-": + if text == "--" or not self.option_like(text) or text in ("-", "+"): return False - if not text.startswith("--") and letter in text[1:]: + if text.startswith("-") and not text.startswith("--") and letter in text[1:]: return True - if isinstance(self.taker(text, position + 1), CommandOption): - position += 1 + if not (text.startswith("--") and "=" in text): + taker = self.taker(text, position + 1) + if taker is None: + return None + if isinstance(taker, CommandOption): + position += 1 position += 1 return False @@ -509,7 +556,14 @@ def same(self, written: str, option: str) -> bool: return written.lower() == option.lower() if self.program in _ANY_CASE else written == option def option_like(self, text: str) -> bool: - return text.startswith("-") or any( + """Whether ``text`` reads as an option of the program: ``-x``, a shell's ``+x``, or + an option the model names (``/c``).""" + + if text.startswith("-") or ( + self.program in _SHELLS and text.startswith("+") and text != "+" + ): + return True + return any( self.same(text, option.option) for option in OPTIONS if option.program == self.program ) @@ -561,11 +615,12 @@ def taker(self, before: str, index: int) -> CommandOption | bool | None: option = self.option(before, index) if option is not None: return option if option.takes_value else False - if not before.startswith("-") or before.startswith("--") or len(before) <= 2: + sign = before[:1] + if sign not in ("-", "+") or before.startswith("--") or len(before) <= 2: return None letters = before[1:] for position, letter in enumerate(letters): - option = self.option(f"-{letter}", index) + option = self.option(f"{sign}{letter}", index) if option is None: return None if option.takes_value: @@ -586,16 +641,18 @@ def applicable(self, index: int) -> list[CommandOption]: if self.program is None: return [] subcommand = self.subcommand(index) - return [ - option - for option in OPTIONS - if option.program == self.program - and ( - not option.subcommands - or (subcommand is None and option.inert is None) - or subcommand in option.subcommands - ) - ] + if subcommand not in self._options: + self._options[subcommand] = [ + option + for option in OPTIONS + if option.program == self.program + and ( + not option.subcommands + or (subcommand is None and option.inert is None) + or subcommand in option.subcommands + ) + ] + return self._options[subcommand] def subcommand(self, index: int) -> str | None: """The first operand before ``index``, ``""`` when there is none, or None when an @@ -620,8 +677,8 @@ def _name(program: str) -> str: def _first_program( part: CommandPart, defs: Mapping[Value, Instruction], strings: Mapping[str, str] ) -> str | None: - """The program the first element of a command names: its constant text, or Python - for ``sys.executable``; None when unknown.""" + """The program the first element of a command names: its constant text, Python for + ``sys.executable``, or the name ``shutil.which`` looks up; None when unknown.""" if not part.known: return None @@ -629,7 +686,14 @@ def _first_program( if whole: return text made = defs.get(part.value) - return "python" if isinstance(made, Symbol) and made.symbol_id == _PYTHON_EXECUTABLE else None + if isinstance(made, Symbol) and made.symbol_id == _PYTHON_EXECUTABLE: + return "python" + if isinstance(made, Call) and made.arguments: + callee = defs.get(made.callee) + if isinstance(callee, Symbol) and callee.symbol_id == _WHICH: + name, whole = leading_text(made.arguments[0], defs, strings) + return name if whole else None + return None def _program( diff --git a/tests/test_command_arguments.py b/tests/test_command_arguments.py index 1a15348..d400d15 100644 --- a/tests/test_command_arguments.py +++ b/tests/test_command_arguments.py @@ -41,7 +41,7 @@ def judged(call: str) -> list[tuple[str, str, str]]: """Severity, verdict and confidence of each command-injection finding on ``call``, made with ``x`` read from the command line and ``extra`` of unknown value.""" - text = f"import os\nimport subprocess\nimport sys\n\n\ndef run(extra):\n x = sys.argv[1]\n {call}\n" + text = f"import os\nimport shutil\nimport subprocess\nimport sys\n\n\ndef run(extra):\n x = sys.argv[1]\n {call}\n" findings = engine.check(SourceManager().add_source("app/run.py", text), [PLUGINS]) return [ (f.severity.value, f.metadata["verdict"], f.confidence.value) @@ -162,6 +162,20 @@ def judged(call: str) -> list[tuple[str, str, str]]: # An executable that cannot be read leaves the program unknown. ('subprocess.run(["git", "commit", "-m", x], executable=extra)', MEDIUM_TO_REVIEW), ('subprocess.run(["x", "-c", f"echo {x}"], executable=sys.executable)', HIGH), + # Third review: options a shell reads, assignments before a wrapper's program. + ('subprocess.run(["bash", "-O", "extglob", "-c", f"echo {x}"])', HIGH), + ('subprocess.run(["bash", "--rcfile", "f", "-c", f"echo {x}"])', HIGH), + ('subprocess.run(["sh", "+e", "-c", f"echo {x}"])', HIGH), + ('subprocess.run(["bash", "+o", "history", "-c", f"echo {x}"])', HIGH), + ('subprocess.run(["env", "PATH=/usr/bin", x])', HIGH), + ('subprocess.run(["sudo", "A=1", x])', HIGH), + ('subprocess.run(["env", "A=1", "sh", "-c", f"echo {x}"])', HIGH), + ('subprocess.run(["env", "A=1", "ls", x])', MEDIUM), + # An unknown program may run its operands: kept for review, never refuted. + ('subprocess.run([extra, f"echo {x}"])', MEDIUM_TO_REVIEW), + ('subprocess.run(["ls", f"echo {x}"], -1, extra)', MEDIUM_TO_REVIEW), + ('subprocess.run([shutil.which("ssh"), "host", f"cat {x}"])', HIGH), + ('subprocess.run([shutil.which("bash"), "-c", f"echo {x}"])', HIGH), # Unpacked parts keep their place in the command. ('subprocess.run(["xcrun", "notarytool", *(["--keychain", x])])', MEDIUM_TO_REVIEW), ("subprocess.run([*extra, x])", HIGH), @@ -187,3 +201,8 @@ def test_an_unpacked_element_keeps_its_place_in_a_list_or_tuple_display() -> Non assert built_list.unpacked_at == (1, 3) assert built_tuple.unpacked_at == (0,) + + +def test_a_deeply_nested_command_is_still_read() -> None: + wrappers = '"sudo", ' * 500 + assert judged(f"subprocess.run([{wrappers}x])") == [HIGH] From 3f5fe0556b63938f4365f2f4e664ed977ce407b0 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 00:16:20 +0200 Subject: [PATCH 5/8] fix(command-injection): report the script an interpreter runs, to review A script path chosen by the input was refuted as an operand when it had a constant prefix (bash f"/tmp/{x}.sh"), unlike the program of a wrapper. Interpreters now declare the script their first operand names: sh, bash, dash, zsh, ksh, python, perl, ruby, node and php, unless an option gives the code (-c, -m, -e, --eval, -r). The input there is high, kept for review: execution is possible, and its exploitability depends on the file the input chooses; so is the module of python -m. The code options of perl, ruby, node and php run their value, and python's flags are modelled so that its options can be read. Refs #208 --- docs/usage.md | 2 +- .../command_injection/command_injection.py | 162 +++++++++++++----- tests/test_command_arguments.py | 18 ++ 3 files changed, 136 insertions(+), 46 deletions(-) diff --git a/docs/usage.md b/docs/usage.md index e0394c8..43fa513 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -115,7 +115,7 @@ variable is not treated that way yet. | Rule | Reached sink | |---|---| -| `command-injection` | Shell or process execution (`os.system`, `subprocess` with a string, …), judged on what the input controls in the command. A string a shell runs, a string naming the program, the program itself, or an operand the program runs as a command (after the host for `ssh`, the program of `sudo`, `env`, `xargs`, `timeout`…, any argument of a batch file) is `high`. Otherwise an element of an argument list run without a shell can only be read as an option: `medium`, or `high` when a documented option of the program runs a command it chooses (`git push --receive-pack=…`, `ssh -o ProxyCommand=…`). A value a documented option takes is judged by what the option does with it, whatever its prefix: run it (`sh -c`, `cmd /c`, `python -c`: `high`), keep it as data (`git commit -m`: refuted). After an option the rule does not know, the finding stays a hotspot to review. A constant prefix that keeps a standalone input from being an option (`f"v{version}"`), or an operand after `--` for a program that honours it, is refuted. | +| `command-injection` | Shell or process execution (`os.system`, `subprocess` with a string, …), judged on what the input controls in the command. A string a shell runs, a string naming the program, the program itself, or an operand the program runs as a command (after the host for `ssh`, the program of `sudo`, `env`, `xargs`, `timeout`…, any argument of a batch file) is `high`. The script an interpreter runs (`bash run.sh`, `python tool.py`, its first operand when no option such as `-c` gives the code), or the module of `python -m`, is `high` and kept for review: execution is possible, and its exploitability depends on the file the input chooses. Otherwise an element of an argument list run without a shell can only be read as an option: `medium`, or `high` when a documented option of the program runs a command it chooses (`git push --receive-pack=…`, `ssh -o ProxyCommand=…`). A value a documented option takes is judged by what the option does with it, whatever its prefix: run it (`sh -c`, `cmd /c`, `python -c`: `high`), keep it as data (`git commit -m`: refuted). After an option the rule does not know, the finding stays a hotspot to review. A constant prefix that keeps a standalone input from being an option (`f"v{version}"`), or an operand after `--` for a program that honours it, is refuted. | | `code-injection` | The code `eval` or `exec` runs; the globals and locals passed with it are data. `dangerous-eval` reports every such call, this rule the ones attacker input reaches. | | `sql-injection` | A database statement (`cursor.execute`, SQLAlchemy `text`, Django `raw`, …). Parameters of a parameterised query are not statements. | | `nosql-injection` | A NoSQL query, such as a MongoDB filter, given attacker input that may hold a structure: a JSON body giving `{"$ne": null}` matches any value. The input that may hold a structure is JSON, the raw bodies a `json.loads` decodes, files, WebSocket messages and standard input. Text cannot hold an operator: form fields, query parameters, headers, cookies, URL parameters, the command line, the environment, and an entry-point parameter annotated with scalars (`str`, `int`, `Optional[str]`, `Annotated[str, Query()]`, `list[str]`, …), which FastAPI validates. A Django, REST framework, aiohttp or tornado view reads a request object: its text attributes (`GET`, `POST`, `headers`, `query_params`, `query`, `match_info`, `arguments`, …) and the URL parameters after it are text, its body, files and `data` are not. Limits: a helper function the request is passed to reads it as a whole; the keys of a dict are not followed, so a string concatenated into `$where` is not reported. The analyzer ships no NoSQL sink: model plugins declare them. | diff --git a/src/coretrace_python/bundled/security/command_injection/command_injection.py b/src/coretrace_python/bundled/security/command_injection/command_injection.py index 2fdd1d5..ba75eff 100644 --- a/src/coretrace_python/bundled/security/command_injection/command_injection.py +++ b/src/coretrace_python/bundled/security/command_injection/command_injection.py @@ -5,7 +5,10 @@ the program itself, or an operand that the program runs as shell text (``ssh`` after the host, a shell's first operand after ``-c``, any argument of a batch file) is a command injection, ``high``. Operands that are a command of their own (the program a wrapper such -as ``sudo`` runs, what follows ``find -exec``) are read again as that command. An element +as ``sudo`` runs, what follows ``find -exec``) are read again as that command. The script +an interpreter runs (its first operand when no option gives the code, as ``bash run.sh`` or +``python tool.py``), or the module ``python -m`` runs, is ``high`` to review: execution is +possible, and whether it is exploitable depends on the file the input chooses. An element of an argument list run without a shell otherwise injects at most an option: ``medium``, or ``high`` when ``OPTIONS`` establishes that the program runs a command chosen through an option that fits there. @@ -52,8 +55,9 @@ class CommandOption: and its value; with ``rest``, every element after it is its value, and with ``opens`` the elements after it, up to ``;`` or ``+``, are a command of their own. What the program does with the value decides: with ``runs``, it runs a command the value - chooses; with ``inert``, it only keeps it as data (a commit message); with neither, - the effect is not established. An option that takes no value leaves the next element + chooses (``examine``: a choice among code already there, such as a module, whose + exploitability is to examine); with ``inert``, it only keeps it as data (a commit + message); with neither, the effect is not established. An option that takes no value leaves the next element free.""" program: str @@ -65,21 +69,25 @@ class CommandOption: inert: str | None = None rest: bool = False opens: bool = False + examine: bool = False @dataclass(frozen=True) class CommandOperands: - """Operands a program runs, counted from 0 after its options: from operand ``first`` - on, either shell ``text`` the program has a shell run, or a program and its arguments, - a command of its own. With ``after``, only when that letter is among the program's - short options, and then operand ``first`` only (``sh -c 'cmd' name arguments``). With - ``assignments``, ``NAME=VALUE`` operands before the program set its environment and do - not count (``env A=1 cmd``).""" + """Operands a program runs, counted from 0 after its options, of one ``kind``: from + operand ``first`` on, shell ``text`` the program has a shell run, or a ``command`` of + its own (a program and its arguments); or operand ``first`` only, a ``script`` file + the program runs, whose contents decide what runs. With ``after``, only when one of + those options comes before the operands, and then operand ``first`` only (``sh -c + 'cmd' name arguments``); with ``unless``, only when none of them does (``python -c`` + gives the code, not a script). With ``assignments``, ``NAME=VALUE`` operands before + the program set its environment and do not count (``env A=1 cmd``).""" first: int - text: bool + kind: str how: str - after: str | None = None + after: tuple[str, ...] = () + unless: tuple[str, ...] = () assignments: bool = False @@ -157,6 +165,16 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: for option in _flags(shell, *(f"-{letter}" for letter in "abCcefhilmnsuvx")) ), CommandOption("python", "-c", "runs its value as Python code"), + CommandOption("python", "-m", "runs the module it names", examine=True), + *(CommandOption("python", option, single=False) for option in ("-W", "-X")), + *_flags("python", *(f"-{letter}" for letter in "bBdEhiIOPqsSuvVx")), + *(CommandOption("perl", option, "runs its value as Perl code") for option in ("-e", "-E")), + CommandOption("ruby", "-e", "runs its value as Ruby code"), + *( + CommandOption("node", option, "runs its value as JavaScript code") + for option in ("-e", "--eval", "-p", "--print") + ), + CommandOption("php", "-r", "runs its value as PHP code"), *( CommandOption( "cmd", option, "runs the rest of the command line as a command", single=False, rest=True @@ -191,20 +209,44 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: ) # Programs documented to read ``--`` as the end of their options. END_OF_OPTIONS = frozenset({"git", "tar", "rsync", "rm"}) -OPERANDS: Mapping[str, CommandOperands] = { - "ssh": CommandOperands(1, True, "is part of the command the remote shell runs"), - "powershell": CommandOperands(0, True, "is part of the command PowerShell runs"), - **{ - shell: CommandOperands(0, True, "is the command the shell runs after -c", after="c") - for shell in _SHELLS - }, - "timeout": CommandOperands(1, False, "names the program timeout runs"), +# Interpreters run the script their first operand names, unless one of these options +# gives the code instead. +_SCRIPTS: Mapping[str, tuple[str, ...]] = { + **{shell: ("-c",) for shell in _SHELLS}, + "python": ("-c", "-m"), + "perl": ("-e", "-E"), + "ruby": ("-e",), + "node": ("-e", "--eval", "-p", "--print"), + "php": ("-r",), +} + + +def _interpreter(name: str, code: tuple[str, ...]) -> tuple[CommandOperands, ...]: + script = CommandOperands(0, "script", f"names the script {name} runs", unless=code) + if name not in _SHELLS: + return (script,) + return ( + CommandOperands(0, "text", "is the command the shell runs after -c", after=("-c",)), + script, + ) + + +OPERANDS: Mapping[str, tuple[CommandOperands, ...]] = { + "ssh": (CommandOperands(1, "text", "is part of the command the remote shell runs"),), + "powershell": (CommandOperands(0, "text", "is part of the command PowerShell runs"),), + "timeout": (CommandOperands(1, "command", "names the program timeout runs"),), **{ - wrapper: CommandOperands( - 0, False, f"names the program {wrapper} runs", assignments=wrapper in ("env", "sudo") + wrapper: ( + CommandOperands( + 0, + "command", + f"names the program {wrapper} runs", + assignments=wrapper in ("env", "sudo"), + ), ) for wrapper in _WRAPPERS }, + **{interpreter: _interpreter(interpreter, code) for interpreter, code in _SCRIPTS.items()}, } # Windows runs a batch file through cmd.exe, which parses its arguments again; its own # programs read their options whatever the case. @@ -449,18 +491,27 @@ def operand(self, index: int) -> _Reading | None: """The reading of an element among the operands the program runs, or None when the element is not one of them.""" - if self.program not in OPERANDS: - return None - spec = OPERANDS[self.program] + for spec in OPERANDS.get(self.program or "", ()): + reading = self.operand_of(spec, index) + if reading is not None: + return reading + return None + + def operand_of(self, spec: CommandOperands, index: int) -> _Reading | None: + script = spec.kind == "script" unclear = _Reading( - f"the input may be an operand that {spec.how}", Severity.HIGH, command=True + f"the input may be an operand that {spec.how}", + Severity.HIGH, + command=not script, + to_review=script, ) - if spec.after is not None: - given = self.letter_given(spec.after, index) - if given is None: - return unclear - if not given: - return None + for names, wanted in ((spec.after, True), (spec.unless, False)): + if names: + given = self.given(names, index) + if given is None: + return unclear + if given is not wanted: + return None operands = self.operands(index, spec.first + 1, spec.assignments) if operands is None: return unclear @@ -469,14 +520,18 @@ def operand(self, index: int) -> _Reading | None: count = len(operands.positions) operand_like = operands.ended or not self.prefix(index).startswith("-") if count > spec.first: - if not spec.text: + if spec.kind == "command": return self.inside(operands.positions[spec.first], None, index) - return ( - None - if spec.after is not None - else _Reading(f"the input {spec.how}", Severity.HIGH, command=True) - ) + if spec.kind == "text" and not spec.after: + return _Reading(f"the input {spec.how}", Severity.HIGH, command=True) + return None if count == spec.first and operand_like: + if script: + return _Reading( + f"the input {spec.how}, whose contents decide what runs", + Severity.HIGH, + to_review=True, + ) return _Reading(f"the input {spec.how}", Severity.HIGH, command=True) return None @@ -527,25 +582,36 @@ def operands(self, index: int, enough: int, assignments: bool = False) -> _Opera position += 1 return _Operands(tuple(positions)) - def letter_given(self, letter: str, index: int) -> bool | None: - """Whether a short option ``letter`` comes among the options before ``index``, None - when an element of unknown value comes first.""" + def given(self, names: tuple[str, ...], index: int) -> bool | None: + """Whether one of the options ``names`` comes among the options before ``index``, + read as getopt reads them (a short option that takes a value takes the rest of its + element); None when an element of unknown value, or an option that may take one, + comes first.""" position = 1 while position < index: text = self.text(position) if text is None: return None - if text == "--" or not self.option_like(text) or text in ("-", "+"): + if text == "--" or text in ("-", "+") or not self.option_like(text): return False - if text.startswith("-") and not text.startswith("--") and letter in text[1:]: + if text[0] in "-+" and not text.startswith("--") and len(text) > 2: + for offset, letter in enumerate(text[1:], start=1): + option = self.option(text[0] + letter, position + 1) + if text[0] + letter in names: + return True + if option is None: + return None + if option.takes_value: + position += offset == len(text) - 1 + break + elif text.split("=", 1)[0] in names: return True - if not (text.startswith("--") and "=" in text): + elif not (text.startswith("--") and "=" in text): taker = self.taker(text, position + 1) if taker is None: return None - if isinstance(taker, CommandOption): - position += 1 + position += isinstance(taker, CommandOption) position += 1 return False @@ -595,6 +661,12 @@ def option_value(self, index: int, name: str) -> _Reading: return self.value_of(option, name) def value_of(self, option: CommandOption, given: str) -> _Reading: + if option.runs is not None and option.examine: + return _Reading( + f"the input is the value of {self.program} {given}, which {option.runs}", + Severity.HIGH, + to_review=True, + ) if option.runs is not None: return _Reading( f"the input is the value of {self.program} {given}, which {option.runs}", diff --git a/tests/test_command_arguments.py b/tests/test_command_arguments.py index d400d15..cc82218 100644 --- a/tests/test_command_arguments.py +++ b/tests/test_command_arguments.py @@ -176,6 +176,24 @@ def judged(call: str) -> list[tuple[str, str, str]]: ('subprocess.run(["ls", f"echo {x}"], -1, extra)', MEDIUM_TO_REVIEW), ('subprocess.run([shutil.which("ssh"), "host", f"cat {x}"])', HIGH), ('subprocess.run([shutil.which("bash"), "-c", f"echo {x}"])', HIGH), + # An interpreter runs the script its first operand names, unless code is given: + # execution is possible, exploitability depends on the file, so high to review. + ('subprocess.run(["bash", f"/tmp/{x}.sh"])', HIGH_TO_REVIEW), + ('subprocess.run(["sh", x])', HIGH_TO_REVIEW), + ('subprocess.run(["bash", "-x", "--", f"run-{x}.sh"])', HIGH_TO_REVIEW), + ('subprocess.run(["python3", f"scripts/{x}.py"])', HIGH_TO_REVIEW), + ('subprocess.run([sys.executable, f"tools/{x}.py", "--flag"])', HIGH_TO_REVIEW), + ('subprocess.run(["python3", "-m", f"pkg.{x}"])', HIGH_TO_REVIEW), + ('subprocess.run(["perl", f"{x}.pl"])', HIGH_TO_REVIEW), + ('subprocess.run(["node", f"{x}.js"])', HIGH_TO_REVIEW), + # Code given by an option runs; the operands after it are the script's arguments. + ('subprocess.run(["perl", "-e", f"print {x}"])', HIGH), + ('subprocess.run(["ruby", "-e", x])', HIGH), + ('subprocess.run(["node", "--eval", x])', HIGH), + ('subprocess.run(["php", "-r", x])', HIGH), + ('subprocess.run(["python3", "-c", "import sys; print(sys.argv[1])", f"v{x}"])', None), + ('subprocess.run(["bash", "script.sh", f"v{x}"])', None), + ('subprocess.run(["python3", "script.py", f"--name={x}"])', MEDIUM), # Unpacked parts keep their place in the command. ('subprocess.run(["xcrun", "notarytool", *(["--keychain", x])])', MEDIUM_TO_REVIEW), ("subprocess.run([*extra, x])", HIGH), From 6f34e0c01d9d72f96dcba6c298a0ba945d0523c9 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 00:42:27 +0200 Subject: [PATCH 6/8] fix(command-injection): read script arguments apart from interpreter options Review of the interpreter script model: python -W, which imports the module its category names, was medium; a whole first operand of python or node, which may be a code option and its value, became a hotspot; the interpreter's own options were applied to the script's arguments (node app.js -e x read as node -e); versioned or distribution names (nodejs, pypy3, perl5.36) were not recognised; perl -M, ruby -r, node --require and php -B/-R/-E were missing; bash -s reads its script from the standard input. Arguments after a script or code are now read as the script's own: its options, of unknown meaning, kept for review after an option and refuted behind a constant prefix. An input that starts the script operand keeps the high reading of a code option when the interpreter has one. Refs #208 --- docs/usage.md | 2 +- .../command_injection/command_injection.py | 132 ++++++++++++++---- tests/test_command_arguments.py | 28 +++- 3 files changed, 127 insertions(+), 35 deletions(-) diff --git a/docs/usage.md b/docs/usage.md index 43fa513..9b96275 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -115,7 +115,7 @@ variable is not treated that way yet. | Rule | Reached sink | |---|---| -| `command-injection` | Shell or process execution (`os.system`, `subprocess` with a string, …), judged on what the input controls in the command. A string a shell runs, a string naming the program, the program itself, or an operand the program runs as a command (after the host for `ssh`, the program of `sudo`, `env`, `xargs`, `timeout`…, any argument of a batch file) is `high`. The script an interpreter runs (`bash run.sh`, `python tool.py`, its first operand when no option such as `-c` gives the code), or the module of `python -m`, is `high` and kept for review: execution is possible, and its exploitability depends on the file the input chooses. Otherwise an element of an argument list run without a shell can only be read as an option: `medium`, or `high` when a documented option of the program runs a command it chooses (`git push --receive-pack=…`, `ssh -o ProxyCommand=…`). A value a documented option takes is judged by what the option does with it, whatever its prefix: run it (`sh -c`, `cmd /c`, `python -c`: `high`), keep it as data (`git commit -m`: refuted). After an option the rule does not know, the finding stays a hotspot to review. A constant prefix that keeps a standalone input from being an option (`f"v{version}"`), or an operand after `--` for a program that honours it, is refuted. | +| `command-injection` | Shell or process execution (`os.system`, `subprocess` with a string, …), judged on what the input controls in the command. A string a shell runs, a string naming the program, the program itself, or an operand the program runs as a command (after the host for `ssh`, the program of `sudo`, `env`, `xargs`, `timeout`…, any argument of a batch file) is `high`. The script an interpreter runs (`bash run.sh`, `python tool.py`, its first operand when no option such as `-c` gives the code), or the module of `python -m`, is `high` and kept for review: execution is possible, and its exploitability depends on the file the input chooses. The arguments after a script or code are the script's own: options of unknown meaning, reported `medium` (a hotspot after an option of the script), refuted behind a constant prefix. Otherwise an element of an argument list run without a shell can only be read as an option: `medium`, or `high` when a documented option of the program runs a command it chooses (`git push --receive-pack=…`, `ssh -o ProxyCommand=…`). A value a documented option takes is judged by what the option does with it, whatever its prefix: run it (`sh -c`, `cmd /c`, `python -c`: `high`), keep it as data (`git commit -m`: refuted). After an option the rule does not know, the finding stays a hotspot to review. A constant prefix that keeps a standalone input from being an option (`f"v{version}"`), or an operand after `--` for a program that honours it, is refuted. | | `code-injection` | The code `eval` or `exec` runs; the globals and locals passed with it are data. `dangerous-eval` reports every such call, this rule the ones attacker input reaches. | | `sql-injection` | A database statement (`cursor.execute`, SQLAlchemy `text`, Django `raw`, …). Parameters of a parameterised query are not statements. | | `nosql-injection` | A NoSQL query, such as a MongoDB filter, given attacker input that may hold a structure: a JSON body giving `{"$ne": null}` matches any value. The input that may hold a structure is JSON, the raw bodies a `json.loads` decodes, files, WebSocket messages and standard input. Text cannot hold an operator: form fields, query parameters, headers, cookies, URL parameters, the command line, the environment, and an entry-point parameter annotated with scalars (`str`, `int`, `Optional[str]`, `Annotated[str, Query()]`, `list[str]`, …), which FastAPI validates. A Django, REST framework, aiohttp or tornado view reads a request object: its text attributes (`GET`, `POST`, `headers`, `query_params`, `query`, `match_info`, `arguments`, …) and the URL parameters after it are text, its body, files and `data` are not. Limits: a helper function the request is passed to reads it as a whole; the keys of a dict are not followed, so a string concatenated into `$where` is not reported. The analyzer ships no NoSQL sink: model plugins declare them. | diff --git a/src/coretrace_python/bundled/security/command_injection/command_injection.py b/src/coretrace_python/bundled/security/command_injection/command_injection.py index ba75eff..25ea4ed 100644 --- a/src/coretrace_python/bundled/security/command_injection/command_injection.py +++ b/src/coretrace_python/bundled/security/command_injection/command_injection.py @@ -8,7 +8,9 @@ as ``sudo`` runs, what follows ``find -exec``) are read again as that command. The script an interpreter runs (its first operand when no option gives the code, as ``bash run.sh`` or ``python tool.py``), or the module ``python -m`` runs, is ``high`` to review: execution is -possible, and whether it is exploitable depends on the file the input chooses. An element +possible, and whether it is exploitable depends on the file the input chooses. The +arguments that follow a script or code are the script's: its own options, whose meaning is +unknown, kept for review after an option and refuted behind a constant prefix. An element of an argument list run without a shell otherwise injects at most an option: ``medium``, or ``high`` when ``OPTIONS`` establishes that the program runs a command chosen through an option that fits there. @@ -166,15 +168,28 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: ), CommandOption("python", "-c", "runs its value as Python code"), CommandOption("python", "-m", "runs the module it names", examine=True), - *(CommandOption("python", option, single=False) for option in ("-W", "-X")), + CommandOption("python", "-W", "imports the module its warning category names", examine=True), + CommandOption("python", "-X", single=False), *_flags("python", *(f"-{letter}" for letter in "bBdEhiIOPqsSuvVx")), *(CommandOption("perl", option, "runs its value as Perl code") for option in ("-e", "-E")), + *( + CommandOption("perl", option, "runs the import of its value as Perl code") + for option in ("-M", "-m") + ), CommandOption("ruby", "-e", "runs its value as Ruby code"), + CommandOption("ruby", "-r", "requires the library it names", examine=True), + *( + CommandOption("node", option, "loads the module it names", examine=True) + for option in ("-r", "--require", "--import") + ), *( CommandOption("node", option, "runs its value as JavaScript code") for option in ("-e", "--eval", "-p", "--print") ), - CommandOption("php", "-r", "runs its value as PHP code"), + *( + CommandOption("php", option, "runs its value as PHP code") + for option in ("-r", "-B", "-R", "-E") + ), *( CommandOption( "cmd", option, "runs the rest of the command line as a command", single=False, rest=True @@ -210,9 +225,9 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: # Programs documented to read ``--`` as the end of their options. END_OF_OPTIONS = frozenset({"git", "tar", "rsync", "rm"}) # Interpreters run the script their first operand names, unless one of these options -# gives the code instead. +# gives the code instead (or, for a shell's ``-s``, reads it from the standard input). _SCRIPTS: Mapping[str, tuple[str, ...]] = { - **{shell: ("-c",) for shell in _SHELLS}, + **{shell: ("-c", "-s") for shell in _SHELLS}, "python": ("-c", "-m"), "perl": ("-e", "-E"), "ruby": ("-e",), @@ -268,7 +283,9 @@ def _interpreter(name: str, code: tuple[str, ...]) -> tuple[CommandOperands, ... _COMMAND_ARGUMENT = frozenset({(0, None), (None, "args")}) _EXECUTABLE, _SHELL_POSITION = 2, 8 _SHELL, _NO_SHELL = frozenset({"True", "1"}), frozenset({"False", "None", "0"}) -_PYTHON = re.compile(r"python[0-9.]*") +# Versioned and distribution names of interpreters: python3.12, pythonw, pypy3, nodejs, +# perl5.36, ruby3.2, php8.2. +_INTERPRETER = re.compile(r"(python|pypy|perl|ruby|php|node)(?:js)?[0-9.]*w?") _PYTHON_EXECUTABLE = SymbolId("python.sys.executable") @@ -505,13 +522,18 @@ def operand_of(self, spec: CommandOperands, index: int) -> _Reading | None: command=not script, to_review=script, ) - for names, wanted in ((spec.after, True), (spec.unless, False)): - if names: - given = self.given(names, index) - if given is None: - return unclear - if given is not wanted: - return None + if spec.after: + given = self.given(spec.after, index) + if given is None: + return unclear + if given is False: + return None + if spec.unless: + given = self.given(spec.unless, index) + if given is None: + return unclear + if given is not False: + return self.argument(given + 1, index) if index > given else None operands = self.operands(index, spec.first + 1, spec.assignments) if operands is None: return unclear @@ -524,17 +546,50 @@ def operand_of(self, spec: CommandOperands, index: int) -> _Reading | None: return self.inside(operands.positions[spec.first], None, index) if spec.kind == "text" and not spec.after: return _Reading(f"the input {spec.how}", Severity.HIGH, command=True) - return None + return self.argument(operands.positions[spec.first] + 1, index) if count == spec.first and operand_like: - if script: - return _Reading( - f"the input {spec.how}, whose contents decide what runs", - Severity.HIGH, - to_review=True, - ) - return _Reading(f"the input {spec.how}", Severity.HIGH, command=True) + if not script: + return _Reading(f"the input {spec.how}", Severity.HIGH, command=True) + if not operands.ended and self.prefix(index) == "": + # A whole element there may as well be a code option and its value. + free = self.free(index, "the input is a whole element") + if free.severity is Severity.HIGH: + return free + return _Reading( + f"the input {spec.how}, whose contents decide what runs", + Severity.HIGH, + to_review=True, + ) return None + def argument(self, start: int, index: int) -> _Reading: + """An argument of a script or of code given by an option, the arguments starting at + ``start``, which the engine does not read: an option of it, whose meaning is + unknown, unless a constant prefix keeps it from being one.""" + + before = self.text(index - 1) if index > start else None + if ( + before is not None + and before.startswith("-") + and before not in ("-", "--") + and "=" not in before + ): + return _Reading( + f"the input follows {before}, an option of the script it may be the value of", + Severity.MEDIUM, + to_review=True, + ) + prefix = self.prefix(index) + if self.parts[index].known and prefix and not prefix.startswith("-"): + return _Reading( + f"the input is an argument of the script, behind the constant prefix {prefix!r}", + None, + ) + return _Reading( + "the input is an argument of the script, which it may read as an option", + Severity.MEDIUM, + ) + def inside(self, start: int, end: int | None, index: int) -> _Reading: """The reading of the element at ``index`` in the command the elements from ``start`` (to ``end``) spell, which the program runs.""" @@ -582,11 +637,12 @@ def operands(self, index: int, enough: int, assignments: bool = False) -> _Opera position += 1 return _Operands(tuple(positions)) - def given(self, names: tuple[str, ...], index: int) -> bool | None: - """Whether one of the options ``names`` comes among the options before ``index``, - read as getopt reads them (a short option that takes a value takes the rest of its - element); None when an element of unknown value, or an option that may take one, - comes first.""" + def given(self, names: tuple[str, ...], index: int) -> int | bool | None: + """Where one of the options ``names`` ends among the options before ``index``: the + position of the element holding its value, or of the option itself when it takes + none; False when none comes, None when an element of unknown value, or an option + that may take one, comes first. Options are read as getopt reads them: a short + option that takes a value takes the rest of its element.""" position = 1 while position < index: @@ -598,15 +654,25 @@ def given(self, names: tuple[str, ...], index: int) -> bool | None: if text[0] in "-+" and not text.startswith("--") and len(text) > 2: for offset, letter in enumerate(text[1:], start=1): option = self.option(text[0] + letter, position + 1) + last = offset == len(text) - 1 if text[0] + letter in names: - return True + return ( + position + 1 + if last and option is not None and option.takes_value + else position + ) if option is None: return None if option.takes_value: - position += offset == len(text) - 1 + position += last break elif text.split("=", 1)[0] in names: - return True + option = self.option(text, position + 1) + return ( + position + 1 + if "=" not in text and option is not None and option.takes_value + else position + ) elif not (text.startswith("--") and "=" in text): taker = self.taker(text, position + 1) if taker is None: @@ -740,10 +806,14 @@ def subcommand(self, index: int) -> str | None: def _name(program: str) -> str: """The name a program is known by in ``OPTIONS``: without its directory (POSIX or - Windows) or ``.exe``, in lower case, any ``python3.x`` as ``python``.""" + Windows) or ``.exe``, in lower case, an interpreter without its version (``pypy`` and + the Windows launcher ``py`` as ``python``).""" name = re.split(r"[\\/]", program)[-1].lower().removesuffix(".exe") - return "python" if _PYTHON.fullmatch(name) else name + interpreter = _INTERPRETER.fullmatch(name) + if interpreter is not None: + name = interpreter.group(1) + return "python" if name in ("pypy", "py") else name def _first_program( diff --git a/tests/test_command_arguments.py b/tests/test_command_arguments.py index cc82218..6d935ed 100644 --- a/tests/test_command_arguments.py +++ b/tests/test_command_arguments.py @@ -177,15 +177,18 @@ def judged(call: str) -> list[tuple[str, str, str]]: ('subprocess.run([shutil.which("ssh"), "host", f"cat {x}"])', HIGH), ('subprocess.run([shutil.which("bash"), "-c", f"echo {x}"])', HIGH), # An interpreter runs the script its first operand names, unless code is given: - # execution is possible, exploitability depends on the file, so high to review. + # execution is possible, exploitability depends on the file, so high to review. An + # input that starts the element may as well be a code option (-e#.pl): high. ('subprocess.run(["bash", f"/tmp/{x}.sh"])', HIGH_TO_REVIEW), ('subprocess.run(["sh", x])', HIGH_TO_REVIEW), ('subprocess.run(["bash", "-x", "--", f"run-{x}.sh"])', HIGH_TO_REVIEW), ('subprocess.run(["python3", f"scripts/{x}.py"])', HIGH_TO_REVIEW), ('subprocess.run([sys.executable, f"tools/{x}.py", "--flag"])', HIGH_TO_REVIEW), ('subprocess.run(["python3", "-m", f"pkg.{x}"])', HIGH_TO_REVIEW), - ('subprocess.run(["perl", f"{x}.pl"])', HIGH_TO_REVIEW), - ('subprocess.run(["node", f"{x}.js"])', HIGH_TO_REVIEW), + ('subprocess.run(["perl", f"lib/{x}.pl"])', HIGH_TO_REVIEW), + ('subprocess.run(["perl", f"{x}.pl"])', HIGH), + ('subprocess.run(["node", f"app/{x}.js"])', HIGH_TO_REVIEW), + ('subprocess.run(["node", f"{x}.js"])', HIGH), # Code given by an option runs; the operands after it are the script's arguments. ('subprocess.run(["perl", "-e", f"print {x}"])', HIGH), ('subprocess.run(["ruby", "-e", x])', HIGH), @@ -194,6 +197,25 @@ def judged(call: str) -> list[tuple[str, str, str]]: ('subprocess.run(["python3", "-c", "import sys; print(sys.argv[1])", f"v{x}"])', None), ('subprocess.run(["bash", "script.sh", f"v{x}"])', None), ('subprocess.run(["python3", "script.py", f"--name={x}"])', MEDIUM), + # Fourth review: options that choose code, arguments of a script, interpreter names. + ('subprocess.run(["python", "-W", x, "s.py"])', HIGH_TO_REVIEW), + ('subprocess.run(["python", f"-W{x}"])', HIGH_TO_REVIEW), + ('subprocess.run(["python", x])', HIGH), + ('subprocess.run(["node", x])', HIGH), + ('subprocess.run(["node", "app.js", x])', MEDIUM), + ('subprocess.run(["node", "app.js", "-e", x])', MEDIUM_TO_REVIEW), + ('subprocess.run(["perl", "x.pl", "-e", x])', MEDIUM_TO_REVIEW), + ('subprocess.run(["python3", "script.py", x])', MEDIUM), + ('subprocess.run(["python", "-m", "pkg", x])', MEDIUM), + ('subprocess.run(["node", "--eval=code", f"v{x}"])', None), + ('subprocess.run(["nodejs", f"s/{x}"])', HIGH_TO_REVIEW), + ('subprocess.run(["pypy3", f"tools/{x}.py"])', HIGH_TO_REVIEW), + ('subprocess.run(["perl5.36", f"lib/{x}.pl"])', HIGH_TO_REVIEW), + ('subprocess.run(["perl", f"-M{x}", "s.pl"])', HIGH), + ('subprocess.run(["ruby", "-r", x, "s.rb"])', HIGH_TO_REVIEW), + ('subprocess.run(["php", "-R", f"echo {x};"])', HIGH), + ('subprocess.run(["bash", "-s", f"v{x}"])', None), + ('subprocess.run(["bash", "-s", x])', MEDIUM), # Unpacked parts keep their place in the command. ('subprocess.run(["xcrun", "notarytool", *(["--keychain", x])])', MEDIUM_TO_REVIEW), ("subprocess.run([*extra, x])", HIGH), From 011adbbc3def5c6219b0683779755662c5d06bb7 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 01:08:53 +0200 Subject: [PATCH 7/8] fix(command-injection): read the options interpreters keep reading after code Review of the script arguments: perl, ruby, node and php read options after -e CODE until their first operand, so perl -e CODE -e x is Perl code, while the arguments reading took it for the script's; python and the shells read none after the code. A shell's lone - ends its options, and the operand after it is the script. The flags and value options of perl, ruby, node, php and PowerShell are modelled, so that a script argument behind them is no longer kept for review; php -f, -d, -c, -z and PowerShell -File choose code to examine; pwsh reads its first operand as a script; node short options take no glued value; pyw and -dbg names are recognised. Whole-word options are matched before being read as grouped short options. Refs #208 --- .../command_injection/command_injection.py | 126 +++++++++++++++--- tests/test_command_arguments.py | 22 +++ 2 files changed, 127 insertions(+), 21 deletions(-) diff --git a/src/coretrace_python/bundled/security/command_injection/command_injection.py b/src/coretrace_python/bundled/security/command_injection/command_injection.py index 25ea4ed..863cce3 100644 --- a/src/coretrace_python/bundled/security/command_injection/command_injection.py +++ b/src/coretrace_python/bundled/security/command_injection/command_injection.py @@ -170,7 +170,31 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: CommandOption("python", "-m", "runs the module it names", examine=True), CommandOption("python", "-W", "imports the module its warning category names", examine=True), CommandOption("python", "-X", single=False), - *_flags("python", *(f"-{letter}" for letter in "bBdEhiIOPqsSuvVx")), + *_flags("python", *(f"-{letter}" for letter in "23bBdEhiIOPqsSuvVx")), + *_flags("perl", *(f"-{letter}" for letter in "acnpsStTUwWX")), + CommandOption("perl", "-I", single=True), + *_flags("ruby", *(f"-{letter}" for letter in "acdlnpsvwWy")), + *(CommandOption("ruby", option) for option in ("-I", "-C", "-E")), + *_flags( + "node", + "-c", + "--check", + "-i", + "--interactive", + "--inspect", + "--inspect-brk", + "--no-warnings", + "--no-deprecation", + "--trace-warnings", + "--enable-source-maps", + "--expose-gc", + "--preserve-symlinks", + ), + *_flags("php", *(f"-{letter}" for letter in "aehHilmnqsvw")), + CommandOption("php", "-f", "runs the script it names", examine=True), + CommandOption("php", "-d", "sets any ini directive, auto_prepend_file included", examine=True), + CommandOption("php", "-c", "loads the php.ini it names", examine=True), + CommandOption("php", "-z", "loads the Zend extension it names", examine=True), *(CommandOption("perl", option, "runs its value as Perl code") for option in ("-e", "-E")), *( CommandOption("perl", option, "runs the import of its value as Perl code") @@ -178,12 +202,21 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: ), CommandOption("ruby", "-e", "runs its value as Ruby code"), CommandOption("ruby", "-r", "requires the library it names", examine=True), + # node reads the value of a short option from the next element only. *( - CommandOption("node", option, "loads the module it names", examine=True) + CommandOption( + "node", + option, + "loads the module it names", + examine=True, + single=option.startswith("--"), + ) for option in ("-r", "--require", "--import") ), *( - CommandOption("node", option, "runs its value as JavaScript code") + CommandOption( + "node", option, "runs its value as JavaScript code", single=option.startswith("--") + ) for option in ("-e", "--eval", "-p", "--print") ), *( @@ -207,6 +240,22 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: for powershell in ("powershell", "pwsh") for option in ("-command", "-c", "-encodedcommand", "-e", "-ec") ), + *( + option + for powershell in ("powershell", "pwsh") + for option in ( + CommandOption( + powershell, "-file", "runs the script it names", single=False, examine=True + ), + *_flags( + powershell, "-noprofile", "-noninteractive", "-nologo", "-noexit", "-sta", "-mta" + ), + *( + CommandOption(powershell, value, single=False) + for value in ("-executionpolicy", "-windowstyle", "-inputformat", "-outputformat") + ), + ) + ), *_flags( "rm", "-d", @@ -225,15 +274,21 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: # Programs documented to read ``--`` as the end of their options. END_OF_OPTIONS = frozenset({"git", "tar", "rsync", "rm"}) # Interpreters run the script their first operand names, unless one of these options -# gives the code instead (or, for a shell's ``-s``, reads it from the standard input). +# gives the code or the script instead (or, for a shell's ``-s``, reads it from the +# standard input). +_POWERSHELL_CODE = ("-command", "-c", "-encodedcommand", "-e", "-ec", "-file") _SCRIPTS: Mapping[str, tuple[str, ...]] = { **{shell: ("-c", "-s") for shell in _SHELLS}, "python": ("-c", "-m"), "perl": ("-e", "-E"), "ruby": ("-e",), "node": ("-e", "--eval", "-p", "--print"), - "php": ("-r",), + "php": ("-r", "-f", "-B", "-R", "-E"), + "pwsh": _POWERSHELL_CODE, } +# These read their options after the code too, up to their first operand: what follows +# ``perl -e CODE`` may be another ``-e``. Python and the shells read none after it. +_OPTIONS_AFTER_CODE = frozenset({"perl", "ruby", "node", "php"}) def _interpreter(name: str, code: tuple[str, ...]) -> tuple[CommandOperands, ...]: @@ -248,7 +303,9 @@ def _interpreter(name: str, code: tuple[str, ...]) -> tuple[CommandOperands, ... OPERANDS: Mapping[str, tuple[CommandOperands, ...]] = { "ssh": (CommandOperands(1, "text", "is part of the command the remote shell runs"),), - "powershell": (CommandOperands(0, "text", "is part of the command PowerShell runs"),), + "powershell": ( + CommandOperands(0, "text", "is part of the command PowerShell runs", unless=("-file",)), + ), "timeout": (CommandOperands(1, "command", "names the program timeout runs"),), **{ wrapper: ( @@ -283,9 +340,9 @@ def _interpreter(name: str, code: tuple[str, ...]) -> tuple[CommandOperands, ... _COMMAND_ARGUMENT = frozenset({(0, None), (None, "args")}) _EXECUTABLE, _SHELL_POSITION = 2, 8 _SHELL, _NO_SHELL = frozenset({"True", "1"}), frozenset({"False", "None", "0"}) -# Versioned and distribution names of interpreters: python3.12, pythonw, pypy3, nodejs, -# perl5.36, ruby3.2, php8.2. -_INTERPRETER = re.compile(r"(python|pypy|perl|ruby|php|node)(?:js)?[0-9.]*w?") +# Versioned and distribution names of interpreters: python3.12, pythonw, python3-dbg, +# pypy3, the Windows launchers py and pyw, nodejs, perl5.36, ruby3.2, php8.2. +_INTERPRETER = re.compile(r"(python|pypy|perl|ruby|php|node|py)(?:js)?[0-9.]*w?(?:-dbg)?") _PYTHON_EXECUTABLE = SymbolId("python.sys.executable") @@ -533,7 +590,7 @@ def operand_of(self, spec: CommandOperands, index: int) -> _Reading | None: if given is None: return unclear if given is not False: - return self.argument(given + 1, index) if index > given else None + return self.after_code(given, index) operands = self.operands(index, spec.first + 1, spec.assignments) if operands is None: return unclear @@ -621,7 +678,7 @@ def operands(self, index: int, enough: int, assignments: bool = False) -> _Opera text = self.text(position) if text is None: return None - if text == "--": + if text == "--" or (text == "-" and self.program in _SHELLS): return _Operands((*positions, *range(position + 1, index))[:enough], ended=True) if self.option_like(text) and text != "-": if not (text.startswith("--") and "=" in text): @@ -651,11 +708,20 @@ def given(self, names: tuple[str, ...], index: int) -> int | bool | None: return None if text == "--" or text in ("-", "+") or not self.option_like(text): return False - if text[0] in "-+" and not text.startswith("--") and len(text) > 2: + whole = text.split("=", 1)[0] + if self.named(whole, names): + option = self.option(text, position + 1) + return ( + position + 1 + if "=" not in text and option is not None and option.takes_value + else position + ) + grouped = text[0] in "-+" and not text.startswith("--") and len(text) > 2 + if grouped and self.option(whole, position + 1) is None: for offset, letter in enumerate(text[1:], start=1): option = self.option(text[0] + letter, position + 1) last = offset == len(text) - 1 - if text[0] + letter in names: + if self.named(text[0] + letter, names): return ( position + 1 if last and option is not None and option.takes_value @@ -666,13 +732,6 @@ def given(self, names: tuple[str, ...], index: int) -> int | bool | None: if option.takes_value: position += last break - elif text.split("=", 1)[0] in names: - option = self.option(text, position + 1) - return ( - position + 1 - if "=" not in text and option is not None and option.takes_value - else position - ) elif not (text.startswith("--") and "=" in text): taker = self.taker(text, position + 1) if taker is None: @@ -681,6 +740,31 @@ def given(self, names: tuple[str, ...], index: int) -> int | bool | None: position += 1 return False + def after_code(self, given: int, index: int) -> _Reading | None: + """The reading of the element at ``index``, once the option ending at ``given`` gave + the code: an argument of it, except that the interpreters of + ``_OPTIONS_AFTER_CODE`` read options up to their first operand, which the usual + reading of options judges (None).""" + + if index <= given: + return None + if self.program not in _OPTIONS_AFTER_CODE: + return self.argument(given + 1, index) + operands = self.operands(index, 1) + if operands is None or operands.consumed: + return None + if operands.positions: + return self.argument(operands.positions[0], index) + prefix = self.prefix(index) + if operands.ended or (prefix and not prefix.startswith("-")): + return self.argument(index, index) + if prefix.startswith("-"): + return None + return self.free(index, "the input is a whole element where options are still read") + + def named(self, written: str, names: tuple[str, ...]) -> bool: + return any(self.same(written, name) for name in names) + def review(self, reason: str, free: _Reading) -> _Reading: return _Reading(reason, free.severity, to_review=True) @@ -707,7 +791,7 @@ def free(self, index: int, how: str) -> _Reading: option for option in self.applicable(index) if option.runs is not None and option.single ] if dangerous: - option = dangerous[0] + option = min(dangerous, key=lambda found: found.examine) return _Reading( f"{how}, and an option such as {self.program} {_spelled(option.option)} {option.runs}", Severity.HIGH, diff --git a/tests/test_command_arguments.py b/tests/test_command_arguments.py index 6d935ed..f5f6a62 100644 --- a/tests/test_command_arguments.py +++ b/tests/test_command_arguments.py @@ -216,6 +216,28 @@ def judged(call: str) -> list[tuple[str, str, str]]: ('subprocess.run(["php", "-R", f"echo {x};"])', HIGH), ('subprocess.run(["bash", "-s", f"v{x}"])', None), ('subprocess.run(["bash", "-s", x])', MEDIUM), + # Fifth review: perl, ruby, node and php read options after the code, until an + # operand; python and the shells do not. A shell's lone - ends its options. + ('subprocess.run(["perl", "-e", "CODE", "-e", f"c{x}"])', HIGH), + ('subprocess.run(["perl", "-e", "CODE", x])', HIGH), + ('subprocess.run(["ruby", "-e", "CODE", x])', HIGH), + ('subprocess.run(["node", "-e", "CODE", f"--eval={x}"])', HIGH), + ('subprocess.run(["perl", "-e", "CODE", f"v{x}"])', None), + ('subprocess.run(["perl", "-e", "CODE", "arg", f"v{x}"])', None), + ('subprocess.run(["python", "-c", "CODE", "-c", x])', MEDIUM_TO_REVIEW), + ('subprocess.run(["bash", "-", f"s/{x}.sh"])', HIGH_TO_REVIEW), + ('subprocess.run(["sh", "-", x])', HIGH_TO_REVIEW), + ('subprocess.run(["perl", "-w", "s.pl", f"v{x}"])', None), + ('subprocess.run(["perl", "-I", "lib", "s.pl", f"v{x}"])', None), + ('subprocess.run(["ruby", "-w", "s.rb", f"v{x}"])', None), + ('subprocess.run(["node", "--inspect", "app.js", f"v{x}"])', None), + ('subprocess.run(["php", "-f", "s.php", f"v{x}"])', None), + ('subprocess.run(["php", "-f", f"s/{x}.php"])', HIGH_TO_REVIEW), + ('subprocess.run(["pwsh", f"s/{x}.ps1"])', HIGH_TO_REVIEW), + ('subprocess.run(["pwsh", "-File", f"s/{x}.ps1"])', HIGH_TO_REVIEW), + ('subprocess.run(["powershell", "-File", "s.ps1", f"v{x}"])', None), + ('subprocess.run(["pyw", f"s/{x}.py"])', HIGH_TO_REVIEW), + ('subprocess.run(["python3-dbg", f"s/{x}.py"])', HIGH_TO_REVIEW), # Unpacked parts keep their place in the command. ('subprocess.run(["xcrun", "notarytool", *(["--keychain", x])])', MEDIUM_TO_REVIEW), ("subprocess.run([*extra, x])", HIGH), From 9106b176a0ab88f1ce0c7852967896d88d93e3e2 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 01:34:01 +0200 Subject: [PATCH 8/8] fix(command-injection): keep pwsh -Command elements as the command Review of the interpreter options: pwsh -Command, -c and -EncodedCommand were read as giving code followed by script arguments, while every element after them is the command, so pwsh -Command Get-Item v$x was refuted. A code option that takes the rest of the command line now leaves its elements to that reading. node -pe, an alias of --print --eval, takes its code from the next element. Refs #208 --- .../security/command_injection/command_injection.py | 9 +++++++-- tests/test_command_arguments.py | 8 ++++++++ 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/src/coretrace_python/bundled/security/command_injection/command_injection.py b/src/coretrace_python/bundled/security/command_injection/command_injection.py index 863cce3..56802e6 100644 --- a/src/coretrace_python/bundled/security/command_injection/command_injection.py +++ b/src/coretrace_python/bundled/security/command_injection/command_injection.py @@ -217,7 +217,7 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: CommandOption( "node", option, "runs its value as JavaScript code", single=option.startswith("--") ) - for option in ("-e", "--eval", "-p", "--print") + for option in ("-e", "--eval", "-p", "--print", "-pe") ), *( CommandOption("php", option, "runs its value as PHP code") @@ -282,7 +282,7 @@ def _flags(program: str, *options: str) -> tuple[CommandOption, ...]: "python": ("-c", "-m"), "perl": ("-e", "-E"), "ruby": ("-e",), - "node": ("-e", "--eval", "-p", "--print"), + "node": ("-e", "--eval", "-p", "--print", "-pe"), "php": ("-r", "-f", "-B", "-R", "-E"), "pwsh": _POWERSHELL_CODE, } @@ -748,6 +748,11 @@ def after_code(self, given: int, index: int) -> _Reading | None: if index <= given: return None + for position in range(1, given + 1): + option = self.option(self.text(position) or "", position) + if option is not None and option.rest: + # Every element after it is the command (``pwsh -Command``). + return None if self.program not in _OPTIONS_AFTER_CODE: return self.argument(given + 1, index) operands = self.operands(index, 1) diff --git a/tests/test_command_arguments.py b/tests/test_command_arguments.py index f5f6a62..ca492d4 100644 --- a/tests/test_command_arguments.py +++ b/tests/test_command_arguments.py @@ -238,6 +238,14 @@ def judged(call: str) -> list[tuple[str, str, str]]: ('subprocess.run(["powershell", "-File", "s.ps1", f"v{x}"])', None), ('subprocess.run(["pyw", f"s/{x}.py"])', HIGH_TO_REVIEW), ('subprocess.run(["python3-dbg", f"s/{x}.py"])', HIGH_TO_REVIEW), + # Sixth review: every element after pwsh -Command is the command; node -pe. + ('subprocess.run(["pwsh", "-Command", "Get-Item", f"v{x}"])', HIGH), + ('subprocess.run(["pwsh", "-NoProfile", "-Command", "Get-Item", f"v{x}"])', HIGH), + ('subprocess.run(["pwsh", "-ec", "AAA", f"v{x}"])', HIGH), + ('subprocess.run(["pwsh", "-c", "echo", x])', HIGH), + ('subprocess.run(["sudo", "pwsh", "-c", "Get-Item", x])', HIGH), + ('subprocess.run(["node", "-pe", f"console.log({x})"])', HIGH), + ('subprocess.run(["node", "-pe", "CODE", x])', HIGH), # Unpacked parts keep their place in the command. ('subprocess.run(["xcrun", "notarytool", *(["--keychain", x])])', MEDIUM_TO_REVIEW), ("subprocess.run([*extra, x])", HIGH),