From f5ec6dc7dd5fbf4785007365fd89e1d09304afd5 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Mon, 5 May 2025 12:55:48 +0100 Subject: [PATCH 1/4] Add in initial version of breakpointing --- volatility3/cli/volshell/generic.py | 32 +++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/volatility3/cli/volshell/generic.py b/volatility3/cli/volshell/generic.py index b9811e7d5..7ff30909f 100644 --- a/volatility3/cli/volshell/generic.py +++ b/volatility3/cli/volshell/generic.py @@ -3,6 +3,7 @@ # import binascii import code +import functools import io import random import string @@ -187,6 +188,7 @@ class Volshell(interfaces.plugins.PluginInterface): def construct_locals(self) -> List[Tuple[List[str], Any]]: """Returns a listing of the functions to be added to the environment.""" return [ + (["bp", "breakpoint"], self.breakpoint), (["dt", "display_type"], self.display_type), (["db", "display_bytes"], self.display_bytes), (["dw", "display_words"], self.display_words), @@ -797,6 +799,36 @@ class Volshell(interfaces.plugins.PluginInterface): return constructed + def breakpoint(self, address: int, layer_name: Optional[str] = None) -> None: + """Sets a breakpoint on a particular address (within a specific layer)""" + if layer_name is None: + if self.current_layer is None: + raise ValueError("Current layer must be set") + layer_name = self.current_layer + + layer: interfaces.layers.DataLayerInterface = self.context.layers[layer_name] + # Check if the read value is already overloaded + if not hasattr(layer.read, "breakpoints"): + # Layer read is not yet wrapped + def wrapped_read(offset: int, length: int, pad: bool = False) -> bytes: + original_read = getattr(wrapped_read, "original_read") + for breakpoint in getattr(wrapped_read, "breakpoints"): + if (offset <= breakpoint) and (breakpoint < offset + length): + import pdb + + pdb.set_trace() + print("Hit breakpoint") + return original_read(offset, length, pad) + + setattr(wrapped_read, "breakpoints", set()) + setattr(wrapped_read, "original_read", layer.read) + setattr(layer, "read", wrapped_read) + + # Add the new breakpoint + breakpoints = getattr(layer.read, "breakpoints") + breakpoints.add(address) + setattr(layer.read, "breakpoints", breakpoints) + class NullFileHandler(io.BytesIO, interfaces.plugins.FileHandlerInterface): """Null FileHandler that swallows files whole without consuming memory""" From ae08c8ecfc8dd85f3c33af3a4307dd1cd86df243 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Mon, 5 May 2025 13:14:19 +0100 Subject: [PATCH 2/4] Support setting the breakpoint on the lowest layer --- volatility3/cli/volshell/generic.py | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/volatility3/cli/volshell/generic.py b/volatility3/cli/volshell/generic.py index 7ff30909f..20fbae417 100644 --- a/volatility3/cli/volshell/generic.py +++ b/volatility3/cli/volshell/generic.py @@ -799,7 +799,9 @@ class Volshell(interfaces.plugins.PluginInterface): return constructed - def breakpoint(self, address: int, layer_name: Optional[str] = None) -> None: + def breakpoint( + self, address: int, layer_name: Optional[str] = None, lowest: bool = False + ) -> None: """Sets a breakpoint on a particular address (within a specific layer)""" if layer_name is None: if self.current_layer is None: @@ -807,6 +809,18 @@ class Volshell(interfaces.plugins.PluginInterface): layer_name = self.current_layer layer: interfaces.layers.DataLayerInterface = self.context.layers[layer_name] + + if lowest: + while isinstance(layer, interfaces.layers.TranslationLayerInterface): + mapping = layer.mapping(address, 1) + if not mapping: + raise ValueError( + "Offset cannot be mapped lower, cannot break at lowest layer" + ) + _, _, mapped_offset, _, mapped_layer_name = next(mapping) + layer = self.context.layers[mapped_layer_name] + address = mapped_offset + # Check if the read value is already overloaded if not hasattr(layer.read, "breakpoints"): # Layer read is not yet wrapped @@ -825,6 +839,7 @@ class Volshell(interfaces.plugins.PluginInterface): setattr(layer, "read", wrapped_read) # Add the new breakpoint + print(f"Setting breakpoint {address:x} on {layer.name}") breakpoints = getattr(layer.read, "breakpoints") breakpoints.add(address) setattr(layer.read, "breakpoints", breakpoints) From 5bb4cd90c16da4b2248a97ad80a04f0b852ecf51 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Mon, 5 May 2025 13:16:29 +0100 Subject: [PATCH 3/4] Remove no longer needed functools import --- volatility3/cli/volshell/generic.py | 1 - 1 file changed, 1 deletion(-) diff --git a/volatility3/cli/volshell/generic.py b/volatility3/cli/volshell/generic.py index 20fbae417..4c90c39c9 100644 --- a/volatility3/cli/volshell/generic.py +++ b/volatility3/cli/volshell/generic.py @@ -3,7 +3,6 @@ # import binascii import code -import functools import io import random import string From 6269091be761e3906ab184665d33cd3a85cdf1f5 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Sun, 18 May 2025 20:08:25 +0100 Subject: [PATCH 4/4] CLI: Support listing/clearing breakpoints in volshell --- volatility3/cli/volshell/generic.py | 44 +++++++++++++++++++++++++++-- 1 file changed, 42 insertions(+), 2 deletions(-) diff --git a/volatility3/cli/volshell/generic.py b/volatility3/cli/volshell/generic.py index 4c90c39c9..e68b0334e 100644 --- a/volatility3/cli/volshell/generic.py +++ b/volatility3/cli/volshell/generic.py @@ -187,6 +187,8 @@ class Volshell(interfaces.plugins.PluginInterface): def construct_locals(self) -> List[Tuple[List[str], Any]]: """Returns a listing of the functions to be added to the environment.""" return [ + (["bc", "breakpoint_clear"], self.breakpoint_clear), + (["bl", "breakpoint_list"], self.breakpoint_list), (["bp", "breakpoint"], self.breakpoint), (["dt", "display_type"], self.display_type), (["db", "display_bytes"], self.display_bytes), @@ -827,10 +829,13 @@ class Volshell(interfaces.plugins.PluginInterface): original_read = getattr(wrapped_read, "original_read") for breakpoint in getattr(wrapped_read, "breakpoints"): if (offset <= breakpoint) and (breakpoint < offset + length): + print( + "Hit breakpoint, entering python debugger. To continue running without the debugger use the command continue" + ) import pdb pdb.set_trace() - print("Hit breakpoint") + _ = "First statement after the breakpoint, use u(p), d(own) and list to navigate through the execution frames" return original_read(offset, length, pad) setattr(wrapped_read, "breakpoints", set()) @@ -838,11 +843,46 @@ class Volshell(interfaces.plugins.PluginInterface): setattr(layer, "read", wrapped_read) # Add the new breakpoint - print(f"Setting breakpoint {address:x} on {layer.name}") + print(f"Setting breakpoint {address:#x} on {layer.name}") breakpoints = getattr(layer.read, "breakpoints") breakpoints.add(address) setattr(layer.read, "breakpoints", breakpoints) + def breakpoint_list(self, layer_names: Optional[List[str]] = None): + """List available breakpoints for a set of layers""" + if not layer_names: + layer_names = [layer_name for layer_name in self.context.layers] + + print("Listing breakpoints:") + for layer_name in layer_names: + print(f" {layer_name}") + layer = self.context.layers.get(layer_name, None) + if layer and hasattr(layer.read, "breakpoints"): + for breakpoint in layer.read.breakpoints: + print(f" {breakpoint:#x}") + + def breakpoint_clear( + self, offset: Optional[int] = None, layer_name: Optional[str] = None + ): + """Clears a offset breakpoint on a layer (or all breakpoints if offset or layer not specified) + + Args: + offset: Address of the breakpoint to clear (or all if None) + layer_name: Layer to clear breakpoints from (or all if None) + """ + print("Clearing breakpoints:") + for candidate_layer_name in self.context.layers: + candidate_layer = self.context.layers[candidate_layer_name] + if layer_name is None or layer_name == candidate_layer_name: + print(f" {candidate_layer_name}") + if hasattr(candidate_layer.read, "breakpoints"): + breakpoints_to_remove = set() + for breakpoint in candidate_layer.read.breakpoints: + if offset is None or offset == breakpoint: + print(f" clearing {breakpoint:#x}") + breakpoints_to_remove.add(breakpoint) + candidate_layer.read.breakpoints -= breakpoints_to_remove + class NullFileHandler(io.BytesIO, interfaces.plugins.FileHandlerInterface): """Null FileHandler that swallows files whole without consuming memory"""