From 577c2365cfcd33d0b2f70d5af8656a7cb9d914ae Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 13 Oct 2026 06:11:02 +0200 Subject: [PATCH 1/2] fix(refutation): prove only the allowlists that are constant and stable Closes #239 --- docs/usage.md | 23 +- .../abstract/module_constants.py | 121 ++++++ src/coretrace_python/abstract/strings.py | 54 +-- src/coretrace_python/engine.py | 2 + src/coretrace_python/findings/refutation.py | 111 +++++- src/coretrace_python/taint/engine.py | 33 +- tests/test_allowlist_shapes.py | 358 ++++++++++++++++++ 7 files changed, 651 insertions(+), 51 deletions(-) create mode 100644 src/coretrace_python/abstract/module_constants.py create mode 100644 tests/test_allowlist_shapes.py diff --git a/docs/usage.md b/docs/usage.md index e6cb6cb..c2a2af5 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -133,7 +133,7 @@ variable is not treated that way yet. | `exploitable-vulnerability` | An API affected by an advisory of a vulnerable requirement, see Dependencies. | Each flow is judged before it is reported. A dominating guard that proves the value -safe (`isdigit()`, membership in a constant collection, equality with a constant, a +safe (`isdigit()`, membership in a constant allowlist, equality with a constant, a numeric value proven by `int()`, `len()` or bounded arithmetic, a validator such as `re.fullmatch`) refutes it and nothing is reported. A hotspot is reported at medium confidence. Two things make a flow a hotspot: @@ -149,6 +149,27 @@ a vulnerability. When the flow passes the object whole, to the sink or to a func a guard on any of its attributes still counts. The verdict and its evidence are in the finding's metadata. +A membership test (`in`, `not in`) is a constant allowlist when every member is a constant +that cannot change before the test: + +- a tuple, list or set display written in the function, or a dict display by its keys, + tested directly or through `.keys()`; +- `frozenset`, `set`, `tuple` or `list` of one of these; +- a module-level tuple, or `frozenset` of a display, of constants, bound once in its + module and never rebound there, by another assignment, an augmented assignment or a + `global` statement. + +A list, set or dict display of the function counts only while the function only reads it: +changed by a method, stored, or passed to a call before the test, it may hold anything. A +module-level list, set or dict can be changed from any module, so it never counts, and +the guard leaves a hotspot. Rebinding a module-level name from another module +(`settings.ALLOWED = ...`) is not seen, as for the module-level strings that fix a URL. + +A lookup in a dict display, `{"list": "ls"}[key]` or `.get(key, default)`, gives one of +its values or the default, never the key: input that only selects the entry is not +passed on. The same lookup in a module-level dict passes the key on, as any other +subscript does. + Django's `render_to_string(name, context)` returns HTML in which autoescaping escaped every variable, so data reaching a response through it is no `xss`, if the template really escapes. On a directory check, the analyzer reads the project's templates to decide it. diff --git a/src/coretrace_python/abstract/module_constants.py b/src/coretrace_python/abstract/module_constants.py new file mode 100644 index 0000000..f2cae96 --- /dev/null +++ b/src/coretrace_python/abstract/module_constants.py @@ -0,0 +1,121 @@ +"""Module-level names whose value is known (architecture ยง18): the names bound once and +never rebound, and among them the ones bound to an immutable collection of constants, +which an allowlist reads.""" + +from __future__ import annotations + +from collections.abc import Mapping +from types import MappingProxyType +from typing import ClassVar + +from coretrace_python.analysis import Analysis, AnalysisContext, AnyAnalysis +from coretrace_python.cfg import CFGError +from coretrace_python.hir import nodes +from coretrace_python.ir.lowering import ( + MODULE_BODY, + LoweringError, + PyIRAnalysis, + analyzable_functions, +) +from coretrace_python.ir.model import ( + BuildDict, + BuildList, + BuildSet, + BuildTuple, + Call, + Constant, + ForNext, + Instruction, + SetGlobal, + StoreLocal, + Symbol, + Value, +) +from coretrace_python.semantic.symbols import SymbolId + +FROZENSET = SymbolId("python.builtins.frozenset") + + +def module_bindings(ctx: AnalysisContext) -> tuple[Mapping[str, Instruction], Mapping[Value, Instruction]]: + """The module-level names bound once and never rebound, by no other statement of the + module body, no function assigning them as ``global``, no definition of that name, + each with the instruction that makes its value; and the definitions of the module + body's values. A module that does not lower entirely has none.""" + + try: + functions = [ctx.get(PyIRAnalysis, function) for function in analyzable_functions(ctx.module)] + except (LoweringError, CFGError): + return MappingProxyType({}), MappingProxyType({}) + body = next((function for function in functions if function.name == MODULE_BODY), None) + if body is None: + return MappingProxyType({}), MappingProxyType({}) + defs = {i.result: i for block in body.blocks for i in block.instructions if i.result is not None} + stored: dict[str, list[Value | None]] = {} + for block in body.blocks: + for instruction in block.instructions: + if isinstance(instruction, StoreLocal): + stored.setdefault(instruction.name, []).append(instruction.value) + if isinstance(block.terminator, ForNext): + # A ``for`` binds its target to each item. + stored.setdefault(block.terminator.target, []).append(None) + rebound = { + i.name for function in functions for block in function.blocks for i in block.instructions + if isinstance(i, SetGlobal) + } + rebound |= {s.name for s in ctx.module.body if isinstance(s, nodes.Function | nodes.Class)} + bound: dict[str, Instruction] = {} + for name, values in stored.items(): + value = values[0] if len(values) == 1 and name not in rebound else None + made = defs.get(value) if value is not None else None + if made is not None: + bound[name] = made + return MappingProxyType(bound), MappingProxyType(defs) + + +def constant_members(display: Instruction | None, defs: Mapping[Value, Instruction]) -> bool: + """Whether ``display`` is a tuple, list or set display whose items are all constants, + or a dict display whose keys are, with nothing unpacked into it: the members a + membership test reads.""" + + if isinstance(display, BuildList | BuildTuple | BuildSet): + return not display.unpacked and all(isinstance(defs.get(e), Constant) for e in display.elements) + if isinstance(display, BuildDict): + return not display.unpacked and all(isinstance(defs.get(key), Constant) for key, _ in display.items) + return False + + +def converts(call: Call, defs: Mapping[Value, Instruction], symbols: frozenset[SymbolId]) -> Value | None: + """The one positional argument of ``call`` when it calls one of ``symbols`` with + nothing else (``frozenset(x)``), otherwise ``None``.""" + + callee = defs.get(call.callee) + if not isinstance(callee, Symbol) or callee.symbol_id not in symbols: + return None + if len(call.arguments) != 1 or call.keywords or call.starred: + return None + return call.arguments[0] + + +class ModuleCollectionsAnalysis(Analysis[frozenset[str]]): + """The module-level names bound once and never rebound (``module_bindings``) to an + immutable collection of constants: a tuple display, or ``frozenset`` of a display + whose members are constants (``constant_members``). A list, a set or a dict can still + change after it is bound, so it is not one.""" + + name: ClassVar[str] = "abstract.module_collections" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({PyIRAnalysis}) + + @classmethod + def compute(cls, ctx: AnalysisContext) -> frozenset[str]: + bound, defs = module_bindings(ctx) + found: set[str] = set() + for name, made in bound.items(): + collection: Instruction | None = made + if isinstance(made, Call): + argument = converts(made, defs, frozenset({FROZENSET})) + collection = defs.get(argument) if argument is not None else None + elif not isinstance(made, BuildTuple): + continue + if constant_members(collection, defs): + found.add(name) + return frozenset(found) diff --git a/src/coretrace_python/abstract/strings.py b/src/coretrace_python/abstract/strings.py index ccf2714..6387381 100644 --- a/src/coretrace_python/abstract/strings.py +++ b/src/coretrace_python/abstract/strings.py @@ -9,24 +9,15 @@ from types import MappingProxyType from typing import ClassVar +from coretrace_python.abstract.module_constants import module_bindings from coretrace_python.analysis import Analysis, AnalysisContext, AnyAnalysis -from coretrace_python.cfg import CFGError -from coretrace_python.hir import nodes -from coretrace_python.ir.lowering import ( - MODULE_BODY, - LoweringError, - PyIRAnalysis, - analyzable_functions, -) +from coretrace_python.ir.lowering import PyIRAnalysis from coretrace_python.ir.model import ( BinaryOp, BuildString, Constant, - ForNext, Global, Instruction, - SetGlobal, - StoreLocal, Value, ) @@ -59,40 +50,19 @@ def leading_text(value: Value, defs: Mapping[Value, Instruction], strings: Mappi class ModuleStringsAnalysis(Analysis[Mapping[str, str]]): - """The module-level names bound once, to a string literal, and never rebound: by no - other statement of the module body, no function assigning them as ``global``, no - definition of that name. A module that does not lower entirely has none.""" + """The module-level names bound once and never rebound (``module_bindings``) to a + string literal, with that literal.""" name: ClassVar[str] = "abstract.module_strings" requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({PyIRAnalysis}) @classmethod def compute(cls, ctx: AnalysisContext) -> Mapping[str, str]: - try: - functions = [ctx.get(PyIRAnalysis, function) for function in analyzable_functions(ctx.module)] - except (LoweringError, CFGError): - return MappingProxyType({}) - body = next((function for function in functions if function.name == MODULE_BODY), None) - if body is None: - return MappingProxyType({}) - defs = {i.result: i for block in body.blocks for i in block.instructions if i.result is not None} - stored: dict[str, list[Value | None]] = {} - for block in body.blocks: - for instruction in block.instructions: - if isinstance(instruction, StoreLocal): - stored.setdefault(instruction.name, []).append(instruction.value) - if isinstance(block.terminator, ForNext): - # A ``for`` binds its target to each item. - stored.setdefault(block.terminator.target, []).append(None) - rebound = { - i.name for function in functions for block in function.blocks for i in block.instructions - if isinstance(i, SetGlobal) - } - rebound |= {s.name for s in ctx.module.body if isinstance(s, nodes.Function | nodes.Class)} - found: dict[str, str] = {} - for name, values in stored.items(): - value = values[0] if len(values) == 1 and name not in rebound else None - made = defs.get(value) if value is not None else None - if isinstance(made, Constant) and isinstance(made.value, str): - found[name] = made.value - return MappingProxyType(found) + bound, _ = module_bindings(ctx) + return MappingProxyType( + { + name: made.value + for name, made in bound.items() + if isinstance(made, Constant) and isinstance(made.value, str) + } + ) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 9a3a56c..d10e1ed 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -19,6 +19,7 @@ from coretrace_python import __version__ from coretrace_python.abstract import ConstantPropagation, HeapAnalysis, RangeAnalysis +from coretrace_python.abstract.module_constants import ModuleCollectionsAnalysis from coretrace_python.abstract.strings import ModuleStringsAnalysis from coretrace_python.analysis import ( AnalysisContext, @@ -158,6 +159,7 @@ ConstantPropagation, RangeAnalysis, ModuleStringsAnalysis, + ModuleCollectionsAnalysis, HeapAnalysis, CallGraphAnalysis, SummaryAnalysis, diff --git a/src/coretrace_python/findings/refutation.py b/src/coretrace_python/findings/refutation.py index d38d2e8..a9487d3 100644 --- a/src/coretrace_python/findings/refutation.py +++ b/src/coretrace_python/findings/refutation.py @@ -26,9 +26,15 @@ from typing import ClassVar from coretrace_python.abstract import ConstantPropagation, RangeAnalysis, RangeFacts +from coretrace_python.abstract.module_constants import ( + ModuleCollectionsAnalysis, + constant_members, + converts, +) from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis from coretrace_python.cfg import CFG, BlockId, CFGAnalysis, DominanceAnalysis, DominatorTree from coretrace_python.hir import nodes +from coretrace_python.hir.visitors import Node, children from coretrace_python.interprocedural import ( CallGraph, CallGraphAnalysis, @@ -36,9 +42,10 @@ project_symbol, ) from coretrace_python.ir.model import ( + BinaryOp, BoolOp, Branch, - BuildList, + BuildDict, BuildTuple, Call, Compare, @@ -46,13 +53,17 @@ FunctionIR, GetAttr, GetItem, + GetIter, + Global, Instruction, + Symbol, UnaryOp, Value, ) from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.semantic.scopes import ScopeAnalysis, ScopeTable from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId, SymbolTable +from coretrace_python.source import SourceSpan from coretrace_python.taint import ( AuthorizationGuard, ModelTable, @@ -66,6 +77,17 @@ VALIDATORS = frozenset( {"isdigit", "isnumeric", "isdecimal", "isalnum", "isalpha", "isidentifier", "isascii"} ) +# The builtins that make a collection of the members of the one collection they are given. +_CONVERSIONS = frozenset(SymbolId(f"python.builtins.{name}") for name in ("frozenset", "set", "tuple", "list")) +# The methods of a list, a set or a dict that read it without changing it. +_READING_METHODS = frozenset( + { + "copy", "count", "index", + "difference", "intersection", "isdisjoint", "issubset", "issuperset", + "symmetric_difference", "union", + "get", "items", "keys", "values", + } +) class Status(Enum): @@ -115,6 +137,7 @@ def __init__( models: ModelTable | None = None, symbols: Mapping[Value, SymbolId] | None = None, authorization: AuthorizationGuard | None = None, + collection_loads: frozenset[SourceSpan] = frozenset(), ) -> None: self.function = function self.cfg = cfg @@ -124,11 +147,13 @@ def __init__( self.models = models or ModelTable((), (), ()) self.symbols = symbols or {} self.authorization = authorization + self.collection_loads = collection_loads self.blocks = {block.id: block for block in function.blocks} self.defs: dict[Value, Instruction] = { i.result: i for block in function.blocks for i in block.instructions if i.result } self._closures: dict[Value, frozenset[Value]] = {} + self._uses: dict[Value, list[Instruction]] | None = None # ------------------------------------------------------------------ dependencies @@ -342,12 +367,55 @@ def access(self, value: Value) -> tuple[Value, str, object] | None: return None def is_constant_collection(self, value: Value) -> bool: + """Whether every member of ``value`` is a constant that cannot change before the + test: a string constant, a display whose members are constants + (``constant_members``), the keys of a dict display (``.keys()``), ``frozenset``, + ``set``, ``tuple`` or ``list`` of such a collection, or a module-level name bound + once to an immutable one, read where ``collection_loads`` says. A collection that + may change after it is built, such as a module-level list, set or dict, is not + (#239).""" + definition = self.defs.get(value) if isinstance(definition, Constant): return isinstance(definition.value, str | bytes) - if isinstance(definition, BuildList | BuildTuple): - return all(isinstance(self.defs.get(e), Constant) for e in definition.elements) - return False + if isinstance(definition, Global | Symbol): + return definition.location in self.collection_loads + if isinstance(definition, Call): + argument = converts(definition, self.defs, _CONVERSIONS) + if argument is not None: + return self.is_constant_collection(argument) + callee = self.defs.get(definition.callee) + if not isinstance(callee, GetAttr) or callee.attribute != "keys": + return False + plain = not (definition.arguments or definition.keywords or definition.starred) + return plain and isinstance(self.defs.get(callee.object), BuildDict) and self.is_constant_collection(callee.object) + if not constant_members(definition, self.defs): + return False + return isinstance(definition, BuildTuple) or self.only_read(value) + + def only_read(self, value: Value) -> bool: + """Whether every use of the list, set or dict ``value`` reads it: a comparison, an + item or an iteration, an operator, a method that does not change it, or a + conversion such as ``frozenset(value)``. Stored, passed on, merged or changed by a + method, it may hold anything by the test.""" + + if self._uses is None: + uses: dict[Value, list[Instruction]] = {} + for block in self.function.blocks: + for instruction in block.instructions: + for operand in instruction.operands(): + uses.setdefault(operand, []).append(instruction) + self._uses = uses + for use in self._uses.get(value, ()): + if isinstance(use, GetAttr): + if use.attribute not in _READING_METHODS: + return False + elif isinstance(use, Call): + if converts(use, self.defs, _CONVERSIONS) != value: + return False + elif not isinstance(use, Compare | GetItem | GetIter | BinaryOp | UnaryOp): + return False + return True # ------------------------------------------------------------------ authorization @@ -442,8 +510,9 @@ def judge_flows( models: ModelTable | None = None, symbols: Mapping[Value, SymbolId] | None = None, authorization: AuthorizationGuard | None = None, + collection_loads: frozenset[SourceSpan] = frozenset(), ) -> Verdicts: - judge = _Judge(function, cfg, tree, taint, ranges, models, symbols, authorization) + judge = _Judge(function, cfg, tree, taint, ranges, models, symbols, authorization, collection_loads) verdicts = [] for flow in taint.flows: sink = judge.sink_block(flow) @@ -451,6 +520,36 @@ def judge_flows( return Verdicts(tuple(verdicts)) +def module_collection_loads( + function: nodes.Function, collections: frozenset[str], scopes: ScopeTable +) -> frozenset[SourceSpan]: + """Where the body of ``function`` reads a module-level name of ``collections`` + (``ModuleCollectionsAnalysis``): the spans of those names. The IR loads such a name as + a ``Global``, or, when the semantic tables know it as an instance + (``ALLOWED = frozenset(...)``), as the ``Symbol`` of what built it, which no longer + names it. Nested definitions are functions of their own; decorators and defaults are + read by the enclosing one.""" + + if not collections: + return frozenset() + scope = scopes.scope_for(function).id + module = scopes.module_scope.id + found: set[SourceSpan] = set() + pending: list[Node] = list(function.body) + while pending: + node = pending.pop() + if isinstance(node, nodes.Function | nodes.Class | nodes.Lambda): + continue + if ( + isinstance(node, nodes.Name) + and node.identifier in collections + and scopes.resolve(scope, node.identifier).scope == module + ): + found.add(node.span) + pending.extend(children(node)) + return frozenset(found) + + def authorization_of( function: nodes.Function, models: ModelTable, scopes: ScopeTable, symbols: SymbolTable ) -> AuthorizationGuard | None: @@ -480,6 +579,7 @@ class RefutationAnalysis(FunctionAnalysis[Verdicts]): CallGraphAnalysis, ScopeAnalysis, SymbolAnalysis, + ModuleCollectionsAnalysis, } ) @@ -500,6 +600,7 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> Verdicts: models, _with_project_callees(ssa, graph, name, ctx.module.name), authorization_of(function, models, ctx.get(ScopeAnalysis), ctx.get(SymbolAnalysis)), + module_collection_loads(function, ctx.get(ModuleCollectionsAnalysis), ctx.get(ScopeAnalysis)), ) diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index f38c672..051f92a 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -224,6 +224,26 @@ def loaded(self, instruction: GetAttr | GetItem | GetIter, state: Mapping[Key, T taint = taint.join(state.get(location, Taint.none())) return taint + def looked_up(self, instruction: Call | GetItem, state: Mapping[Key, Taint]) -> Taint | None: + """What a lookup in a dict display gives, ``d[key]``, ``d.get(key)`` or + ``d.get(key, default)``: one of its values, whatever was stored in it since, or the + default, never the key, which only selects (#239). ``None`` for anything else.""" + + if isinstance(instruction, GetItem): + mapping, default = instruction.object, None + else: + method = self.defs.get(instruction.callee) + if not isinstance(method, GetAttr) or method.attribute != "get": + return None + if instruction.keywords or instruction.starred or not 1 <= len(instruction.arguments) <= 2: + return None + mapping = method.object + default = instruction.arguments[1] if len(instruction.arguments) == 2 else None + if not isinstance(self.defs.get(mapping), BuildDict): + return None + taint = self.deep(mapping, state) + return taint if default is None else taint.join(self.deep(default, state)) + def keyed(self, value: Value, state: Mapping[Key, Taint]) -> Taint: """The taint of the keys of the mappings ``value`` points to.""" @@ -281,12 +301,19 @@ def evaluate( taint = taint.join(incoming[predecessor].get(value, Taint.none())) elif isinstance(instruction, Call): taint = self.call(instruction, state, flows) + looked_up = self.looked_up(instruction, state) + if looked_up is not None: + taint = looked_up elif isinstance(instruction, NonlocalResult): taint = self.nonlocal_result(instruction, state) else: - taint = self.instruction(instruction, state) - if isinstance(instruction, GetAttr | GetItem | GetIter): - taint = taint.join(self.loaded(instruction, state)) + looked_up = self.looked_up(instruction, state) if isinstance(instruction, GetItem) else None + if looked_up is not None: + taint = looked_up + else: + taint = self.instruction(instruction, state) + if isinstance(instruction, GetAttr | GetItem | GetIter): + taint = taint.join(self.loaded(instruction, state)) state[instruction.result] = taint terminator = block.terminator if isinstance(terminator, ForNext) and terminator.result is not None: diff --git a/tests/test_allowlist_shapes.py b/tests/test_allowlist_shapes.py new file mode 100644 index 0000000..303a08b --- /dev/null +++ b/tests/test_allowlist_shapes.py @@ -0,0 +1,358 @@ +"""Acceptance tests for issue #239: which allowlists prove a value, through the bundled +command-injection rule. + +A membership test proves the value when the collection is constant and stable: a tuple, +list, set or dict literal at the point of use, ``frozenset``/``set``/``tuple``/``list`` of +such a literal, the keys of a dict literal, or a module-level tuple or ``frozenset`` of +constants bound once and never rebound. A lookup with an input key in a dict literal of +constants gives one of its constants. A module-level list, set or dict is mutable: its +stability is not shown, so it keeps today's verdict (a hotspot for a membership test, a +vulnerability for a lookup), as does any collection with an element that is not a +constant. +""" + +from __future__ import annotations + +import contextlib +import io +import json + +import pytest + +from coretrace_python.cli import main + +APP = """\ +import os + +from flask import Flask, request + +app = Flask(__name__) + +ALLOWED_TUPLE = ("ls", "date") +ALLOWED_FROZENSET = frozenset({"ls", "date"}) +ALLOWED_SET = {"ls", "date"} +ALLOWED_LIST = ["ls", "date"] +COMMANDS = {"list": "ls", "where": "date"} + +MUTATED = ["ls", "date"] +REBOUND = ("ls", "date") +MIXED = ("ls", os.environ.get("EXTRA_COMMAND", "date")) +FROZEN_MIXED = frozenset({"ls", os.environ.get("EXTRA_COMMAND", "date")}) +EXTENDED = ("ls",) +EXTENDED += ("date",) + + +def extend(extra): + MUTATED.append(extra) + + +def rebind(extra): + global REBOUND + REBOUND = (extra,) + + +@app.route("/literal-tuple") +def literal_tuple(): + cmd = request.args["c"] + if cmd in ("ls", "date"): + os.system(cmd) + + +@app.route("/literal-list") +def literal_list(): + cmd = request.args["c"] + if cmd in ["ls", "date"]: + os.system(cmd) + + +@app.route("/literal-set") +def literal_set(): + cmd = request.args["c"] + if cmd in {"ls", "date"}: + os.system(cmd) + + +@app.route("/literal-frozenset") +def literal_frozenset(): + cmd = request.args["c"] + if cmd in frozenset(("ls", "date")): + os.system(cmd) + + +@app.route("/literal-frozenset-of-set") +def literal_frozenset_of_set(): + cmd = request.args["c"] + if cmd in frozenset({"ls", "date"}): + os.system(cmd) + + +@app.route("/literal-dict-keys") +def literal_dict_keys(): + cmd = request.args["c"] + if cmd in {"ls": "list", "date": "where"}: + os.system(cmd) + + +@app.route("/literal-dict-keys-method") +def literal_dict_keys_method(): + cmd = request.args["c"] + if cmd in {"ls": "list", "date": "where"}.keys(): + os.system(cmd) + + +@app.route("/literal-dict-lookup") +def literal_dict_lookup(): + os.system({"list": "ls", "where": "date"}[request.args["c"]]) + + +@app.route("/literal-dict-get-default") +def literal_dict_get_default(): + os.system({"list": "ls", "where": "date"}.get(request.args["c"], "true")) + + +@app.route("/module-tuple") +def module_tuple(): + cmd = request.args["c"] + if cmd in ALLOWED_TUPLE: + os.system(cmd) + + +@app.route("/module-frozenset") +def module_frozenset(): + cmd = request.args["c"] + if cmd in ALLOWED_FROZENSET: + os.system(cmd) + + +@app.route("/unpacked-element") +def unpacked_element(): + cmd = request.args["c"] + extra = request.args["e"].split(",") + if cmd in ("ls", *extra): + os.system(cmd) + + +@app.route("/non-constant-element") +def non_constant_element(): + cmd = request.args["c"] + if cmd in ("ls", request.args["extra"]): + os.system(cmd) + + +@app.route("/module-tuple-mixed") +def module_tuple_mixed(): + cmd = request.args["c"] + if cmd in MIXED: + os.system(cmd) + + +@app.route("/module-frozenset-mixed") +def module_frozenset_mixed(): + cmd = request.args["c"] + if cmd in FROZEN_MIXED: + os.system(cmd) + + +@app.route("/module-tuple-extended") +def module_tuple_extended(): + cmd = request.args["c"] + if cmd in EXTENDED: + os.system(cmd) + + +@app.route("/rebound-tuple") +def rebound_tuple(): + cmd = request.args["c"] + if cmd in REBOUND: + os.system(cmd) + + +@app.route("/module-set") +def module_set(): + cmd = request.args["c"] + if cmd in ALLOWED_SET: + os.system(cmd) + + +@app.route("/module-list") +def module_list(): + cmd = request.args["c"] + if cmd in ALLOWED_LIST: + os.system(cmd) + + +@app.route("/mutated-list") +def mutated_list(): + cmd = request.args["c"] + if cmd in MUTATED: + os.system(cmd) + + +@app.route("/dict-keys") +def dict_keys(): + cmd = request.args["c"] + if cmd in COMMANDS: + os.system(cmd) + + +@app.route("/dict-keys-method") +def dict_keys_method(): + cmd = request.args["c"] + if cmd in COMMANDS.keys(): + os.system(cmd) + + +@app.route("/dict-lookup") +def dict_lookup(): + os.system(COMMANDS[request.args["c"]]) + + +@app.route("/dict-get-default") +def dict_get_default(): + os.system(COMMANDS.get(request.args["c"], "true")) + + +@app.route("/literal-dict-non-constant-value") +def literal_dict_non_constant_value(): + os.system({"list": "ls", "custom": request.args["x"]}[request.args["c"]]) + + +@app.route("/literal-dict-get-tainted-default") +def literal_dict_get_tainted_default(): + os.system({"list": "ls"}.get(request.args["c"], request.args["fallback"])) + +def register(values): + values.append(os.environ.get("EXTRA_COMMAND", "date")) + + +@app.route("/local-list") +def local_list(): + cmd = request.args["c"] + allowed = ["ls", "date"] + if cmd in allowed: + os.system(cmd) + + +@app.route("/local-dict-keys-read") +def local_dict_keys_read(): + cmd = request.args["c"] + commands = {"ls": "list", "date": "when"} + print(commands.get("ls")) + if cmd in commands.keys(): + os.system(cmd) + + +@app.route("/local-list-appended") +def local_list_appended(): + cmd = request.args["c"] + allowed = ["ls"] + allowed.append(request.args["extra"]) + if cmd in allowed: + os.system(cmd) + + +@app.route("/local-set-added") +def local_set_added(): + cmd = request.args["c"] + allowed = {"ls"} + allowed.add(request.args["extra"]) + if cmd in allowed: + os.system(cmd) + + +@app.route("/local-list-passed") +def local_list_passed(): + cmd = request.args["c"] + allowed = ["ls"] + register(allowed) + if cmd in allowed: + os.system(cmd) + + +@app.route("/local-dict-key-written") +def local_dict_key_written(): + cmd = request.args["c"] + commands = {"ls": "list"} + commands[request.args["k"]] = "custom" + if cmd in commands: + os.system(cmd) + + +@app.route("/local-dict-value-written") +def local_dict_value_written(): + commands = {"list": "ls"} + commands["custom"] = request.args["x"] + os.system(commands[request.args["c"]]) +""" + +NOT_REPORTED = None +HOTSPOT = ("hotspot", "medium") +VULNERABILITY = ("vulnerability", "high") + +EXPECTED = { + # Constant and stable: refuted, or no taint left. + "literal_tuple": NOT_REPORTED, + "literal_list": NOT_REPORTED, + "literal_set": NOT_REPORTED, + "literal_frozenset": NOT_REPORTED, + "literal_frozenset_of_set": NOT_REPORTED, + "literal_dict_keys": NOT_REPORTED, + "literal_dict_keys_method": NOT_REPORTED, + "literal_dict_lookup": NOT_REPORTED, + "literal_dict_get_default": NOT_REPORTED, + "module_tuple": NOT_REPORTED, + "module_frozenset": NOT_REPORTED, + # An element that is not a constant. + "unpacked_element": HOTSPOT, + "non_constant_element": HOTSPOT, + "module_tuple_mixed": HOTSPOT, + "module_frozenset_mixed": HOTSPOT, + # Rebound in the module body or by a ``global`` assignment. + "module_tuple_extended": HOTSPOT, + "rebound_tuple": HOTSPOT, + # Mutable module-level collections: stability not shown, today's verdict kept. + "module_set": HOTSPOT, + "module_list": HOTSPOT, + "mutated_list": HOTSPOT, + "dict_keys": HOTSPOT, + "dict_keys_method": HOTSPOT, + "dict_lookup": VULNERABILITY, + "dict_get_default": VULNERABILITY, + # A display of the function that is only read. + "local_list": NOT_REPORTED, + "local_dict_keys_read": NOT_REPORTED, + # A display mutated, or passed on, before the test. + "local_list_appended": HOTSPOT, + "local_set_added": HOTSPOT, + "local_list_passed": HOTSPOT, + "local_dict_key_written": HOTSPOT, + # A lookup whose result may be input. + "local_dict_value_written": VULNERABILITY, + "literal_dict_non_constant_value": VULNERABILITY, + "literal_dict_get_tainted_default": VULNERABILITY, +} + + +@pytest.fixture(scope="module") +def verdicts(tmp_path_factory: pytest.TempPathFactory) -> dict[str, tuple[str, str]]: + root = tmp_path_factory.mktemp("allowlists") + (root / "app.py").write_text(APP, encoding="utf-8") + out = io.StringIO() + with contextlib.redirect_stdout(out): + main(["--check", str(root), "--format", "json"]) + found: dict[str, tuple[str, str]] = {} + for finding in json.loads(out.getvalue())["findings"]: + if finding["rule_id"] != "command-injection": + continue + assert finding["function"] not in found, finding + found[finding["function"]] = (finding["metadata"]["verdict"], finding["confidence"]) + return found + + +@pytest.mark.parametrize("route", sorted(EXPECTED)) +def test_allowlist_shape_verdict(verdicts: dict[str, tuple[str, str]], route: str) -> None: + assert verdicts.get(route) == EXPECTED[route] + + +def test_every_reported_route_is_expected(verdicts: dict[str, tuple[str, str]]) -> None: + assert set(verdicts) <= set(EXPECTED) From 0bf94f8dc148bb5e2bddb238590530f9c6e0660a Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 13 Oct 2026 07:08:51 +0200 Subject: [PATCH 2/2] fix(refutation): check converted collections for changes and keep dict keys out of lookups A list or set built by list() or set() proves a membership test only while the function only reads it; a lookup in a dict display carries the taint of its values, unpacked mappings and later stores, never of its keys. --- docs/usage.md | 9 +++-- src/coretrace_python/findings/refutation.py | 18 ++++++--- src/coretrace_python/taint/engine.py | 16 ++++++-- tests/test_allowlist_shapes.py | 44 +++++++++++++++++++++ 4 files changed, 73 insertions(+), 14 deletions(-) diff --git a/docs/usage.md b/docs/usage.md index c2a2af5..20422fb 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -159,15 +159,16 @@ that cannot change before the test: module and never rebound there, by another assignment, an augmented assignment or a `global` statement. -A list, set or dict display of the function counts only while the function only reads it: -changed by a method, stored, or passed to a call before the test, it may hold anything. A +A list, set or dict of the function, a display or the `set` or `list` of one, counts only +while the function only reads it: changed by a method, stored, or passed to a call before +the test, it may hold anything. A module-level list, set or dict can be changed from any module, so it never counts, and the guard leaves a hotspot. Rebinding a module-level name from another module (`settings.ALLOWED = ...`) is not seen, as for the module-level strings that fix a URL. A lookup in a dict display, `{"list": "ls"}[key]` or `.get(key, default)`, gives one of -its values or the default, never the key: input that only selects the entry is not -passed on. The same lookup in a module-level dict passes the key on, as any other +its values, of what was stored in it since, or the default, never a key: input that only +selects the entry, or that is itself a key of the display, is not passed on. The same lookup in a module-level dict passes the key on, as any other subscript does. Django's `render_to_string(name, context)` returns HTML in which autoescaping escaped diff --git a/src/coretrace_python/findings/refutation.py b/src/coretrace_python/findings/refutation.py index a9487d3..945af8c 100644 --- a/src/coretrace_python/findings/refutation.py +++ b/src/coretrace_python/findings/refutation.py @@ -77,8 +77,10 @@ VALIDATORS = frozenset( {"isdigit", "isnumeric", "isdecimal", "isalnum", "isalpha", "isidentifier", "isascii"} ) -# The builtins that make a collection of the members of the one collection they are given. -_CONVERSIONS = frozenset(SymbolId(f"python.builtins.{name}") for name in ("frozenset", "set", "tuple", "list")) +# The builtins that make a collection of the members of the one collection they are given; +# a ``set`` or a ``list`` they make can still change afterwards. +_IMMUTABLE_CONVERSIONS = frozenset(SymbolId(f"python.builtins.{name}") for name in ("frozenset", "tuple")) +_CONVERSIONS = _IMMUTABLE_CONVERSIONS | frozenset(SymbolId(f"python.builtins.{name}") for name in ("set", "list")) # The methods of a list, a set or a dict that read it without changing it. _READING_METHODS = frozenset( { @@ -371,9 +373,9 @@ def is_constant_collection(self, value: Value) -> bool: test: a string constant, a display whose members are constants (``constant_members``), the keys of a dict display (``.keys()``), ``frozenset``, ``set``, ``tuple`` or ``list`` of such a collection, or a module-level name bound - once to an immutable one, read where ``collection_loads`` says. A collection that - may change after it is built, such as a module-level list, set or dict, is not - (#239).""" + once to an immutable one, read where ``collection_loads`` says. A list, set or + dict counts only while the function only reads it (``only_read``); a module-level + one may change anywhere, so it never counts (#239).""" definition = self.defs.get(value) if isinstance(definition, Constant): @@ -383,7 +385,11 @@ def is_constant_collection(self, value: Value) -> bool: if isinstance(definition, Call): argument = converts(definition, self.defs, _CONVERSIONS) if argument is not None: - return self.is_constant_collection(argument) + if not self.is_constant_collection(argument): + return False + callee = self.defs.get(definition.callee) + immutable = isinstance(callee, Symbol) and callee.symbol_id in _IMMUTABLE_CONVERSIONS + return immutable or self.only_read(value) callee = self.defs.get(definition.callee) if not isinstance(callee, GetAttr) or callee.attribute != "keys": return False diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 051f92a..ac03a94 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -226,8 +226,9 @@ def loaded(self, instruction: GetAttr | GetItem | GetIter, state: Mapping[Key, T def looked_up(self, instruction: Call | GetItem, state: Mapping[Key, Taint]) -> Taint | None: """What a lookup in a dict display gives, ``d[key]``, ``d.get(key)`` or - ``d.get(key, default)``: one of its values, whatever was stored in it since, or the - default, never the key, which only selects (#239). ``None`` for anything else.""" + ``d.get(key, default)``: one of its values, of the mappings it unpacks, of whatever + was stored in it since, or the default; never a key, its own or the one looked up, + which only select (#239). ``None`` for anything else.""" if isinstance(instruction, GetItem): mapping, default = instruction.object, None @@ -239,9 +240,16 @@ def looked_up(self, instruction: Call | GetItem, state: Mapping[Key, Taint]) -> return None mapping = method.object default = instruction.arguments[1] if len(instruction.arguments) == 2 else None - if not isinstance(self.defs.get(mapping), BuildDict): + display = self.defs.get(mapping) + if not isinstance(display, BuildDict): return None - taint = self.deep(mapping, state) + taint = Taint.none() + for _, value in display.items: + taint = taint.join(state.get(value, Taint.none())) + for unpacked in display.unpacked: + taint = taint.join(self.deep(unpacked, state)) + for location in self.heap.locations(mapping, ELEMENTS): + taint = taint.join(state.get(location, Taint.none())) return taint if default is None else taint.join(self.deep(default, state)) def keyed(self, value: Value, state: Mapping[Key, Taint]) -> Taint: diff --git a/tests/test_allowlist_shapes.py b/tests/test_allowlist_shapes.py index 303a08b..50ae68d 100644 --- a/tests/test_allowlist_shapes.py +++ b/tests/test_allowlist_shapes.py @@ -283,6 +283,43 @@ def local_dict_value_written(): commands = {"list": "ls"} commands["custom"] = request.args["x"] os.system(commands[request.args["c"]]) + +@app.route("/converted-list") +def converted_list(): + cmd = request.args["c"] + allowed = list(("ls", "date")) + if cmd in allowed: + os.system(cmd) + + +@app.route("/converted-list-appended") +def converted_list_appended(): + allowed = list(("ls",)) + allowed.append(request.args["extra"]) + cmd = request.args["c"] + if cmd in allowed: + os.system(cmd) + + +@app.route("/converted-set-added") +def converted_set_added(): + allowed = set(("ls",)) + allowed.add(request.args["extra"]) + cmd = request.args["c"] + if cmd in allowed: + os.system(cmd) + + +@app.route("/literal-dict-input-key") +def literal_dict_input_key(): + os.system({request.args["k"]: "ls"}[request.args["c"]]) + + +@app.route("/local-dict-updated") +def local_dict_updated(): + commands = {"list": "ls"} + commands.update(request.args) + os.system(commands[request.args["c"]]) """ NOT_REPORTED = None @@ -321,13 +358,20 @@ def local_dict_value_written(): # A display of the function that is only read. "local_list": NOT_REPORTED, "local_dict_keys_read": NOT_REPORTED, + # A ``list`` or ``set`` built from constants and only read. + "converted_list": NOT_REPORTED, + # A key that is input only selects the value of a dict display. + "literal_dict_input_key": NOT_REPORTED, # A display mutated, or passed on, before the test. + "converted_list_appended": HOTSPOT, + "converted_set_added": HOTSPOT, "local_list_appended": HOTSPOT, "local_set_added": HOTSPOT, "local_list_passed": HOTSPOT, "local_dict_key_written": HOTSPOT, # A lookup whose result may be input. "local_dict_value_written": VULNERABILITY, + "local_dict_updated": VULNERABILITY, "literal_dict_non_constant_value": VULNERABILITY, "literal_dict_get_tainted_default": VULNERABILITY, }