feat(command-injection): title option injections and record the injection kind (#220)

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`
 

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. |

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."""
 

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}"