From 78f1327b2c4b2f945407c7da21a6e136dcf80b82 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 3 Sep 2026 16:59:14 +0200 Subject: [PATCH 1/2] test: define data-flow framework and constant propagation behavior --- tests/test_architecture.py | 12 +- tests/test_constants.py | 228 +++++++++++++++++++++++++++++++++++++ tests/test_dataflow.py | 221 +++++++++++++++++++++++++++++++++++ 3 files changed, 455 insertions(+), 6 deletions(-) create mode 100644 tests/test_constants.py create mode 100644 tests/test_dataflow.py diff --git a/tests/test_architecture.py b/tests/test_architecture.py index 6ada799..af0c79a 100644 --- a/tests/test_architecture.py +++ b/tests/test_architecture.py @@ -32,12 +32,12 @@ "cfg": 4, "ir": 5, "dataflow": 6, - "abstract": 6, - "interprocedural": 6, - "taint": 7, - "findings": 8, - "plugins": 9, - "reporters": 10, + "abstract": 7, + "interprocedural": 7, + "taint": 8, + "findings": 9, + "plugins": 10, + "reporters": 11, } # Top-level modules that wire the pipeline together and may import any layer. diff --git a/tests/test_constants.py b/tests/test_constants.py new file mode 100644 index 0000000..041b397 --- /dev/null +++ b/tests/test_constants.py @@ -0,0 +1,228 @@ +"""Acceptance tests for abstract values and constant propagation (§18, §38 Phase 4). + +``AbstractValue`` records what is known about one SSA value: its constant (a flat +lattice), its possible types and its truthiness. ``ConstantPropagation`` is the first +client of the data-flow solver: a forward problem over the SSA form that folds pure +operations on constants, joins at phis over executable edges only, and prunes the +branches a constant condition never takes. + +Expected to remain red until ``coretrace_python.abstract`` exists. +""" + +from __future__ import annotations + +import pytest + +from coretrace_python.analysis import AnalysisManager +from coretrace_python.cfg import BlockId, CFGAnalysis, DominanceAnalysis +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.ir.lowering import PyIRAnalysis +from coretrace_python.ir.model import FunctionIR, Phi, Return +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.semantic import SEMANTIC_ANALYSES +from coretrace_python.source import SourceManager + +try: + from coretrace_python.abstract import ( + AbstractValue, + ConstantFacts, + ConstantPropagation, + Truth, + ) + from coretrace_python.dataflow import BOTTOM, TOP +except ImportError as error: # pragma: no cover - red until abstract values land + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_abstract() -> None: + if MISSING is not None: + pytest.fail(f"abstract values are not implemented yet: {MISSING}") + + +def analyze(source_text: str) -> tuple[FunctionIR, ConstantFacts]: + module = build_hir(SourceManager().add_source("const.py", source_text)) + manager = AnalysisManager(module) + manager.register( + *SEMANTIC_ANALYSES, + CFGAnalysis, + DominanceAnalysis, + PyIRAnalysis, + SSAAnalysis, + ConstantPropagation, + ) + function = next(s for s in module.body if isinstance(s, nodes.Function)) + return manager.get(SSAAnalysis, function), manager.get(ConstantPropagation, function) + + +def returned(function: FunctionIR, facts: ConstantFacts, block: str = "entry") -> AbstractValue: + terminator = next(b for b in function.blocks if b.id == BlockId(block)).terminator + assert isinstance(terminator, Return) and terminator.value is not None + return facts.value(terminator.value) + + +# --------------------------------------------------------------------------- abstract values + + +def test_abstract_value_of_a_constant() -> None: + value = AbstractValue.of(3) + + assert value.constant == 3 + assert value.types == frozenset({"int"}) + assert value.truthiness is Truth.TRUE + assert AbstractValue.of(0).truthiness is Truth.FALSE + assert AbstractValue.of("").truthiness is Truth.FALSE + assert AbstractValue.of(None).types == frozenset({"NoneType"}) + + +def test_unknown_abstract_value() -> None: + value = AbstractValue.unknown() + + assert value.constant is TOP + assert value.types is None + assert value.truthiness is Truth.UNKNOWN + assert AbstractValue.bottom().constant is BOTTOM + + +def test_join_keeps_agreement_and_widens_disagreement() -> None: + assert AbstractValue.of(1).join(AbstractValue.of(1)) == AbstractValue.of(1) + joined = AbstractValue.of(1).join(AbstractValue.of(2)) + assert joined.constant is TOP + assert joined.types == frozenset({"int"}) + assert joined.truthiness is Truth.UNKNOWN + assert AbstractValue.of(1).join(AbstractValue.of("a")).types == frozenset({"int", "str"}) + assert AbstractValue.of(1).join(AbstractValue.unknown()).types is None + assert AbstractValue.bottom().join(AbstractValue.of(1)) == AbstractValue.of(1) + + +def test_truthiness_survives_a_lost_constant_when_types_agree() -> None: + joined = AbstractValue.of(1).join(AbstractValue.of(2)) + assert joined.truthiness is Truth.UNKNOWN + known_truthy = AbstractValue.of(1).join(AbstractValue.of(True)) + assert known_truthy.truthiness is Truth.UNKNOWN + + +def test_abstract_values_are_immutable() -> None: + import dataclasses + + with pytest.raises(dataclasses.FrozenInstanceError): + AbstractValue.of(1).constant = 2 # type: ignore[misc] + + +# --------------------------------------------------------------------------- folding + + +def test_constants_propagate_through_pure_operations() -> None: + function, facts = analyze("def f():\n x = 1\n y = x + 2\n return y * 3\n") + assert returned(function, facts).constant == 9 + + +@pytest.mark.parametrize( + "expression, expected", + [ + ("2 - 5", -3), + ("7 // 2", 3), + ("7 % 3", 1), + ("6 / 3", 2.0), + ("'a' + 'b'", "ab"), + ("3 < 4", True), + ("'x' == 'x'", True), + ("None is None", True), + ("not 0", True), + ("-4", -4), + ("6 & 3", 2), + ("6 | 3", 7), + ], +) +def test_folds_arithmetic_comparisons_and_unary_operators(expression: str, expected: object) -> None: + function, facts = analyze(f"def f():\n return {expression}\n") + value = returned(function, facts) + assert value.constant == expected + assert type(value.constant) is type(expected) + + +@pytest.mark.parametrize("expression", ["1 / 0", "2 ** 100", "1 << 3", "'a' * 3"]) +def test_unsafe_or_unbounded_operations_are_not_folded(expression: str) -> None: + function, facts = analyze(f"def f():\n return {expression}\n") + assert returned(function, facts).constant is TOP + + +def test_parameters_globals_and_calls_are_unknown() -> None: + function, facts = analyze("def f(a):\n b = a + 1\n c = g(b)\n return c\n") + assert facts.value(function.parameters[0]).constant is TOP + assert returned(function, facts).constant is TOP + assert returned(function, facts).types is None + + +def test_types_are_tracked_without_constants() -> None: + function, facts = analyze("def f(c):\n if c:\n x = 1\n else:\n x = 2\n return x\n") + value = returned(function, facts, "merge_1") + + assert value.constant is TOP + assert value.types == frozenset({"int"}) + + +# --------------------------------------------------------------------------- control flow + + +def test_phi_of_equal_constants_stays_constant() -> None: + function, facts = analyze("def f(c):\n if c:\n x = 1\n else:\n x = 1\n return x\n") + assert returned(function, facts, "merge_1").constant == 1 + + +def test_constant_condition_prunes_the_dead_branch() -> None: + function, facts = analyze( + "def f():\n if True:\n x = 1\n else:\n x = 2\n return x\n" + ) + merge = next(b for b in function.blocks if b.id == BlockId("merge_1")) + phi = merge.instructions[0] + assert isinstance(phi, Phi) + + assert facts.reachable(BlockId("then_1")) + assert not facts.reachable(BlockId("else_1")) + assert facts.value(phi.result).constant == 1 + + +def test_computed_constant_condition_also_prunes() -> None: + function, facts = analyze( + "def f():\n debug = 1 > 2\n if debug:\n x = 'dev'\n else:\n x = 'prod'\n return x\n" + ) + assert returned(function, facts, "merge_1").constant == "prod" + assert not facts.reachable(BlockId("then_1")) + + +def test_loop_counter_is_not_constant_but_keeps_its_type() -> None: + function, facts = analyze("def f():\n n = 0\n while n < 10:\n n = n + 1\n return n\n") + value = returned(function, facts, "exit_1") + + assert value.constant is TOP + assert value.types == frozenset({"int"}) + + +def test_loop_invariant_stays_constant() -> None: + function, facts = analyze( + "def f(items):\n limit = 10\n for item in items:\n use(item, limit)\n return limit\n" + ) + assert returned(function, facts, "exit_1").constant == 10 + + +def test_values_in_unreached_blocks_have_no_facts() -> None: + function, facts = analyze("def f():\n if False:\n x = 1\n return x\n return 0\n") + + assert not facts.reachable(BlockId("then_1")) + then_block = next(b for b in function.blocks if b.id == BlockId("then_1")) + assert facts.value(then_block.instructions[0].result).constant is BOTTOM + + +# --------------------------------------------------------------------------- analysis wiring + + +def test_constant_propagation_is_declared_through_the_manager() -> None: + assert ConstantPropagation.name == "abstract.constants" + assert {SSAAnalysis, CFGAnalysis} <= ConstantPropagation.requires + function, facts = analyze("def f():\n return 1\n") + assert isinstance(facts, ConstantFacts) + assert facts.reachable(function.entry) diff --git a/tests/test_dataflow.py b/tests/test_dataflow.py new file mode 100644 index 0000000..7ccd0d1 --- /dev/null +++ b/tests/test_dataflow.py @@ -0,0 +1,221 @@ +"""Acceptance tests for the generic data-flow framework (``docs/architecture.md`` §37). + +``dataflow.lattice`` gives the lattice vocabulary (``BOTTOM``, ``TOP``, ``FlatLattice``). +``dataflow.solver`` runs a worklist over a CFG in either direction. A problem receives, +for each block, the states arriving on its executable incoming edges and returns the +states it sends along outgoing edges; edges it does not return are pruned, which is how +constant branches remove paths. + +Expected to remain red until ``coretrace_python.dataflow`` exists. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import ClassVar + +import pytest + +from coretrace_python.cfg import CFG, BlockId, build_cfg +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.source import SourceManager + +try: + from coretrace_python.dataflow import ( + BOTTOM, + ENTRY, + TOP, + DataflowProblem, + Direction, + FlatLattice, + Solution, + solve, + ) +except ImportError as error: # pragma: no cover - red until the framework lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_dataflow() -> None: + if MISSING is not None: + pytest.fail(f"data-flow framework is not implemented yet: {MISSING}") + + +def cfg_for(source_text: str) -> CFG: + module = build_hir(SourceManager().add_source("flow.py", source_text)) + return build_cfg(next(s for s in module.body if isinstance(s, nodes.Function))) + + +def b(name: str) -> BlockId: + return BlockId(name) + + +IF_ELSE = "def f(c):\n if c:\n x = 1\n else:\n x = 2\n return x\n" +WHILE = "def f(n):\n while n:\n n = n - 1\n return n\n" +EARLY_RETURN = "def f(a):\n if a:\n return 1\n return 0\n" + + +# --------------------------------------------------------------------------- lattice + + +def test_flat_lattice_joins_equal_elements_and_tops_different_ones() -> None: + lattice: FlatLattice[int] = FlatLattice() + + assert lattice.join(BOTTOM, 1) == 1 + assert lattice.join(1, BOTTOM) == 1 + assert lattice.join(1, 1) == 1 + assert lattice.join(1, 2) is TOP + assert lattice.join(TOP, 1) is TOP + assert lattice.join(BOTTOM, BOTTOM) is BOTTOM + + +def test_flat_lattice_partial_order() -> None: + lattice: FlatLattice[str] = FlatLattice() + + assert lattice.leq(BOTTOM, "a") + assert lattice.leq("a", "a") + assert lattice.leq("a", TOP) + assert not lattice.leq("a", "b") + assert not lattice.leq(TOP, "a") + assert lattice.bottom is BOTTOM + assert lattice.top is TOP + + +def test_lattice_elements_distinguish_types() -> None: + lattice: FlatLattice[object] = FlatLattice() + + assert lattice.join(1, True) is TOP + assert lattice.join(0, False) is TOP + assert lattice.join(1.0, 1) is TOP + + +# --------------------------------------------------------------------------- solver + + +if MISSING is None: + + class Reaching(DataflowProblem[frozenset[BlockId]]): + """Forward: which blocks lie on some path to each block.""" + + direction: ClassVar[Direction] = Direction.FORWARD + + def initial(self) -> frozenset[BlockId]: + return frozenset() + + def join(self, a: frozenset[BlockId], b: frozenset[BlockId]) -> frozenset[BlockId]: + return a | b + + def flow( + self, + cfg: CFG, + block: BlockId, + incoming: Mapping[BlockId, frozenset[BlockId]], + ) -> Mapping[BlockId, frozenset[BlockId]]: + state = frozenset().union(*incoming.values()) | {block} + return {successor: state for successor in cfg.successors(block)} + + class Exiting(DataflowProblem[frozenset[BlockId]]): + """Backward: which blocks can be reached from each block.""" + + direction: ClassVar[Direction] = Direction.BACKWARD + + def initial(self) -> frozenset[BlockId]: + return frozenset() + + def join(self, a: frozenset[BlockId], b: frozenset[BlockId]) -> frozenset[BlockId]: + return a | b + + def flow( + self, + cfg: CFG, + block: BlockId, + incoming: Mapping[BlockId, frozenset[BlockId]], + ) -> Mapping[BlockId, frozenset[BlockId]]: + state = frozenset().union(*incoming.values()) | {block} + return {predecessor: state for predecessor in cfg.predecessors(block)} + + class PruneElse(Reaching): + """Forward, but never follows the else edge of the entry branch.""" + + def flow( + self, + cfg: CFG, + block: BlockId, + incoming: Mapping[BlockId, frozenset[BlockId]], + ) -> Mapping[BlockId, frozenset[BlockId]]: + out = dict(super().flow(cfg, block, incoming)) + if block == cfg.entry: + out.pop(b("else_1")) + return out + + class Counting(Reaching): + visits: ClassVar[list[BlockId]] = [] + + def flow( + self, + cfg: CFG, + block: BlockId, + incoming: Mapping[BlockId, frozenset[BlockId]], + ) -> Mapping[BlockId, frozenset[BlockId]]: + Counting.visits.append(block) + return super().flow(cfg, block, incoming) + + +def test_forward_problem_reaches_a_fixpoint() -> None: + solution = solve(Reaching(), cfg_for(IF_ELSE)) + + assert isinstance(solution, Solution) + assert solution.incoming(b("entry")) == {ENTRY: frozenset()} + assert solution.state(b("then_1")) == frozenset({b("entry")}) + assert solution.state(b("merge_1")) == frozenset({b("entry"), b("then_1"), b("else_1")}) + assert solution.edge(b("then_1"), b("merge_1")) == frozenset({b("entry"), b("then_1")}) + + +def test_backward_problem_starts_from_the_exits() -> None: + solution = solve(Exiting(), cfg_for(EARLY_RETURN)) + + assert solution.incoming(b("then_1")) == {ENTRY: frozenset()} + assert solution.incoming(b("merge_1")) == {ENTRY: frozenset()} + assert solution.state(b("entry")) == frozenset({b("then_1"), b("merge_1")}) + + +def test_loops_converge() -> None: + solution = solve(Reaching(), cfg_for(WHILE)) + + assert solution.state(b("loop_1")) == frozenset({b("entry"), b("loop_1"), b("body_1")}) + assert solution.state(b("exit_1")) == frozenset({b("entry"), b("loop_1"), b("body_1")}) + + +def test_pruned_edges_make_blocks_unreached() -> None: + solution = solve(PruneElse(), cfg_for(IF_ELSE)) + + assert solution.reached(b("then_1")) + assert not solution.reached(b("else_1")) + assert solution.incoming(b("else_1")) == {} + assert solution.state(b("merge_1")) == frozenset({b("entry"), b("then_1")}) + assert solution.incoming(b("merge_1")) == { + b("then_1"): frozenset({b("entry"), b("then_1")}), + } + + +def test_unreachable_code_is_never_visited() -> None: + Counting.visits = [] + solve(Counting(), cfg_for("def f():\n return 1\n x = 2\n")) + + assert Counting.visits == [b("entry")] + + +def test_blocks_are_revisited_only_when_their_inputs_change() -> None: + Counting.visits = [] + solve(Counting(), cfg_for(IF_ELSE)) + + assert Counting.visits == [b("entry"), b("then_1"), b("else_1"), b("merge_1")] + + +def test_solutions_are_immutable() -> None: + solution = solve(Reaching(), cfg_for(IF_ELSE)) + with pytest.raises(TypeError): + solution.incoming(b("merge_1"))[b("x")] = frozenset() # type: ignore[index] From e5b38c2ef96ca29b5f65ebac75a6e8db6a1a8884 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 3 Sep 2026 16:59:15 +0200 Subject: [PATCH 2/2] feat: add data-flow solver, abstract values and constant propagation --- src/coretrace_python/abstract/__init__.py | 16 ++ src/coretrace_python/abstract/constants.py | 225 +++++++++++++++++++++ src/coretrace_python/abstract/values.py | 56 +++++ src/coretrace_python/dataflow/__init__.py | 29 +++ src/coretrace_python/dataflow/lattice.py | 78 +++++++ src/coretrace_python/dataflow/solver.py | 96 +++++++++ 6 files changed, 500 insertions(+) create mode 100644 src/coretrace_python/abstract/__init__.py create mode 100644 src/coretrace_python/abstract/constants.py create mode 100644 src/coretrace_python/abstract/values.py create mode 100644 src/coretrace_python/dataflow/__init__.py create mode 100644 src/coretrace_python/dataflow/lattice.py create mode 100644 src/coretrace_python/dataflow/solver.py diff --git a/src/coretrace_python/abstract/__init__.py b/src/coretrace_python/abstract/__init__.py new file mode 100644 index 0000000..b0ee686 --- /dev/null +++ b/src/coretrace_python/abstract/__init__.py @@ -0,0 +1,16 @@ +"""Abstract values and the analyses that compute them (architecture §18).""" + +from coretrace_python.abstract.constants import ( + ConstantFacts, + ConstantPropagation, + propagate_constants, +) +from coretrace_python.abstract.values import AbstractValue, Truth + +__all__ = [ + "AbstractValue", + "ConstantFacts", + "ConstantPropagation", + "Truth", + "propagate_constants", +] diff --git a/src/coretrace_python/abstract/constants.py b/src/coretrace_python/abstract/constants.py new file mode 100644 index 0000000..4ffe733 --- /dev/null +++ b/src/coretrace_python/abstract/constants.py @@ -0,0 +1,225 @@ +"""Constant propagation: the reference client of the data-flow solver (§18, §38).""" + +from __future__ import annotations + +import operator +from collections.abc import Callable, Mapping +from types import MappingProxyType +from typing import Any, ClassVar + +from coretrace_python.abstract.values import AbstractValue, Truth +from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis +from coretrace_python.cfg import CFG, BlockId, CFGAnalysis +from coretrace_python.dataflow import BOTTOM, TOP, DataflowProblem, Direction, solve +from coretrace_python.hir import nodes +from coretrace_python.ir.model import ( + BasicBlock, + BinaryOp, + Branch, + Compare, + Constant, + ForNext, + FunctionIR, + Instruction, + Jump, + Phi, + UnaryOp, + Value, +) +from coretrace_python.ir.ssa import SSAAnalysis + +State = Mapping[Value, AbstractValue] + +_SAFE_TYPES = (int, float, bool, str, bytes, type(None)) +_NUMERIC = frozenset({"int", "float", "bool"}) + +_BINARY: Mapping[str, Callable[..., Any]] = { + "add": operator.add, + "sub": operator.sub, + "mul": operator.mul, + "div": operator.truediv, + "floor_div": operator.floordiv, + "mod": operator.mod, + "bit_or": operator.or_, + "bit_xor": operator.xor, + "bit_and": operator.and_, +} +_COMPARE: Mapping[str, Callable[..., Any]] = { + "eq": operator.eq, + "not_eq": operator.ne, + "lt": operator.lt, + "lt_eq": operator.le, + "gt": operator.gt, + "gt_eq": operator.ge, + "is": operator.is_, + "is_not": operator.is_not, + "in": lambda a, b: a in b, + "not_in": lambda a, b: a not in b, +} +_UNARY: Mapping[str, Callable[..., Any]] = { + "neg": operator.neg, + "pos": operator.pos, + "invert": operator.invert, +} + + +class ConstantFacts: + def __init__(self, values: Mapping[Value, AbstractValue], reachable: frozenset[BlockId]): + self._values = MappingProxyType(dict(values)) + self._reachable = reachable + + def value(self, value: Value) -> AbstractValue: + return self._values.get(value, AbstractValue.bottom()) + + def reachable(self, block: BlockId) -> bool: + return block in self._reachable + + +class _ConstantProblem(DataflowProblem[State]): + direction: ClassVar[Direction] = Direction.FORWARD + + def __init__(self, function: FunctionIR) -> None: + self.function = function + self.blocks = {block.id: block for block in function.blocks} + + def initial(self) -> State: + return MappingProxyType({p: AbstractValue.unknown() for p in self.function.parameters}) + + def join(self, a: State, b: State) -> State: + merged = dict(a) + for value, fact in b.items(): + merged[value] = merged[value].join(fact) if value in merged else fact + return MappingProxyType(merged) + + def evaluate(self, block: BasicBlock, incoming: Mapping[BlockId, State]) -> State: + """State after ``block`` given the states on its executable incoming edges.""" + + states = list(incoming.values()) + state = dict(states[0]) + for other in states[1:]: + state = dict(self.join(state, other)) + for instruction in block.instructions: + if instruction.result is None: + continue + state[instruction.result] = ( + self.phi(instruction, incoming) + if isinstance(instruction, Phi) + else self.instruction(instruction, state) + ) + if isinstance(block.terminator, ForNext) and block.terminator.result is not None: + state[block.terminator.result] = AbstractValue.unknown() + return MappingProxyType(state) + + def flow(self, cfg: CFG, block_id: BlockId, incoming: Mapping[BlockId, State]) -> Mapping[BlockId, State]: + block = self.blocks[block_id] + state = self.evaluate(block, incoming) + terminator = block.terminator + if isinstance(terminator, Branch): + truth = state[terminator.condition].truthiness + targets = { + Truth.TRUE: (terminator.then_block,), + Truth.FALSE: (terminator.else_block,), + Truth.UNKNOWN: (terminator.then_block, terminator.else_block), + }[truth] + return {target: state for target in targets} + if isinstance(terminator, Jump): + return {terminator.target: state} + if isinstance(terminator, ForNext): + return {terminator.body: state, terminator.exit: state} + return {} + + # ------------------------------------------------------------------ transfer + + def phi(self, phi: Phi, incoming: Mapping[BlockId, State]) -> AbstractValue: + result = AbstractValue.bottom() + for value, predecessor in phi.incoming: + if predecessor in incoming: + result = result.join(incoming[predecessor].get(value, AbstractValue.unknown())) + return result + + def instruction(self, instruction: Instruction, state: Mapping[Value, AbstractValue]) -> AbstractValue: + if isinstance(instruction, Constant): + return AbstractValue.of(instruction.value) + if isinstance(instruction, BinaryOp): + return self.binary(instruction, state[instruction.left], state[instruction.right]) + if isinstance(instruction, Compare): + return self.fold(_COMPARE[instruction.operator], state[instruction.left], state[instruction.right]) + if isinstance(instruction, UnaryOp): + return self.unary(instruction, state[instruction.operand]) + return AbstractValue.unknown() + + def binary(self, op: BinaryOp, left: AbstractValue, right: AbstractValue) -> AbstractValue: + folder = _BINARY.get(op.operator) + if folder is not None: + numeric_only = op.operator == "mul" + folded = self.fold(folder, left, right, numeric_only=numeric_only) + if folded.constant is not TOP: + return folded + return AbstractValue.unknown(self.result_types(op.operator, left, right)) + + @staticmethod + def result_types(operator_name: str, left: AbstractValue, right: AbstractValue) -> frozenset[str] | None: + if left.types is None or right.types is None: + return None + if left.types <= _NUMERIC and right.types <= _NUMERIC: + if operator_name == "div" or "float" in left.types | right.types: + return frozenset({"float"}) + if operator_name in _BINARY: + return frozenset({"int"}) + if operator_name == "add" and left.types == right.types == frozenset({"str"}): + return frozenset({"str"}) + return None + + def unary(self, op: UnaryOp, operand: AbstractValue) -> AbstractValue: + if op.operator == "not": + if operand.truthiness is Truth.UNKNOWN: + return AbstractValue.unknown(frozenset({"bool"})) + return AbstractValue.of(operand.truthiness is Truth.FALSE) + folder = _UNARY.get(op.operator) + if folder is None or operand.constant is TOP or operand.constant is BOTTOM: + return AbstractValue.unknown() + if not isinstance(operand.constant, (int, float, bool)): + return AbstractValue.unknown() + try: + return AbstractValue.of(folder(operand.constant)) + except (TypeError, ValueError, ArithmeticError): + return AbstractValue.unknown() + + @staticmethod + def fold( + folder: Callable[..., Any], + left: AbstractValue, + right: AbstractValue, + *, + numeric_only: bool = False, + ) -> AbstractValue: + for side in (left.constant, right.constant): + if side is TOP or side is BOTTOM or not isinstance(side, _SAFE_TYPES): + return AbstractValue.unknown() + if numeric_only and not all(isinstance(c, (int, float, bool)) for c in (left.constant, right.constant)): + return AbstractValue.unknown() + try: + return AbstractValue.of(folder(left.constant, right.constant)) + except (TypeError, ValueError, ArithmeticError): + return AbstractValue.unknown() + + +def propagate_constants(function: FunctionIR, cfg: CFG) -> ConstantFacts: + problem = _ConstantProblem(function) + solution = solve(problem, cfg) + values: dict[Value, AbstractValue] = {} + reachable: set[BlockId] = set() + for block in function.blocks: + if solution.reached(block.id): + reachable.add(block.id) + values.update(problem.evaluate(block, solution.incoming(block.id))) + return ConstantFacts(values, frozenset(reachable)) + + +class ConstantPropagation(FunctionAnalysis[ConstantFacts]): + name: ClassVar[str] = "abstract.constants" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, CFGAnalysis}) + + @classmethod + def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> ConstantFacts: + return propagate_constants(ctx.get(SSAAnalysis, function), ctx.get(CFGAnalysis, function)) diff --git a/src/coretrace_python/abstract/values.py b/src/coretrace_python/abstract/values.py new file mode 100644 index 0000000..94ae0a3 --- /dev/null +++ b/src/coretrace_python/abstract/values.py @@ -0,0 +1,56 @@ +"""Abstract values (architecture §18). + +An ``AbstractValue`` records what is known about one SSA value: its constant on a flat +lattice, the set of Python types it may have (``None`` when unknown) and the truthiness +that follows. Taints, ranges, string constraints and nullability join this record as +their analyses arrive. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + +from coretrace_python.dataflow import BOTTOM, TOP, Element, FlatLattice + +_CONSTANTS: FlatLattice[object] = FlatLattice() + + +class Truth(Enum): + TRUE = "true" + FALSE = "false" + UNKNOWN = "unknown" + + +@dataclass(frozen=True) +class AbstractValue: + constant: Element[object] + types: frozenset[str] | None + + @classmethod + def of(cls, value: object) -> AbstractValue: + return cls(value, frozenset({type(value).__name__})) + + @classmethod + def unknown(cls, types: frozenset[str] | None = None) -> AbstractValue: + return cls(TOP, types) + + @classmethod + def bottom(cls) -> AbstractValue: + return cls(BOTTOM, frozenset()) + + @property + def truthiness(self) -> Truth: + if self.constant is TOP or self.constant is BOTTOM: + if self.types == frozenset({"NoneType"}): + return Truth.FALSE + return Truth.UNKNOWN + return Truth.TRUE if self.constant else Truth.FALSE + + def join(self, other: AbstractValue) -> AbstractValue: + if self.constant is BOTTOM: + return other + if other.constant is BOTTOM: + return self + types = None if self.types is None or other.types is None else self.types | other.types + return AbstractValue(_CONSTANTS.join(self.constant, other.constant), types) diff --git a/src/coretrace_python/dataflow/__init__.py b/src/coretrace_python/dataflow/__init__.py new file mode 100644 index 0000000..8141730 --- /dev/null +++ b/src/coretrace_python/dataflow/__init__.py @@ -0,0 +1,29 @@ +"""Generic data-flow framework: lattices and a bidirectional worklist solver.""" + +from coretrace_python.dataflow.lattice import ( + BOTTOM, + TOP, + Bottom, + Element, + FlatLattice, + Lattice, + Top, + same_element, +) +from coretrace_python.dataflow.solver import ENTRY, DataflowProblem, Direction, Solution, solve + +__all__ = [ + "BOTTOM", + "ENTRY", + "TOP", + "Bottom", + "DataflowProblem", + "Direction", + "Element", + "FlatLattice", + "Lattice", + "Solution", + "Top", + "same_element", + "solve", +] diff --git a/src/coretrace_python/dataflow/lattice.py b/src/coretrace_python/dataflow/lattice.py new file mode 100644 index 0000000..3d0f0c8 --- /dev/null +++ b/src/coretrace_python/dataflow/lattice.py @@ -0,0 +1,78 @@ +"""Lattice vocabulary shared by data-flow problems (architecture §37 dataflow/lattice). + +``BOTTOM`` means "no information yet" and ``TOP`` means "anything". A ``FlatLattice`` +places every other element between the two: equal elements join to themselves, different +ones to ``TOP``. Elements are compared by type as well as value, so ``1`` and ``True`` +are different constants. +""" + +from __future__ import annotations + +from typing import Final, Generic, Protocol, TypeAlias, TypeVar + + +class Bottom: + __slots__ = () + + def __repr__(self) -> str: + return "BOTTOM" + + +class Top: + __slots__ = () + + def __repr__(self) -> str: + return "TOP" + + +BOTTOM: Final = Bottom() +TOP: Final = Top() + +T = TypeVar("T") +Element: TypeAlias = T | Bottom | Top + + +def same_element(a: object, b: object) -> bool: + """Equality that also requires equal types, so ``1 == True`` does not conflate.""" + + return type(a) is type(b) and bool(a == b) + + +class Lattice(Protocol[T]): + @property + def bottom(self) -> Element[T]: ... + + @property + def top(self) -> Element[T]: ... + + def join(self, a: Element[T], b: Element[T]) -> Element[T]: ... + + def leq(self, a: Element[T], b: Element[T]) -> bool: ... + + +class FlatLattice(Generic[T]): + """Bottom < every element < Top, with no order between elements.""" + + @property + def bottom(self) -> Element[T]: + return BOTTOM + + @property + def top(self) -> Element[T]: + return TOP + + def join(self, a: Element[T], b: Element[T]) -> Element[T]: + if a is BOTTOM: + return b + if b is BOTTOM: + return a + if a is TOP or b is TOP: + return TOP + return a if same_element(a, b) else TOP + + def leq(self, a: Element[T], b: Element[T]) -> bool: + if a is BOTTOM or b is TOP: + return True + if a is TOP or b is BOTTOM: + return False + return same_element(a, b) diff --git a/src/coretrace_python/dataflow/solver.py b/src/coretrace_python/dataflow/solver.py new file mode 100644 index 0000000..048ff8a --- /dev/null +++ b/src/coretrace_python/dataflow/solver.py @@ -0,0 +1,96 @@ +"""Worklist solver over control-flow graphs (architecture §37 dataflow/worklist, solver). + +A problem sees, for each block, the states arriving on its executable incoming edges +(keyed by the block they come from, or ``ENTRY`` for the initial state) and returns the +states it sends along outgoing edges. Edges a problem does not return are pruned: the +blocks behind them stay unreached unless another edge reaches them. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import Mapping +from enum import Enum +from types import MappingProxyType +from typing import ClassVar, Generic, TypeVar + +from coretrace_python.cfg import CFG, BlockId + +S = TypeVar("S") + +ENTRY = BlockId("") + + +class Direction(Enum): + FORWARD = "forward" + BACKWARD = "backward" + + +class DataflowProblem(ABC, Generic[S]): + direction: ClassVar[Direction] = Direction.FORWARD + + @abstractmethod + def initial(self) -> S: + """State entering the entry block (forward) or leaving the exits (backward).""" + + @abstractmethod + def join(self, a: S, b: S) -> S: + """Combine the states of two incoming edges.""" + + @abstractmethod + def flow(self, cfg: CFG, block: BlockId, incoming: Mapping[BlockId, S]) -> Mapping[BlockId, S]: + """States sent to the next blocks in the flow direction; omitted edges are pruned.""" + + +class Solution(Generic[S]): + def __init__(self, problem: DataflowProblem[S], edges: Mapping[BlockId, Mapping[BlockId, S]]): + self._problem = problem + self._incoming: Mapping[BlockId, Mapping[BlockId, S]] = MappingProxyType( + {block: MappingProxyType(dict(found)) for block, found in edges.items()} + ) + + def incoming(self, block: BlockId) -> Mapping[BlockId, S]: + return self._incoming.get(block, MappingProxyType({})) + + def reached(self, block: BlockId) -> bool: + return bool(self._incoming.get(block)) + + def state(self, block: BlockId) -> S: + """Join of every state arriving at ``block``; raises for an unreached block.""" + + states = list(self.incoming(block).values()) + if not states: + raise KeyError(block) + result = states[0] + for other in states[1:]: + result = self._problem.join(result, other) + return result + + def edge(self, source: BlockId, target: BlockId) -> S: + return self._incoming[target][source] + + +def solve(problem: DataflowProblem[S], cfg: CFG) -> Solution[S]: + forward = problem.direction is Direction.FORWARD + order = {block: position for position, block in enumerate(cfg.blocks)} + if forward: + starts = [cfg.entry] + else: + starts = [b for b in cfg.blocks if b in cfg.reachable() and not cfg.successors(b)] + + incoming: dict[BlockId, dict[BlockId, S]] = {start: {ENTRY: problem.initial()} for start in starts} + worklist = list(starts) + queued = set(starts) + while worklist: + worklist.sort(key=lambda block: order[block], reverse=True) + block = worklist.pop() + queued.discard(block) + for target, state in problem.flow(cfg, block, MappingProxyType(incoming[block])).items(): + edges = incoming.setdefault(target, {}) + if block in edges and edges[block] == state: + continue + edges[block] = state + if target not in queued: + worklist.append(target) + queued.add(target) + return Solution(problem, incoming)