From 9d253f6e7ef365634d11eaa8d2a8e98f507902ae Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Mon, 16 Sep 2019 01:00:03 +0100 Subject: [PATCH] Develop volshell with display_* functions. --- volatility/cli/volshell/shellplugin.py | 154 +++++++++++++++++++++---- volatility/cli/volshell/windows.py | 81 +++++-------- 2 files changed, 159 insertions(+), 76 deletions(-) diff --git a/volatility/cli/volshell/shellplugin.py b/volatility/cli/volshell/shellplugin.py index 398a66c81..289ad252b 100644 --- a/volatility/cli/volshell/shellplugin.py +++ b/volatility/cli/volshell/shellplugin.py @@ -1,18 +1,31 @@ # This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0 # which is available at https://www.volatilityfoundation.org/license/vsl_v1.0 # - +import binascii import code -import inspect -from typing import Any, Callable, Dict, List +import struct +import sys +from typing import Any, Dict, List, Optional from volatility.framework import renderers, interfaces from volatility.framework.configuration import requirements +from volatility.framework.layers import intel + +try: + import capstone + + has_capstone = True +except ImportError: + has_capstone = False class Volshell(interfaces.plugins.PluginInterface): """Shell environment to directly interact with a memory image.""" + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.__current_layer = None # type: Optional[str] + @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: return [ @@ -27,23 +40,7 @@ class Volshell(interfaces.plugins.PluginInterface): Return a TreeGrid but this is always empty since the point of this plugin is to run interactively """ - # Provide some OS-agnostic convenience elements for ease - context = self.context - config = self.config - layer_name = self.config['primary'] - kvo = context.layers[layer_name].config.get('kernel_virtual_offset') - members = lambda x: list(sorted(x.vol.members)) - - # Determine locals - curframe = inspect.currentframe() - vars = {} # type: Dict[str, Any] - if curframe: - vars = curframe.f_globals.copy() - vars.update(curframe.f_locals) - if additional_locals is not None: - vars.update(additional_locals) - - vars.update(self.load_functions()) + self._current_layer = self.config['primary'] # Try to enable tab completion try: @@ -52,21 +49,128 @@ class Volshell(interfaces.plugins.PluginInterface): pass else: import rlcompleter - completer = rlcompleter.Completer(namespace = vars) + completer = rlcompleter.Completer(namespace = self.construct_locals()) readline.set_completer(completer.complete) readline.parse_and_bind("tab: complete") print("Readline imported successfully") # TODO: provide help, consider generic functions (pslist?) and/or providing windows/linux functions - code.interact(local = vars) + sys.ps1 = "({}) >>> ".format(self.current_layer) + code.interact(local = self.construct_locals()) - return renderers.TreeGrid([], None) + return renderers.TreeGrid([("Terminating", str)], None) - def load_functions(self) -> Dict[str, Callable]: + def construct_locals(self) -> Dict[str, Any]: """Returns a dictionary listing the functions to be added to the environment.""" - return {"dt": self.display_type} + return { + 'dt': self.display_type, + 'display_type': self.display_type, + 'db': self.display_bytes, + 'display_bytes': self.display_bytes, + 'dw': self.display_words, + 'display_words': self.display_words, + 'dd': self.display_doublewords, + 'display_doublewords': self.display_doublewords, + 'dq': self.display_quadwords, + 'display_quadwords': self.display_quadwords, + 'dis': self.disassemble, + 'disassemble': self.disassemble, + 'cl': self.change_layer, + 'change_layer': self.change_layer, + 'context': self.context, + 'self': self + } + + def _read_data(self, offset, count = 128, layer_name = None): + """Reads the bytes necessary for the display_* methods""" + return self.context.layers[layer_name or self.current_layer].read(offset, count) + + def _display_data(self, offset: int, remaining_data: bytes, format_string: str = "B", ascii: bool = True): + """Display a series of bytes""" + chunk_size = struct.calcsize(format_string) + data_length = len(remaining_data) + remaining_data = remaining_data[:data_length - (data_length % chunk_size)] + + while remaining_data: + current_line, remaining_data = remaining_data[:16], remaining_data[16:] + offset += 16 + + data_blocks = [current_line[chunk_size * i:chunk_size * (i + 1)] for i in range(16 // chunk_size)] + data_blocks = [x for x in data_blocks if x != b''] + valid_data = [("{:0" + str(2 * chunk_size) + "x}").format(struct.unpack(format_string, x)[0]) + for x in data_blocks] + padding_data = [" " * 2 * chunk_size for _ in range((16 - len(current_line)) // chunk_size)] + hex_data = " ".join(valid_data + padding_data) + + ascii_data = "" + if ascii: + connector = " " + if chunk_size < 2: + connector = "" + ascii_data = connector.join([self._ascii_bytes(x) for x in valid_data]) + + print(hex(offset), " ", hex_data, " ", ascii_data) + + @staticmethod + def _ascii_bytes(bytes): + """Converts bytes into an ascii string""" + return "".join([chr(x) if 32 < x < 127 else '.' for x in binascii.unhexlify(bytes)]) + + @property + def current_layer(self): + return self._current_layer + + def change_layer(self, layer_name = None): + """Changes the current default layer""" + if not layer_name: + layer_name = self.config['primary'] + self._current_layer = layer_name + sys.ps1 = "({}) >>> ".format(self.current_layer) + + def display_bytes(self, offset, count = 128, layer_name = None): + """Displays byte values and ASCII characters. Each display line shows the address of the + first byte in the line, followed by up to 16 hexadecimal byte values. The byte + values are immediately followed by the corresponding ASCII values. + """ + remaining_data = self._read_data(offset, count = count, layer_name = layer_name) + self._display_data(offset, remaining_data) + + def display_quadwords(self, offset, count = 128, layer_name = None): + """Displays quad-word values (8 bytes) and corresponding ASCII characters""" + remaining_data = self._read_data(offset, count = count, layer_name = layer_name) + self._display_data(offset, remaining_data, format_string = "Q") + + def display_doublewords(self, offset, count = 128, layer_name = None): + """Displays double-word values (4 bytes) and corresponding ASCII characters""" + remaining_data = self._read_data(offset, count = count, layer_name = layer_name) + self._display_data(offset, remaining_data, format_string = "I") + + def display_words(self, offset, count = 128, layer_name = None): + """Displays word values (2 bytes) and corresponding ASCII characters""" + remaining_data = self._read_data(offset, count = count, layer_name = layer_name) + self._display_data(offset, remaining_data, format_string = "H") + + def disassemble(self, offset, count = 128, layer_name = None, architecture = None): + """Disassembles a number of instructions from the code at offset""" + remaining_data = self._read_data(offset, count = count, layer_name = layer_name) + if not has_capstone: + print("Capstone not available - please install it to use the disassemble command") + else: + if isinstance(self.context.layers[layer_name or self.current_layer], intel.Intel32e): + architecture = 'intel64' + elif isinstance(self.context.layers[layer_name or self.current_layer], intel.Intel): + architecture = 'intel' + disasm_types = { + 'intel': capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_32), + 'intel64': capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_64), + 'arm': capstone.Cs(capstone.CS_ARCH_ARM, capstone.CS_MODE_ARM), + 'arm64': capstone.Cs(capstone.CS_ARCH_ARM64, capstone.CS_MODE_ARM) + } + if architecture is not None: + for i in disasm_types[architecture].disasm(remaining_data, offset): + print("0x%x:\t%s\t%s" % (i.address, i.mnemonic, i.op_str)) @staticmethod def display_type(object: interfaces.objects.ObjectInterface): diff --git a/volatility/cli/volshell/windows.py b/volatility/cli/volshell/windows.py index f6bbe1047..54114f571 100644 --- a/volatility/cli/volshell/windows.py +++ b/volatility/cli/volshell/windows.py @@ -2,72 +2,51 @@ # which is available at https://www.volatilityfoundation.org/license/vsl_v1.0 # -import inspect -from typing import Callable, Dict +from typing import Dict, Any from volatility.cli.volshell import shellplugin from volatility.framework.configuration import requirements +from volatility.plugins.windows import pslist class Volshell(shellplugin.Volshell): """Shell environment to directly interact with a windows memory image.""" + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + @classmethod def get_requirements(cls): return (super().get_requirements() + [ requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"), + requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)), requirements.IntRequirement(name = 'pid', description = "Process ID", optional = True) ]) + def change_process(self, pid = None): + """Change the current process and layer, based on a process ID""" + processes = self.list_processes() + for process in processes: + if process.UniqueProcessId == pid: + process_layer = process.add_process_layer() + self.change_layer(process_layer) + return + print("No process with process ID {} found".format(pid)) + def list_processes(self): - """Lists all the processes in the primary layer.""" + """Returns a list of EPROCESS objects from the primary layer""" + # We always use the main kernel memory and the symbols + return list(pslist.PsList.list_processes(self.context, self.config['primary'], self.config['nt_symbols'])) - # We only use the object factory to demonstrate how to use one - layer_name = self.config['primary'] - kvo = self.context.layers[layer_name].config['kernel_virtual_offset'] - ntkrnlmp = self.context.module(self.config['nt_symbols'], layer_name = layer_name, offset = kvo) - - ps_aph_offset = ntkrnlmp.get_symbol("PsActiveProcessHead").address - list_entry = ntkrnlmp.object(object_type = "_LIST_ENTRY", offset = ps_aph_offset) - - # This is example code to demonstrate how to use symbol_space directly, rather than through a module: - # - # ``` - # reloff = self.context.symbol_space.get_type( - # self.config['nt_symbols'] + constants.BANG + "_EPROCESS").relative_child_offset( - # "ActiveProcessLinks") - # ``` - # - # Note: "nt!_EPROCESS" could have been used, but would rely on the "nt" symbol table not already - # having been present. Strictly, the value of the requirement should be joined with the BANG character - # defined in the constants file - reloff = ntkrnlmp.get_type("_EPROCESS").relative_child_offset("ActiveProcessLinks") - eproc = ntkrnlmp.object(object_type = "_EPROCESS", offset = list_entry.vol.offset - reloff, absolute = True) - - for proc in eproc.ActiveProcessLinks: - yield proc - - def load_functions(self) -> Dict[str, Callable]: - result = super().load_functions() - result.update({'ps': lambda: list(self.list_processes())}) + def construct_locals(self) -> Dict[str, Any]: + result = super().construct_locals() + result.update({ + 'cp': self.change_process, + 'change_process': self.change_process, + 'ps': self.list_processes, + 'list_processes': self.list_processes, + 'symbols': self.context.symbol_space[self.config['nt_symbols']] + }) + if self.config.get('pid', None) is not None: + self.change_process(self.config['pid']) return result - - def run(self, additional_locals = None): - # Determine locals - curframe = inspect.currentframe() - - # Provide some OS-agnostic convenience elements for ease - layer_name = self.config['primary'] - kvo = self.context.layers[layer_name].config['kernel_virtual_offset'] - nt = self.context.module(self.config['nt_symbols'], layer_name = layer_name, offset = kvo) - ps = lambda: list(self.list_processes()) - - pid = self.config.get('pid', None) - eproc = None - if pid: - for _x in ps(): - if _x.UniqueProcessId == pid: - eproc = _x - break - - return super().run(curframe.f_locals)