mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-08-31 04:09:40 +02:00
Refactor several configuration/requirement structures to avoid import loops.
This commit is contained in:
Mike Auty
@@ -3,135 +3,4 @@ Created on 7 May 2013
|
||||
|
||||
@author: mike
|
||||
"""
|
||||
import collections
|
||||
import copy
|
||||
import json
|
||||
import logging
|
||||
|
||||
from volatility.framework.configuration.requirements import MultiRequirement
|
||||
from volatility.framework.interfaces.configuration import CONFIG_SEPARATOR
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class HierarchicalDict(collections.Mapping):
|
||||
def __init__(self, initial_dict = None, separator = CONFIG_SEPARATOR):
|
||||
if not (isinstance(separator, str) and len(separator) == 1):
|
||||
raise TypeError("Separator must be a one character string")
|
||||
self._separator = separator
|
||||
self._data = {}
|
||||
self._subdict = {}
|
||||
if isinstance(initial_dict, str):
|
||||
initial_dict = json.loads(initial_dict)
|
||||
if isinstance(initial_dict, dict):
|
||||
for k, v in initial_dict.items():
|
||||
self[k] = v
|
||||
elif initial_dict is not None:
|
||||
raise TypeError("Initial_dict must be a dictionary or JSON string containing a dictionary")
|
||||
|
||||
@property
|
||||
def separator(self):
|
||||
return self._separator
|
||||
|
||||
@property
|
||||
def data(self):
|
||||
return self._data.copy()
|
||||
|
||||
def _key_head(self, key):
|
||||
"""Returns the first division of a key based on the dict separator,
|
||||
or the full key if the separator is not present
|
||||
"""
|
||||
if self.separator in key:
|
||||
return key[:key.index(self.separator)]
|
||||
else:
|
||||
return key
|
||||
|
||||
def _key_tail(self, key):
|
||||
"""Returns all but the first division of a key based on the dict separator,
|
||||
or None if the separator is not in the key
|
||||
"""
|
||||
if self.separator in key:
|
||||
return key[key.index(self.separator) + 1:]
|
||||
return None
|
||||
|
||||
def __iter__(self):
|
||||
"""Returns an iterator object that supports the iterator protocol"""
|
||||
return self.generator()
|
||||
|
||||
def generator(self):
|
||||
"""Yields the next element in the iterator"""
|
||||
for key in self._data:
|
||||
yield key
|
||||
for subdict_key in self._subdict:
|
||||
for key in self._subdict[subdict_key]:
|
||||
yield subdict_key + self.separator + key
|
||||
|
||||
def __getitem__(self, key):
|
||||
"""Gets an item, traversing down the trees to get to the final value"""
|
||||
try:
|
||||
if self.separator in key:
|
||||
subdict = self._subdict[self._key_head(key)]
|
||||
return subdict[self._key_tail(key)]
|
||||
else:
|
||||
return self._data[key]
|
||||
except KeyError:
|
||||
raise KeyError(key)
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
"""Sets an item or creates a subdict and sets the item within that"""
|
||||
self._setitem(key, value)
|
||||
|
||||
def _setitem(self, key, value, is_data = True):
|
||||
"""Set an item or appends a whole subtree at a key location"""
|
||||
if self.separator in key:
|
||||
subdict = self._subdict.get(self._key_head(key), HierarchicalDict(separator = self.separator))
|
||||
subdict[self._key_tail(key)] = value
|
||||
self._subdict[self._key_head(key)] = subdict
|
||||
else:
|
||||
if is_data:
|
||||
self._data[key] = value
|
||||
else:
|
||||
if not isinstance(value, HierarchicalDict) and value is not None:
|
||||
raise TypeError("HierarchicalDicts can only store HierarchicalDicts within their structure")
|
||||
self._subdict[key] = value
|
||||
|
||||
def __delitem__(self, key):
|
||||
"""Deletes an item from the hierarchical dict"""
|
||||
try:
|
||||
if self.separator in key:
|
||||
subdict = self._subdict[self._key_head(key)]
|
||||
del subdict[self._key_tail(key)]
|
||||
if not subdict:
|
||||
del self._subdict[self._key_head(key)]
|
||||
except KeyError:
|
||||
raise KeyError(key)
|
||||
|
||||
def __contains__(self, key):
|
||||
"""Determines whether the key is present in the hierarchy"""
|
||||
if self.separator in key:
|
||||
try:
|
||||
subdict = self._subdict[self._key_head(key)]
|
||||
return self._key_tail(key) in subdict
|
||||
except KeyError:
|
||||
return False
|
||||
else:
|
||||
return key in self._data
|
||||
|
||||
def __len__(self):
|
||||
"""Returns the length of all items"""
|
||||
return len(self._data) + sum([len(subdict) for subdict in self._subdict])
|
||||
|
||||
def branch(self, key):
|
||||
"""Returns the HierarchicalDict housed under the key"""
|
||||
if self.separator in key:
|
||||
return self._subdict[self._key_head(key)].branch(self._key_tail(key))
|
||||
else:
|
||||
return self._subdict[key]
|
||||
|
||||
def clone(self):
|
||||
"""Duplicate the configuration, allowing changes without affecting the original"""
|
||||
return copy.deepcopy(self)
|
||||
|
||||
def __str__(self):
|
||||
"""Turns the Hierarchical dict into a string representation"""
|
||||
return json.dumps(dict([(key, self[key]) for key in self.generator()]), indent = 2)
|
||||
from volatility.framework.configuration import requirements
|
||||
|
||||
@@ -4,6 +4,11 @@ from volatility.framework import interfaces
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
# Allow these two to be imported directly from requirements
|
||||
# This helps prevent import loops since other interfaces need to be able to check instances of this
|
||||
TranslationLayerRequirement = interfaces.configuration.TranslationLayerRequirement
|
||||
SymbolRequirement = interfaces.configuration.SymbolRequirement
|
||||
|
||||
|
||||
class MultiRequirement(interfaces.configuration.RequirementInterface):
|
||||
"""Class to hold multiple requirements
|
||||
@@ -45,108 +50,6 @@ class BytesRequirement(InstanceRequirement):
|
||||
instance_type = bytes
|
||||
|
||||
|
||||
class TranslationLayerRequirement(interfaces.configuration.ConstructableRequirementInterface):
|
||||
"""Class maintaining the limitations on what sort of address spaces are acceptable"""
|
||||
|
||||
def __init__(self, name, description = None, default = None, optional = False):
|
||||
"""Constructs a Translation Layer Requirement
|
||||
|
||||
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:
|
||||
"""
|
||||
super().__init__(name, description, default, optional)
|
||||
|
||||
# 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):
|
||||
"""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.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)!!!
|
||||
|
||||
self._check_class(context, config_path)
|
||||
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"""
|
||||
# Determine the layer name
|
||||
name = self.name
|
||||
counter = 2
|
||||
while name in context.memory:
|
||||
name = self.name + str(counter)
|
||||
counter += 1
|
||||
|
||||
config_path = interfaces.configuration.path_join(config_path, self.name)
|
||||
|
||||
args = {"context": context,
|
||||
"config_path": config_path,
|
||||
"name": name}
|
||||
|
||||
if not all([subreq.validate(context, config_path) for subreq in self.requirements.values() if
|
||||
not subreq.optional]):
|
||||
return False
|
||||
|
||||
obj = self._construct_class(context, config_path, args)
|
||||
if obj is None:
|
||||
return False
|
||||
context.add_layer(obj)
|
||||
return True
|
||||
|
||||
|
||||
class SymbolRequirement(interfaces.configuration.ConstructableRequirementInterface):
|
||||
"""Class maintaining the limitations on what sort of symbol spaces are acceptable"""
|
||||
|
||||
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):
|
||||
vollog.debug("TypeError - SymbolRequirement only accepts string labels")
|
||||
return False
|
||||
if value not in context.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
|
||||
|
||||
def construct(self, context, config_path):
|
||||
"""Constructs the symbol space within the context based on the subrequirements"""
|
||||
# Determine the space name
|
||||
name = self.name
|
||||
if name in context.symbol_space:
|
||||
raise ValueError("Symbol space already contains a SymbolTable by the same name")
|
||||
|
||||
config_path = interfaces.configuration.path_join(config_path, self.name)
|
||||
args = {"context": context,
|
||||
"config_path": config_path,
|
||||
"name": name}
|
||||
|
||||
if not all([subreq.validate(context, config_path) for subreq in self.requirements.values() if
|
||||
not subreq.optional]):
|
||||
return False
|
||||
|
||||
obj = self._construct_class(context, config_path, args)
|
||||
if obj is None:
|
||||
return False
|
||||
context.symbol_space.append(obj)
|
||||
return True
|
||||
|
||||
|
||||
class ChoiceRequirement(interfaces.configuration.RequirementInterface):
|
||||
"""Allows one from a choice of strings"""
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
from volatility.framework import interfaces, symbols
|
||||
from volatility.framework.configuration import HierarchicalDict
|
||||
from volatility.framework.interfaces.configuration import HierarchicalDict
|
||||
|
||||
__author__ = 'mike'
|
||||
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
import collections
|
||||
import copy
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
from abc import ABCMeta, abstractmethod
|
||||
|
||||
@@ -8,6 +12,8 @@ __author__ = 'mike'
|
||||
|
||||
CONFIG_SEPARATOR = "."
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def path_join(*args):
|
||||
"""Joins the config paths together"""
|
||||
@@ -203,3 +209,241 @@ class ConfigurableInterface(validity.ValidityRoutines, metaclass = ABCMeta):
|
||||
def validate(cls, context, config_path):
|
||||
return all([requirement.validate(context, config_path) for requirement in cls.get_requirements() if
|
||||
not requirement.optional])
|
||||
|
||||
|
||||
class HierarchicalDict(collections.Mapping):
|
||||
def __init__(self, initial_dict = None, separator = CONFIG_SEPARATOR):
|
||||
if not (isinstance(separator, str) and len(separator) == 1):
|
||||
raise TypeError("Separator must be a one character string")
|
||||
self._separator = separator
|
||||
self._data = {}
|
||||
self._subdict = {}
|
||||
if isinstance(initial_dict, str):
|
||||
initial_dict = json.loads(initial_dict)
|
||||
if isinstance(initial_dict, dict):
|
||||
for k, v in initial_dict.items():
|
||||
self[k] = v
|
||||
elif initial_dict is not None:
|
||||
raise TypeError("Initial_dict must be a dictionary or JSON string containing a dictionary")
|
||||
|
||||
@property
|
||||
def separator(self):
|
||||
return self._separator
|
||||
|
||||
@property
|
||||
def data(self):
|
||||
return self._data.copy()
|
||||
|
||||
def _key_head(self, key):
|
||||
"""Returns the first division of a key based on the dict separator,
|
||||
or the full key if the separator is not present
|
||||
"""
|
||||
if self.separator in key:
|
||||
return key[:key.index(self.separator)]
|
||||
else:
|
||||
return key
|
||||
|
||||
def _key_tail(self, key):
|
||||
"""Returns all but the first division of a key based on the dict separator,
|
||||
or None if the separator is not in the key
|
||||
"""
|
||||
if self.separator in key:
|
||||
return key[key.index(self.separator) + 1:]
|
||||
return None
|
||||
|
||||
def __iter__(self):
|
||||
"""Returns an iterator object that supports the iterator protocol"""
|
||||
return self.generator()
|
||||
|
||||
def generator(self):
|
||||
"""Yields the next element in the iterator"""
|
||||
for key in self._data:
|
||||
yield key
|
||||
for subdict_key in self._subdict:
|
||||
for key in self._subdict[subdict_key]:
|
||||
yield subdict_key + self.separator + key
|
||||
|
||||
def __getitem__(self, key):
|
||||
"""Gets an item, traversing down the trees to get to the final value"""
|
||||
try:
|
||||
if self.separator in key:
|
||||
subdict = self._subdict[self._key_head(key)]
|
||||
return subdict[self._key_tail(key)]
|
||||
else:
|
||||
return self._data[key]
|
||||
except KeyError:
|
||||
raise KeyError(key)
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
"""Sets an item or creates a subdict and sets the item within that"""
|
||||
self._setitem(key, value)
|
||||
|
||||
def _setitem(self, key, value, is_data = True):
|
||||
"""Set an item or appends a whole subtree at a key location"""
|
||||
if self.separator in key:
|
||||
subdict = self._subdict.get(self._key_head(key), HierarchicalDict(separator = self.separator))
|
||||
subdict[self._key_tail(key)] = value
|
||||
self._subdict[self._key_head(key)] = subdict
|
||||
else:
|
||||
if is_data:
|
||||
self._data[key] = value
|
||||
else:
|
||||
if not isinstance(value, HierarchicalDict) and value is not None:
|
||||
raise TypeError("HierarchicalDicts can only store HierarchicalDicts within their structure")
|
||||
self._subdict[key] = value
|
||||
|
||||
def __delitem__(self, key):
|
||||
"""Deletes an item from the hierarchical dict"""
|
||||
try:
|
||||
if self.separator in key:
|
||||
subdict = self._subdict[self._key_head(key)]
|
||||
del subdict[self._key_tail(key)]
|
||||
if not subdict:
|
||||
del self._subdict[self._key_head(key)]
|
||||
except KeyError:
|
||||
raise KeyError(key)
|
||||
|
||||
def __contains__(self, key):
|
||||
"""Determines whether the key is present in the hierarchy"""
|
||||
if self.separator in key:
|
||||
try:
|
||||
subdict = self._subdict[self._key_head(key)]
|
||||
return self._key_tail(key) in subdict
|
||||
except KeyError:
|
||||
return False
|
||||
else:
|
||||
return key in self._data
|
||||
|
||||
def __len__(self):
|
||||
"""Returns the length of all items"""
|
||||
return len(self._data) + sum([len(subdict) for subdict in self._subdict])
|
||||
|
||||
def branch(self, key):
|
||||
"""Returns the HierarchicalDict housed under the key"""
|
||||
if self.separator in key:
|
||||
return self._subdict[self._key_head(key)].branch(self._key_tail(key))
|
||||
else:
|
||||
return self._subdict[key]
|
||||
|
||||
def splice(self, key, value):
|
||||
"""Splices an existing HierarchicalDictionary under a key"""
|
||||
if not isinstance(key, str) or not isinstance(value, HierarchicalDict):
|
||||
raise TypeError("Splice requires a string key and HierarchicalDict value")
|
||||
self._setitem(key, value, False)
|
||||
|
||||
def clone(self):
|
||||
"""Duplicate the configuration, allowing changes without affecting the original"""
|
||||
return copy.deepcopy(self)
|
||||
|
||||
def __str__(self):
|
||||
"""Turns the Hierarchical dict into a string representation"""
|
||||
return json.dumps(dict([(key, self[key]) for key in self.generator()]), indent = 2)
|
||||
|
||||
|
||||
class TranslationLayerRequirement(ConstructableRequirementInterface):
|
||||
"""Class maintaining the limitations on what sort of address spaces are acceptable"""
|
||||
|
||||
def __init__(self, name, description = None, default = None, optional = False):
|
||||
"""Constructs a Translation Layer Requirement
|
||||
|
||||
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:
|
||||
"""
|
||||
super().__init__(name, description, default, optional)
|
||||
|
||||
# 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):
|
||||
"""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.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)!!!
|
||||
|
||||
self._check_class(context, config_path)
|
||||
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"""
|
||||
# Determine the layer name
|
||||
name = self.name
|
||||
counter = 2
|
||||
while name in context.memory:
|
||||
name = self.name + str(counter)
|
||||
counter += 1
|
||||
|
||||
config_path = path_join(config_path, self.name)
|
||||
|
||||
args = {"context": context,
|
||||
"config_path": config_path,
|
||||
"name": name}
|
||||
|
||||
if not all([subreq.validate(context, config_path) for subreq in self.requirements.values() if
|
||||
not subreq.optional]):
|
||||
return False
|
||||
|
||||
obj = self._construct_class(context, config_path, args)
|
||||
if obj is None:
|
||||
return False
|
||||
context.add_layer(obj)
|
||||
return True
|
||||
|
||||
|
||||
class SymbolRequirement(ConstructableRequirementInterface):
|
||||
"""Class maintaining the limitations on what sort of symbol spaces are acceptable"""
|
||||
|
||||
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):
|
||||
vollog.debug("TypeError - SymbolRequirement only accepts string labels")
|
||||
return False
|
||||
if value not in context.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
|
||||
|
||||
def construct(self, context, config_path):
|
||||
"""Constructs the symbol space within the context based on the subrequirements"""
|
||||
# Determine the space name
|
||||
name = self.name
|
||||
if name in context.symbol_space:
|
||||
raise ValueError("Symbol space already contains a SymbolTable by the same name")
|
||||
|
||||
config_path = path_join(config_path, self.name)
|
||||
args = {"context": context,
|
||||
"config_path": config_path,
|
||||
"name": name}
|
||||
|
||||
if not all([subreq.validate(context, config_path) for subreq in self.requirements.values() if
|
||||
not subreq.optional]):
|
||||
return False
|
||||
|
||||
# Fill out the parameter for class creation
|
||||
cls = self.requirements["class"].cls
|
||||
node_config = context.config.branch(config_path)
|
||||
for req in cls.get_requirements():
|
||||
if req.name in node_config.data and req.name != "class":
|
||||
args[req.name] = node_config.data[req.name]
|
||||
|
||||
obj = self._construct_class(context, config_path, args)
|
||||
if obj is None:
|
||||
return False
|
||||
context.symbol_space.append(obj)
|
||||
return True
|
||||
|
||||
@@ -118,6 +118,8 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR
|
||||
"""Returns a list of Requirement objects for this type of layer"""
|
||||
return []
|
||||
|
||||
# ## General scanning methods
|
||||
|
||||
def _pre_scan(self, context, min_address, max_address, progress_callback, scanner):
|
||||
"""Prepares the scanner based on standard procedures shared between TranslationLayers and DataLayers"""
|
||||
if progress_callback is not None:
|
||||
@@ -205,6 +207,8 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
|
||||
self._context.memory.write(layer, mapped_offset, length)
|
||||
current_offset += length
|
||||
|
||||
# ## Scan implementation with knowledge of pages
|
||||
|
||||
def scan(self, context, scanner, progress_callback = None, min_address = None, max_address = None):
|
||||
"""Scans a Translation layer by chunk
|
||||
|
||||
|
||||
Reference in New Issue
Block a user