mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-28 04:24:53 +02:00
Refactor volshell from a plugin to a standalone program.
This commit is contained in:
Mike Auty
+81
-50
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
@@ -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)
|
||||
@@ -0,0 +1,7 @@
|
||||
#!/usr/bin/env python3.5
|
||||
|
||||
from volatility.cli import volshell
|
||||
|
||||
if __name__ == '__main__':
|
||||
volshell.main()
|
||||
|
||||
Reference in New Issue
Block a user