From d7a3ebda786ee02e1b239f8794209eb217f8e4d1 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 15 Oct 2026 09:19:13 +0200 Subject: [PATCH 1/3] fix(summaries): keep nothing of the blocks constant propagation proves unreachable Closes #258 --- docs/usage.md | 6 +- .../interprocedural/summaries.py | 13 ++- tests/test_summary_reachability.py | 81 +++++++++++++++++++ 3 files changed, 95 insertions(+), 5 deletions(-) create mode 100644 tests/test_summary_reachability.py diff --git a/docs/usage.md b/docs/usage.md index 94c739f..37ef46d 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -132,7 +132,11 @@ variable is not treated that way yet. | `plaintext-credential-storage` | A parameter named like a password stored in a database without hashing. Medium confidence, since a name is a hint. | | `exploitable-vulnerability` | An API affected by an advisory of a vulnerable requirement, see Dependencies. | -Each flow is judged before it is reported. A dominating guard that proves the value +Each flow is judged before it is reported. Code constant propagation proves can never +run (under `if False:`, `while False:`) reports nothing, in the function and through the +project functions that contain it: their summaries keep nothing of such code, neither a +sink nor a returned, merged or stored value. Code that may run is kept, whatever decides +it at run time, a parameter or a module-level name. 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 sink that selects code by name, diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index 028dd1c..30e460d 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -509,6 +509,11 @@ def evaluate(self, block: BasicBlock, incoming: Mapping[BlockId, State]) -> Stat return MappingProxyType(state) def flow(self, cfg: CFG, block_id: BlockId, incoming: Mapping[BlockId, State]) -> Mapping[BlockId, State]: + # A block constant propagation proves unreachable has no effect: no store, call, + # return or write, and nothing it would pass on reaches its successors or their + # phis (#258). A block that may run is kept, whatever decides it. + if not self.reachable(block_id): + return {} block = self.blocks[block_id] state = self.evaluate(block, incoming) exits = {target: state for target in block.exception_targets} @@ -576,9 +581,9 @@ def call(self, call: Call, state: dict[Key, Dep]) -> Dep: given = self.graph.arguments_at(self.name, call.location) self.record(target.symbol, arguments, named, call.location, None, given) # Where its result is called: facts the summary keeps whatever the models; - # the consumer decides which kinds of a sink apply there (#240). An invocation - # is kept only when it and the call can run, by constant propagation. - if call.result is not None and self.runs(call): + # the consumer decides which kinds of a sink apply there (#240). The call runs, + # as its block is evaluated; an invocation is kept only when it can run too. + if call.result is not None: for invocation in invocations(self.function, self.uses, call.result): if self.runs(invocation): self.record(target.symbol, arguments, named, call.location, None, given, invocation.location) @@ -769,7 +774,7 @@ def summarize( problem.mutated_globals = set() problem.nonlocal_stores = {} for block in function.blocks: - if solution.reached(block.id): + if solution.reached(block.id) and problem.reachable(block.id): problem.evaluate(block, solution.incoming(block.id)) mutations: list[Mutation] = [] for index, parameter in enumerate(function.parameters): diff --git a/tests/test_summary_reachability.py b/tests/test_summary_reachability.py new file mode 100644 index 0000000..f8ed947 --- /dev/null +++ b/tests/test_summary_reachability.py @@ -0,0 +1,81 @@ +"""Acceptance tests for issue #258: a summary keeps nothing of the blocks constant +propagation proves unreachable. + +A project function's summary carries its sinks, its return, the arguments it mutates +and the values merged at a join to its callers. A block that can never run contributes +none of them: no flow is reported at a caller for code under ``if False:`` or +``while False:``, through one or several functions. A block that may run is kept, +whatever decides it at run time: a parameter, or a module-level name the analysis of +one function does not resolve. +""" + +from __future__ import annotations + +import pytest + +from coretrace_python import engine +from coretrace_python.source import SourceManager + + +def injections(body: str) -> list[tuple[int, str]]: + text = "import os\n\n\n" + body + findings = engine.check(SourceManager().add_source("app.py", text), [engine.BUNDLED_PLUGINS]) + return [(f.span.start_line, f.metadata["verdict"]) for f in findings if f.rule_id == "command-injection"] + + +@pytest.mark.parametrize( + "body", + [ + # A sink. + "def helper(cmd):\n if False:\n os.system(cmd)\n\n\ndef caller():\n helper(input())\n", + "def helper(cmd):\n while False:\n os.system(cmd)\n\n\ndef caller():\n helper(input())\n", + # A return. + "def helper(cmd):\n if False:\n return cmd\n return 'ls'\n\n\ndef caller():\n os.system(helper(input()))\n", + # A value merged at a join from the block that cannot run. + ( + "def helper(cmd):\n result = 'ls'\n if False:\n result = cmd\n return result\n\n\n" + "def caller():\n os.system(helper(input()))\n" + ), + # A mutated argument. + ( + "def helper(items, cmd):\n if False:\n items.append(cmd)\n\n\n" + "def caller():\n items = []\n helper(items, input())\n os.system(items[0])\n" + ), + # Through two functions. + ( + "def inner(cmd):\n if False:\n os.system(cmd)\n\n\ndef outer(cmd):\n inner(cmd)\n\n\n" + "def caller():\n outer(input())\n" + ), + ], +) +def test_nothing_of_a_block_that_cannot_run_reaches_the_callers(body: str) -> None: + assert injections(body) == [] + + +@pytest.mark.parametrize( + ("body", "line"), + [ + # Decided at run time by a parameter. + ("def helper(cmd, flag):\n if flag:\n os.system(cmd)\n\n\ndef caller(flag):\n helper(input(), flag)\n", 10), + # A module-level name, not resolved within the function. + ("DEBUG = False\n\n\ndef helper(cmd):\n if DEBUG:\n os.system(cmd)\n\n\ndef caller():\n helper(input())\n", 13), + # The same effects in blocks that run. + ("def helper(cmd):\n if True:\n os.system(cmd)\n\n\ndef caller():\n helper(input())\n", 10), + ( + ( + "def helper(cmd):\n result = 'ls'\n if True:\n result = cmd\n return result\n\n\n" + "def caller():\n os.system(helper(input()))\n" + ), + 12, + ), + ( + ( + "def helper(items, cmd):\n if True:\n items.append(cmd)\n\n\n" + "def caller():\n items = []\n helper(items, input())\n os.system(items[0])\n" + ), + 12, + ), + ], +) +def test_a_block_that_may_run_keeps_its_effects(body: str, line: int) -> None: + assert injections(body) == [(line, "vulnerability")] From b2fe0787a9afa81dd76983e48dd2d85ed9bc11c0 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 15 Oct 2026 10:51:32 +0200 Subject: [PATCH 2/3] test(cache): pin that a new engine version reuses nothing cached The fix of #258 keeps every cache key between commits of one version; the engine version, part of every key, is what delivers it to cached projects. --- tests/test_cache_engine_version.py | 42 ++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) create mode 100644 tests/test_cache_engine_version.py diff --git a/tests/test_cache_engine_version.py b/tests/test_cache_engine_version.py new file mode 100644 index 0000000..e4ca1a3 --- /dev/null +++ b/tests/test_cache_engine_version.py @@ -0,0 +1,42 @@ +"""What makes a fix reach a cached project (#258, review of PR #259). + +A change to the analysis that keeps the cache format and the plugins, as the summaries +dropping unreachable blocks does, keeps every key between commits of one version: a +cache built before it would still serve the old results. What delivers it is the +version of the engine, part of every key: once it changes, nothing cached is reused and +every module is analysed again with the engine that reads it. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cache import ProjectCache + +APP = "import os\n\n\ndef helper(cmd):\n if False:\n os.system(cmd)\n\n\ndef caller():\n helper(input())\n" + + +def project(tmp_path: Path) -> Path: + root = tmp_path / "project" + (root / "app").mkdir(parents=True) + (root / "app" / "__init__.py").write_text("", encoding="utf-8") + (root / "app" / "main.py").write_text(APP, encoding="utf-8") + return root + + +def test_a_new_engine_version_reuses_nothing_cached(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = project(tmp_path) + cache = ProjectCache(tmp_path / "cache") + built = engine.analyze_project(root, [engine.BUNDLED_PLUGINS], cache=cache) + same_version = engine.analyze_project(root, [engine.BUNDLED_PLUGINS], cache=cache) + + monkeypatch.setattr(engine, "__version__", "0.0.0+next") + next_version = engine.analyze_project(root, [engine.BUNDLED_PLUGINS], cache=cache) + + assert set(same_version.reused) == set(built.keys) == {"app", "app.main"} + assert next_version.reused == () + assert set(next_version.keys.values()).isdisjoint(built.keys.values()) + assert [f for f in next_version.findings if f.rule_id == "command-injection"] == [] From 1fe578d0bbe09f071c5469ec039d72f5a6320d2a Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 16 Oct 2026 01:28:57 +0200 Subject: [PATCH 3/3] fix(summaries): keep a nonlocal's value where a nested function may leave it unwritten NonlocalWrite.kept says the function may return without writing the name, and no record means it never does; NonlocalResult carries the variable's value before the call, which the caller keeps there, joined with what may be written. A write that cannot run, a write on some paths only, or none at all no longer erases the caller's taint. PyIR schema 2, cache format 16. Part of #258. --- docs/usage.md | 5 +- src/coretrace_python/cache.py | 14 ++- .../interprocedural/summaries.py | 57 +++++++-- src/coretrace_python/ir/lowering.py | 3 +- src/coretrace_python/ir/model.py | 7 +- src/coretrace_python/ir/printer.py | 5 +- src/coretrace_python/taint/engine.py | 16 +-- tests/test_nonlocal_kept_value.py | 116 ++++++++++++++++++ 8 files changed, 198 insertions(+), 25 deletions(-) create mode 100644 tests/test_nonlocal_kept_value.py diff --git a/docs/usage.md b/docs/usage.md index 37ef46d..b1fe7e4 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -917,7 +917,10 @@ such as `metaclass=`, `with` targets of any assignable form and `match` with eve pattern kind. `finally` is modelled on the normal path only. A `nonlocal` write made by a nested function flows back to the enclosing variable when the enclosing function calls that nested function directly by name; a nested function passed -around and called elsewhere keeps its writes to itself. Methods of module-level classes are +around and called elsewhere keeps its writes to itself. Where the nested function may +return without writing the name (no write on some path, a write constant propagation +proves cannot run, or an exception before it), the variable keeps the value it had +before the call, joined with what may be written; only a write on every path replaces it. Methods of module-level classes are analysed like functions. ## Looking at the intermediate representation diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index 6bf2d39..a76a4c9 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -42,7 +42,7 @@ from coretrace_python.semantic.symbols import SymbolId from coretrace_python.source import FileLocation, Location, SourceId, SourceSpan -CACHE_FORMAT = 15 +CACHE_FORMAT = 16 @dataclass(frozen=True) @@ -300,7 +300,12 @@ def _encode_summary(summary: FunctionSummary) -> dict[str, Any]: ], "side_effects": sorted(summary.side_effects), "nonlocal_writes": [ - {"name": w.name, "dependencies": sorted(w.dependencies), "externals": sorted(str(s) for s in w.externals)} + { + "name": w.name, + "dependencies": sorted(w.dependencies), + "externals": sorted(str(s) for s in w.externals), + "kept": w.kept, + } for w in summary.nonlocal_writes ], "static": summary.static, @@ -328,7 +333,10 @@ def _decode_summary(data: Mapping[str, Any]) -> FunctionSummary: frozenset(_string(name) for name in data["side_effects"]), tuple( NonlocalWrite( - _string(w["name"]), _indices(w["dependencies"]), frozenset(SymbolId(_string(s)) for s in w["externals"]) + _string(w["name"]), + _indices(w["dependencies"]), + frozenset(SymbolId(_string(s)) for s in w["externals"]), + bool(w["kept"]), ) for w in data["nonlocal_writes"] ), diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index 30e460d..2250c6d 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -332,11 +332,15 @@ class Mutation: @dataclass(frozen=True) class NonlocalWrite: """What the function stores into its ``nonlocal`` ``name``: data of its parameters - and results of external symbols, written back to the caller's variable.""" + and results of external symbols, written back to the caller's variable. ``kept`` + says it may also return without writing it, on a path with no reachable write, so + the caller's variable may keep the value it had. A function with no record for a + ``nonlocal`` never writes it (#258).""" name: str dependencies: Dependencies externals: frozenset[SymbolId] + kept: bool = False @dataclass(frozen=True) @@ -602,14 +606,17 @@ def call(self, call: Call, state: dict[Key, Dep]) -> Dep: return everything | state.get(call.callee, EMPTY) def nonlocal_result(self, instruction: NonlocalResult, state: dict[Key, Dep]) -> Dep: - """What the nested callee left in its nonlocal ``name``, in the caller's terms.""" + """What the nested callee left in its nonlocal ``name``, in the caller's terms: + what it may write, and the value before the call wherever it may leave the name + unwritten, or when what it does is not known.""" + previous = self.deep(instruction.previous, state) call = self.defs.get(instruction.call) if not isinstance(call, Call): - return EMPTY + return previous target = self.graph.target_at(self.name, call.location) if not isinstance(target, KnownFunction) or target.name not in self.table: - return EMPTY + return previous callee = self.table[target.name] arguments = tuple(self.deep(a, state) for a in call.arguments) made = self.defs.get(call.callee) @@ -620,10 +627,9 @@ def nonlocal_result(self, instruction: NonlocalResult, state: dict[Key, Dep]) -> keywords = EMPTY for value in (*call.starred, *(v for _, v in call.keywords)): keywords |= self.deep(value, state) - deps = EMPTY - for write in callee.nonlocal_writes: - if write.name != instruction.name: - continue + written = [write for write in callee.nonlocal_writes if write.name == instruction.name] + deps = previous if not written or any(write.kept for write in written) else EMPTY + for write in written: for index in write.dependencies: deps |= arguments[index] if index < len(arguments) else keywords deps |= Dep(externals=write.externals) @@ -686,6 +692,38 @@ def mapped(indices: Dependencies, cleared: Cleared = ()) -> Dep: def uses(self) -> DefUse: return def_use(self.function) + def may_return_unwritten(self, name: str) -> bool: + """Whether a path from the entry to a return writes no ``nonlocal`` ``name``, + through the blocks that can run. An exception edge leaves a block before its + write, so its handler is reached unwritten.""" + + writes = { + block.id + for block in self.function.blocks + if any(isinstance(i, SetNonlocal) and i.name == name for i in block.instructions) + } + pending = [self.function.entry] + seen: set[BlockId] = set() + while pending: + block_id = pending.pop() + if block_id in seen or not self.reachable(block_id): + continue + seen.add(block_id) + block = self.blocks[block_id] + pending.extend(block.exception_targets) + if block_id in writes: + continue + terminator = block.terminator + if isinstance(terminator, Return): + return True + if isinstance(terminator, Branch): + pending.extend((terminator.then_block, terminator.else_block)) + elif isinstance(terminator, Jump): + pending.append(terminator.target) + elif isinstance(terminator, ForNext): + pending.extend((terminator.body, terminator.exit)) + return False + def runs(self, call: Call) -> bool: """Whether ``call`` can run: its block is reachable by constant propagation.""" @@ -793,9 +831,8 @@ def summarize( mutations=tuple(mutations), side_effects=frozenset(problem.mutated_globals), nonlocal_writes=tuple( - NonlocalWrite(name, deps.parameters, deps.externals) + NonlocalWrite(name, deps.parameters, deps.externals, problem.may_return_unwritten(name)) for name, deps in sorted(problem.nonlocal_stores.items()) - if deps.parameters or deps.externals ), static=static, return_cleared=problem.returns.cleared, diff --git a/src/coretrace_python/ir/lowering.py b/src/coretrace_python/ir/lowering.py index 78d495e..01cde42 100644 --- a/src/coretrace_python/ir/lowering.py +++ b/src/coretrace_python/ir/lowering.py @@ -251,7 +251,8 @@ def write_back_nonlocals(self, node: nodes.Call, call: Value) -> None: return bindings = self.scopes.scope_for(nested).bindings for name in sorted(n for n, b in bindings.items() if b.kind is BindingKind.NONLOCAL): - written = self.emit(NonlocalResult(self.new_value(), node.span, call, name)) + previous = self.expression(nodes.Name(name, node.span)) + written = self.emit(NonlocalResult(self.new_value(), node.span, call, name, previous)) self.store(nodes.Name(name, node.span), written) # ------------------------------------------------------------------ statements diff --git a/src/coretrace_python/ir/model.py b/src/coretrace_python/ir/model.py index d6a67b0..528f77f 100644 --- a/src/coretrace_python/ir/model.py +++ b/src/coretrace_python/ir/model.py @@ -9,7 +9,7 @@ from coretrace_python.source import SourceSpan # Bumped whenever the instruction set or its meaning changes; part of the cache key. -PYIR_SCHEMA_VERSION = 1 +PYIR_SCHEMA_VERSION = 2 @dataclass(frozen=True) @@ -245,12 +245,15 @@ class Yield(Operands): @dataclass(frozen=True) class NonlocalResult(Operands): """The value a nested function left in its ``nonlocal`` ``name`` when ``call`` - returned; the caller stores it back into its own variable.""" + returned; the caller stores it back into its own variable. ``previous`` is the value + that variable held before the call, which it keeps wherever the function may leave + the name unwritten (``NonlocalWrite.kept``).""" result: Value location: SourceSpan call: Value name: str + previous: Value @dataclass(frozen=True) diff --git a/src/coretrace_python/ir/printer.py b/src/coretrace_python/ir/printer.py index ccb2793..58239ec 100644 --- a/src/coretrace_python/ir/printer.py +++ b/src/coretrace_python/ir/printer.py @@ -129,7 +129,10 @@ def _instruction(instruction: Instruction) -> str: suffix = "" if instruction.value is None else f" {_value(instruction.value)}" return f"{_value(instruction.result)} = yield{suffix}" if isinstance(instruction, NonlocalResult): - return f"{_value(instruction.result)} = nonlocal_result {_value(instruction.call)}, {instruction.name!r}" + return ( + f"{_value(instruction.result)} = nonlocal_result {_value(instruction.call)}, {instruction.name!r}" + f" else {_value(instruction.previous)}" + ) if isinstance(instruction, SetAttr): return f"set_attr {_value(instruction.object)}, {instruction.attribute!r}, {_value(instruction.value)}" if isinstance(instruction, SetItem): diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index efca2a1..b45f1dc 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -464,14 +464,17 @@ def constructed_keys(self, call: Call, state: Mapping[Key, Taint]) -> Taint: return taint def nonlocal_result(self, instruction: NonlocalResult, state: dict[Key, Taint]) -> Taint: - """The taint a nested callee left in its nonlocal ``name``, in the caller's terms.""" + """The taint a nested callee left in its nonlocal ``name``, in the caller's terms: + what it may write, and the value before the call wherever it may leave the name + unwritten (``NonlocalWrite.kept``), or when what it does is not known.""" + previous = self.deep(instruction.previous, state) call = self.defs.get(instruction.call) if not isinstance(call, Call): - return Taint.none() + return previous target = self.graph.target_at(self.name, call.location) if not isinstance(target, KnownFunction): - return Taint.none() + return previous summary = self.summaries.summary(target.name) arguments = ( *(self.deep(r, state) for r in (() if summary.static else self.receiver(call, target.name))), @@ -481,10 +484,9 @@ def nonlocal_result(self, instruction: NonlocalResult, state: dict[Key, Taint]) keywords = Taint.none() for value in (*call.starred, *(v for _, v in call.keywords)): keywords = keywords.join(self.deep(value, state)) - result = Taint.none() - for write in summary.nonlocal_writes: - if write.name != instruction.name: - continue + written = [write for write in summary.nonlocal_writes if write.name == instruction.name] + result = previous if not written or any(write.kept for write in written) else Taint.none() + for write in written: for index in sorted(write.dependencies): result = result.join(arguments[index] if index < len(arguments) else keywords) for symbol in sorted(write.externals, key=str): diff --git a/tests/test_nonlocal_kept_value.py b/tests/test_nonlocal_kept_value.py new file mode 100644 index 0000000..768bc0c --- /dev/null +++ b/tests/test_nonlocal_kept_value.py @@ -0,0 +1,116 @@ +"""Review of #258 (PR #259): a nested function that may leave a ``nonlocal`` unwritten +leaves the caller's variable as it was before the call. + +The summary of a nested function says, for each ``nonlocal`` it writes, what it writes +and whether it may also return without writing it (``NonlocalWrite.kept``); no record +means it never writes it. The caller then keeps the value its variable held before the +call wherever the name may be left unwritten, joined with what the function may write: +a write under ``if False:``, on some paths only, after an exception, or no write at +all. Only a write on every path that returns replaces the value. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cache import ProjectCache +from coretrace_python.source import SourceManager + + +def injections(body: str) -> list[tuple[int, str]]: + text = "import os\n\n\n" + body + findings = engine.check(SourceManager().add_source("app.py", text), [engine.BUNDLED_PLUGINS]) + return [(f.span.start_line, f.metadata["verdict"]) for f in findings if f.rule_id == "command-injection"] + + +def nested(write: str, before_call: str = "") -> str: + """``run`` reads input into ``cmd``, defines ``keep`` writing it as ``write`` says, + calls it and runs ``cmd``; the call to ``os.system`` is the last line.""" + + lines = ["def run(flag):", " cmd = input()", " def keep():", " nonlocal cmd"] + lines += [f" {line}" for line in write.splitlines()] + lines += [f" {line}" for line in before_call.splitlines()] + lines += [" keep()", " os.system(cmd)"] + return "\n".join(lines) + "\n" + + +def last_line(body: str) -> int: + return ("import os\n\n\n" + body).count("\n") + + +@pytest.mark.parametrize( + "write", + [ + # The review's cases: the only write cannot run. + "if False:\n cmd = cmd", + "while False:\n cmd = cmd", + "if False:\n cmd = 'ls'", + # No write at all. + "print(cmd)", + # A write on some paths only. + "if flag:\n cmd = 'ls'", + # A write an exception may skip. + "try:\n print(len(cmd))\n cmd = 'ls'\nexcept ValueError:\n pass", + ], +) +def test_a_value_the_nested_function_may_leave_unwritten_keeps_its_taint(write: str) -> None: + body = nested(write) + + assert injections(body) == [(last_line(body), "vulnerability")] + + +@pytest.mark.parametrize( + "write", + [ + "cmd = 'ls'", + "if flag:\n cmd = 'ls'\nelse:\n cmd = 'pwd'", + "cmd = 'ls'\nif flag:\n cmd = 'pwd'", + ], +) +def test_a_safe_value_written_on_every_path_replaces_it(write: str) -> None: + assert injections(nested(write)) == [] + + +def test_what_the_nested_function_writes_still_reaches_the_caller() -> None: + body = ( + "def run():\n cmd = 'ls'\n def keep():\n nonlocal cmd\n cmd = input()\n" + " keep()\n os.system(cmd)\n" + ) + + assert injections(body) == [(last_line(body), "vulnerability")] + + +def test_the_value_kept_is_the_one_before_the_call_not_at_the_definition() -> None: + # Safe when ``keep`` is defined, input when it is called. + body = ( + "def run():\n cmd = 'ls'\n def keep():\n nonlocal cmd\n if False:\n cmd = cmd\n" + " cmd = input()\n keep()\n os.system(cmd)\n" + ) + + assert injections(body) == [(last_line(body), "vulnerability")] + + +def test_the_kept_value_reaches_the_caller_of_the_enclosing_function() -> None: + body = ( + "def run(value):\n cmd = value\n def keep():\n nonlocal cmd\n if False:\n cmd = 'ls'\n" + " keep()\n return cmd\n\n\ndef caller():\n os.system(run(input()))\n" + ) + + assert injections(body) == [(last_line(body), "vulnerability")] + + +def test_the_kept_flag_survives_the_cache(tmp_path: Path) -> None: + root = tmp_path / "project" + (root / "app").mkdir(parents=True) + (root / "app" / "__init__.py").write_text("", encoding="utf-8") + (root / "app" / "main.py").write_text("import os\n\n\n" + nested("if flag:\n cmd = 'ls'"), encoding="utf-8") + cache = ProjectCache(tmp_path / "cache") + + analysis = engine.analyze_project(root, [engine.BUNDLED_PLUGINS], cache=cache) + entry = cache.load(analysis.keys["app.main"]) + + assert entry is not None + assert [(w.name, w.kept) for w in entry.summaries["run.keep"].nonlocal_writes] == [("cmd", True)]