diff --git a/volatility3/cli/volshell/__init__.py b/volatility3/cli/volshell/__init__.py index 769e958fd..52bb5bde5 100644 --- a/volatility3/cli/volshell/__init__.py +++ b/volatility3/cli/volshell/__init__.py @@ -7,11 +7,13 @@ import json import logging import os import sys +from typing import Any, List, Tuple import volatility3.plugins import volatility3.symbols from volatility3 import cli, framework from volatility3.cli.volshell import generic, linux, mac, windows +from volatility3.cli.volshell.generic import VolshellShellPlugin from volatility3.framework import automagic, constants, contexts, exceptions, interfaces, plugins # Make sure we log everything @@ -260,6 +262,17 @@ class VolShell(cli.CommandLine): parser.exit(1, f"Unable to validate the plugin requirements: {[x for x in excp.unsatisfied]}\n") +class VolshellPlugin: + def __init__(self, parent: VolshellShellPlugin): + super().__init__() + self._parent = parent + + def construct_locals(self) -> List[Tuple[List[str], Any]]: + """Returns a list of tuples that contain a list of aliases and what those aliases point to + """ + return [] + + def main(): """A convenience function for constructing and running the :class:`CommandLine`'s run method.""" diff --git a/volatility3/cli/volshell/generic.py b/volatility3/cli/volshell/generic.py index 19e263a03..de6309c13 100644 --- a/volatility3/cli/volshell/generic.py +++ b/volatility3/cli/volshell/generic.py @@ -4,6 +4,7 @@ import binascii import code import io +import logging import random import string import struct @@ -11,11 +12,14 @@ import sys from typing import Any, Dict, Iterable, List, Optional, Tuple, Type, Union from urllib import parse, request +from volatility3 import framework from volatility3.cli import text_renderer, volshell from volatility3.framework import exceptions, interfaces, objects, plugins, renderers from volatility3.framework.configuration import requirements from volatility3.framework.layers import intel, physical, resources +vollog = logging.getLogger(__name__) + try: import capstone @@ -24,19 +28,7 @@ except ImportError: has_capstone = False -class Volshell(interfaces.plugins.PluginInterface): - """Shell environment to directly interact with a memory image.""" - _required_framework_version = (2, 0, 0) - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.__current_layer: Optional[str] = None - self.__current_symbol_table: Optional[str] = None - self.__current_kernel_name: Optional[str] = None - self.__console = None - - def random_string(self, length: int = 32) -> str: - return ''.join(random.sample(string.ascii_uppercase + string.digits, length)) +class VolshellShellPlugin(interfaces.plugins.PluginInterface): @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: @@ -52,6 +44,88 @@ class Volshell(interfaces.plugins.PluginInterface): requirements.TranslationLayerRequirement(name = 'primary', description = 'Memory layer for the kernel'), ] + def run(self) -> interfaces.renderers.TreeGrid: + pass + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._current_layer: Optional[str] = None + self._current_symbol_table: Optional[str] = None + self._current_kernel_name: Optional[str] = None + self._console = None + self._plugins = [] + + def random_string(self, length: int = 32) -> str: + return ''.join(random.sample(string.ascii_uppercase + string.digits, length)) + + @property + def current_layer(self): + if self._current_layer is None: + self._current_layer = self.config['primary'] + return self._current_layer + + @property + def current_symbol_table(self): + if self._current_symbol_table is None and self.kernel: + self._current_symbol_table = self.kernel.symbol_table_name + return self._current_symbol_table + + @property + def current_kernel_name(self): + if self._current_kernel_name is None: + self._current_kernel_name = self.config.get('kernel', None) + return self._current_kernel_name + + @property + def kernel(self): + """Returns the current kernel object""" + if self.current_kernel_name not in self.context.modules: + return None + return self.context.modules[self.current_kernel_name] + + def change_layer(self, layer_name: str = None): + """Changes the current default layer""" + if not layer_name: + layer_name = self.current_layer + if layer_name not in self.context.layers: + print(f"Layer {layer_name} not present in context") + else: + self._current_layer = layer_name + sys.ps1 = f"({self.current_layer}) >>> " + + def generate_treegrid(self, plugin: Type[interfaces.plugins.PluginInterface], + **kwargs) -> Optional[interfaces.renderers.TreeGrid]: + """Generates a TreeGrid based on a specific plugin passing in kwarg configuration values""" + path_join = interfaces.configuration.path_join + + # Generate a temporary configuration path + plugin_config_suffix = self.random_string() + plugin_path = path_join(self.config_path, plugin_config_suffix) + + # Populate the configuration + for name, value in kwargs.items(): + self.config[path_join(plugin_config_suffix, plugin.__name__, name)] = value + + try: + constructed = plugins.construct_plugin(self.context, [], plugin, plugin_path, None, NullFileHandler) + return constructed.run() + except exceptions.UnsatisfiedException as excp: + print(f"Unable to validate the plugin requirements: {[x for x in excp.unsatisfied]}\n") + return None + + def render_treegrid(self, + treegrid: interfaces.renderers.TreeGrid, + renderer: Optional[interfaces.renderers.Renderer] = None) -> None: + """Renders a treegrid as produced by generate_treegrid""" + if renderer is None: + renderer = text_renderer.QuickTextRenderer() + renderer.render(treegrid) + + +class Volshell(VolshellShellPlugin): + """Shell environment to directly interact with a memory image.""" + _required_framework_version = (2, 0, 0) + def run(self, additional_locals: Dict[str, Any] = None) -> interfaces.renderers.TreeGrid: """Runs the interactive volshell plugin. @@ -59,6 +133,8 @@ class Volshell(interfaces.plugins.PluginInterface): Return a TreeGrid but this is always empty since the point of this plugin is to run interactively """ + # Try to load plugins + # Try to enable tab completion try: import readline @@ -73,6 +149,12 @@ class Volshell(interfaces.plugins.PluginInterface): # TODO: provide help, consider generic functions (pslist?) and/or providing windows/linux functions + # Load generic Volshell addons + framework.import_files(volshell) + for plugin in framework.class_subclasses(volshell.VolshellPlugin): + plugin_instanve = plugin(self) + self._plugins.append(plugin_instanve) + mode = self.__module__.split('.')[-1] mode = mode[0].upper() + mode[1:] @@ -86,13 +168,13 @@ class Volshell(interfaces.plugins.PluginInterface): """ sys.ps1 = f"({self.current_layer}) >>> " - self.__console = code.InteractiveConsole(locals = self._construct_locals_dict()) + self._console = code.InteractiveConsole(locals = self._construct_locals_dict()) # Since we have to do work to add the option only once for all different modes of volshell, we can't # rely on the default having been set if self.config.get('script', None) is not None: self.run_script(location = self.config['script']) - self.__console.interact(banner = banner) + self._console.interact(banner = banner) return renderers.TreeGrid([("Terminating", str)], None) @@ -119,20 +201,35 @@ class Volshell(interfaces.plugins.PluginInterface): def construct_locals(self) -> List[Tuple[List[str], Any]]: """Returns a dictionary listing the functions to be added to the environment.""" - return [(['dt', 'display_type'], self.display_type), (['db', 'display_bytes'], self.display_bytes), - (['dw', 'display_words'], self.display_words), (['dd', - 'display_doublewords'], self.display_doublewords), - (['dq', 'display_quadwords'], self.display_quadwords), (['dis', 'disassemble'], self.disassemble), - (['cl', 'change_layer'], self.change_layer), - (['cs', 'change_symboltable'], self.change_symbol_table), - (['ck', 'change_kernel'], self.change_kernel), - (['context'], self.context), (['self'], self), - (['dpo', 'display_plugin_output'], self.display_plugin_output), - (['gt', 'generate_treegrid'], self.generate_treegrid), (['rt', - 'render_treegrid'], self.render_treegrid), - (['ds', 'display_symbols'], self.display_symbols), (['hh', 'help'], self.help), - (['cc', 'create_configurable'], self.create_configurable), (['lf', 'load_file'], self.load_file), - (['rs', 'run_script'], self.run_script)] + locals_dict = [(['dt', 'display_type'], self.display_type), (['db', 'display_bytes'], self.display_bytes), + (['dw', 'display_words'], self.display_words), (['dd', + 'display_doublewords'], + self.display_doublewords), + (['dq', 'display_quadwords'], self.display_quadwords), + (['dis', 'disassemble'], self.disassemble), + (['cl', 'change_layer'], self.change_layer), + (['cs', 'change_symboltable'], self.change_symbol_table), + (['ck', 'change_kernel'], self.change_kernel), + (['context'], self.context), (['self'], self), + (['dpo', 'display_plugin_output'], self.display_plugin_output), + (['gt', 'generate_treegrid'], self.generate_treegrid), (['rt', + 'render_treegrid'], + self.render_treegrid), + (['ds', 'display_symbols'], self.display_symbols), (['hh', 'help'], self.help), + (['cc', 'create_configurable'], self.create_configurable), (['lf', 'load_file'], self.load_file), + (['rs', 'run_script'], self.run_script)] + + used_aliases = set([item for sublist, _ in locals_dict for item in sublist]) + + # Load generic plugins + for plugin in self._plugins: + for plugin_aliases, plugin_command in plugin.construct_locals(): + for plugin_alias in plugin_aliases: + if plugin_alias in used_aliases: + vollog.info(f"Overwriting alias {plugin_alias} with value from {plugin.__class__.__name__}") + locals_dict.append((plugin_aliases, plugin_command)) + + return locals_dict def _construct_locals_dict(self) -> Dict[str, Any]: """Returns a dictionary of the locals """ @@ -140,6 +237,7 @@ class Volshell(interfaces.plugins.PluginInterface): for aliases, value in self.construct_locals(): for alias in aliases: result[alias] = value + return result def _read_data(self, offset, count = 128, layer_name = None): @@ -177,41 +275,6 @@ class Volshell(interfaces.plugins.PluginInterface): """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): - if self.__current_layer is None: - self.__current_layer = self.config['primary'] - return self.__current_layer - - @property - def current_symbol_table(self): - if self.__current_symbol_table is None and self.kernel: - self.__current_symbol_table = self.kernel.symbol_table_name - return self.__current_symbol_table - - @property - def current_kernel_name(self): - if self.__current_kernel_name is None: - self.__current_kernel_name = self.config.get('kernel', None) - return self.__current_kernel_name - - @property - def kernel(self): - """Returns the current kernel object""" - if self.current_kernel_name not in self.context.modules: - return None - return self.context.modules[self.current_kernel_name] - - def change_layer(self, layer_name: str = None): - """Changes the current default layer""" - if not layer_name: - layer_name = self.current_layer - if layer_name not in self.context.layers: - print(f"Layer {layer_name} not present in context") - else: - self.__current_layer = layer_name - sys.ps1 = f"({self.current_layer}) >>> " - def change_symbol_table(self, symbol_table_name: str = None): """Changes the current_symbol_table""" if not symbol_table_name: @@ -219,16 +282,17 @@ class Volshell(interfaces.plugins.PluginInterface): if symbol_table_name not in self.context.symbol_space: print(f"Symbol table {symbol_table_name} not present in context symbol_space") else: - self.__current_symbol_table = symbol_table_name + self._current_symbol_table = symbol_table_name print(f"Current Symbol Table: {self.current_symbol_table}") def change_kernel(self, kernel_name: str = None): + """Alters the kernel module""" if not kernel_name: print("No kernel module name provided, not changing current kernel") if kernel_name not in self.context.modules: print(f"Kernel module {kernel_name} not found in the context module list") else: - self.__current_kernel_name = kernel_name + self._current_kernel_name = kernel_name print(f"Current kernel : {self.current_kernel_name}") def display_bytes(self, offset, count = 128, layer_name = None): @@ -332,34 +396,6 @@ class Volshell(interfaces.plugins.PluginInterface): else: return hex(value.vol.offset) - def generate_treegrid(self, plugin: Type[interfaces.plugins.PluginInterface], - **kwargs) -> Optional[interfaces.renderers.TreeGrid]: - """Generates a TreeGrid based on a specific plugin passing in kwarg configuration values""" - path_join = interfaces.configuration.path_join - - # Generate a temporary configuration path - plugin_config_suffix = self.random_string() - plugin_path = path_join(self.config_path, plugin_config_suffix) - - # Populate the configuration - for name, value in kwargs.items(): - self.config[path_join(plugin_config_suffix, plugin.__name__, name)] = value - - try: - constructed = plugins.construct_plugin(self.context, [], plugin, plugin_path, None, NullFileHandler) - return constructed.run() - except exceptions.UnsatisfiedException as excp: - print(f"Unable to validate the plugin requirements: {[x for x in excp.unsatisfied]}\n") - return None - - def render_treegrid(self, - treegrid: interfaces.renderers.TreeGrid, - renderer: Optional[interfaces.renderers.Renderer] = None) -> None: - """Renders a treegrid as produced by generate_treegrid""" - if renderer is None: - renderer = text_renderer.QuickTextRenderer() - renderer.render(treegrid) - def display_plugin_output(self, plugin: Type[interfaces.plugins.PluginInterface], **kwargs) -> None: """Displays the output for a particular plugin (with keyword arguments)""" treegrid = self.generate_treegrid(plugin, **kwargs) @@ -391,7 +427,7 @@ class Volshell(interfaces.plugins.PluginInterface): print(f"Running code from {location}\n") accessor = resources.ResourceAccessor() with io.TextIOWrapper(accessor.open(url = location), encoding = 'utf-8') as fp: - self.__console.runsource(fp.read(), symbol = 'exec') + self._console.runsource(fp.read(), symbol = 'exec') print("\nCode complete") def load_file(self, location: str): @@ -442,6 +478,9 @@ class Volshell(interfaces.plugins.PluginInterface): class NullFileHandler(io.BytesIO, interfaces.plugins.FileHandlerInterface): """Null FileHandler that swallows files whole without consuming memory""" + def close(self): + pass + def __init__(self, preferred_name: str): interfaces.plugins.FileHandlerInterface.__init__(self, preferred_name) super().__init__() diff --git a/volatility3/cli/volshell/linux.py b/volatility3/cli/volshell/linux.py index 97a488743..95694922e 100644 --- a/volatility3/cli/volshell/linux.py +++ b/volatility3/cli/volshell/linux.py @@ -67,6 +67,6 @@ class Volshell(generic.Volshell): @property def current_layer(self): - if self.__current_layer is None: - self.__current_layer = self.kernel.layer_name - return self.__current_layer + if self._current_layer is None: + self._current_layer = self.kernel.layer_name + return self._current_layer diff --git a/volatility3/cli/volshell/windows.py b/volatility3/cli/volshell/windows.py index 2cc5d3e1d..948dcb776 100644 --- a/volatility3/cli/volshell/windows.py +++ b/volatility3/cli/volshell/windows.py @@ -64,6 +64,6 @@ class Volshell(generic.Volshell): @property def current_layer(self): - if self.__current_layer is None: - self.__current_layer = self.kernel.layer_name - return self.__current_layer + if self._current_layer is None: + self._current_layer = self.kernel.layer_name + return self._current_layer