From 717c07498f5413aa8c0ca0516f645a002b8f80a6 Mon Sep 17 00:00:00 2001 From: sebas8866 Date: Fri, 16 Oct 2026 21:38:47 -0400 Subject: [PATCH 1/5] fix: canonicalize external calls in summary fixpoints --- src/coretrace_python/cache.py | 4 +- .../interprocedural/summaries.py | 26 +++- tests/test_summary_external_call_order.py | 113 ++++++++++++++++++ 3 files changed, 141 insertions(+), 2 deletions(-) create mode 100644 tests/test_summary_external_call_order.py diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index a76a4c9..e8fc550 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -42,7 +42,9 @@ from coretrace_python.semantic.symbols import SymbolId from coretrace_python.source import FileLocation, Location, SourceId, SourceSpan -CACHE_FORMAT = 16 +# Cached summaries from earlier versions preserve unstable external-call +# insertion order. They must be recomputed with canonical ordering (#261). +CACHE_FORMAT = 17 @dataclass(frozen=True) diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index 2250c6d..726ccee 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -318,6 +318,30 @@ def named(self) -> Iterable[tuple[str | None, Dependencies, Cleared]]: return ((name, deps, c) for (name, deps), c in _aligned(self.keyword_dependencies, self.keyword_cleared)) +def _canonical_external_calls(calls: Iterable[ExternalCall]) -> tuple[ExternalCall, ...]: + """Order by call identity so recursive summaries reach stable equality. + + A callee's external calls get copied into its callers, potentially in a + different insertion order on each fixpoint round even when the content + is already stable. Comparing unsorted tuples can therefore loop forever. + """ + def span_key(span: SourceSpan | None) -> tuple[str, int, int, int, int]: + if span is None: + return ("", 0, 0, 0, 0) + return ( + span.source_id.value, span.start_line, span.start_column, + span.end_line or 0, span.end_column or 0, + ) + + return tuple(sorted( + calls, + key=lambda call: ( + span_key(call.location), span_key(call.call_site), + span_key(call.invoked_at), call.symbol.canonical_name, + ), + )) + + @dataclass(frozen=True) class Mutation: """What the function stores into ``field`` of its ``parameter``: data of other @@ -826,7 +850,7 @@ def summarize( name, len(function.parameters), problem.returns.parameters, - tuple(problem.external.values()), + _canonical_external_calls(problem.external.values()), return_externals=problem.returns.externals, mutations=tuple(mutations), side_effects=frozenset(problem.mutated_globals), diff --git a/tests/test_summary_external_call_order.py b/tests/test_summary_external_call_order.py new file mode 100644 index 0000000..690eb88 --- /dev/null +++ b/tests/test_summary_external_call_order.py @@ -0,0 +1,113 @@ +"""Regression for issue #261: an external-call cycle must reach a fixpoint. + +The child-process timeout makes the old nontermination observable without +leaving a hanging pytest process on CI. The source is a fixed, local fixture. +""" + +from __future__ import annotations + +import subprocess +import sys +import tempfile +import textwrap +import unittest +from pathlib import Path + + +SOURCE = textwrap.dedent("""\ + def a(x): + b(x) + print(x) + + def b(x): + c(x) + print(x) + + def c(x): + a(x) +""") + + +class ExternalCallOrderTests(unittest.TestCase): + def test_old_order_dependent_summary_cache_is_invalidated(self): + from coretrace_python.cache import CACHE_FORMAT, decode + + self.assertGreater(CACHE_FORMAT, 16) + with self.assertRaisesRegex(ValueError, "unsupported cache format"): + decode({"format": 16}) + + def test_three_function_cycle_converges_with_external_calls(self): + program = ( + "from coretrace_python import engine\n" + "from coretrace_python.frontend import build_hir\n" + "from coretrace_python.source import SourceManager\n" + "from coretrace_python.interprocedural import SummaryAnalysis\n" + f"source = {SOURCE!r}\n" + "table = engine.build_manager(build_hir(SourceManager().add_source('cycle.py', source))).get(SummaryAnalysis)\n" + "for name in ('a', 'b', 'c'):\n" + " calls = table.summary(name).external_calls\n" + " assert any(str(c.symbol) == 'python.builtins.print' for c in calls), name\n" + "print('converged')\n" + ) + try: + proc = subprocess.run( + [sys.executable, "-c", program], capture_output=True, + text=True, timeout=20, check=False, + ) + except subprocess.TimeoutExpired as exc: + self.fail(f"summary fixpoint did not terminate within 20 seconds: {exc}") + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertEqual(proc.stdout.strip(), "converged") + + def test_external_call_order_is_independent_of_insertion_order(self): + from coretrace_python.interprocedural.summaries import ( + ExternalCall, _canonical_external_calls, + ) + from coretrace_python.semantic.symbols import SymbolId + from coretrace_python.source import SourceId, SourceSpan + + span = SourceSpan(SourceId("cycle.py"), 3, 5) + earlier = ExternalCall(SymbolId("python.builtins.print"), (), (), span, None) + later = ExternalCall( + SymbolId("python.builtins.len"), (), (), + SourceSpan(SourceId("cycle.py"), 6, 5), span, + ) + self.assertEqual( + _canonical_external_calls([earlier, later]), + _canonical_external_calls([later, earlier]), + ) + self.assertEqual(_canonical_external_calls([later, earlier]), (earlier, later)) + + def test_public_check_command_finishes_for_recursive_external_calls(self): + with tempfile.TemporaryDirectory() as root: + path = Path(root) / "cycle.py" + path.write_text(SOURCE, encoding="utf-8") + try: + proc = subprocess.run( + [sys.executable, "-m", "coretrace_python", "--check", str(path)], + capture_output=True, text=True, timeout=20, check=False, + ) + except subprocess.TimeoutExpired as exc: + self.fail(f"CLI --check stalled on 3-function call cycle: {exc}") + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + + def test_order_is_stable_across_call_site_and_symbol_ties(self): + from coretrace_python.interprocedural.summaries import ( + ExternalCall, _canonical_external_calls, + ) + from coretrace_python.semantic.symbols import SymbolId + from coretrace_python.source import SourceId, SourceSpan + + location = SourceSpan(SourceId("cycle.py"), 3, 5) + caller = SourceSpan(SourceId("cycle.py"), 6, 5) + first = ExternalCall(SymbolId("python.builtins.len"), (), (), location, None) + second = ExternalCall(SymbolId("python.builtins.print"), (), (), location, None) + third = ExternalCall(SymbolId("python.builtins.len"), (), (), location, caller) + self.assertEqual( + _canonical_external_calls((third, second, first)), + (first, second, third), + ) + + +if __name__ == "__main__": + unittest.main() From e20c149ecf9593ad64d101e7f7720b119aeb2c61 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 20 Oct 2026 00:49:02 +0200 Subject: [PATCH 2/5] test(summaries): cover the cycle shapes and the source order of external calls The cycles of #261 that never ended, three and four functions and ansible's shape with exception handlers, are checked end to end with their finding, in a subprocess with a timeout. A summary lists its external calls in source order, and a caller's summary does not depend on the order of its callee's. --- tests/test_summary_fixpoint.py | 184 +++++++++++++++++++++++++++++++++ 1 file changed, 184 insertions(+) create mode 100644 tests/test_summary_fixpoint.py diff --git a/tests/test_summary_fixpoint.py b/tests/test_summary_fixpoint.py new file mode 100644 index 0000000..5ef402a --- /dev/null +++ b/tests/test_summary_fixpoint.py @@ -0,0 +1,184 @@ +"""Acceptance tests for issue #261: the summary fixpoint of a module terminates. + +A function's summary lists the external calls it reaches, its own and those of the +project functions it calls. Those of a callee used to be copied in the callee's order, so +around a cycle of three functions or more each function's order depended on the order +its callee had in the previous round: the content settled, the order rotated, and the +order-sensitive comparison that ends the fixpoint never held. The analysis then never +returned. A summary now lists its external calls in source order, whatever the order in +which the rounds met them. +""" + +from __future__ import annotations + +import dataclasses +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cfg import CFGAnalysis +from coretrace_python.frontend import build_hir +from coretrace_python.interprocedural import CallGraphAnalysis, SummaryAnalysis +from coretrace_python.interprocedural.summaries import summarize +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.source import SourceManager + +# A converging analysis of these files takes about a second: a run past this is one that +# would not end. +TIMEOUT = 60 + +THREE_FUNCTIONS = """\ +import os + + +def a(x): + b(x) + os.system(x) + + +def b(x): + c(x) + print(x) + + +def c(x): + a(x) + + +def run(): + a(input()) +""" + +FOUR_FUNCTIONS = """\ +import os + + +def a(x): + b(x) + os.system(x) + + +def b(x): + c(x) + print(x) + + +def c(x): + d(x) + + +def d(x): + a(x) + + +def run(): + a(input()) +""" + +# The shape of ansible's ``Role.load``, ``_load_role_data`` and ``_load_dependencies``. +EXCEPTION_HANDLERS = """\ +import os + + +def load(include): + try: + load_data(include) + except RuntimeError: + raise ValueError(include.ds) + + +def load_data(include): + load_dependencies(include) + try: + os.system(include) + except AssertionError: + raise ValueError(include.name) + + +def load_dependencies(include): + load(include) + + +def run(): + load(input()) +""" + + +def checked(tmp_path: Path, source: str) -> list[tuple[str, int, str | None]]: + """The findings of ``--check`` on a project of one module, failing the test instead of + waiting when the analysis does not end.""" + + root = tmp_path / "project" + root.mkdir() + (root / "app.py").write_text(source, encoding="utf-8") + try: + completed = subprocess.run( + [sys.executable, "-m", "coretrace_python", "--check", str(root), "--format", "json"], + capture_output=True, + text=True, + check=False, + timeout=TIMEOUT, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"the analysis did not end within {TIMEOUT} s") + report = json.loads(completed.stdout) + return [(f["rule_id"], f["location"]["line"], f["metadata"].get("verdict")) for f in report["findings"]] + + +@pytest.mark.parametrize( + ("source", "line"), + [(THREE_FUNCTIONS, 19), (FOUR_FUNCTIONS, 23), (EXCEPTION_HANDLERS, 24)], + ids=["three-functions", "four-functions", "exception-handlers"], +) +def test_a_call_cycle_with_external_calls_after_the_recursive_call_is_checked( + tmp_path: Path, source: str, line: int +) -> None: + assert checked(tmp_path, source) == [("command-injection", line, "vulnerability")] + + +CALLEE_FIRST = """\ +import os + + +def run(cmd): + helper(cmd) + os.system(cmd) + + +def helper(cmd): + print(cmd) + os.remove(cmd) +""" + + +def test_a_summary_lists_its_external_calls_in_source_order() -> None: + # ``run`` meets ``helper``'s calls first, at line 5, then its own at line 6. + table = engine.build_manager(build_hir(SourceManager().add_source("app.py", CALLEE_FIRST))).get(SummaryAnalysis) + + assert [(str(c.symbol), c.location.start_line) for c in table.summary("run").external_calls] == [ + ("python.os.system", 6), + ("python.builtins.print", 10), + ("python.os.remove", 11), + ] + + +def test_a_summary_does_not_depend_on_the_order_of_its_callee_s_external_calls() -> None: + manager = engine.build_manager(build_hir(SourceManager().add_source("app.py", CALLEE_FIRST))) + graph = manager.get(CallGraphAnalysis) + summaries = manager.get(SummaryAnalysis) + table = {name: summaries.summary(name) for name in summaries.names} + helper = table["helper"] + reversed_helper = dataclasses.replace(helper, external_calls=helper.external_calls[::-1]) + run = graph.definitions["run"] + + def run_summary(callee): + return summarize( + "run", manager.get(SSAAnalysis, run), manager.get(CFGAnalysis, run), graph, {**table, "helper": callee} + ) + + assert helper.external_calls != reversed_helper.external_calls + assert run_summary(reversed_helper) == run_summary(helper) From 8e7e9d7a541795c8042e71513c36e6e9df83d9a6 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 20 Oct 2026 00:49:02 +0200 Subject: [PATCH 3/5] fix(engine): start an import cycle's fixpoint from empty summaries A call between two modules of an import cycle was a call to an unknown symbol on the first iteration and a project call on the next, so around a cycle of three modules the iteration shifted by one module each time and ran to MAX_PROJECT_ITERATIONS. The members' functions now start from the summaries SummaryAnalysis starts from, shared as starting_summaries. Closes #261. --- src/coretrace_python/engine.py | 16 ++++++- .../interprocedural/__init__.py | 2 + .../interprocedural/summaries.py | 40 ++++++++++------ tests/test_summary_fixpoint.py | 46 ++++++++++++++++++- 4 files changed, 87 insertions(+), 17 deletions(-) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index c7f1648..3291d45 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -96,6 +96,7 @@ discover_sources, name_modules, project_symbol, + starting_summaries, ) from coretrace_python.ir import PYIR_SCHEMA_VERSION from coretrace_python.ir.defuse import DefUseAnalysis @@ -831,7 +832,20 @@ def _analyse_managers( counts = Counter(manager.module.name for manager in managers.values()) names = {name: manager.module.name for name, manager in managers.items() if counts[manager.module.name] == 1} - index = seed + # The members' functions start from the summaries ``SummaryAnalysis`` starts from. + # Left out, a call from one member to another is a call to an unknown symbol on the + # first iteration and a project call on the next: around a cycle of modules, the + # iteration then shifts by one module each time and never settles (#261). + index = SummaryIndex( + { + **seed.summaries, + **{ + project_symbol(module, function): summary + for name, module in names.items() + for function, summary in starting_summaries(managers[name].get(CallGraphAnalysis)).items() + }, + } + ) for manager in managers.values(): manager.provide(ProjectSummaries, index) for _ in range(MAX_PROJECT_ITERATIONS): diff --git a/src/coretrace_python/interprocedural/__init__.py b/src/coretrace_python/interprocedural/__init__.py index fc053b2..e27a800 100644 --- a/src/coretrace_python/interprocedural/__init__.py +++ b/src/coretrace_python/interprocedural/__init__.py @@ -40,6 +40,7 @@ TemplateCallsAnalysis, TemplateFilter, cleared_by, + starting_summaries, ) __all__ = [ @@ -78,4 +79,5 @@ "discover_sources", "name_modules", "project_symbol", + "starting_summaries", ] diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index 726ccee..ecedf09 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -863,6 +863,21 @@ def summarize( ) +def starting_summaries(graph: CallGraph) -> dict[str, FunctionSummary]: + """The summary each function of ``graph`` starts from in a fixpoint: a supported + function reaches nothing yet, the bottom its summary grows from; an unsupported one + passes every parameter to its return value and keeps that summary.""" + + table: dict[str, FunctionSummary] = {} + for name, function in graph.definitions.items(): + count = len(function.parameters) + if name in graph.unsupported: + table[name] = FunctionSummary(name, count, frozenset(range(count)), (), True) + else: + table[name] = FunctionSummary(name, count, NONE, ()) + return table + + class SummaryAnalysis(Analysis[SummaryTable]): name: ClassVar[str] = "interprocedural.summaries" requires: ClassVar[frozenset[AnyAnalysis]] = frozenset( @@ -884,20 +899,17 @@ def compute(cls, ctx: AnalysisContext) -> SummaryTable: project = ctx.get(ProjectSummaries).summaries clearing = ctx.get(ClearingAnalysis) templates = ctx.get(TemplateCallsAnalysis) - table: dict[str, FunctionSummary] = {} - supported: dict[str, tuple[FunctionIR, CFG, HeapFacts, Callable[[BlockId], bool]]] = {} - for name, function in graph.definitions.items(): - if name in graph.unsupported: - count = len(function.parameters) - table[name] = FunctionSummary(name, count, frozenset(range(count)), (), True) - else: - supported[name] = ( - ctx.get(SSAAnalysis, function), - ctx.get(CFGAnalysis, function), - ctx.get(HeapAnalysis, function), - ctx.get(ConstantPropagation, function).reachable, - ) - table[name] = FunctionSummary(name, len(function.parameters), NONE, ()) + table = starting_summaries(graph) + supported: dict[str, tuple[FunctionIR, CFG, HeapFacts, Callable[[BlockId], bool]]] = { + name: ( + ctx.get(SSAAnalysis, function), + ctx.get(CFGAnalysis, function), + ctx.get(HeapAnalysis, function), + ctx.get(ConstantPropagation, function).reachable, + ) + for name, function in graph.definitions.items() + if name not in graph.unsupported + } changed = True while changed: changed = False diff --git a/tests/test_summary_fixpoint.py b/tests/test_summary_fixpoint.py index 5ef402a..7ed97b9 100644 --- a/tests/test_summary_fixpoint.py +++ b/tests/test_summary_fixpoint.py @@ -1,4 +1,5 @@ -"""Acceptance tests for issue #261: the summary fixpoint of a module terminates. +"""Acceptance tests for issue #261: the summary fixpoints terminate, a module's and a +project's. A function's summary lists the external calls it reaches, its own and those of the project functions it calls. Those of a callee used to be copied in the callee's order, so @@ -7,6 +8,10 @@ order-sensitive comparison that ends the fixpoint never held. The analysis then never returned. A summary now lists its external calls in source order, whatever the order in which the rounds met them. + +The modules of an import cycle iterate their summaries together. Their own functions +now start from the summaries a module's fixpoint starts from, so the iteration grows to +its fixpoint instead of running to its bound. """ from __future__ import annotations @@ -15,14 +20,16 @@ import json import subprocess import sys +from collections import Counter from pathlib import Path import pytest from coretrace_python import engine +from coretrace_python.analysis import AnalysisManager from coretrace_python.cfg import CFGAnalysis from coretrace_python.frontend import build_hir -from coretrace_python.interprocedural import CallGraphAnalysis, SummaryAnalysis +from coretrace_python.interprocedural import CallGraphAnalysis, FunctionSummary, SummaryAnalysis from coretrace_python.interprocedural.summaries import summarize from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.source import SourceManager @@ -182,3 +189,38 @@ def run_summary(callee): assert helper.external_calls != reversed_helper.external_calls assert run_summary(reversed_helper) == run_summary(helper) + + +# The modules of an import cycle are analysed together, their summaries iterated until +# the index of the project's summaries stops changing. ``THREE_FUNCTIONS`` across three +# modules: a call from one member to another used to be a call to an unknown symbol on +# the first iteration and a project call on the next, and the iteration never settled +# before its bound. +CYCLE_OF_MODULES = { + "pkg/__init__.py": "", + "pkg/a.py": "import os\n\nfrom pkg import b\n\n\ndef a(x):\n b.b(x)\n os.system(x)\n\n\ndef run():\n a(input())\n", + "pkg/b.py": "from pkg import c\n\n\ndef b(x):\n c.c(x)\n print(x)\n", + "pkg/c.py": "from pkg import a\n\n\ndef c(x):\n a.a(x)\n", +} + + +def test_a_call_cycle_across_modules_converges_before_the_bound(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = tmp_path / "project" + for path, text in CYCLE_OF_MODULES.items(): + (root / path).parent.mkdir(parents=True, exist_ok=True) + (root / path).write_text(text, encoding="utf-8") + extractions: Counter[str] = Counter() + summaries_of = engine._summaries_of + + def counted(manager: AnalysisManager) -> dict[str, FunctionSummary]: + extractions[manager.module.name] += 1 + return summaries_of(manager) + + monkeypatch.setattr(engine, "_summaries_of", counted) + analysis = engine.analyze_project(root, [engine.BUNDLED_PLUGINS]) + + # Once per iteration, and once more for the module's results. + assert max(extractions[name] for name in ("pkg.a", "pkg.b", "pkg.c")) <= engine.MAX_PROJECT_ITERATIONS + assert [ + (f.rule_id, Path(str(f.span.source_id)).name, f.span.start_line, f.metadata["verdict"]) for f in analysis.findings + ] == [("command-injection", "a.py", 12, "vulnerability")] From e5e51de4ef9024fb720a4db15b1d242783fe5e31 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 20 Oct 2026 00:49:09 +0200 Subject: [PATCH 4/5] style(tests): sort the imports of the external-call order tests ruff check . reports I001 three times in tests/test_summary_external_call_order.py, added in #265. This is ruff's own fix, a blank line and one imported name per line; the tests are unchanged. --- tests/test_summary_external_call_order.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/test_summary_external_call_order.py b/tests/test_summary_external_call_order.py index 690eb88..9b60932 100644 --- a/tests/test_summary_external_call_order.py +++ b/tests/test_summary_external_call_order.py @@ -13,7 +13,6 @@ import unittest from pathlib import Path - SOURCE = textwrap.dedent("""\ def a(x): b(x) @@ -61,7 +60,8 @@ def test_three_function_cycle_converges_with_external_calls(self): def test_external_call_order_is_independent_of_insertion_order(self): from coretrace_python.interprocedural.summaries import ( - ExternalCall, _canonical_external_calls, + ExternalCall, + _canonical_external_calls, ) from coretrace_python.semantic.symbols import SymbolId from coretrace_python.source import SourceId, SourceSpan @@ -93,7 +93,8 @@ def test_public_check_command_finishes_for_recursive_external_calls(self): def test_order_is_stable_across_call_site_and_symbol_ties(self): from coretrace_python.interprocedural.summaries import ( - ExternalCall, _canonical_external_calls, + ExternalCall, + _canonical_external_calls, ) from coretrace_python.semantic.symbols import SymbolId from coretrace_python.source import SourceId, SourceSpan From f0cab6c593737359e449109d513fce5e83d7e008 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 20 Oct 2026 05:21:53 +0200 Subject: [PATCH 5/5] fix(engine): iterate an import cycle's summaries until they settle Starting from empty summaries, the iteration stopped at MAX_PROJECT_ITERATIONS published summaries that missed what later iterations add, so a dependency still on its way read as no taint: a parameter returned at the end of a 22-module cycle was lost, and a sink at the end of a 30-module cycle was missed, on main too. The loop now runs until the index settles and the bound is gone; a comment at the loop states why it settles. The cross-module test runs --check in a subprocess with a timeout instead of asserting the bound, and two 30-module cycles are checked. --- src/coretrace_python/engine.py | 13 +++++- tests/test_summary_fixpoint.py | 80 +++++++++++++++++++++++++--------- 2 files changed, 70 insertions(+), 23 deletions(-) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 3291d45..9952e6e 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -185,7 +185,6 @@ _PROJECT_DEPENDANTS: frozenset[AnyAnalysis] = frozenset( a for a in ALL_ANALYSES if a.name in _PROJECT_DEPENDANT_NAMES ) -MAX_PROJECT_ITERATIONS = 20 class ProjectSummariesUpdated(TransformationPass): @@ -848,7 +847,17 @@ def _analyse_managers( ) for manager in managers.values(): manager.provide(ProjectSummaries, index) - for _ in range(MAX_PROJECT_ITERATIONS): + # Iterated until the index settles, with no bound: stopped earlier, summaries that + # started empty would miss what later iterations add, a dependency still on its way + # reading as no taint. It settles because each iteration can only add: the index holds + # the same symbols from the start, and in every round of a module's fixpoint the rules + # that read the index join what they read (a call's dependencies and records, heap + # stores, merges), while those that do not join (a nonlocal record not summarized yet, + # ``kept``, reachability, ``unsupported``, ``static``) depend on the code and the round + # only. Each round's table thus grows with the index, the fixpoint too, in a finite + # domain: the parameters, symbols and call records of the code. A rule reading the + # index without joining would break this, and the loop's termination with it. + while True: updated = SummaryIndex( { **seed.summaries, diff --git a/tests/test_summary_fixpoint.py b/tests/test_summary_fixpoint.py index 7ed97b9..af048d3 100644 --- a/tests/test_summary_fixpoint.py +++ b/tests/test_summary_fixpoint.py @@ -10,8 +10,9 @@ which the rounds met them. The modules of an import cycle iterate their summaries together. Their own functions -now start from the summaries a module's fixpoint starts from, so the iteration grows to -its fixpoint instead of running to its bound. +now start from the summaries a module's fixpoint starts from, and the iteration goes on +until the index settles, however deep the cycle: no bound stops it while a dependency is +still on its way. """ from __future__ import annotations @@ -20,16 +21,15 @@ import json import subprocess import sys -from collections import Counter +from collections.abc import Mapping from pathlib import Path import pytest from coretrace_python import engine -from coretrace_python.analysis import AnalysisManager from coretrace_python.cfg import CFGAnalysis from coretrace_python.frontend import build_hir -from coretrace_python.interprocedural import CallGraphAnalysis, FunctionSummary, SummaryAnalysis +from coretrace_python.interprocedural import CallGraphAnalysis, SummaryAnalysis from coretrace_python.interprocedural.summaries import summarize from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.source import SourceManager @@ -194,8 +194,7 @@ def run_summary(callee): # The modules of an import cycle are analysed together, their summaries iterated until # the index of the project's summaries stops changing. ``THREE_FUNCTIONS`` across three # modules: a call from one member to another used to be a call to an unknown symbol on -# the first iteration and a project call on the next, and the iteration never settled -# before its bound. +# the first iteration and a project call on the next, and the iteration never settled. CYCLE_OF_MODULES = { "pkg/__init__.py": "", "pkg/a.py": "import os\n\nfrom pkg import b\n\n\ndef a(x):\n b.b(x)\n os.system(x)\n\n\ndef run():\n a(input())\n", @@ -204,23 +203,62 @@ def run_summary(callee): } -def test_a_call_cycle_across_modules_converges_before_the_bound(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: +def project_findings(tmp_path: Path, files: Mapping[str, str]) -> list[tuple[str, str, int, str | None]]: + """The findings of ``--check`` on a project of several files, by path in the project, + failing the test instead of waiting when the analysis does not end.""" + root = tmp_path / "project" - for path, text in CYCLE_OF_MODULES.items(): + for path, text in files.items(): (root / path).parent.mkdir(parents=True, exist_ok=True) (root / path).write_text(text, encoding="utf-8") - extractions: Counter[str] = Counter() - summaries_of = engine._summaries_of + try: + completed = subprocess.run( + [sys.executable, "-m", "coretrace_python", "--check", str(root), "--format", "json"], + capture_output=True, + text=True, + check=False, + timeout=TIMEOUT, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"the analysis did not end within {TIMEOUT} s") + report = json.loads(completed.stdout) + return [ + (f["rule_id"], f["location"]["path"], f["location"]["line"], f["metadata"].get("verdict")) + for f in report["findings"] + ] + + +def test_a_call_cycle_across_modules_converges(tmp_path: Path) -> None: + assert project_findings(tmp_path, CYCLE_OF_MODULES) == [("command-injection", "pkg/a.py", 12, "vulnerability")] + + +# Starting from empty summaries, an index that has not settled yet misses what later +# iterations add: stopped there, a dependency still on its way would read as no taint. +# A cycle deeper than the 20 iterations the loop used to stop at. +DEPTH = 30 + + +def deep_cycle(last: str, caller: str) -> dict[str, str]: + """``DEPTH`` modules in an import cycle, each ``m.f`` returning ``m.f(x)``, + the last doing ``last`` with ``x`` first; ``pkg/app.py`` holds ``caller``.""" + + files = {"pkg/__init__.py": "", "pkg/app.py": caller} + for i in range(DEPTH): + following = (i + 1) % DEPTH + own = last if i == DEPTH - 1 else "" + files[f"pkg/m{i}.py"] = f"import os\n\nfrom pkg import m{following}\n\n\ndef f(x):\n{own} return m{following}.f(x)\n" + return files + + +def test_a_parameter_returned_at_the_end_of_a_deep_cycle_reaches_the_caller(tmp_path: Path) -> None: + caller = "import os\n\nfrom pkg import m0\n\n\ndef run():\n os.system(m0.f(input()))\n" + findings = project_findings(tmp_path, deep_cycle(" if x:\n return x\n", caller)) + + assert findings == [("command-injection", "pkg/app.py", 7, "vulnerability")] - def counted(manager: AnalysisManager) -> dict[str, FunctionSummary]: - extractions[manager.module.name] += 1 - return summaries_of(manager) - monkeypatch.setattr(engine, "_summaries_of", counted) - analysis = engine.analyze_project(root, [engine.BUNDLED_PLUGINS]) +def test_a_sink_at_the_end_of_a_deep_cycle_is_reported_at_the_caller(tmp_path: Path) -> None: + caller = "from pkg import m0\n\n\ndef run():\n m0.f(input())\n" + findings = project_findings(tmp_path, deep_cycle(" if x:\n os.system(x)\n", caller)) - # Once per iteration, and once more for the module's results. - assert max(extractions[name] for name in ("pkg.a", "pkg.b", "pkg.c")) <= engine.MAX_PROJECT_ITERATIONS - assert [ - (f.rule_id, Path(str(f.span.source_id)).name, f.span.start_line, f.metadata["verdict"]) for f in analysis.findings - ] == [("command-injection", "a.py", 12, "vulnerability")] + assert findings == [("command-injection", "pkg/app.py", 5, "vulnerability")]