156 slicing cancellation takes too long to respond (#161)

* Added contributor

* Added symbol names used by xcode

* Fixing code x-ref issue

* Fixes for PE files

* Improved function slicing

* Only log source/sink info if we process it

* Store all instructions at the code x-refs address

* Restructure function handling

* Pass canceled callable to MediumLevelILBackwardSlicer

* Pass canceled callable to MediumLevelILBackwardSlicer

* Improved cancellation

* 151 failed to find code references in macho binaries (#160)

* Added contributor

* Added symbol names used by xcode

* Fixing code x-ref issue

* Fixes for PE files

* Improved function slicing

* Only log source/sink info if we process it

* Store all instructions at the code x-refs address

* Restructure function handling

* Moved to class method

* Handle no founds found

* Improved function slicing

* Restructure function handling

* Pass canceled callable to MediumLevelILBackwardSlicer

* Improved cancellation

* Fixing follow_params

* Remove uneeded function follow_params

* Version update
This commit is contained in:
Damian Pfammatter
2025-06-27 20:58:43 +02:00
committed by GitHub
parent de6f59069c
commit 8cd94d484c
7 changed files with 40 additions and 41 deletions
+4 -4
View File
@@ -216,7 +216,7 @@ class PathController:
log.error(tag, f"Failed to load path #{i + 1:d}: {str(e):s}")
finally:
self._thread.progress = (
f"Mole loads paths: {i + 1:d}/{len(s_paths):d}"
f"Mole loaded path: {i + 1:d}/{len(s_paths):d}"
)
except KeyError:
pass
@@ -270,7 +270,7 @@ class PathController:
log.error(tag, f"Failed to save path #{i + 1:d}: {str(e):s}")
finally:
self._thread.progress = (
f"Mole saves paths: {i + 1:d}/{len(paths):d}"
f"Mole saved path: {i + 1:d}/{len(paths):d}"
)
self._bv.store_metadata("mole_paths", json.dumps(s_paths))
except Exception as e:
@@ -349,7 +349,7 @@ class PathController:
)
finally:
self._thread.progress = (
f"Mole imports paths: {i + 1:d}/{cnt_total_paths:d}"
f"Mole imported path: {i + 1:d}/{cnt_total_paths:d}"
)
except Exception as e:
log.error(tag, f"Failed to import paths: {str(e):s}")
@@ -429,7 +429,7 @@ class PathController:
)
finally:
self._thread.progress = (
f"Mole exports paths: {i + 1:d}/{len(path_ids):d}"
f"Mole exported path: {i + 1:d}/{len(path_ids):d}"
)
f.write("\n]")
except Exception as e:
+17 -14
View File
@@ -200,7 +200,7 @@ class SourceFunction(Function):
return False
return super().__eq__(other)
def find_targets(self, bv: bn.BinaryView, canceled: Callable[[], bool]) -> None:
def find_targets(self, bv: bn.BinaryView, cancelled: Callable[[], bool]) -> None:
"""
This method finds a set of target instructions that a static backward slice should hit on.
"""
@@ -211,11 +211,11 @@ class SourceFunction(Function):
code_refs = SymbolHelper.get_code_refs(bv, self.symbols)
# Iterate code references
for src_sym_name, src_insts in code_refs.items():
if canceled():
if cancelled():
break
# Iterate source instructions
for src_inst in src_insts:
if canceled():
if cancelled():
break
# Ignore everything but call instructions
if not (
@@ -241,7 +241,7 @@ class SourceFunction(Function):
)
# Iterate source instruction's parameters
for src_par_idx, src_par_var in enumerate(src_call_inst.params):
if canceled():
if cancelled():
break
src_par_idx += 1
log.debug(
@@ -272,7 +272,9 @@ class SourceFunction(Function):
)
continue
# Create backward slicer
src_slicer = MediumLevelILBackwardSlicer(bv, custom_tag, 0)
src_slicer = MediumLevelILBackwardSlicer(
bv, custom_tag, 0, cancelled
)
# Add edge between call and parameter instructions
src_slicer.inst_graph.add_node(
src_call_inst, 0, src_call_inst.function, origin="src"
@@ -285,7 +287,8 @@ class SourceFunction(Function):
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
if not cancelled():
src_par_map[(src_par_idx, src_par_var)] = src_slicer.inst_graph
return
@@ -310,7 +313,7 @@ class SinkFunction(Function):
max_call_level: int,
max_slice_depth: int,
found_path: Callable[[Path], None],
canceled: Callable[[], bool],
cancelled: Callable[[], bool],
) -> List[Path]:
"""
This method tries to find paths, starting from the current sink and ending in one of the
@@ -324,11 +327,11 @@ class SinkFunction(Function):
code_refs = SymbolHelper.get_code_refs(bv, self.symbols)
# Iterate code references
for snk_sym_name, snk_insts in code_refs.items():
if canceled():
if cancelled():
break
# Iterate sink instructions
for snk_inst in snk_insts:
if canceled():
if cancelled():
break
# Ignore everything but call instructions
if not (
@@ -351,7 +354,7 @@ class SinkFunction(Function):
continue
# Iterate sink instruction's parameters
for snk_par_idx, snk_par_var in enumerate(snk_call_inst.params):
if canceled():
if cancelled():
break
snk_par_idx += 1
log.debug(
@@ -385,7 +388,7 @@ class SinkFunction(Function):
if self.par_slice_fun(snk_par_idx):
# Create backward slicer
snk_slicer = MediumLevelILBackwardSlicer(
bv, custom_tag, max_call_level
bv, custom_tag, max_call_level, cancelled
)
snk_inst_graph = snk_slicer.inst_graph
snk_call_graph = snk_slicer.call_graph
@@ -402,7 +405,7 @@ class SinkFunction(Function):
snk_slicer.slice_backwards(snk_par_var)
# Iterate sources
for source in sources:
if canceled():
if cancelled():
break
# Iterate source instructions
for (
@@ -410,13 +413,13 @@ class SinkFunction(Function):
src_sym_name,
src_call_inst,
), src_par_map in source.src_map.items():
if canceled():
if cancelled():
break
# Iterate source instruction's parameters
for (src_par_idx, src_par_var), (
src_inst_graph
) in src_par_map.items():
if canceled():
if cancelled():
break
# Source parameter was not sliced
if not source.par_slice_fun(src_par_idx):
+13 -11
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, List, Set, Tuple
from typing import Any, Callable, Dict, List, Set, Tuple
import binaryninja as bn
import networkx as nx
@@ -158,7 +158,11 @@ class MediumLevelILBackwardSlicer:
"""
def __init__(
self, bv: bn.BinaryView, custom_tag: str = "", max_call_level: int = -1
self,
bv: bn.BinaryView,
custom_tag: str = "",
max_call_level: int = -1,
cancelled: Callable[[], bool] = None,
) -> None:
"""
This method initializes a backward slicer for for MLIL instructions.
@@ -171,6 +175,7 @@ class MediumLevelILBackwardSlicer:
elif "snk" in self._tag.lower():
self._origin = "snk"
self._max_call_level: int = max_call_level
self._cancelled = cancelled
self._inst_visited: Set[bn.MediumLevelILInstruction] = set()
self.inst_graph: MediumLevelILInstructionGraph = MediumLevelILInstructionGraph()
self.call_graph: MediumLevelILFunctionGraph = MediumLevelILFunctionGraph()
@@ -274,6 +279,8 @@ class MediumLevelILBackwardSlicer:
expected to be `inst`'s level within the call stack. Parameter `caller_site` is expected to
be the function that called `inst.function`.
"""
if self._cancelled and self._cancelled():
return
info = InstructionHelper.get_inst_info(inst)
# Maxium call level
if self._max_call_level >= 0 and abs(call_level) > self._max_call_level:
@@ -522,16 +529,16 @@ class MediumLevelILBackwardSlicer:
symb = func.source_function.symbol
for func_inst in func.instructions:
# TODO: Support all return instructions
func_inst_info = InstructionHelper.get_inst_info(
func_inst, False
)
match func_inst:
case (
bn.MediumLevelILRet()
| bn.MediumLevelILTailcallSsa()
):
# Function
if symb.type == bn.SymbolType.FunctionSymbol:
if symb.type in [
bn.SymbolType.FunctionSymbol,
bn.SymbolType.LibraryFunctionSymbol,
]:
ret_info = InstructionHelper.get_inst_info(
func_inst, False
)
@@ -579,11 +586,6 @@ class MediumLevelILBackwardSlicer:
self._tag,
f"Function '{call_info:s}' has an unexpected type '{str(symb.type):s}'",
)
case _:
log.warn(
self._tag,
f"[{call_level:+d}] {func_inst_info:s}: Missing handler for function return instruction",
)
# Indirect function calls
case bn.MediumLevelILVarSsa():
self._slice_params(inst, call_level, caller_site)
-2
View File
@@ -410,8 +410,6 @@ Be proactive in exploring upstream paths, analyzing data/control dependencies, a
tasks[task] = path_id
# Wait for tasks to complete
for cnt, task in enumerate(futures.as_completed(tasks)):
if self.cancelled:
break
self.progress = f"Mole analyzed path {cnt + 1:d}/{len(self._paths):d}"
path_id = tasks[task]
# Collect vulnerability reports from task results
+4 -8
View File
@@ -106,12 +106,9 @@ class PathService(BackgroundTask):
)
)
# Wait for tasks to complete
self.progress = f"Mole processes {len(tasks):d} source functions"
for cnt, _ in enumerate(futures.as_completed(tasks)):
if self.cancelled:
break
self.progress = (
f"Mole processed source {cnt + 1:d}/{len(src_funs):d}"
)
self.progress = f"Mole processed source {cnt + 1:d}/{len(tasks):d}"
# Backward slice sink functions
with futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# Submit tasks
@@ -131,10 +128,9 @@ class PathService(BackgroundTask):
)
)
# Wait for tasks to complete and collect paths
self.progress = f"Mole processes {len(tasks):d} sink functions"
for cnt, task in enumerate(futures.as_completed(tasks)):
if self.cancelled:
break
self.progress = f"Mole processed sink {cnt + 1:d}/{len(snk_funs):d}"
self.progress = f"Mole processed sink {cnt + 1:d}/{len(tasks):d}"
# Collect paths from task results
if task.done() and not task.exception():
paths = task.result()
+1 -1
View File
@@ -24,7 +24,7 @@
"Linux": "",
"Windows": ""
},
"version": "0.2.0",
"version": "0.2.1",
"author": "Damian Pfammatter and Sergio Paganoni",
"minimumbinaryninjaversion": 6455
}
+1 -1
View File
@@ -11,7 +11,7 @@ include = ["mole*"]
[project]
name = "mole"
version = "0.2.0"
version = "0.2.1"
description = "A Binary Ninja plugin to identify interesting paths using static backward slicing"
authors = [
{name = "Damian Pfammatter"},