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..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, …). | +| `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 f43942b..56802e6 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,371 @@ -"""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, +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. 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. 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. + +A value an option takes is judged by what ``OPTIONS`` establishes the option does with +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 +import re +from collections.abc import Mapping +from dataclasses import dataclass 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 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 +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, and ``single`` says one element can carry the option + 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 (``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 + option: str + runs: str | None = None + subcommands: tuple[str, ...] = () + single: bool = True + takes_value: bool = True + 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, 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 + kind: str + how: str + after: tuple[str, ...] = () + unless: tuple[str, ...] = () + assignments: 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") +_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. +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")), + ("-u", ("clone", "ls-remote")), + ("--receive-pack", ("push",)), + ("--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")), + 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"), + *( + CommandOption("find", option, "runs the command that follows", single=False, opens=True) + for option in ("-exec", "-execdir", "-ok", "-okdir") + ), + *(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 + 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", "-W", "imports the module its warning category names", examine=True), + CommandOption("python", "-X", single=False), + *_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") + for option in ("-M", "-m") + ), + 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, + single=option.startswith("--"), + ) + for option in ("-r", "--require", "--import") + ), + *( + CommandOption( + "node", option, "runs its value as JavaScript code", single=option.startswith("--") + ) + for option in ("-e", "--eval", "-p", "--print", "-pe") + ), + *( + 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 + ) + for option in ("/c", "/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", "-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", + "-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", "tar", "rsync", "rm"}) +# Interpreters run the script their first operand names, unless one of these options +# 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", "-pe"), + "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, ...]: + 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", unless=("-file",)), + ), + "timeout": (CommandOperands(1, "command", "names the program timeout runs"),), + **{ + 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. +_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}") + 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"}) +# 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") + + +@dataclass(frozen=True) +class _Reading: + """What one tainted element of a command lets the attacker do: run a ``command`` + (``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 + + +@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): @@ -15,3 +374,607 @@ 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 + arguments = flow.sink_arguments + if arguments is None or not flow.passed_as or not flow.passed_as <= _COMMAND_ARGUMENT: + return default + shell = _shell(arguments) + if shell is True: + return default + if shell is None: + 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 + } + parts = command_parts(flow.argument, defs, ctx.get(DefUseAnalysis, function)) + if not parts: + return default + taint = ctx.get(TaintAnalysis, function) + command = _Command( + parts, defs, ctx.get(ModuleStringsAnalysis), _program(arguments, parts[0], defs) + ) + 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: + """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], + 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] + 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.""" + + 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] + 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. 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) + if self.batch: + 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 + 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: + 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, + ) + if index >= 2: + before = self.text(index - 1) + if before is None: + free = self.free(index, "the input is a whole element") + return self.review( + f"cannot tell what the element before the input is; if not an option, {free.reason}", + free, + ) + 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] != "-": + 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 + ) + 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( + index, "the input follows '--', which the program is not known to honour" + ) + 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.""" + + 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=not script, + to_review=script, + ) + 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.after_code(given, index) + operands = self.operands(index, spec.first + 1, spec.assignments) + 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 spec.kind == "command": + 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 self.argument(operands.positions[spec.first] + 1, index) + if count == spec.first and operand_like: + 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.""" + + if index == start: + 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}", + reading.severity, + reading.command, + reading.to_review, + ) + + 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 + while position < index and len(positions) < enough: + text = self.text(position) + if text is None: + return None + 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): + 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) + elif not (assignments and _ASSIGNMENT.match(text)): + positions.append(position) + position += 1 + return _Operands(tuple(positions)) + + 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: + text = self.text(position) + if text is None: + return None + if text == "--" or text in ("-", "+") or not self.option_like(text): + return False + 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 self.named(text[0] + letter, names): + 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 += last + break + elif not (text.startswith("--") and "=" in text): + taker = self.taker(text, position + 1) + if taker is None: + return None + position += isinstance(taker, CommandOption) + 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 + 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) + 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) + + 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: + """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 + ) + + 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 = 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, + ) + 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 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}", + 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 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 + 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"{sign}{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 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. 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 [] + subcommand = self.subcommand(index) + 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 + 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 _name(program: str) -> str: + """The name a program is known by in ``OPTIONS``: without its directory (POSIX or + 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") + 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( + part: CommandPart, defs: Mapping[Value, Instruction], strings: Mapping[str, str] +) -> str | None: + """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 + text, whole = leading_text(part.value, defs, strings) + if whole: + return text + made = defs.get(part.value) + 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( + 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 + decide.""" + + given = arguments.given("shell", _SHELL_POSITION) + 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..ca492d4 --- /dev/null +++ b/tests/test_command_arguments.py @@ -0,0 +1,278 @@ +"""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 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. +""" + +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 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) + 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), + # 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), + # 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), + # 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), + # 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. 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"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), + ('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), + # 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), + # 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), + # 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), + ('subprocess.run(["ls", *extra, x])', MEDIUM_TO_REVIEW), + ('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,) + + +def test_a_deeply_nested_command_is_still_read() -> None: + wrappers = '"sudo", ' * 500 + assert judged(f"subprocess.run([{wrappers}x])") == [HIGH] 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",)