diff --git a/angr/analyses/decompiler/clinic.py b/angr/analyses/decompiler/clinic.py index 312fd059a..2abb758a4 100644 --- a/angr/analyses/decompiler/clinic.py +++ b/angr/analyses/decompiler/clinic.py @@ -16,7 +16,20 @@ import networkx from angr import ailment from angr.ailment import AILBlockRewriter, Assignment, Block, Statement from angr.ailment.block_walker import AILBlockViewer -from angr.ailment.expression import Array, Call, FunctionLikeMacro, Let, RustEnum, Struct, VirtualVariable +from angr.ailment.expression import ( + Array, + Call, + Expression, + FunctionLikeMacro, + Let, + RustEnum, + Struct, + Tmp, + VirtualVariable, +) +from angr.ailment.expression import ( + Register as AILRegister, +) from angr.analyses.analysis import Analysis, register_analysis from angr.analyses.cfg.cfg_base import CFGBase from angr.analyses.decompiler.block_simplifier import BlockSimplifier, PeepholeOptimizationBundle @@ -28,6 +41,7 @@ from angr.analyses.s_reaching_definitions.s_rda_model import SRDAModel from angr.analyses.stack_pointer_tracker import OffsetVal, Register from angr.analyses.typehoon import Typehoon from angr.analyses.typehoon.simple_solver import SimpleSolver +from angr.block import Block as VEXBlock from angr.calling_conventions import ( SimCCUsercall, SimComboArg, @@ -244,6 +258,77 @@ class _VLABufferBinder(AILBlockViewer): return super()._handle_VirtualVariable(expr_idx, expr, stmt_idx, stmt, block) +class _ITStateDependencyWalker(AILBlockViewer): + """Follow local temporary definitions to find a function-entry ITSTATE read.""" + + def __init__(self, tmp_definitions: dict[int, Expression], itstate_offset: int, entry_addr: int): + super().__init__() + self._tmp_definitions = tmp_definitions + self._itstate_offset = itstate_offset + self._entry_addr = entry_addr + self._visiting_tmp_indices: set[int] = set() + self.depends_on_itstate = False + + def _handle_Register( + self, expr_idx: int, expr: AILRegister, stmt_idx: int, stmt: Statement | None, block: Block | None + ) -> None: + if expr.reg_offset == self._itstate_offset and expr.tags.get("ins_addr") == self._entry_addr: + self.depends_on_itstate = True + + def _handle_Tmp(self, expr_idx: int, expr: Tmp, stmt_idx: int, stmt: Statement | None, block: Block | None) -> None: + if self.depends_on_itstate or expr.tmp_idx in self._visiting_tmp_indices: + return + definition = self._tmp_definitions.get(expr.tmp_idx) + if definition is None: + return + + self._visiting_tmp_indices.add(expr.tmp_idx) + self._handle_expr(0, definition, stmt_idx, stmt, block) + self._visiting_tmp_indices.remove(expr.tmp_idx) + + +def _remove_thumb_entry_itstate_guard( + block: VEXBlock, statements: list[Statement], itstate_offset: int +) -> list[Statement]: + """Remove LibVEX's synthetic guard around the first Thumb instruction, if present.""" + + instruction_addrs = block.instruction_addrs + if not instruction_addrs or instruction_addrs[0] != block.addr: + return statements + + next_instruction_addr = instruction_addrs[1] if len(instruction_addrs) > 1 else block.addr + block.size + tmp_definitions = {} + new_statements = [] + for stmt in statements: + if ( + isinstance(stmt, ailment.Stmt.Assignment) + and isinstance(stmt.dst, ailment.Expr.Tmp) + and stmt.tags.get("ins_addr") == block.addr + ): + tmp_definitions[stmt.dst.tmp_idx] = stmt.src + + if ( + isinstance(stmt, ailment.Stmt.ConditionalJump) + and stmt.tags.get("ins_addr") == block.addr + and isinstance(stmt.true_target, ailment.Expr.Const) + and stmt.true_target.value == next_instruction_addr + and ( + stmt.false_target is None + or ( + isinstance(stmt.false_target, ailment.Expr.Const) + and stmt.false_target.value == next_instruction_addr + ) + ) + ): + dependency_walker = _ITStateDependencyWalker(tmp_definitions, itstate_offset, block.addr) + dependency_walker.walk_expression(stmt.condition, stmt=stmt, block=None) + if dependency_walker.depends_on_itstate: + continue + + new_statements.append(stmt) + return new_statements + + class Clinic(Analysis, Serializable): """ A Clinic deals with AILments: it lifts a function to AIL and runs the decompiler's simplification pipeline on it. @@ -1519,6 +1604,25 @@ class Clinic(Analysis, Serializable): self._ail_manager.next_atom(), dflag, forward, ins_addr=block.addr ) converted.statements.insert(0, dflag_assignment) + elif block.addr == self.function.addr and block.thumb is True and "itstate" in self.project.arch.registers: + itstate_offset, itstate_size = self.project.arch.registers["itstate"] + # LibVEX may infer stale ITSTATE from bytes before a context-free Thumb lift. Remove only the entry guard + # that skips the first instruction; genuine IT blocks later in the function remain untouched. + converted.statements = _remove_thumb_entry_itstate_guard(block, converted.statements, itstate_offset) + + itstate = ailment.Expr.Register( + self._ail_manager.next_atom(), + itstate_offset, + itstate_size * self.project.arch.byte_width, + ins_addr=block.addr, + ) + cleared = ailment.Expr.Const( + self._ail_manager.next_atom(), 0, itstate_size * self.project.arch.byte_width, ins_addr=block.addr + ) + itstate_assignment = ailment.Stmt.Assignment( + self._ail_manager.next_atom(), itstate, cleared, ins_addr=block.addr + ) + converted.statements.insert(0, itstate_assignment) return converted diff --git a/tests/analyses/decompiler/test_clinic_entry_state.py b/tests/analyses/decompiler/test_clinic_entry_state.py new file mode 100644 index 000000000..e54a6f4e4 --- /dev/null +++ b/tests/analyses/decompiler/test_clinic_entry_state.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +# pylint:disable=missing-class-docstring,protected-access +from __future__ import annotations + +import unittest +from collections.abc import Callable, Sequence +from types import SimpleNamespace +from typing import Any, cast +from unittest import mock + +import archinfo +from archinfo import Arch + +from angr import ailment +from angr.analyses.decompiler.clinic import Clinic +from angr.codenode import BlockNode + + +class TestClinicEntryState(unittest.TestCase): + @staticmethod + def _assignment( + manager: ailment.Manager, reg_offset: int, bits: int, value: int, *, ins_addr: int | None = None + ) -> ailment.Stmt.Assignment: + return ailment.Stmt.Assignment( + manager.next_atom(), + ailment.Expr.Register(manager.next_atom(), reg_offset, bits, ins_addr=ins_addr), + ailment.Expr.Const(manager.next_atom(), value, bits, ins_addr=ins_addr), + ins_addr=ins_addr, + ) + + @staticmethod + def _guarded_statements( + manager: ailment.Manager, + arch: Arch, + block_addr: int, + condition_reg_offset: int, + true_target: int, + false_target: int | None = None, + ) -> list[ailment.Stmt.Statement]: + source_tmp = ailment.Expr.Tmp(manager.next_atom(), 0, arch.bits, ins_addr=block_addr) + condition_tmp = ailment.Expr.Tmp(manager.next_atom(), 1, arch.bits, ins_addr=block_addr) + return [ + ailment.Stmt.Assignment( + manager.next_atom(), + source_tmp, + ailment.Expr.Register(manager.next_atom(), condition_reg_offset, arch.bits, ins_addr=block_addr), + ins_addr=block_addr, + ), + ailment.Stmt.Assignment(manager.next_atom(), condition_tmp, source_tmp, ins_addr=block_addr), + ailment.Stmt.ConditionalJump( + manager.next_atom(), + condition_tmp, + ailment.Expr.Const(manager.next_atom(), true_target, arch.bits, ins_addr=block_addr), + ( + ailment.Expr.Const(manager.next_atom(), false_target, arch.bits, ins_addr=block_addr) + if false_target is not None + else None + ), + ins_addr=block_addr, + ), + ] + + @staticmethod + def _convert( + function_addr: int, + block_addr: int, + thumb: bool, + original_statements: Callable[[ailment.Manager, Arch], Sequence[ailment.Stmt.Statement]], + *, + block_size: int = 8, + instruction_addrs: list[int] | None = None, + ) -> tuple[Arch, ailment.Block, list[ailment.Stmt.Statement]]: + arch = archinfo.ArchARMCortexM() + manager = ailment.Manager(arch=arch) + block = SimpleNamespace( + addr=block_addr, + size=block_size, + thumb=thumb, + instruction_addrs=instruction_addrs if instruction_addrs is not None else [block_addr, block_addr + 4], + ) + clinic = cast(Any, Clinic.__new__(Clinic)) + clinic.project = SimpleNamespace( + arch=arch, + factory=SimpleNamespace(block=mock.Mock(return_value=block)), + ) + clinic.function = SimpleNamespace(addr=function_addr) + clinic._ail_manager = manager + lifted_statements = list(original_statements(manager, arch)) + converted = ailment.Block(block_addr, block.size, statements=list(lifted_statements)) + + with mock.patch.object(Clinic, "_convert_vex", return_value=converted): + result = clinic._convert(BlockNode(block_addr, block.size)) + return arch, result, lifted_statements + + def test_thumb_entry_clears_itstate_before_lifted_statements(self): + def original(manager, arch): + return [self._assignment(manager, arch.registers["r0"][0], arch.bits, 1)] + + arch, result, original_statements = self._convert(0x1001, 0x1001, True, original) + + self.assertEqual(len(result.statements), 2) + assignment = result.statements[0] + assert isinstance(assignment, ailment.Stmt.Assignment) + destination = cast(Any, assignment.dst) + source = cast(Any, assignment.src) + self.assertIsInstance(destination, ailment.Expr.Register) + self.assertEqual(destination.reg_offset, arch.registers["itstate"][0]) + self.assertIsInstance(source, ailment.Expr.Const) + self.assertEqual(source.value, 0) + self.assertIs(result.statements[1], original_statements[0]) + + def test_thumb_entry_removes_only_itstate_guard_skipping_first_instruction(self): + def original(manager, arch): + statements = self._guarded_statements( + manager, + arch, + 0x1001, + arch.registers["itstate"][0], + 0x1005, + ) + statements.append(self._assignment(manager, arch.registers["r0"][0], arch.bits, 1, ins_addr=0x1001)) + return statements + + _, result, original_statements = self._convert(0x1001, 0x1001, True, original) + + self.assertNotIn(original_statements[2], result.statements) + self.assertIn(original_statements[3], result.statements) + + def test_single_instruction_thumb_entry_removes_itstate_guard_to_block_end(self): + def original(manager, arch): + return self._guarded_statements( + manager, + arch, + 0x1001, + arch.registers["itstate"][0], + 0x1005, + false_target=0x1005, + ) + + _, result, original_statements = self._convert( + 0x1001, + 0x1001, + True, + original, + block_size=4, + instruction_addrs=[0x1001], + ) + + self.assertNotIn(original_statements[2], result.statements) + + def test_thumb_entry_keeps_guard_not_depending_on_itstate(self): + def original(manager, arch): + return self._guarded_statements(manager, arch, 0x1001, arch.registers["r0"][0], 0x1005) + + _, result, original_statements = self._convert(0x1001, 0x1001, True, original) + + self.assertIn(original_statements[2], result.statements) + + def test_thumb_entry_keeps_itstate_guard_not_skipping_to_next_instruction(self): + def original(manager, arch): + return self._guarded_statements(manager, arch, 0x1001, arch.registers["itstate"][0], 0x1011) + + _, result, original_statements = self._convert(0x1001, 0x1001, True, original) + + self.assertIn(original_statements[2], result.statements) + + def test_thumb_entry_keeps_itstate_conditional_jump_with_two_targets(self): + def original(manager, arch): + return self._guarded_statements( + manager, + arch, + 0x1001, + arch.registers["itstate"][0], + 0x1005, + false_target=0x1011, + ) + + _, result, original_statements = self._convert(0x1001, 0x1001, True, original) + + self.assertIn(original_statements[2], result.statements) + + def test_thumb_interior_block_keeps_lifted_itstate_guard(self): + def original(manager, arch): + return self._guarded_statements(manager, arch, 0x1011, arch.registers["itstate"][0], 0x1015) + + _, result, original_statements = self._convert(0x1001, 0x1011, True, original) + + self.assertEqual(result.statements, original_statements) + + def test_arm_entry_does_not_seed_thumb_itstate(self): + def original(manager, arch): + return [self._assignment(manager, arch.registers["r0"][0], arch.bits, 1)] + + _, result, original_statements = self._convert(0x1000, 0x1000, False, original) + + self.assertEqual(result.statements, original_statements) + + +if __name__ == "__main__": + unittest.main()