Add dynmically generated command line parameters for the CLI.

Note this now requires the file to be specified with single-location,
it also requires a specific ordering of parameters.
This commit is contained in:
Mike Auty
2017-01-08 02:31:31 +00:00
parent 282af4db99
commit cafc09a6d3
3 changed files with 100 additions and 37 deletions
+98 -34
View File
@@ -9,14 +9,16 @@
"""
import argparse
import inspect
import json
import logging
import os
import sys
import volatility.framework
import volatility.plugins
from volatility.framework import automagic, constants, contexts, interfaces
from volatility.framework import automagic
from volatility.framework import constants, contexts, interfaces
from volatility.framework.configuration import requirements
from volatility.framework.interfaces.configuration import HierarchicalDict
from volatility.framework.renderers.text import TextRenderer
@@ -49,69 +51,90 @@ class CommandLine(object):
volatility.framework.require_interface_version(0, 0, 0)
# TODO: Get CLI config options
# Do the initialization
ctx = contexts.Context() # Construct a blank context
volatility.framework.import_files(volatility.plugins)
automagics = automagic.available(ctx)
plugin_list = {}
for plugin in volatility.framework.class_subclasses(interfaces.plugins.PluginInterface):
plugin_name = plugin.__module__ + "." + plugin.__name__
if plugin_name.startswith("volatility.plugins."):
plugin_name = plugin_name[len("volatility.plugins."):]
plugin_list[plugin_name] = plugin
# TODO: Choose a plugin
parser = argparse.ArgumentParser(prog = 'volatility',
description = "An open-source memory forensics framework")
parser.add_argument("-p", "--plugin", help = "Run the following plugin", default = "windows.pslist.PsList")
parser.add_argument("file", help = "Temporary method for changing the file", default = None)
parser.add_argument("-c", "--config", help = "Load the configuration from a json file", default = None,
type = str)
parser.add_argument("-v", "--verbosity", help = "Increase output verbosity", default = 0, action = "count")
# argparse_adapter.adapt_config(context.config, parser)
# Run the argparser
args = parser.parse_args()
console.setLevel(10 - min(3, args.verbosity))
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
print("PLUGIN", args.plugin)
plug_class = args.plugin.split(".")[-1]
plug_mod = ".".join(args.plugin.split(".")[:-1])
plug_name = "volatility.plugins." + plug_mod
plugin = None
for module in sys.modules:
if plug_name == module:
plugin = getattr(sys.modules[module], plug_class)
break
else:
raise RuntimeError("Invalid plugin requested: {}".format(plug_name))
config_path = interfaces.configuration.path_join("plugins", plugin.__name__.lower())
subparser = parser.add_subparsers(title = "Plugins", dest = "plugin")
for plugin in plugin_list:
plugin_parser = subparser.add_parser(plugin, help = plugin.__doc__)
self.populate_requirements_argparse(plugin_parser, plugin_list[plugin])
configurables_list[plugin] = plugin_list[plugin]
###
# PASS TO UI
###
# Hand the plugin requirements over to the CLI (us) and let it construct the config tree
# UI fills in the config:
ctx = contexts.Context()
# Run the argparser
args = parser.parse_args()
if args.plugin is None:
parser.error("Please select a plugin to run")
console.setLevel(10 - min(3, args.verbosity))
plugin = plugin_list[args.plugin]
plugin_config_path = interfaces.configuration.path_join('plugins', plugin.__name__)
# 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():
value = vargs.get(requirement.name, None)
if value is not None:
if not inspect.isclass(configurables_list[configurable]):
config_path = configurables_list[configurable].config_path
else:
# 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
# UI fills in the config:
if args.config:
with open(args.config, "r") as f:
json_val = json.load(f)
ctx.config.splice("plugins.pslist", HierarchicalDict(json_val))
if not args.file or not os.path.exists(args.file):
raise RuntimeError("Please provide a valid filename")
else:
ctx.config["automagic.LayerStacker.single_location"] = "file://" + os.path.abspath(args.file)
pass
ctx.config.splice(plugin_config_path, HierarchicalDict(json_val))
###
# BACK TO THE FRAMEWORK
###
# Clever magic figures out how to fulfill each requirement that might not be fulfilled
automagics = automagic.available(ctx)
automagic.run(automagics, ctx, plugin, "plugins", progress_callback = progress_callback)
print("CONFIG", ctx.config)
# Check all the requirements and/or go back to the automagic step
if not plugin.validate(ctx, config_path):
if not plugin.validate(ctx, plugin_config_path):
raise RuntimeError("Unable to validate the plugin configuration")
print("\n\n")
constructed = plugin(ctx, config_path)
constructed = plugin(ctx, plugin_config_path)
with open("config.json", "w") as f:
json.dump(dict(constructed.build_configuration()), f, sort_keys = True, indent = 2)
@@ -119,6 +142,47 @@ class CommandLine(object):
# Construct and run the plugin
TextRenderer().render(constructed.run())
def populate_requirements_argparse(self, parser, configurable):
"""Adds the plugin's simple requirements to the provided parser
:param parser: The parser to add the plugin's (simple) requirements to
:type parser: argparse.ArgumentParser
:param configurable: The plugin object to pull the requirements from
:type configurable: volatility.framework.interfaces.plugins.PluginInterface
"""
if not issubclass(configurable, interfaces.configuration.ConfigurableInterface):
raise TypeError("Expected ConfigurableInterface type, not: {}".format(type(configurable)))
# Construct an argparse group
for requirement in configurable.get_requirements():
additional = {}
if not isinstance(requirement, interfaces.configuration.RequirementInterface):
raise TypeError(
"Plugin contains requirements that are not RequirementInterfaces: {}".format(configurable.__name__))
if isinstance(requirement, interfaces.configuration.InstanceRequirement):
additional["type"] = requirement.instance_type
if isinstance(requirement, requirements.BooleanRequirement):
additional["action"] = "store_true"
elif isinstance(requirement, requirements.ListRequirement):
if requirement.min_elements != requirement.max_elements:
if requirement.min_elements > 0:
additional["nargs"] = "+"
additional["nargs"] = "*"
# We can't test for min_elements > 1 or going over max_elements but rather than warning here,
# we expect the plugin to fail validation instead
else:
additional["nargs"] = requirement.max_elements
additional["type"] = requirement.element_type.instance_type
elif isinstance(requirement, requirements.ChoiceRequirement):
additional["type"] = str
additional["choices"] = requirement.choices
else:
continue
parser.add_argument("--" + requirement.name.replace('_', '-'), help = requirement.description,
default = requirement.default, dest = requirement.name,
required = not requirement.optional, **additional)
def progress_callback(progress, description = None):
""" A sinmple function for providing text-based feedback
+1 -1
View File
@@ -60,7 +60,7 @@ def run(automagics, context, configurable, config_path, progress_callback = None
configurable_class = configurable
if isinstance(configurable, interfaces.configuration.ConfigurableInterface):
configurable_class = configurable.__class__
requirement = requirements.MultiRequirement(name = configurable_class.__name__.lower())
requirement = requirements.MultiRequirement(name = configurable_class.__name__)
for req in configurable.get_requirements():
requirement.add_requirement(req)
+1 -2
View File
@@ -57,8 +57,7 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
new_context = context.clone()
current_layer_name = context.memory.free_layer_name("FileLayer")
current_config_path = interfaces.configuration.path_join("automagic", "layer_stacker", "stack",
current_layer_name)
current_config_path = interfaces.configuration.path_join(config_path, "stack", current_layer_name)
# This must be specific to get us started, setup the config and run
new_context.config[interfaces.configuration.path_join(current_config_path, "filename")] = self.local_store
new_context.add_layer(physical.FileLayer(new_context, current_config_path, current_layer_name))