diff --git a/CHANGELOG.md b/CHANGELOG.md index 935943e..eef5fcb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # 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). +- 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/docs/plugins.md b/docs/plugins.md index 8816b8d..36a4cf9 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -195,8 +195,12 @@ 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 -the [usage guide](usage.md) describes. +`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. 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..bfaf109 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 @@ -471,6 +494,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 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..11e455f 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 @@ -187,6 +189,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 @@ -226,9 +230,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): @@ -251,6 +266,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 +276,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..8e36cf1 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 @@ -75,7 +125,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 +157,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) @@ -137,13 +195,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 +230,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 +243,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..10a8c21 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: @@ -204,13 +207,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, 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) 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), + ]