feat: decide an argument condition on the argument being passed

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

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`

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

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,14 +253,25 @@ 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"]),
         None if entry.get("argument") is None else str(entry["argument"]),
         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
     ):

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)

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:

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)