admin/cyber-defence-campus-mole
1
0
Fork 0
mirror of https://github.com/cyber-defence-campus/mole synced 2026-06-20 13:19:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

58 Wrong Instructions For Path Near The Source Call (#104)

Browse Source 
* 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 <sergio.paganoni@gmail.com>
This commit is contained in:
Damian Pfammatter
2025-04-09 13:01:44 +02:00
committed by GitHub
co-authored by wizche
parent d346504415
commit c5b5578382
12 changed files with 1072 additions and 395 deletions
Download Patch File Download Diff File Expand all files Collapse all files
conf/001-settings.yml
+17 -3
View File
@@ -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
mole/controllers/path.py
+41 -14
View File
@@ -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:
mole/core/data.py
+314 -177
View File
@@ -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
mole/core/slice.py
+133 -114
View File
@@ -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(
mole/models/path.py
+24 -16
View File
@@ -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,
mole/services/config.py
+1 -1
View File
@@ -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
mole/views/config.py
+1 -1
View File
@@ -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
mole/views/graph.py
+2 -2
View File
@@ -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
mole/views/path_tree.py
+5 -1
View File
@@ -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 [
test/src/simple_http_server-04.c
+180
View File
@@ -0,0 +1,180 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#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;
}
test/test_data.py
+35 -18
View File
@@ -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:
test/test_slicing.py
+319 -48
View File
@@ -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(