Merge pull request #147 from CoreTrace/feat/advisory-attacker-arguments

feat: let advisory entry points name the arguments the attacker must control

Author: SizzleUnrlsd

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+### Precision
+
+- An advisory entry point may name the `attacker_arguments` through which attacker input exploits it: by keyword, and by position when the argument may be passed positionally, a variadic one taking every later position. Attacker input in another argument leaves the call reachable instead of exploitable, so an advisory for `send_from_directory` that names its `path` no longer counts `download_name=`, and one for `requests.get` that names its `url` no longer counts the body. An argument unpacked with `*` or `**` may fill any of them: like an argument condition it leaves undecided, it does not rule the call out, so a wrapper forwarding `*args, **kwargs` still reaches the entry point. An entry point naming no argument counts every argument, as before. Each taint flow now records how its value is passed to the sink (#145).
+
 ## 0.8.0 (2026-09-25)
 
 ### Precision

docs/usage.md

@@ -336,6 +336,19 @@ 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.
 
+Most vulnerabilities need attacker input in one argument: the path `send_from_directory`
+serves, not its `download_name`; the URL `requests.get` fetches, not its body. An entry
+point may name them in `attacker_arguments`, by keyword `argument` and, when it may be
+passed positionally, by `position`, counted as for argument conditions; `"variadic": true`
+extends a position to every later one, as `safe_join(directory, *pathnames)` takes
+(`{"position": 1, "variadic": true}`). A call is then exploitable only when attacker
+input reaches one of them, `"attacker_arguments": [{"argument": "url", "position": 0}]`
+for `requests.get`; attacker input in another argument leaves it reachable. An argument
+unpacked with `*args` or `**kwargs` may fill any of them: like an argument condition it
+leaves undecided, it does not rule the call out, so it counts, and a wrapper forwarding
+`*args, **kwargs` to the entry point still reaches it. An entry point without
+`attacker_arguments` is exploitable through any argument.
+
 An `argument` condition is decided at each call. It names the argument by keyword and,
 when it may be passed positionally, by `position` (counted from 0, the receiver of a
 method excluded); `values` are what makes the call affected, written as the analyzer

src/coretrace_python/dependency/__init__.py

@@ -12,6 +12,7 @@
     DEPENDENCY_FILES,
     Advisory,
     AdvisoryEntryPoint,
+    AttackerArgument,
     Condition,
     DependencyAnalysis,
     DependencyGraph,
@@ -31,6 +32,7 @@
     "Advisory",
     "AdvisoryEntryPoint",
     "AdvisoryFileError",
+    "AttackerArgument",
     "Condition",
     "DependencyAnalysis",
     "DependencyGraph",

src/coretrace_python/dependency/advisories.py

@@ -17,7 +17,13 @@
 from pathlib import Path
 from typing import Any
 
-from coretrace_python.dependency.graph import Advisory, AdvisoryEntryPoint, Condition, normalize
+from coretrace_python.dependency.graph import (
+    Advisory,
+    AdvisoryEntryPoint,
+    AttackerArgument,
+    Condition,
+    normalize,
+)
 from coretrace_python.findings import Severity
 from coretrace_python.semantic.symbols import SymbolId
 
@@ -155,6 +161,19 @@ def _entry_point_entry(entry_point: AdvisoryEntryPoint) -> dict[str, Any]:
     }
     if entry_point.read:
         entry["read"] = True
+    if entry_point.attacker_arguments:
+        entry["attacker_arguments"] = [_attacker_argument_entry(a) for a in entry_point.attacker_arguments]
+    return entry
+
+
+def _attacker_argument_entry(argument: AttackerArgument) -> dict[str, Any]:
+    entry: dict[str, Any] = {}
+    if argument.argument is not None:
+        entry["argument"] = argument.argument
+    if argument.position is not None:
+        entry["position"] = argument.position
+    if argument.variadic:
+        entry["variadic"] = True
     return entry
 
 
@@ -206,9 +225,25 @@ def _entry_point(entry: Mapping[str, Any]) -> AdvisoryEntryPoint:
         str(entry["justification"]),
         tuple(_condition(c) for c in entry.get("conditions") or []),
         read,
+        tuple(_attacker_argument(a) for a in entry.get("attacker_arguments") or []),
     )
 
 
+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):
+        raise TypeError(f"attacker argument name must be a string, got {argument!r}")
+    if position is not None and (not isinstance(position, int) or isinstance(position, bool) or position < 0):
+        raise TypeError(f"attacker argument position must be a non-negative integer, got {position!r}")
+    if not isinstance(variadic, bool):
+        raise TypeError(f"attacker argument variadic must be true or false, got {variadic!r}")
+    if argument is None and position is None:
+        raise ValueError("an attacker argument needs a name, a position or both")
+    if variadic and position is None:
+        raise ValueError("a variadic attacker argument needs the position it starts at")
+    return AttackerArgument(argument, position, variadic)
+
+
 def _condition(entry: Mapping[str, Any]) -> Condition:
     position = entry.get("position")
     if position is not None and (not isinstance(position, int) or isinstance(position, bool) or position < 0):

src/coretrace_python/dependency/correlation.py

@@ -133,7 +133,10 @@ def correlate(
             continue
         hotspot = verdict is not None and verdict.status is Status.HOTSPOT
         for advisory in affected.get(flow.sink.symbol, ()):
-            check = check_conditions(advisory.entry_point(flow.sink.symbol), flow.sink_arguments)
+            entry = advisory.entry_point(flow.sink.symbol)
+            if entry is not None and not any(entry.exploitable_through(p, k) for p, k in flow.passed_as):
+                continue
+            check = check_conditions(entry, flow.sink_arguments)
             if check.contradicted is None:
                 findings.append(_exploitable(function, flow, advisory, verdict, hotspot, check))
     return tuple(findings)

src/coretrace_python/dependency/graph.py

@@ -166,19 +166,52 @@ def checkable(self) -> bool:
         return self.kind == "argument"
 
 
+@dataclass(frozen=True)
+class AttackerArgument:
+    """An argument of an entry point that must carry the attacker's input: passed by
+    keyword ``argument``, or at ``position`` (from 0, a method's receiver excluded) when it
+    may be passed positionally; a ``variadic`` one, such as ``*pathnames``, also takes every
+    later position."""
+
+    argument: str | None = None
+    position: int | None = None
+    variadic: bool = False
+
+    def receives(self, position: int | None, keyword: str | None) -> bool:
+        """Whether the argument passed at ``position`` or as ``keyword`` is this one. An
+        argument unpacked with ``*`` or ``**`` has neither and may fill this one: like an
+        argument condition it leaves undecided, it does not rule the call out, so it is."""
+
+        if keyword is not None:
+            return keyword == self.argument
+        if position is None:
+            return True
+        if self.position is None:
+            return False
+        return position == self.position or (self.variadic and position > self.position)
+
+
 @dataclass(frozen=True)
 class AdvisoryEntryPoint:
     """A public API through which a project reaches an affected symbol, justified by the
     fixing commit or by a call path, with the conditions under which it is affected.
 
     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."""
+    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."""
 
     symbol: SymbolId
     justification: str
     conditions: tuple[Condition, ...] = ()
     read: bool = False
+    attacker_arguments: tuple[AttackerArgument, ...] = ()
+
+    def exploitable_through(self, position: int | None, keyword: str | None) -> bool:
+        """Whether attacker input in the argument at ``position`` or ``keyword`` can
+        exploit the vulnerability."""
+
+        return not self.attacker_arguments or any(a.receives(position, keyword) for a in self.attacker_arguments)
 
 
 DIRECT = "affected symbol, changed by the fix"

src/coretrace_python/taint/engine.py

@@ -9,7 +9,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Iterable, Mapping
 from dataclasses import dataclass, replace
 from types import MappingProxyType
 from typing import ClassVar
@@ -119,6 +119,9 @@ class TaintFlow:
     sink_location: SourceSpan | None = None
     # What the arguments of the sink call denote, to decide the conditions on them.
     sink_arguments: Arguments | None = None
+    # How the tainted value is passed to the sink call: at a position, by a keyword, or
+    # (None, None) unpacked with ``*`` or ``**``; ``f(x, key=x)`` passes it twice.
+    passed_as: frozenset[tuple[int | None, str | None]] = frozenset()
 
 
 Key = Value | HeapLocation
@@ -559,6 +562,7 @@ def report(
                     through,
                     call.location if sink_location is None else sink_location,
                     arguments,
+                    frozenset({(position, keyword)}),
                 )
             )
 
@@ -819,7 +823,18 @@ def propagate_taint(
             state, found = problem.evaluate(block, solution.incoming(block.id))
             taints.update(state)
             flows.extend(found)
-    return TaintFacts(taints, tuple(dict.fromkeys(flows)))
+    return TaintFacts(taints, _merged(flows))
+
+
+def _merged(flows: Iterable[TaintFlow]) -> tuple[TaintFlow, ...]:
+    """One flow per source, sink, kinds and tainted value, however many arguments of the
+    sink call it is passed as."""
+
+    passed: dict[TaintFlow, frozenset[tuple[int | None, str | None]]] = {}
+    for flow in flows:
+        key = replace(flow, passed_as=frozenset())
+        passed[key] = passed.get(key, frozenset()) | flow.passed_as
+    return tuple(replace(flow, passed_as=ways) for flow, ways in passed.items())
 
 
 class TaintAnalysis(FunctionAnalysis[TaintFacts]):