WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-29 13:04:58 +02:00
Code Issues Packages Projects Releases Wiki Activity

Rework configuration classes to move towards a schema system.

Browse Source
This commit is contained in:
Mike Auty
2015-11-30 21:21:15 +00:00
parent d258c56991
commit 58582715db
6 changed files with 89 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility/cli/__init__.py
+1 -1
View File
@@ -65,7 +65,7 @@ class CommandLine(object):
context.config.add_item(req, plugin.__name__)
if isinstance(req, configuration.TranslationLayerRequirement):
# Choose an appropriate LayerFactory (add layer to the req.name so we don't blat the requirement itself
namespace = interfaces.configuration.namespace_join([plugin.__name__, req.name + "_layer"])
namespace = interfaces.configuration.schema_name_join([plugin.__name__, req.name + "_layer"])
factory = self.construct_translation_layer_factory(namespace, req)
req_mapping[req] = factory
for facreq in factory.requirements():
volatility/cli/argparse_adapter.py
+5 -4
View File
@@ -1,9 +1,10 @@
import argparse
from volatility.framework import interfaces, configuration
__author__ = 'mike'
from volatility.framework import interfaces
from volatility.framework import configuration
def StoreItemFactory(config_item):
class StoreItemAction(argparse.Action):
@@ -18,7 +19,7 @@ def StoreItemFactory(config_item):
def adapt_config(config, parser, group = None):
"""Constructs an argument parser based on a volatility configuration"""
if not group and not isinstance(config, interfaces.configuration.ConfigurationGroup):
if not group and not isinstance(config, interfaces.configuration.ConfigurationSchemaGroup):
raise TypeError("adapt_config expects a ConfigurationItem, not a " + type(config).__name__)
for item in flatten_configuration(config):
@@ -32,7 +33,7 @@ def adapt_config(config, parser, group = None):
def flatten_configuration(config):
output = {}
for item in config:
if isinstance(config[item], interfaces.configuration.ConfigurationGroup):
if isinstance(config[item], interfaces.configuration.ConfigurationSchemaGroup):
for k, v in flatten_configuration(config[item]).items():
output[item + "." + k] = v
else:
volatility/framework/configuration.py
+16 -23
View File
@@ -4,15 +4,13 @@ Created on 7 May 2013
@author: mike
"""
from volatility.framework.interfaces.configuration import GenericRequirement
from volatility.framework.interfaces.configuration import ConfigurationSchemaNode, Configurable
# Intentionally reimport namespace_join so it's accessible from the main config module
class InstanceRequirement(GenericRequirement):
class InstanceRequirement(ConfigurationSchemaNode):
instance_type = bool
def validate_input(self, value, context):
def validate(self, value, _context):
if not isinstance(value, self.instance_type):
raise TypeError(self.name + " input only accepts " + self.instance_type.__name__ + " type")
@@ -26,53 +24,48 @@ class StringRequirement(InstanceRequirement):
instance_type = str
class TranslationLayerRequirement(GenericRequirement):
class TranslationLayerRequirement(ConfigurationSchemaNode, Configurable):
"""Class maintaining the limitations on what sort of address spaces are acceptable"""
# TODO: derive acceptable OSes from the address_space information
# TODO: derive acceptable arches from the available layers
def __init__(self, name, layer_type, os_type, architectures, *args, **kwargs):
GenericRequirement.__init__(self, name, *args, **kwargs)
self.layer_type = layer_type
self.os = os_type
self.arches = architectures
# TODO: Add requirements: acceptable OSes from the address_space information
# TODO: Add requirements: acceptable arches from the available layers
def validate_input(self, value, context):
def validate(self, value, context):
"""Validate that the value is a valid layer name and that the layer adheres to the requirements"""
if value not in context.memory:
raise IndexError((value or "") + " is not a memory layer")
class ChoiceRequirement(GenericRequirement):
class ChoiceRequirement(ConfigurationSchemaNode):
"""Allows one from a choice of strings
"""
def __init__(self, choices, *args, **kwargs):
GenericRequirement.__init__(self, *args, **kwargs)
ConfigurationSchemaNode.__init__(self, *args, **kwargs)
if not isinstance(choices, list) or any([not isinstance(choice, str) for choice in choices]):
raise TypeError("ChoiceRequirement takes a list of strings as choices")
self._choices = choices
def validate_input(self, value, context):
def validate(self, value, context):
"""Validates the provided value to ensure it is one of the available choices"""
if value not in self._choices:
raise ValueError("Value is not within the set of available choices")
class ListRequirement(GenericRequirement):
def __init__(self, min_elements, max_elements, element_type, *args, **kwargs):
GenericRequirement.__init__(self, *args, **kwargs)
class ListRequirement(ConfigurationSchemaNode):
def __init__(self, element_type, max_elements, min_elements, *args, **kwargs):
ConfigurationSchemaNode.__init__(self, *args, **kwargs)
if isinstance(element_type, ListRequirement):
raise TypeError("ListRequirements cannot contain ListRequirements")
self.element_type = self._type_check(element_type, GenericRequirement)
self.element_type = self._type_check(element_type, ConfigurationSchemaNode)
self.min_elements = min_elements
self.max_elements = max_elements
def validate_input(self, value, context):
def validate(self, value, context):
"""Check the types on each of the returned values and then call the element type's check for each one"""
self._type_check(value, list)
if not all([self._type_check(element, self.element_type) for element in value]):
raise TypeError("At least one element in the list is not of the correct type.")
if not (self.min_elements <= len(value) <= self.max_elements):
raise TypeError("List option provided more or less elements than allowed.")
[self.element_type.validate_input(element, context) for element in value]
[self.element_type.validate(element, context) for element in value]
volatility/framework/contexts/__init__.py
+4 -4
View File
@@ -34,7 +34,7 @@ class LayerFactory(validity.ValidityRoutines, list):
groups = []
for index in range(len(self)):
modifier = self[index]
group = interfaces.configuration.ConfigurationGroup(modifier.__name__ + str(index))
group = interfaces.configuration.ConfigurationSchemaGroup(modifier.__name__ + str(index))
for req in modifier.requirements():
group.add_item(req)
groups.append(group)
@@ -46,7 +46,7 @@ class LayerFactory(validity.ValidityRoutines, list):
Returns a new context with all appropriate modifications (symbols, layers, etc)
"""
for index in range(len(self)):
namespace = interfaces.configuration.namespace_join([self.name, self[index].__name__ + str(index)])
namespace = interfaces.configuration.schema_name_join([self.name, self[index].__name__ + str(index)])
self[index](namespace).modify_context(context = context)
return context
@@ -71,7 +71,7 @@ class Context(interfaces.context.ContextInterface):
interfaces.context.ContextInterface.__init__(self)
self._symbol_space = symbols.SymbolSpace(natives)
self._memory = layers.Memory()
self._config = interfaces.configuration.ConfigurationGroup(name = 'volatility')
self._config = interfaces.configuration.ConfigurationSchemaGroup(name = 'volatility')
# ## Symbol Space Functions
@@ -82,7 +82,7 @@ class Context(interfaces.context.ContextInterface):
@config.setter
def config(self, value):
if not isinstance(value, interfaces.configuration.ConfigurationGroup):
if not isinstance(value, interfaces.configuration.ConfigurationSchemaGroup):
raise TypeError("Configuration must of type ConfigurationGroup")
self._config = value
volatility/framework/interfaces/configuration.py
+57 -66
View File
@@ -1,101 +1,100 @@
import collections.abc
from abc import ABCMeta, abstractmethod
from volatility.framework import validity
__author__ = 'mike'
NAMESPACE_DIVIDER = "."
SCHEMA_NAME_DIVIDER = "."
def namespace_join(pathlist):
return NAMESPACE_DIVIDER.join(pathlist)
def schema_name_join(pathlist):
return SCHEMA_NAME_DIVIDER.join(pathlist)
class ConfigurationItem(validity.ValidityRoutines):
class ConfigurationSchemaNode(validity.ValidityRoutines):
"""Class to distinguish configuration elements from everything else"""
def __init__(self, name, optional):
def __init__(self, name, description = None, default = None, optional = False):
validity.ValidityRoutines.__init__(self)
self._type_check(name, str)
if NAMESPACE_DIVIDER in name:
raise ValueError("Name cannot contain the namespace divider (" + NAMESPACE_DIVIDER + ")")
if SCHEMA_NAME_DIVIDER in name:
raise ValueError("Name cannot contain the namespace divider (" + SCHEMA_NAME_DIVIDER + ")")
self._name = name
self._description = description or ""
self._default = default
self._optional = optional
self._children = {}
@property
def name(self):
"""The name of the Option."""
return self._name
@abstractmethod
def validate(self, context):
"""Validates the currently set value"""
pass
@property
def description(self):
"""A short description of what the Option is designed to affect or achieve."""
return self._description
@property
def default(self):
"""Returns the default value if one is set"""
return self._default
@property
def optional(self):
"""Whether the option is required for or not"""
return self._optional
# Child operations
class ConfigurationGroup(ConfigurationItem, collections.abc.Mapping):
"""Class to hold and provide a namespace for plugins and core options"""
def __init__(self, name):
ConfigurationItem.__init__(self, name, optional = False)
self._namespace = {}
def add_item(self, item, namespace = None):
if not isinstance(item, ConfigurationItem):
def add_item(self, item):
"""Add a child to the configuration schema"""
if not isinstance(item, ConfigurationSchemaNode):
raise TypeError("Only ConfigurationItem objects can be added to a ConfigurationGroup")
if namespace:
ns_split = namespace.split(NAMESPACE_DIVIDER)
if ns_split[0] not in self:
self._namespace[ns_split[0]] = ConfigurationGroup(ns_split[0])
return self._namespace[ns_split[0]].add_item(item, namespace_join(ns_split[1:]))
self._namespace[item.name] = item
self._children[item.name] = item
def __iter__(self):
return iter(self._namespace)
"""Iterate through all the child configuration schemas"""
return iter(self._children)
def __getitem__(self, item):
"""Returns a single child configuration schema by name"""
self._type_check(item, str)
item_split = item.split(NAMESPACE_DIVIDER)
item_split = item.split(SCHEMA_NAME_DIVIDER)
if len(item_split) > 1:
return self._namespace[item_split[0]][namespace_join(item_split[1:])]
return self._children[item_split[0]][schema_name_join(item_split[1:])]
# Let namespace produce the index error if necessary
return self._namespace[item_split[0]]
def get_value(self, item):
"""Returns the value of the requirement, not the requirement itself"""
return self.get(item).value
return self._children[item_split[0]]
def __contains__(self, item):
item_split = item.split(NAMESPACE_DIVIDER)
"""Determine membership"""
item_split = item.split(SCHEMA_NAME_DIVIDER)
if len(item_split) > 1:
if item_split[0] in self._namespace:
return namespace_join(item_split[1:]) in self._namespace[item_split[0]]
if item_split[0] in self._children:
return schema_name_join(item_split[1:]) in self._children[item_split[0]]
else:
return False
return item in self._namespace
return item in self._children
def __len__(self):
return len(self._namespace)
return len(self._children)
def validate(self, context):
"""Validates the current value, which for groups cannot be set, so always returns True"""
return all([self[subitem].validate(context) for subitem in self._namespace if not self[subitem].optional])
# Validation routines
@abstractmethod
def validate(self, value, context):
"""Method to validate the value for the configuration object against a context
Raises a ValueError if the value provided is invalid for some reason.
"""
class GenericRequirement(ConfigurationItem, metaclass = ABCMeta):
class GenericRequirement(ConfigurationSchemaNode, metaclass = ABCMeta):
"""Class to handle a single specific configuration option"""
def __init__(self, name, description = None, default = None, optional = None):
"""Creates a new option"""
ConfigurationItem.__init__(self, name, optional)
self._default = default
self._description = description
ConfigurationSchemaNode.__init__(self, name, description = description, default = default, optional = optional)
self._value = None
@property
@@ -110,23 +109,15 @@ class GenericRequirement(ConfigurationItem, metaclass = ABCMeta):
"""Sets the value to that of the input data"""
self._value = data
@property
def default(self):
"""Returns the default value if one is set"""
return self._default
@property
def description(self):
"""A short description of what the Option is designed to affect or achieve."""
return self._description
@abstractmethod
def validate_input(self, value, context):
"""Validates the value against a context
Throws exceptions if the valid is invalid"""
pass
def validate(self, context):
def self_validate(self, context):
"""Validates the currently set value"""
return self.validate_input(self.value, context)
return self.validate(self.value, context)
class Configurable(metaclass = ABCMeta):
"""Class to allow objects to have requirements and populate the context config tree"""
@classmethod
@abstractmethod
def get_schema(self):
"""Returns a list of configuration schema nodes for this object"""
volatility/framework/interfaces/plugins.py
+6 -6
View File
@@ -5,8 +5,8 @@ Created on 6 May 2013
"""
from abc import abstractmethod, ABCMeta
from volatility.framework import interfaces
from volatility.framework import validity
from volatility.framework.interfaces import context as interfaces_context
#
@@ -22,11 +22,11 @@ from volatility.framework.interfaces import context as interfaces_context
# The plugin accepts the context and modifies as necessary
# The plugin runs and produces a TreeGrid output
class PluginInterface(validity.ValidityRoutines, metaclass = ABCMeta):
class PluginInterface(validity.ValidityRoutines, interfaces.configuration.Configurable, metaclass = ABCMeta):
"""Class that defines the interface all Plugins must maintain"""
def __init__(self, context):
self._type_check(context, interfaces_context.ContextInterface)
self._type_check(context, interfaces.context.ContextInterface)
self._context = context
self.validate_inputs()
@@ -36,8 +36,8 @@ class PluginInterface(validity.ValidityRoutines, metaclass = ABCMeta):
@classmethod
@abstractmethod
def requirements(cls):
"""Returns a list of requirements options"""
def get_schema(cls):
"""Returns a list of configuration schema items"""
return []
@property
@@ -47,7 +47,7 @@ class PluginInterface(validity.ValidityRoutines, metaclass = ABCMeta):
return self._context.config.get(self.__class__.__name__)
def validate_inputs(self):
for option in self.requirements():
for option in self.get_schema():
if not option.optional:
option.validate_input(self.config.get_value(option.name), self.context)