feat: match curated entry points by suffix for untypeable receivers

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

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`
 

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

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,37 +32,36 @@ 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)
             for function in graph.functions:
                 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),
     )

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,15 +63,16 @@ 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)
         for function in graph.functions:
             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

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,14 +266,25 @@ 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"]),
         None if entry.get("argument") is None else str(entry["argument"]),
         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
     ):