denuvo-ref/angr
1
0
Fork 0
mirror of https://github.com/angr/angr synced 2026-08-17 12:23:11 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Decompiler: reset Thumb IT state at function entry

Browse source
This commit is contained in:
Yan 2026-08-15 09:47:18 +00:00
parent 503b1be066
commit 0b01658449
2 changed files with 305 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

106
angr/analyses/decompiler/clinic.py
View file

@ -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

200
tests/analyses/decompiler/test_clinic_entry_state.py Normal file
View file

@ -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()