Skip to content
30 changes: 28 additions & 2 deletions chb/ast/ASTProvenance.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,10 @@ def __init__(self) -> None:
self._reaching_definitions: Dict[int, List[int]] = {}
self._flag_reaching_definitions: Dict[int, List[int]] = {}
self._definitions_used: Dict[int, List[int]] = {}
# Liveness, keyed by instruction address:
# {"0xNNNN": {"live-in": [...], "live-out": [...]}}
self._flag_liveness: Dict[str, Dict[str, List[str]]] = {}
self._register_liveness: Dict[str, Dict[str, List[str]]] = {}

@property
def instruction_mapping(self) -> Mapping[int, List[int]]:
Expand All @@ -63,6 +67,24 @@ def flag_reaching_definitions(self) -> Mapping[int, List[int]]:
def definitions_used(self) -> Mapping[int, List[int]]:
return self._definitions_used

@property
def flag_liveness(self) -> Mapping[str, Dict[str, List[str]]]:
return self._flag_liveness

@flag_liveness.setter
def flag_liveness(
self, liveness: Dict[str, Dict[str, List[str]]]) -> None:
self._flag_liveness = liveness

@property
def register_liveness(self) -> Mapping[str, Dict[str, List[str]]]:
return self._register_liveness

@register_liveness.setter
def register_liveness(
self, liveness: Dict[str, Dict[str, List[str]]]) -> None:
self._register_liveness = liveness

def has_expression_mapping(self, exprid: int) -> bool:
return exprid in self.expression_mapping

Expand Down Expand Up @@ -105,14 +127,16 @@ def add_definitions_used(self, lvalid: int, instrids: List[int]) -> None:
if instrid not in self.definitions_used[lvalid]:
self._definitions_used[lvalid].append(instrid)

def serialize(self) -> Mapping[str, Mapping[int, Union[int, List[int]]]]:
result: Dict[str, Mapping[int, Union[int, List[int]]]] = {}
def serialize(self) -> Mapping[str, Any]:
result: Dict[str, Any] = {}
result["instruction-mapping"] = self.instruction_mapping
result["expression-mapping"] = self.expression_mapping
result["lval-mapping"] = self.lval_mapping
result["reaching-definitions"] = self.reaching_definitions
result["flag-reaching-definitions"] = self.flag_reaching_definitions
result["definitions-used"] = self.definitions_used
result["flag-liveness"] = self.flag_liveness
result["register-liveness"] = self.register_liveness
return result

def deserialize(self, d: Dict[str, Any]) -> None:
Expand All @@ -128,3 +152,5 @@ def deserialize(self, d: Dict[str, Any]) -> None:
int(i): v for (i, v) in d["flag-reaching-definitions"].items()}
self._definitions_used = {
int(i): v for (i, v) in d["definitions-used"].items()}
self._flag_liveness = d.get("flag-liveness", {})
self._register_liveness = d.get("register-liveness", {})
236 changes: 236 additions & 0 deletions chb/astinterface/ASTILiveness.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,236 @@
# ------------------------------------------------------------------------------
# CodeHawk Binary Analyzer
# Author: Dan Phung
# ------------------------------------------------------------------------------
# The MIT License (MIT)
#
# Copyright (c) 2024-2025 Aarno Labs LLC
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# ------------------------------------------------------------------------------
"""Address-keyed liveness derived from per-instruction reaching-def facts.

The reaching-definition facts that CodeHawk attaches to each instruction record,
at each USE site, the addresses that DEFINE the value used there (for GP
registers via the reaching-def facts, for NZCV flags via the flag-reaching-def
facts). From those facts this class builds per-address use/kill sets and runs a
standard backward live-variable fixpoint over the function CFG, producing
live-in/live-out sets keyed by instruction address.

This is intentionally sound-by-over-approximation: every use recorded in the
facts is honored (never dropped), while a def that reaches no use may be absent
from the kill set, which can only make a variable appear live longer -- never
shorter. Consumers that use liveness to gate a transformation therefore never
get a false "dead".
"""

from collections import defaultdict

from typing import (
Callable, Dict, List, Mapping, Optional, Sequence, Set, TYPE_CHECKING,
Tuple, Union)

from chb.util.loggingutil import chklogger

if TYPE_CHECKING:
from chb.app.Function import Function
from chb.app.Instruction import Instruction
from chb.invariants.VarInvariantFact import (
FlagReachingDefFact, ReachingDefFact)


class ASTILiveness:
"""Derives address-keyed liveness for one CodeHawk function."""

def __init__(self, fn: "Function") -> None:
self._fn = fn

@property
def fn(self) -> "Function":
return self._fn

def flag_liveness(self) -> Dict[str, Dict[str, List[str]]]:
"""NZCV flag live-in/live-out per instruction address."""
(use, kill) = self._use_kill(
lambda instr: instr.xdata.flag_reachingdefs, warn_on_init=True)
return self._liveness(use, kill)

def register_liveness(
self, registers: Set[str]) -> Dict[str, Dict[str, List[str]]]:
"""GP-register live-in/live-out per instruction address.

registers is the set of architecture register names (e.g. the keys of
the register-size map), used to exclude spill slots and stack
temporaries.
"""
(use, kill) = self._use_kill(
lambda instr: instr.xdata.reachingdefs, set(registers))
return self._liveness(use, kill)

def _use_kill(
self,
get_facts: Callable[
["Instruction"],
Sequence[Optional[Union["FlagReachingDefFact",
"ReachingDefFact"]]]],
names: Optional[Set[str]] = None,
warn_on_init: bool = False
) -> Tuple[Dict[str, Set[str]], Dict[str, Set[str]]]:
"""Build per-address use and kill (def) sets from reaching-def facts.

get_facts is called once per instruction and returns that instruction's
reaching-def facts: instr.xdata.flag_reachingdefs for flags,
instr.xdata.reachingdefs for registers. Each fact names the variable
used at that instruction and the addresses that defined the value used
there, so the fact contributes a use at the instruction and a kill at
each of those def addresses.

When names is given, only variables in that set are considered. "PC" is
always excluded regardless of names. It is not a value a consumer can
treat as live or dead.

warn_on_init reports uses whose value is defined on function entry
rather than by an instruction.
"""

def is_real_def_site(defloc: str) -> bool:
"""True if defloc is an instruction address that can carry a kill.

Only "init" is excluded since it is the analysis's marker for a
value defined on function entry rather than by an instruction, so
there is no instruction there to do the killing.
"""

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am somewhat confused about this comment. It seems incorrect, but raises an interesting issue, at least for flag definitions. Flags are not preserved across function calls. I realize this is an omission also on the ocaml side, where flag definitions are currently not clobbered by calls, like registers R0-R3 are; I will fix that. However, if you encounter a def-site prefixed with F, then that location does kill, because the execution sequence determines what flags are set. If indeed a flag def-site is in the inlined (presumably a payload) function, then that is what the downstream code sees. For the same reason, an "init" def-site for a flag definition is suspicious if that is an actual reaching definition for a user; it should probably generate a warning.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, sorry that was my misunderstanding of the ABI and the Flag preservation. I've now removed that and added a warning.

return defloc != "init"

use: Dict[str, Set[str]] = defaultdict(set)
kill: Dict[str, Set[str]] = defaultdict(set)
warned_init: Set[Tuple[str, str]] = set()
for (iaddr, instr) in self.fn.instructions.items():
for fact in get_facts(instr):
if fact is None:
continue
name = str(fact.variable)
if name == "PC":
continue
if names is not None and name not in names:
continue
use[iaddr].add(name)
for d in fact.deflocations:
da = str(d)
if not is_real_def_site(da):
if warn_on_init and (iaddr, name) not in warned_init:
warned_init.add((iaddr, name))
chklogger.logger.info(
"flag %s used at %s in function %s is reached by "
"an '%s' definition: its value predates function "
"entry.",
name, iaddr, self.fn.faddr, da)
continue
kill[da].add(name)
return (use, kill)

def _blocks(self) -> Dict[str, List[str]]:
"""Map block address to its instruction addresses in execution order.

Sorted lexicographically, the same ordering BasicBlock.lastaddr uses.
Sorting the addresses as numbers instead would raise an exception
because not every address is plain hex: the analysis writes an inlined
instruction's address as "F:0x...._0x....".
"""
result: Dict[str, List[str]] = {}
for (baddr, block) in self.fn.blocks.items():
result[baddr] = sorted(block.instructions.keys())
return result

def _liveness(
self,
use: Dict[str, Set[str]],
kill: Dict[str, Set[str]]) -> Dict[str, Dict[str, List[str]]]:
blocks = self._blocks()
# Read successors through the cfg.edges property, not cfg.successors,
# which reads the backing map directly and returns nothing until the
# property has lazily loaded it from XML.
edges = self.fn.cfg.edges
(live_in, live_out) = self._backward(blocks, edges, use, kill)
result: Dict[str, Dict[str, List[str]]] = {}
for iaddrs in blocks.values():
for ia in iaddrs:
lin = sorted(live_in.get(ia, set()))
lout = sorted(live_out.get(ia, set()))
if lin or lout:
result[ia] = {"live-in": lin, "live-out": lout}
return result

def _visit_order(self, blocks: Dict[str, List[str]]) -> List[str]:
"""Block addresses in the order the fixpoint should visit them.

A backward analysis converges fastest visiting blocks in reverse of
reverse-postorder, so successors are settled before their predecessors.
cfg.rpo_sorted_nodes supplies the reverse-postorder. Blocks it omits
(it is derived from the graph reachable from the entry) are appended, so
every block is still visited; order only affects how many rounds the
fixpoint takes, never the result.
"""
try:
rpo = list(self.fn.cfg.rpo_sorted_nodes)
except Exception:
return list(blocks)
ordered = [b for b in reversed(rpo) if b in blocks]
seen = set(ordered)
return ordered + [b for b in blocks if b not in seen]

def _backward(
self,
blocks: Dict[str, List[str]],
edges: Mapping[str, Sequence[str]],
use: Dict[str, Set[str]],
kill: Dict[str, Set[str]]
) -> Tuple[Dict[str, Set[str]], Dict[str, Set[str]]]:
"""Standard iterative backward live-variable analysis.

Returns (live_in, live_out), each instruction address -> set of live
names. CodeHawk has no dataflow framework to reuse for the fixpoint
itself; the CFG traversal it does provide is used via _visit_order.
"""
block_in: Dict[str, Set[str]] = {b: set() for b in blocks}
live_in: Dict[str, Set[str]] = {}
live_out: Dict[str, Set[str]] = {}
order = self._visit_order(blocks)

changed = True
while changed:
changed = False
for b in order:
iaddrs = blocks[b]
# live-out of the block = union of successors' block-entry sets
cur_out: Set[str] = set()
for s in edges.get(b, []):
cur_out |= block_in.get(s, set())
# walk the block backwards, threading live-out -> live-in
for ia in reversed(iaddrs):
live_out[ia] = set(cur_out)
lin = use.get(ia, set()) | (cur_out - kill.get(ia, set()))
live_in[ia] = lin
cur_out = lin
# cur_out is now the live-in at the block's first instruction
if cur_out != block_in[b]:
block_in[b] = cur_out
changed = True

return (live_in, live_out)
27 changes: 27 additions & 0 deletions chb/astinterface/ASTInterfaceFunction.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@

from chb.astinterface.ASTICodeTransformer import ASTICodeTransformer
from chb.astinterface.ASTICPrettyPrinter import ASTICPrettyPrinter
from chb.astinterface.ASTILiveness import ASTILiveness
from chb.astinterface.ASTInterface import ASTInterface
from chb.astinterface.ASTInterfaceBasicBlock import ASTInterfaceBasicBlock
from chb.astinterface.ASTInterfaceInstruction import ASTInterfaceInstruction
Expand Down Expand Up @@ -162,6 +163,8 @@ def mk_asts(self, support: CustomASTSupport) -> List[ASTStmt]:

# transfer provenance data to the AST abstract syntaxtree
self.astinterface.set_ast_provenance()
self.set_flag_liveness()
self.set_register_liveness(support)
self.set_invariants()
self.set_return_sequences()

Expand Down Expand Up @@ -249,6 +252,30 @@ def complete_instruction_connections(self) -> None:
for ll_instr in instr.ll_ast_instructions:
self.astinterface.add_instr_mapping(hl_instr, ll_instr)

def set_flag_liveness(self) -> None:
# Derive address-keyed NZCV flag liveness from the reaching-def facts
# and attach it to the provenance. This is auxiliary, so a failure here
# must not abort AST generation.
try:
liveness = ASTILiveness(self.function).flag_liveness()
self.astinterface.astree.provenance.flag_liveness = liveness
except Exception:
chklogger.logger.exception(
"flag-liveness computation failed for %s", self.function.faddr)

def set_register_liveness(self, support: CustomASTSupport) -> None:
# Derive address-keyed GP-register liveness from the reaching-def facts
# and attach it to the provenance. This is auxiliary, so a failure here
# must not abort AST generation.
try:
liveness = ASTILiveness(self.function).register_liveness(
set(support.register_sizes.keys()))
self.astinterface.astree.provenance.register_liveness = liveness
except Exception:
chklogger.logger.exception(
"register-liveness computation failed for %s",
self.function.faddr)

def set_invariants(self) -> None:
invariants = self.function.invariants
aexprs: Dict[str, Dict[str, Tuple[int, int, str]]] = {}
Expand Down