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]): 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)