diff --git a/volatility/cli/__init__.py b/volatility/cli/__init__.py index 6e1b7fc27..b6193ee86 100644 --- a/volatility/cli/__init__.py +++ b/volatility/cli/__init__.py @@ -3,9 +3,10 @@ import logging import sys import volatility.framework +import volatility.framework.automagic import volatility.plugins -from volatility.cli import argparse_adapter -from volatility.framework import configuration, contexts +from volatility.framework import contexts +from volatility.framework.interfaces import configuration as config_interface from volatility.framework.renderers.text import TextRenderer __author__ = 'mike' @@ -44,38 +45,41 @@ class CommandLine(object): # Run the argparser parser.parse_args() + config_path = config_interface.path_join("plugins", plugin.__name__.lower()) - # Determine the selected plugin - # Resolve the dependencies on that plugin - dldr = depresolver.DependencyResolver() - deptree = dldr.build_tree(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() - ctx.config["pslist.primary.memory_layer.filename"] = "/run/media/mike/disk/memory/xp-laptop-2005-07-04-1430.img" - ctx.config["pslist.offset"] = 0x823c87c0 + ctx.config["plugins.pslist.primary.class"] = "volatility.framework.layers.intel.Intel" + ctx.config[ + "plugins.pslist.primary.memory_layer.filename"] = "/run/media/mike/disk/memory/xp-laptop-2005-07-04-1430.img" + ctx.config["plugins.pslist.offset"] = 0x823c87c0 - ctx.config["pslist.primary.memory_layer.filename"] = "/run/media/mike/disk/memory/private/jon-fres.dmp" - ctx.config["pslist.offset"] = 0x81bcc830 + ctx.config["plugins.pslist.primary.memory_layer.class"] = "volatility.framework.layers.physical.FileLayer" + ctx.config["plugins.pslist.primary.memory_layer.filename"] = "/run/media/mike/disk/memory/private/jon-fres.dmp" + ctx.config["plugins.pslist.offset"] = 0x81bcc830 - # ctx.config["pslist.primary.page_map_offset"] = 0x39000 + ctx.config["plugins.pslist.primary.page_map_offset"] = 0x39000 - config_path = plugin.__name__.lower() + ### + # BACK TO THE FRAMEWORK + ### + # Clever magic figures out how to fulfill each requirement that might not be fulfilled + volatility.framework.automagic.automagic(ctx, plugin, "plugins") - windows = True - if windows: - # Traverse the dependency tree and tag the config with the appropriate page_map_offset values where not already applied - deptree.traverse(windows_automagic.PageMapOffsetHelper(context = ctx), - config_path = config_path, - short_circuit = False) + import pdb + pdb.set_trace() - # Walk down the tree attempting to fulfil each requirement (recursive) and backtrack when necessary - # Translate the parsed args to a context configuration - if dldr.validate_dependencies(deptree, context = ctx, path = config_path): - # Construct and run the plugin - TextRenderer().render(plugin(ctx, config_path).run()) - else: - raise DependencyError("Unable to validate all the dependencies, please check configuration parameters") + # Check all the requirements and/or go back to the automagic step + if not plugin.validate(ctx, config_path): + raise RuntimeError("Unable to validate the plugin configuration") + + # Construct and run the plugin + TextRenderer().render(plugin(ctx, config_path).run()) def main(): diff --git a/volatility/framework/automagic/__init__.py b/volatility/framework/automagic/__init__.py index e69de29bb..61ca55f77 100644 --- a/volatility/framework/automagic/__init__.py +++ b/volatility/framework/automagic/__init__.py @@ -0,0 +1,29 @@ +import sys + +from volatility.framework import class_subclasses, import_files +from volatility.framework.configuration import MultiRequirement +from volatility.framework.interfaces import automagic as automagic_interface +from volatility.framework.interfaces.configuration import ConfigurableInterface + + +def automagic(context, configurable, config_path = ""): + """Runs through all the appropriate automagic capabilities on the configurable + + This is where any automagic is allowed to run, and alter the context in order to satisfy/improve all requirements + """ + import_files(sys.modules[__name__]) + if not isinstance(configurable, ConfigurableInterface) and not issubclass(configurable, ConfigurableInterface): + raise TypeError("Automagic operates on configurables only") + automagics = [clazz() for clazz in class_subclasses(automagic_interface.AutomagicInterface)] + + # TODO: Fix need for top level config element just because we're using a MultiRequirement to group the + # configurable's config requirements + configurable_class = configurable + if isinstance(configurable, ConfigurableInterface): + configurable_class = configurable.__class__ + requirement = MultiRequirement(name = configurable_class.__name__.lower()) + for req in configurable.get_requirements(): + requirement.add_requirement(req) + + for automagic in sorted(automagics, key = lambda x: x.priority): + automagic(context, requirement, config_path) diff --git a/volatility/framework/automagic/construct_layers.py b/volatility/framework/automagic/construct_layers.py new file mode 100644 index 000000000..43ffc27bb --- /dev/null +++ b/volatility/framework/automagic/construct_layers.py @@ -0,0 +1,27 @@ +from volatility.framework.configuration import requirements +from volatility.framework.interfaces import automagic as automagic_interface, configuration as config_interface + + +class ConstructLayers(automagic_interface.AutomagicInterface): + """Runs through the requirement tree and from the bottom up attempts to construct all TranslationLayerRequirements""" + + def __call__(self, context, requirement, config_path): + print("Processing", config_path, requirement.name) + if not requirement.validate(context, config_path): + # Having called validate at the top level tells us both that we need to dig deeper + # but also ensures that TranslationLayerRequirements have got the correct subrequirements if their class is populated + + success = True + for subreq in requirement.requirements: + subreq_config_path = config_interface.path_join(config_path, requirement.name) + self(context, subreq, subreq_config_path) + valid = subreq.validate(context, subreq_config_path) + # We want to traverse optional paths, so don't check until we've tried to validate + if not valid and not subreq.optional: + success = False + if not success: + return False + elif isinstance(requirement, requirements.TranslationLayerRequirement): + # We know all the subrequirements are filled, so let's populate + requirement.construct(context, config_path) + return True diff --git a/volatility/framework/automagic/windows.py b/volatility/framework/automagic/windows.py index 9df46a48e..2ddba07dc 100644 --- a/volatility/framework/automagic/windows.py +++ b/volatility/framework/automagic/windows.py @@ -1,3 +1,5 @@ +from volatility.framework.configuration import requirements + if __name__ == "__main__": import os import sys @@ -6,8 +8,8 @@ if __name__ == "__main__": import struct -from volatility.framework import interfaces, layers, validity, configuration -from volatility.framework.configuration import depresolver +from volatility.framework import interfaces, layers, validity +from volatility.framework.interfaces import automagic as automagic_interface PAGE_SIZE = 0x1000 @@ -127,9 +129,10 @@ class SelfReferentialTest(object): return response -class PageMapOffsetHelper(interfaces.configuration.HierachicalVisitor): - def __init__(self, context): - self.ctx = self._check_type(context, interfaces.context.ContextInterface) +class PageMapOffsetHelper(automagic_interface.AutomagicInterface): + priority = 20 + + def __init__(self): self.tests = dict([(test.layer_type, test) for test in [DtbTest32bit(), DtbTest64bit(), DtbTestPae()]]) def branch_leave(self, node, config_path): @@ -137,28 +140,23 @@ class PageMapOffsetHelper(interfaces.configuration.HierachicalVisitor): self(node, config_path) return True - def __call__(self, node, config_path): - if isinstance(node, depresolver.RequirementTreeChoice): - useful = [] - for candidate in node.candidates: - if candidate in self.tests: - useful.append(self.tests[candidate]) - if useful: - depresolver.DependencyResolver().validate_dependencies(node.candidates[useful[0].layer_type], self.ctx, - config_path) - prefix = config_path + configuration.CONFIG_SEPARATOR - memory_layer = self.ctx.config.get(prefix + "memory_layer", None) - page_table_offset = self.ctx.config.get(prefix + "page_map_offset", None) - if page_table_offset is None and memory_layer is not None: - hits = scan(self.ctx, memory_layer, useful) - for test in useful: - if hits.get(test.layer_type, []): - self.ctx.config[prefix + "page_map_offset"] = hits[test.layer_type][0] - else: - # Delete the node rather than fixing the constraints, - # since the requirements haven't changed, but some of the candidates are no longer valid - # If the constraints were global across the tree, then tagging the constraints may be more useful - del node.candidates[test.layer_type] + def __call__(self, context, requirement, config_path): + if isinstance(requirement, requirements.TranslationLayerRequirement): + # Determine if a class has been chosen + # If a class hasn't been chosen, look through the underlying config for appropriate parameters + # If possible run scan and choose an appropriate class + # Once an appropriate class has been chosen, attempt to determine the page_map_offset value + pass + + # hits = scan(self.context, memory_layer, useful) + # for test in useful: + # if hits.get(test.layer_type, []): + # self.context.config[prefix + "page_map_offset"] = hits[test.layer_type][0] + # else: + # # Delete the node rather than fixing the constraints, + # # since the requirements haven't changed, but some of the candidates are no longer valid + # # If the constraints were global across the tree, then tagging the constraints may be more useful + # del node.candidates[test.layer_type] return True diff --git a/volatility/framework/configuration/depresolver.py b/volatility/framework/configuration/depresolver.py deleted file mode 100644 index 9f2dde29b..000000000 --- a/volatility/framework/configuration/depresolver.py +++ /dev/null @@ -1,284 +0,0 @@ -import logging -from collections import OrderedDict - -import volatility.framework as framework -from volatility.framework import validity, interfaces - - -def satisfies(provider, requirement): - """Takes the requirement (which should always be a TranslationLayerRequirement) and determines if the - layer_class satisfies it""" - satisfied = True - for k, v in requirement.constraints.items(): - if k in provider.provides: - satisfied = satisfied and bool(common_provision(provider.provides[k], v)) - return satisfied - - -def common_provision(value1, value2): - """Normalizes individual values down to singleton lists, then tests for overlap between the two lists""" - if not isinstance(value1, list): - value1 = [value1] - if not isinstance(value2, list): - value2 = [value2] - set1 = set(value1) - set2 = set(value2) - return set1.intersection(set2) - - -class DependencyResolver(validity.ValidityRoutines): - def __init__(self): - # Maintain a cache of translation layers - self.configurable_cache = [] - self.provides = {} - self.providers_cache = sorted(list(self._build_caches(interfaces.configuration.ProviderInterface)), - key = lambda x: -x.priority) - - def _build_caches(self, clazz): - self.provides = {} - cache = set() - for provider in framework.class_subclasses(clazz): - for k, v in provider.provides.items(): - if not isinstance(v, list): - new_v = self.provides.get(k, set()) - new_v.add(v) - else: - new_v = self.provides.get(k, set()).union(set(v)) - self.provides[k] = new_v - cache.add(provider) - return cache - - def validate_dependencies(self, deptree, context, path = None): - """Takes a dependency tree and attempts to resolve the tree by validating each branch and using the first that successfully validates - - DEPTREE = [ REQUIREMENTS ... ] - REQUIREMENT = ( NODE | LEAF ) - NODE = req, { candidate : DEPTREE, ... } - LEAF = req - - @param path: A path to access the deptree's configuration details - """ - if path is None: - path = "" - - self._check_type(deptree, interfaces.configuration.RequirementTreeNode) - visitor = ValidatorVisitor(context) - deptree.traverse(visitor, path, short_circuit = True) - return visitor.is_valid() - - def build_tree(self, configurable): - """Takes a configurable and produces a priority ordered tree of possible solutions to satisfy the various requirements - - @param configurable: A configurable class that requires its dependency tree constructing - @param path: A path indicating where the configurable resides in the config namespace - @return deptree: The returned tree should include each of the potential nodes (and requirements, including optional ones) allowing the UI - to decide the layer build-path and get all the necessary variables from the user for that path. - """ - self._check_class(configurable, interfaces.configuration.ConfigurableInterface) - - deptree = [] - - for subreq in configurable.get_requirements(): - # Find all the different ways to fulfill it (recursively) - # TODO: Ensure no cycles or loops - if not isinstance(subreq, interfaces.configuration.ConstraintInterface): - deptree.append(RequirementTreeReq(requirement = subreq)) - else: - candidates = OrderedDict() - satisfiable = False - for potential in self.providers_cache: - if satisfies(potential, subreq): - try: - candidate = self.build_tree(potential) - candidates[potential] = candidate - satisfiable = True - except DependencyError: - pass - # Check we've satisfied one of the possibilities, exception if we haven't - if not satisfiable: - raise DependencyError("No solutions to fulfill requirement " + repr(subreq)) - # Construct the appropriate Requirement node - if candidates: - deptree.append(RequirementTreeChoice(requirement = subreq, candidates = candidates)) - return RequirementTreeList(deptree) - - -class DependencyError(Exception): - pass - - -########################## -# Visitors -########################## - - -class ValidatorVisitor(interfaces.configuration.HierachicalVisitor): - def __init__(self, context): - self.ctx = context - self.stack = [(None, [])] - - def is_valid(self): - _, result_list = self.stack[0] - return result_list[0] - - def branch_enter(self, node, config_path): - self.stack.append((node, [])) - return True - - def branch_leave(self, node, config_path): - (_, child_results), self.stack = self.stack[-1], self.stack[:-1] - _, stack = self.stack[-1] - if isinstance(node, RequirementTreeChoice): - # Don't bother validating if the choice didn't have one success - if not any(child_results): - # Choice requirements can still be valid even if their requirements failed if they are optional - stack.append(node.requirement.optional) - return True - else: - # Don't bother validating if the list failed - if not all(child_results): - # List requirements always fail if one inside fails to validate - # (since optional requirements in the list should validate as true) - stack.append(False) - return True - # If we haven't already determined the result - if node.requirement is not None: - return self(node, config_path) - else: - stack.append(True) - return True - - def __call__(self, node, config_path): - """Returns whether a node is valid""" - # Determine if we should - branch_node, branch_results = self.stack[-1] - - if isinstance(branch_node, RequirementTreeChoice) and branch_results and branch_results[-1] == True: - return False - if isinstance(branch_node, RequirementTreeList) and branch_results and branch_results[-1] == False: - return False - - # Attempt to fulfill the provider - if isinstance(node, RequirementTreeChoice) and not node.requirement.optional: - # Only try to provide when we're not already sorted - if self.ctx.config.get(config_path, None) is None: - for provider in node.candidates: - # Recheck the requirements in case the deptree has changed - if satisfies(provider, node.requirement): - try: - provider.fulfill(self.ctx, node.requirement, config_path) - break - except Exception as e: - pass - else: - logging.debug( - "Unable to fulfill requirement " + repr(node.requirement) + " - no fulfillable candidates") - branch_results.append(False) - return True - - try: - value = self.ctx.config[config_path] - node.requirement.validate(value, self.ctx) - branch_results.append(True) - except Exception as e: - if not node.requirement.optional: - logging.debug( - "Unable to fulfill non-optional requirement " + repr(node.requirement) + " [" + str(e) + "]") - branch_results.append(node.requirement.optional) - return True - - -class PrettyPrinter(interfaces.configuration.HierachicalVisitor): - def __init__(self): - self.lines = [] - - def run(self, deptree): - deptree.traverse(self, - config_path = "pprinter", - short_circuit = False) - for line in self.lines: - print(*line) - - def branch_leave(self, node, config_path): - return self(node, config_path) - - def __call__(self, node, config_path): - depth = config_path.count(interfaces.configuration.CONFIG_SEPARATOR) - lines = [("." * depth, config_path, type(node))] - if node.requirement is not None: - lines.append((" " * depth, node.requirement)) - self.lines = lines + self.lines - return True - - -########################## -# Requirement tree classes -########################## - - -class RequirementTreeReq(interfaces.configuration.RequirementTreeNode): - def __repr__(self): - return "" - - def traverse(self, visitor, config_path = None, short_circuit = False): - if config_path is None: - config_path = self.requirement.name - else: - self._check_type(config_path, str) - config_path += interfaces.configuration.CONFIG_SEPARATOR + self.requirement.name - - return visitor(self, config_path) - - -class RequirementTreeChoice(RequirementTreeReq): - def __init__(self, requirement = None, candidates = None): - RequirementTreeReq.__init__(self, requirement) - for k in candidates: - self._check_class(k, interfaces.configuration.ProviderInterface) - self._check_type(candidates[k], RequirementTreeList) - self.candidates = candidates - if candidates is None: - self.candidates = OrderedDict() - - def __repr__(self): - return "" - - def traverse(self, visitor, config_path = None, short_circuit = False): - if config_path is None: - config_path = self.requirement.name - else: - self._check_type(config_path, str) - config_path += interfaces.configuration.CONFIG_SEPARATOR + self.requirement.name - - if visitor.branch_enter(self, config_path): - for node in self.candidates.values(): - cont = node.traverse(visitor, config_path, short_circuit) - if not cont: - break - - return visitor.branch_leave(self, config_path) - - -class RequirementTreeList(interfaces.configuration.RequirementTreeNode): - def __init__(self, children = None): - interfaces.configuration.RequirementTreeNode.__init__(self, None) - self._check_type(children, list) - for child in children: - self._check_type(child, interfaces.configuration.RequirementTreeNode) - self.children = children - - def __repr__(self): - return "" - - def traverse(self, visitor, config_path = None, short_circuit = False): - if config_path is None: - config_path = "" - self._check_type(config_path, str) - - if visitor.branch_enter(self, config_path): - for node in self.children: - cont = node.traverse(visitor, config_path, short_circuit) - if not cont: - break - - return visitor.branch_leave(self, config_path) diff --git a/volatility/framework/configuration/requirements.py b/volatility/framework/configuration/requirements.py index dacd04cd9..625d8bcec 100644 --- a/volatility/framework/configuration/requirements.py +++ b/volatility/framework/configuration/requirements.py @@ -3,11 +3,30 @@ import sys from volatility.framework.interfaces import configuration as config_interface +vollog = logging.getLogger(__name__) + + +class MultiRequirement(config_interface.RequirementInterface): + """Class to hold multiple requirements + + Technically the Interface could handle this, but it's an interface, so this is a concrete implementation + """ + + def validate(self, context, config_path): + return self.validate_children(context, config_path) + class InstanceRequirement(config_interface.RequirementInterface): instance_type = bool - def validate(self, value, _context): + def add_requirement(self, requirement): + raise TypeError("Instance Requirements cannot have subrequirements") + + def remove_requirement(self, requirement): + raise TypeError("Instance Requirements cannot have subrequirements") + + def validate(self, context, config_path): + value = self.config_value(context, config_path, None) if not isinstance(value, self.instance_type): vollog.debug("TypeError - " + self.name + " input only accepts " + self.instance_type.__name__ + " type") return False @@ -27,11 +46,39 @@ class BytesRequirement(InstanceRequirement): instance_type = bytes -class TranslationLayerRequirement(config_interface.ConstraintInterface): +class ClassRequirement(config_interface.RequirementInterface): + """Requires a specific class""" + + def __init__(self, *args, **kwargs): + config_interface.RequirementInterface.__init__(self, *args, **kwargs) + self._cls = None + + @property + def cls(self): + return self._cls + + def validate(self, context, config_path): + """Checks to see if a class can be recovered""" + value = self.config_value(context, config_path, None) + self._cls = None + if value is not None: + if "." in value: + # TODO: consider importing the prefix + module = sys.modules.get(value[:value.rindex(".")], None) + class_name = value[value.rindex(".") + 1:] + if hasattr(module, class_name): + self._cls = getattr(module, class_name) + else: + if value in globals(): + self._cls = globals()[value] + return self._cls is not None + + +class TranslationLayerRequirement(config_interface.RequirementInterface): """Class maintaining the limitations on what sort of address spaces are acceptable""" def __init__(self, name, description = None, default = None, - optional = False, layer_name = None, constraints = None): + optional = False, constraints = None): """Constructs a Translation Layer Requirement The configuration option's value will be the name of the layer once it exists in the store @@ -40,38 +87,103 @@ class TranslationLayerRequirement(config_interface.ConstraintInterface): :param layer_name: String detailing the expected name of the required layer, this can be None if it is to be randomly generated :return: """ - config_interface.ConstraintInterface.__init__(self, name, description, default, optional, constraints) - self._layer_name = layer_name + config_interface.RequirementInterface.__init__(self, name, description, default, optional) + self.add_requirement(ClassRequirement("class", "Class of the translation layer")) + self._current_class_requirements = set() # TODO: Add requirements: acceptable OSes from the address_space information # TODO: Add requirements: acceptable arches from the available layers - def validate(self, value, context): + def validate(self, context, config_path): """Validate that the value is a valid layer name and that the layer adheres to the requirements""" - if not isinstance(value, str): - raise TypeError("TranslationLayerRequirements only accepts string labels") - if value not in context.memory: - raise IndexError((value or "") + " is not a memory layer") + value = self.config_value(context, config_path, None) + if isinstance(value, str): + if value not in context.memory: + vollog.debug("IndexError - Layer " + value + " not found in memory space") + return False + return True + + if value is not None: + vollog.debug("TypeError - TranslationLayerRequirements only accepts string labels") + return False + + # TODO: check that the space in the context lives up to the requirements for arch/os etc + + ### NOTE: This validate method has side effects (the dependencies can change)!!! + + # See if our class is valid and if so populate the other requirements + # (but no need to validate, since we're invalid already) + class_req = self._requirements['class'] + subreq_config_path = config_interface.path_join(config_path, self.name) + if class_req.validate(context, subreq_config_path): + # We have a class, and since it's validated we can construct our requirements from it + if issubclass(class_req.cls, config_interface.ConfigurableInterface): + # In case the class has changed, clear out the old requirements + for old_req in self._current_class_requirements.copy(): + del self._requirements[old_req] + self._current_class_requirements.remove(old_req) + # And add the new ones + for requirement in class_req.cls.get_requirements(): + self._current_class_requirements.add(requirement.name) + self.add_requirement(requirement) + vollog.debug("IndexError - No configuration provided for layer") + return False + + def construct(self, context, config_path): + """Constructs the appropriate layer and adds it based on the class parameter""" + config_path = config_interface.path_join(config_path, self.name) + if not all([subreq.validate(context, config_path) for subreq in self.requirements if not subreq.optional]): + return False + + cls = self._requirements["class"].cls + node_config = context.config.branch(config_path) + + # Determine the layer name + layer_name = self.name + counter = 2 + while layer_name in context.memory: + layer_name = self.name + str(counter) + counter += 1 + + # Construct the layer + requirement_dict = {} + for req in cls.get_requirements(): + if req.name in node_config.data and req.name != "class": + requirement_dict[req.name] = node_config.data[req.name] + # Fulfillment must happen, exceptions happening here mean the requirements aren't correct + # and these need to be raised and fixed, rather than caught and ignored + layer = cls(context, config_path, layer_name, **requirement_dict) + context.add_layer(layer) + context.config[config_path] = layer_name + return True -class SymbolRequirement(config_interface.ConstraintInterface): +class SymbolRequirement(config_interface.RequirementInterface): """Class maintaining the limitations on what sort of symbol spaces are acceptable""" def __init__(self, name, description = None, default = None, optional = False, constraints = None): - config_interface.ConstraintInterface.__init__(self, name, description, default, optional, constraints) + config_interface.RequirementInterface.__init__(self, name, description, default, optional) - def validate(self, value, context): + def validate(self, context, config_path): """Validate that the value is a valid within the symbol space of the provided context""" + value = self.config_value(context, config_path, None) if not isinstance(value, str): - raise TypeError("SymbolRequirement only accepts string labels") + vollog.debug("TypeError - SymbolRequirement only accepts string labels") + return False if value not in context.symbol_space: - raise IndexError((value or "") + " is not present in the symbol space") + # This is an expected situation, so return False rather than raise + vollog.debug("IndexError - " + (value or "") + " is not present in the symbol space") + return False + return True class NativeSymbolRequirement(SymbolRequirement): - def validate(self, value, context): + def validate(self, context, config_path): + value = self.config_value(context, config_path) if not isinstance(value, str): - raise TypeError("SymbolRequirement only accepts string labels") + vollog.debug("TypeError - SymbolRequirement only accepts string labels") + return False + return True class ChoiceRequirement(config_interface.RequirementInterface): @@ -83,10 +195,13 @@ class ChoiceRequirement(config_interface.RequirementInterface): raise TypeError("ChoiceRequirement takes a list of strings as choices") self._choices = choices - def validate(self, value, context): + def validate(self, context, config_path): """Validates the provided value to ensure it is one of the available choices""" + value = self.config_value(context, config_path) if value not in self._choices: - raise ValueError("Value is not within the set of available choices") + vollog.debug("ValueError - Value is not within the set of available choices") + return False + return True class ListRequirement(config_interface.RequirementInterface): @@ -98,12 +213,14 @@ class ListRequirement(config_interface.RequirementInterface): self.min_elements = min_elements self.max_elements = max_elements - def validate(self, value, context): + def validate(self, context, config_path): """Check the types on each of the returned values and then call the element type's check for each one""" + value = self.config_value(context, config_path) self._check_type(value, list) if not all([self._check_type(element, self.element_type) for element in value]): - raise TypeError("At least one element in the list is not of the correct type.") + vollog.debug("TypeError - At least one element in the list is not of the correct type.") + return False if not (self.min_elements <= len(value) <= self.max_elements): - raise TypeError("List option provided more or less elements than allowed.") - for element in value: - self.element_type.validate(element, context) + vollog.debug("TypeError - List option provided more or less elements than allowed.") + return False + return all([self.element_type.validate(context, element) for element in value]) diff --git a/volatility/framework/interfaces/automagic.py b/volatility/framework/interfaces/automagic.py new file mode 100644 index 000000000..add3cc8de --- /dev/null +++ b/volatility/framework/interfaces/automagic.py @@ -0,0 +1,13 @@ +from abc import ABCMeta, abstractmethod + +from volatility.framework import validity + + +class AutomagicInterface(validity.ValidityRoutines, metaclass = ABCMeta): + """Class that defines an automagic component that can help fulfill a Requirement""" + + priority = 10 + + @abstractmethod + def __call__(self, context, config_path, configurable): + """Runs the automagic over the configurable""" diff --git a/volatility/framework/interfaces/configuration.py b/volatility/framework/interfaces/configuration.py index a312addaf..a9e60d942 100644 --- a/volatility/framework/interfaces/configuration.py +++ b/volatility/framework/interfaces/configuration.py @@ -1,7 +1,5 @@ from abc import ABCMeta, abstractmethod -# We must import interfaces.context this way, since we can't import our parent without cause a loop -from volatility.framework.interfaces import context as interfaces_context from volatility.framework import validity __author__ = 'mike' @@ -9,6 +7,11 @@ __author__ = 'mike' CONFIG_SEPARATOR = "." +def path_join(*args): + """Joins the config paths together""" + return CONFIG_SEPARATOR.join(args) + + class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): """Class to distinguish configuration elements from everything else""" @@ -21,6 +24,7 @@ class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): self._description = description or "" self._default = default self._optional = optional + self._requirements = {} def __repr__(self): return "<" + self.__class__.__name__ + ": " + self.name + ">" @@ -45,71 +49,58 @@ class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): """Whether the option is required for or not""" return self._optional + def config_value(self, context, config_path, default = None): + """Returns the value for this element from its config path""" + return context.config.get(path_join(config_path, self.name), default) + + # Child operations + @property + def requirements(self): + """Returns an iterator of all the child requirements""" + for child in self._requirements: + yield self._requirements[child] + + def add_requirement(self, requirement): + """Adds a child to the list of requirements""" + self._check_type(requirement, RequirementInterface) + self._requirements[requirement.name] = requirement + + def remove_requirement(self, requirement): + """Removes a child from the list of requirements""" + self._check_type(requirement, RequirementInterface) + del self._requirements[requirement.name] + + def validate_children(self, context, config_path): + """Method that will validate all child requirements""" + return all([requirement.validate(context, path_join(config_path, self._name)) for requirement in + self.requirements if not requirement.optional]) + # Validation routines - @abstractmethod - def validate(self, value, context): - """Method to validate the value stored at config_location for the configuration object against a context + def validate(self, context, config_path): + """Method to validate the value stored at config_path for the configuration object against a context - Raises a ValueError based on whether the item is valid or not + Returns False when an item is invalid """ class ConfigurableInterface(validity.ValidityRoutines): """Class to allow objects to have requirements and read configuration data from the context config tree""" - def __init__(self, context, config_path): + def __init__(self, config_path): + """Basic initializer that allows configurables to access their own config settings""" validity.ValidityRoutines.__init__(self) - self._context = self._check_type(context, interfaces_context.ContextInterface) self._config_path = self._check_type(config_path, str) - @property - def context(self): - return self._context - - @property - def config_path(self): - return self._config_path - @classmethod def get_requirements(cls): - """Returns a list of configuration schema nodes for this object""" + """Returns a list of RequirementInterface objects required by this object""" return [] - @property - def config(self): - return self._context.config.branch(self._config_path) - - -class ConstraintInterface(RequirementInterface): - """Class that specifies capabilities that must be provided to succeed""" - - def __init__(self, name, description = None, default = None, optional = False, constraints = None): - if constraints is None: - constraints = {} - RequirementInterface.__init__(self, name, description = description, default = default, optional = optional) - if not self._check_type(constraints, dict): - raise TypeError("Constraints must be a dictionary") - self._constraints = constraints - - @property - def constraints(self): - """Returns a dictionary of requirements that must be met by a provider""" - return self._constraints.copy() - - -class ProviderInterface(ConfigurableInterface): - """Class that allows providers to meet constraints on requirements - - All providers are configurable, but having the interfaces as separate classes - would allow us to disentangle them in the future if necessary. - """ - provides = {} - priority = 10 - @classmethod - def fulfill(cls, context, requirement, config_path): - """Fulfills a context's requirement, altering the context appropriately""" + def validate(cls, context, config_path): + return all([requirement.validate(context, config_path) for requirement in cls.get_requirements() if + not requirement.optional]) class RequirementTreeNode(validity.ValidityRoutines): diff --git a/volatility/framework/interfaces/layers.py b/volatility/framework/interfaces/layers.py index 47ba2fe8f..b5f9197ed 100644 --- a/volatility/framework/interfaces/layers.py +++ b/volatility/framework/interfaces/layers.py @@ -4,11 +4,10 @@ Created on 4 May 2013 @author: mike """ import collections +from abc import ABCMeta, abstractmethod, abstractproperty from volatility.framework import exceptions, validity -# We can't just import interfaces because we'd have a cycle going from volatility.framework.interfaces import configuration, context -from abc import ABCMeta, abstractmethod, abstractproperty class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta): @@ -63,14 +62,15 @@ class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta): """ -class DataLayerInterface(configuration.ProviderInterface, validity.ValidityRoutines, metaclass = ABCMeta): +class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityRoutines, metaclass = ABCMeta): """A Layer that directly holds data (and does not translate it""" provides = {"type": "interface"} def __init__(self, context, config_path, name): - configuration.ProviderInterface.__init__(self, context, config_path) + configuration.ConfigurableInterface.__init__(self, config_path) validity.ValidityRoutines.__init__(self) + self._context = context self._check_type(name, str) self._name = name diff --git a/volatility/framework/interfaces/plugins.py b/volatility/framework/interfaces/plugins.py index cf1872f42..e8afc3714 100644 --- a/volatility/framework/interfaces/plugins.py +++ b/volatility/framework/interfaces/plugins.py @@ -3,11 +3,10 @@ Created on 6 May 2013 @author: mike """ -from abc import abstractmethod, ABCMeta +from abc import ABCMeta, abstractmethod from volatility.framework import validity -from volatility.framework.interfaces import configuration as configuration_interface -from volatility.framework.interfaces import context as context_interface +from volatility.framework.interfaces import configuration as configuration_interface, context as context_interface # @@ -28,23 +27,30 @@ class PluginInterface(configuration_interface.ConfigurableInterface, validity.Va def __init__(self, context, config_path): validity.ValidityRoutines.__init__(self) - configuration_interface.ConfigurableInterface.__init__(self, context, config_path) - self._check_type(context, context_interface.ContextInterface) - # self.validate_inputs() + configuration_interface.ConfigurableInterface.__init__(self, config_path) + self._context = self._check_type(context, context_interface.ContextInterface) + # self.validate() @property def context(self): return self._context + @property + def config(self): + return self._context.config.branch(self._config_path) + @classmethod def get_requirements(cls): """Returns a list of Requirement objects for this plugin""" return [] - def validate_inputs(self): - for option in self.get_schema(): - if not option.optional: - option.validate(self.config.get_value(option.name), self.context) + @classmethod + def validate(self, context, config_path): + print(config_path) + result_set = [requirement.validate(context, config_path) for requirement in self.get_requirements() if + not requirement.optional] + print(result_set) + return all(result_set) @abstractmethod def run(self): diff --git a/volatility/framework/interfaces/symbols.py b/volatility/framework/interfaces/symbols.py index e8ab029cf..4898c86f5 100644 --- a/volatility/framework/interfaces/symbols.py +++ b/volatility/framework/interfaces/symbols.py @@ -144,5 +144,5 @@ class NativeTableInterface(SymbolTableInterface): return [] -class SymbolTableProviderInterface(configuration.ProviderInterface): +class SymbolTableProviderInterface(configuration.ConfigurableInterface): pass