mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-08-17 20:35:40 +02:00
Bulk out the config/inputs code.
This commit is contained in:
@@ -36,7 +36,7 @@ def require_version(*args):
|
||||
".".join([str(x) for x in args[0:2]]))
|
||||
|
||||
|
||||
from volatility.framework import interfaces, symbols, layers, contexts
|
||||
from volatility.framework import interfaces, symbols, layers, contexts, config
|
||||
|
||||
|
||||
class Context(interfaces.context.ContextInterface):
|
||||
@@ -59,6 +59,7 @@ class Context(interfaces.context.ContextInterface):
|
||||
interfaces.context.ContextInterface.__init__(self)
|
||||
self._symbol_space = symbols.SymbolSpace(natives)
|
||||
self._memory = layers.Memory()
|
||||
self.config = config.Config()
|
||||
|
||||
# ## Symbol Space Functions
|
||||
|
||||
|
||||
@@ -5,46 +5,87 @@ Created on 7 May 2013
|
||||
"""
|
||||
import re
|
||||
|
||||
from volatility.framework.interfaces.config import GenericRequirement
|
||||
from volatility.framework.interfaces.config import GenericInput, ConfigInterface
|
||||
|
||||
class Config(ConfigInterface):
|
||||
"""Class to hold and provide a namespace for plugins and core options"""
|
||||
def __init__(self):
|
||||
self._namespace = {'core': {}}
|
||||
|
||||
class BooleanRequirement(GenericRequirement):
|
||||
def check_value(self, value, context):
|
||||
if not isinstance(self.value, bool):
|
||||
raise TypeError(self.name + " requirement only accepts a boolean type")
|
||||
def add_item(self, namespace, item):
|
||||
self._type_check(namespace, str)
|
||||
self._type_check(item, GenericInput)
|
||||
subconfig = self._namespace.get(namespace,{})
|
||||
subconfig[item.name] = item
|
||||
self._namespace[namespace] = subconfig
|
||||
|
||||
class TranslationLayerRequirement(GenericRequirement):
|
||||
def __contains__(self, item):
|
||||
return (item in self._namespace)
|
||||
|
||||
def __len__(self):
|
||||
return len(self._namespace)
|
||||
|
||||
class InstanceInput(GenericInput):
|
||||
instance_type = bool
|
||||
|
||||
def validate_input(self, value, context):
|
||||
if not isinstance(value, self.instance_type):
|
||||
raise TypeError(self.name + " input only accepts " + self.instance_type.__name__+ " type")
|
||||
|
||||
class IntInput(InstanceInput):
|
||||
instance_type = int
|
||||
|
||||
class StringInput(InstanceInput):
|
||||
#TODO: Maybe add string length limits?
|
||||
instance_type = str
|
||||
|
||||
class TranslationLayerInput(GenericInput):
|
||||
"""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)
|
||||
GenericInput.__init__(self, name, *args, **kwargs)
|
||||
self.layer_type = layer_type
|
||||
self.os = os_type
|
||||
self.arches = architectures
|
||||
|
||||
def check_value(self, value, context):
|
||||
def validate_input(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 + " is not memory layer")
|
||||
|
||||
|
||||
class ListRequirement(GenericRequirement):
|
||||
class ChoiceInput(GenericInput):
|
||||
"""Allows one from a choice of strings
|
||||
"""
|
||||
def __init__(self, choices, *args, **kwargs):
|
||||
GenericInput.__init__(*args, **kwargs)
|
||||
if not isinstance(choices, list) or any([not isinstance(choice, str) for choice in choices]):
|
||||
raise TypeError("ChoiceInput takes a list of strings as choices")
|
||||
self._choices = choices
|
||||
|
||||
def validate_input(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 ListInput(GenericInput):
|
||||
def __init__(self, min_elements, max_elements, element_type, *args, **kwargs):
|
||||
GenericRequirement.__init__(self, *args, **kwargs)
|
||||
if isinstance(element_type, ListRequirement):
|
||||
raise TypeError("ListRequirements cannot contain ListRequirements")
|
||||
self.element_type = self._type_check(element_type, GenericRequirement)
|
||||
GenericInput.__init__(self, *args, **kwargs)
|
||||
if isinstance(element_type, ListInput):
|
||||
raise TypeError("ListInputs cannot contain ListInputs")
|
||||
self.element_type = self._type_check(element_type, GenericInput)
|
||||
self.min_elements = min_elements
|
||||
self.max_elements = max_elements
|
||||
|
||||
def check_value(self, value, context):
|
||||
def validate_input(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.check_value(element, context) for element in value]
|
||||
[self.element_type.validate_input(element, context) for element in value]
|
||||
|
||||
|
||||
# TODO: OptionTypes such as choice, list and so on
|
||||
|
||||
|
||||
@@ -6,7 +6,7 @@ from volatility.framework import validity
|
||||
__author__ = 'mike'
|
||||
|
||||
|
||||
class GenericRequirement(validity.ValidityRoutines, metaclass = ABCMeta):
|
||||
class GenericInput(validity.ValidityRoutines, metaclass = ABCMeta):
|
||||
"""Class to handle a single specific configuration option"""
|
||||
|
||||
def __init__(self, name, description = None, default = None, optional = False):
|
||||
@@ -16,15 +16,24 @@ class GenericRequirement(validity.ValidityRoutines, metaclass = ABCMeta):
|
||||
self._type_check(name, str)
|
||||
self._name = name
|
||||
self._description = description
|
||||
self.value = None
|
||||
self._value = None
|
||||
self._optional = optional
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self._name
|
||||
def value(self):
|
||||
"""Returns the value or the default if the value is not set"""
|
||||
if self._value is None:
|
||||
return self._default
|
||||
return self._value
|
||||
|
||||
@value.setter
|
||||
def value(self, data):
|
||||
"""Sets the value to that of the input data"""
|
||||
self._value = data
|
||||
|
||||
@property
|
||||
def optional(self):
|
||||
"""Whether the option is required for or not"""
|
||||
return self._optional
|
||||
|
||||
@property
|
||||
@@ -38,11 +47,26 @@ class GenericRequirement(validity.ValidityRoutines, metaclass = ABCMeta):
|
||||
return self._description
|
||||
|
||||
@abstractmethod
|
||||
def check_value(self, value, context):
|
||||
def validate_input(self, value, context):
|
||||
"""Validates the value against a context
|
||||
|
||||
Returns True if the value is valid
|
||||
Throws exceptions if the valid is invalid"""
|
||||
pass
|
||||
|
||||
def validate(self, context):
|
||||
"""Validates the currently set value"""
|
||||
return self.validate_input(self.value, context)
|
||||
|
||||
class ConfigInterface(validity.ValidityRoutines):
|
||||
"""Class to hold and provide a namespace for plugins and core options"""
|
||||
@abstractmethod
|
||||
def add_item(self, namespace, item):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def __contains__(self, item):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def __len__(self):
|
||||
pass
|
||||
@@ -27,6 +27,7 @@ class PluginInterface(validity.ValidityRoutines, metaclass = ABCMeta):
|
||||
def __init__(self, context):
|
||||
self._type_check(context, context_module.ContextInterface)
|
||||
self._context = context
|
||||
self.validate_inputs()
|
||||
|
||||
@property
|
||||
def context(self):
|
||||
@@ -34,13 +35,18 @@ class PluginInterface(validity.ValidityRoutines, metaclass = ABCMeta):
|
||||
|
||||
@abstractmethod
|
||||
@classmethod
|
||||
def requirements(cls):
|
||||
def inputs(cls):
|
||||
"""Returns a list of requirements options"""
|
||||
return []
|
||||
|
||||
def check_requirements(self):
|
||||
for requirement in self.requirements():
|
||||
requirement.check_value(requirement.value, self._context)
|
||||
def get_input(self, name, core = False):
|
||||
if core:
|
||||
return self._context.config.get("core", name)
|
||||
return self._context.config.get(self.__name__, name)
|
||||
|
||||
def validate_inputs(self):
|
||||
for option in self.inputs():
|
||||
option.validate_input(option.value, self.context)
|
||||
|
||||
@abstractmethod
|
||||
def __call__(self):
|
||||
|
||||
@@ -6,9 +6,10 @@ Created on 4 May 2013
|
||||
|
||||
from volatility.framework import validity, interfaces, exceptions
|
||||
from volatility.framework.layers import physical, intel
|
||||
import collections.abc
|
||||
|
||||
|
||||
class Memory(validity.ValidityRoutines):
|
||||
class Memory(validity.ValidityRoutines, collections.abc.Mapping):
|
||||
"""Container for multiple layers of data"""
|
||||
|
||||
def __init__(self):
|
||||
@@ -57,6 +58,12 @@ class Memory(validity.ValidityRoutines):
|
||||
"""Returns the layer of specified name"""
|
||||
return self._layers[name]
|
||||
|
||||
def __len__(self):
|
||||
return len(self._layers)
|
||||
|
||||
def __iter__(self):
|
||||
return iter(self._layers)
|
||||
|
||||
def check_cycles(self):
|
||||
"""Runs through the available layers and identifies if there are cycles in the DAG"""
|
||||
# TODO: Is having a cycle check necessary?
|
||||
|
||||
Reference in New Issue
Block a user