From 035bada7b734a1e77ad65ca845f510633f2374d9 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Sun, 6 May 2018 18:11:15 +0100 Subject: [PATCH] Refactor volshell from a plugin to a standalone program. --- volatility/cli/__init__.py | 131 ++++++++------ volatility/cli/volshell/__init__.py | 162 ++++++++++++++++++ .../volshell/shellplugin.py} | 0 .../volshell.py => cli/volshell/windows.py} | 6 +- volshell.py | 7 + 5 files changed, 253 insertions(+), 53 deletions(-) create mode 100644 volatility/cli/volshell/__init__.py rename volatility/{plugins/volshell.py => cli/volshell/shellplugin.py} (100%) rename volatility/{plugins/windows/volshell.py => cli/volshell/windows.py} (93%) create mode 100644 volshell.py diff --git a/volatility/cli/__init__.py b/volatility/cli/__init__.py index 1036b2484..00c11fbab 100644 --- a/volatility/cli/__init__.py +++ b/volatility/cli/__init__.py @@ -22,7 +22,7 @@ import volatility.framework.configuration.requirements import volatility.plugins from volatility import framework from volatility.cli import text_renderer -from volatility.framework import automagic, constants, contexts, interfaces +from volatility.framework import automagic, constants, contexts, interfaces, exceptions from volatility.framework.configuration import requirements # Make sure we log everything @@ -164,8 +164,78 @@ class CommandLine(interfaces.plugins.FileConsumerInterface): json_val = json.load(f) ctx.config.splice(plugin_config_path, interfaces.configuration.HierarchicalDict(json_val)) - # Populate the context config based on the returned args - # We have already determined these elements must be descended from ConfigurableInterface + self.populate_config(ctx, configurables_list, args, plugin_config_path) + + if args.extend: + for extension in args.extend: + if '=' not in extension: + raise ValueError( + "Invalid extension (extensions must be of the format \"conf.path.value='value'\")") + address, value = extension[:extension.find('=')], json.loads(extension[extension.find('=') + 1:]) + ctx.config[address] = value + + # It should be up to the UI to determine which automagics to run, so this is before BACK TO THE FRAMEWORK + automagics = automagic.choose_automagic(automagics, plugin) + self.output_dir = args.output_dir + + ### + # BACK TO THE FRAMEWORK + ### + try: + constructed = self.run_plugin(ctx, + automagics, + plugin, + plugin_config_path, + quiet = args.quiet, + write_config = args.write_config) + + # Construct and run the plugin + text_renderer.QuickTextRenderer().render(constructed.run()) + except UnsatisfiedException as excp: + parser.exit(1, "Unable to validate the plugin requirements: {}\n".format(excp.unsatisfied)) + + def run_plugin(self, + context: interfaces.context.ContextInterface, + automagics: typing.List[interfaces.automagic.AutomagicInterface], + plugin: typing.Type[interfaces.plugins.PluginInterface], + plugin_config_path: str, + write_config: bool = False, + quiet: bool = False): + """Run the actual plugin based on the parameters + + Clever magic figures out how to fulfill each requirement that might not be fulfilled + """ + progress_callback = PrintedProgress() + if quiet: + progress_callback = None + errors = automagic.run(automagics, context, plugin, "plugins", progress_callback = progress_callback) + + # Check all the requirements and/or go back to the automagic step + unsatisfied = plugin.unsatisfied(context, plugin_config_path) + if unsatisfied: + for error in errors: + error_string = [x for x in error.format_exception_only()][-1] + vollog.warning("Automagic exception occured: {}".format(error_string[:-1])) + vollog.log(constants.LOGLEVEL_V, "".join(error.format(chain = True))) + raise UnsatisfiedException(unsatisfied) + + print("\n\n") + + constructed = plugin(context, plugin_config_path, progress_callback = progress_callback) + if write_config: + vollog.debug("Writing out configuration data to config.json") + with open("config.json", "w") as f: + json.dump(dict(constructed.build_configuration()), f, sort_keys = True, indent = 2) + constructed.set_file_consumer(self) + return constructed + + def populate_config(self, + context: interfaces.context.ContextInterface, + configurables_list: typing.Dict[str, interfaces.configuration.ConfigurableInterface], + args: argparse.Namespace, + plugin_config_path: str): + """Populate the context config based on the returned args + We have already determined these elements must be descended from ConfigurableInterface""" vargs = vars(args) for configurable in configurables_list: for requirement in configurables_list[configurable].get_requirements(): @@ -181,52 +251,7 @@ class CommandLine(interfaces.plugins.FileConsumerInterface): # We must be the plugin, so name it appropriately: config_path = plugin_config_path extended_path = interfaces.configuration.path_join(config_path, requirement.name) - ctx.config[extended_path] = value - - if args.extend: - for extension in args.extend: - if '=' not in extension: - raise ValueError( - "Invalid extension (extensions must be of the format \"conf.path.value='value'\")") - address, value = extension[:extension.find('=')], json.loads(extension[extension.find('=') + 1:]) - ctx.config[address] = value - - # It should be up to the UI to determine which automagics to run, so this is before BACK TO THE FRAMEWORK - automagics = automagic.choose_automagic(automagics, plugin) - - ### - # BACK TO THE FRAMEWORK - ### - # Clever magic figures out how to fulfill each requirement that might not be fulfilled - progress_callback = PrintedProgress() - if args.quiet: - progress_callback = None - - errors = automagic.run(automagics, ctx, plugin, "plugins", progress_callback = progress_callback) - - # Check all the requirements and/or go back to the automagic step - unsatisfied = plugin.unsatisfied(ctx, plugin_config_path) - if unsatisfied: - for error in errors: - error_string = [x for x in error.format_exception_only()][-1] - vollog.warning("Automagic exception occured: {}".format(error_string[:-1])) - vollog.log(constants.LOGLEVEL_V, "".join(error.format(chain = True))) - parser.exit(1, "Unable to validate the plugin requirements: {}\n".format(unsatisfied)) - - print("\n\n") - - constructed = plugin(ctx, plugin_config_path, progress_callback = progress_callback) - - if args.write_config: - vollog.debug("Writing out configuration data to config.json") - with open("config.json", "w") as f: - json.dump(dict(constructed.build_configuration()), f, sort_keys = True, indent = 2) - - self.output_dir = args.output_dir - constructed.set_file_consumer(self) - - # Construct and run the plugin - text_renderer.QuickTextRenderer().render(constructed.run()) + context.config[extended_path] = value def consume_file(self, filedata: interfaces.plugins.FileInterface): """Consumes a file as produced by a plugin""" @@ -246,7 +271,7 @@ class CommandLine(interfaces.plugins.FileConsumerInterface): vollog.warning("Refusing to overwrite an existing file: {}".format(output_filename)) def populate_requirements_argparse(self, - parser: argparse.ArgumentParser, + parser: typing.Union[argparse.ArgumentParser, argparse._ArgumentGroup], configurable: typing.Type[interfaces.configuration.ConfigurableInterface]): """Adds the plugin's simple requirements to the provided parser @@ -330,6 +355,12 @@ class HelpfulSubparserAction(argparse._SubParsersAction): getattr(namespace, argparse._UNRECOGNIZED_ARGS_ATTR).extend(arg_strings) +class UnsatisfiedException(exceptions.VolatilityException): + def __init__(self, unsatisfied): + super().__init__() + self.unsatisfied = unsatisfied + + def main(): """A convenience function for constructing and running the :class:`CommandLine`'s run method""" CommandLine().run() diff --git a/volatility/cli/volshell/__init__.py b/volatility/cli/volshell/__init__.py new file mode 100644 index 000000000..c4e20f11a --- /dev/null +++ b/volatility/cli/volshell/__init__.py @@ -0,0 +1,162 @@ +import argparse +import json +import logging +import os +import sys +from urllib import request + +import volatility.framework +import volatility.plugins +from volatility import cli, framework +from volatility.cli import text_renderer +from volatility.cli.volshell import shellplugin, windows +from volatility.framework import constants, contexts, automagic, interfaces + +# Make sure we log everything +vollog = logging.getLogger() +vollog.setLevel(0) +# Trim the console down by default +console = logging.StreamHandler() +console.setLevel(logging.WARNING) +formatter = logging.Formatter('%(levelname)-8s %(name)-12s: %(message)s') +console.setFormatter(formatter) +vollog.addHandler(console) + + +class VolShell(cli.CommandLine): + """Program to allow interactive """ + + def run(self): + sys.stdout.write("Volshell (Volatility Framework) {}\n".format(constants.PACKAGE_VERSION)) + + volatility.framework.require_interface_version(0, 0, 0) + + parser = argparse.ArgumentParser(prog = 'volshell', + description = "A tool for interactivate forensic analysis of memory images") + parser.add_argument("-c", "--config", help = "Load the configuration from a json file", default = None, + type = str) + parser.add_argument("-e", "--extend", help = "Extend the configuration with a new (or changed) setting", + default = None, action = 'append') + parser.add_argument("-p", "--plugins", help = "Semi-colon separated list of paths to find plugins", + default = "", type = str) + parser.add_argument("-v", "--verbosity", help = "Increase output verbosity", default = 0, action = "count") + parser.add_argument("-o", "--output-dir", help = "Directory in which to output any generated files", + default = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')), type = str) + parser.add_argument("--log", help = "Log output to a file as well as the console", default = None, type = str) + parser.add_argument("-f", metavar = "FILE", default = None, type = str, + help = "Shorthand for --single-location=file://FILE if single-location is not defined") + + # Volshell specific flags + parser.add_argument("-w", "--windows", default = False, action = "store_true", help = "Run a Windows volshell") + parser.add_argument("-l", "--linux", default = False, action = "store_true", help = "Run a Linux volshell") + + # We have to filter out help, otherwise parse_known_args will trigger the help message before having + # processed the plugin choice or had the plugin subparser added. + known_args = [arg for arg in sys.argv if arg != '--help' and arg != '-h'] + partial_args, _ = parser.parse_known_args(known_args) + if partial_args.plugins: + volatility.plugins.__path__ = partial_args.plugins.split(";") + constants.PLUGINS_PATH + + if partial_args.log: + file_logger = logging.FileHandler(partial_args.log) + file_logger.setLevel(0) + file_formatter = logging.Formatter(datefmt = '%y-%m-%d %H:%M:%S', + fmt = '%(asctime)s %(name)-12s %(levelname)-8s %(message)s') + file_logger.setFormatter(file_formatter) + vollog.addHandler(file_logger) + vollog.info("Logging started") + + # Do the initialization + ctx = contexts.Context() # Construct a blank context + framework.import_files(volatility.plugins) # Will not log as console's default level is WARNING + automagics = automagic.available(ctx) + + # Initialize the list of plugins in case volshell needs it + framework.list_plugins() + + seen_automagics = set() + configurables_list = {} + for amagic in automagics: + if amagic in seen_automagics: + continue + seen_automagics.add(amagic) + if isinstance(amagic, interfaces.configuration.ConfigurableInterface): + self.populate_requirements_argparse(parser, amagic.__class__) + configurables_list[amagic.__class__.__name__] = amagic + + # We don't list plugin arguments, because they can be provided within python + volshell_plugin_list = {'generic': shellplugin.Volshell, + 'windows': windows.Volshell} + print(repr(volshell_plugin_list)) + for plugin in volshell_plugin_list: + subparser = parser.add_argument_group(title = plugin.capitalize(), + description = "Configuration options based on {} options".format( + plugin.capitalize())) + self.populate_requirements_argparse(subparser, volshell_plugin_list[plugin]) + configurables_list[plugin] = volshell_plugin_list[plugin] + + # Run the argparser + args = parser.parse_args() + if args.verbosity < 3: + console.setLevel(30 - (args.verbosity * 10)) + else: + console.setLevel(10 - (args.verbosity - 2)) + + vollog.log(constants.LOGLEVEL_VVV, "Cache directory used: {}".format(constants.CACHE_PATH)) + + plugin = shellplugin.Volshell + if args.windows: + plugin = windows.Volshell + + plugin_config_path = interfaces.configuration.path_join('plugins', plugin.__name__) + + # Special case the -f argument because people use is so frequently + # It has to go here so it can be overridden by single-location if it's defined + # NOTE: This will *BREAK* if LayerStacker, or the automagic configuration system, changes at all + ### + if args.f: + file_name = os.path.abspath(args.f) + if not os.path.exists(file_name): + vollog.log(logging.INFO, "File does not exist: {}".format(file_name)) + else: + single_location = "file:" + request.pathname2url(file_name) + ctx.config['automagic.LayerStacker.single_location'] = single_location + + # UI fills in the config, here we load it from the config file and do it before we process the CL parameters + if args.config: + with open(args.config, "r") as f: + json_val = json.load(f) + ctx.config.splice(plugin_config_path, interfaces.configuration.HierarchicalDict(json_val)) + + self.populate_config(ctx, configurables_list, args, plugin_config_path) + + if args.extend: + for extension in args.extend: + if '=' not in extension: + raise ValueError( + "Invalid extension (extensions must be of the format \"conf.path.value='value'\")") + address, value = extension[:extension.find('=')], json.loads(extension[extension.find('=') + 1:]) + ctx.config[address] = value + + # It should be up to the UI to determine which automagics to run, so this is before BACK TO THE FRAMEWORK + automagics = automagic.choose_automagic(automagics, plugin) + self.output_dir = args.output_dir + + ### + # BACK TO THE FRAMEWORK + ### + try: + constructed = self.run_plugin(ctx, + automagics, + plugin, + plugin_config_path) + + # Construct and run the plugin + text_renderer.QuickTextRenderer().render(constructed.run()) + except cli.UnsatisfiedException as excp: + parser.exit(1, "Unable to validate the plugin requirements: {}\n".format(excp.unsatisfied)) + + +def main(): + """A convenience function for constructing and running the :class:`CommandLine`'s run method""" + VolShell().run() diff --git a/volatility/plugins/volshell.py b/volatility/cli/volshell/shellplugin.py similarity index 100% rename from volatility/plugins/volshell.py rename to volatility/cli/volshell/shellplugin.py diff --git a/volatility/plugins/windows/volshell.py b/volatility/cli/volshell/windows.py similarity index 93% rename from volatility/plugins/windows/volshell.py rename to volatility/cli/volshell/windows.py index 16f61888d..bc19ad390 100644 --- a/volatility/plugins/windows/volshell.py +++ b/volatility/cli/volshell/windows.py @@ -1,8 +1,8 @@ import inspect +from volatility.cli.volshell import shellplugin from volatility.framework.configuration import requirements from volatility.framework.interfaces import plugins -from volatility.plugins import volshell class Volshell(plugins.PluginInterface): @@ -10,7 +10,7 @@ class Volshell(plugins.PluginInterface): @classmethod def get_requirements(cls): - return (volshell.Volshell.get_requirements() + + return (shellplugin.Volshell.get_requirements() + [requirements.SymbolRequirement(name = "nt_symbols", description = "Windows OS"), requirements.IntRequirement(name = 'pid', description = "Process ID", @@ -63,4 +63,4 @@ class Volshell(plugins.PluginInterface): eproc = _x break - return volshell.Volshell(self.context, "plugins.Volshell").run(curframe.f_locals) + return shellplugin.Volshell(self.context, "plugins.Volshell").run(curframe.f_locals) diff --git a/volshell.py b/volshell.py new file mode 100644 index 000000000..be553efd1 --- /dev/null +++ b/volshell.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python3.5 + +from volatility.cli import volshell + +if __name__ == '__main__': + volshell.main() +