From c5b5578382baac90731b9c5f4d3786f4fdd10eaf Mon Sep 17 00:00:00 2001 From: Damian Pfammatter Date: Wed, 9 Apr 2025 13:01:44 +0200 Subject: [PATCH] 58 Wrong Instructions For Path Near The Source Call (#104) * Path equality includes sink call instruction * Tag instruction origin * Refactoring source slicing (incomplete) * Merge source and sink instruction graphs (incomplete) * Merge source/sink graphs before path finding * Reverse source-originating graphs once * Handler for MLIL_BOOL_TO_INT (#106) * Handler for MLIL_CMP_ULT (#108) * Path equality includes sink call instruction * Tag instruction origin * Refactoring source slicing (incomplete) * Merge source and sink instruction graphs (incomplete) * Merge source/sink graphs before path finding * Reverse source-originating graphs once * new testcase with phi * new testcase for duplicate paths * Duplicate reduction * Fixing the call graphs * Log origin in path differences * Added new unittests * Different colors for highlighting source/sink instructions * Added comments for path equality * Without merging source/sink graphs * Reverse path instead of graph * Lazy path initialization * Fix navigation to src parm --------- Co-authored-by: wizche --- conf/001-settings.yml | 20 +- mole/controllers/path.py | 55 +++- mole/core/data.py | 491 ++++++++++++++++++++----------- mole/core/slice.py | 247 +++++++++------- mole/models/path.py | 40 ++- mole/services/config.py | 2 +- mole/views/config.py | 2 +- mole/views/graph.py | 4 +- mole/views/path_tree.py | 6 +- test/src/simple_http_server-04.c | 180 +++++++++++ test/test_data.py | 53 ++-- test/test_slicing.py | 367 ++++++++++++++++++++--- 12 files changed, 1072 insertions(+), 395 deletions(-) create mode 100644 test/src/simple_http_server-04.c diff --git a/conf/001-settings.yml b/conf/001-settings.yml index 3b32c51..47a38f5 100644 --- a/conf/001-settings.yml +++ b/conf/001-settings.yml @@ -17,9 +17,23 @@ settings: value: -1 min_value: -1 max_value: 9999 - highlight_color: - name: highlight_color - help: color used to highlight paths + src_highlight_color: + name: src_highlight_color + help: color used to highlight instructions originating from slicing a source function + value: Orange + items: + - Blue + - Green + - Cyan + - Red + - Magenta + - Yellow + - Orange + - White + - Black + snk_highlight_color: + name: snk_highlight_color + help: color used to highlight instructions originating from slicing a sink function value: Red items: - Blue diff --git a/mole/controllers/path.py b/mole/controllers/path.py index 13fef16..a3cf36a 100644 --- a/mole/controllers/path.py +++ b/mole/controllers/path.py @@ -1,6 +1,7 @@ from __future__ import annotations -from mole.core.data import InstructionHelper, Path +from mole.common.help import InstructionHelper from mole.common.task import BackgroundTask +from mole.core.data import Path from mole.services.path import PathService from mole.views.graph import GraphWidget from mole.views.path import PathView @@ -459,18 +460,24 @@ class PathController: log.info(tag, msg) if reverse: log.debug(tag, "--- Forward Slice ---") + src_inst_idx = len(path.insts) - path.src_inst_idx insts = reversed(path.insts) else: log.debug(tag, "--- Backward Slice ---") + src_inst_idx = path.src_inst_idx insts = path.insts basic_block = None - for inst in insts: + for i, inst in enumerate(insts): + if (not reverse and i < src_inst_idx) or (reverse and i >= src_inst_idx): + custom_tag = f"{tag}] [Snk" + else: + custom_tag = f"{tag}] [Src" if inst.il_basic_block != basic_block: basic_block = inst.il_basic_block fun_name = basic_block.function.name bb_addr = basic_block[0].address - log.debug(tag, f"- FUN: '{fun_name:s}', BB: 0x{bb_addr:x}") - log.debug(tag, InstructionHelper.get_inst_info(inst)) + log.debug(custom_tag, f"- FUN: '{fun_name:s}', BB: 0x{bb_addr:x}") + log.debug(custom_tag, InstructionHelper.get_inst_info(inst)) log.debug(tag, "----------------------") log.debug(tag, msg) return @@ -492,17 +499,27 @@ class PathController: if not path_0: return path_0_id = path_ids[0] - path_0_insts = [ - InstructionHelper.get_inst_info(inst, False) for inst in path_0.insts - ] + path_0_insts = [] + for i, inst in enumerate(path_0.insts): + if i < path_0.src_inst_idx: + ori = "[Snk]" + else: + ori = "[Src]" + info = InstructionHelper.get_inst_info(inst, False) + path_0_insts.append(f"{ori:s} {info}") # Get instructions of path 1 path_1 = self.path_tree_view.get_path(path_ids[1]) if not path_1: return path_1_id = path_ids[1] - path_1_insts = [ - InstructionHelper.get_inst_info(inst, False) for inst in path_1.insts - ] + path_1_insts = [] + for i, inst in enumerate(path_1.insts): + if i < path_1.src_inst_idx: + ori = "[Snk]" + else: + ori = "[Src]" + info = InstructionHelper.get_inst_info(inst, False) + path_1_insts.append(f"{ori:s} {info}") # Get terminal width and calculate column width ter_width = shu.get_terminal_size().columns col_width = ter_width // 2 - 2 @@ -602,12 +619,22 @@ class PathController: highlighted_path = path insts_colors = {} try: - setting = self.config_ctr.get_setting("highlight_color") + setting = self.config_ctr.get_setting("src_highlight_color") color_name = setting.widget.currentText().capitalize() - color = bn.HighlightStandardColor[f"{color_name:s}HighlightColor"] + src_color = bn.HighlightStandardColor[f"{color_name:s}HighlightColor"] except Exception as _: - color = bn.HighlightStandardColor.RedHighlightColor - for inst in path.insts: + src_color = bn.HighlightStandardColor.RedHighlightColor + try: + setting = self.config_ctr.get_setting("snk_highlight_color") + color_name = setting.widget.currentText().capitalize() + snk_color = bn.HighlightStandardColor[f"{color_name:s}HighlightColor"] + except Exception as _: + snk_color = bn.HighlightStandardColor.RedHighlightColor + for i, inst in enumerate(path.insts): + if i < path.src_inst_idx: + color = snk_color + else: + color = src_color func = inst.function.source_function addr = inst.address if addr not in insts_colors: diff --git a/mole/core/data.py b/mole/core/data.py index 6d0bdb4..42883c5 100644 --- a/mole/core/data.py +++ b/mole/core/data.py @@ -1,11 +1,16 @@ from __future__ import annotations from mole.common.help import InstructionHelper, SymbolHelper -from mole.core.slice import MediumLevelILBackwardSlicer, MediumLevelILFunctionGraph +from mole.core.slice import ( + MediumLevelILBackwardSlicer, + MediumLevelILFunctionGraph, + MediumLevelILInstructionGraph, +) from dataclasses import dataclass, field from mole.common.log import log -from typing import Callable, Dict, List, Tuple +from typing import Callable, Dict, List, Optional, Tuple import binaryninja as bn -import hashlib as hashlib +import hashlib +import networkx as nx import PySide6.QtWidgets as qtw @@ -174,9 +179,17 @@ class SourceFunction(Function): This class is a representation of the data associated with source functions. """ - target_insts: Dict[Tuple[int, str], List[bn.MediumLevelILInstruction]] = field( - default_factory=dict - ) + src_map: Dict[ + Tuple[ + int, # src_sym_addr + str, # src_sym_name + bn.MediumLevelILInstruction, # src_call_inst + ], + Dict[ + Tuple[int, bn.MediumLevelILInstruction], # src_par_idx, src_par_var + MediumLevelILInstructionGraph, # src_inst_graph + ], + ] = field(default_factory=dict) def __eq__(self, other: Function) -> bool: if not isinstance(other, SourceFunction): @@ -190,78 +203,92 @@ class SourceFunction(Function): """ This method finds a set of target instructions that a static backward slice should hit on. """ - custom_tag = f"Mole.Src.{self.name:s}" - self.target_insts.clear() + custom_tag = f"{tag:s}.Src.{self.name:s}" + # Clear map + self.src_map.clear() + # Get code references of symbols code_refs = SymbolHelper.get_code_refs( bv, self.symbols, [bn.SymbolType.FunctionSymbol, bn.SymbolType.ImportedFunctionSymbol], ) - for src_name, src_insts in code_refs.items(): + # Iterate code references + for src_sym_name, src_insts in code_refs.items(): if canceled(): break + # Iterate source instructions for src_inst in src_insts: if canceled(): break + src_sym_addr = src_inst.address log.info( custom_tag, - f"Analyze source function '0x{src_inst.address:x} {src_name:s}'", + f"Analyze source function '0x{src_sym_addr:x} {src_sym_name:s}'", ) # Ignore everything but call instructions - match src_inst: - case bn.MediumLevelILCallSsa() | bn.MediumLevelILTailcallSsa(): - self.target_insts.setdefault( - (src_inst.address, src_name), [] - ).append(src_inst) - case _: - continue + if not ( + isinstance(src_inst, bn.MediumLevelILCallSsa) + or isinstance(src_inst, bn.MediumLevelILTailcallSsa) + ): + continue + src_call_inst = src_inst # Ignore calls with an invalid number of parameters - if not self.par_cnt_fun(len(src_inst.params)): + if not self.par_cnt_fun(len(src_call_inst.params)): log.warn( custom_tag, - f"0x{src_inst.address:x} Ignore arguments of call '0x{src_inst.address:x} {src_name:s}' due to an unexpected amount", + f"0x{src_sym_addr:x} Ignore call '0x{src_sym_addr:x} {src_sym_name:s}' due to an invalid number of arguments", ) continue - # Analyze parameters - for par_idx, par_var in enumerate(src_inst.params): + src_par_map = self.src_map.setdefault( + (src_sym_addr, src_sym_name, src_call_inst), {} + ) + # Iterate source instruction's parameters + for src_par_idx, src_par_var in enumerate(src_call_inst.params): if canceled(): break - par_idx += 1 + src_par_idx += 1 log.debug( custom_tag, - f"Analyze argument 'arg#{par_idx:d}:{str(par_var):s}'", + f"Analyze argument 'arg#{src_par_idx:d}:{str(src_par_var):s}'", ) - # Perform dataflow analysis - if self.par_dataflow_fun(par_idx): + # Perform dataflow analysis on the parameter + if self.par_dataflow_fun(src_par_idx): # Ignore constant parameters - if par_var.operation != bn.MediumLevelILOperation.MLIL_VAR_SSA: + if ( + src_par_var.operation + != bn.MediumLevelILOperation.MLIL_VAR_SSA + ): log.debug( custom_tag, - f"0x{src_inst.address:x} Ignore constant argument 'arg#{par_idx:d}:{str(par_var):s}'", + f"0x{src_sym_addr:x} Ignore constant argument 'arg#{src_par_idx:d}:{str(src_par_var):s}'", ) continue # Ignore parameters that can be determined with dataflow analysis - possible_sizes = par_var.possible_values + possible_sizes = src_par_var.possible_values if ( possible_sizes.type != bn.RegisterValueType.UndeterminedValue ): log.debug( custom_tag, - f"0x{src_inst.address:x} Ignore dataflow determined argument 'arg#{par_idx:d}:{str(par_var):s}'", + f"0x{src_sym_addr:x} Ignore dataflow determined argument 'arg#{src_par_idx:d}:{str(src_par_var):s}'", ) continue - # Backward slice the parameter - if self.par_slice_fun(par_idx): - slicer = MediumLevelILBackwardSlicer(bv, custom_tag, 0) - slicer.slice_backwards(par_var) - # Add sliced instructions to the target instructions - addr_src_list = self.target_insts.setdefault( - (src_inst.address, src_name), [] - ) - for inst in slicer.get_insts(): - if inst not in addr_src_list: - addr_src_list.append(inst) + # Create backward slicer + src_slicer = MediumLevelILBackwardSlicer(bv, custom_tag, 0) + # Add edge between call and parameter instructions + src_slicer.inst_graph.add_node( + src_call_inst, 0, src_call_inst.function, origin="src" + ) + src_slicer.inst_graph.add_node( + src_par_var, 0, src_par_var.function, origin="src" + ) + src_slicer.inst_graph.add_edge(src_call_inst, src_par_var) + # Perform backward slicing of the parameter + if self.par_slice_fun(src_par_idx): + src_slicer.slice_backwards(src_par_var) + # Store the instruction graph + src_par_map[(src_par_idx, src_par_var)] = src_slicer.inst_graph return @@ -292,153 +319,210 @@ class SinkFunction(Function): This method tries to find paths, starting from the current sink and ending in one of the given `sources` using static backward slicing. """ - paths = [] - custom_tag = f"Mole.Snk.{self.name:s}" + paths: List[Path] = [] + custom_tag = f"{tag:s}.Snk.{self.name:s}" + # Calculate SHA1 hash of binary sha1_hash = hashlib.sha1(bv.file.raw.read(0, bv.file.raw.end)).hexdigest() + # Get code references of symbols code_refs = SymbolHelper.get_code_refs( bv, self.symbols, [bn.SymbolType.FunctionSymbol, bn.SymbolType.ImportedFunctionSymbol], ) - for snk_name, snk_insts in code_refs.items(): + # Iterate code references + for snk_sym_name, snk_insts in code_refs.items(): if canceled(): break + # Iterate sink instructions for snk_inst in snk_insts: if canceled(): break + snk_sym_addr = snk_inst.address log.info( custom_tag, - f"Analyze sink function '0x{snk_inst.address:x} {snk_name:s}'", + f"Analyze sink function '0x{snk_sym_addr:x} {snk_sym_name:s}'", ) # Ignore everything but call instructions - match snk_inst: - case bn.MediumLevelILCallSsa() | bn.MediumLevelILTailcallSsa(): - pass - case _: - continue + if not ( + isinstance(snk_inst, bn.MediumLevelILCallSsa) + or isinstance(snk_inst, bn.MediumLevelILTailcallSsa) + ): + continue + snk_call_inst = snk_inst # Ignore calls with an invalid number of parameters - if not self.par_cnt_fun(len(snk_inst.params)): + if not self.par_cnt_fun(len(snk_call_inst.params)): log.warn( custom_tag, - f"0x{snk_inst.address:x} Ignore call '0x{snk_inst.address:x} {snk_name:s}' due to invalid number of arguments", + f"0x{snk_sym_addr:x} Ignore call '0x{snk_sym_addr:x} {snk_sym_name:s}' due to an invalid number of arguments", ) continue - # Analyze parameters - for par_idx, par_var in enumerate(snk_inst.params): + # Iterate sink instruction's parameters + for snk_par_idx, snk_par_var in enumerate(snk_call_inst.params): if canceled(): break - par_idx += 1 + snk_par_idx += 1 log.debug( custom_tag, - f"Analyze argument 'arg#{par_idx:d}:{str(par_var):s}'", + f"Analyze argument 'arg#{snk_par_idx:d}:{str(snk_par_var):s}'", ) - # Perform dataflow analysis - if self.par_dataflow_fun(par_idx): + # Perform dataflow analysis on the parameter + if self.par_dataflow_fun(snk_par_idx): # Ignore constant parameters - if par_var.operation != bn.MediumLevelILOperation.MLIL_VAR_SSA: + if ( + snk_par_var.operation + != bn.MediumLevelILOperation.MLIL_VAR_SSA + ): log.debug( custom_tag, - f"0x{snk_inst.address:x} Ignore constant argument 'arg#{par_idx:d}:{str(par_var):s}'", + f"0x{snk_sym_addr:x} Ignore constant argument 'arg#{snk_par_idx:d}:{str(snk_par_var):s}'", ) continue # Ignore parameters that can be determined with dataflow analysis - possible_sizes = par_var.possible_values + possible_sizes = snk_par_var.possible_values if ( possible_sizes.type != bn.RegisterValueType.UndeterminedValue ): log.debug( custom_tag, - f"0x{snk_inst.address:x} Ignore dataflow determined argument 'arg#{par_idx:d}:{str(par_var):s}'", + f"0x{snk_sym_addr:x} Ignore dataflow determined argument 'arg#{snk_par_idx:d}:{str(snk_par_var):s}'", ) continue - # Backward slice the parameter - if self.par_slice_fun(par_idx): - slicer = MediumLevelILBackwardSlicer( + # Peform backward slicing of the parameter + if self.par_slice_fun(snk_par_idx): + # Create backward slicer + snk_slicer = MediumLevelILBackwardSlicer( bv, custom_tag, max_call_level ) - slicer.slice_backwards(par_var) + snk_inst_graph = snk_slicer.inst_graph + snk_call_graph = snk_slicer.call_graph + # Add edge between call and parameter instructions + snk_inst_graph.add_node( + snk_call_inst, 0, snk_call_inst.function, origin="snk" + ) + snk_inst_graph.add_node( + snk_par_var, 0, snk_par_var.function, origin="snk" + ) + snk_inst_graph.add_edge(snk_call_inst, snk_par_var) + # Backward slice the parameter instruction + snk_slicer.slice_backwards(snk_par_var) + # Iterate sources for source in sources: if canceled(): break + # Iterate source instructions for ( src_sym_addr, src_sym_name, - ), src_insts in source.target_insts.items(): + src_call_inst, + ), src_par_map in source.src_map.items(): if canceled(): break - for src_inst in src_insts: + # Iterate source instruction's parameters + for (src_par_idx, src_par_var), ( + src_inst_graph + ) in src_par_map.items(): if canceled(): break - # Find paths - for insts, call_graph in slicer.find_paths( - par_var, src_inst, max_slice_depth - ): - # Prepend sink instruction - insts.insert(0, snk_inst) - # Find split between sink and source originating instructions - src_inst_idx = len(insts) - for src_inst_idx in range( - src_inst_idx - 1, -1, -1 - ): - if insts[src_inst_idx] not in src_insts: - break - src_inst_idx += 1 - # Add additional attributes to call graph - if snk_inst.function in call_graph: - call_graph.nodes[snk_inst.function][ - "snk" - ] = f"snk: {snk_name:s} | {str(par_var):s}" - if src_inst.function in call_graph: - call_graph.nodes[src_inst.function][ - "src" - ] = f"src: {src_sym_name:s}" - # Create path - path = Path( - src_sym_addr=src_sym_addr, - src_sym_name=src_sym_name, - snk_sym_addr=snk_inst.address, - snk_sym_name=snk_name, - snk_par_idx=par_idx, - snk_par_var=par_var, - src_inst_idx=src_inst_idx, - comment="", - sha1_hash=sha1_hash, - insts=insts, - call_graph=call_graph, - ) - # Found the same path before - if path in paths: + # Iterate source instructions (order of backward slicing) + for src_inst in src_inst_graph.nodes(): + # Ignore source instructions that were not sliced in the sink + if src_inst not in snk_inst_graph: continue - # Store path - paths.append(path) - if found_path: - found_path(path) - # Log path - t_log = f"Interesting path: {str(path):s}" - t_log = f"{t_log:s} [L:{len(insts):d},P:{len(path.phiis):d},B:{len(path.bdeps):d}]!" - log.info(custom_tag, t_log) - log.debug(custom_tag, "--- Backward Slice ---") - basic_block = None - for idx, inst in enumerate(insts): - if idx == src_inst_idx: + # Adjust negative `max_slice_depth` values + if ( + max_slice_depth is not None + and max_slice_depth < 0 + ): + max_slice_depth = None + # Find all simple paths starting at the sink's call + # instruction and ending in the current source instruction + snk_paths: List[ + List[bn.MediumLevelILInstruction] + ] = [] + try: + snk_paths = nx.all_simple_paths( + snk_inst_graph, + snk_call_inst, + src_inst, + max_slice_depth, + ) + except (nx.NodeNotFound, nx.NetworkXNoPath): + # Go to the next source instruction if no path found + continue + # Find shortest path starting at the source's call + # instruction and ending in the current source instruction + src_path: List[bn.MediumLevelILInstruction] = [] + try: + src_path = nx.shortest_path( + src_inst_graph, src_call_inst, src_inst + ) + except (nx.NodeNotFound, nx.NetworkXNoPath): + # Go to the next source instruction if no path found + continue + # Reverse the source path so it can be appended to the sink path + src_path = list(reversed(src_path)) + # Iterate found paths + for snk_path in snk_paths: + # Create a new path object + path = Path( + src_sym_addr=src_sym_addr, + src_sym_name=src_sym_name, + src_par_idx=src_par_idx, + src_par_var=src_par_var, + src_inst_idx=len(snk_path), + snk_sym_addr=snk_sym_addr, + snk_sym_name=snk_sym_name, + snk_par_idx=snk_par_idx, + snk_par_var=snk_par_var, + insts=snk_path + src_path[1:], + comment="", + sha1_hash=sha1_hash, + ) + # Ignore the path if we found it before + if path in paths: + continue + # Fully initialize the path + path.init( + src_call_inst.function, + f"src: {src_sym_name:s} | {str(src_par_var):s}", + snk_call_inst.function, + f"snk: {snk_sym_name:s} | {str(snk_par_var):s}", + snk_call_graph, + ) + # Store the path + paths.append(path) + # Execute callback on a newly found path + if found_path: + found_path(path) + # Log newly found path + t_log = f"Interesting path: {str(path):s}" + t_log = f"{t_log:s} [L:{len(path.insts):d},P:{len(path.phiis):d},B:{len(path.bdeps):d}]!" + log.info(custom_tag, t_log) + log.debug( + custom_tag, "--- Backward Slice ---" + ) + basic_block = None + for inst in path.insts: + if inst.il_basic_block != basic_block: + basic_block = inst.il_basic_block + fun_name = basic_block.function.name + bb_addr = basic_block[0].address + log.debug( + custom_tag, + f"- FUN: '{fun_name:s}', BB: 0x{bb_addr:x}", + ) log.debug( custom_tag, - "--- Source Function ---", - ) - if inst.il_basic_block != basic_block: - basic_block = inst.il_basic_block - fun_name = basic_block.function.name - bb_addr = basic_block[0].address - log.debug( - custom_tag, - f"- FUN: '{fun_name:s}', BB: 0x{bb_addr:x}", + InstructionHelper.get_inst_info( + inst + ), ) log.debug( - custom_tag, - InstructionHelper.get_inst_info(inst), + custom_tag, "-----------------------" ) - log.debug(custom_tag, "-----------------------") + # Ignore all other source instructions since a path was found + break return paths @@ -450,70 +534,52 @@ class Path: src_sym_addr: int src_sym_name: str + src_par_idx: int + src_par_var: bn.MediumLevelILInstruction + src_inst_idx: int snk_sym_addr: int snk_sym_name: str snk_par_idx: int - snk_par_var: bn.MediumLevelILVarSsa - src_inst_idx: int - comment: str - sha1_hash: str + snk_par_var: bn.MediumLevelILInstruction insts: List[bn.MediumLevelILInstruction] = field(default_factory=list) phiis: List[bn.MediumLevelILInstruction] = field(default_factory=list) bdeps: Dict[int, bn.ILBranchDependence] = field(default_factory=dict) + calls: List[Tuple[int, str, int]] = field(default_factory=list) call_graph: MediumLevelILFunctionGraph = field( default_factory=MediumLevelILFunctionGraph ) + comment: str = "" + sha1_hash: str = "" def __init__( self, src_sym_addr: int, src_sym_name: str, + src_par_idx: int, + src_par_var: bn.MediumLevelILInstruction, + src_inst_idx: int, snk_sym_addr: int, snk_sym_name: str, snk_par_idx: int, - snk_par_var: bn.MediumLevelILVarSsa, - src_inst_idx: int, - comment: str, - sha1_hash: str, + snk_par_var: bn.MediumLevelILInstruction, insts: List[bn.MediumLevelILInstruction] = field(default_factory=list), - call_graph: MediumLevelILFunctionGraph = field( - default_factory=MediumLevelILFunctionGraph - ), + call_graph: MediumLevelILFunctionGraph = MediumLevelILFunctionGraph(), + comment: str = "", + sha1_hash: str = "", ) -> None: self.src_sym_addr = src_sym_addr self.src_sym_name = src_sym_name + self.src_par_idx = src_par_idx + self.src_par_var = src_par_var + self.src_inst_idx = src_inst_idx self.snk_sym_addr = snk_sym_addr self.snk_sym_name = snk_sym_name self.snk_par_idx = snk_par_idx self.snk_par_var = snk_par_var - self.src_inst_idx = src_inst_idx - self.comment = comment - self.sha1_hash = sha1_hash self.insts = insts self.call_graph = call_graph - self._init_metrics() - self._init_calls() - return - - def _init_metrics(self) -> None: - self.phiis = [] - self.bdeps = {} - for inst in self.insts: - if isinstance(inst, bn.MediumLevelILVarPhi): - self.phiis.append(inst) - for bch_idx, bch_dep in inst.branch_dependence.items(): - self.bdeps.setdefault(bch_idx, bch_dep) - return - - def _init_calls(self) -> None: - self.calls = [] - for inst in self.insts: - func_name = inst.function.source_function.name - if len(self.calls) == 0 or self.calls[-1][1] != func_name: - call_level = self.call_graph.nodes.get(inst.function, {}).get( - "call_level", 0 - ) - self.calls.append((inst.address, func_name, call_level)) + self.comment = comment + self.sha1_hash = sha1_hash return def __eq__(self, other: Path) -> bool: @@ -523,23 +589,89 @@ class Path: except Exception as _: return False return ( + # Equal source self.src_sym_addr == other.src_sym_addr and self.src_sym_name == other.src_sym_name + and self.src_par_idx == other.src_par_idx + and self.src_par_var == other.src_par_var + # Equal sink and self.snk_sym_addr == other.snk_sym_addr and self.snk_sym_name == other.snk_sym_name and self.snk_par_idx == other.snk_par_idx and self.snk_par_var == other.snk_par_var + # Equal instructions (ignoring the ones originating from slicing the source, only + # considering the source's call instruction) + and self.src_inst_idx == other.src_inst_idx + and self.insts[: self.src_inst_idx - 1] + == other.insts[: self.src_inst_idx - 1] + and self.insts[-1] == other.insts[-1] + # Equal binary and self.sha1_hash == other.sha1_hash - and self.insts[1 : self.src_inst_idx - 1] - == other.insts[1 : other.src_inst_idx - 1] ) def __str__(self) -> str: src = f"0x{self.src_sym_addr:x} {self.src_sym_name:s}" + src = f"{src:s}(arg#{self.src_par_idx:d}:{str(self.src_par_var):s})" snk = f"0x{self.snk_sym_addr:x} {self.snk_sym_name:s}" snk = f"{snk:s}(arg#{self.snk_par_idx:d}:{str(self.snk_par_var):s})" return f"{src:s} --> {snk:s}" + def init( + self, + src_func: bn.MediumLevelILFunction, + src_info: str, + snk_func: bn.MediumLevelILFunction, + snk_info: str, + call_graph: MediumLevelILFunctionGraph, + ) -> None: + self.phiis = [] + self.bdeps = {} + self.calls = [] + self.call_graph = MediumLevelILFunctionGraph() + # Copy all nodes with added attribute `in_path=False` + for node, attrs in call_graph.nodes(data=True): + new_attrs = {**attrs, "in_path": False} + self.call_graph.add_node(node, **new_attrs) + # Change node attribute to `in_path=True` where functions are in the path + old_func_name = None + for inst in self.insts: + # Phi-instructions + if isinstance(inst, bn.MediumLevelILVarPhi): + self.phiis.append(inst) + # Branch dependencies + for bch_idx, bch_dep in inst.branch_dependence.items(): + self.bdeps.setdefault(bch_idx, bch_dep) + # Function information + func = inst.function + func_name = func.source_function.name + # Continue if the function does not change + if func_name == old_func_name: + continue + # Function calls + call_level = self.call_graph.nodes.get(func, {}).get("call_level", 0) + self.calls.append((inst.address, func_name, call_level)) + # Function calls graph + if func in self.call_graph: + self.call_graph.nodes[func]["in_path"] = True + # Store old function name + old_func_name = func_name + # Copy all edges with added attribute `in_path` stating whether or not both nodes have + # `in_path == True` + for node_from, node_to, attrs in call_graph.edges(data=True): + in_path = ( + self.call_graph.nodes[node_from]["in_path"] + and self.call_graph.nodes[node_to]["in_path"] + ) + new_attrs = {**attrs, "in_path": in_path} + self.call_graph.add_edge(node_from, node_to, **new_attrs) + # Add `src` node attribute + if src_func in self.call_graph: + self.call_graph.nodes[src_func]["src"] = src_info + # Add `snk` node attribute + if snk_func in self.call_graph: + self.call_graph.nodes[snk_func]["snk"] = snk_info + return + def to_dict(self) -> Dict: # Serialize instructions insts: List[Tuple[int, int]] = [] @@ -548,39 +680,44 @@ class Path: return { "src_sym_addr": hex(self.src_sym_addr), "src_sym_name": self.src_sym_name, + "src_par_idx": self.src_par_idx, + "src_inst_idx": self.src_inst_idx, "snk_sym_addr": hex(self.snk_sym_addr), "snk_sym_name": self.snk_sym_name, "snk_par_idx": self.snk_par_idx, - "src_inst_idx": self.src_inst_idx, - "comment": self.comment, - "sha1_hash": self.sha1_hash, "insts": insts, "call_graph": self.call_graph.to_dict(), + "comment": self.comment, + "sha1_hash": self.sha1_hash, } @classmethod - def from_dict(cls: Path, bv: bn.BinaryView, d: Dict) -> Path | None: + def from_dict(cls: Path, bv: bn.BinaryView, d: Dict) -> Optional[Path]: # Deserialize instructions insts: List[bn.MediumLevelILInstruction] = [] for func_addr, expr_idx in d["insts"]: func = bv.get_function_at(int(func_addr, 0)) inst = func.mlil.ssa_form.get_expr(expr_idx) insts.append(inst) - # Deserialize sink parameter variable + # Deserialize parameter variables + src_par_idx = d["src_par_idx"] + src_par_var = insts[-1].params[src_par_idx - 1] snk_par_idx = d["snk_par_idx"] snk_par_var = insts[0].params[snk_par_idx - 1] path = cls( src_sym_addr=int(d["src_sym_addr"], 0), src_sym_name=d["src_sym_name"], + src_par_idx=src_par_idx, + src_par_var=src_par_var, + src_inst_idx=d["src_inst_idx"], snk_sym_addr=int(d["snk_sym_addr"], 0), snk_sym_name=d["snk_sym_name"], snk_par_idx=snk_par_idx, snk_par_var=snk_par_var, - src_inst_idx=d["src_inst_idx"], - comment=d["comment"], - sha1_hash=d["sha1_hash"], insts=insts, call_graph=MediumLevelILFunctionGraph.from_dict(bv, d["call_graph"]), + comment=d["comment"], + sha1_hash=d["sha1_hash"], ) return path diff --git a/mole/core/slice.py b/mole/core/slice.py index ba31d1a..e9338b4 100644 --- a/mole/core/slice.py +++ b/mole/core/slice.py @@ -2,7 +2,7 @@ from __future__ import annotations from mole.common.help import FunctionHelper, InstructionHelper, VariableHelper from functools import lru_cache from mole.common.log import log -from typing import Any, Dict, Generator, List, Set, Tuple +from typing import Any, Dict, List, Set, Tuple import binaryninja as bn import networkx as nx @@ -56,6 +56,18 @@ class MediumLevelILInstructionGraph(nx.DiGraph): super().add_edge(from_inst, to_inst, **attr) return + def reverse(self) -> MediumLevelILInstructionGraph: + """ + This method returns a copy of the graph with the directions of edges reversed. All node and + edge attributes are preserved. + """ + reversed_graph = MediumLevelILInstructionGraph() + for node, attrs in self.nodes(data=True): + reversed_graph.add_node(node, **attrs) + for from_node, to_node, edge_attrs in self.edges(data=True): + reversed_graph.add_edge(to_node, from_node, **edge_attrs) + return reversed_graph + class MediumLevelILFunctionGraph(nx.DiGraph): """ @@ -103,9 +115,9 @@ class MediumLevelILFunctionGraph(nx.DiGraph): """ This method returns a copy of the graph. """ - call_graph = MediumLevelILFunctionGraph() - call_graph.update(self) - return call_graph + graph = MediumLevelILFunctionGraph() + graph.update(self) + return graph def to_dict(self) -> Dict: """ @@ -165,12 +177,15 @@ class MediumLevelILBackwardSlicer: """ self._bv: bn.BinaryView = bv self._tag = custom_tag if custom_tag else tag + self._origin = None + if "src" in self._tag.lower(): + self._origin = "src" + elif "snk" in self._tag.lower(): + self._origin = "snk" self._max_call_level: int = max_call_level self._inst_visited: Set[bn.MediumLevelILInstruction] = set() - self._inst_graph: MediumLevelILInstructionGraph = ( - MediumLevelILInstructionGraph() - ) - self._call_graph: MediumLevelILFunctionGraph = MediumLevelILFunctionGraph() + self.inst_graph: MediumLevelILInstructionGraph = MediumLevelILInstructionGraph() + self.call_graph: MediumLevelILFunctionGraph = MediumLevelILFunctionGraph() return def _slice_ssa_var_definition( @@ -191,15 +206,17 @@ class MediumLevelILBackwardSlicer: # Try finding the definition withing the current function inst_def = inst.function.get_ssa_var_definition(ssa_var) if inst_def: - self._inst_graph.add_node(inst, call_level, caller_site) - self._inst_graph.add_node(inst_def, call_level, caller_site) - self._inst_graph.add_edge(inst, inst_def) + self.inst_graph.add_node(inst, call_level, caller_site, origin=self._origin) + self.inst_graph.add_node( + inst_def, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_edge(inst, inst_def) self._slice_backwards(inst_def, call_level, caller_site) return # Try finding the definition in another function if self._max_call_level >= 0 and abs(call_level) > self._max_call_level: return - caller_level = self._call_graph.nodes.get(caller_site, {}).get( + caller_level = self.call_graph.nodes.get(caller_site, {}).get( "call_level", None ) for parm_idx, parm_var in enumerate( @@ -221,12 +238,16 @@ class MediumLevelILBackwardSlicer: self._tag, f"Follow parameter '{var_info:s}' to caller '{cs_info:s}'", ) - self._inst_graph.add_node(inst, call_level, caller_site) - self._inst_graph.add_node(cs_parm, call_level - 1, inst.function) - self._inst_graph.add_edge(inst, cs_parm) - self._call_graph.add_node(cs_inst.function, call_level - 1) - self._call_graph.add_node(inst.function, call_level) - self._call_graph.add_edge(cs_inst.function, inst.function) + self.inst_graph.add_node( + inst, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_node( + cs_parm, call_level - 1, inst.function, origin=self._origin + ) + self.inst_graph.add_edge(inst, cs_parm) + self.call_graph.add_node(cs_inst.function, call_level - 1) + self.call_graph.add_node(inst.function, call_level) + self.call_graph.add_edge(cs_inst.function, inst.function) self._slice_backwards(cs_parm, call_level - 1, inst.function) except Exception as _: continue @@ -289,13 +310,19 @@ class MediumLevelILBackwardSlicer: self._tag, f"Follow '{mem_def_inst_info:s}' since it uses '0x{inst.constant:x}'", ) - self._inst_graph.add_node( - inst, call_level, caller_site + self.inst_graph.add_node( + inst, + call_level, + caller_site, + origin=self._origin, ) - self._inst_graph.add_node( - mem_def_inst, call_level, caller_site + self.inst_graph.add_node( + mem_def_inst, + call_level, + caller_site, + origin=self._origin, ) - self._inst_graph.add_edge(inst, mem_def_inst) + self.inst_graph.add_edge(inst, mem_def_inst) self._slice_backwards( mem_def_inst, call_level, caller_site ) @@ -343,11 +370,16 @@ class MediumLevelILBackwardSlicer: self._tag, f"Follow '{mem_def_inst_info:s}' since it uses '{var_addr_ass_inst_info:s}'", ) - self._inst_graph.add_node(inst, call_level, caller_site) - self._inst_graph.add_node( - mem_def_inst, call_level, caller_site + self.inst_graph.add_node( + inst, call_level, caller_site, origin=self._origin ) - self._inst_graph.add_edge(inst, mem_def_inst) + self.inst_graph.add_node( + mem_def_inst, + call_level, + caller_site, + origin=self._origin, + ) + self.inst_graph.add_edge(inst, mem_def_inst) self._slice_backwards( mem_def_inst, call_level, caller_site ) @@ -373,9 +405,13 @@ class MediumLevelILBackwardSlicer: | bn.MediumLevelILFneg() | bn.MediumLevelILFloatConv() ): - self._inst_graph.add_node(inst, call_level, caller_site) - self._inst_graph.add_node(inst.src, call_level, caller_site) - self._inst_graph.add_edge(inst, inst.src) + self.inst_graph.add_node( + inst, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_node( + inst.src, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_edge(inst, inst.src) self._slice_backwards(inst.src, call_level, caller_site) case ( bn.MediumLevelILAdd() @@ -402,19 +438,31 @@ class MediumLevelILBackwardSlicer: | bn.MediumLevelILFdiv() | bn.MediumLevelILCmpUlt() ): - self._inst_graph.add_node(inst, call_level, caller_site) - self._inst_graph.add_node(inst.left, call_level, caller_site) - self._inst_graph.add_edge(inst, inst.left) + self.inst_graph.add_node( + inst, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_node( + inst.left, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_edge(inst, inst.left) self._slice_backwards(inst.left, call_level, caller_site) - self._inst_graph.add_node(inst, call_level, caller_site) - self._inst_graph.add_node(inst.right, call_level, caller_site) - self._inst_graph.add_edge(inst, inst.right) + self.inst_graph.add_node( + inst, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_node( + inst.right, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_edge(inst, inst.right) self._slice_backwards(inst.right, call_level, caller_site) case bn.MediumLevelILRet(): for ret in inst.src: - self._inst_graph.add_node(inst, call_level, caller_site) - self._inst_graph.add_node(ret, call_level, caller_site) - self._inst_graph.add_edge(inst, ret) + self.inst_graph.add_node( + inst, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_node( + ret, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_edge(inst, ret) self._slice_backwards(ret, call_level, caller_site) case ( bn.MediumLevelILSetVarSsa() @@ -423,9 +471,13 @@ class MediumLevelILBackwardSlicer: | bn.MediumLevelILSetVarSsaField() | bn.MediumLevelILSetVarSplitSsa() ): - self._inst_graph.add_node(inst, call_level, caller_site) - self._inst_graph.add_node(inst.src, call_level, caller_site) - self._inst_graph.add_edge(inst, inst.src) + self.inst_graph.add_node( + inst, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_node( + inst.src, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_edge(inst, inst.src) self._slice_backwards(inst.src, call_level, caller_site) case bn.MediumLevelILVarSplitSsa(): self._slice_ssa_var_definition(inst.high, inst, call_level, caller_site) @@ -464,20 +516,26 @@ class MediumLevelILBackwardSlicer: self._tag, f"Follow return instruction '{ret_info:s}' of function '{call_info:s}'", ) - self._inst_graph.add_node( - inst, call_level, caller_site + self.inst_graph.add_node( + inst, + call_level, + caller_site, + origin=self._origin, ) - self._inst_graph.add_node( - func_inst, call_level + 1, inst.function + self.inst_graph.add_node( + func_inst, + call_level + 1, + inst.function, + origin=self._origin, ) - self._inst_graph.add_edge(inst, func_inst) - self._call_graph.add_node( + self.inst_graph.add_edge(inst, func_inst) + self.call_graph.add_node( inst.function, call_level ) - self._call_graph.add_node( + self.call_graph.add_node( func, call_level + 1 ) - self._call_graph.add_edge( + self.call_graph.add_edge( inst.function, func ) self._slice_backwards( @@ -498,13 +556,19 @@ class MediumLevelILBackwardSlicer: self._tag, f"Follow parameter {par_idx:d} '{par_info:s}' of imported function '{call_info:s}'", ) - self._inst_graph.add_node( - inst, call_level, caller_site + self.inst_graph.add_node( + inst, + call_level, + caller_site, + origin=self._origin, ) - self._inst_graph.add_node( - par, call_level, caller_site + self.inst_graph.add_node( + par, + call_level, + caller_site, + origin=self._origin, ) - self._inst_graph.add_edge(inst, par) + self.inst_graph.add_edge(inst, par) self._slice_backwards( par, call_level, caller_site ) @@ -524,9 +588,13 @@ class MediumLevelILBackwardSlicer: self._tag, f"Follow parameter {par_idx:d} '{par_info:s}' of indirect function call '{call_info:s}'", ) - self._inst_graph.add_node(inst, call_level, caller_site) - self._inst_graph.add_node(par, call_level, caller_site) - self._inst_graph.add_edge(inst, par) + self.inst_graph.add_node( + inst, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_node( + par, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_edge(inst, par) self._slice_backwards(par, call_level, caller_site) case _: log.warn( @@ -535,9 +603,13 @@ class MediumLevelILBackwardSlicer: ) case bn.MediumLevelILSyscallSsa(): for par in inst.params: - self._inst_graph.add_node(inst, call_level, caller_site) - self._inst_graph.add_node(par, call_level, caller_site) - self._inst_graph.add_edge(inst, par) + self.inst_graph.add_node( + inst, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_node( + par, call_level, caller_site, origin=self._origin + ) + self.inst_graph.add_edge(inst, par) self._slice_backwards(par, call_level, caller_site) case _: log.warn(self._tag, f"[{call_level:+d}] {info:s}: Missing handler") @@ -551,59 +623,6 @@ class MediumLevelILBackwardSlicer: pass return - def get_insts(self) -> Generator[bn.MediumLevelILInstruction]: - """ - This method returns all sliced instructions. - """ - return self._inst_graph.nodes() - - def find_paths( - self, - snk_inst: bn.MediumLevelILInstruction, - src_inst: bn.MediumLevelILInstruction, - max_slice_depth: int, - ) -> List[Tuple[List[bn.MediumLevelILInstruction], MediumLevelILFunctionGraph]]: - """ - This method finds all simple paths from `snk_inst` to `src_inst`, with optionally limiting - path length by `max_slice_depth`. For each found path, the following is returned: First, a - list of instructions belonging to the path. And second, a function call graph, where nodes - and edges belonging to the path, have an attribute `in_path` set to `True`. - """ - paths = [] - # Find all simple paths - try: - if max_slice_depth is not None and max_slice_depth < 0: - max_slice_depth = None - simple_paths: List[List[bn.MediumLevelILInstruction]] = list( - nx.all_simple_paths( - self._inst_graph, snk_inst, src_inst, max_slice_depth - ) - ) - except (nx.NodeNotFound, nx.NetworkXNoPath): - return paths - - # Process all simple paths - for simple_path in simple_paths: - # Copy the call graph - call_graph = self._call_graph.copy() - # Add attribute `in_path = False` to all nodes - for node in call_graph.nodes(): - call_graph.nodes[node]["in_path"] = False - # Change attribute to `in_path = True` where functions are part of the path - for inst in simple_path: - func = inst.function - if func in call_graph: - call_graph.nodes[func]["in_path"] = True - # Add attribute `ìn_path` to edges where both nodes have `in_path = True` - for node_from, node_to in call_graph.edges(): - call_graph[node_from][node_to]["in_path"] = ( - call_graph.nodes[node_from]["in_path"] - and call_graph.nodes[node_to]["in_path"] - ) - # Add path and call graph - paths.append((simple_path, call_graph)) - return paths - @staticmethod @lru_cache(maxsize=None) def _get_mem_definitions( diff --git a/mole/models/path.py b/mole/models/path.py index c40bc5f..3c23486 100644 --- a/mole/models/path.py +++ b/mole/models/path.py @@ -11,13 +11,14 @@ PATH_COLS = { "Id": 0, "Src Addr": 1, "Src Func": 2, - "Snk Addr": 3, - "Snk Func": 4, - "Snk Parm": 5, - "Insts": 6, - "Phis": 7, - "Branches": 8, - "Comment": 9, + "Src Parm": 3, + "Snk Addr": 4, + "Snk Func": 5, + "Snk Parm": 6, + "Insts": 7, + "Phis": 8, + "Branches": 9, + "Comment": 10, } # Custom roles for tree items @@ -157,9 +158,9 @@ class PathTreeModel(qtui.QStandardItemModel): parent_item = self.group_items[internal_id] # Create path items - index_item = qtui.QStandardItem(f"{self.path_id:d}") - index_item.setData(self.path_id, PATH_ID_ROLE) - index_item.setData(True, IS_PATH_ITEM_ROLE) + id_item = qtui.QStandardItem(f"{self.path_id:d}") + id_item.setData(self.path_id, PATH_ID_ROLE) + id_item.setData(True, IS_PATH_ITEM_ROLE) # Only store hex values as UserRole data for proper sorting src_addr_item = qtui.QStandardItem(f"0x{path.src_sym_addr:x}") @@ -169,6 +170,11 @@ class PathTreeModel(qtui.QStandardItemModel): src_func_item = qtui.QStandardItem(path.src_sym_name) src_func_item.setData(True, IS_PATH_ITEM_ROLE) + src_parm_item = qtui.QStandardItem( + f"arg#{path.src_par_idx:d}:{str(path.src_par_var):s}" + ) + src_parm_item.setData(True, IS_PATH_ITEM_ROLE) + snk_addr_item = qtui.QStandardItem(f"0x{path.snk_sym_addr:x}") snk_addr_item.setData(path.snk_sym_addr, qtc.Qt.UserRole) snk_addr_item.setData(True, IS_PATH_ITEM_ROLE) @@ -181,8 +187,8 @@ class PathTreeModel(qtui.QStandardItemModel): ) snk_parm_item.setData(True, IS_PATH_ITEM_ROLE) - insts_item = qtui.QStandardItem(str(len(path.insts))) - insts_item.setData(True, IS_PATH_ITEM_ROLE) + inst_item = qtui.QStandardItem(str(len(path.insts))) + inst_item.setData(True, IS_PATH_ITEM_ROLE) phis_item = qtui.QStandardItem(str(len(path.phiis))) phis_item.setData(True, IS_PATH_ITEM_ROLE) @@ -195,13 +201,14 @@ class PathTreeModel(qtui.QStandardItemModel): # Set items as non-editable (except for comment) for item in [ - index_item, + id_item, src_addr_item, src_func_item, + src_parm_item, snk_addr_item, snk_func_item, snk_parm_item, - insts_item, + inst_item, phis_item, bdeps_item, ]: @@ -209,13 +216,14 @@ class PathTreeModel(qtui.QStandardItemModel): # Create path row and append to parent item (lowest level group) path_row = [ - index_item, + id_item, src_addr_item, src_func_item, + src_parm_item, snk_addr_item, snk_func_item, snk_parm_item, - insts_item, + inst_item, phis_item, bdeps_item, comment_item, diff --git a/mole/services/config.py b/mole/services/config.py index f025658..1e51825 100644 --- a/mole/services/config.py +++ b/mole/services/config.py @@ -202,7 +202,7 @@ class ConfigService: ) } ) - for name in ["highlight_color", "path_grouping"]: + for name in ["src_highlight_color", "snk_highlight_color", "path_grouping"]: setting = settings.get(name, None) if not setting: continue diff --git a/mole/views/config.py b/mole/views/config.py index b052a92..ea18609 100644 --- a/mole/views/config.py +++ b/mole/views/config.py @@ -147,7 +147,7 @@ class ConfigView(qtw.QWidget): pth_wid = qtw.QWidget() pth_lay = qtw.QFormLayout() - for name in ["highlight_color", "path_grouping"]: + for name in ["src_highlight_color", "snk_highlight_color", "path_grouping"]: setting: ComboboxSetting = self.config_ctr.get_setting(name) if not setting: continue diff --git a/mole/views/graph.py b/mole/views/graph.py index 826e431..46fe2d5 100644 --- a/mole/views/graph.py +++ b/mole/views/graph.py @@ -452,9 +452,9 @@ class GraphView(qtw.QGraphicsView): dest = self._nodes_map[b] self.scene().addItem(Edge(source, dest, self.get_node_color)) - # layout the graph + # Layout the graph self.layout() - # fit the view to the graph once animation is over + # Fit the view to the graph once animation is over self.animations.finished.connect(self.fit_to_window) return diff --git a/mole/views/path_tree.py b/mole/views/path_tree.py index 2f32b21..cec6ea2 100644 --- a/mole/views/path_tree.py +++ b/mole/views/path_tree.py @@ -370,7 +370,11 @@ class PathTreeView(qtw.QTreeView): path = self.get_path(path_id) if path: # Navigate to source address - if col in [PATH_COLS["Src Addr"], PATH_COLS["Src Func"]]: + if col in [ + PATH_COLS["Src Addr"], + PATH_COLS["Src Func"], + PATH_COLS["Src Parm"], + ]: vf.navigate(bv, path.src_sym_addr) # Navigate to sink address elif col in [ diff --git a/test/src/simple_http_server-04.c b/test/src/simple_http_server-04.c new file mode 100644 index 0000000..c378d74 --- /dev/null +++ b/test/src/simple_http_server-04.c @@ -0,0 +1,180 @@ +#include +#include +#include +#include +#include + +#define PORT 8080 +#define BUFFER_SIZE 256 + +/* +Testcase Description: +- server example +- disallow function inlining +- generate duplicates path (phis in process_post_request) +*/ + + +__attribute__ ((noinline)) +void process_post_request(int client_socket, const char *buffer); + +__attribute__ ((noinline)) +void handle_client(int client_socket); + +__attribute__ ((noinline)) +void execute_cgi_command(const char *buffer); + +__attribute__ ((noinline)) +void send_response(int client_socket, const char *response); + +__attribute__ ((noinline)) +int create_server_socket(struct sockaddr_in *address); + +__attribute__ ((noinline)) +void handle_post_request(int client_socket); + +__attribute__ ((noinline)) +char* receive_data(int client_socket, int *size); + +int main() { + int server_fd, client_socket; + struct sockaddr_in address; + int addrlen = sizeof(address); + + server_fd = create_server_socket(&address); + + while (1) { + if ((client_socket = accept(server_fd, (struct sockaddr *)&address, (socklen_t*)&addrlen)) < 0) { + perror("accept"); + close(server_fd); + exit(EXIT_FAILURE); + } + handle_client(client_socket); + } + + close(server_fd); + return 0; +} + +void handle_client(int client_socket) { + int size; + char *method = receive_data(client_socket, &size); + + if (method == NULL) { + close(client_socket); + return; + } + + if (strncmp(method, "POST", 4) == 0) { + handle_post_request(client_socket); + } else { + send_response(client_socket, "HTTP/1.1 405 Method Not Allowed\r\nContent-Type: text/plain\r\n\r\nMethod Not Allowed.\n"); + close(client_socket); + } + + free(method); +} + +void process_post_request(int client_socket, const char *buffer) { + // extract body from buffer + char *body = strstr(buffer, "\r\n\r\n"); + if (body) { + body += 4; // skip the "\r\n\r\n" + } else { + body = (char*)buffer; + } + while (1) { + + // for each line in body execute the command + char *line = strstr(body, "\r\n"); + if(line == NULL) { + break; + } + char* cmd = strstr(line, "EXECUTE"); + if (cmd) { + cmd += 8; // skip the "EXECUTE " + } else { + cmd = line; + } + execute_cgi_command(cmd); + + } +} + + +void handle_post_request(int client_socket) { + int size; + char *buffer = receive_data(client_socket, &size); + + if (buffer == NULL) { + close(client_socket); + return; + } + + process_post_request(client_socket, buffer); + send_response(client_socket, "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nPOST request received.\n"); + close(client_socket); + free(buffer); +} + +void execute_cgi_command(const char *buffer) { + char *cgi_start = strstr(buffer, "/cgi-bin/"); + if (cgi_start) { + cgi_start += strlen("/cgi-bin/"); + char *cgi_end = strchr(cgi_start, ' '); + if (cgi_end) { + *cgi_end = '\0'; + system(cgi_start); + } + } +} + +void send_response(int client_socket, const char *response) { + write(client_socket, response, strlen(response)); +} + +int create_server_socket(struct sockaddr_in *address) { + int server_fd; + + if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) { + perror("socket failed"); + exit(EXIT_FAILURE); + } + + address->sin_family = AF_INET; + address->sin_addr.s_addr = INADDR_ANY; + address->sin_port = htons(PORT); + + if (bind(server_fd, (struct sockaddr *)address, sizeof(*address)) < 0) { + perror("bind failed"); + close(server_fd); + exit(EXIT_FAILURE); + } + + if (listen(server_fd, 3) < 0) { + perror("listen"); + close(server_fd); + exit(EXIT_FAILURE); + } + + return server_fd; +} + +char* receive_data(int client_socket, int *size) { + char *buffer = (char *)malloc(BUFFER_SIZE); + if (buffer == NULL) { + perror("malloc"); + return NULL; + } + + int bytes_read = recv(client_socket, buffer, BUFFER_SIZE - 1, 0); + if (bytes_read < 0) { + perror("recv"); + free(buffer); + return NULL; + } + + buffer[bytes_read] = '\0'; + *size = bytes_read; + return buffer; +} \ No newline at end of file diff --git a/test/test_data.py b/test/test_data.py index dd51beb..98953fe 100644 --- a/test/test_data.py +++ b/test/test_data.py @@ -79,8 +79,8 @@ class TestData(unittest.TestCase): max_value=9999, help="maximum slice depth to stop the search", ), - "highlight_color": ComboboxSetting( - name="highlight_color", + "src_highlight_color": ComboboxSetting( + name="src_highlight_color", value="Red", items=[ "Blue", @@ -93,7 +93,23 @@ class TestData(unittest.TestCase): "White", "Black", ], - help="color used to highlight paths", + help="color used to highlight instructions originating from slicing a source function", + ), + "snk_highlight_color": ComboboxSetting( + name="snk_highlight_color", + value="Red", + items=[ + "Blue", + "Green", + "Cyan", + "Red", + "Magenta", + "Yellow", + "Orange", + "White", + "Black", + ], + help="color used to highlight instructions originating from slicing a sink function", ), }, ) @@ -217,21 +233,22 @@ class TestData(unittest.TestCase): return def test_serialize_combobox_settings(self) -> None: - setting = self.config.settings["highlight_color"] - # Serialize - yaml.safe_dump( - setting.to_dict(), - self.tf, - sort_keys=False, - default_style=None, - default_flow_style=None, - encoding="utf-8", - ) - # Deserialize - self.tf.seek(0) - ydoc = yaml.safe_load(self.tf) - # Assert - self.assertEqual(ydoc, setting, "Serialization error of 'ComboboxSetting'") + for name in ["src_highlight_color", "snk_highlight_color"]: + setting = self.config.settings.get(name, None) + # Serialize + yaml.safe_dump( + setting.to_dict(), + self.tf, + sort_keys=False, + default_style=None, + default_flow_style=None, + encoding="utf-8", + ) + # Deserialize + self.tf.seek(0) + ydoc = yaml.safe_load(self.tf) + # Assert + self.assertEqual(ydoc, setting, "Serialization error of 'ComboboxSetting'") return def tearDown(self) -> None: diff --git a/test/test_slicing.py b/test/test_slicing.py index f14eeec..5dd969b 100644 --- a/test/test_slicing.py +++ b/test/test_slicing.py @@ -69,8 +69,16 @@ class TestVarious(TestCase): self.assertEqual(path, Path.from_dict(bv, path.to_dict()), "serialization") self.assertIn(path.src_sym_name, ["gets"], "source has symbol 'gets'") self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["gets"], "sink has symbol 'gets'") self.assertTrue( @@ -83,7 +91,7 @@ class TestVarious(TestCase): self.assertEqual(path.snk_par_idx, 1, "arg1") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] self.assertEqual(calls, ["main"], "calls") @@ -106,8 +114,16 @@ class TestVarious(TestCase): ) self.assertIn(path.src_sym_name, ["gets"], "source has symbol 'gets'") self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertTrue( path.snk_sym_name in ["gets", "memcpy"], @@ -139,8 +155,16 @@ class TestVarious(TestCase): path.src_sym_name, ["getenv"], "source has symbol 'getenv'" ) self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn( path.snk_sym_name, @@ -157,7 +181,7 @@ class TestVarious(TestCase): self.assertEqual(path.snk_par_idx, 1, "arg1") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] self.assertEqual(calls, ["main"], "calls") @@ -177,8 +201,16 @@ class TestVarious(TestCase): self.assertEqual(path, Path.from_dict(bv, path.to_dict()), "serialization") self.assertIn(path.src_sym_name, ["getenv"], "source has symbol 'getenv'") self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["memcpy"], "sink has symbol 'memcpy'") self.assertTrue( @@ -191,7 +223,7 @@ class TestVarious(TestCase): self.assertEqual(path.snk_par_idx, 3, "arg3") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] self.assertEqual(calls, ["main"], "calls") @@ -215,8 +247,16 @@ class TestVarious(TestCase): path.src_sym_name, ["getenv"], "source has symbol 'getenv'" ) self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["memcpy"], "sink has symbol 'memcpy'") self.assertTrue( @@ -229,7 +269,7 @@ class TestVarious(TestCase): self.assertTrue(path.snk_par_idx in [2, 3], "arg2 or arg3") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] self.assertEqual(calls, ["main"], "calls") @@ -249,8 +289,16 @@ class TestVarious(TestCase): self.assertEqual(path, Path.from_dict(bv, path.to_dict()), "serialization") self.assertIn(path.src_sym_name, ["getenv"], "source has symbol 'getenv'") self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["memcpy"], "sink has symbol 'memcpy'") self.assertTrue( @@ -263,7 +311,7 @@ class TestVarious(TestCase): self.assertEqual(path.snk_par_idx, 1, "arg1") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] self.assertEqual(calls, ["main"], "calls") @@ -283,8 +331,16 @@ class TestVarious(TestCase): self.assertEqual(path, Path.from_dict(bv, path.to_dict()), "serialization") self.assertIn(path.src_sym_name, ["getenv"], "source has symbol 'getenv'") self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["memcpy"], "sink has symbol 'memcpy'") self.assertTrue( @@ -297,7 +353,7 @@ class TestVarious(TestCase): self.assertEqual(path.snk_par_idx, 3, "arg3") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] self.assertEqual(calls, ["main", "my_getenv"], "calls") @@ -321,8 +377,16 @@ class TestVarious(TestCase): path.src_sym_name, ["getenv"], "source has symbol 'getenv'" ) self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["memcpy"], "sink has symbol 'memcpy'") self.assertTrue( @@ -335,7 +399,7 @@ class TestVarious(TestCase): self.assertTrue(path.snk_par_idx in [2, 3], "arg2 or arg3") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] self.assertEqual(calls, ["main", "my_getenv"], "calls") @@ -392,8 +456,16 @@ class TestFunctionCalling(TestCase): path.src_sym_name, ["getenv"], "source has symbol 'getenv'" ) self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["system"], "sink has symbol 'system'") self.assertTrue( @@ -406,7 +478,7 @@ class TestFunctionCalling(TestCase): self.assertEqual(path.snk_par_idx, 1, "arg1") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] self.assertTrue("system_1b" not in calls, "system_1b not called") @@ -434,8 +506,16 @@ class TestFunctionCalling(TestCase): path.src_sym_name, ["getenv"], "source has symbol 'getenv'" ) self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["system"], "sink has symbol 'system'") self.assertTrue( @@ -448,7 +528,7 @@ class TestFunctionCalling(TestCase): self.assertEqual(path.snk_par_idx, 1, "arg1") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] call_paths.append(calls) @@ -488,8 +568,16 @@ class TestFunctionCalling(TestCase): self.assertEqual(path, Path.from_dict(bv, path.to_dict()), "serialization") self.assertIn(path.src_sym_name, ["getenv"], "source has symbol 'getenv'") self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["system"], "sink has symbol 'system'") self.assertTrue( @@ -502,7 +590,7 @@ class TestFunctionCalling(TestCase): self.assertEqual(path.snk_par_idx, 1, "arg1") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] self.assertEqual(calls, ["main", "func", "main"], "calls") @@ -550,8 +638,16 @@ class TestPointerAnalysis(TestCase): self.assertEqual(path, Path.from_dict(bv, path.to_dict()), "serialization") self.assertIn(path.src_sym_name, ["getenv"], "source has symbol 'getenv'") self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["system"], "sink has symbol 'system'") self.assertTrue( @@ -564,7 +660,7 @@ class TestPointerAnalysis(TestCase): self.assertEqual(path.snk_par_idx, 1, "arg1") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] self.assertEqual(calls, ["main"], "calls") @@ -600,8 +696,16 @@ class TestPointerAnalysis(TestCase): path.src_sym_name, ["getenv"], "source has symbol 'getenv'" ) self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["system"], "sink has symbol 'system'") self.assertTrue( @@ -646,8 +750,16 @@ class TestPointerAnalysis(TestCase): path.src_sym_name, ["getenv"], "source has symbol 'getenv'" ) self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["memcpy"], "sink has symbol 'memcpy'") self.assertTrue( @@ -674,8 +786,16 @@ class TestPointerAnalysis(TestCase): self.assertEqual(path, Path.from_dict(bv, path.to_dict()), "serialization") self.assertIn(path.src_sym_name, ["getenv"], "source has symbol 'getenv'") self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["memcpy"], "sink has symbol 'system'") self.assertTrue( @@ -688,7 +808,7 @@ class TestPointerAnalysis(TestCase): self.assertEqual(path.snk_par_idx, 2, "arg2") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] self.assertEqual(calls, ["main", "my_getenv"], "calls") @@ -730,8 +850,16 @@ class TestSimpleServer(TestCase): ) self.assertIn(path.src_sym_name, ["recv"], "source has symbol 'recv'") self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 2, "arg2") + self.assertTrue( + isinstance(path.src_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["system"], "sink has symbol 'system'") self.assertTrue( @@ -744,7 +872,7 @@ class TestSimpleServer(TestCase): self.assertEqual(path.snk_par_idx, 1, "arg1") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] call_paths.append(calls) @@ -773,8 +901,16 @@ class TestSimpleServer(TestCase): ) self.assertIn(path.src_sym_name, ["recv"], "source has symbol 'recv'") self.assertTrue( - isinstance(path.insts[-1], bn.MediumLevelILInstruction), - "source is a MLIL instruction", + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 2, "arg2") + self.assertTrue( + isinstance(path.src_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", ) self.assertIn(path.snk_sym_name, ["system"], "sink has symbol 'system'") self.assertTrue( @@ -787,7 +923,7 @@ class TestSimpleServer(TestCase): self.assertEqual(path.snk_par_idx, 1, "arg1") self.assertTrue( isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), - "argument is a MLIL variable", + "sink argument is a MLIL variable", ) calls = [call[1] for call in path.calls] call_paths.append(calls) @@ -802,6 +938,141 @@ class TestSimpleServer(TestCase): bv.file.close() return + def test_simple_http_server_03( + self, filenames: List[str] = ["simple_http_server-03"] + ) -> None: + for file in TestCase.load_files(filenames): + # Load and analyze test binary with Binary Ninja + bv = bn.load(file) + bv.update_analysis_and_wait() + # Analyze test binary + paths = self.get_paths(bv) + # Assert results + self.assertTrue(len(paths) == 4, "4 paths identified") + call_paths = [] + for path in paths: + self.assertEqual( + path, Path.from_dict(bv, path.to_dict()), "serialization" + ) + self.assertIn(path.src_sym_name, ["recv"], "source has symbol 'recv'") + self.assertTrue( + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 2, "arg2") + self.assertTrue( + isinstance(path.src_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", + ) + self.assertIn(path.snk_sym_name, ["system"], "sink has symbol 'system'") + self.assertTrue( + ( + isinstance(path.insts[0], bn.MediumLevelILCallSsa) + or isinstance(path.insts[0], bn.MediumLevelILTailcallSsa) + ), + "sink is a MLIL call instruction", + ) + self.assertEqual(path.snk_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "sink argument is a MLIL variable", + ) + calls = [call[1] for call in path.calls] + call_paths.append(calls) + self.assertCountEqual( + call_paths, + [ + [ + "execute_cgi_command", + "wrap_and_execute", + "process_request", + "handle_get_request", + "receive_data", + ], + [ + "execute_cgi_command", + "wrap_and_execute", + "process_request", + "handle_post_request", + "receive_data", + ], + [ + "execute_cgi_command", + "wrap_and_execute", + "process_request", + "handle_put_request", + "receive_data", + ], + [ + "execute_cgi_command", + "wrap_and_execute", + "process_request", + "handle_delete_request", + "receive_data", + ], + ], + "calls", + ) + bv.file.close() + return + + def test_simple_http_server_04( + self, filenames: List[str] = ["simple_http_server-04"] + ) -> None: + for file in TestCase.load_files(filenames): + # Load and analyze test binary with Binary Ninja + bv = bn.load(file) + bv.update_analysis_and_wait() + # Analyze test binary + paths = self.get_paths(bv) + # Assert results + self.assertTrue(len(paths) >= 1, "at least 1 path identified") + for path in paths: + self.assertEqual( + path, Path.from_dict(bv, path.to_dict()), "serialization" + ) + self.assertIn(path.src_sym_name, ["recv"], "source has symbol 'recv'") + self.assertTrue( + ( + isinstance(path.insts[-1], bn.MediumLevelILCallSsa) + or isinstance(path.insts[-1], bn.MediumLevelILTailcallSsa) + ), + "source is a MLIL call instruction", + ) + self.assertEqual(path.src_par_idx, 2, "arg2") + self.assertTrue( + isinstance(path.src_par_var, bn.MediumLevelILVarSsa), + "source argument is a MLIL variable", + ) + self.assertIn(path.snk_sym_name, ["system"], "sink has symbol 'system'") + self.assertTrue( + ( + isinstance(path.insts[0], bn.MediumLevelILCallSsa) + or isinstance(path.insts[0], bn.MediumLevelILTailcallSsa) + ), + "sink is a MLIL call instruction", + ) + self.assertEqual(path.snk_par_idx, 1, "arg1") + self.assertTrue( + isinstance(path.snk_par_var, bn.MediumLevelILVarSsa), + "sink argument is a MLIL variable", + ) + calls = [call[1] for call in path.calls] + self.assertEqual( + calls, + [ + "execute_cgi_command", + "process_post_request", + "handle_post_request", + "receive_data", + ], + ) + bv.file.close() + return + class TestMultiThreading(TestCase): def test_consistency_01(