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

ssa: add check_ail_graph(), and regression-test it on known-bad functions

Browse source 
The "is this graph legal" logic existed only as one-off snippets in throwaway
scripts while these repairs were being written. Promote it to
angr/utils/ssa/validate.py: check_ail_graph() returns a list of GraphProblem, one
per violation, each carrying the block, statement index and enough detail to go
straight into an assertion message.

Six checks: a vvar assigned in more than one place, a phi operand naming a block
that is not in the graph, one naming a live block that is not a predecessor, a
predecessor no operand mentions, two blocks sharing an (addr, idx), and a graph
without exactly one entry block. Every one of them is invisible where it happens
and only surfaces later as an unrelated failure, which is why they are worth
asserting on directly.

Tests come in two layers. The checker itself is exercised on graphs built to
break exactly one rule each, including that narrowing the check set really does
suppress the others. Then nine real functions from mv_0 and grep_gcc17.0.0_O2 --
found by validating every function of those binaries with the repairs disabled --
are decompiled and validated. With the repairs no-op'd all nine fail, so they
test what they claim to; the scoped CFG the tests use does reproduce the original
breakage. One scoped CFG per binary rather than per function keeps that under 20s.

The check set excludes the "predecessor with no operand" class for now: it is a
real defect but no pass repairs it yet, and leaving it in would mask everything
else.
This commit is contained in:
Fish Wang 2026-08-14 06:17:34 +00:00
parent 728d77dafb
commit 5ed9a98767
2 changed files with 372 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

151
angr/utils/ssa/validate.py Normal file
View file

@ -0,0 +1,151 @@
"""Structural checks an AIL graph in (partial) SSA form has to satisfy.
Every problem reported here is one that survives quietly and surfaces much later
as an unrelated failure: a vvar with two definitions makes ``SReachingDefinitions``
and ``GraphDephicationVVarMapping`` record only the last one (both key definitions
by variable id in a plain dict), and a phi naming a block that is no longer a
predecessor sends de-phi looking for a definition along an edge that does not
exist. Passes that add, copy or delete blocks are what break these, so checking a
graph after a pass has run is the cheapest way to catch the damage where it
happened.
"""
from __future__ import annotations
from collections import defaultdict
from dataclasses import dataclass
from typing import TYPE_CHECKING
from angr.ailment.expression import VirtualVariable
from angr.ailment.statement import Assignment
from . import is_phi_assignment
if TYPE_CHECKING:
from angr.ailment import Block
Address = tuple[int, "int | None"]
#: two blocks share one ``(addr, idx)``; consumers key blocks by that pair in a
#: plain dict, so one silently shadows the other
DUPLICATE_BLOCK = "duplicate-block"
#: the graph does not have exactly one entry block at the function's address
BAD_ENTRY = "bad-entry"
#: a virtual variable is assigned in more than one place
VVAR_REDEFINED = "vvar-redefined"
#: a phi operand names a block that is not in the graph at all
PHI_SOURCE_REMOVED = "phi-source-removed"
#: a phi operand names a live block that is not a predecessor of the phi's block
PHI_SOURCE_NOT_PREDECESSOR = "phi-source-not-predecessor"
#: a predecessor that no operand of the phi mentions
PHI_MISSING_PREDECESSOR = "phi-missing-predecessor"
ALL_CHECKS = frozenset(
{
DUPLICATE_BLOCK,
BAD_ENTRY,
VVAR_REDEFINED,
PHI_SOURCE_REMOVED,
PHI_SOURCE_NOT_PREDECESSOR,
PHI_MISSING_PREDECESSOR,
}
)
@dataclass(frozen=True)
class GraphProblem:
"""One violation, identified well enough to go straight into an assertion message."""
kind: str
block: Address
detail: str
stmt_idx: int | None = None
def __str__(self):
where = f"{self.block[0]:#x}.{self.block[1]}"
if self.stmt_idx is not None:
where += f"[{self.stmt_idx}]"
return f"{self.kind} at {where}: {self.detail}"
def check_ail_graph(graph, func_addr: int | None = None, checks=ALL_CHECKS) -> list[GraphProblem]:
"""Report every structural problem in ``graph``.
:param graph: The AIL graph to check.
:param func_addr: Function address, for the entry-block check. Omit to skip it.
:param checks: Which checks to run; see the module constants. Narrow this when a
known-unfixed class would otherwise drown out what you are testing.
:return: The problems found, ordered by block and statement.
"""
problems: list[GraphProblem] = []
locations: dict[Address, Block] = {}
for block in graph:
loc = (block.addr, block.idx)
if loc in locations and DUPLICATE_BLOCK in checks:
problems.append(GraphProblem(DUPLICATE_BLOCK, loc, "two blocks share this location"))
locations[loc] = block
if func_addr is not None and BAD_ENTRY in checks:
entries = [b for b in graph if b.addr == func_addr and b.idx is None]
if len(entries) != 1:
problems.append(
GraphProblem(BAD_ENTRY, (func_addr, None), f"expected exactly one entry block, found {len(entries)}")
)
definitions: dict[int, list[tuple[Address, int]]] = defaultdict(list)
for block in graph:
loc = (block.addr, block.idx)
predecessors = {(p.addr, p.idx) for p in graph.predecessors(block)}
for stmt_idx, stmt in enumerate(block.statements):
if isinstance(stmt, Assignment) and isinstance(stmt.dst, VirtualVariable):
definitions[stmt.dst.varid].append((loc, stmt_idx))
if not is_phi_assignment(stmt):
continue
named = set()
for src, _ in stmt.src.src_and_vvars:
named.add(src)
if src not in locations:
if PHI_SOURCE_REMOVED in checks:
problems.append(
GraphProblem(
PHI_SOURCE_REMOVED,
loc,
f"operand names {src[0]:#x}.{src[1]}, which is not in the graph",
stmt_idx,
)
)
elif src not in predecessors and PHI_SOURCE_NOT_PREDECESSOR in checks:
problems.append(
GraphProblem(
PHI_SOURCE_NOT_PREDECESSOR,
loc,
f"operand names {src[0]:#x}.{src[1]}, which is not a predecessor",
stmt_idx,
)
)
if PHI_MISSING_PREDECESSOR in checks:
for missing in sorted(predecessors - named):
problems.append(
GraphProblem(
PHI_MISSING_PREDECESSOR,
loc,
f"predecessor {missing[0]:#x}.{missing[1]} has no operand",
stmt_idx,
)
)
if VVAR_REDEFINED in checks:
for varid, sites in sorted(definitions.items()):
if len(sites) > 1:
where = ", ".join(f"{loc[0]:#x}.{loc[1]}[{i}]" for loc, i in sites[:4])
problems.append(
GraphProblem(VVAR_REDEFINED, sites[0][0], f"vvar {varid} is defined {len(sites)} times: {where}")
)
problems.sort(key=lambda p: (p.block[0], -1 if p.block[1] is None else p.block[1], p.stmt_idx or -1, p.kind))
return problems

221
tests/analyses/decompiler/test_ail_graph_validity.py Normal file
View file

@ -0,0 +1,221 @@
#!/usr/bin/env python3
# pylint: disable=missing-class-docstring,no-self-use
from __future__ import annotations
__package__ = __package__ or "tests.analyses.decompiler" # pylint:disable=redefined-builtin
import os.path
import unittest
import networkx
from angr.ailment import Block
from angr.ailment.expression import Const, Phi, VirtualVariable, VirtualVariableCategory
from angr.ailment.manager import Manager
from angr.ailment.statement import Assignment, Jump
from angr.utils.ssa.validate import (
ALL_CHECKS,
BAD_ENTRY,
DUPLICATE_BLOCK,
PHI_MISSING_PREDECESSOR,
PHI_SOURCE_NOT_PREDECESSOR,
PHI_SOURCE_REMOVED,
VVAR_REDEFINED,
check_ail_graph,
)
from tests.common import bin_location, load_project_with_scoped_cfg
test_location = os.path.join(bin_location, "tests")
# A phi that does not mention one of its predecessors is a real defect, but no pass
# repairs it yet and the right value to supply is an open question, so it would mask
# everything else here. Everything the repairs do cover is checked.
REPAIRED_CHECKS = ALL_CHECKS - {PHI_MISSING_PREDECESSOR}
# Functions whose decompiled graph was malformed before the phi-source and
# duplicate-definition repairs landed. Found by validating every function of these
# binaries with the repairs disabled; kept as the regression corpus because the
# breakage only shows up on real switch lowering, dead-block removal and return
# duplication, which is awkward to fake.
KNOWN_BAD = [
("mv_0", "rpl_fchmodat", 0x40BB60),
("mv_0", "fts_build", 0x40D200),
("mv_0", "prompt", 0x403E80),
("mv_0", "areadlinkat_with_size", 0x40AC20),
("grep_gcc17.0.0_O2", "rpl_fopen", 0x48ABC0),
("grep_gcc17.0.0_O2", "grepbuf", 0x4115A0),
("grep_gcc17.0.0_O2", "excluded_file_name", 0x4482B0),
("grep_gcc17.0.0_O2", "mbscasecmp", 0x463180),
("grep_gcc17.0.0_O2", "xstrtoimax", 0x4888A0),
]
def _vvar(manager, varid, bits=32):
return VirtualVariable(manager.next_atom(), varid, bits, VirtualVariableCategory.REGISTER, oident=16)
class TestCheckAILGraph(unittest.TestCase):
"""The checker itself, on graphs built to break exactly one rule each."""
@staticmethod
def _two_arm_graph(manager):
head = Block(0x1000, 1, statements=[Jump(manager.next_atom(), Const(manager.next_atom(), 0x2000, 64))])
left = Block(0x2000, 1, statements=[Jump(manager.next_atom(), Const(manager.next_atom(), 0x4000, 64))])
right = Block(0x3000, 1, statements=[Jump(manager.next_atom(), Const(manager.next_atom(), 0x4000, 64))])
join = Block(0x4000, 1, statements=[])
graph = networkx.DiGraph()
graph.add_edge(head, left)
graph.add_edge(head, right)
graph.add_edge(left, join)
graph.add_edge(right, join)
return graph, head, left, right, join
def test_clean_graph_reports_nothing(self):
manager = Manager(arch=None)
graph, head, left, right, join = self._two_arm_graph(manager)
join.statements = [
Assignment(
manager.next_atom(),
_vvar(manager, 100),
Phi(
manager.next_atom(),
32,
[((left.addr, left.idx), _vvar(manager, 10)), ((right.addr, right.idx), _vvar(manager, 11))],
),
)
]
assert check_ail_graph(graph, head.addr) == []
def test_detects_a_redefined_vvar(self):
manager = Manager(arch=None)
graph, head, left, right, _join = self._two_arm_graph(manager)
for block in (left, right):
block.statements.insert(
0, Assignment(manager.next_atom(), _vvar(manager, 100), Const(manager.next_atom(), 1, 32))
)
kinds = {p.kind for p in check_ail_graph(graph, head.addr)}
assert VVAR_REDEFINED in kinds
def test_detects_a_phi_naming_a_removed_block(self):
manager = Manager(arch=None)
graph, head, _left, _right, join = self._two_arm_graph(manager)
join.statements = [
Assignment(
manager.next_atom(),
_vvar(manager, 100),
Phi(manager.next_atom(), 32, [((0xDEAD, None), _vvar(manager, 10))]),
)
]
problems = check_ail_graph(graph, head.addr, checks={PHI_SOURCE_REMOVED})
assert [p.kind for p in problems] == [PHI_SOURCE_REMOVED]
assert "0xdead" in str(problems[0])
def test_detects_a_phi_naming_a_non_predecessor(self):
manager = Manager(arch=None)
graph, head, _left, _right, join = self._two_arm_graph(manager)
join.statements = [
Assignment(
manager.next_atom(),
_vvar(manager, 100),
# head is in the graph but does not flow straight into the join
Phi(manager.next_atom(), 32, [((head.addr, head.idx), _vvar(manager, 10))]),
)
]
problems = check_ail_graph(graph, head.addr, checks={PHI_SOURCE_NOT_PREDECESSOR})
assert [p.kind for p in problems] == [PHI_SOURCE_NOT_PREDECESSOR]
def test_detects_a_predecessor_with_no_operand(self):
manager = Manager(arch=None)
graph, head, left, _right, join = self._two_arm_graph(manager)
join.statements = [
Assignment(
manager.next_atom(),
_vvar(manager, 100),
Phi(manager.next_atom(), 32, [((left.addr, left.idx), _vvar(manager, 10))]),
)
]
problems = check_ail_graph(graph, head.addr, checks={PHI_MISSING_PREDECESSOR})
assert [p.kind for p in problems] == [PHI_MISSING_PREDECESSOR]
def test_detects_duplicate_block_locations(self):
manager = Manager(arch=None)
graph, head, _left, _right, _join = self._two_arm_graph(manager)
graph.add_edge(head, Block(0x2000, 1, statements=[]))
assert DUPLICATE_BLOCK in {p.kind for p in check_ail_graph(graph, head.addr)}
def test_detects_a_missing_entry_block(self):
manager = Manager(arch=None)
graph, head, _left, _right, _join = self._two_arm_graph(manager)
assert BAD_ENTRY in {p.kind for p in check_ail_graph(graph, 0x9999)}
assert BAD_ENTRY not in {p.kind for p in check_ail_graph(graph, head.addr)}
def test_checks_can_be_narrowed(self):
manager = Manager(arch=None)
graph, head, _left, _right, join = self._two_arm_graph(manager)
join.statements = [
Assignment(
manager.next_atom(),
_vvar(manager, 100),
Phi(manager.next_atom(), 32, [((0xDEAD, None), _vvar(manager, 10))]),
)
]
assert check_ail_graph(graph, head.addr, checks={VVAR_REDEFINED}) == []
class TestDecompiledGraphsAreValid(unittest.TestCase):
"""Real functions that used to come out of the decompiler malformed."""
@classmethod
def setUpClass(cls):
# one scoped CFG per binary covering all of its targets: building nine separate
# ones costs about ten times as much as building two
cls.projects = {}
by_binary: dict[str, list[int]] = {}
for binary, _name, addr in KNOWN_BAD:
by_binary.setdefault(binary, []).append(addr)
for binary, addrs in by_binary.items():
cls.projects[binary] = load_project_with_scoped_cfg(
os.path.join(test_location, "x86_64", binary), addrs[0], extra_func_addrs=addrs[1:]
)
def test_known_bad_functions_are_now_valid(self):
for binary, name, addr in KNOWN_BAD:
with self.subTest(binary=binary, function=name):
proj, cfg = self.projects[binary]
func = cfg.functions[addr]
dec = proj.analyses.Decompiler(func, cfg=cfg.model)
assert dec.ail_graph is not None, f"{binary}:{addr:#x} produced no AIL graph"
problems = check_ail_graph(dec.ail_graph, func.addr, checks=REPAIRED_CHECKS)
assert not problems, f"{binary} {name}: " + "; ".join(str(p) for p in problems[:5])
def test_a_whole_binary_decompiles_to_valid_graphs(self):
"""A net over every function of a small binary, to catch a future pass that
starts producing malformed graphs somewhere the corpus above does not reach.
1after909 is clean today even with the repairs switched off."""
import angr # pylint:disable=import-outside-toplevel
proj = angr.Project(os.path.join(test_location, "x86_64", "1after909"), auto_load_libs=False)
cfg = proj.analyses.CFG(normalize=True)
proj.analyses.CompleteCallingConventions()
checked = 0
for func in cfg.functions.values():
if func.is_simprocedure or func.is_plt or func.is_alignment:
continue
try:
dec = proj.analyses.Decompiler(func, cfg=cfg.model)
except Exception: # pylint:disable=broad-except
continue
if dec.ail_graph is None:
continue
checked += 1
problems = check_ail_graph(dec.ail_graph, func.addr, checks=REPAIRED_CHECKS)
assert not problems, f"{func.name}: " + "; ".join(str(p) for p in problems[:5])
assert checked > 5, f"only {checked} functions decompiled; the net is not catching anything"
if __name__ == "__main__":
unittest.main()