diff --git a/CHANGELOG.md b/CHANGELOG.md index 935943e..8be993d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # 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). +- 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/docs/plugins.md b/docs/plugins.md index 8816b8d..a14ad1f 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`, `keyword_name` or `sequence`, 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..dc70fbf 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 @@ -514,6 +547,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 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..53e7ead 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) + 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)) + 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..347943c 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, 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..19c9ac5 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,10 @@ def _condition_entry(condition: Condition) -> dict[str, Any]: entry["position"] = condition.position if condition.default: entry["default"] = True + if condition.present: + entry["present"] = True + if condition.method is not None: + entry["method"] = condition.method return entry @@ -226,9 +232,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 +268,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 +278,22 @@ def _condition(entry: Mapping[str, Any]) -> Condition: tuple(str(v) for v in entry.get("values") or []), position, default, + present, + None if entry.get("method") is None else str(entry["method"]), ) + 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 == "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..2759fd2 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/src/coretrace_python/dependency/correlation.py @@ -21,9 +21,9 @@ ) 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 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 @@ -67,6 +117,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 +126,15 @@ 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. 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,10 +160,40 @@ 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) 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) @@ -116,6 +205,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,18 +247,31 @@ 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})" -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,9 +292,12 @@ def correlate( verdicts: Verdicts | None, affected: Affected, urls: Mapping[TaintFlow, UrlProof] | None = None, + suffixes: SuffixIndex | 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: @@ -176,25 +307,32 @@ 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) + 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)) + 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..4d42e62 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -151,13 +151,21 @@ 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, 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 +173,8 @@ class Condition: values: tuple[str, ...] = () position: int | None = None default: bool = False + present: bool = False + method: str | None = None @property def checkable(self) -> bool: @@ -204,13 +214,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..98f63a8 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).""" @@ -875,8 +891,17 @@ 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, suffixes, 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) 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), + ] 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