feat: taint the keys of a **mapping as the keyword names of the call

CHANGELOG.md

@@ -1,5 +1,19 @@
 # Changelog
 
+## Unreleased
+
+### Advisories
+
+- An advisory condition of kind `keyword_name` says the attacker must choose a keyword name of the call, as Django's ORM takes column aliases and lookups (`annotate(**{alias: expr})`); it carries its text and nothing else. The engine decides it from the keys of what the call expands with `**`, apart from the mapping's values: keys carrying attacker input meet it and the call is exploitable; constant keys, a dict literal with string keys, `dict(total=...)` or plain keywords, contradict it whatever the values carry, so the call is ruled out (`app.views:12 python.app.models.Item.objects.annotate(keywords=total)`, `(keywords absent)` for a call passing no keyword) and a project passing only those stays `not_affected`; keys the engine cannot establish, a `**extra` parameter, a mapping built elsewhere, `dict(zip(...))`, leave it pending review, so the call is reachable, and exploitable when the mapping's values carry input (#193).
+
+### Analysis
+
+- The keys of a mapping carry taint of their own, apart from its values: a dict literal `{alias: expr}`, `dict([(alias, expr)])`, a comprehension `{k: expr for k in input}` and a subscript assignment `mapping[alias] = expr` taint the `keys` location of the mapping's abstract object, which `{**mapping}`, `dict(mapping)` and `dict(**mapping)` copy; the value taint of a literal is what it was. A call expanding such a mapping with `**` passes its keys as the keyword names of the call, a way of passing only an advisory sink reads. Keys through a project wrapper are not followed; `mapping.update(other)`, `mapping.setdefault(k, v)`, `for k in mapping` and `mapping.keys()` carry no key taint; a mapping bound by a `Phi` is unknown (#193).
+
+### Plugins
+
+- `Arguments.keyword_unpacked` says a call expands a `**mapping` whose keys it does not write; a `**` literal with constant keys that nothing else uses is recorded as its keywords, so `given()` decides safe arguments and argument conditions for it. The cache format changes.
+
 ## 0.19.0 (2026-09-29)
 
 ### Analysis

docs/plugins.md

@@ -194,7 +194,9 @@ A model plugin may also carry `advisories`, a tuple of `Advisory` values
 (`coretrace_python.dependency`) with the package, the vulnerable range and the affected
 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.
+calls become exploitable ones. An advisory's entry points carry `conditions`, of kind
+`argument`, `host`, `semantic` or `keyword_name`, and `attacker_arguments`, decided as
+the [usage guide](usage.md) describes.
 
 ### Reclassifying findings: `ProjectPlugin.refine`
 

docs/usage.md

@@ -488,6 +488,32 @@ attacker's input is passed in that argument. The outcome has three cases:
 
 The other conditions, the environment included, stay pending as before.
 
+A `keyword_name` condition says the attacker must choose a keyword name of the call,
+as Django's ORM takes column aliases and lookups (`annotate(**{alias: expr})`):
+`{"kind": "keyword_name", "text": "…"}`, and nothing else. The engine decides it from
+the keys of what the call expands with `**`; the values of the mapping do not decide
+it. The outcome has three cases:
+
+- Keys carrying attacker input meet it, and the call is exploitable. Key taint is
+  carried, apart from value taint, for a dict literal `{alias: expr}`,
+  `dict([(alias, expr)])`, a comprehension `{k: expr for k in request.GET}` and a
+  subscript assignment `mapping[alias] = expr`, and copied by `{**mapping}` and
+  `dict(mapping)`.
+- Constant keys — a dict literal with string keys, `dict(total=…)`, plain keywords —
+  contradict it, whatever the values carry: the call is ruled out
+  (`app.views:12 python.app.models.Item.objects.annotate(keywords=total)`, or
+  `(keywords absent)` for a call passing no keyword), and a project passing only those
+  stays `not_affected`.
+- Keys the engine cannot establish — a `**extra` parameter, a mapping built elsewhere
+  or returned by a function, `dict(zip(...))`, `**request.GET` — leave it pending
+  review, as any unknown argument does: the call is reachable, and exploitable when the
+  mapping's values carry attacker input.
+
+Keys passed through a project wrapper stay pending, and `mapping.update(other)`,
+`mapping.setdefault(k, v)`, `for k in mapping` and `mapping.keys()` carry no key
+taint. A literal the function fills after building it (`m = {}; m[alias] = expr`)
+has keys the literal does not write, so the call is not ruled out.
+
 Every dependency finding records the highest level of evidence established in its
 `level` metadata: `declared` (the requirement allows a vulnerable version), `imported`
 (a module of the package is imported somewhere), `reachable` (an entry point or affected

src/coretrace_python/abstract/__init__.py

@@ -8,6 +8,7 @@
 from coretrace_python.abstract.heap import (
     ATTRIBUTES,
     ELEMENTS,
+    KEYS,
     MUTATORS,
     AbstractObject,
     AliasSet,
@@ -24,6 +25,7 @@
 __all__ = [
     "ATTRIBUTES",
     "ELEMENTS",
+    "KEYS",
     "MUTATORS",
     "AbstractObject",
     "AbstractValue",

src/coretrace_python/abstract/heap.py

@@ -6,7 +6,8 @@
 Each value points to an ``AliasSet`` of objects, computed by a flow-insensitive
 points-to fixpoint over the SSA form; each object has two ``HeapLocation`` fields,
 ``elements`` for its items and ``attributes`` for its attributes, field-insensitive
-within each. Taint and dependence analyses key their states by these locations, so a
+within each, and a third, ``keys``, for the keys of a mapping, which only the taint
+engine fills. Taint and dependence analyses key their states by these locations, so a
 store, a mutating method call or a load on any alias reads and writes the same place.
 """
 
@@ -47,6 +48,8 @@
 
 ELEMENTS = "elements"
 ATTRIBUTES = "attributes"
+# The keys of a mapping, stored by the taint engine; the points-to solver never fills it.
+KEYS = "keys"
 
 # Method names that store their arguments into the receiver's elements.
 MUTATORS = frozenset(

src/coretrace_python/cache.py

@@ -42,7 +42,7 @@
 from coretrace_python.semantic.symbols import SymbolId
 from coretrace_python.source import SourceId, SourceSpan
 
-CACHE_FORMAT = 11
+CACHE_FORMAT = 12
 
 
 @dataclass(frozen=True)
@@ -368,18 +368,20 @@ def _encode_arguments(arguments: Arguments) -> dict[str, Any]:
         "positional": list(arguments.positional),
         "keywords": [[name, value] for name, value in arguments.keywords],
         "unpacked": arguments.unpacked,
+        "keyword_unpacked": arguments.keyword_unpacked,
     }
 
 
 def _decode_arguments(data: Mapping[str, Any]) -> Arguments:
     def denoted(value: Any) -> str | None:
         return None if value is None else _string(value)
 
-    unpacked = data["unpacked"]
-    if not isinstance(unpacked, bool):
-        raise TypeError(f"expected a boolean, got {unpacked!r}")
+    unpacked, keyword_unpacked = data["unpacked"], data["keyword_unpacked"]
+    if not isinstance(unpacked, bool) or not isinstance(keyword_unpacked, bool):
+        raise TypeError(f"expected booleans, got {unpacked!r} and {keyword_unpacked!r}")
     return Arguments(
         tuple(denoted(value) for value in data["positional"]),
         tuple((_string(name), denoted(value)) for name, value in data["keywords"]),
         unpacked,
+        keyword_unpacked,
     )

src/coretrace_python/dependency/advisories.py

@@ -263,4 +263,8 @@ def _condition(entry: Mapping[str, Any]) -> Condition:
         (condition.argument is None and condition.position is None) or condition.values or condition.default
     ):
         raise ValueError("a host condition names the URL argument, by keyword or position, and nothing else")
+    if condition.kind == "keyword_name" and (
+        condition.argument is not None or condition.position is not None or condition.values or condition.default
+    ):
+        raise ValueError("a keyword_name condition carries its text and nothing else")
     return condition

src/coretrace_python/dependency/correlation.py

@@ -23,7 +23,7 @@
 from coretrace_python.findings.refutation import Status, Verdict, Verdicts
 from coretrace_python.interprocedural import Arguments, CallSite, ExternalSymbol
 from coretrace_python.semantic.symbols import SymbolId
-from coretrace_python.taint import Sink, TaintFlow, TaintKind
+from coretrace_python.taint import KEYWORD_NAMES, Sink, TaintFlow, TaintKind
 from coretrace_python.taint.urls import UrlProof, redirects_disabled
 
 Affected = Mapping[SymbolId, tuple[Advisory, ...]]
@@ -72,13 +72,25 @@ def check_conditions(
     a ``host`` condition, what ``url`` proves of the value the attacker's input is passed
     in (``passed_as``). A host the URL's constant text fixes rules the call out when the
     call disables redirects; otherwise a redirect may still lead to a host the attacker
-    controls, and the condition stays pending with that uncertainty."""
+    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."""
 
     if entry is None:
         return ConditionCheck()
     met: list[Condition] = []
     pending: list[Condition] = []
     for condition in entry.conditions:
+        if condition.kind == "keyword_name":
+            if any(keyword == KEYWORD_NAMES for _, keyword in passed_as):
+                met.append(condition)
+            elif arguments is None or arguments.keyword_unpacked:
+                pending.append(condition)
+            else:
+                names = ", ".join(sorted({name for name, _ in arguments.keywords}))
+                return ConditionCheck(tuple(met), tuple(pending), condition, names or None)
+            continue
         if condition.kind == "host":
             origin = url.origin if url is not None and _passed_in(condition, passed_as) else None
             if origin is None:
@@ -119,7 +131,8 @@ def ruled_out(module: str, site: CallSite, check: ConditionCheck) -> str:
     python.yaml.load(Loader=python.yaml.SafeLoader)``."""
 
     assert check.contradicted is not None and isinstance(site.target, ExternalSymbol)
-    argument = check.contradicted.argument or f"#{check.contradicted.position}"
+    contradicted = check.contradicted
+    argument = contradicted.argument or ("keywords" if contradicted.kind == "keyword_name" else f"#{contradicted.position}")
     passed = f"{argument} absent" if check.passed is None else f"{argument}={check.passed}"
     return f"{module}:{site.location.start_line} {site.target.symbol}({passed})"
 

src/coretrace_python/dependency/graph.py

@@ -154,7 +154,10 @@ class Condition:
     ``default`` says an absent argument means a vulnerable value. 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
-    ``semantic`` condition cannot be checked and is reported as pending review."""
+    ``keyword_name`` condition says the attacker must choose a keyword name of the call,
+    as an ORM alias expanded from a mapping; the engine decides it from the keys of what
+    the call expands with ``**``. A ``semantic`` condition cannot be checked and is
+    reported as pending review."""
 
     kind: str
     text: str

src/coretrace_python/interprocedural/callgraph.py

@@ -18,6 +18,7 @@
 from coretrace_python.analysis import Analysis, AnalysisContext, AnyAnalysis
 from coretrace_python.cfg import CFGError
 from coretrace_python.hir import nodes
+from coretrace_python.ir.defuse import DefUse, def_use
 from coretrace_python.ir.lowering import (
     MODULE_BODY,
     LoweringError,
@@ -26,6 +27,7 @@
 )
 from coretrace_python.ir.model import (
     Await,
+    BuildDict,
     Call,
     Constant,
     FunctionIR,
@@ -68,11 +70,15 @@ class Arguments:
     (``python.yaml.FullLoader``), a constant as Python writes it (``True``, ``'/static'``),
     or None for anything else (a parameter, an expression). ``unpacked`` says the call
     unpacks ``*args`` or ``**kwargs``, so an argument it does not give explicitly may
-    still be given; after ``*args`` no position is known, and ``positional`` is empty."""
+    still be given; after ``*args`` no position is known, and ``positional`` is empty.
+    ``keyword_unpacked`` says the call expands a ``**mapping`` whose keys it does not
+    write, so a keyword name may be anything; a literal with constant keys is written out
+    as its keywords, and ``unpacked`` stays true for it only when ``*args`` is also given."""
 
     positional: tuple[str | None, ...] = ()
     keywords: tuple[tuple[str, str | None], ...] = ()
     unpacked: bool = False
+    keyword_unpacked: bool = False
 
     def given(self, keyword: str | None, position: int | None = None) -> tuple[bool, str | None] | None:
         """Whether the call gives an argument, by ``keyword`` or at ``position``, and what it
@@ -245,8 +251,45 @@ def derive_symbols(
 # passed as an argument is not recorded, so call sites stay small in the cache.
 MAX_CONSTANT_LENGTH = 64
 
+_DICT = SymbolId("python.builtins.dict")
+
+
+def _keyword_entries(
+    value: Value, symbols: Mapping[Value, SymbolId], defs: Mapping[Value, Instruction], uses: DefUse
+) -> tuple[tuple[str, Value], ...] | None:
+    """The keywords a ``**value`` gives when every name is established: the string keys
+    of a dict literal, the keywords of ``dict(...)``, through what they unpack in turn;
+    None when a key is computed, the mapping is built elsewhere, or anything but the
+    call uses it, since a store or a mutating call may add a key the literal does not
+    write."""
+
+    if len(uses.uses(value)) != 1:
+        return None
+    made = defs.get(value)
+    entries: list[tuple[str, Value]] = []
+    if isinstance(made, BuildDict):
+        for key, item in made.items:
+            constant = defs.get(key)
+            if not isinstance(constant, Constant) or not isinstance(constant.value, str):
+                return None
+            entries.append((constant.value, item))
+        nested = made.unpacked
+    elif isinstance(made, Call) and symbols.get(made.callee) == _DICT and not made.arguments and not made.starred:
+        nested = tuple(v for name, v in made.keywords if name is None)
+        entries.extend((name, v) for name, v in made.keywords if name is not None)
+    else:
+        return None
+    for inner in nested:
+        found = _keyword_entries(inner, symbols, defs, uses)
+        if found is None:
+            return None
+        entries.extend(found)
+    return tuple(entries)
+
 
-def _arguments(call: Call, symbols: Mapping[Value, SymbolId], defs: Mapping[Value, Instruction]) -> Arguments:
+def _arguments(
+    call: Call, symbols: Mapping[Value, SymbolId], defs: Mapping[Value, Instruction], uses: DefUse
+) -> Arguments:
     def denoted(value: Value) -> str | None:
         symbol = symbols.get(value)
         if symbol is not None:
@@ -257,10 +300,22 @@ def denoted(value: Value) -> str | None:
         written = repr(made.value)
         return written if len(written) <= MAX_CONSTANT_LENGTH else None
 
+    keywords: list[tuple[str, str | None]] = []
+    keyword_unpacked = False
+    for name, value in call.keywords:
+        if name is not None:
+            keywords.append((name, denoted(value)))
+            continue
+        entries = _keyword_entries(value, symbols, defs, uses)
+        if entries is None:
+            keyword_unpacked = True
+        else:
+            keywords.extend((key, denoted(item)) for key, item in entries)
     return Arguments(
         () if call.starred else tuple(denoted(value) for value in call.arguments),
-        tuple((name, denoted(value)) for name, value in call.keywords if name is not None),
-        bool(call.starred) or any(name is None for name, _ in call.keywords),
+        tuple(keywords),
+        bool(call.starred) or keyword_unpacked,
+        keyword_unpacked,
     )
 
 
@@ -480,6 +535,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph:
                 table=table,
             )
             defs = {i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None}
+            uses = def_use(ssa)
             found: list[CallSite] = []
             for block in ssa.blocks:
                 for instruction in block.instructions:
@@ -489,7 +545,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph:
                                 name,
                                 instruction.location,
                                 targets.get(instruction.callee, UnknownTarget()),
-                                _arguments(instruction, symbols[name], defs),
+                                _arguments(instruction, symbols[name], defs, uses),
                             )
                         )
             sites[name] = tuple(found)

src/coretrace_python/taint/__init__.py

@@ -2,6 +2,7 @@
 
 from coretrace_python.semantic.symbols import Members
 from coretrace_python.taint.engine import (
+    KEYWORD_NAMES,
     EntryPointAnalysis,
     Taint,
     TaintAnalysis,
@@ -46,6 +47,7 @@
 
 __all__ = [
     "FILTER_FUNCTIONS",
+    "KEYWORD_NAMES",
     "TEXT_KINDS",
     "AuthorizationGuard",
     "BasePattern",