diff --git a/README.md b/README.md index 85e1cc7..51e503c 100644 --- a/README.md +++ b/README.md @@ -106,6 +106,12 @@ detectors consume the shared taint result, so adding a framework model makes eve detector aware of it. Taint follows calls between functions of the same module through function summaries: a tainted argument passed to a helper that reaches a sink is reported at the call site, and a helper returning attacker-controlled data taints its result. +Taint also follows objects, not only values: a coarse heap abstraction gives every +allocation site one abstract object with an `elements` and an `attributes` location, so +`b = a; b.append(user_input); sink(a[0])`, dictionary and attribute stores, `extend`, +`insert`, `add`, `update` and iteration all carry taint through the container. Function +summaries record the parameters a function mutates and the module globals it touches, so +a helper that fills a list taints the caller's list, across files too. Each flow is then judged: a dominating guard that proves the value safe (`isdigit()`, membership in a constant allowlist, equality with a constant) refutes it and it is not reported; a guard that only mentions the value makes it a hotspot with medium confidence; diff --git a/src/coretrace_python/abstract/__init__.py b/src/coretrace_python/abstract/__init__.py index 5e2233f..7e156e8 100644 --- a/src/coretrace_python/abstract/__init__.py +++ b/src/coretrace_python/abstract/__init__.py @@ -5,17 +5,41 @@ ConstantPropagation, propagate_constants, ) +from coretrace_python.abstract.heap import ( + ATTRIBUTES, + ELEMENTS, + MUTATORS, + AbstractObject, + AliasSet, + AllocationSite, + HeapAnalysis, + HeapFacts, + HeapLocation, + analyze_heap, + mutated_by, +) from coretrace_python.abstract.ranges import Interval, RangeAnalysis, RangeFacts, analyze_ranges from coretrace_python.abstract.values import AbstractValue, Truth __all__ = [ + "ATTRIBUTES", + "ELEMENTS", + "MUTATORS", + "AbstractObject", "AbstractValue", + "AliasSet", + "AllocationSite", "ConstantFacts", "ConstantPropagation", + "HeapAnalysis", + "HeapFacts", + "HeapLocation", "Interval", "RangeAnalysis", "RangeFacts", "Truth", + "analyze_heap", "analyze_ranges", + "mutated_by", "propagate_constants", ] diff --git a/src/coretrace_python/abstract/heap.py b/src/coretrace_python/abstract/heap.py new file mode 100644 index 0000000..dc56238 --- /dev/null +++ b/src/coretrace_python/abstract/heap.py @@ -0,0 +1,242 @@ +"""Coarse heap and aliasing abstraction (architecture §22). + +SSA names values, not objects. This domain gives every allocation site one +``AbstractObject``: containers and calls at their instruction, parameters, module +globals and imported symbols by name, and the fields loaded from an object by field. +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 +store, a mutating method call or a load on any alias reads and writes the same place. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from types import MappingProxyType +from typing import ClassVar + +from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis +from coretrace_python.hir import nodes +from coretrace_python.ir.model import ( + Await, + BuildDict, + BuildList, + BuildTuple, + Call, + Catch, + ForNext, + FunctionIR, + GetAttr, + GetItem, + GetIter, + Global, + Instruction, + Phi, + SetAttr, + SetItem, + Symbol, + Value, + WithEnter, + Yield, +) +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.source import SourceSpan + +ELEMENTS = "elements" +ATTRIBUTES = "attributes" + +# Method names that store their arguments into the receiver's elements. +MUTATORS = frozenset( + {"append", "appendleft", "extend", "extendleft", "insert", "add", "update", "setdefault", "put"} +) + + +@dataclass(frozen=True) +class AllocationSite: + kind: str + location: SourceSpan + name: str = "" + ordinal: int = 0 + + def __str__(self) -> str: + text = f"{self.kind}@{self.location.source_id}:{self.location.start_line}" + if self.name: + text += f":{self.name}" + if self.ordinal: + text += f"#{self.ordinal}" + return text + + +@dataclass(frozen=True) +class AbstractObject: + site: AllocationSite + + def __str__(self) -> str: + return str(self.site) + + +@dataclass(frozen=True) +class HeapLocation: + object: AbstractObject + field: str + + def __str__(self) -> str: + return f"{self.object}.{self.field}" + + +AliasSet = frozenset[AbstractObject] +NOTHING: AliasSet = frozenset() + + +class HeapFacts: + def __init__(self, objects: Mapping[Value, AliasSet]) -> None: + self._objects: Mapping[Value, AliasSet] = MappingProxyType(dict(objects)) + self.values = tuple(objects) + + def objects(self, value: Value) -> AliasSet: + return self._objects.get(value, NOTHING) + + def locations(self, value: Value, field: str) -> tuple[HeapLocation, ...]: + return tuple(HeapLocation(o, field) for o in sorted(self.objects(value), key=str)) + + +def mutated_by(call: Call, defs: Mapping[Value, Instruction]) -> Value | None: + """The receiver of a call to a mutating method (``xs.append(v)``), if any.""" + + callee = defs.get(call.callee) + if isinstance(callee, GetAttr) and callee.attribute in MUTATORS: + return callee.object + return None + + +class _PointsTo: + def __init__(self, function: FunctionIR) -> None: + self.function = function + self.defs: dict[Value, Instruction] = { + i.result: i for block in function.blocks for i in block.instructions if i.result + } + self.points: dict[Value, set[AbstractObject]] = {} + self.heap: dict[HeapLocation, set[AbstractObject]] = {} + self.named: dict[tuple[str, str], AbstractObject] = {} + for index, parameter in enumerate(function.parameters): + self.points[parameter] = { + AbstractObject(AllocationSite("parameter", function.location, ordinal=index)) + } + + def named_object(self, kind: str, name: str) -> AbstractObject: + key = (kind, name) + if key not in self.named: + self.named[key] = AbstractObject(AllocationSite(kind, self.function.location, name)) + return self.named[key] + + @staticmethod + def field_object(owner: AbstractObject, field: str) -> AbstractObject: + # A field of a field collapses onto itself, so chains such as ``node.next`` + # in a loop stay finite. + if owner.site.kind == "field": + return owner + return AbstractObject(AllocationSite("field", owner.site.location, f"{owner.site}.{field}")) + + def of(self, value: Value) -> set[AbstractObject]: + return self.points.setdefault(value, set()) + + def at(self, objects: set[AbstractObject], field: str) -> set[AbstractObject]: + found: set[AbstractObject] = set() + for owner in objects: + found |= self.heap.setdefault(HeapLocation(owner, field), set()) + found.add(self.field_object(owner, field)) + return found + + def store(self, objects: set[AbstractObject], field: str, value: Value) -> bool: + changed = False + for owner in objects: + location = self.heap.setdefault(HeapLocation(owner, field), set()) + before = len(location) + location |= self.of(value) + changed |= len(location) != before + return changed + + def solve(self) -> None: + changed = True + while changed: + changed = False + for block in self.function.blocks: + for instruction in block.instructions: + changed |= self.instruction(instruction) + terminator = block.terminator + if isinstance(terminator, ForNext) and terminator.result is not None: + changed |= self.assign(terminator.result, self.at(self.of(terminator.iterator), ELEMENTS)) + + def assign(self, value: Value, objects: set[AbstractObject]) -> bool: + current = self.of(value) + before = len(current) + current |= objects + return len(current) != before + + def instruction(self, instruction: Instruction) -> bool: + if isinstance(instruction, SetAttr): + return self.store(self.of(instruction.object), ATTRIBUTES, instruction.value) + if isinstance(instruction, SetItem): + return self.store(self.of(instruction.object), ELEMENTS, instruction.value) + result = instruction.result + if result is None: + return False + if isinstance(instruction, BuildList | BuildTuple | BuildDict | Call | WithEnter | Catch | Yield): + kind = { + BuildList: "list", + BuildTuple: "tuple", + BuildDict: "dict", + Call: "call", + WithEnter: "context", + Catch: "exception", + Yield: "sent", + }[type(instruction)] + site = AbstractObject(AllocationSite(kind, instruction.location)) + changed = self.assign(result, {site}) + if isinstance(instruction, BuildList | BuildTuple): + for element in instruction.elements: + changed |= self.store({site}, ELEMENTS, element) + elif isinstance(instruction, BuildDict): + for _, value in instruction.items: + changed |= self.store({site}, ELEMENTS, value) + elif isinstance(instruction, Call): + receiver = mutated_by(instruction, self.defs) + if receiver is not None: + for argument in instruction.argument_values(): + changed |= self.store(self.of(receiver), ELEMENTS, argument) + return changed + if isinstance(instruction, Global): + return self.assign(result, {self.named_object("global", instruction.name)}) + if isinstance(instruction, Symbol): + return self.assign(result, {self.named_object("symbol", instruction.symbol_id.canonical_name)}) + if isinstance(instruction, Phi): + objects: set[AbstractObject] = set() + for value, _ in instruction.incoming: + objects |= self.of(value) + return self.assign(result, objects) + if isinstance(instruction, GetAttr): + return self.assign(result, self.at(self.of(instruction.object), ATTRIBUTES)) + if isinstance(instruction, GetItem): + return self.assign(result, self.at(self.of(instruction.object), ELEMENTS)) + if isinstance(instruction, GetIter): + return self.assign(result, self.of(instruction.iterable)) + if isinstance(instruction, Await): + return self.assign(result, self.of(instruction.value)) + return False + + +def analyze_heap(function: FunctionIR) -> HeapFacts: + solver = _PointsTo(function) + solver.solve() + return HeapFacts({value: frozenset(objects) for value, objects in solver.points.items() if objects}) + + +class HeapAnalysis(FunctionAnalysis[HeapFacts]): + name: ClassVar[str] = "abstract.heap" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis}) + + @classmethod + def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> HeapFacts: + return analyze_heap(ctx.get(SSAAnalysis, function)) diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index c7350e8..17c3833 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -28,6 +28,7 @@ FunctionSummary, KnownFunction, ModuleGraph, + Mutation, SummaryIndex, Target, UnknownTarget, @@ -35,7 +36,7 @@ from coretrace_python.semantic.symbols import SymbolId from coretrace_python.source import SourceId, SourceSpan -CACHE_FORMAT = 1 +CACHE_FORMAT = 2 @dataclass(frozen=True) @@ -236,6 +237,16 @@ def _encode_summary(summary: FunctionSummary) -> dict[str, Any]: "external_calls": [_encode_call(call) for call in summary.external_calls], "unsupported": summary.unsupported, "return_externals": sorted(str(s) for s in summary.return_externals), + "mutations": [ + { + "parameter": m.parameter, + "field": m.field, + "dependencies": sorted(m.dependencies), + "externals": sorted(str(s) for s in m.externals), + } + for m in summary.mutations + ], + "side_effects": sorted(summary.side_effects), } @@ -247,6 +258,16 @@ def _decode_summary(data: Mapping[str, Any]) -> FunctionSummary: tuple(_decode_call(call) for call in data["external_calls"]), bool(data["unsupported"]), frozenset(SymbolId(_string(s)) for s in data["return_externals"]), + tuple( + Mutation( + _integer(m["parameter"]), + _string(m["field"]), + _indices(m["dependencies"]), + frozenset(SymbolId(_string(s)) for s in m["externals"]), + ) + for m in data["mutations"] + ), + frozenset(_string(name) for name in data["side_effects"]), ) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index cd838a6..227e357 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -15,7 +15,7 @@ from typing import Any, ClassVar from coretrace_python import __version__ -from coretrace_python.abstract import ConstantPropagation, RangeAnalysis +from coretrace_python.abstract import ConstantPropagation, HeapAnalysis, RangeAnalysis from coretrace_python.analysis import ( AnalysisContext, AnalysisManager, @@ -109,6 +109,7 @@ DefUseAnalysis, ConstantPropagation, RangeAnalysis, + HeapAnalysis, CallGraphAnalysis, SummaryAnalysis, ProjectSummaries, diff --git a/src/coretrace_python/interprocedural/__init__.py b/src/coretrace_python/interprocedural/__init__.py index b5d0533..53838ca 100644 --- a/src/coretrace_python/interprocedural/__init__.py +++ b/src/coretrace_python/interprocedural/__init__.py @@ -18,6 +18,7 @@ from coretrace_python.interprocedural.summaries import ( ExternalCall, FunctionSummary, + Mutation, ProjectSummaries, SummaryAnalysis, SummaryIndex, @@ -33,6 +34,7 @@ "FunctionSummary", "KnownFunction", "ModuleGraph", + "Mutation", "ProjectSummaries", "SummaryAnalysis", "SummaryIndex", diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index af4d1f8..41f096f 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -14,6 +14,14 @@ from types import MappingProxyType from typing import ClassVar +from coretrace_python.abstract import ( + ATTRIBUTES, + ELEMENTS, + HeapAnalysis, + HeapFacts, + HeapLocation, + mutated_by, +) from coretrace_python.analysis import Analysis, AnalysisContext, AnyAnalysis from coretrace_python.cfg import CFG, BlockId, CFGAnalysis from coretrace_python.dataflow import DataflowProblem, Direction, solve @@ -31,11 +39,16 @@ Constant, ForNext, FunctionIR, + GetAttr, + GetItem, + GetIter, Global, Instruction, Jump, Phi, Return, + SetAttr, + SetItem, Symbol, Undefined, Value, @@ -73,6 +86,17 @@ class ExternalCall: call_site: SourceSpan | None +@dataclass(frozen=True) +class Mutation: + """What the function stores into ``field`` of its ``parameter``: data of other + parameters, and results of external symbols (§22).""" + + parameter: int + field: str + dependencies: Dependencies + externals: frozenset[SymbolId] + + @dataclass(frozen=True) class FunctionSummary: name: str @@ -81,6 +105,8 @@ class FunctionSummary: external_calls: tuple[ExternalCall, ...] unsupported: bool = False return_externals: frozenset[SymbolId] = frozenset() + mutations: tuple[Mutation, ...] = () + side_effects: frozenset[str] = frozenset() class SummaryIndex: @@ -122,7 +148,8 @@ def summary(self, name: str) -> FunctionSummary: return self._summaries[name] -State = Mapping[Value, Dep] +Key = Value | HeapLocation +State = Mapping[Key, Dep] _CallKey = tuple[SymbolId, SourceSpan, SourceSpan | None] @@ -136,15 +163,40 @@ def __init__( graph: CallGraph, table: Mapping[str, FunctionSummary], project: Mapping[SymbolId, FunctionSummary] | None = None, + heap: HeapFacts | None = None, ) -> None: self.name = name self.function = function self.graph = graph self.table = table self.project = project or {} + self.heap = heap or HeapFacts({}) 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.external: dict[_CallKey, ExternalCall] = {} self.returns: Dep = EMPTY + self.stores: dict[HeapLocation, Dep] = {} + self.mutated_globals: set[str] = set() + + # ------------------------------------------------------------------ heap + + def deep(self, value: Value, state: Mapping[Key, Dep]) -> Dep: + """What a value depends on, contents of the objects it points to included.""" + + deps = state.get(value, EMPTY) + for field in (ELEMENTS, ATTRIBUTES): + for location in self.heap.locations(value, field): + deps |= state.get(location, EMPTY) + return deps + + def store(self, state: dict[Key, Dep], receiver: Value, field: str, deps: Dep) -> None: + for location in self.heap.locations(receiver, field): + state[location] = state.get(location, EMPTY) | deps + self.stores[location] = self.stores.get(location, EMPTY) | deps + if location.object.site.kind == "global": + self.mutated_globals.add(location.object.site.name) def initial(self) -> State: return MappingProxyType( @@ -159,10 +211,14 @@ def join(self, a: State, b: State) -> State: def evaluate(self, block: BasicBlock, incoming: Mapping[BlockId, State]) -> State: states = list(incoming.values()) - state: dict[Value, Dep] = dict(states[0]) if states else {} + state: dict[Key, Dep] = dict(states[0]) if states else {} for other in states[1:]: state = dict(self.join(state, other)) for instruction in block.instructions: + if isinstance(instruction, SetAttr): + self.store(state, instruction.object, ATTRIBUTES, state.get(instruction.value, EMPTY)) + elif isinstance(instruction, SetItem): + self.store(state, instruction.object, ELEMENTS, state.get(instruction.value, EMPTY)) if instruction.result is None: continue if isinstance(instruction, Phi): @@ -174,10 +230,12 @@ def evaluate(self, block: BasicBlock, incoming: Mapping[BlockId, State]) -> Stat deps = self.call(instruction, state) else: deps = self.instruction(instruction, state) + if isinstance(instruction, GetAttr | GetItem | GetIter): + deps |= self.loaded(instruction, state) state[instruction.result] = deps terminator = block.terminator if isinstance(terminator, ForNext) and terminator.result is not None: - state[terminator.result] = state.get(terminator.iterator, EMPTY) + state[terminator.result] = self.deep(terminator.iterator, state) if isinstance(terminator, Return) and terminator.value is not None: self.returns |= state.get(terminator.value, EMPTY) return MappingProxyType(state) @@ -198,7 +256,7 @@ def flow(self, cfg: CFG, block_id: BlockId, incoming: Mapping[BlockId, State]) - # ------------------------------------------------------------------ transfer @staticmethod - def instruction(instruction: Instruction, state: Mapping[Value, Dep]) -> Dep: + def instruction(instruction: Instruction, state: Mapping[Key, Dep]) -> Dep: if isinstance(instruction, Constant | Global | Symbol | Undefined): return EMPTY deps = EMPTY @@ -206,20 +264,37 @@ def instruction(instruction: Instruction, state: Mapping[Value, Dep]) -> Dep: deps |= state.get(operand, EMPTY) return deps - def call(self, call: Call, state: Mapping[Value, Dep]) -> Dep: - arguments = tuple(state.get(a, EMPTY) for a in call.arguments) + def loaded(self, instruction: GetAttr | GetItem | GetIter, state: Mapping[Key, Dep]) -> Dep: + """The contents a load reads from the objects it reads from.""" + + if isinstance(instruction, GetAttr): + receiver, field = instruction.object, ATTRIBUTES + elif isinstance(instruction, GetItem): + receiver, field = instruction.object, ELEMENTS + else: + receiver, field = instruction.iterable, ELEMENTS + deps = EMPTY + for location in self.heap.locations(receiver, field): + deps |= state.get(location, EMPTY) + return deps + + def call(self, call: Call, state: dict[Key, Dep]) -> Dep: + arguments = tuple(self.deep(a, state) for a in call.arguments) keywords = EMPTY for _, value in call.keywords: - keywords |= state.get(value, EMPTY) + keywords |= self.deep(value, state) everything = keywords for deps in arguments: everything |= deps + receiver = mutated_by(call, self.defs) + if receiver is not None: + self.store(state, receiver, ELEMENTS, everything) target = self.graph.target_at(self.name, call.location) if isinstance(target, ExternalSymbol): project = self.project.get(target.symbol) if project is not None: - return self.known(project, arguments, keywords, everything, call) + return self.known(project, arguments, keywords, everything, call, state) self.record( target.symbol, tuple(a.parameters for a in arguments), @@ -229,7 +304,7 @@ def call(self, call: Call, state: Mapping[Value, Dep]) -> Dep: ) return everything | Dep(externals=frozenset({target.symbol})) if isinstance(target, KnownFunction): - return self.known(self.table[target.name], arguments, keywords, everything, call) + return self.known(self.table[target.name], arguments, keywords, everything, call, state) assert isinstance(target, UnknownTarget) return everything | state.get(call.callee, EMPTY) @@ -240,6 +315,7 @@ def known( keywords: Dep, everything: Dep, call: Call, + state: dict[Key, Dep], ) -> Dep: """Dependencies of a call to a function whose summary is known.""" @@ -260,6 +336,10 @@ def mapped(indices: Dependencies) -> Dep: reached.location, call.location, ) + for mutation in callee.mutations: + if mutation.parameter < len(call.arguments): + deps = mapped(mutation.dependencies) | Dep(externals=mutation.externals) + self.store(state, call.arguments[mutation.parameter], mutation.field, deps) return mapped(callee.return_dependencies) | Dep(externals=callee.return_externals) def record( @@ -287,27 +367,40 @@ def summarize( graph: CallGraph, table: Mapping[str, FunctionSummary], project: Mapping[SymbolId, FunctionSummary] | None = None, + heap: HeapFacts | None = None, ) -> FunctionSummary: - problem = _DependenceProblem(name, function, graph, table, project) + problem = _DependenceProblem(name, function, graph, table, project, heap) solution = solve(problem, cfg) problem.external = {} problem.returns = EMPTY + problem.stores = {} + problem.mutated_globals = set() for block in function.blocks: if solution.reached(block.id): problem.evaluate(block, solution.incoming(block.id)) + mutations: list[Mutation] = [] + for index, parameter in enumerate(function.parameters): + for field in (ELEMENTS, ATTRIBUTES): + deps = EMPTY + for location in problem.heap.locations(parameter, field): + deps |= problem.stores.get(location, EMPTY) + if deps.parameters or deps.externals: + mutations.append(Mutation(index, field, deps.parameters, deps.externals)) return FunctionSummary( name, len(function.parameters), problem.returns.parameters, tuple(problem.external.values()), return_externals=problem.returns.externals, + mutations=tuple(mutations), + side_effects=frozenset(problem.mutated_globals), ) class SummaryAnalysis(Analysis[SummaryTable]): name: ClassVar[str] = "interprocedural.summaries" requires: ClassVar[frozenset[AnyAnalysis]] = frozenset( - {CallGraphAnalysis, SSAAnalysis, CFGAnalysis, ProjectSummaries} + {CallGraphAnalysis, SSAAnalysis, CFGAnalysis, ProjectSummaries, HeapAnalysis} ) @classmethod @@ -315,19 +408,23 @@ def compute(cls, ctx: AnalysisContext) -> SummaryTable: graph = ctx.get(CallGraphAnalysis) project = ctx.get(ProjectSummaries).summaries table: dict[str, FunctionSummary] = {} - supported: dict[str, tuple[FunctionIR, CFG]] = {} + supported: dict[str, tuple[FunctionIR, CFG, HeapFacts]] = {} for name, function in graph.definitions.items(): if name in graph.unsupported: count = len(function.parameters) table[name] = FunctionSummary(name, count, frozenset(range(count)), (), True) else: - supported[name] = (ctx.get(SSAAnalysis, function), ctx.get(CFGAnalysis, function)) + supported[name] = ( + ctx.get(SSAAnalysis, function), + ctx.get(CFGAnalysis, function), + ctx.get(HeapAnalysis, function), + ) table[name] = FunctionSummary(name, len(function.parameters), NONE, ()) changed = True while changed: changed = False - for name, (ssa, cfg) in supported.items(): - updated = summarize(name, ssa, cfg, graph, table, project) + for name, (ssa, cfg, heap) in supported.items(): + updated = summarize(name, ssa, cfg, graph, table, project, heap) if updated != table[name]: table[name] = updated changed = True diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index b00dacb..f366f31 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -14,6 +14,14 @@ from types import MappingProxyType from typing import ClassVar +from coretrace_python.abstract import ( + ATTRIBUTES, + ELEMENTS, + HeapAnalysis, + HeapFacts, + HeapLocation, + mutated_by, +) from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis from coretrace_python.cfg import CFG, BlockId, CFGAnalysis from coretrace_python.dataflow import DataflowProblem, Direction, solve @@ -36,9 +44,14 @@ Compare, ForNext, FunctionIR, + GetAttr, + GetItem, + GetIter, Instruction, Jump, Phi, + SetAttr, + SetItem, Symbol, Value, ) @@ -90,16 +103,24 @@ class TaintFlow: sink_location: SourceSpan | None = None +Key = Value | HeapLocation + + class TaintFacts: - def __init__(self, taints: Mapping[Value, Taint], flows: tuple[TaintFlow, ...]) -> None: + def __init__(self, taints: Mapping[Key, Taint], flows: tuple[TaintFlow, ...]) -> None: self._taints = MappingProxyType(dict(taints)) self.flows = flows def taint(self, value: Value) -> Taint: return self._taints.get(value, Taint.none()) + def heap(self, location: HeapLocation) -> Taint: + """The taint stored in one field of one abstract object (§22).""" + + return self._taints.get(location, Taint.none()) -State = Mapping[Value, Taint] + +State = Mapping[Key, Taint] class _TaintProblem(DataflowProblem[State]): @@ -114,6 +135,7 @@ def __init__( summaries: SummaryTable, parameters: Mapping[int, Source] | None = None, project: SummaryIndex | None = None, + heap: HeapFacts | None = None, ) -> None: self.name = name self.function = function @@ -122,9 +144,40 @@ def __init__( self.summaries = summaries self.parameters = parameters or {} self.project = project or SummaryIndex() + self.heap = heap or HeapFacts({}) 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.symbols = graph.symbols(name) + # ------------------------------------------------------------------ heap + + def deep(self, value: Value, state: Mapping[Key, Taint]) -> Taint: + """The taint of a value and of the contents of the objects it points to.""" + + taint = state.get(value, Taint.none()) + for field in (ELEMENTS, ATTRIBUTES): + for location in self.heap.locations(value, field): + taint = taint.join(state.get(location, Taint.none())) + return taint + + def store(self, state: dict[Key, Taint], receiver: Value, field: str, taint: Taint) -> None: + for location in self.heap.locations(receiver, field): + state[location] = state.get(location, Taint.none()).join(taint) + + def loaded(self, instruction: GetAttr | GetItem | GetIter, state: Mapping[Key, Taint]) -> Taint: + if isinstance(instruction, GetAttr): + receiver, field = instruction.object, ATTRIBUTES + elif isinstance(instruction, GetItem): + receiver, field = instruction.object, ELEMENTS + else: + receiver, field = instruction.iterable, ELEMENTS + taint = Taint.none() + for location in self.heap.locations(receiver, field): + taint = taint.join(state.get(location, Taint.none())) + return taint + def initial(self) -> State: return MappingProxyType( { @@ -144,11 +197,15 @@ def evaluate( self, block: BasicBlock, incoming: Mapping[BlockId, State] ) -> tuple[State, list[TaintFlow]]: states = list(incoming.values()) - state: dict[Value, Taint] = dict(states[0]) if states else {} + state: dict[Key, Taint] = dict(states[0]) if states else {} for other in states[1:]: state = dict(self.join(state, other)) flows: list[TaintFlow] = [] for instruction in block.instructions: + if isinstance(instruction, SetAttr): + 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())) if instruction.result is None: continue if isinstance(instruction, Phi): @@ -160,10 +217,12 @@ def evaluate( taint = self.call(instruction, state, flows) 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: - state[terminator.result] = state.get(terminator.iterator, Taint.none()) + state[terminator.result] = self.deep(terminator.iterator, state) return MappingProxyType(state), flows def flow(self, cfg: CFG, block_id: BlockId, incoming: Mapping[BlockId, State]) -> Mapping[BlockId, State]: @@ -181,7 +240,7 @@ def flow(self, cfg: CFG, block_id: BlockId, incoming: Mapping[BlockId, State]) - # ------------------------------------------------------------------ transfer - def instruction(self, instruction: Instruction, state: Mapping[Value, Taint]) -> Taint: + def instruction(self, instruction: Instruction, state: Mapping[Key, Taint]) -> Taint: taint = Taint.none() symbol = self.symbols.get(instruction.result) if instruction.result else None if symbol is not None: @@ -194,25 +253,28 @@ def instruction(self, instruction: Instruction, state: Mapping[Value, Taint]) -> taint = taint.join(state.get(operand, Taint.none())) return taint - def call(self, call: Call, state: Mapping[Value, Taint], flows: list[TaintFlow]) -> Taint: - arguments = tuple(state.get(a, Taint.none()) for a in call.arguments) + def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> Taint: + arguments = tuple(self.deep(a, state) for a in call.arguments) keywords = Taint.none() for _, value in call.keywords: - keywords = keywords.join(state.get(value, Taint.none())) + keywords = keywords.join(self.deep(value, state)) everything = keywords for taint in arguments: everything = everything.join(taint) + receiver = mutated_by(call, self.defs) + if receiver is not None: + self.store(state, receiver, ELEMENTS, everything) target = self.graph.target_at(self.name, call.location) if isinstance(target, ExternalSymbol): project = self.project.summary(target.symbol) if project is not None: through = target.symbol.canonical_name.removeprefix("python.") - return self.known(project, through, arguments, keywords, everything, call, flows) + return self.known(project, through, arguments, keywords, everything, call, flows, state) sink = self.models.sink(target.symbol) if sink is not None: for argument in call.argument_values(): - self.report(flows, sink, state.get(argument, Taint.none()), argument, call, None, None) + self.report(flows, sink, self.deep(argument, state), argument, call, None, None) sanitizer = self.models.sanitizer(target.symbol) if sanitizer is not None: return everything.without(sanitizer.kinds) @@ -224,7 +286,7 @@ def call(self, call: Call, state: Mapping[Value, Taint], flows: list[TaintFlow]) return everything if isinstance(target, KnownFunction): summary = self.summaries.summary(target.name) - return self.known(summary, target.name, arguments, keywords, everything, call, flows) + return self.known(summary, target.name, arguments, keywords, everything, call, flows, state) return everything.join(state.get(call.callee, Taint.none())) def known( @@ -236,6 +298,7 @@ def known( everything: Taint, call: Call, flows: list[TaintFlow], + state: dict[Key, Taint], ) -> Taint: """Flows and result taint of a call to a function whose summary is known.""" @@ -259,6 +322,14 @@ def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: taint, witness = mapped(deps) if witness is not None: self.report(flows, sink, taint, witness, call, through, reached.location) + for mutation in summary.mutations: + if mutation.parameter < len(call.arguments): + stored = mapped(mutation.dependencies)[0] + for symbol in sorted(mutation.externals, key=str): + stored_source = self.models.source(symbol) + if stored_source is not None: + stored = stored.join(Taint(stored_source.kinds, frozenset({stored_source}))) + self.store(state, call.arguments[mutation.parameter], mutation.field, stored) result = mapped(summary.return_dependencies)[0] for symbol in sorted(summary.return_externals, key=str): returned_source = self.models.source(symbol) @@ -364,10 +435,11 @@ def propagate_taint( summaries: SummaryTable, parameters: Mapping[int, Source] | None = None, project: SummaryIndex | None = None, + heap: HeapFacts | None = None, ) -> TaintFacts: - problem = _TaintProblem(name, function, models, graph, summaries, parameters, project) + problem = _TaintProblem(name, function, models, graph, summaries, parameters, project, heap) solution = solve(problem, cfg) - taints: dict[Value, Taint] = {} + taints: dict[Key, Taint] = {} flows: list[TaintFlow] = [] for block in function.blocks: if solution.reached(block.id): @@ -391,6 +463,7 @@ class TaintAnalysis(FunctionAnalysis[TaintFacts]): ScopeAnalysis, SymbolAnalysis, ProjectSummaries, + HeapAnalysis, } ) @@ -409,4 +482,5 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> TaintFacts: function, ctx.module, models, ctx.get(ScopeAnalysis), ctx.get(SymbolAnalysis) ), ctx.get(ProjectSummaries), + ctx.get(HeapAnalysis, function), ) diff --git a/tests/test_architecture.py b/tests/test_architecture.py index 7da12c0..140c6d1 100644 --- a/tests/test_architecture.py +++ b/tests/test_architecture.py @@ -22,7 +22,8 @@ # Dependency direction of the pipeline (§2, §37). A package may import only packages # with a strictly lower layer number, or itself. ``analysis`` is infrastructure every -# provider subclasses, so it sits below the analyses it manages (§8). +# provider subclasses, so it sits below the analyses it manages (§8). Function summaries +# consume the heap domain (§22), so ``interprocedural`` sits above ``abstract``. LAYERS = { "source": 0, "hir": 1, @@ -33,12 +34,12 @@ "ir": 5, "dataflow": 6, "abstract": 7, - "interprocedural": 7, - "taint": 8, - "findings": 9, - "dependency": 10, - "plugins": 11, - "reporters": 12, + "interprocedural": 8, + "taint": 9, + "findings": 10, + "dependency": 11, + "plugins": 12, + "reporters": 13, } # Top-level modules that wire the pipeline together and may import any layer. diff --git a/tests/test_heap.py b/tests/test_heap.py new file mode 100644 index 0000000..ea8bf9c --- /dev/null +++ b/tests/test_heap.py @@ -0,0 +1,331 @@ +"""Acceptance tests for the heap and aliasing abstraction (``docs/architecture.md`` §22, +§19; roadmap issue #35). + +SSA names values, not objects: ``b = a; b.append(x); sink(a[0])`` is invisible to a +value-only taint. The coarse abstraction of §22 adds one ``AbstractObject`` per +allocation site (containers, calls, parameters, globals, loaded fields), an ``AliasSet`` +of objects per value computed by a flow-insensitive points-to fixpoint, and two +``HeapLocation`` fields per object, ``elements`` and ``attributes``. Taint and +dependence flow through stores, mutating method calls and loads on those locations, and +function summaries record the parameters a function mutates and the module globals it +touches, so mutations cross calls and files. + +Expected to remain red until ``abstract.heap``, heap-aware taint and summaries with +``mutations`` and ``side_effects`` exist. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.analysis import AnalysisManager +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.interprocedural import FunctionSummary, SummaryAnalysis +from coretrace_python.ir.model import ( + BuildList, + FunctionIR, + GetAttr, + GetItem, + Instruction, + Phi, + Value, +) +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager +from coretrace_python.taint import ( + SecurityModelAnalysis, + SecurityModelRegistry, + Sink, + Source, + TaintAnalysis, + TaintKind, +) + +try: + from coretrace_python.abstract import HeapAnalysis, HeapFacts + from coretrace_python.abstract.heap import AllocationSite, HeapLocation + from coretrace_python.interprocedural import Mutation +except ImportError as error: # pragma: no cover - red until the heap lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_heap() -> None: + if MISSING is not None: + pytest.fail(f"heap abstraction is not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + +MODELS = ( + Source(SymbolId("python.builtins.input"), "stdin"), + Sink(SymbolId("python.os.system"), TaintKind.COMMAND), +) +PRELUDE = "import os\nfrom config import Config\n\n" + + +def manager_for(body: str, prelude: str = PRELUDE) -> 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 heap_of(body: str, name: str = "f") -> tuple[FunctionIR, HeapFacts]: + manager = manager_for(body) + function = function_named(manager, name) + return manager.get(SSAAnalysis, function), manager.get(HeapAnalysis, function) + + +def instructions(ssa: FunctionIR, kind: type[Instruction]) -> list[Instruction]: + return [i for block in ssa.blocks for i in block.instructions if isinstance(i, kind)] + + +def result_of(instruction: Instruction) -> Value: + assert instruction.result is not None + return instruction.result + + +def flow_lines(body: str, name: str = "f") -> list[int]: + manager = manager_for(body) + function = function_named(manager, name) + return sorted(flow.location.start_line for flow in manager.get(TaintAnalysis, function).flows) + + +def summary_of(body: str, name: str) -> FunctionSummary: + manager = manager_for(body, "import os\n\n") + return manager.get(SummaryAnalysis).summary(name) + + +def check_lines(text: str) -> list[tuple[str, int]]: + findings = engine.check(SourceManager().add_source("app.py", text), [PLUGINS]) + return sorted((f.rule_id, f.span.start_line) for f in findings) + + +# --------------------------------------------------------------------------- points-to + + +def test_containers_get_one_object_per_allocation_site() -> None: + ssa, heap = heap_of("def f():\n a = []\n b = []\n return a, b\n") + first, second = instructions(ssa, BuildList) + + (object_a,) = heap.objects(result_of(first)) + (object_b,) = heap.objects(result_of(second)) + + assert object_a != object_b + assert object_a.site == AllocationSite("list", first.location) + assert str(object_a.site) == "list@m.py:5" + assert str(HeapLocation(object_a, "elements")) == "list@m.py:5.elements" + assert HeapAnalysis.name == "abstract.heap" + assert SSAAnalysis in HeapAnalysis.requires + assert HeapAnalysis in engine.ALL_ANALYSES + + +def test_aliases_merge_through_phis_and_stay_separate_otherwise() -> None: + ssa, heap = heap_of( + "def f(c):\n a = []\n b = []\n if c:\n x = a\n else:\n x = b\n return x, a\n" + ) + lists = instructions(ssa, BuildList) + (phi_value,) = instructions(ssa, Phi) + + objects = heap.objects(result_of(phi_value)) + + assert objects == heap.objects(result_of(lists[0])) | heap.objects(result_of(lists[1])) + assert len(objects) == 2 + + +def test_parameters_and_globals_are_objects_of_their_own() -> None: + ssa, heap = heap_of("cache = []\n\ndef f(items, other):\n cache.append(items)\n return other\n") + + (items,) = heap.objects(ssa.parameters[0]) + (other,) = heap.objects(ssa.parameters[1]) + + assert items.site.kind == "parameter" and other.site.kind == "parameter" + assert items != other + assert any(o.site.kind == "global" and o.site.name == "cache" for value in heap.values for o in heap.objects(value)) + + +def test_loaded_fields_are_deterministic_objects() -> None: + ssa, heap = heap_of("def f():\n cfg = Config()\n first = cfg.cmd\n second = cfg.cmd\n return first, second\n") + first, second = instructions(ssa, GetAttr) + + assert heap.objects(result_of(first)) == heap.objects(result_of(second)) + (field,) = heap.objects(result_of(first)) + assert field.site.kind == "field" + + +def test_stores_feed_later_loads() -> None: + ssa, heap = heap_of("def f():\n inner = []\n box = {}\n box['k'] = inner\n out = box['k']\n return out\n") + (inner,) = instructions(ssa, BuildList) + (load,) = instructions(ssa, GetItem) + + assert heap.objects(result_of(inner)) <= heap.objects(result_of(load)) + + +def test_field_chains_in_loops_terminate() -> None: + _, heap = heap_of("def f(node):\n while node:\n node = node.next\n return node\n") + assert heap is not None + + +# --------------------------------------------------------------------------- taint through the heap + + +def test_the_issue_example_is_a_flow() -> None: + assert flow_lines("def f():\n a = []\n b = a\n b.append(input())\n os.system(a[0])\n") == [8] + + +@pytest.mark.parametrize( + "body", + [ + "def f():\n d = {}\n d['x'] = input()\n os.system(d['x'])\n", + "def f():\n cfg = Config()\n cfg.cmd = input()\n os.system(cfg.cmd)\n", + "def f():\n a = []\n a.extend([input()])\n os.system(a[0])\n", + "def f():\n a = []\n a.insert(0, input())\n os.system(a[-1])\n", + "def f():\n s = set()\n s.add(input())\n os.system(list(s)[0])\n", + "def f():\n d = {}\n d.update(x=input())\n os.system(d['x'])\n", + "def f():\n a = []\n a.append(input())\n for item in a:\n os.system(item)\n", + "def f():\n a = []\n a.append(input())\n os.system(' '.join(a))\n", + "def f(c):\n a = []\n b = []\n if c:\n x = a\n else:\n x = b\n x.append(input())\n os.system(a[0])\n", + ], +) +def test_stores_mutators_and_loads_carry_taint(body: str) -> None: + assert len(flow_lines(body)) == 1 + + +@pytest.mark.parametrize( + "body", + [ + "def f():\n a = []\n b = []\n a.append(input())\n os.system(b[0])\n", + "def f():\n a = []\n a.append('ls')\n os.system(a[0])\n", + "def f():\n cfg = Config()\n other = Config()\n cfg.cmd = input()\n os.system(other.cmd)\n", + ], +) +def test_unrelated_objects_and_clean_stores_stay_clean(body: str) -> None: + assert flow_lines(body) == [] + + +def test_heap_taint_is_exposed_by_location() -> None: + manager = manager_for("def f():\n a = []\n a.append(input())\n return a\n") + function = function_named(manager, "f") + ssa = manager.get(SSAAnalysis, function) + heap = manager.get(HeapAnalysis, function) + (allocation,) = instructions(ssa, BuildList) + (obj,) = heap.objects(result_of(allocation)) + + facts = manager.get(TaintAnalysis, function) + + assert facts.heap(HeapLocation(obj, "elements")).kinds == TaintKind.ALL + assert not facts.heap(HeapLocation(obj, "attributes")) + assert facts.taint(result_of(allocation)).kinds == TaintKind.NONE + + +# --------------------------------------------------------------------------- summaries + + +def test_summaries_record_parameter_mutations() -> None: + summary = summary_of("def fill(items, value):\n items.append(value)\n", "fill") + + assert summary.mutations == (Mutation(0, "elements", frozenset({1}), frozenset()),) + assert summary.side_effects == frozenset() + + +def test_summaries_record_sources_stored_into_parameters_and_attribute_stores() -> None: + fill = summary_of("def fill(items):\n items.append(input())\n", "fill") + configure = summary_of("def configure(cfg, cmd):\n cfg.command = cmd\n", "configure") + + assert fill.mutations == (Mutation(0, "elements", frozenset(), frozenset({SymbolId("python.builtins.input")})),) + assert configure.mutations == (Mutation(0, "attributes", frozenset({1}), frozenset()),) + + +def test_summaries_record_side_effects_on_module_globals() -> None: + summary = summary_of("cache = []\n\ndef touch(x):\n cache.append(x)\n", "touch") + + assert summary.side_effects == frozenset({"cache"}) + assert summary.mutations == () + + +def test_pure_functions_have_no_mutations() -> None: + summary = summary_of("def add(a, b):\n return a + b\n", "add") + assert summary.mutations == () and summary.side_effects == frozenset() + + +def test_mutations_compose_through_callers() -> None: + summary = summary_of( + "def fill(items, value):\n items.append(value)\n\ndef wrap(xs, v):\n fill(xs, v)\n", + "wrap", + ) + assert summary.mutations == (Mutation(0, "elements", frozenset({1}), frozenset()),) + + +def test_mutations_round_trip_through_the_cache_codec() -> None: + from coretrace_python.cache import CachedModule, decode, encode + + summary = summary_of("def fill(items):\n items.append(input())\n", "fill") + + restored = decode(json.loads(json.dumps(encode(CachedModule(("fill",), {"fill": summary}, (), ()))))) + + assert restored.summaries["fill"] == summary + + +# --------------------------------------------------------------------------- interprocedural + + +def test_taint_flows_through_a_callee_that_mutates_its_argument() -> None: + body = ( + "def fill(items, value):\n items.append(value)\n\n" + "def run():\n a = []\n fill(a, input())\n os.system(a[0])\n" + ) + assert flow_lines(body, "run") == [10] + + +def test_taint_flows_from_a_source_stored_by_the_callee() -> None: + body = "def fill(items):\n items.append(input())\n\ndef run():\n a = []\n fill(a)\n os.system(a[0])\n" + assert flow_lines(body, "run") == [10] + + +def test_callees_that_do_not_mutate_leave_arguments_clean() -> None: + body = "def peek(items, value):\n return len(items)\n\ndef run():\n a = []\n peek(a, input())\n os.system(a[0])\n" + assert flow_lines(body, "run") == [] + + +def test_mutations_cross_files_through_the_project_index(tmp_path: Path) -> None: + (tmp_path / "helpers.py").write_text("def fill(items, value):\n items.append(value)\n", encoding="utf-8") + (tmp_path / "main.py").write_text( + "import os\nfrom helpers import fill\n\ndef run():\n a = []\n fill(a, input())\n os.system(a[0])\n", + encoding="utf-8", + ) + + findings = engine.analyze_project(tmp_path, [PLUGINS]).findings + + assert [(f.rule_id, f.span.start_line, f.metadata.get("through")) for f in findings] == [ + ("command-injection", 7, None) + ] + + +# --------------------------------------------------------------------------- with the rest + + +def test_refutation_still_judges_container_borne_values() -> None: + assert check_lines( + "import os\n\ndef f():\n a = []\n a.append(input())\n x = a[0]\n if x.isdigit():\n os.system(x)\n" + ) == [] + assert check_lines("import os\n\ndef f():\n a = []\n a.append(input())\n os.system(a[0])\n") == [ + ("command-injection", 6) + ] + assert check_lines("import os\n\ndef f():\n a = []\n a.append(input())\n os.system(len(a))\n") == []