fix(command-injection): judge process commands by what the input controls

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

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

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]

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)

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)

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

src/coretrace_python/plugins/detectors.py

@@ -1,14 +1,16 @@
 """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.
 """
 
 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)

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)