feat(code-injection): report dynamic imports and runpy reached by input

docs/usage.md

@@ -121,7 +121,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. 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, and as the only property of its SARIF result) 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. |
+| `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. Also the module `importlib.import_module`, `__import__` or `runpy.run_module` imports, and the file `runpy.run_path` runs, by position or by keyword (`name`, `mod_name`, `path_name`); the package of a relative name, the globals and the run name are data. An import runs the top-level code of a module the input chooses among those that can be found, a script path the file it chooses: execution is possible, and how far it goes depends on what the input can reach, so these findings are titled `Untrusted module import` and `Untrusted script execution`. A constant package prefix (`"app.plugins." + name`) confines an import to that package: a hotspot. The environment and the command line reach these sinks as they reach `eval`. |
 | `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. |
 | `path-traversal` | A file system path (`open`, `send_file`, `os.remove`, …). |
@@ -135,7 +135,12 @@ variable is not treated that way yet.
 Each flow is judged before it is reported. A dominating guard that proves the value
 safe (`isdigit()`, membership in a constant allowlist, equality with a constant, a
 numeric value proven by `int()`, `len()` or bounded arithmetic, a validator such as
-`re.fullmatch`) refutes it and nothing is reported. A hotspot is reported at medium
+`re.fullmatch`) refutes it and nothing is reported. A sink that selects code by name,
+the module of `importlib.import_module`, `__import__` or `runpy.run_module` and the file
+of `runpy.run_path`, is proven only by a check that limits the value to constants:
+membership in a constant allowlist or equality with a constant. A check of its shape
+(`isidentifier()`, `isalnum()`, `re.fullmatch`) limits its characters, not the modules
+or files it can name, and leaves a hotspot. A hotspot is reported at medium
 confidence. Two things make a flow a hotspot:
 
 - A guard that examines the data without proving it safe. That covers a test of the value

src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py

@@ -62,6 +62,18 @@ class PythonStdlibModels(ModelPlugin):
         # The code is the first argument; the globals and locals it runs with are data.
         Sink(_sym("builtins.eval"), TaintKind.CODE, ((TaintKind.CODE, (0,)),)),
         Sink(_sym("builtins.exec"), TaintKind.CODE, ((TaintKind.CODE, (0,)),)),
+        # The module or the file the call imports or runs; its globals, its run name and
+        # the package of a relative name are data (#240).
+        *(
+            Sink(_sym(symbol), TaintKind.CODE, ((TaintKind.CODE, (0,)),), ((TaintKind.CODE, (keyword,)),), selects=True)
+            for symbol, keyword in (
+                ("importlib.import_module", "name"),
+                ("builtins.__import__", "name"),
+                ("importlib.__import__", "name"),
+                ("runpy.run_module", "mod_name"),
+                ("runpy.run_path", "path_name"),
+            )
+        ),
         Sink(_sym("builtins.open"), TaintKind.PATH),
         Sink(_sym("os.remove"), TaintKind.PATH),
         Sink(_sym("os.unlink"), TaintKind.PATH),

src/coretrace_python/bundled/security/code_injection/code_injection.py

@@ -1,14 +1,33 @@
 """Code injection: attacker-controlled input reaching a CODE sink, such as the code
-``eval`` or ``exec`` runs. ``dangerous-eval`` reports every such call; this rule reports
-the ones attacker input reaches."""
+``eval`` or ``exec`` runs, the module ``importlib.import_module``, ``__import__`` or
+``runpy.run_module`` imports, or the file ``runpy.run_path`` runs. ``dangerous-eval``
+reports every ``eval`` or ``exec``; this rule reports the calls attacker input reaches.
+
+An import runs the top-level code of a module the input chooses among those that can be
+found, and a script path runs the file the input chooses: execution is possible, and
+how far it goes depends on what the input can reach. Their findings are titled for
+that. A constant package prefix (``"app.plugins." + name``) confines an import to the
+modules of that package: a hotspot."""
 
 from __future__ import annotations
 
 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.ir.ssa import SSAAnalysis
+from coretrace_python.plugins import Assessment, PluginContext, TaintDetector
+from coretrace_python.semantic.symbols import SymbolId
+from coretrace_python.taint import TaintFlow, TaintKind
+
+_IMPORTS = frozenset(
+    SymbolId(f"python.{name}")
+    for name in ("importlib.import_module", "builtins.__import__", "importlib.__import__", "runpy.run_module")
+)
+_RUN_PATH = SymbolId("python.runpy.run_path")
 
 
 class CodeInjectionPlugin(TaintDetector):
@@ -17,3 +36,23 @@ class CodeInjectionPlugin(TaintDetector):
     kind: ClassVar[TaintKind] = TaintKind.CODE
     severity: ClassVar[Severity] = Severity.HIGH
     title: ClassVar[str] = "Code injection"
+    requires: ClassVar[frozenset[AnyAnalysis]] = TaintDetector.requires | {SSAAnalysis, ModuleStringsAnalysis}
+
+    def assess(
+        self, ctx: PluginContext, function: nodes.Function, flow: TaintFlow, verdict: Verdict
+    ) -> Assessment:
+        symbol = flow.sink.symbol
+        if symbol == _RUN_PATH:
+            return Assessment(verdict, self.severity, "Untrusted script execution")
+        if symbol not in _IMPORTS:
+            return Assessment(verdict, self.severity)
+        if verdict.status is Status.VULNERABILITY and flow.through is None:
+            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}
+            prefix, whole = leading_text(flow.argument, defs, ctx.get(ModuleStringsAnalysis))
+            package = prefix.rpartition(".")[0]
+            if not whole and package and not prefix.startswith("."):
+                verdict = Verdict(
+                    flow, Status.HOTSPOT, f"the package is fixed by {prefix!r}, but the input chooses a module in it"
+                )
+        return Assessment(verdict, self.severity, "Untrusted module import")

src/coretrace_python/bundled/security/code_injection/plugin.toml

@@ -1,7 +1,7 @@
 name = "code-injection"
 version = "1.0.0"
 plugin_api = ">=1,<2"
-requires = ["taint.flows", "findings.refutation"]
+requires = ["taint.flows", "findings.refutation", "ir.ssa", "abstract.module_strings"]
 provides = ["vulnerability.code-injection"]
 
 [entrypoint]

src/coretrace_python/findings/refutation.py

@@ -247,16 +247,16 @@ def guards(self, sink: BlockId) -> list[_Guard]:
 
     def interpret(
         self, condition: Value, truth: bool | None
-    ) -> tuple[dict[Value, tuple[str, TaintKind]], set[Value]]:
-        """Values the condition proves safe (with the reason and the kinds it proves them
-        safe for), and values it mentions without proving anything. ``truth`` is ``None``
-        when it is not fixed at the sink.
+    ) -> tuple[dict[Value, tuple[str, TaintKind, bool]], set[Value]]:
+        """Values the condition proves safe (with the reason, the kinds it proves them
+        safe for, and whether it limits them to constants), and values it mentions without
+        proving anything. ``truth`` is ``None`` when it is not fixed at the sink.
 
         A recognised check evaluated the wrong way (``isdigit()`` known false) yields
         nothing: it is neither a proof nor a reassuring guard."""
 
         definition = self.defs.get(condition)
-        proven: dict[Value, tuple[str, TaintKind]] = {}
+        proven: dict[Value, tuple[str, TaintKind, bool]] = {}
         mentioned: set[Value] = set()
         if isinstance(definition, UnaryOp) and definition.operator == "not":
             return self.interpret(definition.operand, None if truth is None else not truth)
@@ -270,29 +270,37 @@ def interpret(
         recognised = self.recognise(definition)
         if recognised is None:
             return proven, set(self.closure(condition)) | {condition}
-        tested, reason, when_true, kinds = recognised
+        tested, reason, when_true, kinds, exact = recognised
         if truth is None:
             mentioned = set(self.closure(condition)) | {condition}
         elif truth == when_true:
             # The check proves ``tested``; whatever else it reads is merely mentioned.
-            proven[tested] = (reason, kinds)
+            proven[tested] = (reason, kinds, exact)
             mentioned = set(self.closure(condition)) | {condition}
         return proven, mentioned
 
-    def recognise(self, definition: Instruction | None) -> tuple[Value, str, bool, TaintKind] | None:
-        """``(validated value, reason, truth that validates, kinds it proves)`` for known
-        check shapes. A string check, a constant allowlist or an equality with a constant
-        proves every kind; a ``Validator`` only the kinds it declares."""
+    def recognise(self, definition: Instruction | None) -> tuple[Value, str, bool, TaintKind, bool] | None:
+        """``(validated value, reason, truth that validates, kinds it proves, whether it
+        limits the value to constants)`` for known check shapes. A string check, a constant
+        allowlist or an equality with a constant proves every kind; a ``Validator`` only
+        the kinds it declares. Only the allowlist and the equality limit the value to
+        constants: a string check or a validator checks its shape."""
 
         if isinstance(definition, Call) and not definition.arguments:
             callee = self.defs.get(definition.callee)
             if isinstance(callee, GetAttr) and callee.attribute in VALIDATORS:
-                return callee.object, f"guarded by {callee.attribute}()", True, TaintKind.ALL
+                return callee.object, f"guarded by {callee.attribute}()", True, TaintKind.ALL, False
         if isinstance(definition, Call):
             symbol = self.symbols.get(definition.callee)
             validator = self.models.validator(symbol) if symbol is not None else None
             if validator is not None and validator.argument < len(definition.arguments):
-                return definition.arguments[validator.argument], f"validated by {symbol}", True, validator.kinds
+                return (
+                    definition.arguments[validator.argument],
+                    f"validated by {symbol}",
+                    True,
+                    validator.kinds,
+                    False,
+                )
         if isinstance(definition, Compare):
             left, right = definition.left, definition.right
             if definition.operator in ("in", "not_in") and self.is_constant_collection(right):
@@ -301,11 +309,12 @@ def recognise(self, definition: Instruction | None) -> tuple[Value, str, bool, T
                     "allowlisted by a membership check on constants",
                     definition.operator == "in",
                     TaintKind.ALL,
+                    True,
                 )
             if definition.operator in ("eq", "not_eq"):
                 for tested, other in ((left, right), (right, left)):
                     if isinstance(self.defs.get(other), Constant):
-                        return tested, "equals a constant", definition.operator == "eq", TaintKind.ALL
+                        return tested, "equals a constant", definition.operator == "eq", TaintKind.ALL, True
         return None
 
     def examined(
@@ -458,10 +467,12 @@ def judge(self, flow: TaintFlow, reachable: bool) -> Verdict:
             return Verdict(flow, Status.REFUTED, "sink unreachable by constant propagation")
         origins = self.origins(flow.argument)
         chain = self.closure(flow.argument) | {flow.argument}
+        # A sink that selects code by name is proven only by constants (``Sink.selects``).
+        selects = flow.sink.selects
         proven: dict[Value, str] = {
             value: f"numeric value within {interval}"
             for value, interval in self.ranges.at(sink).items()
-            if value in chain
+            if value in chain and not selects
         }
         mentions: dict[Value, int] = {}
         authorization: str | None = (
@@ -471,9 +482,9 @@ def judge(self, flow: TaintFlow, reachable: bool) -> Verdict:
         )
         for guard in self.guards(sink):
             found, mentioned = self.interpret(guard.condition, guard.truth)
-            for value, (reason, kinds) in found.items():
+            for value, (reason, kinds, exact) in found.items():
                 # A check proving some kinds only leaves the others reaching the sink.
-                if value in chain and not flow.kinds & ~kinds:
+                if value in chain and not flow.kinds & ~kinds and (exact or not selects):
                     proven.setdefault(value, reason)
             examined = self.examined(guard.condition, mentioned, flow.argument, chain)
             for origin in origins:

src/coretrace_python/semantic/symbols.py

@@ -14,7 +14,8 @@
 from coretrace_python.semantic.imports import ImportAnalysis, ImportTable
 from coretrace_python.semantic.scopes import ResolutionKind, ScopeAnalysis, ScopeId, ScopeTable
 
-BUILTIN_NAMES = frozenset(name for name in dir(builtins) if not name.startswith("_"))
+# ``__import__`` is the one dunder of the builtins a program calls by name (#240).
+BUILTIN_NAMES = frozenset(name for name in dir(builtins) if not name.startswith("_")) | {"__import__"}
 
 
 @dataclass(frozen=True)

src/coretrace_python/taint/models.py

@@ -65,12 +65,16 @@ class Sink:
     """A callable whose arguments must not carry the given taint kinds. ``positions`` and
     ``keywords`` restrict some kinds to the arguments they list: a SQL statement is the
     first argument of ``execute``, its parameter tuple is not a statement; the target of
-    ``requests.get`` is its first argument or ``url=``, its body is no destination."""
+    ``requests.get`` is its first argument or ``url=``, its body is no destination.
+    ``selects`` marks a sink whose argument chooses by name the code that runs, a module
+    or a file: only a check that limits the value to constants proves it, not a check of
+    its shape (#240)."""
 
     symbol: SymbolId
     kinds: TaintKind
     positions: tuple[tuple[TaintKind, tuple[int, ...]], ...] = ()
     keywords: tuple[tuple[TaintKind, tuple[str, ...]], ...] = ()
+    selects: bool = False
 
     def kinds_at(self, position: int | None, keyword: str | None = None) -> TaintKind:
         """The kinds that must not reach the argument at ``position`` or passed as
@@ -90,13 +94,15 @@ def kinds_at(self, position: int | None, keyword: str | None = None) -> TaintKin
         return self.kinds & ~(restricted & ~allowed)
 
     def merged(self, other: Sink) -> Sink:
-        """This sink with ``other``'s kinds, positions and keywords added."""
+        """This sink with ``other``'s kinds, positions and keywords added; it selects code
+        by name when either does, whatever the order the declarations merge in."""
 
         return Sink(
             self.symbol,
             self.kinds | other.kinds,
             self.positions + other.positions,
             self.keywords + other.keywords,
+            self.selects or other.selects,
         )
 
 

tests/test_detectors.py

@@ -202,8 +202,15 @@ 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", "code-injection", "nosql-injection", "path-traversal", "xss"):
+    for rule in ("sql-injection", "nosql-injection", "path-traversal", "xss"):
         assert by_name[rule].requires == ("taint.flows", "findings.refutation")
+    # ``code-injection`` also reads how the name a dynamic import receives is built (#240).
+    assert by_name["code-injection"].requires == (
+        "taint.flows",
+        "findings.refutation",
+        "ir.ssa",
+        "abstract.module_strings",
+    )
     # ``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.