From 911dcf68fdd2d43f5fd46a0171e7a5af3451b681 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 30 Sep 2026 10:16:16 +0200 Subject: [PATCH 1/9] test: specify the present form of argument conditions --- tests/test_argument_present_conditions.py | 205 ++++++++++++++++++++++ 1 file changed, 205 insertions(+) create mode 100644 tests/test_argument_present_conditions.py diff --git a/tests/test_argument_present_conditions.py b/tests/test_argument_present_conditions.py new file mode 100644 index 0000000..7abad45 --- /dev/null +++ b/tests/test_argument_present_conditions.py @@ -0,0 +1,205 @@ +"""An advisory condition may require an argument to be passed, whatever its value. + +Several aiohttp advisories ride on an argument the project passes: a ``\\r`` in +``web.Response(reason=...)`` splits the status line whatever the value denotes, so a +call giving ``reason`` — positionally, by keyword, or through a folded ``**`` literal — +meets the condition, a call without it is ruled out (``(reason absent)``) and the +project stays ``not_affected``, and a call the engine cannot read (``*args``, an unknown +``**mapping``) leaves it pending review. The form is exclusive with ``values`` and +``default``: the argument being passed is the whole condition. +""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import pytest + +from coretrace_python import engine +from coretrace_python.dependency import ( + Advisory, + AdvisoryEntryPoint, + AdvisoryFileError, + Condition, + dump_advisories, + load_advisories, + render_vex, +) +from coretrace_python.dependency.correlation import check_conditions +from coretrace_python.engine import ProjectAnalysis +from coretrace_python.findings import Finding, Severity +from coretrace_python.semantic.symbols import SymbolId + +PLUGINS = engine.BUNDLED_PLUGINS +NOW = datetime(2026, 9, 30, 12, 0, tzinfo=UTC) + +RESPONSE = "python.aiohttp.web.Response" +TEXT = "reason is passed" +CONDITION = Condition("argument", TEXT, argument="reason", position=1, present=True) +ADVISORY = Advisory( + "CVE-2026-34519", + "aiohttp", + "<3.13.4", + "a carriage return in the reason splits the status line", + Severity.HIGH, + entry_points=(AdvisoryEntryPoint(SymbolId(RESPONSE), "the reason is written into the status line", (CONDITION,)),), + modules=("aiohttp",), +) + +HANDLERS = "from aiohttp import web\n\n\n" +ISSUE = HANDLERS + ( + "async def status(request: web.Request) -> web.Response:\n" + ' return web.Response(status=400, reason=request.query["r"])\n' + "\n\n" + "async def plain(request: web.Request) -> web.Response:\n" + ' return web.Response(text="ok")\n' +) +PLAIN = HANDLERS + 'async def plain(request: web.Request) -> web.Response:\n return web.Response(text="ok")\n' + +# The lock file shows no other package requires aiohttp, so a project whose calls are +# all ruled out can reach ``not_affected`` in the VEX document. +LOCK = """version = 1 + +[[package]] +name = "app" +version = "0.1.0" +source = { virtual = "." } +dependencies = [{ name = "aiohttp" }] + +[[package]] +name = "aiohttp" +version = "3.13.2" +source = { registry = "https://pypi.org/simple" } +""" + + +def analyse(root: Path, source: str) -> ProjectAnalysis: + for relative, text in {"requirements.txt": "aiohttp==3.13.2\n", "uv.lock": LOCK, "app.py": source}.items(): + (root / relative).write_text(text, encoding="utf-8") + (root / "advisories.json").write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + return engine.analyze_project(root, [PLUGINS]) + + +def of_rule(analysis: ProjectAnalysis, rule_id: str) -> list[Finding]: + return sorted((f for f in analysis.findings if f.rule_id == rule_id), key=lambda f: f.span.start_line) + + +def statement(analysis: ProjectAnalysis, root: Path) -> dict[str, Any]: + evidence = (*analysis.findings, *analysis.suppressed, *analysis.accepted) + document = render_vex( + analysis.dependencies, analysis.advisories, evidence, analysis.coverage, root, "coretrace", "0.6.0", NOW + ) + return next(s for s in json.loads(document)["statements"] if s["vulnerability"]["name"] == ADVISORY.id) + + +def handler(call: str) -> str: + return HANDLERS + f"async def status(request: web.Request) -> web.Response:\n return web.Response({call})\n" + + +# --------------------------------------------------------------------------- the reproduction + + +def test_a_call_passing_the_argument_meets_the_condition_and_is_exploitable(tmp_path: Path) -> None: + analysis = analyse(tmp_path, ISSUE) + + (found,) = of_rule(analysis, "exploitable-vulnerability") + assert found.span.start_line == 5 + assert found.metadata["symbol"] == RESPONSE + assert found.metadata["conditions_met"] == TEXT + assert "conditions_pending_review" not in found.metadata + assert statement(analysis, tmp_path)["status"] == "affected" + + +def test_a_call_without_the_argument_is_ruled_out_and_the_project_not_affected(tmp_path: Path) -> None: + ruled_out = f"app:5 {RESPONSE}(reason absent)" + + analysis = analyse(tmp_path, PLAIN) + + assert of_rule(analysis, "reachable-vulnerability") == [] + assert of_rule(analysis, "exploitable-vulnerability") == [] + (declared,) = of_rule(analysis, "vulnerable-dependency") + assert declared.metadata["level"] == "imported" + assert declared.metadata["ruled_out"] == ruled_out + found = statement(analysis, tmp_path) + assert found["status"] == "not_affected" + assert found["impact_statement"].endswith(f" Calls ruled out by their arguments: {ruled_out}.") + + +# --------------------------------------------------------------------------- how the argument is passed + + +@pytest.mark.parametrize( + "call", + [ + 'status=400, reason=request.query["r"]', + 'b"", request.query["r"]', + '**{"reason": request.query["r"]}', + ], + ids=["keyword", "positional", "folded-literal"], +) +def test_the_argument_counts_however_the_call_gives_it(tmp_path: Path, call: str) -> None: + analysis = analyse(tmp_path, handler(call)) + + (found,) = of_rule(analysis, "exploitable-vulnerability") + assert found.metadata["symbol"] == RESPONSE + assert found.metadata["conditions_met"] == TEXT + assert "conditions_pending_review" not in found.metadata + + +@pytest.mark.parametrize("call", ["**options", "*rest"], ids=["unknown-mapping", "starred"]) +def test_a_call_the_engine_cannot_read_leaves_the_condition_pending(tmp_path: Path, call: str) -> None: + source = f"from aiohttp import web\n\ndef build(options, rest):\n return web.Response({call})\n" + + analysis = analyse(tmp_path, source) + + (reached,) = of_rule(analysis, "reachable-vulnerability") + assert reached.metadata["symbol"] == RESPONSE + assert reached.metadata["conditions_pending_review"] == TEXT + assert "conditions_met" not in reached.metadata + assert of_rule(analysis, "exploitable-vulnerability") == [] + + +def test_without_call_arguments_the_condition_stays_pending() -> None: + check = check_conditions(ADVISORY.entry_points[0], None) + + assert check.met == () + assert check.pending == (CONDITION,) + assert check.contradicted is None + + +# --------------------------------------------------------------------------- advisory files + + +def test_the_condition_round_trips_through_advisory_files(tmp_path: Path) -> None: + path = tmp_path / "advisories.json" + path.write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + + assert load_advisories(path) == (ADVISORY,) + (entry,) = json.loads(path.read_text())["advisories"][0]["entry_points"] + assert entry["conditions"] == [ + {"kind": "argument", "text": TEXT, "argument": "reason", "position": 1, "present": True} + ] + + +@pytest.mark.parametrize( + "condition", + [ + {"kind": "argument", "text": TEXT, "argument": "reason", "present": True, "values": ["'x'"]}, + {"kind": "argument", "text": TEXT, "argument": "reason", "present": True, "default": True}, + {"kind": "argument", "text": TEXT, "present": True}, + {"kind": "semantic", "text": TEXT, "present": True}, + {"kind": "argument", "text": TEXT, "argument": "reason", "present": "yes"}, + ], + ids=["values", "default", "no-argument", "not-argument-kind", "not-boolean"], +) +def test_a_malformed_present_condition_is_rejected(tmp_path: Path, condition: dict[str, Any]) -> None: + document = json.loads(dump_advisories((ADVISORY,))) + document["advisories"][0]["entry_points"][0]["conditions"] = [condition] + path = tmp_path / "advisories.json" + path.write_text(json.dumps(document), encoding="utf-8") + + with pytest.raises(AdvisoryFileError, match="present"): + load_advisories(path) From 887b3ee3b086c955a7ab1e7b09fc91cc99141bbc Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 30 Sep 2026 10:16:16 +0200 Subject: [PATCH 2/9] feat: decide an argument condition on the argument being passed Add Condition.present: the argument being given at all, whatever its value, is the condition, exclusive with values and default. A call giving it, positionally, by keyword or through a folded ** literal, meets it even when the value is unknown; a call surely without it is ruled out (reason absent); *args or an unreadable **mapping leaves it pending review. Closes #195. --- CHANGELOG.md | 6 ++++++ src/coretrace_python/dependency/advisories.py | 13 +++++++++++++ src/coretrace_python/dependency/correlation.py | 10 +++++++++- src/coretrace_python/dependency/graph.py | 5 ++++- 4 files changed, 32 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 935943e..c065fb2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased + +### Advisories + +- An `argument` condition may carry `present: true`: the argument being passed at all, whatever its value, is the condition, as a `\r` in `web.Response(reason=…)` needs `reason` given, constant or not. It names the argument by keyword or position as today and excludes `values` and `default`. A call giving the argument — positionally, by keyword, or through a `**` literal with constant keys, folded into the keywords — meets it even when the value is unknown, so attacker input there makes the call exploitable; a call surely not giving it is ruled out (`app:5 python.aiohttp.web.Response(reason absent)`) and a project passing only such calls stays `not_affected`; a call unpacking `*args` or a `**mapping` whose keys are not written leaves it pending review (#195). + ## 0.20.0 (2026-09-30) ### Advisories diff --git a/src/coretrace_python/dependency/advisories.py b/src/coretrace_python/dependency/advisories.py index b0bab34..9a615e3 100644 --- a/src/coretrace_python/dependency/advisories.py +++ b/src/coretrace_python/dependency/advisories.py @@ -187,6 +187,8 @@ def _condition_entry(condition: Condition) -> dict[str, Any]: entry["position"] = condition.position if condition.default: entry["default"] = True + if condition.present: + entry["present"] = True return entry @@ -251,6 +253,9 @@ def _condition(entry: Mapping[str, Any]) -> Condition: default = entry.get("default", False) if not isinstance(default, bool): raise TypeError(f"condition default must be true or false, got {default!r}") + present = entry.get("present", False) + if not isinstance(present, bool): + raise TypeError(f"condition present must be true or false, got {present!r}") condition = Condition( str(entry["kind"]), str(entry["text"]), @@ -258,7 +263,15 @@ def _condition(entry: Mapping[str, Any]) -> Condition: tuple(str(v) for v in entry.get("values") or []), position, default, + present, ) + if condition.present and ( + condition.kind != "argument" + or (condition.argument is None and condition.position is None) + or condition.values + or condition.default + ): + raise ValueError("a present condition names the argument that must be passed, by keyword or position, and nothing else") if condition.kind == "host" and ( (condition.argument is None and condition.position is None) or condition.values or condition.default ): diff --git a/src/coretrace_python/dependency/correlation.py b/src/coretrace_python/dependency/correlation.py index 69d0130..6b552d6 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/src/coretrace_python/dependency/correlation.py @@ -75,7 +75,9 @@ def check_conditions( controls, and the condition stays pending with that uncertainty. A ``keyword_name`` condition is met when the attacker's input is passed as the keys of a ``**`` mapping, pending when the call expands a mapping whose keys are not established, and - contradicted by a call writing every keyword name.""" + contradicted by a call writing every keyword name. An ``argument`` condition with + ``present`` is met when the call gives the argument, whatever it denotes, and + contradicted when the call surely does not give it.""" if entry is None: return ConditionCheck() @@ -105,6 +107,12 @@ def check_conditions( if given is None: pending.append(condition) continue + if condition.present: + if given[0]: + met.append(condition) + else: + return ConditionCheck(tuple(met), tuple(pending), condition, None) + continue explicit, value = given if explicit and value is None: pending.append(condition) diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index ed4af09..e1360db 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -151,7 +151,9 @@ class Condition: ``argument`` condition names the argument by keyword, and by ``position`` when it may be passed positionally, and lists the ``values`` that satisfy it — symbols (``python.yaml.FullLoader``) or constants as Python writes them (``True``); - ``default`` says an absent argument means a vulnerable value. A ``host`` condition + ``default`` says an absent argument means a vulnerable value; ``present`` says the + argument being passed at all, whatever its value, is the condition, so it excludes + ``values`` and ``default``. A ``host`` condition says the attacker must choose the host of the URL passed as ``argument`` (or at ``position``); the engine decides it from what the URL's constant text proves. A ``keyword_name`` condition says the attacker must choose a keyword name of the call, @@ -165,6 +167,7 @@ class Condition: values: tuple[str, ...] = () position: int | None = None default: bool = False + present: bool = False @property def checkable(self) -> bool: From 8f769e9ffb80ab5080e02350379e18a59fd58269 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 30 Sep 2026 10:16:16 +0200 Subject: [PATCH 3/9] docs: describe the present form of argument conditions --- docs/plugins.md | 3 ++- docs/usage.md | 10 ++++++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/docs/plugins.md b/docs/plugins.md index 8816b8d..d64cf63 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -195,7 +195,8 @@ A model plugin may also carry `advisories`, a tuple of `Advisory` values symbols, as the shipped `sample-advisories` plugin does. Requirements matching them are reported, calls to the affected symbols become reachable vulnerabilities and tainted calls become exploitable ones. An advisory's entry points carry `conditions`, of kind -`argument`, `host`, `semantic` or `keyword_name`, and `attacker_arguments`, decided as +`argument` (on the value, or with `present` on the argument being passed at all), +`host`, `semantic` or `keyword_name`, and `attacker_arguments`, decided as the [usage guide](usage.md) describes. ### Reclassifying findings: `ProjectPlugin.refine` diff --git a/docs/usage.md b/docs/usage.md index e4271c3..dadaa5e 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -471,6 +471,16 @@ a call passing a variable, or unpacking `*args` or `**kwargs`, leaves the condit pending review. A `semantic` condition — the document's tags, the platform, the template's origin — cannot be decided and is always pending review, never dropped. +An `argument` condition with `"present": true` says the argument being passed at all, +whatever its value, is the condition, as a `\r` in `web.Response(reason=…)` needs +`reason` given, constant or not: `{"kind": "argument", "text": "…", "argument": +"reason", "position": 1, "present": true}`, exclusive with `values` and `default`. A +call giving the argument — positionally, by keyword, or through a `**` literal with +constant keys, which is folded into the keywords — meets it even when the value is +unknown; a call surely not giving it does not reach the vulnerability and is ruled out +(`(reason absent)`); a call unpacking `*args` or a `**mapping` whose keys the engine +cannot read leaves it pending review. + A `host` condition says the attacker must choose the host of the URL passed as `argument` (or at `position`), as credentials leaking to the host of a crafted URL need: `{"kind": "host", "text": "…", "argument": "url", "position": 0}`. It is From 92ede92260420a72b8f2155eb36c1d97bda336b3 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 30 Sep 2026 10:36:20 +0200 Subject: [PATCH 4/9] test: specify suffix-matched curated entry points Cover issue #197 before the implementation: the reproduction (Item.objects.annotate through a manager the engine cannot type), a declared queryset chain, an undeclared chain and a non-project receiver that must not match, ruled-out and pending conditions through the suffix, the advisories.json round trip with validation, and the ModelTable and SuffixIndex unit behaviour. --- tests/test_entry_point_suffixes.py | 259 +++++++++++++++++++++++++++++ 1 file changed, 259 insertions(+) create mode 100644 tests/test_entry_point_suffixes.py diff --git a/tests/test_entry_point_suffixes.py b/tests/test_entry_point_suffixes.py new file mode 100644 index 0000000..b4c93be --- /dev/null +++ b/tests/test_entry_point_suffixes.py @@ -0,0 +1,259 @@ +"""A curated entry point matched by a suffix of the derived symbol. + +A curated advisory names its entry points by canonical symbol, +``python.django.db.models.QuerySet.annotate``, but a project reaches the method through +a manager the engine cannot type: ``Item.objects.annotate(...)`` derives +``python.app.models.Item.objects.annotate``, the project's own symbol, and the entry +point never matches. An entry point may therefore declare ``suffixes``: a symbol of the +project's own modules ending with a dot plus the suffix reaches the entry point, as a +``SuffixSink`` matches a sink. A queryset chain is declared explicitly +(``objects.filter.annotate``); another package's symbol never matches; the finding names +the advisory's symbol as ``entry_point`` and the derived one as ``symbol``. +""" + +from __future__ import annotations + +import json +from dataclasses import replace +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import pytest + +from coretrace_python import engine +from coretrace_python.dependency import ( + Advisory, + AdvisoryEntryPoint, + AdvisoryFileError, + Condition, + dump_advisories, + load_advisories, + render_vex, +) +from coretrace_python.dependency.correlation import affected_entries, suffix_index +from coretrace_python.engine import ProjectAnalysis +from coretrace_python.findings import Finding, Severity +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import ModelTable, Sink, SuffixSink, TaintKind + +PLUGINS = engine.BUNDLED_PLUGINS +NOW = datetime(2026, 9, 30, 12, 0, tzinfo=UTC) + +QUERYSET = "python.django.db.models.QuerySet.annotate" +DERIVED = "python.app.models.Item.objects.annotate" +TEXT = "the alias is a keyword name the attacker chooses" +CONDITION = Condition("keyword_name", TEXT) + + +def advisory(*suffixes: str) -> Advisory: + return Advisory( + "CVE-2025-57833", + "django", + "<4.2.24 || >=5.0,<5.1.12 || >=5.2,<5.2.6", + "FilteredRelation alias through annotate()/alias() allows SQL injection", + Severity.HIGH, + entry_points=( + AdvisoryEntryPoint( + SymbolId(QUERYSET), + "FilteredRelation alias through annotate()/alias()", + (CONDITION,), + suffixes=suffixes, + ), + ), + modules=("django",), + ) + + +ADVISORY = advisory("objects.annotate", "objects.filter.annotate") + +LOCK = """version = 1 + +[[package]] +name = "app" +version = "0.1.0" +source = { virtual = "." } +dependencies = [{ name = "django" }] + +[[package]] +name = "django" +version = "5.2.5" +source = { registry = "https://pypi.org/simple" } +""" +MODELS = "from django.db import models\n\n\nclass Item(models.Model):\n pass\n" +VIEWS = "from django.db.models import F\nfrom django.http import HttpRequest\n\nfrom .models import Item\n\n\n" +ISSUE = VIEWS + ( + "def listing(request: HttpRequest):\n" + ' return Item.objects.annotate(**{request.GET["alias"]: F("id")})\n' +) +CHAIN = VIEWS + ( + "def listing(request: HttpRequest):\n" + ' return Item.objects.filter(name="x").annotate(**{request.GET["alias"]: F("id")})\n' +) +OTHER = ( + "import other_lib\n" + "from django.http import HttpRequest\n\n\n" + "def listing(request: HttpRequest):\n" + ' return other_lib.objects.annotate(**{request.GET["alias"]: 1})\n' +) +FIXED = VIEWS + 'def fixed(request: HttpRequest):\n return Item.objects.annotate(**{"total": F("id")})\n' +PENDING = VIEWS + ( + "def options():\n return {}\n\n\n" + "def listing(request: HttpRequest):\n opts = options()\n return Item.objects.annotate(**opts)\n" +) +RULED_OUT = f"app.views:8 {DERIVED}(keywords=total)" + + +def analyse(root: Path, views: str, curated: Advisory = ADVISORY) -> ProjectAnalysis: + files = {"uv.lock": LOCK, "app/__init__.py": "", "app/models.py": MODELS, "app/views.py": views} + for relative, text in files.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + (root / "advisories.json").write_text(dump_advisories((curated,)), encoding="utf-8") + return engine.analyze_project(root, [PLUGINS]) + + +def of_rule(analysis: ProjectAnalysis, rule_id: str) -> list[Finding]: + return sorted((f for f in analysis.findings if f.rule_id == rule_id), key=lambda f: f.span.start_line) + + +def statement(analysis: ProjectAnalysis, root: Path) -> dict[str, Any]: + evidence = (*analysis.findings, *analysis.suppressed, *analysis.accepted) + document = render_vex( + analysis.dependencies, analysis.advisories, evidence, analysis.coverage, root, "coretrace", "0.6.0", NOW + ) + return next(s for s in json.loads(document)["statements"] if s["vulnerability"]["name"] == ADVISORY.id) + + +# --------------------------------------------------------------------------- the reproduction + + +def test_a_suffix_matches_the_receiver_the_engine_cannot_type(tmp_path: Path) -> None: + analysis = analyse(tmp_path, ISSUE) + + (found,) = of_rule(analysis, "exploitable-vulnerability") + assert found.metadata["symbol"] == DERIVED + assert found.metadata["entry_point"] == QUERYSET + assert found.metadata["conditions_met"] == TEXT + assert statement(analysis, tmp_path)["status"] == "affected" + + +def test_a_declared_queryset_chain_matches_through_its_own_suffix(tmp_path: Path) -> None: + analysis = analyse(tmp_path, CHAIN) + + (found,) = of_rule(analysis, "exploitable-vulnerability") + assert found.metadata["symbol"].endswith("objects.filter.annotate") + assert found.metadata["entry_point"] == QUERYSET + assert statement(analysis, tmp_path)["status"] == "affected" + + +def test_a_chain_not_declared_does_not_match(tmp_path: Path) -> None: + analysis = analyse(tmp_path, CHAIN, advisory("objects.annotate")) + + assert of_rule(analysis, "reachable-vulnerability") == [] + assert of_rule(analysis, "exploitable-vulnerability") == [] + + +def test_another_packages_receiver_never_matches(tmp_path: Path) -> None: + analysis = analyse(tmp_path, OTHER) + + assert of_rule(analysis, "reachable-vulnerability") == [] + assert of_rule(analysis, "exploitable-vulnerability") == [] + (declared,) = of_rule(analysis, "vulnerable-dependency") + assert declared.metadata["advisory"] == ADVISORY.id + + +# --------------------------------------------------------------------------- conditions through the suffix + + +def test_constant_keys_through_the_suffix_rule_the_call_out(tmp_path: Path) -> None: + analysis = analyse(tmp_path, FIXED) + + assert of_rule(analysis, "reachable-vulnerability") == [] + assert of_rule(analysis, "exploitable-vulnerability") == [] + (declared,) = of_rule(analysis, "vulnerable-dependency") + assert declared.metadata["ruled_out"] == RULED_OUT + assert statement(analysis, tmp_path)["status"] == "not_affected" + + +def test_an_unknown_mapping_reaches_with_the_condition_pending(tmp_path: Path) -> None: + analysis = analyse(tmp_path, PENDING) + + (reached,) = of_rule(analysis, "reachable-vulnerability") + assert reached.metadata["symbol"] == DERIVED + assert reached.metadata["entry_point"] == QUERYSET + assert reached.metadata["conditions_pending_review"] == TEXT + assert of_rule(analysis, "exploitable-vulnerability") == [] + + +# --------------------------------------------------------------------------- advisory files + + +def test_suffixes_round_trip_through_advisory_files(tmp_path: Path) -> None: + path = tmp_path / "advisories.json" + path.write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + + assert load_advisories(path) == (ADVISORY,) + (entry,) = json.loads(path.read_text())["advisories"][0]["entry_points"] + assert entry["suffixes"] == ["objects.annotate", "objects.filter.annotate"] + + +def test_an_entry_point_without_suffixes_dumps_no_suffixes_key() -> None: + (entry,) = json.loads(dump_advisories((advisory(),)))["advisories"][0]["entry_points"] + assert "suffixes" not in entry + + +@pytest.mark.parametrize( + "value", + [42, "", "annotate", ".objects.annotate", "objects..annotate", "objects.annotate."], + ids=["not-a-string", "empty", "one-name", "leading-dot", "empty-segment", "trailing-dot"], +) +def test_an_invalid_suffix_is_rejected(tmp_path: Path, value: Any) -> None: + document = json.loads(dump_advisories((ADVISORY,))) + document["advisories"][0]["entry_points"][0]["suffixes"] = [value] + path = tmp_path / "advisories.json" + path.write_text(json.dumps(document), encoding="utf-8") + + with pytest.raises(AdvisoryFileError): + load_advisories(path) + + +# --------------------------------------------------------------------------- the shared matching machinery + + +def test_an_exact_advisory_sink_composes_with_a_suffix_sink() -> None: + table = ModelTable((), (), (), suffix_sinks=(SuffixSink("objects.extra", TaintKind.SQL),)) + extended = table.extended(Sink(SymbolId("python.app.models.Item.objects.extra"), TaintKind.ADVISORY)) + + found = extended.sink(SymbolId("python.app.models.Item.objects.extra")) + assert found is not None and found.kinds == TaintKind.SQL | TaintKind.ADVISORY + + +def test_extended_takes_suffix_sinks_that_resolve_derived_symbols() -> None: + table = ModelTable((), (), ()).extended(suffixes=(SuffixSink("objects.annotate", TaintKind.ADVISORY),)) + + found = table.sink(SymbolId(DERIVED)) + assert found is not None and found.kinds == TaintKind.ADVISORY + + +def test_a_suffix_matches_project_symbols_on_dot_boundaries_only() -> None: + affected = {SymbolId(QUERYSET): (ADVISORY,)} + index = suffix_index(affected, ("app", "app.models", "app.views")) + + (match,) = index.matches(SymbolId(DERIVED)) + assert match == (ADVISORY, ADVISORY.entry_points[0]) + assert index.matches(SymbolId("python.app.models.Item.myobjects.annotate")) == () + assert index.matches(SymbolId("python.other_lib.objects.annotate")) == () + + +def test_two_advisories_declaring_the_same_suffix_both_match() -> None: + other = replace(advisory("objects.annotate"), id="CVE-2026-0001") + affected = {SymbolId(QUERYSET): (ADVISORY, other)} + index = suffix_index(affected, ("app",)) + + found = affected_entries(SymbolId(DERIVED), affected, index) + assert [(a.id, None if e is None else str(e.symbol)) for a, e in found] == [ + (ADVISORY.id, QUERYSET), + (other.id, QUERYSET), + ] From 9f9c37706d90269098ded7bf6c0d3f47a8b8e7ae Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 30 Sep 2026 10:36:20 +0200 Subject: [PATCH 5/9] feat: match curated entry points by suffix for untypeable receivers An advisory entry point may declare suffixes: a derived symbol of the project's own modules ending with a dot plus the suffix reaches the entry point, as a SuffixSink matches a sink. Suffixes load, dump and validate in advisories.json (at least two dot-separated names); SuffixIndex and affected_entries resolve exact and suffix matches for correlate(), the reachable-vulnerability detector and the vulnerable-dependency exclusions; advisory_suffix_sinks feeds the taint engine, and findings keep the advisory's symbol as entry_point and the derived one as symbol. ModelTable.sink now merges an exact sink with every matching suffix sink instead of returning the first, so an advisory sink on a derived symbol composes with objects.extra's SQL kind. No cache-key change: the advisory repr already carries the suffixes, and the project-prefix test can only flip for a module when the prefix module enters or leaves its transitive import closure, which module_keys already keys on. Closes #197. --- CHANGELOG.md | 6 ++ .../reachable_vulnerability.py | 30 ++++--- .../vulnerable_dependency.py | 13 ++- src/coretrace_python/dependency/advisories.py | 15 +++- .../dependency/correlation.py | 82 +++++++++++++++++-- src/coretrace_python/dependency/graph.py | 8 +- src/coretrace_python/engine.py | 32 ++++++-- src/coretrace_python/taint/models.py | 17 ++-- 8 files changed, 163 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 935943e..81bd857 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased + +### Advisories + +- A curated entry point may declare `suffixes`, for receivers the engine cannot type: `Item.objects.annotate(...)` derives the project's own `python.app.models.Item.objects.annotate`, which the canonical `python.django.db.models.QuerySet.annotate` entry point never matches. A suffix matches any symbol of the project's own modules ending with a dot plus the suffix, on dot boundaries as a `SuffixSink` matches, never another package's symbol; a queryset chain is declared explicitly (`objects.filter.annotate`), and a suffix is at least two dot-separated names. The finding keeps the advisory's symbol as `entry_point` and the derived one as `symbol`; conditions, `attacker_arguments` and the ruled-out evidence of contradicted calls apply through a suffix as through the exact symbol. An exact advisory sink whose symbol a bundled suffix sink also matches (`objects.extra`) now composes its kinds with the suffix sink's instead of shadowing them (#197). + ## 0.20.0 (2026-09-30) ### Advisories diff --git a/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py b/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py index be80c3e..e16c871 100644 --- a/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py +++ b/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py @@ -8,12 +8,14 @@ from typing import ClassVar from coretrace_python.analysis import AnyAnalysis -from coretrace_python.dependency import Advisory, DependencyAnalysis +from coretrace_python.dependency import Advisory, AdvisoryEntryPoint, DependencyAnalysis from coretrace_python.dependency.correlation import ( ConditionCheck, + affected_entries, affected_symbols, check_conditions, evidence, + suffix_index, ) from coretrace_python.findings import Confidence, Finding from coretrace_python.interprocedural import CallGraphAnalysis, ExternalSymbol @@ -30,6 +32,7 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: affected = affected_symbols(ctx.dependencies, ctx.advisories) if not affected: return () + suffixes = suffix_index(affected, ctx.modules) findings: list[Finding] = [] for module in ctx.modules: graph = ctx.call_graph(module) @@ -37,30 +40,28 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: for site in graph.sites(function): if not isinstance(site.target, ExternalSymbol): continue - for advisory in affected.get(site.target.symbol, ()): - entry = advisory.entry_point(site.target.symbol) + for advisory, entry in affected_entries(site.target.symbol, affected, suffixes): if entry is not None and entry.read: continue # a call reads its callee first: reported with the reads check = check_conditions(entry, site.arguments) if check.contradicted is not None: continue - findings.append(_reached(advisory, site.target.symbol, site.location, function, check)) + findings.append(_reached(advisory, site.target.symbol, site.location, function, check, entry)) # Reading ``request.form.get`` reads ``request.form``: one finding per entry # point and function, where the function first reads it. reported: set[tuple[Advisory, SymbolId]] = set() for read in graph.reads(function): for symbol in _read_through(read.symbol): - for advisory in affected.get(symbol, ()): - entry = advisory.entry_point(symbol) + for advisory, entry in affected_entries(symbol, affected, suffixes): if entry is None or not entry.read or (advisory, symbol) in reported: continue reported.add((advisory, symbol)) check = check_conditions(entry, None) - findings.append(_reached(advisory, symbol, read.location, function, check)) + findings.append(_reached(advisory, symbol, read.location, function, check, entry)) for call in ctx.templates.calls: - for advisory in affected.get(call.symbol, ()): - check = check_conditions(advisory.entry_point(call.symbol), None) - findings.append(_reached(advisory, call.symbol, call.span, None, check)) + for advisory, entry in affected_entries(call.symbol, affected, suffixes): + check = check_conditions(entry, None) + findings.append(_reached(advisory, call.symbol, call.span, None, check, entry)) return findings @@ -73,7 +74,12 @@ def _read_through(symbol: SymbolId) -> Iterator[SymbolId]: def _reached( - advisory: Advisory, symbol: SymbolId, span: SourceSpan, function: str | None, check: ConditionCheck + advisory: Advisory, + symbol: SymbolId, + span: SourceSpan, + function: str | None, + check: ConditionCheck, + entry: AdvisoryEntryPoint | None, ) -> Finding: return Finding( rule_id="reachable-vulnerability", @@ -85,5 +91,5 @@ def _reached( confidence=Confidence.HIGH, span=span, function=function, - metadata=evidence(advisory, symbol, "reachable", check), + metadata=evidence(advisory, symbol, "reachable", check, entry), ) diff --git a/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py b/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py index edf40a2..0f8881b 100644 --- a/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py +++ b/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py @@ -7,7 +7,13 @@ from coretrace_python.analysis import AnyAnalysis from coretrace_python.dependency import DependencyAnalysis -from coretrace_python.dependency.correlation import affected_symbols, check_conditions, ruled_out +from coretrace_python.dependency.correlation import ( + affected_entries, + affected_symbols, + check_conditions, + ruled_out, + suffix_index, +) from coretrace_python.findings import Confidence, Finding from coretrace_python.interprocedural import CallGraphAnalysis, ExternalSymbol from coretrace_python.plugins import ProjectContext, ProjectPlugin @@ -57,6 +63,7 @@ def _ruled_out(ctx: ProjectContext) -> dict[str, list[str]]: conditions: the evidence that the requirement, imported, is not reached there.""" affected = affected_symbols(ctx.dependencies, ctx.advisories) + suffixes = suffix_index(affected, ctx.modules) excluded: dict[str, list[str]] = {} for module in sorted(ctx.modules): graph = ctx.call_graph(module) @@ -64,8 +71,8 @@ def _ruled_out(ctx: ProjectContext) -> dict[str, list[str]]: for site in graph.sites(function): if not isinstance(site.target, ExternalSymbol): continue - for advisory in affected.get(site.target.symbol, ()): - check = check_conditions(advisory.entry_point(site.target.symbol), site.arguments) + for advisory, entry in affected_entries(site.target.symbol, affected, suffixes): + check = check_conditions(entry, site.arguments) if check.contradicted is not None: excluded.setdefault(advisory.id, []).append(ruled_out(module, site, check)) return excluded diff --git a/src/coretrace_python/dependency/advisories.py b/src/coretrace_python/dependency/advisories.py index b0bab34..e5d7c33 100644 --- a/src/coretrace_python/dependency/advisories.py +++ b/src/coretrace_python/dependency/advisories.py @@ -6,7 +6,7 @@ that a project keeps at its root as ``advisories.json`` or passes with ``--advisories``. OSV records name no affected APIs, so imported advisories feed the requirement checks and the SBOM; a file completed by hand with ``affected_symbols``, ``entry_points`` and -their ``conditions`` also feeds the reachability and correlation checks. +their ``conditions`` and ``suffixes`` also feeds the reachability and correlation checks. """ from __future__ import annotations @@ -161,6 +161,8 @@ def _entry_point_entry(entry_point: AdvisoryEntryPoint) -> dict[str, Any]: } if entry_point.read: entry["read"] = True + if entry_point.suffixes: + entry["suffixes"] = list(entry_point.suffixes) if entry_point.attacker_arguments: entry["attacker_arguments"] = [_attacker_argument_entry(a) for a in entry_point.attacker_arguments] return entry @@ -226,9 +228,20 @@ def _entry_point(entry: Mapping[str, Any]) -> AdvisoryEntryPoint: tuple(_condition(c) for c in entry.get("conditions") or []), read, tuple(_attacker_argument(a) for a in entry.get("attacker_arguments") or []), + tuple(_suffix(s) for s in entry.get("suffixes") or []), ) +def _suffix(value: Any) -> str: + if not isinstance(value, str): + raise TypeError(f"an entry point suffix must be a string, got {value!r}") + parts = value.split(".") + # A single name (``annotate``) would match every project method of that name. + if len(parts) < 2 or not all(p.isidentifier() for p in parts): + raise ValueError(f"an entry point suffix is at least two dot-separated names, got {value!r}") + return value + + def _attacker_argument(entry: Mapping[str, Any]) -> AttackerArgument: argument, position, variadic = entry.get("argument"), entry.get("position"), entry.get("variadic", False) if argument is not None and not isinstance(argument, str): diff --git a/src/coretrace_python/dependency/correlation.py b/src/coretrace_python/dependency/correlation.py index 69d0130..8fb1d5c 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/src/coretrace_python/dependency/correlation.py @@ -23,7 +23,7 @@ from coretrace_python.findings.refutation import Status, Verdict, Verdicts from coretrace_python.interprocedural import Arguments, CallSite, ExternalSymbol from coretrace_python.semantic.symbols import SymbolId -from coretrace_python.taint import KEYWORD_NAMES, Sink, TaintFlow, TaintKind +from coretrace_python.taint import KEYWORD_NAMES, Sink, SuffixSink, TaintFlow, TaintKind from coretrace_python.taint.urls import UrlProof, redirects_disabled Affected = Mapping[SymbolId, tuple[Advisory, ...]] @@ -49,6 +49,56 @@ def advisory_sinks(affected: Affected) -> tuple[Sink, ...]: return tuple(Sink(symbol, TaintKind.ADVISORY) for symbol in affected) +@dataclass(frozen=True) +class SuffixIndex: + """The curated entry-point suffixes of the matching advisories, and the project's + module prefixes: a derived symbol matches a suffix on a dot boundary, as a + ``SuffixSink`` does, and only when it belongs to a project module — another + package's ``objects.annotate`` stays out.""" + + entries: tuple[tuple[str, Advisory, AdvisoryEntryPoint], ...] = () + prefixes: tuple[str, ...] = () + + def matches(self, symbol: SymbolId) -> tuple[tuple[Advisory, AdvisoryEntryPoint], ...]: + name = symbol.canonical_name + found = tuple((a, e) for tail, a, e in self.entries if name.endswith(tail)) + return found if found and name.startswith(self.prefixes) else () + + +def suffix_index(affected: Affected, modules: Iterable[str]) -> SuffixIndex: + advisories = {a: None for found in affected.values() for a in found} + entries = tuple((f".{s}", a, e) for a in advisories for e in a.entry_points for s in e.suffixes) + return SuffixIndex(entries, tuple(f"python.{m}." for m in sorted(set(modules)))) + + +def affected_entries( + symbol: SymbolId, affected: Affected, suffixes: SuffixIndex | None = None +) -> tuple[tuple[Advisory, AdvisoryEntryPoint | None], ...]: + """The advisories reached through ``symbol``, each with the entry point it is reached + by: the advisories affecting the exact symbol, then those a suffix matches.""" + + found: list[tuple[Advisory, AdvisoryEntryPoint | None]] = [ + (a, a.entry_point(symbol)) for a in affected.get(symbol, ()) + ] + if suffixes is not None: + seen = {(a.id, a.package, None if e is None else e.symbol) for a, e in found} + for advisory, entry in suffixes.matches(symbol): + key = (advisory.id, advisory.package, entry.symbol) + if key not in seen: + seen.add(key) + found.append((advisory, entry)) + return tuple(found) + + +def advisory_suffix_sinks(affected: Affected) -> tuple[SuffixSink, ...]: + """A suffix sink per curated entry-point suffix, so the taint engine tracks flows + into the derived symbols; ``correlate`` then keeps only the project's.""" + + advisories = {a: None for found in affected.values() for a in found} + suffixes = {s: None for a in advisories for e in a.entry_points for s in e.suffixes} + return tuple(SuffixSink(s, TaintKind.ADVISORY) for s in suffixes) + + @dataclass(frozen=True) class ConditionCheck: """What one call tells of an entry point's conditions: those it meets, those left to @@ -137,13 +187,21 @@ def ruled_out(module: str, site: CallSite, check: ConditionCheck) -> str: return f"{module}:{site.location.start_line} {site.target.symbol}({passed})" -def evidence(advisory: Advisory, symbol: SymbolId, level: str, check: ConditionCheck) -> dict[str, str]: +def evidence( + advisory: Advisory, + symbol: SymbolId, + level: str, + check: ConditionCheck, + entry: AdvisoryEntryPoint | None = None, +) -> dict[str, str]: """What a finding keeps of the advisory for ``symbol``: the level of evidence established, how the symbol relates to the vulnerability, and its conditions — those - the call meets and those left to review.""" + the call meets and those left to review. ``entry`` names the entry point ``symbol`` + was matched by when it is not the advisory's own symbol, as a suffix match derives + it from the project; the metadata then keeps both.""" metadata = {"advisory": advisory.id, "package": advisory.package, "symbol": str(symbol), "level": level} - entry = advisory.entry_point(symbol) + entry = entry if entry is not None else advisory.entry_point(symbol) if entry is None: metadata["justification"] = DIRECT return metadata @@ -164,6 +222,7 @@ def correlate( verdicts: Verdicts | None, affected: Affected, urls: Mapping[TaintFlow, UrlProof] | None = None, + suffixes: SuffixIndex | None = None, ) -> tuple[Finding, ...]: """Exploitable-vulnerability findings for the non-refuted ADVISORY flows of a function, one per advisory the sink is affected by; ``urls`` holds what each flow's URL proves.""" @@ -176,25 +235,30 @@ def correlate( if verdict is not None and verdict.status is Status.REFUTED: continue hotspot = verdict is not None and verdict.status is Status.HOTSPOT - for advisory in affected.get(flow.sink.symbol, ()): - entry = advisory.entry_point(flow.sink.symbol) + for advisory, entry in affected_entries(flow.sink.symbol, affected, suffixes): if entry is not None and not any(entry.exploitable_through(p, k) for p, k in flow.passed_as): continue check = check_conditions(entry, flow.sink_arguments, (urls or {}).get(flow), flow.passed_as) if check.contradicted is None: - findings.append(_exploitable(function, flow, advisory, verdict, hotspot, check)) + findings.append(_exploitable(function, flow, advisory, verdict, hotspot, check, entry)) return tuple(findings) def _exploitable( - function: str, flow: TaintFlow, advisory: Advisory, verdict: Verdict | None, hotspot: bool, check: ConditionCheck + function: str, + flow: TaintFlow, + advisory: Advisory, + verdict: Verdict | None, + hotspot: bool, + check: ConditionCheck, + entry: AdvisoryEntryPoint | None, ) -> Finding: message = ( f"{advisory.id}: {flow.source.label} input reaches {flow.sink.symbol}, affected in " f"the required {advisory.package} {advisory.vulnerable}: {advisory.summary}" ) metadata = { - **evidence(advisory, flow.sink.symbol, "exploitable", check), + **evidence(advisory, flow.sink.symbol, "exploitable", check, entry), "source": str(flow.source.symbol), "source_label": flow.source.label, "verdict": "hotspot" if hotspot else "vulnerability", diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index ed4af09..4391ff1 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -204,13 +204,19 @@ class AdvisoryEntryPoint: An entry point is reached by a call, or — when ``read`` is set, for an attribute whose getter runs the affected code, such as a lazily parsed request body — by any read of it or of an attribute of it. When it names ``attacker_arguments``, attacker input - exploits it only through one of them; otherwise through any argument.""" + exploits it only through one of them; otherwise through any argument. + + A suffix names the tail of a derived project symbol through which the entry point is + reached when the engine cannot type the receiver: ``Item.objects.annotate`` derives + ``python.app.models.Item.objects.annotate``, which ``objects.annotate`` matches. + Suffixes match on dot boundaries, and only symbols of the project's own modules.""" symbol: SymbolId justification: str conditions: tuple[Condition, ...] = () read: bool = False attacker_arguments: tuple[AttackerArgument, ...] = () + suffixes: tuple[str, ...] = () def exploitable_through(self, position: int | None, keyword: str | None) -> bool: """Whether attacker input in the argument at ``position`` or ``keyword`` can diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 2a2bbd4..4588cf4 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -53,9 +53,12 @@ ) from coretrace_python.dependency.correlation import ( Affected, + SuffixIndex, advisory_sinks, + advisory_suffix_sinks, affected_symbols, correlate, + suffix_index, ) from coretrace_python.findings import ( ADVISORIES, @@ -424,7 +427,9 @@ def analyze_project( components = _components(registry, root, (path for path, _ in file_advisories)) advisories, origins = _merge_advisories(_contributions(registry, plugins, root, file_advisories)) affected = affected_symbols(dependencies, advisories) - models = plugin_models(all_plugins, root).extended(*advisory_sinks(affected)) + models = plugin_models(all_plugins, root).extended( + *advisory_sinks(affected), suffixes=advisory_suffix_sinks(affected) + ) members = models.members_by_class() for manager in managers.values(): manager.provide(SecurityModelAnalysis, models) @@ -480,6 +485,7 @@ def analyze_project( reused = tuple(sorted(results)) module_plugins = tuple(p for p in all_plugins if not isinstance(p, ProjectPlugin)) + suffixes = suffix_index(affected, analysable) pool = None if jobs > 1: pool = concurrent.futures.ProcessPoolExecutor( @@ -492,7 +498,9 @@ def analyze_project( if pool is None: for component in pending: batch = {name: analysable[name] for name in sorted(component)} - computed.update(_analyse_managers(batch, _seed(results, graph, component), module_plugins, affected)) + computed.update( + _analyse_managers(batch, _seed(results, graph, component), module_plugins, affected, suffixes) + ) for manager in batch.values(): manager.run(ResultsEvicted) else: @@ -509,6 +517,7 @@ def analyze_project( _encode_routes(routes), tuple(sorted(escaped)), template_calls, + project=tuple(sorted(analysable)), ), ) for component in pending @@ -734,6 +743,7 @@ def _analyse_managers( seed: SummaryIndex, plugins: tuple[Plugin, ...], affected: Affected, + suffixes: SuffixIndex, ) -> dict[str, CachedModule]: """Analyse one component: iterate its summaries to a fixpoint over ``seed`` (§21), then extract what the rest of the run needs from each module. Each member registers @@ -762,7 +772,7 @@ def _analyse_managers( for manager in managers.values(): manager.run(ProjectSummariesUpdated) manager.provide(ProjectSummaries, index) - return {name: _analyse_module(manager, plugins, affected) for name, manager in managers.items()} + return {name: _analyse_module(manager, plugins, affected, suffixes) for name, manager in managers.items()} @dataclass(frozen=True) @@ -780,6 +790,9 @@ class _Batch: routes: tuple[tuple[str, str, str, int], ...] = () escaped: tuple[str, ...] = () template_calls: TemplateCalls = field(default_factory=TemplateCalls) + # Every analysable module of the project, not only the component's: a curated + # entry-point suffix matches derived symbols of any project module. + project: tuple[str, ...] = () def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]: @@ -796,7 +809,9 @@ def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]: continue advisories, _ = _merge_advisories(_contributions(registry, batch.plugins, batch.root, file_advisories)) affected = affected_symbols(dependencies, advisories) - models = plugin_models(all_plugins, batch.root).extended(*advisory_sinks(affected)) + models = plugin_models(all_plugins, batch.root).extended( + *advisory_sinks(affected), suffixes=advisory_suffix_sinks(affected) + ) routes = _decode_routes(batch.routes) escaped = frozenset(batch.escaped) members = models.members_by_class() @@ -810,7 +825,8 @@ def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]: manager.provide(ClearingAnalysis, models.clearing(escaped)) manager.provide(TemplateCallsAnalysis, batch.template_calls) module_plugins = tuple(p for p in all_plugins if not isinstance(p, ProjectPlugin)) - results = _analyse_managers(managers, decode_index(batch.seed), module_plugins, affected) + suffixes = suffix_index(affected, batch.project) + results = _analyse_managers(managers, decode_index(batch.seed), module_plugins, affected, suffixes) return {name: encode(entry) for name, entry in results.items()} @@ -860,7 +876,7 @@ def _configuration_key( def _analyse_module( - manager: AnalysisManager, plugins: tuple[Plugin, ...], affected: Affected + manager: AnalysisManager, plugins: tuple[Plugin, ...], affected: Affected, suffixes: SuffixIndex ) -> CachedModule: """One module's findings, summaries, call sites and symbol reads: what the cache keeps (§11).""" @@ -876,7 +892,9 @@ def _analyse_module( defs = {i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None} urls = {flow: flow_url(flow, defs, strings) for flow in flows if flow.kinds & TaintKind.ADVISORY} correlated.extend( - correlate(graph.name_of(function), flows, manager.get(RefutationAnalysis, function), affected, urls) + correlate( + graph.name_of(function), flows, manager.get(RefutationAnalysis, function), affected, urls, suffixes + ) ) return CachedModule( manager.get(EntryPointAnalysis), diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index 2469238..3efd2cd 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -356,20 +356,23 @@ def source_covering(self, symbol: SymbolId) -> Source | None: return None def sink(self, symbol: SymbolId) -> Sink | None: + # An exact sink and the matching suffix sinks compose: an advisory sink on a + # derived symbol must not shadow ``objects.extra``'s SQL kind, nor conversely. found = self._by_symbol[Sink].get(symbol) - if isinstance(found, Sink): - return found + merged = found if isinstance(found, Sink) else None for suffix in self.suffix_sinks: if symbol.canonical_name.endswith(f".{suffix.suffix}"): - return Sink(symbol, suffix.kinds, suffix.positions) - return None + matched = Sink(symbol, suffix.kinds, suffix.positions) + merged = matched if merged is None else merged.merged(matched) + return merged def route_registrar(self, symbol: SymbolId) -> RouteRegistrar | None: found = self._by_symbol[RouteRegistrar].get(symbol) return found if isinstance(found, RouteRegistrar) else None - def extended(self, *sinks: Sink) -> ModelTable: - """A table with extra sinks; a sink already present gains the new kinds.""" + def extended(self, *sinks: Sink, suffixes: tuple[SuffixSink, ...] = ()) -> ModelTable: + """A table with extra sinks and suffix sinks; a sink already present gains the + new kinds, a suffix sink already present is kept once.""" merged = {sink.symbol: sink for sink in self.sinks} for sink in sinks: @@ -385,7 +388,7 @@ def extended(self, *sinks: Sink) -> ModelTable: self.authorizations, self.named_parameters, self.route_registrars, - self.suffix_sinks, + self.suffix_sinks + tuple(s for s in suffixes if s not in self.suffix_sinks), self.safe_arguments, self.template_renders, self.members, From 699dfae8f72050948ade41d6309dfefca2548e27 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 30 Sep 2026 10:36:20 +0200 Subject: [PATCH 6/9] docs: describe entry-point suffixes in the usage and plugin guides --- docs/plugins.md | 5 ++++- docs/usage.md | 23 +++++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/docs/plugins.md b/docs/plugins.md index 8816b8d..3bfaa77 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -196,7 +196,10 @@ symbols, as the shipped `sample-advisories` plugin does. Requirements matching t reported, calls to the affected symbols become reachable vulnerabilities and tainted calls become exploitable ones. An advisory's entry points carry `conditions`, of kind `argument`, `host`, `semantic` or `keyword_name`, and `attacker_arguments`, decided as -the [usage guide](usage.md) describes. +the [usage guide](usage.md) describes. An entry point may also declare `suffixes`, +matched against the tail of a derived symbol of the project's own modules on dot +boundaries, as a `SuffixSink` matches, for receivers the engine cannot type (see the +usage guide). ### Reclassifying findings: `ProjectPlugin.refine` diff --git a/docs/usage.md b/docs/usage.md index e4271c3..684478b 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -370,6 +370,29 @@ it is reachable, called or not — `request.form['name']`, `for key in request.f `request.form.get('name')` all read it. A function reading it several times is reported once, where it first reads it. +An entry point may declare `suffixes`, for receivers the engine cannot type. A project +reaches `python.django.db.models.QuerySet.annotate` through a manager on its own model, +`Item.objects.annotate(...)`, whose call derives the project's own symbol +`python.app.models.Item.objects.annotate` — `Item` is a project class, `objects` an +attribute the engine does not type — so the canonical entry point never matches. A +suffix matches any symbol of the project's own modules ending with a dot plus the +suffix: + +```json +{"symbol": "python.django.db.models.QuerySet.annotate", "justification": "…", + "suffixes": ["objects.annotate", "objects.filter.annotate"]} +``` + +A queryset chain is declared explicitly: `Item.objects.filter(...).annotate(...)` +derives `…objects.filter.annotate`, which only the second suffix matches. A suffix is +at least two dot-separated names, so a bare method name, matching every project method +of that name, cannot be declared; another package's symbol, such as +`other_lib.objects.annotate`, never matches. The finding names the advisory's symbol as +`entry_point` and the derived one as `symbol`, and the entry point's conditions and +`attacker_arguments` apply through a suffix as through the exact symbol. A project +attribute chain genuinely ending with a declared suffix matches too — the vulnerable +version range and the conditions still gate what is reported. + A filter of Django's own libraries that a project template applies is a call to the function behind it, placed at the template's line: `{{ bio|striptags }}` calls `python.django.template.defaultfilters.striptags`, and so do filters in tag arguments From 8f1429dbf54fbe594339f6f3f121afb4f6ea6d09 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 30 Sep 2026 10:43:07 +0200 Subject: [PATCH 7/9] test: cover sequence conditions across receivers, order and the cache --- tests/test_sequence_conditions.py | 461 ++++++++++++++++++++++++++++++ 1 file changed, 461 insertions(+) create mode 100644 tests/test_sequence_conditions.py diff --git a/tests/test_sequence_conditions.py b/tests/test_sequence_conditions.py new file mode 100644 index 0000000..359710d --- /dev/null +++ b/tests/test_sequence_conditions.py @@ -0,0 +1,461 @@ +"""An advisory condition of kind ``sequence`` relates two calls on one receiver. + +Some flaws span a call sequence: ``Session.get(url, verify=False)`` disables +certificate verification for the pooled connection a later ``Session.get(other)`` on +the same session reuses (requests CVE-2024-35195), and a tornado +``CurlAsyncHTTPClient`` keeps the first fetch's credentials on the reused curl handle. +The entry point is the second call; the condition names the method the first call must +be and, optionally, the argument it must carry. A prior call on the same receiver, on +every path to the entry call, meets it; a receiver followed whole with no such prior +call rules the entry call out, so a project with only single calls stays +``not_affected``; a receiver the engine cannot follow, an order a branch or a loop +leaves uncertain, or an entry call that could be its own prior across loop iterations, +stays pending review. +""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import pytest + +from coretrace_python import engine +from coretrace_python.cache import CachedModule, ProjectCache, decode, encode +from coretrace_python.dependency import ( + Advisory, + AdvisoryEntryPoint, + AdvisoryFileError, + Condition, + dump_advisories, + load_advisories, + render_vex, +) +from coretrace_python.dependency.correlation import check_conditions +from coretrace_python.engine import ProjectAnalysis +from coretrace_python.findings import Finding, Severity +from coretrace_python.interprocedural import Arguments, CallSite, ExternalSymbol, PriorCall +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceId, SourceSpan + +PLUGINS = engine.BUNDLED_PLUGINS +NOW = datetime(2026, 9, 30, 12, 0, tzinfo=UTC) + +GET = "python.requests.Session.get" +TEXT = "a prior call on the same session disabled certificate verification" +CONDITION = Condition("sequence", TEXT, "verify", ("False",), method="Session.get") +ADVISORY = Advisory( + "CVE-2024-35195", + "requests", + "<2.32.0", + "Session keeps verify=False for the connection later requests reuse", + Severity.MEDIUM, + entry_points=( + AdvisoryEntryPoint(SymbolId(GET), "the second request reuses the unverified connection", (CONDITION,)), + ), +) + +LOCK = """version = 1 + +[[package]] +name = "app" +version = "0.1.0" +source = { virtual = "." } +dependencies = [{ name = "requests" }] + +[[package]] +name = "requests" +version = "2.31.0" +source = { registry = "https://pypi.org/simple" } +""" + +HEADER = "import requests\n\n\n" +FETCH_BOTH = HEADER + ( + "def fetch_both(url: str, other: str):\n" + " s = requests.Session()\n" + " s.get(url, verify=False)\n" + " return s.get(other) # reuses the unverified connection\n" +) +FETCH_ONE = HEADER + ( + "def fetch_one(other: str):\n" + " s = requests.Session()\n" + " return s.get(other) # nothing to reuse\n" +) +RULED_OUT = f"app.fetch:6 {GET}(no prior Session.get on the receiver)" +TWO_RECEIVERS = HEADER + ( + "def fetch_two(url: str, other: str):\n" + " s1 = requests.Session()\n" + " s1.get(url, verify=False)\n" + " s2 = requests.Session()\n" + " return s2.get(other)\n" +) +REVERSED = HEADER + ( + "def fetch_reversed(url: str, other: str):\n" + " s = requests.Session()\n" + " s.get(other)\n" + " return s.get(url, verify=False)\n" +) +PARAMETER = HEADER + "def fetch(s: requests.Session, other: str):\n return s.get(other)\n" +BRANCHED = HEADER + ( + "def fetch_maybe(url: str, other: str, flag: bool):\n" + " s = requests.Session()\n" + " if flag:\n" + " s.get(url, verify=False)\n" + " return s.get(other)\n" +) +ESCAPED = HEADER + ( + "def configure(s):\n" + " pass\n" + "\n" + "\n" + "def fetch_helped(url: str, other: str):\n" + " s = requests.Session()\n" + " s.get(url, verify=False)\n" + " configure(s)\n" + " return s.get(other)\n" +) +UNPACKED = HEADER + ( + "def fetch_kw(url: str, other: str, **kw):\n" + " s = requests.Session()\n" + " s.get(url, **kw)\n" + " return s.get(other)\n" +) +LOOP = HEADER + ( + "def fetch_all(urls):\n" + " s = requests.Session()\n" + " for u in urls:\n" + " s.get(u, verify=False)\n" +) + +CURL_FETCH = "python.tornado.httpclient.CurlAsyncHTTPClient.fetch" +CURL_TEXT = "an earlier fetch on the same client left credentials on the reused curl handle" +CURL_CONDITION = Condition("sequence", CURL_TEXT, method="CurlAsyncHTTPClient.fetch") +CURL_ADVISORY = Advisory( + "CVE-2026-91992", + "tornado", + "<6.5.7", + "CurlAsyncHTTPClient reuses the previous fetch's credentials, certificate or proxy", + Severity.HIGH, + entry_points=( + AdvisoryEntryPoint(SymbolId(CURL_FETCH), "the reused curl handle keeps per-request state", (CURL_CONDITION,)), + ), +) +CURL_LOCK = """version = 1 + +[[package]] +name = "app" +version = "0.1.0" +source = { virtual = "." } +dependencies = [{ name = "tornado" }] + +[[package]] +name = "tornado" +version = "6.5.0" +source = { registry = "https://pypi.org/simple" } +""" +CURL = ( + "from tornado.httpclient import CurlAsyncHTTPClient\n" + "\n" + "\n" + "async def fetch_pair(first, second):\n" + " client = CurlAsyncHTTPClient()\n" + " await client.fetch(first)\n" + " return await client.fetch(second)\n" +) + + +def analyse( + root: Path, + source: str, + *, + lock: str = LOCK, + advisories: tuple[Advisory, ...] = (ADVISORY,), + cache: ProjectCache | None = None, +) -> ProjectAnalysis: + files = {"uv.lock": lock, "app/__init__.py": "", "app/fetch.py": source} + for relative, text in files.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + (root / "advisories.json").write_text(dump_advisories(advisories), encoding="utf-8") + return engine.analyze_project(root, [PLUGINS], cache=cache) + + +def of_rule(analysis: ProjectAnalysis, rule_id: str) -> list[Finding]: + return sorted((f for f in analysis.findings if f.rule_id == rule_id), key=lambda f: f.span.start_line) + + +def statement(analysis: ProjectAnalysis, root: Path, advisory: Advisory = ADVISORY) -> dict[str, Any]: + evidence = (*analysis.findings, *analysis.suppressed, *analysis.accepted) + document = render_vex( + analysis.dependencies, analysis.advisories, evidence, analysis.coverage, root, "coretrace", "0.6.0", NOW + ) + return next(s for s in json.loads(document)["statements"] if s["vulnerability"]["name"] == advisory.id) + + +# --------------------------------------------------------------------------- the reproduction + + +def test_a_prior_call_disabling_verification_meets_the_condition(tmp_path: Path) -> None: + analysis = analyse(tmp_path, FETCH_BOTH) + + first, second = of_rule(analysis, "reachable-vulnerability") + assert second.span.start_line == 7 + assert second.metadata["symbol"] == GET + assert second.metadata["conditions_met"] == f"{TEXT} (prior Session.get at line 6)" + assert "conditions_pending_review" not in second.metadata + # The first call could be its own prior across the iterations of a loop. + assert first.span.start_line == 6 + assert first.metadata["conditions_pending_review"] == TEXT + assert "conditions_met" not in first.metadata + (declared,) = of_rule(analysis, "vulnerable-dependency") + assert "ruled_out" not in declared.metadata + assert statement(analysis, tmp_path)["status"] == "affected" + + +def test_a_single_call_without_the_argument_is_ruled_out(tmp_path: Path) -> None: + analysis = analyse(tmp_path, FETCH_ONE) + + assert of_rule(analysis, "reachable-vulnerability") == [] + assert of_rule(analysis, "exploitable-vulnerability") == [] + (declared,) = of_rule(analysis, "vulnerable-dependency") + assert declared.metadata["level"] == "imported" + assert declared.metadata["ruled_out"] == RULED_OUT + found = statement(analysis, tmp_path) + assert found["status"] == "not_affected" + assert found["impact_statement"].endswith(f" Calls ruled out by their arguments: {RULED_OUT}.") + + +def test_calls_on_different_receivers_do_not_relate(tmp_path: Path) -> None: + analysis = analyse(tmp_path, TWO_RECEIVERS) + + # The fetching call has its own receiver and no prior call on it: ruled out. The + # disabling call itself carries verify=False, so it is pending, never ruled out. + (disabling,) = of_rule(analysis, "reachable-vulnerability") + assert disabling.span.start_line == 6 + assert disabling.metadata["conditions_pending_review"] == TEXT + assert "conditions_met" not in disabling.metadata + (declared,) = of_rule(analysis, "vulnerable-dependency") + assert declared.metadata["ruled_out"] == f"app.fetch:8 {GET}(no prior Session.get on the receiver)" + assert statement(analysis, tmp_path)["status"] == "affected" + + +def test_a_later_matching_call_leaves_the_earlier_one_pending(tmp_path: Path) -> None: + analysis = analyse(tmp_path, REVERSED) + + first, second = of_rule(analysis, "reachable-vulnerability") + assert (first.span.start_line, second.span.start_line) == (6, 7) + for found in (first, second): + assert found.metadata["conditions_pending_review"] == TEXT + assert "conditions_met" not in found.metadata + (declared,) = of_rule(analysis, "vulnerable-dependency") + assert "ruled_out" not in declared.metadata + assert statement(analysis, tmp_path)["status"] == "affected" + + +# --------------------------------------------------------------------------- pending receivers + + +@pytest.mark.parametrize( + "source, line", + [ + (PARAMETER, 5), + (BRANCHED, 8), + (ESCAPED, 12), + (UNPACKED, 7), + ], + ids=["parameter-receiver", "prior-under-if", "receiver-passed-to-helper", "prior-with-unpacked-arguments"], +) +def test_a_receiver_or_an_order_the_engine_cannot_follow_is_pending( + tmp_path: Path, source: str, line: int +) -> None: + analysis = analyse(tmp_path, source) + + found = next(f for f in of_rule(analysis, "reachable-vulnerability") if f.span.start_line == line) + assert found.metadata["conditions_pending_review"] == TEXT + assert "conditions_met" not in found.metadata + (declared,) = of_rule(analysis, "vulnerable-dependency") + assert "ruled_out" not in declared.metadata + assert statement(analysis, tmp_path)["status"] == "affected" + + +def test_a_lone_call_in_a_loop_may_be_its_own_prior(tmp_path: Path) -> None: + analysis = analyse(tmp_path, LOOP) + + (found,) = of_rule(analysis, "reachable-vulnerability") + assert found.span.start_line == 7 + assert found.metadata["conditions_pending_review"] == TEXT + assert "conditions_met" not in found.metadata + (declared,) = of_rule(analysis, "vulnerable-dependency") + assert "ruled_out" not in declared.metadata + assert statement(analysis, tmp_path)["status"] == "affected" + + +def test_a_method_only_condition_relates_the_two_fetches_of_one_client(tmp_path: Path) -> None: + analysis = analyse(tmp_path, CURL, lock=CURL_LOCK, advisories=(CURL_ADVISORY,)) + + first, second = of_rule(analysis, "reachable-vulnerability") + assert second.span.start_line == 7 + assert second.metadata["symbol"] == CURL_FETCH + assert second.metadata["conditions_met"] == f"{CURL_TEXT} (prior CurlAsyncHTTPClient.fetch at line 6)" + assert first.span.start_line == 6 + assert first.metadata["conditions_pending_review"] == CURL_TEXT + assert statement(analysis, tmp_path, CURL_ADVISORY)["status"] == "affected" + + +# --------------------------------------------------------------------------- advisory files + + +def test_the_condition_round_trips_through_advisory_files(tmp_path: Path) -> None: + path = tmp_path / "advisories.json" + path.write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + + assert load_advisories(path) == (ADVISORY,) + (entry,) = json.loads(path.read_text())["advisories"][0]["entry_points"] + assert entry["conditions"] == [ + {"kind": "sequence", "text": TEXT, "argument": "verify", "values": ["False"], "method": "Session.get"} + ] + + +@pytest.mark.parametrize( + "condition", + [ + {"kind": "sequence", "text": TEXT}, + {"kind": "sequence", "text": TEXT, "method": ""}, + {"kind": "sequence", "text": TEXT, "method": "Session.get", "values": ["False"]}, + {"kind": "argument", "text": TEXT, "argument": "verify", "values": ["False"], "method": "Session.get"}, + ], + ids=["no-method", "empty-method", "values-without-argument", "method-on-another-kind"], +) +def test_an_invalid_sequence_condition_is_rejected(tmp_path: Path, condition: dict[str, Any]) -> None: + document = json.loads(dump_advisories((ADVISORY,))) + document["advisories"][0]["entry_points"][0]["conditions"] = [condition] + path = tmp_path / "advisories.json" + path.write_text(json.dumps(document), encoding="utf-8") + + with pytest.raises(AdvisoryFileError): + load_advisories(path) + + +# --------------------------------------------------------------------------- the cache + + +SPAN = SourceSpan(SourceId("/p/a.py"), 7, 12, 7, 24) +PRIOR = PriorCall( + SymbolId(GET), + SourceSpan(SourceId("/p/a.py"), 6, 5, 6, 30), + Arguments((None,), (("verify", "False"),)), + True, +) + + +def test_prior_calls_round_trip_through_the_cache_codec() -> None: + related = CallSite("f", SPAN, ExternalSymbol(SymbolId(GET)), Arguments((None,)), (PRIOR,)) + unfollowed = CallSite("g", SPAN, ExternalSymbol(SymbolId(GET))) + + text = json.dumps(encode(CachedModule((), {}, (related, unfollowed), ()))) + restored = decode(json.loads(text)) + + assert restored.sites == (related, unfollowed) + assert restored.sites[0].prior_calls == (PRIOR,) + assert restored.sites[1].prior_calls is None + + +def test_findings_served_from_the_cache_are_identical(tmp_path: Path) -> None: + cache = ProjectCache(tmp_path / "cache") + + first = analyse(tmp_path / "src", FETCH_BOTH, cache=cache) + second = analyse(tmp_path / "src", FETCH_BOTH, cache=cache) + + assert second.findings == first.findings + met = [f for f in second.findings if f.rule_id == "reachable-vulnerability" and "conditions_met" in f.metadata] + assert len(met) == 1 + + +# --------------------------------------------------------------------------- the decision + + +ENTRY = AdvisoryEntryPoint(SymbolId(GET), "reuses the connection", (CONDITION,)) +BARE = Arguments((None,)) +DISABLING = Arguments((None,), (("verify", "False"),)) +UNDECIDED = Arguments((None,), (), True) + + +def prior(dominates: bool, arguments: Arguments = DISABLING, symbol: str = GET, line: int = 5) -> PriorCall: + return PriorCall(SymbolId(symbol), SourceSpan(SourceId("/p/a.py"), line, 5, line, 30), arguments, dominates) + + +def test_a_dominating_satisfying_prior_call_meets_the_condition() -> None: + check = check_conditions(ENTRY, BARE, prior_calls=(prior(True),)) + + (met,) = check.met + assert met.text == f"{TEXT} (prior Session.get at line 5)" + assert check.pending == () + assert check.contradicted is None + + +def test_the_first_satisfying_dominating_prior_call_names_the_line() -> None: + check = check_conditions(ENTRY, BARE, prior_calls=(prior(True, line=3), prior(True, line=4))) + + (met,) = check.met + assert met.text.endswith("(prior Session.get at line 3)") + + +def test_the_method_may_be_the_full_canonical_name() -> None: + condition = Condition("sequence", TEXT, "verify", ("False",), method=GET) + entry = AdvisoryEntryPoint(SymbolId(GET), "reuses the connection", (condition,)) + + check = check_conditions(entry, BARE, prior_calls=(prior(True),)) + + assert len(check.met) == 1 + assert check.contradicted is None + + +@pytest.mark.parametrize( + "arguments, prior_calls", + [ + (BARE, None), + (BARE, (prior(False),)), + (BARE, (prior(True, UNDECIDED),)), + (DISABLING, ()), + (UNDECIDED, ()), + (None, ()), + ], + ids=[ + "unfollowed-receiver", + "satisfying-but-non-dominating", + "prior-arguments-undecided", + "the-call-is-its-own-prior", + "own-arguments-undecided", + "no-arguments-known", + ], +) +def test_uncertain_priors_leave_the_condition_pending( + arguments: Arguments | None, prior_calls: tuple[PriorCall, ...] | None +) -> None: + check = check_conditions(ENTRY, arguments, prior_calls=prior_calls) + + assert check.met == () + assert check.pending == (CONDITION,) + assert check.contradicted is None + + +@pytest.mark.parametrize( + "prior_calls", + [ + (), + (prior(True, BARE),), + (prior(True, symbol="python.requests.Session.post"),), + ], + ids=["alone-on-the-receiver", "prior-without-the-argument", "another-method"], +) +def test_a_followed_receiver_without_a_satisfying_prior_call_is_contradicted( + prior_calls: tuple[PriorCall, ...], +) -> None: + check = check_conditions(ENTRY, BARE, prior_calls=prior_calls) + + assert check.met == () + assert check.contradicted == CONDITION + assert check.passed is None From bcd49cb091d9c55d3ea94d6b1527a989326a9132 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 30 Sep 2026 10:43:07 +0200 Subject: [PATCH 8/9] feat: relate two calls on one receiver for sequence advisory conditions A sequence condition names the method a prior call on the same receiver must have made, with an optional argument constraint read as an argument condition. The call graph records, per call site, the other external calls on a receiver whose whole lifetime is visible (bound once to a call or a with, read only to call external methods, capped), each with whether it dominates the site; the cache carries them and its format becomes 13. The condition is met by a dominating satisfied prior, contradicted only when the followed receiver shows no call that could satisfy it, the entry call's own arguments included since a loop makes it its own prior, and pending everywhere else: an unfollowed receiver, an undecided or non-dominating prior, a flow through a summary. Closes #194. --- CHANGELOG.md | 10 ++ .../reachable_vulnerability.py | 2 +- .../vulnerable_dependency.py | 4 +- src/coretrace_python/cache.py | 18 ++- src/coretrace_python/dependency/advisories.py | 9 ++ .../dependency/correlation.py | 74 +++++++++- src/coretrace_python/dependency/graph.py | 11 +- src/coretrace_python/engine.py | 9 +- .../interprocedural/__init__.py | 2 + .../interprocedural/callgraph.py | 136 +++++++++++++++++- 10 files changed, 260 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 935943e..b56b47d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## Unreleased + +### Advisories + +- An advisory condition of kind `sequence` says the flaw spans two calls on one receiver: the first call leaves state on the object the entry point — the second call — reuses. It names the `method` the first call must be, by canonical name or `.`-suffix, and, optionally, the argument that call must carry (`{"kind": "sequence", "method": "Session.get", "argument": "verify", "values": ["False"], ...}`), so `s = requests.Session(); s.get(url, verify=False); s.get(other)` marks the second `get` reachable with the prior call as evidence (requests CVE-2024-35195). A receiver followed whole — bound once to the result of a call or a `with` and used only to call its methods — with no matching call on every path to the entry call rules it out (`app.fetch:6 python.requests.Session.get(no prior Session.get on the receiver)`), and a project with only such calls stays `not_affected`. A receiver the engine cannot follow (a parameter, a module-level instance, one passed to another function or mutated through an attribute), a matching call only in a branch or textually later, an unpacking that leaves the first call's arguments undecided, or an entry call that itself carries the named argument — its own prior across the iterations of a loop — leaves the condition pending review (#194). + +### Plugins + +- `CallSite.prior_calls` records, for a receiver the engine follows whole, the other external calls on the same receiver (`PriorCall`: symbol, location, arguments, and whether it surely executes before this one), in source order; `CallGraph.prior_calls_at` reads it by location, and `check_conditions` takes it to decide `sequence` conditions. The cache format changes. + ## 0.20.0 (2026-09-30) ### Advisories diff --git a/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py b/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py index be80c3e..678c253 100644 --- a/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py +++ b/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py @@ -41,7 +41,7 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: entry = advisory.entry_point(site.target.symbol) if entry is not None and entry.read: continue # a call reads its callee first: reported with the reads - check = check_conditions(entry, site.arguments) + check = check_conditions(entry, site.arguments, prior_calls=site.prior_calls) if check.contradicted is not None: continue findings.append(_reached(advisory, site.target.symbol, site.location, function, check)) diff --git a/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py b/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py index edf40a2..3085709 100644 --- a/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py +++ b/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py @@ -65,7 +65,9 @@ def _ruled_out(ctx: ProjectContext) -> dict[str, list[str]]: if not isinstance(site.target, ExternalSymbol): continue for advisory in affected.get(site.target.symbol, ()): - check = check_conditions(advisory.entry_point(site.target.symbol), site.arguments) + check = check_conditions( + advisory.entry_point(site.target.symbol), site.arguments, prior_calls=site.prior_calls + ) if check.contradicted is not None: excluded.setdefault(advisory.id, []).append(ruled_out(module, site, check)) return excluded diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index e15c1b4..e34ae7d 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -34,6 +34,7 @@ ModuleGraph, Mutation, NonlocalWrite, + PriorCall, SummaryIndex, SymbolRead, Target, @@ -42,7 +43,7 @@ from coretrace_python.semantic.symbols import SymbolId from coretrace_python.source import SourceId, SourceSpan -CACHE_FORMAT = 12 +CACHE_FORMAT = 13 @dataclass(frozen=True) @@ -351,18 +352,33 @@ def _encode_site(site: CallSite) -> dict[str, Any]: "location": _encode_span(site.location), "target": _encode_target(site.target), "arguments": _encode_arguments(site.arguments), + "prior": None + if site.prior_calls is None + else [ + [str(p.symbol), _encode_span(p.location), _encode_arguments(p.arguments), p.dominates] + for p in site.prior_calls + ], } def _decode_site(data: Mapping[str, Any]) -> CallSite: + prior = data["prior"] return CallSite( _string(data["caller"]), _decode_span(data["location"]), _decode_target(data["target"]), _decode_arguments(data["arguments"]), + None if prior is None else tuple(_decode_prior(entry) for entry in prior), ) +def _decode_prior(data: Any) -> PriorCall: + symbol, span, arguments, dominates = data + if not isinstance(dominates, bool): + raise TypeError(f"expected a boolean, got {dominates!r}") + return PriorCall(SymbolId(_string(symbol)), _decode_span(span), _decode_arguments(arguments), dominates) + + def _encode_arguments(arguments: Arguments) -> dict[str, Any]: return { "positional": list(arguments.positional), diff --git a/src/coretrace_python/dependency/advisories.py b/src/coretrace_python/dependency/advisories.py index b0bab34..b093958 100644 --- a/src/coretrace_python/dependency/advisories.py +++ b/src/coretrace_python/dependency/advisories.py @@ -187,6 +187,8 @@ def _condition_entry(condition: Condition) -> dict[str, Any]: entry["position"] = condition.position if condition.default: entry["default"] = True + if condition.method is not None: + entry["method"] = condition.method return entry @@ -258,7 +260,14 @@ def _condition(entry: Mapping[str, Any]) -> Condition: tuple(str(v) for v in entry.get("values") or []), position, default, + None if entry.get("method") is None else str(entry["method"]), ) + if condition.kind == "sequence" and not condition.method: + raise ValueError("a sequence condition names the method of the prior call") + if condition.kind == "sequence" and condition.values and condition.argument is None: + raise ValueError("the values of a sequence condition need the argument they constrain") + if condition.kind != "sequence" and condition.method is not None: + raise ValueError(f"only a sequence condition names a method, not {condition.kind!r}") if condition.kind == "host" and ( (condition.argument is None and condition.position is None) or condition.values or condition.default ): diff --git a/src/coretrace_python/dependency/correlation.py b/src/coretrace_python/dependency/correlation.py index 69d0130..2caa47c 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/src/coretrace_python/dependency/correlation.py @@ -21,7 +21,7 @@ ) from coretrace_python.findings import Confidence, Finding, Severity from coretrace_python.findings.refutation import Status, Verdict, Verdicts -from coretrace_python.interprocedural import Arguments, CallSite, ExternalSymbol +from coretrace_python.interprocedural import Arguments, CallSite, ExternalSymbol, PriorCall from coretrace_python.semantic.symbols import SymbolId from coretrace_python.taint import KEYWORD_NAMES, Sink, TaintFlow, TaintKind from coretrace_python.taint.urls import UrlProof, redirects_disabled @@ -67,6 +67,7 @@ def check_conditions( arguments: Arguments | None, url: UrlProof | None = None, passed_as: frozenset[tuple[int | None, str | None]] = frozenset(), + prior_calls: tuple[PriorCall, ...] | None = None, ) -> ConditionCheck: """Decide ``entry``'s conditions against what the call's ``arguments`` denote and, for a ``host`` condition, what ``url`` proves of the value the attacker's input is passed @@ -75,7 +76,13 @@ def check_conditions( controls, and the condition stays pending with that uncertainty. A ``keyword_name`` condition is met when the attacker's input is passed as the keys of a ``**`` mapping, pending when the call expands a mapping whose keys are not established, and - contradicted by a call writing every keyword name.""" + contradicted by a call writing every keyword name. A ``sequence`` condition is met + when one of the ``prior_calls`` on the call's receiver dominates it — surely executes + before it — matches the condition's method and passes what it requires; contradicted + when the receiver is followed whole and no matching call could do so, the call's own + ``arguments`` included, since across the iterations of a loop the call is its own + prior; pending otherwise — an unfollowed receiver (``prior_calls`` is None), an + undecided or non-dominating matching call, or an undecidable self-check.""" if entry is None: return ConditionCheck() @@ -101,6 +108,30 @@ def check_conditions( text = f"the host is fixed by {origin!r}, but a redirect may lead to a host the attacker controls" pending.append(replace(condition, text=text)) continue + if condition.kind == "sequence": + candidates = [ + p + for p in prior_calls or () + if condition.method is not None and _named(p.symbol, condition.method) + ] + satisfied = [_first_call_satisfies(condition, p.arguments) for p in candidates] + hit = next( + (p for p, s in zip(candidates, satisfied, strict=True) if s is True and p.dominates), None + ) + if hit is not None: + text = f"{condition.text} (prior {condition.method} at line {hit.location.start_line})" + met.append(replace(condition, text=text)) + elif ( + prior_calls is None + or condition.method is None + or None in satisfied + or True in satisfied + or _first_call_satisfies(condition, arguments) is not False + ): + pending.append(condition) + else: + return ConditionCheck(tuple(met), tuple(pending), condition, None) + continue given = arguments.given(condition.argument, condition.position) if condition.checkable and arguments is not None else None if given is None: pending.append(condition) @@ -116,6 +147,32 @@ def check_conditions( return ConditionCheck(tuple(met), tuple(pending)) +def _named(symbol: SymbolId, method: str) -> bool: + """Whether ``symbol`` is the ``method`` a sequence condition names, by exact + canonical name or by ``.``-suffix (``Session.get``).""" + + name = symbol.canonical_name + return name == method or name.endswith("." + method) + + +def _first_call_satisfies(condition: Condition, arguments: Arguments | None) -> bool | None: + """Whether ``arguments`` carry what a sequence ``condition`` requires of the first + call: trivially when it names no argument, else as an argument condition reads + ``given()``; None when they cannot tell.""" + + if condition.argument is None and condition.position is None: + return True + if arguments is None: + return None + given = arguments.given(condition.argument, condition.position) + if given is None: + return None + explicit, value = given + if explicit and value is None: + return None + return value in condition.values if explicit else condition.default + + def _passed_in(condition: Condition, passed_as: frozenset[tuple[int | None, str | None]]) -> bool: """Whether a value passed as ``passed_as`` is the argument ``condition`` names.""" @@ -132,6 +189,11 @@ def ruled_out(module: str, site: CallSite, check: ConditionCheck) -> str: assert check.contradicted is not None and isinstance(site.target, ExternalSymbol) contradicted = check.contradicted + if contradicted.kind == "sequence": + return ( + f"{module}:{site.location.start_line} {site.target.symbol}" + f"(no prior {contradicted.method} on the receiver)" + ) argument = contradicted.argument or ("keywords" if contradicted.kind == "keyword_name" else f"#{contradicted.position}") passed = f"{argument} absent" if check.passed is None else f"{argument}={check.passed}" return f"{module}:{site.location.start_line} {site.target.symbol}({passed})" @@ -164,9 +226,11 @@ def correlate( verdicts: Verdicts | None, affected: Affected, urls: Mapping[TaintFlow, UrlProof] | None = None, + prior: Mapping[TaintFlow, tuple[PriorCall, ...] | None] | None = None, ) -> tuple[Finding, ...]: """Exploitable-vulnerability findings for the non-refuted ADVISORY flows of a function, - one per advisory the sink is affected by; ``urls`` holds what each flow's URL proves.""" + one per advisory the sink is affected by; ``urls`` holds what each flow's URL proves + and ``prior`` the other calls on each flow's receiver, for sequence conditions.""" findings: list[Finding] = [] for flow in flows: @@ -180,7 +244,9 @@ def correlate( entry = advisory.entry_point(flow.sink.symbol) if entry is not None and not any(entry.exploitable_through(p, k) for p, k in flow.passed_as): continue - check = check_conditions(entry, flow.sink_arguments, (urls or {}).get(flow), flow.passed_as) + check = check_conditions( + entry, flow.sink_arguments, (urls or {}).get(flow), flow.passed_as, (prior or {}).get(flow) + ) if check.contradicted is None: findings.append(_exploitable(function, flow, advisory, verdict, hotspot, check)) return tuple(findings) diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index ed4af09..31954e4 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -156,8 +156,14 @@ class Condition: ``position``); the engine decides it from what the URL's constant text proves. A ``keyword_name`` condition says the attacker must choose a keyword name of the call, as an ORM alias expanded from a mapping; the engine decides it from the keys of what - the call expands with ``**``. A ``semantic`` condition cannot be checked and is - reported as pending review.""" + the call expands with ``**``. A ``sequence`` condition says the flaw needs a prior + call on the same receiver: it names the ``method`` that call must be, by canonical + name or ``.``-suffix, and, when ``argument`` is named, what it must pass, as an + argument condition does. It is met when such a call dominates the entry call — + surely executes before it — matches ``method`` and passes a value in ``values`` + (``default`` for an absent argument); contradicted when the receiver is fully + followed and no such prior call can precede; pending otherwise. A ``semantic`` + condition cannot be checked and is reported as pending review.""" kind: str text: str @@ -165,6 +171,7 @@ class Condition: values: tuple[str, ...] = () position: int | None = None default: bool = False + method: str | None = None @property def checkable(self) -> bool: diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 2a2bbd4..4adc0d5 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -875,8 +875,15 @@ def _analyse_module( ssa = manager.get(SSAAnalysis, function) defs = {i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None} urls = {flow: flow_url(flow, defs, strings) for flow in flows if flow.kinds & TaintKind.ADVISORY} + name = graph.name_of(function) + # A flow through a callee reaches the sink at another call: no receiver here. + prior = { + flow: graph.prior_calls_at(name, flow.location) + for flow in flows + if flow.kinds & TaintKind.ADVISORY and flow.through is None + } correlated.extend( - correlate(graph.name_of(function), flows, manager.get(RefutationAnalysis, function), affected, urls) + correlate(name, flows, manager.get(RefutationAnalysis, function), affected, urls, prior) ) return CachedModule( manager.get(EntryPointAnalysis), diff --git a/src/coretrace_python/interprocedural/__init__.py b/src/coretrace_python/interprocedural/__init__.py index d1817ad..fc053b2 100644 --- a/src/coretrace_python/interprocedural/__init__.py +++ b/src/coretrace_python/interprocedural/__init__.py @@ -8,6 +8,7 @@ ExternalSymbol, KnownFunction, ModuleFunction, + PriorCall, Signature, SignaturesAnalysis, SymbolRead, @@ -58,6 +59,7 @@ "ModuleGraph", "Mutation", "NonlocalWrite", + "PriorCall", "ProjectSummaries", "Signature", "SignaturesAnalysis", diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index 300656c..de087c0 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -11,12 +11,13 @@ import re from collections.abc import Mapping -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from types import MappingProxyType from typing import ClassVar from coretrace_python.analysis import Analysis, AnalysisContext, AnyAnalysis -from coretrace_python.cfg import CFGError +from coretrace_python.cfg import BlockId, CFGError +from coretrace_python.cfg.dominance import DominanceAnalysis, DominatorTree from coretrace_python.hir import nodes from coretrace_python.ir.defuse import DefUse, def_use from coretrace_python.ir.lowering import ( @@ -27,6 +28,7 @@ ) from coretrace_python.ir.model import ( Await, + BasicBlock, BuildDict, Call, Constant, @@ -93,12 +95,26 @@ def given(self, keyword: str | None, position: int | None = None) -> tuple[bool, return None if self.unpacked else (False, None) +@dataclass(frozen=True) +class PriorCall: + """Another call on the same receiver in this function; ``dominates`` says it surely + executes before this one.""" + + symbol: SymbolId + location: SourceSpan + arguments: Arguments + dominates: bool + + @dataclass(frozen=True) class CallSite: caller: str location: SourceSpan target: Target arguments: Arguments = field(default_factory=Arguments) + # None: no receiver, or one the engine cannot follow; a tuple is the complete set + # of other external calls on the receiver, in source order. + prior_calls: tuple[PriorCall, ...] | None = None @dataclass(frozen=True) @@ -201,6 +217,13 @@ def arguments_at(self, caller: str, location: SourceSpan) -> Arguments: site = self._at.get((caller, location)) return site.arguments if site is not None else Arguments() + def prior_calls_at(self, caller: str, location: SourceSpan) -> tuple[PriorCall, ...] | None: + """The other calls on the receiver of the call at ``location`` in ``caller``; + None when there is no receiver, or one the engine cannot follow.""" + + site = self._at.get((caller, location)) + return site.prior_calls if site is not None else None + def callees(self, caller: str) -> frozenset[str]: return frozenset( site.target.name for site in self.sites(caller) if isinstance(site.target, KnownFunction) @@ -335,6 +358,105 @@ def _position(span: SourceSpan) -> tuple[int, int]: return span.start_line, span.start_column +# A receiver carrying more calls than this is left unfollowed: the sibling pairing +# stays bounded, and such code is generated, not a client bound and used twice. +MAX_RECEIVER_CALLS = 12 + + +def _relate_receivers( + sites: list[CallSite], + calls: list[tuple[Call, BlockId, int]], + function: FunctionIR, + defs: Mapping[Value, Instruction], + uses: DefUse, + symbols: Mapping[Value, SymbolId], + tree: DominatorTree, +) -> tuple[CallSite, ...]: + """The ``sites`` with ``prior_calls`` filled wherever the receiver's whole lifetime + is visible, so the set of its other calls is complete and program order is decided + by dominance; any other receiver leaves ``None``. ``calls`` gives each site's + instruction with its block and index, aligned with ``sites``.""" + + groups: dict[Value, list[int]] = {} + for index, (call, _, _) in enumerate(calls): + callee = defs.get(call.callee) + if isinstance(callee, GetAttr): + groups.setdefault(callee.object, []).append(index) + related = list(sites) + blocks = {block.id: block for block in function.blocks} + for receiver, group in groups.items(): + if not _followed(receiver, group, sites, blocks, defs, uses, symbols): + continue + ordered = sorted(group, key=lambda member: _position(sites[member].location)) + for index in group: + _, block, at = calls[index] + related[index] = replace( + sites[index], + prior_calls=tuple( + PriorCall( + target.symbol, + sites[other].location, + sites[other].arguments, + _executes_before(calls[other], block, at, tree), + ) + for other in ordered + if other != index and isinstance(target := sites[other].target, ExternalSymbol) + ), + ) + return tuple(related) + + +def _followed( + receiver: Value, + group: list[int], + sites: list[CallSite], + blocks: Mapping[BlockId, BasicBlock], + defs: Mapping[Value, Instruction], + uses: DefUse, + symbols: Mapping[Value, SymbolId], +) -> bool: + """Whether the receiver's whole lifetime is visible: bound once to the result of a + call or a ``with`` denoting a symbol, and read only to call external methods. Only + then is ``group`` complete enough to claim a prior call surely missing; a receiver + returned, passed to another call, mutated through an attribute or merged by a + ``Phi`` may be called elsewhere, and stays unfollowed.""" + + if len(group) > MAX_RECEIVER_CALLS or receiver not in symbols: + return False + if any(not isinstance(sites[index].target, ExternalSymbol) for index in group): + return False + made = defs.get(receiver) + while isinstance(made, Await): + made = defs.get(made.value) + if not isinstance(made, Call | WithEnter): + return False + for use in uses.uses(receiver): + if use.index is None: # a terminator use, as a return, escapes + return False + read = blocks[use.block].instructions[use.index] + if not isinstance(read, GetAttr): + return False + for called in uses.uses(read.result): + if called.index is None: + return False + call = blocks[called.block].instructions[called.index] + if not isinstance(call, Call) or call.callee != read.result or read.result in call.argument_values(): + return False + return True + + +def _executes_before( + other: tuple[Call, BlockId, int], block: BlockId, index: int, tree: DominatorTree +) -> bool: + """Whether ``other`` surely executes before the call at ``index`` in ``block``: it + comes earlier in the same block, or its block dominates this one.""" + + _, other_block, other_index = other + if other_block == block: + return other_index < index + return tree.dominates(other_block, block) + + def resolve_targets( function: FunctionIR, scopes: ScopeTable, @@ -471,7 +593,7 @@ def _typed_with_module_classes( class CallGraphAnalysis(Analysis[CallGraph]): name: ClassVar[str] = "interprocedural.callgraph" requires: ClassVar[frozenset[AnyAnalysis]] = frozenset( - {SSAAnalysis, ScopeAnalysis, SymbolAnalysis, SignaturesAnalysis} + {SSAAnalysis, ScopeAnalysis, SymbolAnalysis, SignaturesAnalysis, DominanceAnalysis} ) @classmethod @@ -537,8 +659,9 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: defs = {i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None} uses = def_use(ssa) found: list[CallSite] = [] + calls: list[tuple[Call, BlockId, int]] = [] for block in ssa.blocks: - for instruction in block.instructions: + for index, instruction in enumerate(block.instructions): if isinstance(instruction, Call): found.append( CallSite( @@ -548,6 +671,9 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: _arguments(instruction, symbols[name], defs, uses), ) ) - sites[name] = tuple(found) + calls.append((instruction, block.id, index)) + sites[name] = _relate_receivers( + found, calls, ssa, defs, uses, symbols[name], ctx.get(DominanceAnalysis, function) + ) reads[name] = _reads(name, ssa, symbols[name]) return CallGraph(definitions, sites, frozenset(unsupported), symbols, reads) From 8425a4d14a957984428b78fdf0e7e4e0e87bbef8 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 30 Sep 2026 10:43:07 +0200 Subject: [PATCH 9/9] docs: document the sequence condition and CallSite.prior_calls --- docs/plugins.md | 2 +- docs/usage.md | 41 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 1 deletion(-) diff --git a/docs/plugins.md b/docs/plugins.md index 8816b8d..b4f65af 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -195,7 +195,7 @@ A model plugin may also carry `advisories`, a tuple of `Advisory` values symbols, as the shipped `sample-advisories` plugin does. Requirements matching them are reported, calls to the affected symbols become reachable vulnerabilities and tainted calls become exploitable ones. An advisory's entry points carry `conditions`, of kind -`argument`, `host`, `semantic` or `keyword_name`, and `attacker_arguments`, decided as +`argument`, `host`, `semantic`, `keyword_name` or `sequence`, and `attacker_arguments`, decided as the [usage guide](usage.md) describes. ### Reclassifying findings: `ProjectPlugin.refine` diff --git a/docs/usage.md b/docs/usage.md index e4271c3..b94f983 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -514,6 +514,47 @@ Keys passed through a project wrapper stay pending, and `mapping.update(other)`, taint. A literal the function fills after building it (`m = {}; m[alias] = expr`) has keys the literal does not write, so the call is not ruled out. +A `sequence` condition says the flaw spans two calls on one receiver: a first call +leaves state on the object, and the entry point — the second call — reuses it, as +`Session.get(url, verify=False)` disables certificate verification for the pooled +connection a later `Session.get(other)` on the same session reuses (requests +CVE-2024-35195). It names the `method` the first call must be, by canonical name or +`.`-suffix, and, optionally, the argument that call must carry, read as an argument +condition reads it: + +```json +{ + "kind": "sequence", + "text": "a prior call on the same session disabled certificate verification", + "method": "Session.get", + "argument": "verify", + "values": ["False"] +} +``` + +The engine follows a receiver bound once in the analysed function to the result of a +call or a `with` and used only to call its methods. The outcome has three cases: + +- A prior call on the same receiver, on every path to the entry call, matches the + method and carries a value in `values` (`default` for an absent argument): the + condition is met, and the evidence names that call's line. +- The receiver's whole lifetime is visible and no such prior call can precede — one + call alone on the receiver, or the two calls on different receivers: the condition + is contradicted, the call is ruled out + (`app.fetch:6 python.requests.Session.get(no prior Session.get on the receiver)`) + and a project with only such calls stays `not_affected`. +- Otherwise the condition stays pending review: a receiver the engine cannot follow — + a parameter, a module-level instance, one passed to another function or mutated + through an attribute such as `s.verify = False` — a prior call whose arguments a + `*`/`**` unpacking leaves undecided, or a flow reaching the entry point inside a + callee. + +Only a prior call on every path to the entry call counts as before it: a matching +call in a branch that may not run, or a textually later one, leaves the condition +pending, since a loop may bring it before the entry call. A call that itself carries +what the condition requires of the first call is never ruled out either: across the +iterations of a loop it is its own prior. + Every dependency finding records the highest level of evidence established in its `level` metadata: `declared` (the requirement allows a vulnerable version), `imported` (a module of the package is imported somewhere), `reachable` (an entry point or affected