[Restructuring] Wrong Value in Condition

decompiler/frontend/binaryninja/handlers/constants.py

@@ -53,7 +53,11 @@ def lift_constant_pointer(self, pointer: mediumlevelil.MediumLevelILConstPtr, **
  if symbol is not None and symbol.type in (SymbolType.ImportedFunctionSymbol, SymbolType.ExternalSymbol, SymbolType.FunctionSymbol):
  return self._lift_symbol_pointer(address, symbol)
 
- if not isinstance(pointer, mediumlevelil.MediumLevelILImport) and (symbol is None or symbol.type != SymbolType.DataSymbol) and (string := bv.get_string_at(address, partial=True) or bv.get_ascii_string_at(address, min_length=2)):
+ if (
+ not isinstance(pointer, mediumlevelil.MediumLevelILImport)
+ and (symbol is None or symbol.type != SymbolType.DataSymbol)
+ and (string := bv.get_string_at(address, partial=True) or bv.get_ascii_string_at(address, min_length=2))
+ ):
  return Constant(address, Pointer(Integer.char()), Constant(string.value, Integer.char()))
 
  if (variable := bv.get_data_var_at(address)) is not None:

decompiler/pipeline/controlflowanalysis/restructuring.py

@@ -6,9 +6,11 @@
 import logging
 from typing import List, Optional
 
+from decompiler.pipeline.commons.reaching_definitions import ReachingDefinitions
 from decompiler.pipeline.controlflowanalysis.restructuring_commons.acyclic_restructuring import AcyclicRegionRestructurer
 from decompiler.pipeline.controlflowanalysis.restructuring_commons.cyclic_restructuring import CyclicRegionStructurer
 from decompiler.pipeline.controlflowanalysis.restructuring_commons.empty_basic_block_remover import EmptyBasicBlockRemover
+from decompiler.pipeline.controlflowanalysis.restructuring_commons.side_effect_handling.side_effect_handler import SideEffectHandler
 from decompiler.pipeline.stage import PipelineStage
 from decompiler.structures.ast.syntaxforest import AbstractSyntaxForest
 from decompiler.structures.ast.syntaxtree import AbstractSyntaxTree
@@ -33,6 +35,7 @@ def __init__(self, tcfg: Optional[TransitionCFG] = None, asforest: Optional[Abst
  """
  self.t_cfg: TransitionCFG = tcfg
  self.asforest: AbstractSyntaxForest = asforest
+ self._reaching_definitions: Optional[ReachingDefinitions] = None
 
  def run(self, task: DecompilerTask):
  """
@@ -48,6 +51,7 @@ def run(self, task: DecompilerTask):
  self.asforest.set_current_root(self.t_cfg.root.ast)
  assert (roots := len(self.asforest.get_roots)) == 1, f"After the restructuring the forest should have one root, but it has {roots}!"
  task._ast = AbstractSyntaxTree.from_asforest(self.asforest, self.asforest.current_root)
+ SideEffectHandler.resolve(task)
  task._cfg = None
 
  def restructure_cfg(self) -> None:

decompiler/pipeline/controlflowanalysis/restructuring_commons/side_effect_handling/data_graph.py

@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+from typing import Dict, Iterable, Optional, Type, Union
+
+from decompiler.pipeline.controlflowanalysis.restructuring_commons.side_effect_handling.data_graph_visitor import (
+ ASTDataGraphVisitor,
+ SubtreeProperty,
+)
+from decompiler.structures.ast.ast_nodes import AbstractSyntaxTreeNode, CaseNode, CodeNode, ConditionNode, LoopNode, SwitchNode, TrueNode
+from decompiler.structures.ast.syntaxtree import AbstractSyntaxTree
+from decompiler.structures.graphs.basicblock import BasicBlock
+from decompiler.structures.graphs.branches import BasicBlockEdge, FalseCase, SwitchCase, TrueCase, UnconditionalEdge
+from decompiler.structures.graphs.cfg import ControlFlowGraph
+from decompiler.structures.logic.logic_condition import LogicCondition
+from decompiler.structures.pseudo import Assignment, Branch, Call, Condition, Constant, ListOperation, Variable
+from networkx import MultiDiGraph
+
+
+class DataNode(BasicBlock):
+ def copy(self) -> DataNode:
+ """Return a deep copy of the node."""
+ return DataNode(self._address, [instruction.copy() for instruction in self._instructions], graph=self._graph)
+
+ @classmethod
+ def generate_node_from(
+ cls, idx: int, ast_node: AbstractSyntaxTreeNode, condition_map: Dict[LogicCondition, Condition]
+ ) -> Optional[DataNode]:
+ if isinstance(ast_node, CodeNode):
+ return DataNode(idx, ast_node.instructions)
+ if isinstance(ast_node, ConditionNode):
+ return LogicNode(idx, ast_node, condition_map)
+ if isinstance(ast_node, SwitchNode):
+ return DataNode(idx, [Assignment(ListOperation([]), Call(Variable("switch"), [ast_node.expression]))])
+ if isinstance(ast_node, LoopNode):
+ return LogicNode(idx, ast_node, condition_map)
+ if isinstance(ast_node, CaseNode):
+ constant = ast_node.constant if isinstance(ast_node.constant, Constant) else Constant(ast_node.constant)
+ return DataNode(idx, [Assignment(ListOperation([]), Call(Variable("case"), [constant]))])
+ return None
+
+
+class LogicNode(DataNode):
+ def __init__(
+ self,
+ name: int,
+ ast_node: Union[LoopNode, ConditionNode],
+ condition_map: Dict[LogicCondition, Condition],
+ graph: ControlFlowGraph = None,
+ ):
+ super().__init__(name, [Branch(condition_map[symbol]) for symbol in ast_node.condition.get_symbols()], graph)
+ self._logic_condition = ast_node.condition
+ self._condition_map = condition_map
+ self._type = ast_node.loop_type.value if isinstance(ast_node, LoopNode) else "if"
+
+ def __str__(self) -> str:
+ """Return a string representation of all instructions in the basic block."""
+ return f"{self._type}({self._logic_condition.rich_string_representation(self._condition_map)})"
+
+ def __eq__(self, other: object) -> bool:
+ """Basic Blocks can be equal based on their contained instructions and addresses."""
+ return isinstance(other, LogicNode) and self._address == other._address
+
+ def __hash__(self) -> int:
+ """
+ Basic Blocks should hash the same even then in different graphs.
+
+ Since addresses are supposed to be unique,
+ they are used for hashing in order to identify the same Block with different instruction as equal.
+ """
+ return hash(self._address)
+
+ @property
+ def logic_condition(self) -> LogicCondition:
+ return self._logic_condition
+
+ def copy(self) -> LogicNode:
+ """Return a deep copy of the node."""
+ return LogicNode(self._address, self._logic_condition, self._condition_map, graph=self._graph)
+
+
+class DataGraph(ControlFlowGraph):
+ NODE = DataNode
+ EDGE = BasicBlockEdge
+
+ def __init__(self, graph: Optional[MultiDiGraph] = None, root: Optional[NODE] = None):
+ """
+ Init a new empty instance.
+
+ - translation_dict maps the AST-nodes to the data-nodes of the graph
+ """
+ super().__init__(graph, root)
+ self._translation_dict: Dict[AbstractSyntaxTreeNode, DataNode] = dict()
+
+ @classmethod
+ def generate_from_ast(cls, ast: AbstractSyntaxTree) -> DataGraph:
+ data_graph = cls()
+ property_dict = data_graph.generate_nodes(ast)
+ data_graph.generate_edges(ast, property_dict)
+ return data_graph
+
+ def generate_nodes(self, ast: AbstractSyntaxTree) -> Dict[AbstractSyntaxTreeNode, SubtreeProperty]:
+ """Generate nodes from the given ast for the data graph."""
+ ast_data_graph_visitor: ASTDataGraphVisitor = ASTDataGraphVisitor()
+ idx = 0
+ for ast_node in ast.post_order():
+ base_node = DataNode.generate_node_from(idx, ast_node, ast.condition_map)
+ if base_node is not None:
+ self.add_node(base_node)
+ self._translation_dict[ast_node] = base_node
+ idx += 1
+ ast_data_graph_visitor.visit(ast_node)
+ return ast_data_graph_visitor.property_dict
+
+ def generate_edges(self, ast: AbstractSyntaxTree, property_dict: Dict[AbstractSyntaxTreeNode, SubtreeProperty]) -> None:
+ """Generate edges between the data-graph nodes"""
+ for seq_node in ast.get_sequence_nodes_post_order():
+ for source_child, sink_child in zip(seq_node.children[:-1], seq_node.children[1:]):
+ self._add_edges_between(property_dict[source_child].last_nodes, {property_dict[sink_child].first_node})
+ for cond_node in ast.get_condition_nodes_post_order():
+ for branch in cond_node.children:
+ edge_type = TrueCase if isinstance(branch, TrueNode) else FalseCase
+ self._add_edges_between({cond_node}, {property_dict[branch.child].first_node}, edge_type)
+ for loop_node in ast.get_loop_nodes_post_order():
+ self._add_edges_between(property_dict[loop_node.body].continue_nodes | property_dict[loop_node.body].last_nodes, {loop_node})
+ self._add_edges_between({loop_node}, {property_dict[loop_node.body].first_node}, TrueCase)
+ for switch_node in ast.get_switch_nodes_post_order():
+ self._add_edges_between({switch_node}, switch_node.children, SwitchCase)
+ for source_case, sink_case in zip(switch_node.children[:-1], switch_node.children[1:]):
+ if source_case.break_case:
+ continue
+ self._add_edges_between(property_dict[source_case].last_nodes, {property_dict[sink_case].first_node})
+ # edges for case_nodes
+ for case in switch_node.children:
+ self._add_edges_between({case}, {property_dict[case.child].first_node})
+
+ def _add_edges_between(
+ self,
+ sources: Iterable[AbstractSyntaxTreeNode],
+ sinks: Iterable[AbstractSyntaxTreeNode],
+ edge_type: Type[BasicBlockEdge] = UnconditionalEdge,
+ ):
+ """Add edges between the corresponding base-nodes of the source nodes and the base-nodes of the sink nodes."""
+ for source in sources:
+ for sink in sinks:
+ if edge_type == SwitchCase:
+ assert isinstance(sink, CaseNode)
+ self.add_edge(SwitchCase(self._translation_dict[source], self._translation_dict[sink], [sink.constant]))
+ elif edge_type == UnconditionalEdge and isinstance(source, (ConditionNode, LoopNode)):
+ self.add_edge(FalseCase(self._translation_dict[source], self._translation_dict[sink]))
+ else:
+ self.add_edge(edge_type(self._translation_dict[source], self._translation_dict[sink]))
+
+ def get_logic_nodes(self) -> Iterable[LogicNode]:
+ """Yield all logic nodes of the data-graph"""
+ for node in self.nodes:
+ if isinstance(node, LogicNode):
+ yield node

decompiler/pipeline/controlflowanalysis/restructuring_commons/side_effect_handling/data_graph_visitor.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, Set
+
+from decompiler.structures.ast.ast_nodes import (
+ AbstractSyntaxTreeNode,
+ CaseNode,
+ CodeNode,
+ ConditionNode,
+ DoWhileLoopNode,
+ FalseNode,
+ LoopNode,
+ SeqNode,
+ SwitchNode,
+ TrueNode,
+ VirtualRootNode,
+)
+from decompiler.structures.visitors.interfaces import ASTVisitorInterface
+
+
+@dataclass
+class SubtreeProperty:
+ root: AbstractSyntaxTreeNode
+ first_node: AbstractSyntaxTreeNode
+ last_nodes: Set[AbstractSyntaxTreeNode] = field(default_factory=set)
+ continue_nodes: Set[AbstractSyntaxTreeNode] = field(default_factory=set)
+ break_nodes: Set[AbstractSyntaxTreeNode] = field(default_factory=set)
+
+
+class ASTDataGraphVisitor(ASTVisitorInterface):
+ def __init__(self):
+ super().__init__()
+ self._property_dict: Dict[AbstractSyntaxTreeNode, SubtreeProperty] = dict()
+
+ @property
+ def property_dict(self) -> Dict[AbstractSyntaxTreeNode, SubtreeProperty]:
+ return self._property_dict
+
+ def visit_seq_node(self, node: SeqNode) -> None:
+ first_node = self._property_dict[node.children[0]].first_node
+ last_nodes = self._property_dict[node.children[-1]].last_nodes
+ continue_nodes = set().union(*(self._property_dict[child].continue_nodes for child in node.children))
+ break_nodes = set().union(*(self._property_dict[child].break_nodes for child in node.children))
+ self._property_dict[node] = SubtreeProperty(node, first_node, last_nodes, continue_nodes, break_nodes)
+
+ def visit_code_node(self, node: CodeNode) -> None:
+ first_node = node
+ last_nodes = set() if node.does_end_with_break else {node}
+ continue_nodes = {node} if node.does_end_with_continue else set()
+ break_nodes = {node} if node.does_end_with_break else set()
+ self._property_dict[node] = SubtreeProperty(node, first_node, last_nodes, continue_nodes, break_nodes)
+
+ def visit_condition_node(self, node: ConditionNode) -> None:
+ first_node = node
+ last_nodes = {node} if node.false_branch_child is None else set()
+ last_nodes = last_nodes.union(*(self._property_dict[branch.child].last_nodes for branch in node.children))
+ continue_nodes = set().union(*(self._property_dict[branch.child].continue_nodes for branch in node.children))
+ break_nodes = set().union(*(self._property_dict[branch.child].break_nodes for branch in node.children))
+ self._property_dict[node] = SubtreeProperty(node, first_node, last_nodes, continue_nodes, break_nodes)
+
+ def visit_loop_node(self, node: LoopNode) -> None:
+ first_node = self._property_dict[node.body].first_node if isinstance(node, DoWhileLoopNode) else node
+ last_nodes = set() if node.is_endless_loop else {node}
+ last_nodes |= self._property_dict[node.body].break_nodes
+ continue_nodes = set()
+ break_nodes = set()
+ self._property_dict[node] = SubtreeProperty(node, first_node, last_nodes, continue_nodes, break_nodes)
+
+ def visit_switch_node(self, node: SwitchNode) -> None:
+ first_node = node
+ last_nodes = self._property_dict[node.default].last_nodes if node.default else set()
+ last_nodes = last_nodes.union(*(self._property_dict[case].last_nodes for case in node.cases if case.break_case))
+ continue_nodes = set().union(*(self._property_dict[case].continue_nodes for case in node.children))
+ break_nodes = set()
+ self._property_dict[node] = SubtreeProperty(node, first_node, last_nodes, continue_nodes, break_nodes)
+
+ def visit_case_node(self, node: CaseNode) -> None:
+ first_node = node
+ last_nodes = self._property_dict[node.child].last_nodes
+ continue_nodes = self._property_dict[node.child].continue_nodes
+ break_nodes = {node} if node.break_case else set()
+ self._property_dict[node] = SubtreeProperty(node, first_node, last_nodes, continue_nodes, break_nodes)
+
+ def visit_true_node(self, node: TrueNode) -> None:
+ pass
+
+ def visit_false_node(self, node: FalseNode) -> None:
+ pass
+
+ def visit_root_node(self, node: VirtualRootNode) -> None:
+ pass

decompiler/pipeline/controlflowanalysis/restructuring_commons/side_effect_handling/side_effect_handler.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from decompiler.pipeline.commons.reaching_definitions import ReachingDefinitions
+from decompiler.pipeline.controlflowanalysis.restructuring_commons.side_effect_handling.data_graph import DataGraph
+from decompiler.structures.ast.syntaxtree import AbstractSyntaxTree
+from decompiler.structures.graphs.cfg import ControlFlowGraph
+from decompiler.task import DecompilerTask
+
+
+class SideEffectHandler:
+ def __init__(self, ast: AbstractSyntaxTree, cfg: ControlFlowGraph, data_graph: DataGraph):
+ self._ast: AbstractSyntaxTree = ast
+ self._cfg: ControlFlowGraph = cfg
+ self._data_graph: DataGraph = data_graph
+
+ @classmethod
+ def resolve(cls, task: DecompilerTask) -> None:
+ # return
+ data_graph = DataGraph.generate_from_ast(task.syntax_tree)
+ side_effect_handler = cls(task.syntax_tree, task.graph, data_graph)
+ # DecoratedAST.from_ast(task.syntax_tree).export_plot("/home/eva/Projects/dewolf-decompiler/AST/ast.png")
+ from decompiler.util.decoration import DecoratedAST, DecoratedCFG
+
+ # DecoratedCFG.from_cfg(data_graph).export_plot("/home/eva/Projects/dewolf-decompiler/AST/cfg.png")
+ side_effect_handler.apply()
+
+ def apply(self):
+ reaching_definitions = ReachingDefinitions(self._data_graph)
+ all_defined_ssa_variables = set(var.ssa_name for node in self._data_graph.nodes for var in node.definitions)
+ # TODO: What about variables defined via phi-functions
+ # TODO: How about added variables without SSA-values?
+ # for logic_node in self._data_graph.get_logic_nodes():
+ # definitions: Dict[Variable, List[Instruction]] = dict()
+ # for instruction in reaching_definitions.reach_in_block(logic_node):
+ # for definition in instruction.definitions:
+ # definitions[definition] = definitions.get(definition, list()) + [instruction]
+ # for symbol in logic_node.logic_condition.get_symbols():
+ # pseudo_condition = self._ast.condition_map[symbol]
+ # for used_variable in pseudo_condition.requirements:
+ # if used_variable not in all_defined_ssa_variables:
+ # continue
+ # for def_instruction in definitions[used_variable]:
+ # definition = [def_var for def_var in def_instruction.definitions if def_var == used_variable][0]
+ # if used_variable.ssa_name != definition.ssa_name:
+ # raise "We have to handle side effects!"

decompiler/util/decoration.py

@@ -152,7 +152,7 @@ def _format_label(node: BasicBlock) -> str:
  if node.instructions is None:
  instructions_left_aligned = " "
  else:
- instructions_left_aligned = "\n".join(map(str, node.instructions))
+ instructions_left_aligned = str(node)
  return f"{node.name}.\n{instructions_left_aligned}"
 
  def export_flowgraph(self) -> FlowGraph: