diff --git a/CHANGELOG.md b/CHANGELOG.md index 935943e..c065fb2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # 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). + ## 0.20.0 (2026-09-30) ### Advisories diff --git a/docs/plugins.md b/docs/plugins.md index 8816b8d..d64cf63 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -195,7 +195,8 @@ 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 +`argument` (on the value, or with `present` on the argument being passed at all), +`host`, `semantic` or `keyword_name`, and `attacker_arguments`, decided as the [usage guide](usage.md) describes. ### Reclassifying findings: `ProjectPlugin.refine` diff --git a/docs/usage.md b/docs/usage.md index e4271c3..dadaa5e 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -471,6 +471,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 diff --git a/src/coretrace_python/dependency/advisories.py b/src/coretrace_python/dependency/advisories.py index b0bab34..9a615e3 100644 --- a/src/coretrace_python/dependency/advisories.py +++ b/src/coretrace_python/dependency/advisories.py @@ -187,6 +187,8 @@ def _condition_entry(condition: Condition) -> dict[str, Any]: entry["position"] = condition.position if condition.default: entry["default"] = True + if condition.present: + entry["present"] = True return entry @@ -251,6 +253,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 +263,15 @@ def _condition(entry: Mapping[str, Any]) -> Condition: tuple(str(v) for v in entry.get("values") or []), position, default, + present, ) + if condition.present and ( + condition.kind != "argument" + or (condition.argument is None and condition.position is None) + or condition.values + or condition.default + ): + raise ValueError("a present condition names the argument that must be passed, by keyword or position, and nothing else") if condition.kind == "host" and ( (condition.argument is None and condition.position is None) or condition.values or condition.default ): diff --git a/src/coretrace_python/dependency/correlation.py b/src/coretrace_python/dependency/correlation.py index 69d0130..6b552d6 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/src/coretrace_python/dependency/correlation.py @@ -75,7 +75,9 @@ 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.""" if entry is None: return ConditionCheck() @@ -105,6 +107,12 @@ def check_conditions( 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) diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index ed4af09..e1360db 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -151,7 +151,9 @@ 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, @@ -165,6 +167,7 @@ class Condition: values: tuple[str, ...] = () position: int | None = None default: bool = False + present: bool = False @property def checkable(self) -> bool: 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)