diff --git a/CHANGELOG.md b/CHANGELOG.md index b36f515..21a371e 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/plugins.md b/docs/plugins.md index a765f1b..8816b8d 100644 --- a/docs/plugins.md +++ b/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` diff --git a/docs/usage.md b/docs/usage.md index 3d48bb1..0c95047 100644 --- a/docs/usage.md +++ b/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 diff --git a/src/coretrace_python/abstract/__init__.py b/src/coretrace_python/abstract/__init__.py index 7e156e8..bf94d80 100644 --- a/src/coretrace_python/abstract/__init__.py +++ b/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", diff --git a/src/coretrace_python/abstract/heap.py b/src/coretrace_python/abstract/heap.py index 7baf8e0..71dafbc 100644 --- a/src/coretrace_python/abstract/heap.py +++ b/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( diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index 10d505c..e15c1b4 100644 --- a/src/coretrace_python/cache.py +++ b/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,6 +368,7 @@ 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, } @@ -375,11 +376,12 @@ 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, ) diff --git a/src/coretrace_python/dependency/advisories.py b/src/coretrace_python/dependency/advisories.py index 0a92bd9..b0bab34 100644 --- a/src/coretrace_python/dependency/advisories.py +++ b/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 diff --git a/src/coretrace_python/dependency/correlation.py b/src/coretrace_python/dependency/correlation.py index 7ce0ba4..69d0130 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/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})" diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index 5aafb77..ed4af09 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/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 diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index 6d73469..300656c 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/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) diff --git a/src/coretrace_python/taint/__init__.py b/src/coretrace_python/taint/__init__.py index 6114195..81bb361 100644 --- a/src/coretrace_python/taint/__init__.py +++ b/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", diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index d0ec2f6..f38c672 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -18,6 +18,7 @@ from coretrace_python.abstract import ( ATTRIBUTES, ELEMENTS, + KEYS, HeapAnalysis, HeapFacts, HeapLocation, @@ -130,12 +131,18 @@ class TaintFlow: # 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. + # (None, None) unpacked with ``*`` or ``**``; ``f(x, key=x)`` passes it twice. The + # keyword ``KEYWORD_NAMES`` says the value's keys are the keyword names of the call. passed_as: frozenset[tuple[int | None, str | None]] = frozenset() Key = Value | HeapLocation +# The ``keyword`` of a ``passed_as`` way whose value's keys become the keyword names of +# the call: ``f(**m)`` passes the keys of ``m`` as ``(None, KEYWORD_NAMES)``. +KEYWORD_NAMES = "**" +_DICT = SymbolId("python.builtins.dict") + class TaintFacts: def __init__(self, taints: Mapping[Key, Taint], flows: tuple[TaintFlow, ...]) -> None: @@ -217,6 +224,14 @@ def loaded(self, instruction: GetAttr | GetItem | GetIter, state: Mapping[Key, T taint = taint.join(state.get(location, Taint.none())) return taint + def keyed(self, value: Value, state: Mapping[Key, Taint]) -> Taint: + """The taint of the keys of the mappings ``value`` points to.""" + + taint = Taint.none() + for location in self.heap.locations(value, KEYS): + taint = taint.join(state.get(location, Taint.none())) + return taint + def initial(self) -> State: state: dict[Key, Taint] = { self.function.parameters[index]: Taint(source.kinds, frozenset({source})) @@ -246,6 +261,17 @@ def evaluate( self.store(state, instruction.object, ATTRIBUTES, state.get(instruction.value, Taint.none())) elif isinstance(instruction, SetItem): self.store(state, instruction.object, ELEMENTS, state.get(instruction.value, Taint.none())) + key = state.get(instruction.key, Taint.none()) + if key: + self.store(state, instruction.object, KEYS, key) + elif isinstance(instruction, BuildDict): + keys = Taint.none() + for item_key, _ in instruction.items: + keys = keys.join(state.get(item_key, Taint.none())) + for unpacked in instruction.unpacked: # ``{**d}`` copies d's keys + keys = keys.join(self.keyed(unpacked, state)) + if keys: + self.store(state, instruction.result, KEYS, keys) if instruction.result is None: continue if isinstance(instruction, Phi): @@ -352,6 +378,10 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T project, through, arguments, keywords, everything, call, flows, state, (), bound ) return self.declared_clean(target.symbol, result) + if target.symbol == _DICT: + keys = self.constructed_keys(call, state) + if keys: + self.store(state, call.result, KEYS, keys) symbol = target.symbol if not self.modelled(symbol): # ``get_conn().execute`` derived ``app.database.get_conn.execute``; what @@ -376,6 +406,23 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T return self.external(returned, everything, call, state, flows) return everything.join(state.get(call.callee, Taint.none())) + def constructed_keys(self, call: Call, state: Mapping[Key, Taint]) -> Taint: + """What the keys of ``dict(...)`` carry: the keys of the mapping it copies + (``dict(m)``, ``dict(**m)``) and the first element of each pair of a sequence + literal (``dict([(k, v)])``); a keyword is a constant key, and an iterable built + elsewhere (``zip(...)``, a parameter) gives keys the engine cannot establish.""" + + taint = Taint.none() + for value in (*call.arguments[:1], *(v for name, v in call.keywords if name is None)): + taint = taint.join(self.keyed(value, state)) + made = self.defs.get(value) + if isinstance(made, BuildList | BuildTuple | BuildSet) and not made.unpacked: + for element in made.elements: + pair = self.defs.get(element) + if isinstance(pair, BuildTuple | BuildList) and pair.elements: + taint = taint.join(state.get(pair.elements[0], Taint.none())) + return taint + def nonlocal_result(self, instruction: NonlocalResult, state: dict[Key, Taint]) -> Taint: """The taint a nested callee left in its nonlocal ``name``, in the caller's terms.""" @@ -455,6 +502,11 @@ def external( self.report( flows, sink, self.rendered(argument, state), argument, call, None, None, None, given, keyword=name ) + if name is None: + # The keys of a ``**mapping`` are keyword names of the call: only an + # advisory asks who chooses a name, no other sink reads one. + keys = self.keyed(argument, state).without(~TaintKind.ADVISORY) + self.report(flows, sink, keys, argument, call, None, None, None, given, keyword=KEYWORD_NAMES) self.template_filters(symbol, call, given, state, flows) cleared = self.clearing.clears(symbol, given) if cleared: diff --git a/tests/test_key_taint.py b/tests/test_key_taint.py new file mode 100644 index 0000000..c219b3c --- /dev/null +++ b/tests/test_key_taint.py @@ -0,0 +1,229 @@ +"""The keys of a mapping carry taint of their own (issue #193). + +A dict literal ``{taint: value}``, ``dict([(taint, value)])``, a comprehension ``{k: v for +k in taint}`` and a subscript assignment ``mapping[taint] = value`` taint the ``keys`` +field of the mapping's abstract object, apart from its ``elements`` and from the value +taint the mapping keeps as before; ``{**d}``, ``dict(d)`` and ``dict(**d)`` copy it. A +call expanding such a mapping with ``**`` passes its keys as the keyword names of the +call, the ``(None, "**")`` way of the flow, which only an advisory sink reads. Call sites +record whether every keyword name is written (``Arguments.keyword_unpacked``): a ``**`` +literal with constant keys that nothing else touches is recorded as its keywords. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.abstract import ELEMENTS, HeapAnalysis +from coretrace_python.abstract.heap import AbstractObject, HeapLocation +from coretrace_python.analysis import AnalysisManager +from coretrace_python.cache import CachedModule, ProjectCache +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.interprocedural import Arguments, CallGraphAnalysis, CallSite, ExternalSymbol +from coretrace_python.ir.model import Return +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceId, SourceManager, SourceSpan +from coretrace_python.taint import ( + SecurityModelAnalysis, + SecurityModelRegistry, + Sink, + Source, + TaintAnalysis, + TaintFacts, + TaintFlow, + TaintKind, +) + +try: + from coretrace_python.abstract import KEYS + from coretrace_python.taint import KEYWORD_NAMES +except ImportError as error: # pragma: no cover - red until key taint lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_key_taint() -> None: + if MISSING is not None: + pytest.fail(f"key taint is not implemented yet: {MISSING}") + + +MODELS = ( + Source(SymbolId("python.builtins.input"), "stdin"), + Sink(SymbolId("python.os.system"), TaintKind.COMMAND), + Sink(SymbolId("python.vulnlib.sink"), TaintKind.ADVISORY), +) +PRELUDE = "import os\nimport vulnlib\n\n" + + +def manager_for(body: str) -> AnalysisManager: + module = build_hir(SourceManager().add_source("m.py", PRELUDE + body)) + manager = AnalysisManager(module) + manager.register(*engine.ALL_ANALYSES) + registry = SecurityModelRegistry() + registry.register(*MODELS) + manager.provide(SecurityModelAnalysis, registry.freeze()) + return manager + + +def function_named(manager: AnalysisManager, name: str) -> nodes.Function: + return next(s for s in manager.module.body if isinstance(s, nodes.Function) and s.name == name) + + +def returned_mapping(body: str) -> tuple[TaintFacts, frozenset[AbstractObject]]: + """The taint facts of ``f`` and the objects of the mapping it returns.""" + + manager = manager_for(body) + function = function_named(manager, "f") + ssa = manager.get(SSAAnalysis, function) + returned = next(b.terminator.value for b in ssa.blocks if isinstance(b.terminator, Return)) + assert returned is not None + objects = manager.get(HeapAnalysis, function).objects(returned) + assert objects + return manager.get(TaintAnalysis, function), objects + + +def flows_of(body: str) -> tuple[TaintFlow, ...]: + manager = manager_for(body) + return manager.get(TaintAnalysis, function_named(manager, "f")).flows + + +def arguments_of(call: str, before: str = "") -> Arguments: + manager = manager_for(f"def run(m, xs, k):\n{before} {call}\n") + graph = manager.get(CallGraphAnalysis) + site = next(s for s in graph.sites("run") if s.target == ExternalSymbol(SymbolId("python.vulnlib.f"))) + return site.arguments + + +# --------------------------------------------------------------------------- key taint + + +@pytest.mark.parametrize( + "body", + [ + "def f():\n d = {input(): 1}\n return d\n", + "def f():\n d = dict([(input(), 1)])\n return d\n", + "def f():\n d = {k: 1 for k in input()}\n return d\n", + "def f():\n d = {}\n d[input()] = 1\n return d\n", + "def f():\n m = {input(): 1}\n d = {**m}\n return d\n", + "def f():\n m = {input(): 1}\n d = dict(m)\n return d\n", + "def f():\n m = {input(): 1}\n d = dict(**m)\n return d\n", + ], + ids=["literal", "pairs", "comprehension", "subscript-assignment", "unpacked", "dict-copy", "dict-expanded"], +) +def test_attacker_keys_taint_the_keys_of_the_mapping_and_not_its_elements(body: str) -> None: + facts, objects = returned_mapping(body) + + assert all(facts.heap(HeapLocation(o, KEYS)).kinds == TaintKind.ALL for o in objects) + assert not any(facts.heap(HeapLocation(o, ELEMENTS)) for o in objects) + + +def test_attacker_values_leave_the_keys_clean() -> None: + facts, objects = returned_mapping("def f():\n d = {'k': input()}\n return d\n") + + assert not any(facts.heap(HeapLocation(o, KEYS)) for o in objects) + + +def test_a_subscript_assignment_taints_the_keys_apart_from_the_mapping_itself() -> None: + body = "def f():\n d = {}\n d[input()] = 1\n return d\n" + manager = manager_for(body) + function = function_named(manager, "f") + ssa = manager.get(SSAAnalysis, function) + returned = next(b.terminator.value for b in ssa.blocks if isinstance(b.terminator, Return)) + assert returned is not None + (obj,) = manager.get(HeapAnalysis, function).objects(returned) + + facts = manager.get(TaintAnalysis, function) + + assert facts.heap(HeapLocation(obj, KEYS)).kinds == TaintKind.ALL + assert not facts.heap(HeapLocation(obj, ELEMENTS)) + assert not facts.taint(returned) + + +# --------------------------------------------------------------------------- flows + + +def test_the_keyword_names_way_is_the_double_star() -> None: + assert KEYWORD_NAMES == "**" + + +@pytest.mark.parametrize( + "body, ways", + [ + ("def f():\n d = {input(): 1}\n vulnlib.sink(**d)\n", {(None, None), (None, "**")}), + ("def f():\n d = {}\n d[input()] = 1\n vulnlib.sink(**d)\n", {(None, "**")}), + ("def f():\n d = {'k': input()}\n vulnlib.sink(**d)\n", {(None, None)}), + ], + ids=["tainted-key-literal", "tainted-keys-only", "tainted-value"], +) +def test_keys_expanded_into_an_advisory_sink_are_passed_as_its_keyword_names( + body: str, ways: set[tuple[int | None, str | None]] +) -> None: + (flow,) = flows_of(body) + + assert flow.kinds == TaintKind.ADVISORY + assert flow.passed_as == frozenset(ways) + + +def test_no_other_sink_reads_the_keyword_names() -> None: + assert flows_of("def f():\n d = {}\n d[input()] = 1\n os.system(**d)\n") == () + + +# --------------------------------------------------------------------------- call sites + + +@pytest.mark.parametrize( + "call, before, expected", + [ + ('vulnlib.f(**{"a": 1})', "", ((), (("a", "1"),), False, False)), + ('vulnlib.f(1, **{"a": 1})', "", (("1",), (("a", "1"),), False, False)), + ("vulnlib.f(**m)", "", ((), (), True, True)), + ("vulnlib.f(*xs)", "", ((), (), True, False)), + ('vulnlib.f(*xs, **{"a": 1})', "", ((), (("a", "1"),), True, False)), + ('vulnlib.f(**dict(a=1, **{"b": 2}))', "", ((), (("a", "1"), ("b", "2")), False, False)), + ("vulnlib.f(**{k: 1})", "", ((), (), True, True)), + ("vulnlib.f(**dict(m))", "", ((), (), True, True)), + ("vulnlib.f(**opts)", ' opts = {"a": 1}\n', ((), (("a", "1"),), False, False)), + # A literal filled after its construction has keys the literal does not write. + ("vulnlib.f(**opts)", ' opts = {"a": 1}\n opts[k] = 2\n', ((), (), True, True)), + ("vulnlib.f(**opts)", " opts = {}\n opts[k] = 1\n", ((), (), True, True)), + ], + ids=[ + "literal", + "literal-after-positional", + "parameter", + "starred", + "starred-and-literal", + "dict-with-nested-literal", + "computed-key", + "dict-copy", + "literal-by-name", + "literal-then-assigned", + "empty-literal-then-assigned", + ], +) +def test_call_sites_record_whether_every_keyword_name_is_written( + call: str, before: str, expected: tuple[object, ...] +) -> None: + found = arguments_of(call, before) + + assert (found.positional, found.keywords, found.unpacked, found.keyword_unpacked) == expected + + +def test_keyword_unpacked_round_trips_through_the_cache(tmp_path: Path) -> None: + span = SourceSpan(SourceId("/p/a.py"), 3, 5, 3, 9) + site = CallSite("f", span, ExternalSymbol(SymbolId("python.vulnlib.f")), Arguments((), (), True, True)) + cache = ProjectCache(tmp_path / "cache") + + cache.store("key", CachedModule((), {}, (site,), ())) + restored = cache.load("key") + + assert restored is not None + assert restored.sites == (site,) + assert restored.sites[0].arguments.keyword_unpacked is True diff --git a/tests/test_keyword_name_conditions.py b/tests/test_keyword_name_conditions.py new file mode 100644 index 0000000..29e6171 --- /dev/null +++ b/tests/test_keyword_name_conditions.py @@ -0,0 +1,243 @@ +"""An advisory condition of kind ``keyword_name`` names the keyword names of a call. + +Django's ORM takes column aliases and lookups as keyword names, ``annotate(**{alias: +expr})``, and several advisories are SQL injections through a name the attacker chooses. +The engine decides the condition 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(a=...)`` or plain +keywords, contradict it and the call is ruled out, so a project passing only those stays +``not_affected``; keys the engine cannot establish, a parameter, a mapping built +elsewhere, ``dict(zip(...))``, leave it pending review as any unknown argument does. +""" + +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.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) + +# ``Item.objects.annotate`` derives its symbol from the project's model class. +ANNOTATE = "python.app.models.Item.objects.annotate" +TEXT = "the alias is a keyword name the attacker chooses" +CONDITION = Condition("keyword_name", TEXT) +ADVISORY = Advisory( + "CVE-2025-57833", + "django", + "<4.2.24 || >=5.0,<5.1.12 || >=5.2,<5.2.6", + "FilteredRelation alias through annotate()/alias() allows SQL injection", + Severity.HIGH, + entry_points=( + AdvisoryEntryPoint(SymbolId(ANNOTATE), "FilteredRelation alias through annotate()/alias()", (CONDITION,)), + ), + modules=("django",), +) + +LOCK = """version = 1 + +[[package]] +name = "app" +version = "0.1.0" +source = { virtual = "." } +dependencies = [{ name = "django" }] + +[[package]] +name = "django" +version = "5.2.5" +source = { registry = "https://pypi.org/simple" } +""" +MODELS = "from django.db import models\n\n\nclass Item(models.Model):\n pass\n" +VIEWS = "from django.db.models import F\nfrom django.http import HttpRequest\n\nfrom .models import Item\n\n\n" +ISSUE = VIEWS + ( + "def listing(request: HttpRequest):\n" + ' return Item.objects.annotate(**{request.GET["alias"]: F("id")}) # the alias is the attacker\'s\n' + "\n\n" + "def fixed(request: HttpRequest):\n" + ' return Item.objects.annotate(**{"total": F("id")}) # constant key\n' +) +FIXED = VIEWS + 'def fixed(request: HttpRequest):\n return Item.objects.annotate(**{"total": F("id")})\n' +RULED_OUT = f"app.views:8 {ANNOTATE}(keywords=total)" + + +def analyse(root: Path, views: str) -> ProjectAnalysis: + files = {"uv.lock": LOCK, "app/__init__.py": "", "app/models.py": MODELS, "app/views.py": views} + for relative, text in 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((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 calling(call: str, before: str = "") -> str: + return VIEWS + f"def listing(request: HttpRequest):\n{before} return Item.objects.annotate({call})\n" + + +# --------------------------------------------------------------------------- the reproduction + + +def test_keys_the_attacker_chooses_meet_the_condition_and_make_the_project_affected(tmp_path: Path) -> None: + analysis = analyse(tmp_path, ISSUE) + + (found,) = of_rule(analysis, "exploitable-vulnerability") + assert found.span.start_line == 8 + assert found.metadata["symbol"] == ANNOTATE + assert found.metadata["conditions_met"] == TEXT + assert "conditions_pending_review" not in found.metadata + assert statement(analysis, tmp_path)["status"] == "affected" + + +def test_constant_keys_rule_the_call_out_and_leave_the_project_not_affected(tmp_path: Path) -> None: + analysis = analyse(tmp_path, FIXED) + + 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}.") + + +@pytest.mark.parametrize( + "views", + [ + VIEWS + "def listing(request: HttpRequest, **extra):\n return Item.objects.annotate(**extra)\n", + VIEWS + ( + "def options():\n return {}\n\n\n" + "def listing(request: HttpRequest):\n opts = options()\n return Item.objects.annotate(**opts)\n" + ), + ], + ids=["parameter", "built-elsewhere"], +) +def test_a_mapping_whose_keys_are_not_established_leaves_the_condition_pending(tmp_path: Path, views: str) -> None: + analysis = analyse(tmp_path, views) + + (reached,) = of_rule(analysis, "reachable-vulnerability") + assert reached.metadata["symbol"] == ANNOTATE + assert reached.metadata["conditions_pending_review"] == TEXT + assert "conditions_met" not in reached.metadata + assert of_rule(analysis, "exploitable-vulnerability") == [] + assert statement(analysis, tmp_path)["status"] == "affected" + + +# --------------------------------------------------------------------------- constant keys + + +@pytest.mark.parametrize( + "call", + [ + '**{"total": F(request.GET["x"])}', + 'total=F("id")', + 'total=F(request.GET["x"])', + ], + ids=["literal-with-tainted-value", "keyword", "keyword-with-tainted-value"], +) +def test_constant_keys_are_ruled_out_whatever_the_values_carry(tmp_path: Path, call: str) -> None: + analysis = analyse(tmp_path, calling(call)) + + assert of_rule(analysis, "reachable-vulnerability") == [] + assert of_rule(analysis, "exploitable-vulnerability") == [] + (declared,) = of_rule(analysis, "vulnerable-dependency") + assert declared.metadata["ruled_out"] == RULED_OUT + assert statement(analysis, tmp_path)["status"] == "not_affected" + + +# --------------------------------------------------------------------------- key taint + + +ALIAS = ' alias = request.GET["alias"]\n' + + +@pytest.mark.parametrize( + "call, before", + [ + ('**{alias: F("id")}', ALIAS), + ('**dict([(alias, F("id"))])', ALIAS), + ('**{k: F("id") for k in request.GET}', ""), + ("**m", ALIAS + ' m = {}\n m[alias] = F("id")\n'), + ("**{**m}", ALIAS + ' m = {alias: F("id")}\n'), + ("**dict(m)", ALIAS + ' m = {alias: F("id")}\n'), + ('**dict(a=F("id"), **m)', ALIAS + ' m = {alias: F("id")}\n'), + ], + ids=["literal", "pairs", "comprehension", "subscript-assignment", "unpacked", "dict-copy", "dict-with-keywords"], +) +def test_every_construction_carrying_attacker_keys_meets_the_condition(tmp_path: Path, call: str, before: str) -> None: + analysis = analyse(tmp_path, calling(call, before)) + + (found,) = of_rule(analysis, "exploitable-vulnerability") + assert found.metadata["symbol"] == ANNOTATE + assert found.metadata["conditions_met"] == TEXT + assert "conditions_pending_review" not in found.metadata + # A call the taint engine finds exploitable is not one the call site rules out. + (declared,) = of_rule(analysis, "vulnerable-dependency") + assert "ruled_out" not in declared.metadata + assert statement(analysis, tmp_path)["status"] == "affected" + + +def test_keys_zipped_from_input_are_pending_not_met(tmp_path: Path) -> None: + analysis = analyse(tmp_path, calling('**dict(zip(request.GET, [F("id")]))')) + + (found,) = of_rule(analysis, "exploitable-vulnerability") + assert found.metadata["conditions_pending_review"] == TEXT + assert "conditions_met" not in found.metadata + assert statement(analysis, tmp_path)["status"] == "affected" + + +# --------------------------------------------------------------------------- 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": "keyword_name", "text": TEXT}] + + +@pytest.mark.parametrize( + "extra", + [{"argument": "alias"}, {"position": 0}, {"values": ["total"]}, {"default": True}], + ids=["argument", "position", "values", "default"], +) +def test_a_keyword_name_condition_carrying_more_than_its_text_is_rejected(tmp_path: Path, extra: dict[str, Any]) -> None: + document = json.loads(dump_advisories((ADVISORY,))) + document["advisories"][0]["entry_points"][0]["conditions"] = [{"kind": "keyword_name", "text": TEXT, **extra}] + path = tmp_path / "advisories.json" + path.write_text(json.dumps(document), encoding="utf-8") + + with pytest.raises(AdvisoryFileError): + load_advisories(path)