diff --git a/docs/plugins.md b/docs/plugins.md index ebb442b..b120460 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -130,10 +130,15 @@ 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` +verdict lowers the confidence of the finding, never its severity. An `Assessment` may +also carry a `title` for the finding's message instead of the rule's, and `metadata` +the finding carries besides the flow's; the rule id never changes, so suppressions and +baselines keep matching. `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. +It sets the `injection` metadata (`command` or `option`, the latter titled `Option +injection`) only when it establishes what the input is. ### Security models: `ModelPlugin` diff --git a/docs/usage.md b/docs/usage.md index 6aca604..ec318ad 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. 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. | +| `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. Every case keeps the rule id `command-injection`, so suppressions and baselines match whatever the case. When the rule reads the command element by element and establishes what the input is there, the finding's `injection` metadata (in the JSON report) says so: `command` when the input runs or chooses what runs (a shell string, the program, an operand run as a command, the value of an option that runs it); `option` when it can only reach the program as an option or as the value of an option not established to run it, and the message is then titled `Option injection`. When that cannot be established (an option whose effect is to review, an unknown program or element before the input, a command that is not a list or tuple display, a sink such as `os.system` the rule does not read element by element), the metadata is absent and the title stays `Command injection`. | | `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 4d5f6c4..fcf9570 100644 --- a/src/coretrace_python/bundled/security/command_injection/command_injection.py +++ b/src/coretrace_python/bundled/security/command_injection/command_injection.py @@ -31,7 +31,7 @@ import re from collections.abc import Mapping -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import ClassVar from coretrace_python.abstract.strings import ModuleStringsAnalysis, leading_text @@ -362,16 +362,28 @@ def _interpreter(name: str, code: tuple[str, ...]) -> tuple[CommandOperands, ... _PYTHON_EXECUTABLE = SymbolId("python.sys.executable") +# What the attacker's input is in the command, when the rule establishes it (#220). +_COMMAND, _OPTION = "command", "option" +_OPTION_TITLE = "Option injection" + + @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.""" + with ``severity`` None. ``injection`` says what the input is there, when established: + ``command`` (it runs, or chooses what runs) or ``option`` (an option, or the value of + one the model does not establish to run it).""" reason: str severity: Severity | None command: bool = False to_review: bool = False + injection: str | None = None + + @property + def kind(self) -> str | None: + return _COMMAND if self.command else self.injection @dataclass(frozen=True) @@ -407,7 +419,7 @@ def assess( return default shell = _shell(arguments) if shell is True: - return default + return _of_kind(default, _COMMAND) if shell is None: reason = "an expression chooses whether a shell runs the command" return Assessment(_to_review(default.verdict, reason), Severity.HIGH) @@ -428,8 +440,10 @@ def assess( 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): + if not readings: return default + if any(reading.command for reading in readings): + return _of_kind(default, _COMMAND) injected = [reading for reading in readings if reading.severity is not None] if not injected: return Assessment( @@ -441,9 +455,11 @@ def assess( ) decisive = [r for r in injected if r.severity is severity] established = [reading for reading in decisive if not reading.to_review] + kinds = {reading.kind for reading in established or decisive} + kind = kinds.pop() if len(kinds) == 1 else None if not established: - return Assessment(_to_review(default.verdict, decisive[0].reason), severity) - return Assessment(_explained(default.verdict, established[0].reason), severity) + return _of_kind(Assessment(_to_review(default.verdict, decisive[0].reason), severity), kind) + return _of_kind(Assessment(_explained(default.verdict, established[0].reason), severity), kind) class _Command: @@ -510,7 +526,7 @@ def read(self, index: int) -> _Reading: 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 + f"the program is unknown and may run its operands; {free.reason}", free, None ) for position in range(1, index): option = self.option(self.text(position) or "", position) @@ -529,6 +545,7 @@ def read(self, index: int) -> _Reading: return _Reading( f"the input follows {self.program} {option.option}, which {option.runs}", Severity.HIGH, + injection=_COMMAND, ) if index >= 2: before = self.text(index - 1) @@ -537,6 +554,7 @@ def read(self, index: int) -> _Reading: return self.review( f"cannot tell what the element before the input is; if not an option, {free.reason}", free, + None, ) if before not in ("-", "--") and self.option_like(before): taker = self.taker(before, index) @@ -546,9 +564,11 @@ def read(self, index: int) -> _Reading: return self.value_of(taker, before) if taker is None: free = self.free(index, "the input is a whole element") + # Its value or an option of its own: an option either way. return self.review( f"cannot tell whether {before} takes the input as its value; if not, {free.reason}", free, + _OPTION, ) if not self.parts[index].known: return self.free(index, "the input gives whole elements") @@ -558,6 +578,7 @@ def read(self, index: int) -> _Reading: return _Reading( f"the input follows {self.program} {option.option}, which {option.runs}", Severity.HIGH, + injection=_COMMAND, ) if prefix.startswith("--") and "=" in prefix: return self.option_value(index, prefix[: prefix.index("=")]) @@ -651,6 +672,7 @@ def argument(self, start: int, index: int) -> _Reading: f"the input follows {before}, an option of the script it may be the value of", Severity.MEDIUM, to_review=True, + injection=_OPTION, ) prefix = self.prefix(index) if self.parts[index].known and prefix and not prefix.startswith("-"): @@ -661,6 +683,7 @@ def argument(self, start: int, index: int) -> _Reading: return _Reading( "the input is an argument of the script, which it may read as an option", Severity.MEDIUM, + injection=_OPTION, ) def inside(self, start: int, end: int | None, index: int) -> _Reading: @@ -681,6 +704,7 @@ def inside(self, start: int, end: int | None, index: int) -> _Reading: reading.severity, reading.command, reading.to_review, + reading.injection, ) def operands(self, index: int, enough: int, assignments: bool = False) -> _Operands | None: @@ -789,8 +813,8 @@ def after_code(self, given: int, index: int) -> _Reading | None: 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 review(self, reason: str, free: _Reading, injection: str | None) -> _Reading: + return _Reading(reason, free.severity, to_review=True, injection=injection) def same(self, written: str, option: str) -> bool: return written.lower() == option.lower() if self.program in _ANY_CASE else written == option @@ -819,15 +843,18 @@ def free(self, index: int, how: str) -> _Reading: return _Reading( f"{how}, and an option such as {self.program} {_spelled(option.option)} {option.runs}", Severity.HIGH, + injection=_OPTION, ) - return _Reading(f"{how}, so it may be read as an option", Severity.MEDIUM) + return _Reading(f"{how}, so it may be read as an option", Severity.MEDIUM, injection=_OPTION) 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) + return _Reading( + f"the input is the value of the option {name}", Severity.MEDIUM, injection=_OPTION + ) if not option.takes_value: return self.free( index, f"the input follows {name}, which takes no value, as more options" @@ -845,12 +872,15 @@ def value_of(self, option: CommandOption, given: str) -> _Reading: return _Reading( f"the input is the value of {self.program} {given}, which {option.runs}", Severity.HIGH, + injection=_COMMAND, ) 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) + return _Reading( + f"the input is the value of {self.program} {given}", Severity.MEDIUM, injection=_OPTION + ) def taker(self, before: str, index: int) -> CommandOption | bool | None: """The option ``before`` whose value is the element at ``index``; False when the @@ -986,6 +1016,19 @@ def _shell(arguments: Arguments) -> bool | None: return False if value in _NO_SHELL else None +def _of_kind(assessment: Assessment, kind: str | None) -> Assessment: + """``assessment`` saying what kind of injection it is, when established: an option + injection is titled so; the rule id stays ``command-injection`` either way.""" + + if kind is None: + return assessment + return replace( + assessment, + title=_OPTION_TITLE if kind == _OPTION else None, + metadata={**assessment.metadata, "injection": kind}, + ) + + def _explained(verdict: Verdict, reason: str) -> Verdict: """``verdict`` with what the command lets the attacker do added to its evidence.""" diff --git a/src/coretrace_python/plugins/detectors.py b/src/coretrace_python/plugins/detectors.py index c39534d..557c592 100644 --- a/src/coretrace_python/plugins/detectors.py +++ b/src/coretrace_python/plugins/detectors.py @@ -10,7 +10,7 @@ from __future__ import annotations from collections.abc import Mapping, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import ClassVar from coretrace_python.analysis import AnyAnalysis @@ -27,10 +27,14 @@ @dataclass(frozen=True) class Assessment: - """A rule's verdict on one flow and the severity it reports the flow at.""" + """A rule's verdict on one flow and the severity it reports the flow at; a ``title`` + for the finding's message instead of the rule's, and ``metadata`` the finding carries + besides the flow's, when the rule tells cases of its sinks apart.""" verdict: Verdict severity: Severity + title: str | None = None + metadata: Mapping[str, str] = field(default_factory=dict) class TaintDetector(Plugin): @@ -57,13 +61,15 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: verdict = assessment.verdict if verdict.status is Status.REFUTED: continue - message = f"{self.title}: {flow.source.label} input reaches {flow.sink.symbol}" + title = assessment.title or self.title + message = f"{title}: {flow.source.label} input reaches {flow.sink.symbol}" metadata = { "source": str(flow.source.symbol), "source_label": flow.source.label, "sink": str(flow.sink.symbol), "verdict": verdict.status.value, "evidence": verdict.evidence, + **assessment.metadata, } if flow.through is not None and flow.sink_location is not None: message += f" through {flow.through}" diff --git a/tests/test_command_injection_nature.py b/tests/test_command_injection_nature.py new file mode 100644 index 0000000..27b65e3 --- /dev/null +++ b/tests/test_command_injection_nature.py @@ -0,0 +1,92 @@ +"""Acceptance tests for issue #220: what kind of injection a command-injection finding is. + +The rule id stays ``command-injection`` for every case, so suppressions and baselines +keep matching. When the rule reads a command element by element, it says what the +attacker's input is there in the ``injection`` metadata: + +- ``command``: the input is run, or chooses what runs: a shell string, the program, an + operand the program runs, an element after an option that runs the rest of the + command, the value of an option that runs the command it chooses; +- ``option``: the input can only reach the program as an option it may read (a whole + element, ``--{x}``, an argument of a script) or as the value of an option the model + does not establish to run it (``curl --output=``, an option it does not describe). + Its message is titled ``Option injection``. + +When the kind cannot be established (an option whose effect is to examine, an unknown +program or element before the input, a command that is not a list or tuple the rule +reads, a sink the rule does not read element by element), +the metadata is absent and the title stays ``Command injection``. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.source import SourceManager + +PLUGINS = Path(__file__).resolve().parent.parent / "src" / "coretrace_python" / "bundled" +COMMAND = ("Command injection", "command") +OPTION = ("Option injection", "option") +UNDETERMINED = ("Command injection", None) + + +def kinds(call: str) -> list[tuple[str, str, str | None]]: + """Rule id, message title and ``injection`` metadata of each command-injection finding + on ``call``, made with ``x`` read from the command line and ``extra`` unknown.""" + + 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.rule_id, f.message.split(":", 1)[0], f.metadata.get("injection")) + for f in findings + if f.rule_id == "command-injection" + ] + + +@pytest.mark.parametrize( + ("call", "expected"), + [ + ('subprocess.run(f"git tag {x}", shell=True)', COMMAND), + ('subprocess.run([x, "-l"])', COMMAND), + ('subprocess.run(["sh", "-c", x])', COMMAND), + ('subprocess.run(["python", "-c", x])', COMMAND), + ('subprocess.run(["pwsh", "-Command", "Get-Item", x])', COMMAND), + ('subprocess.run(["ssh", "host", f"cat {x}"])', COMMAND), + ('subprocess.run(["sudo", x])', COMMAND), + ('subprocess.run(["git", "fetch", f"--upload-pack={x}"])', COMMAND), + ('subprocess.run(["ls", x])', OPTION), + ('subprocess.run(["git", "push", "origin", x])', OPTION), + ('subprocess.run(["curl", f"--output={x}"])', OPTION), + ('subprocess.run(["ls", f"--{x}"])', OPTION), + ('subprocess.run(["node", "app.js", x])', OPTION), + ('subprocess.run(["notmodelled", "-c", f"echo {x}"])', OPTION), + ("subprocess.run(x, shell=extra)", UNDETERMINED), + ('subprocess.run(["python", "-m", x])', UNDETERMINED), + ('subprocess.run(["tar", "-I", x, "-xf", "a.tar"])', UNDETERMINED), + ('subprocess.run(["ls", *extra, x])', UNDETERMINED), + ('subprocess.run(["ls", x], executable=extra)', UNDETERMINED), + ('os.system("ls " + x)', UNDETERMINED), + # Not a list the rule reads: a string naming the program, or a list built elsewhere. + ("subprocess.run(x)", UNDETERMINED), + ], +) +def test_the_finding_says_what_kind_of_injection_it_is( + call: str, expected: tuple[str, str | None] +) -> None: + assert kinds(call) == [("command-injection", *expected)] + + +def test_a_suppression_of_the_rule_still_applies_to_an_option_injection() -> None: + text = ( + "import subprocess\nimport sys\n\n\ndef run():\n" + ' subprocess.run(["ls", sys.argv[1]]) # coretrace: ignore[command-injection]\n' + ) + analysis = engine.analyze_file(SourceManager().add_source("app/run.py", text), [PLUGINS]) + + assert [f.rule_id for f in analysis.findings] == [] + assert [(f.rule_id, f.metadata["injection"]) for f in analysis.suppressed] == [ + ("command-injection", "option") + ]