diff --git a/docs/usage.md b/docs/usage.md index 2c44caa..9d28265 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`, …). | @@ -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 e0a52e6..276ea86 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,)),), 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), 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/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/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/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index 3efd2cd..0cf2c94 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 @@ -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, ) 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) 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") 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) == [] 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") + ]