From f4c9f8c29646ae82baa17c1033847105e0888201 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 25 Sep 2026 18:02:13 +0200 Subject: [PATCH 1/2] test: specify the arguments an advisory entry point needs the attacker to control --- tests/test_attacker_arguments.py | 203 +++++++++++++++++++++++++++++++ 1 file changed, 203 insertions(+) create mode 100644 tests/test_attacker_arguments.py diff --git a/tests/test_attacker_arguments.py b/tests/test_attacker_arguments.py new file mode 100644 index 0000000..76fd063 --- /dev/null +++ b/tests/test_attacker_arguments.py @@ -0,0 +1,203 @@ +"""An advisory entry point may name the arguments the attacker must control. + +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 naming its ``attacker_arguments`` — by keyword and, when they may be passed +positionally, by ``position``; a ``variadic`` one takes every position from its own on — +is exploitable only when attacker input reaches one of them, and stays reachable +otherwise. 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 it counts. An +entry point naming none counts every argument, as before. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.dependency import ( + Advisory, + AdvisoryEntryPoint, + AdvisoryFileError, + AttackerArgument, + Condition, + dump_advisories, + load_advisories, +) +from coretrace_python.findings import Finding, Severity +from coretrace_python.semantic.symbols import SymbolId + +PLUGINS = engine.BUNDLED_PLUGINS +CRAFTED = (Condition("semantic", "the URL is crafted to split its host"),) +URL = (AttackerArgument("url", 0),) + +FETCH = Advisory( + "CVE-2099-0601", + "vulnlib", + "<1.1", + "fetch sends the credentials of one host to another for a crafted URL", + Severity.HIGH, + entry_points=( + AdvisoryEntryPoint(SymbolId("python.vulnlib.fetch"), "commit abc1234", CRAFTED, attacker_arguments=URL), + AdvisoryEntryPoint( + SymbolId("python.vulnlib.Session.fetch"), + "call path: Session.fetch -> fetch (commit abc1234)", + CRAFTED, + attacker_arguments=URL, + ), + ), + modules=("vulnlib",), +) +JOIN = Advisory( + "CVE-2099-0602", + "vulnlib", + "<1.1", + "join accepts device names among the parts it joins", + Severity.MEDIUM, + entry_points=( + AdvisoryEntryPoint( + SymbolId("python.vulnlib.join"), + "commit def5678", + (Condition("semantic", "the server runs on Windows"),), + attacker_arguments=(AttackerArgument(position=1, variadic=True),), + ), + ), + modules=("vulnlib",), +) +SLOW = Advisory( + "CVE-2099-0603", + "vulnlib", + "<1.1", + "fetch is slow on any large argument", + Severity.LOW, + entry_points=( + AdvisoryEntryPoint( + SymbolId("python.vulnlib.fetch"), "commit 0a1b2c3", (Condition("semantic", "the argument is large"),) + ), + ), + modules=("vulnlib",), +) + + +def analyse(root: Path, files: dict[str, str]) -> tuple[Finding, ...]: + for relative, text in {"requirements.txt": "vulnlib==1.0\n", **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((FETCH, JOIN, SLOW)), encoding="utf-8") + return engine.analyze_project(root, [PLUGINS]).findings + + +def of_rule(findings: tuple[Finding, ...], rule_id: str) -> list[tuple[int, str]]: + return sorted((f.span.start_line, f.metadata["advisory"]) for f in findings if f.rule_id == rule_id) + + +def exploitable(root: Path, files: dict[str, str]) -> list[tuple[int, str]]: + return of_rule(analyse(root, files), "exploitable-vulnerability") + + +def calling(call: str) -> dict[str, str]: + return {"app.py": f"import vulnlib\n\ndef run():\n data = input()\n return {call}\n"} + + +@pytest.mark.parametrize( + "call, advisories", + [ + ("vulnlib.fetch(data)", ["CVE-2099-0601", "CVE-2099-0603"]), + ("vulnlib.fetch(url=data)", ["CVE-2099-0601", "CVE-2099-0603"]), + ("vulnlib.Session().fetch(data)", ["CVE-2099-0601"]), + ("vulnlib.join('uploads', data)", ["CVE-2099-0602"]), + ("vulnlib.join('uploads', 'reports', data)", ["CVE-2099-0602"]), + ], +) +def test_attacker_input_in_a_named_argument_is_exploitable(tmp_path: Path, call: str, advisories: list[str]) -> None: + assert exploitable(tmp_path, calling(call)) == [(5, a) for a in advisories] + + +@pytest.mark.parametrize( + "call, advisories, reached", + [ + ("vulnlib.fetch('https://example.com', body=data)", ["CVE-2099-0603"], "CVE-2099-0601"), + ("vulnlib.fetch('https://example.com', data)", ["CVE-2099-0603"], "CVE-2099-0601"), + ("vulnlib.Session().fetch('https://example.com', body=data)", [], "CVE-2099-0601"), + ("vulnlib.join(data, 'report.pdf')", [], "CVE-2099-0602"), + ], +) +def test_attacker_input_in_another_argument_leaves_the_call_reachable_only( + tmp_path: Path, call: str, advisories: list[str], reached: str +) -> None: + found = analyse(tmp_path, calling(call)) + + assert of_rule(found, "exploitable-vulnerability") == [(5, a) for a in advisories] + assert (5, reached) in of_rule(found, "reachable-vulnerability") + + +@pytest.mark.parametrize( + "call, advisories", + [ + ("vulnlib.fetch(*data.split(','))", ["CVE-2099-0601", "CVE-2099-0603"]), + ("vulnlib.fetch(**{'url': data})", ["CVE-2099-0601", "CVE-2099-0603"]), + ("vulnlib.join('uploads', *data.split('/'))", ["CVE-2099-0602"]), + ], +) +def test_an_unpacked_argument_may_fill_a_named_one_so_it_counts(tmp_path: Path, call: str, advisories: list[str]) -> None: + assert exploitable(tmp_path, calling(call)) == [(5, a) for a in advisories] + + +def test_a_wrapper_forwarding_its_arguments_unpacked_still_reaches_the_named_one(tmp_path: Path) -> None: + source = ( + "import vulnlib\n\n" + "def fetch(*args, **kwargs):\n return vulnlib.fetch(*args, timeout=5, **kwargs)\n\n" + "def run():\n fetch(input())\n" + ) + + assert exploitable(tmp_path, {"app.py": source}) == [(7, "CVE-2099-0601"), (7, "CVE-2099-0603")] + + +def test_the_argument_of_the_entry_point_call_decides_when_the_flow_goes_through_a_function(tmp_path: Path) -> None: + source = ( + "import vulnlib\n\n" + "def get(url, body):\n return vulnlib.fetch(url, body=body)\n\n" + "def run():\n get('https://example.com', input())\n get(input(), None)\n" + ) + + assert exploitable(tmp_path, {"app.py": source}) == [ + (7, "CVE-2099-0603"), + (8, "CVE-2099-0601"), + (8, "CVE-2099-0603"), + ] + + +def test_attacker_arguments_round_trip_through_advisory_files(tmp_path: Path) -> None: + path = tmp_path / "advisories.json" + path.write_text(dump_advisories((FETCH, JOIN, SLOW)), encoding="utf-8") + + assert load_advisories(path) == (FETCH, JOIN, SLOW) + entries = [e for a in json.loads(path.read_text())["advisories"] for e in a["entry_points"]] + assert entries[0]["attacker_arguments"] == [{"argument": "url", "position": 0}] + assert entries[2]["attacker_arguments"] == [{"position": 1, "variadic": True}] + assert "attacker_arguments" not in entries[3] + + +@pytest.mark.parametrize( + "arguments", + [ + [{}], + [{"position": -1}], + [{"position": True}], + [{"argument": "url", "variadic": True}], + [{"position": 1, "variadic": "yes"}], + [{"argument": 7}], + ], +) +def test_a_malformed_attacker_argument_is_rejected(tmp_path: Path, arguments: list[object]) -> None: + document = json.loads(dump_advisories((FETCH,))) + document["advisories"][0]["entry_points"][0]["attacker_arguments"] = arguments + path = tmp_path / "advisories.json" + path.write_text(json.dumps(document), encoding="utf-8") + + with pytest.raises(AdvisoryFileError): + load_advisories(path) From 3f3aa131e453fb01685001f3a96446820d2cddc1 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 25 Sep 2026 18:02:13 +0200 Subject: [PATCH 2/2] feat: let advisory entry points name the arguments the attacker must control An advisory entry point was an ADVISORY sink on every argument, so attacker data in any argument made the call exploitable. An entry point may now name its attacker_arguments, by keyword and position, a variadic one taking every later position; attacker input elsewhere leaves the call reachable. Each taint flow records how its value is passed to the sink, and flows differing only in that are merged, so detectors see the same flows as before. --- CHANGELOG.md | 6 +++ docs/usage.md | 13 +++++++ src/coretrace_python/dependency/__init__.py | 2 + src/coretrace_python/dependency/advisories.py | 37 ++++++++++++++++++- .../dependency/correlation.py | 5 ++- src/coretrace_python/dependency/graph.py | 35 +++++++++++++++++- src/coretrace_python/taint/engine.py | 19 +++++++++- 7 files changed, 112 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b4c47dc..51cd395 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/usage.md b/docs/usage.md index b881cc4..995baa9 100644 --- a/docs/usage.md +++ b/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 diff --git a/src/coretrace_python/dependency/__init__.py b/src/coretrace_python/dependency/__init__.py index f399fb9..7339c08 100644 --- a/src/coretrace_python/dependency/__init__.py +++ b/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", diff --git a/src/coretrace_python/dependency/advisories.py b/src/coretrace_python/dependency/advisories.py index edfa4fd..9853add 100644 --- a/src/coretrace_python/dependency/advisories.py +++ b/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): diff --git a/src/coretrace_python/dependency/correlation.py b/src/coretrace_python/dependency/correlation.py index f1e98e5..001f3da 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/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) diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index 220ddcd..853dfec 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -166,6 +166,31 @@ 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 @@ -173,12 +198,20 @@ 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.""" + 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" diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 4e00f80..21f9a45 100644 --- a/src/coretrace_python/taint/engine.py +++ b/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]):