mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-11 20:27:38 +02:00
Add a third layer to the requirement system
Now: 1. ConfigSchemaNode Class (pre-plugin instantiation) 2. ConfigSchemaNode Instance (plugin instantiated) 3. ConfigurationItem Instance (tree linking schema nodes to configuration value locations)
This commit is contained in:
@@ -10,7 +10,9 @@ from volatility.framework.interfaces.configuration import ConfigurationSchemaNod
|
||||
class InstanceRequirement(ConfigurationSchemaNode):
|
||||
instance_type = bool
|
||||
|
||||
def validate(self, value, _context):
|
||||
def validate(self, value, _context, valid_children = None):
|
||||
# Child_results can be ignored because no instance class should have children
|
||||
# We don't ban child_results in case someone wants to extend this in some meaningful way
|
||||
if not isinstance(value, self.instance_type):
|
||||
raise TypeError(self.name + " input only accepts " + self.instance_type.__name__ + " type")
|
||||
|
||||
@@ -27,15 +29,21 @@ class StringRequirement(InstanceRequirement):
|
||||
class TranslationLayerRequirement(ConfigurationSchemaNode, Configurable):
|
||||
"""Class maintaining the limitations on what sort of address spaces are acceptable"""
|
||||
|
||||
def __init__(self, name, layer_name):
|
||||
def __init__(self, name, description = None, default = None,
|
||||
optional = False, layer_name = None):
|
||||
"""Constructs a Translation Layer Requirement
|
||||
|
||||
:param name:
|
||||
The configuration option's value will be the name of the layer once it exists in the store
|
||||
|
||||
:param name: Name of the configuration requirement
|
||||
:param layer_name: String detailing the expected name of the required layer, this can be None if it is to be randomly generated
|
||||
:return:
|
||||
"""
|
||||
ConfigurationSchemaNode.__init__(name, description, default, optional)
|
||||
Configurable.__init__(self)
|
||||
self._layer_name = layer_name
|
||||
|
||||
@classmethod
|
||||
def get_schema(cls):
|
||||
# Runs through each of the available translation layers, determining what they're capable of,
|
||||
# and adding their requirements (tagged with their class) to a Disjunction config schema node
|
||||
@@ -44,7 +52,7 @@ class TranslationLayerRequirement(ConfigurationSchemaNode, Configurable):
|
||||
# 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, value, context, valid_children = None):
|
||||
"""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")
|
||||
@@ -62,7 +70,7 @@ class ChoiceRequirement(ConfigurationSchemaNode):
|
||||
raise TypeError("ChoiceRequirement takes a list of strings as choices")
|
||||
self._choices = choices
|
||||
|
||||
def validate(self, value, context):
|
||||
def validate(self, value, context, valid_children = None):
|
||||
"""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")
|
||||
@@ -77,14 +85,15 @@ class ListRequirement(ConfigurationSchemaNode):
|
||||
self.min_elements = min_elements
|
||||
self.max_elements = max_elements
|
||||
|
||||
def validate(self, value, context):
|
||||
def validate(self, value, context, valid_children = None):
|
||||
"""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(element, context) for element in value]
|
||||
for element in value:
|
||||
self.element_type.validate(element, context)
|
||||
|
||||
|
||||
class DisjunctionRequirement(ConfigurationSchemaNode):
|
||||
@@ -96,11 +105,14 @@ class DisjunctionRequirement(ConfigurationSchemaNode):
|
||||
for requirement in requirements:
|
||||
self.add_item(requirement)
|
||||
|
||||
def validate(self, value, context):
|
||||
pass
|
||||
def validate(self, value, context, valid_children = None):
|
||||
if valid_children is None:
|
||||
raise ValueError("A Disjunction requirement cannot exist without children")
|
||||
if not any(valid_children):
|
||||
raise ValueError("No valid children to support the Disjunction Requirement")
|
||||
|
||||
|
||||
class ConjunctionRequierment(ConfigurationSchemaNode):
|
||||
class ConjunctionRequirement(ConfigurationSchemaNode):
|
||||
"""Class requiring all of multiple requirements"""
|
||||
|
||||
def __init__(self, requirements, *args, **kwargs):
|
||||
@@ -109,5 +121,8 @@ class ConjunctionRequierment(ConfigurationSchemaNode):
|
||||
for requirement in requirements:
|
||||
self.add_item(requirement)
|
||||
|
||||
def validate(self, value, context):
|
||||
pass
|
||||
def validate(self, value, context, valid_children = None):
|
||||
if valid_children is None:
|
||||
raise ValueError("A Conjunction requirement cannot exist without children")
|
||||
if not all(valid_children):
|
||||
raise ValueError("An invalid child prevents the Conjunction Requirement")
|
||||
|
||||
@@ -28,9 +28,15 @@ SCHEMA_NAME_DIVIDER = "."
|
||||
|
||||
|
||||
def schema_name_join(pathlist):
|
||||
"""Returns the path string of a list of path components for a schema"""
|
||||
return SCHEMA_NAME_DIVIDER.join(pathlist)
|
||||
|
||||
|
||||
def schema_name_split(path):
|
||||
"""Returns the path components of a schema name"""
|
||||
return path.split(SCHEMA_NAME_DIVIDER)
|
||||
|
||||
|
||||
class ConfigurationSchemaNode(validity.ValidityRoutines):
|
||||
"""Class to distinguish configuration elements from everything else"""
|
||||
|
||||
@@ -69,8 +75,7 @@ class ConfigurationSchemaNode(validity.ValidityRoutines):
|
||||
|
||||
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")
|
||||
self._type_check(item, ConfigurationSchemaNode)
|
||||
self._children[item.name] = item
|
||||
|
||||
def __iter__(self):
|
||||
@@ -102,37 +107,42 @@ class ConfigurationSchemaNode(validity.ValidityRoutines):
|
||||
# Validation routines
|
||||
|
||||
@abstractmethod
|
||||
def validate(self, config_location, context):
|
||||
def validate(self, value, context, valid_children = None):
|
||||
"""Method to validate the value stored at config_location for the configuration object against a context
|
||||
This must validate its own children (so that conjunction/disjunction can work)
|
||||
|
||||
Raises a ValueError if the value provided is invalid for some reason.
|
||||
Raises a ValueError based on whether the item is valid or not
|
||||
"""
|
||||
|
||||
|
||||
class GenericRequirement(ConfigurationSchemaNode, metaclass = ABCMeta):
|
||||
"""Class to handle a single specific configuration option"""
|
||||
class ConfigurationItem(validity.ValidityRoutines):
|
||||
"""Class for wrapping ConfigurationSchemaNodes to create specific configuration items"""
|
||||
|
||||
def __init__(self, name, description = None, default = None, optional = None):
|
||||
"""Creates a new option"""
|
||||
ConfigurationSchemaNode.__init__(self, name, description = description, default = default, optional = optional)
|
||||
self._value = None
|
||||
def __init__(self, schema, config_location_prefix = None):
|
||||
""""""
|
||||
validity.ValidityRoutines.__init__(self)
|
||||
self._schema = self._type_check(schema, ConfigurationSchemaNode)
|
||||
if config_location_prefix is None:
|
||||
config_location_prefix = "core"
|
||||
self._config_location = schema_name_join(
|
||||
schema_name_split(self._type_check(config_location_prefix, str)) +
|
||||
[self._schema.name])
|
||||
|
||||
@property
|
||||
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
|
||||
self._children = []
|
||||
|
||||
@value.setter
|
||||
def value(self, data):
|
||||
"""Sets the value to that of the input data"""
|
||||
self._value = data
|
||||
for child in self._schema:
|
||||
self._children.append(
|
||||
ConfigurationItem(child, self._config_location))
|
||||
|
||||
def self_validate(self, context):
|
||||
"""Validates the currently set value"""
|
||||
return self.validate(self.value, context)
|
||||
def valid(self, context):
|
||||
child_results = []
|
||||
for child in self._children:
|
||||
child_results.append(child.valid(context))
|
||||
try:
|
||||
self._schema.validate(context.config[self._config_location], context, child_results)
|
||||
return True
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
class Configurable(metaclass = ABCMeta):
|
||||
|
||||
Reference in New Issue
Block a user