diff --git a/volatility/cli/__init__.py b/volatility/cli/__init__.py index 2ca3cac2e..510eb4190 100644 --- a/volatility/cli/__init__.py +++ b/volatility/cli/__init__.py @@ -126,11 +126,10 @@ class CommandLine(object): # Clever magic figures out how to fulfill each requirement that might not be fulfilled 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, plugin_config_path): - raise RuntimeError("Unable to validate the plugin configuration") + unsatisfied = plugin.unsatisfied(ctx, plugin_config_path) + if unsatisfied: + raise RuntimeError("Unable to validate the plugin configuration: {}".format(unsatisfied)) print("\n\n") diff --git a/volatility/framework/automagic/construct_layers.py b/volatility/framework/automagic/construct_layers.py index 193ac048e..e761c507e 100644 --- a/volatility/framework/automagic/construct_layers.py +++ b/volatility/framework/automagic/construct_layers.py @@ -19,23 +19,23 @@ class ConstructionMagic(interfaces.automagic.AutomagicInterface): priority = 0 def __call__(self, context, config_path, requirement, progress_callback = None, optional = False): - if not requirement.validate(context, config_path): + result = [] + if requirement.unsatisfied(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 subreq_config_path = interfaces.configuration.path_join(config_path, requirement.name) for subreq in requirement.requirements.values(): self(context, subreq_config_path, subreq, optional or subreq.optional) - valid = subreq.validate(context, subreq_config_path) + invalid = subreq.unsatisfied(context, subreq_config_path) # We want to traverse optional paths, so don't check until we've tried to validate # We also don't want to emit a debug message when a parent is optional, hence the optional parameter - if not valid and not (optional or subreq.optional): + if invalid and not (optional or subreq.optional): vollog.log(constants.LOGLEVEL_V, "Failed on requirement: {}".format(subreq_config_path)) - success = False - if not success: - return False + result.append(interfaces.configuration.path_join(subreq_config_path, subreq.name)) + if result: + return result elif isinstance(requirement, interfaces.configuration.ConstructableRequirementInterface): # We know all the subrequirements are filled, so let's populate requirement.construct(context, config_path) - return True + return [] diff --git a/volatility/framework/automagic/pdbscan.py b/volatility/framework/automagic/pdbscan.py index 596e31647..e59475753 100644 --- a/volatility/framework/automagic/pdbscan.py +++ b/volatility/framework/automagic/pdbscan.py @@ -175,7 +175,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): results = [] if isinstance(requirement, interfaces.configuration.SymbolRequirement): # TODO: check if this is a windows symbol requirement, otherwise ignore it - if not requirement.validate(context, config_path): + if requirement.unsatisfied(context, config_path): results.append((config_path, sub_config_path, requirement)) else: for subreq in requirement.requirements.values(): diff --git a/volatility/framework/automagic/stacker.py b/volatility/framework/automagic/stacker.py index ca5d6a009..334fe99d9 100644 --- a/volatility/framework/automagic/stacker.py +++ b/volatility/framework/automagic/stacker.py @@ -39,12 +39,13 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): """Runs the automagic over the configurable""" # Quick exit if we're not needed - if requirement.validate(context, config_path): + if not requirement.unsatisfied(context, config_path): return # Bow out quickly if the UI hasn't provided a single_location - if not self.validate(self.context, self.config_path): - return + unsatisfied = self.unsatisfied(self.context, self.config_path) + if unsatisfied: + return unsatisfied location = self.config["single_location"] self._check_type(location, str) self._check_type(requirement, interfaces.configuration.RequirementInterface) @@ -111,11 +112,11 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): :rtype: (str, str)""" child_config_path = interfaces.configuration.path_join(config_path, requirement.name) if isinstance(requirement, interfaces.configuration.TranslationLayerRequirement): - if not requirement.validate(context, config_path): + if requirement.unsatisfied(context, config_path): original_setting = context.config.get(child_config_path, None) for layer_name in stacked_layers: context.config[child_config_path] = layer_name - if requirement.validate(context, config_path): + if not requirement.unsatisfied(context, config_path): return child_config_path, layer_name else: # Clean-up to restore the config diff --git a/volatility/framework/automagic/windows.py b/volatility/framework/automagic/windows.py index 1861661fb..583bfaf8d 100644 --- a/volatility/framework/automagic/windows.py +++ b/volatility/framework/automagic/windows.py @@ -234,7 +234,7 @@ class WintelHelper(interfaces.automagic.AutomagicInterface): # Determine if a class has been chosen # Once an appropriate class has been chosen, attempt to determine the page_map_offset value if ("memory_layer" in requirement.requirements and - requirement.requirements["memory_layer"].validate(context, sub_config_path)): + not requirement.requirements["memory_layer"].unsatisfied(context, sub_config_path)): physical_layer = requirement.requirements["memory_layer"].config_value(context, sub_config_path) hits = context.memory[physical_layer].scan(context, PageMapScanner(useful), progress_callback) for test, dtb in hits: diff --git a/volatility/framework/configuration/requirements.py b/volatility/framework/configuration/requirements.py index 88ec6392f..6b6db64d0 100644 --- a/volatility/framework/configuration/requirements.py +++ b/volatility/framework/configuration/requirements.py @@ -24,8 +24,8 @@ class MultiRequirement(interfaces_configuration.RequirementInterface): 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) + def unsatisfied(self, context, config_path): + return self.unsatisfied_children(context, config_path) class BooleanRequirement(interfaces_configuration.InstanceRequirement): @@ -63,13 +63,13 @@ class ChoiceRequirement(interfaces_configuration.RequirementInterface): raise TypeError("ChoiceRequirement takes a list of strings as choices") self.choices = choices - def validate(self, context, config_path): + def unsatisfied(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: vollog.log(constants.LOGLEVEL_V, "ValueError - Value is not within the set of available choices") - return False - return True + return [interfaces_configuration.path_join(config_path, self.name)] + return [] class ListRequirement(interfaces_configuration.RequirementInterface): @@ -99,14 +99,19 @@ class ListRequirement(interfaces_configuration.RequirementInterface): self.min_elements = min_elements self.max_elements = max_elements - def validate(self, context, config_path): + def unsatisfied(self, context, config_path): """Check the types on each of the returned values and their number 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 (self.min_elements <= len(value) <= self.max_elements): vollog.log(constants.LOGLEVEL_V, "TypeError - List option provided more or less elements than allowed.") - return False + return [interfaces_configuration.path_join(config_path, self.name)] if not all([self._check_type(element, self.element_type) for element in value]): vollog.log(constants.LOGLEVEL_V, "TypeError - At least one element in the list is not of the correct type.") - return False - return all([self.element_type.validate(context, element) for element in value]) + return [interfaces_configuration.path_join(config_path, self.name)] + result = [] + for element in value: + subresult = self.element_type.unsatisfied(context, element) + for subvalue in subresult: + result.append(interfaces_configuration.path_join(config_path, subvalue)) + return result diff --git a/volatility/framework/interfaces/configuration.py b/volatility/framework/interfaces/configuration.py index fda0cb7ee..e5fc92c23 100644 --- a/volatility/framework/interfaces/configuration.py +++ b/volatility/framework/interfaces/configuration.py @@ -263,17 +263,22 @@ class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): self._check_type(requirement, RequirementInterface) del self._requirements[requirement.name] - def validate_children(self, context, config_path): + def unsatisfied_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.values() if not requirement.optional]) + result = [] + for requirement in self.requirements.values(): + if not requirement.optional: + subresult = requirement.unsatisfied(context, path_join(config_path, self._name)) + for value in subresult: + result.append(path_join(config_path, value)) + return result # Validation routines @abstractmethod - def validate(self, context, config_path): + def unsatisfied(self, context, config_path): """Method to validate the value stored at config_path for the configuration object against a context - Returns False when an item is invalid + Returns a list containing its own name (or multiple unsatisfied requirement names) when invalid """ @@ -289,7 +294,7 @@ class InstanceRequirement(RequirementInterface): """Always raises a TypeError as instance requirements cannot have children""" raise TypeError("Instance Requirements cannot have subrequirements") - def validate(self, context, config_path): + def unsatisfied(self, context, config_path): """Validates the instance requirement based upon its `instance_type`.""" value = self.config_value(context, config_path, None) if not isinstance(value, self.instance_type): @@ -297,8 +302,8 @@ class InstanceRequirement(RequirementInterface): "TypeError - {} requirements only accept {} type: {}".format(self.name, self.instance_type.__name__, value)) - return False - return True + return [path_join(config_path, self.name)] + return [] class ClassRequirement(RequirementInterface): @@ -313,7 +318,7 @@ class ClassRequirement(RequirementInterface): def cls(self): return self._cls - def validate(self, context, config_path): + def unsatisfied(self, context, config_path): """Checks to see if a class can be recovered""" value = self.config_value(context, config_path, None) self._cls = None @@ -327,7 +332,9 @@ class ClassRequirement(RequirementInterface): else: if value in globals(): self._cls = globals()[value] - return self._cls is not None + if self._cls is None: + return [path_join(config_path, self.name)] + return [] class ConstructableRequirementInterface(RequirementInterface): @@ -349,13 +356,13 @@ class ConstructableRequirementInterface(RequirementInterface): def construct(self, context, config_path): """Method for constructing within the context any required elements from subrequirements""" - def _check_class(self, context, config_path): + def _validate_class(self, context, config_path): """Method to check if the class Requirement 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 = path_join(config_path, self.name) - if class_req.validate(context, subreq_config_path): + if not class_req.unsatisfied(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, ConfigurableInterface): # In case the class has changed, clear out the old requirements @@ -446,9 +453,24 @@ class ConfigurableInterface(validity.ValidityRoutines, metaclass = ABCMeta): return [] @classmethod - def validate(cls, context, config_path): - return all([requirement.validate(context, config_path) for requirement in cls.get_requirements() if - not requirement.optional]) + def unsatisfied(cls, context, config_path): + """Returns a list of the names of all unsatisfied requirements + + Since a satisfied set of requirements will return [], it can be used in tests as follows: + + .. code-block:: python + + unmet = configurable.unsatisfied(context, config_path) + if unmet: + raise RuntimeError("Unsatisfied requirements: {}".format(unmet) + """ + result = [] + for requirement in cls.get_requirements(): + if not requirement.optional: + subresult = requirement.unsatisfied(context, config_path) + for value in subresult: + result.append(path_join(config_path, value)) + return result class TranslationLayerRequirement(ConstructableRequirementInterface): @@ -468,28 +490,28 @@ class TranslationLayerRequirement(ConstructableRequirementInterface): # TODO: Add requirements: acceptable OSes from the address_space information # TODO: Add requirements: acceptable arches from the available layers - def validate(self, context, config_path): + def unsatisfied(self, context, config_path): """Validate that the value is a valid layer name and that the layer adheres to the requirements""" value = self.config_value(context, config_path, None) if isinstance(value, str): if value not in context.memory: vollog.log(9, "IndexError - Layer not found in memory space: {}".format(value)) - return False - return True + return [path_join(config_path, self.name)] + return [] if value is not None: vollog.log(constants.LOGLEVEL_V, "TypeError - Translation Layer Requirement only accepts string labels: {}".format(value)) - return False + return [path_join(config_path, self.name)] # 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)!!! - self._check_class(context, config_path) + self._validate_class(context, config_path) vollog.log(constants.LOGLEVEL_V, "IndexError - No configuration provided: {}".format(config_path + CONFIG_SEPARATOR + self.name)) - return False + return [path_join(config_path, self.name)] def construct(self, context, config_path): """Constructs the appropriate layer and adds it based on the class parameter""" @@ -506,8 +528,8 @@ class TranslationLayerRequirement(ConstructableRequirementInterface): "config_path": config_path, "name": name} - if not all([subreq.validate(context, config_path) for subreq in self.requirements.values() if - not subreq.optional]): + if any([subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if + not subreq.optional]): return False obj = self._construct_class(context, config_path, args) @@ -520,19 +542,19 @@ class TranslationLayerRequirement(ConstructableRequirementInterface): class SymbolRequirement(ConstructableRequirementInterface): """Class maintaining the limitations on what sort of symbol spaces are acceptable""" - def validate(self, context, config_path): + def unsatisfied(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): vollog.log(constants.LOGLEVEL_V, "TypeError - SymbolRequirement only accepts string labels: {}".format(value)) - return False + return [path_join(config_path, self.name)] if value not in context.symbol_space: # This is an expected situation, so return False rather than raise vollog.log(constants.LOGLEVEL_V, "IndexError - Value not present in the symbol space: {}".format(value or "")) - return False - return True + return [path_join(config_path, self.name)] + return [] def construct(self, context, config_path): """Constructs the symbol space within the context based on the subrequirements""" @@ -546,8 +568,8 @@ class SymbolRequirement(ConstructableRequirementInterface): "config_path": config_path, "name": name} - if not all([subreq.validate(context, config_path) for subreq in self.requirements.values() if - not subreq.optional]): + if any([subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if + not subreq.optional]): return False # Fill out the parameter for class creation diff --git a/volatility/framework/interfaces/plugins.py b/volatility/framework/interfaces/plugins.py index c3243a164..30d779b55 100644 --- a/volatility/framework/interfaces/plugins.py +++ b/volatility/framework/interfaces/plugins.py @@ -38,7 +38,7 @@ class PluginInterface(interfaces_configuration.ConfigurableInterface, validity.V super().__init__(context, config_path) # Plugins self validate on construction, it makes it more difficult to work with them, but then # the validation doesn't need to be repeated over and over again by externals - if not self.validate(context, config_path): + if self.unsatisfied(context, config_path): vollog.warning("Plugin failed validation") raise exceptions.PluginRequirementException("The plugin configuration failed to validate") @@ -47,13 +47,6 @@ class PluginInterface(interfaces_configuration.ConfigurableInterface, validity.V """Returns a list of Requirement objects for this plugin""" return [] - @classmethod - def validate(cls, context, config_path): - """Ensures that the plugin's requirements have been met appropriately""" - result_set = [(config_path + "." + requirement.name, requirement.validate(context, config_path)) for requirement - in cls.get_requirements() if not requirement.optional] - return all([r for _, r in result_set]) - @abstractmethod def run(self): """Executes the functionality of the code