From 8f4116ab1523af081dd6eaa8082d26b77058f203 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 13 Oct 2026 08:28:22 +0200 Subject: [PATCH 1/4] feat(code-injection): report dynamic imports and runpy reached by input Part of #240: sys.path and getattr are left for a next step. --- docs/usage.md | 2 +- .../models/python_stdlib/python_stdlib.py | 12 ++ .../security/code_injection/code_injection.py | 47 ++++- .../security/code_injection/plugin.toml | 2 +- src/coretrace_python/semantic/symbols.py | 3 +- tests/test_detectors.py | 9 +- tests/test_dynamic_import_sinks.py | 176 ++++++++++++++++++ 7 files changed, 243 insertions(+), 8 deletions(-) create mode 100644 tests/test_dynamic_import_sinks.py diff --git a/docs/usage.md b/docs/usage.md index 2c44caa..835b7c4 100644 --- a/docs/usage.md +++ b/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`, …). | diff --git a/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py b/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py index e0a52e6..ed7204a 100644 --- a/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py +++ b/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,)),)) + 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), diff --git a/src/coretrace_python/bundled/security/code_injection/code_injection.py b/src/coretrace_python/bundled/security/code_injection/code_injection.py index 495bca4..a33ba3c 100644 --- a/src/coretrace_python/bundled/security/code_injection/code_injection.py +++ b/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") diff --git a/src/coretrace_python/bundled/security/code_injection/plugin.toml b/src/coretrace_python/bundled/security/code_injection/plugin.toml index b0a4e23..06614f6 100644 --- a/src/coretrace_python/bundled/security/code_injection/plugin.toml +++ b/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] diff --git a/src/coretrace_python/semantic/symbols.py b/src/coretrace_python/semantic/symbols.py index 3eec67b..f2b1953 100644 --- a/src/coretrace_python/semantic/symbols.py +++ b/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) diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 59cb396..597fc4c 100644 --- a/tests/test_detectors.py +++ b/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. diff --git a/tests/test_dynamic_import_sinks.py b/tests/test_dynamic_import_sinks.py new file mode 100644 index 0000000..5f41d49 --- /dev/null +++ b/tests/test_dynamic_import_sinks.py @@ -0,0 +1,176 @@ +"""Acceptance tests for issue #240, part 1: the dynamic import and module-execution APIs +are ``code-injection`` sinks. + +``importlib.import_module``, ``__import__`` and ``runpy.run_module`` import or run a module +the input chooses among those that can be found; ``runpy.run_path`` runs the file the +input chooses. Only the argument naming the module or the file is a sink, by position +or keyword; the globals and the run name are data. A constant package prefix confines +the input to the modules of that package: a hotspot. The environment and the command +line reach these sinks as they reach ``eval``. +""" + +from __future__ import annotations + +import contextlib +import io +import json + +import pytest + +from coretrace_python.cli import main + +APP = """\ +import importlib +import os +import runpy +import sys + +from flask import Flask, request + +app = Flask(__name__) +SCRIPTS = "/srv/scripts" + + +def load(name): + return importlib.import_module(name) + + +@app.route("/import-module") +def import_module(): + return importlib.import_module(request.args["m"]) + + +@app.route("/import-module-keyword") +def import_module_keyword(): + return importlib.import_module(name=request.args["m"]) + + +@app.route("/import-module-package") +def import_module_package(): + return importlib.import_module(request.args["m"], package="app") + + +@app.route("/builtin-import") +def builtin_import(): + return __import__(request.args["m"]) + + +@app.route("/importlib-import") +def importlib_import(): + return importlib.__import__(request.args["m"]) + + +@app.route("/run-module") +def run_module(): + return runpy.run_module(request.args["m"]) + + +@app.route("/run-path") +def run_path(): + return runpy.run_path(request.args["p"]) + + +@app.route("/run-path-directory") +def run_path_directory(): + return runpy.run_path(os.path.join(SCRIPTS, request.args["p"])) + + +@app.route("/through-helper") +def through_helper(): + return load(request.args["m"]) + + +@app.route("/package-prefix") +def package_prefix(): + return importlib.import_module("app.plugins." + request.args["m"]) + + +@app.route("/package-prefix-format") +def package_prefix_format(): + return importlib.import_module(f"app.plugins.{request.args['m']}") + + +@app.route("/constant-name") +def constant_name(): + return importlib.import_module("json") + + +@app.route("/allowlisted-name") +def allowlisted_name(): + name = request.args["m"] + if name in ("csv", "json"): + return importlib.import_module(name) + + +@app.route("/data-arguments-only") +def data_arguments_only(): + return runpy.run_path("jobs/report.py", init_globals={"q": request.args["q"]}, run_name=request.args["n"]) + + +def settings_module(): + return importlib.import_module(os.environ["DJANGO_SETTINGS_MODULE"]) + + +def plugin_from_command_line(): + return importlib.import_module(sys.argv[1]) + + +def evaluate_environment(): + return eval(os.environ["EXPRESSION"]) + + +def evaluate_command_line(): + return eval(sys.argv[1]) +""" + +IMPORT = ("vulnerability", "high", "Untrusted module import") +SCRIPT = ("vulnerability", "high", "Untrusted script execution") +EVAL = ("vulnerability", "high", "Code injection") + +EXPECTED = { + "import_module": IMPORT, + "import_module_keyword": IMPORT, + "import_module_package": IMPORT, + "builtin_import": IMPORT, + "importlib_import": IMPORT, + "run_module": IMPORT, + "run_path": SCRIPT, + "run_path_directory": SCRIPT, + "through_helper": IMPORT, + "package_prefix": ("hotspot", "medium", "Untrusted module import"), + "package_prefix_format": ("hotspot", "medium", "Untrusted module import"), + "constant_name": None, + "allowlisted_name": None, + "data_arguments_only": None, + # The environment and the command line carry ``CODE`` (#240, decided). + "settings_module": IMPORT, + "plugin_from_command_line": IMPORT, + "evaluate_environment": EVAL, + "evaluate_command_line": EVAL, +} + + +@pytest.fixture(scope="module") +def reported(tmp_path_factory: pytest.TempPathFactory) -> dict[str, tuple[str, str, str]]: + root = tmp_path_factory.mktemp("dynamic_imports") + (root / "app.py").write_text(APP, encoding="utf-8") + out = io.StringIO() + with contextlib.redirect_stdout(out): + main(["--check", str(root), "--format", "json"]) + found: dict[str, tuple[str, str, str]] = {} + for finding in json.loads(out.getvalue())["findings"]: + if finding["rule_id"] != "code-injection": + continue + assert finding["function"] not in found, finding + title = finding["message"].split(":", 1)[0] + found[finding["function"]] = (finding["metadata"]["verdict"], finding["confidence"], title) + return found + + +@pytest.mark.parametrize("function", sorted(EXPECTED)) +def test_dynamic_import_verdict(reported: dict[str, tuple[str, str, str]], function: str) -> None: + assert reported.get(function) == EXPECTED[function] + + +def test_every_reported_function_is_expected(reported: dict[str, tuple[str, str, str]]) -> None: + assert set(reported) <= set(EXPECTED) From 1fd0e7cbdb9a9c8c1fa5bf0a6a9b9838bcab9450 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 13 Oct 2026 08:32:01 +0200 Subject: [PATCH 2/4] test(code-injection): pin package prefixes, input-controlled names and unchanged eval/exec Part of #240. --- tests/test_dynamic_import_verdicts.py | 101 ++++++++++++++++++++++++++ 1 file changed, 101 insertions(+) create mode 100644 tests/test_dynamic_import_verdicts.py diff --git a/tests/test_dynamic_import_verdicts.py b/tests/test_dynamic_import_verdicts.py new file mode 100644 index 0000000..8a727f8 --- /dev/null +++ b/tests/test_dynamic_import_verdicts.py @@ -0,0 +1,101 @@ +"""The verdicts of #240, one concern at a time: what a constant package prefix proves, a +name the input controls entirely, and ``eval`` and ``exec`` left as they were. + +A constant prefix that ends a package path (``"app.plugins." + name``) confines an import +to the modules of that package: a hotspot. A prefix that fixes no package, a relative +name, input before the constant, or a directory in front of a script path confines +nothing. ``eval`` and ``exec`` run the code they are given: no prefix proves anything +there. +""" + +from __future__ import annotations + +import pytest + +from coretrace_python import engine +from coretrace_python.findings import Finding +from coretrace_python.source import SourceManager + +IMPORTS = "import importlib\nimport runpy\n\nPLUGINS = 'app.plugins.'\n\n\n" + + +def injection(body: str) -> Finding: + text = f"{IMPORTS}def run(prefix):\n {body}\n" + findings = [ + f + for f in engine.check(SourceManager().add_source("app.py", text), [engine.BUNDLED_PLUGINS]) + if f.rule_id == "code-injection" + ] + (finding,) = findings + return finding + + +def verdict(finding: Finding) -> tuple[str, str, str]: + return finding.metadata["verdict"], finding.confidence.value, finding.message.split(":", 1)[0] + + +@pytest.mark.parametrize( + "body", + [ + "importlib.import_module('app.plugins.' + input())", + "importlib.import_module(f'app.plugins.{input()}')", + "importlib.import_module('app.' + input())", + "importlib.import_module(PLUGINS + input())", + "__import__('app.plugins.' + input())", + "runpy.run_module('app.jobs.' + input())", + ], +) +def test_a_constant_package_prefix_makes_a_hotspot(body: str) -> None: + finding = injection(body) + + assert verdict(finding) == ("hotspot", "medium", "Untrusted module import") + assert "the package is fixed by" in finding.metadata["evidence"] + + +@pytest.mark.parametrize( + ("body", "title"), + [ + # No package fixed: ``"app" + name`` can be any module whose name starts so. + ("importlib.import_module('app' + input())", "Untrusted module import"), + # A relative name is resolved against the ``package`` argument, which the rule + # does not read: the package it lands in is not established. + ("importlib.import_module('.' + input(), package='app')", "Untrusted module import"), + ("importlib.import_module('.plugins.' + input(), package='app')", "Untrusted module import"), + ("importlib.import_module(input() + '.plugins')", "Untrusted module import"), + ("importlib.import_module(prefix + input())", "Untrusted module import"), + # ``../`` leaves any directory. + ("runpy.run_path('/srv/scripts/' + input())", "Untrusted script execution"), + ], +) +def test_a_prefix_that_confines_nothing_stays_a_vulnerability(body: str, title: str) -> None: + assert verdict(injection(body)) == ("vulnerability", "high", title) + + +@pytest.mark.parametrize( + ("body", "title"), + [ + ("importlib.import_module(input())", "Untrusted module import"), + ("importlib.import_module(name=input())", "Untrusted module import"), + ("__import__(input())", "Untrusted module import"), + ("importlib.__import__(input())", "Untrusted module import"), + ("runpy.run_module(input())", "Untrusted module import"), + ("runpy.run_module(mod_name=input())", "Untrusted module import"), + ("runpy.run_path(input())", "Untrusted script execution"), + ("runpy.run_path(path_name=input())", "Untrusted script execution"), + ], +) +def test_a_name_the_input_controls_entirely_is_a_vulnerability(body: str, title: str) -> None: + assert verdict(injection(body)) == ("vulnerability", "high", title) + + +@pytest.mark.parametrize( + "body", + [ + "eval(input())", + "exec(input())", + "eval('app.plugins.' + input())", + "exec('import ' + input())", + ], +) +def test_eval_and_exec_are_unchanged(body: str) -> None: + assert verdict(injection(body)) == ("vulnerability", "high", "Code injection") From be52ac94fe2873ab4c5f25f209d498f47c4a302a Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 13 Oct 2026 19:26:20 +0200 Subject: [PATCH 3/4] fix(refutation): prove sinks that select code by name only by constants A check of the shape of a module or file name (isidentifier(), isalnum(), re.fullmatch) limits its characters, not the code it names. For the dynamic import and runpy sinks it leaves a hotspot; only membership in a constant allowlist or equality with a constant proves them. Part of #240. --- docs/usage.md | 7 +- .../models/python_stdlib/python_stdlib.py | 2 +- src/coretrace_python/findings/refutation.py | 45 +++++++----- src/coretrace_python/taint/models.py | 6 +- tests/test_selection_sink_proofs.py | 69 +++++++++++++++++++ 5 files changed, 109 insertions(+), 20 deletions(-) create mode 100644 tests/test_selection_sink_proofs.py diff --git a/docs/usage.md b/docs/usage.md index 835b7c4..9d28265 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -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 diff --git a/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py b/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py index ed7204a..276ea86 100644 --- a/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py +++ b/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py @@ -65,7 +65,7 @@ class PythonStdlibModels(ModelPlugin): # 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,)),)) + Sink(_sym(symbol), TaintKind.CODE, ((TaintKind.CODE, (0,)),), ((TaintKind.CODE, (keyword,)),), selects=True) for symbol, keyword in ( ("importlib.import_module", "name"), ("builtins.__import__", "name"), diff --git a/src/coretrace_python/findings/refutation.py b/src/coretrace_python/findings/refutation.py index 945af8c..5cc44ab 100644 --- a/src/coretrace_python/findings/refutation.py +++ b/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: diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index 3efd2cd..eac0dab 100644 --- a/src/coretrace_python/taint/models.py +++ b/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 diff --git a/tests/test_selection_sink_proofs.py b/tests/test_selection_sink_proofs.py new file mode 100644 index 0000000..47b1179 --- /dev/null +++ b/tests/test_selection_sink_proofs.py @@ -0,0 +1,69 @@ +"""Review of #240 (PR #250): a check of the shape of a name does not prove an import. + +``isidentifier()``, ``isalnum()``, ``isdigit()`` or a ``re.fullmatch`` validator limit the +characters of a name, not the modules or files it can name: the input still chooses +the code that runs. For the sinks that select code by name (``Sink.selects``), only a +check that limits the value to constants proves it: membership in a constant +collection or equality with a constant. A guard that only checks the shape leaves a +hotspot. ``eval``, ``exec`` and every other sink keep the proofs they had. +""" + +from __future__ import annotations + +import pytest + +from coretrace_python import engine +from coretrace_python.findings import Finding +from coretrace_python.source import SourceManager + +HEADER = "import importlib\nimport os\nimport re\nimport runpy\n\n\ndef run():\n name = input()\n" + + +def findings(body: str, rule: str) -> list[Finding]: + text = HEADER + "".join(f" {line}\n" for line in body.splitlines()) + analysed = engine.check(SourceManager().add_source("app.py", text), [engine.BUNDLED_PLUGINS]) + return [f for f in analysed if f.rule_id == rule] + + +SINKS = [ + "importlib.import_module(name)", + "__import__(name)", + "runpy.run_module(name)", + "runpy.run_path(name)", +] + + +@pytest.mark.parametrize("sink", SINKS) +@pytest.mark.parametrize( + "guard", + [ + "name.isidentifier()", + "name.isalnum()", + "name.isdigit()", + "name.isascii()", + "re.fullmatch(r'[a-z_]+', name)", + ], +) +def test_a_check_of_the_shape_of_the_name_leaves_a_hotspot(sink: str, guard: str) -> None: + (finding,) = findings(f"if {guard}:\n {sink}", "code-injection") + + assert finding.metadata["verdict"] == "hotspot" + assert finding.confidence.value == "medium" + + +@pytest.mark.parametrize("sink", SINKS) +@pytest.mark.parametrize("guard", ["name in ('csv', 'json')", "name == 'json'"]) +def test_a_check_against_constants_still_proves_the_name(sink: str, guard: str) -> None: + assert findings(f"if {guard}:\n {sink}", "code-injection") == [] + + +@pytest.mark.parametrize( + ("body", "rule"), + [ + ("if name.isidentifier():\n eval(name)", "code-injection"), + ("if name.isalnum():\n exec(name)", "code-injection"), + ("if name.isalnum():\n os.system(name)", "command-injection"), + ], +) +def test_other_sinks_keep_the_proofs_they_had(body: str, rule: str) -> None: + assert findings(body, rule) == [] From 2641b57e28c654a83c2485dd96a0456691f070a8 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 13 Oct 2026 23:37:26 +0200 Subject: [PATCH 4/4] fix(taint): keep a sink's selection when models merge Sink.merged rebuilt the sink without selects, so another plugin declaring the same import sink brought back shape checks as proofs. The merged sink selects when either declaration does, in either order. Part of #240. --- src/coretrace_python/taint/models.py | 4 +- tests/test_sink_merge.py | 72 ++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+), 1 deletion(-) create mode 100644 tests/test_sink_merge.py diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index eac0dab..0cf2c94 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -94,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, ) diff --git a/tests/test_sink_merge.py b/tests/test_sink_merge.py new file mode 100644 index 0000000..fa83ebf --- /dev/null +++ b/tests/test_sink_merge.py @@ -0,0 +1,72 @@ +"""Review of #240 (PR #250): a sink that selects code by name stays one when the models +of several plugins are merged. + +Two plugins may declare the same sink; the registry merges them (``Sink.merged``). The +merged sink selects code by name when either declaration says so, whatever the order the +plugins load in, so a check of the shape of the name still leaves a hotspot. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import Sink, TaintKind + +IMPORT_MODULE = SymbolId("python.importlib.import_module") + + +@pytest.mark.parametrize( + ("first", "second"), + [(True, False), (False, True), (True, True), (False, False)], +) +def test_a_merged_sink_selects_when_either_declaration_does(first: bool, second: bool) -> None: + left = Sink(IMPORT_MODULE, TaintKind.CODE, ((TaintKind.CODE, (0,)),), selects=first) + right = Sink(IMPORT_MODULE, TaintKind.CODE, ((TaintKind.CODE, (0,)),), selects=second) + + assert left.merged(right).selects is (first or second) + assert right.merged(left).selects is (first or second) + + +MANIFEST = ( + 'name = "import-sink"\nversion = "1.0.0"\nplugin_api = ">=1,<2"\nrequires = []\n' + 'provides = []\n\n[entrypoint]\nmodule = "import_sink"\nclass = "ImportSink"\n' +) +# Another plugin declaring the import sink, without ``selects``. +PLUGIN = ''' +from typing import ClassVar + +from coretrace_python.plugins import ModelPlugin +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import Sink, TaintKind + + +class ImportSink(ModelPlugin): + name: ClassVar[str] = "import-sink" + models: ClassVar = (Sink(SymbolId("python.importlib.import_module"), TaintKind.CODE),) +''' +APP = ( + "import importlib\n\nfrom flask import request\n\n\n" + "def load():\n name = request.args['name']\n if name.isidentifier():\n importlib.import_module(name)\n" +) + + +@pytest.mark.parametrize("extra_first", [True, False], ids=["extra-plugin-first", "bundled-first"]) +def test_another_plugin_declaring_the_sink_keeps_the_hotspot(tmp_path: Path, extra_first: bool) -> None: + plugin = tmp_path / "plugins" / "import_sink" + plugin.mkdir(parents=True) + (plugin / "plugin.toml").write_text(MANIFEST, encoding="utf-8") + (plugin / "import_sink.py").write_text(PLUGIN, encoding="utf-8") + root = tmp_path / "project" + root.mkdir() + (root / "app.py").write_text(APP, encoding="utf-8") + roots = [tmp_path / "plugins", engine.BUNDLED_PLUGINS] + + findings = engine.analyze_project(root, roots if extra_first else roots[::-1]).findings + + assert [(f.metadata["verdict"], f.confidence.value) for f in findings if f.rule_id == "code-injection"] == [ + ("hotspot", "medium") + ]