mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-08 10:47:38 +02:00
Refactor where certain types of requirement live (given they're more concrete than interface).
This commit is contained in:
@@ -3,6 +3,7 @@ import typing
|
|||||||
|
|
||||||
from volatility.framework import interfaces, constants, validity
|
from volatility.framework import interfaces, constants, validity
|
||||||
from volatility.framework.automagic import linux_symbol_cache
|
from volatility.framework.automagic import linux_symbol_cache
|
||||||
|
from volatility.framework.configuration import requirements
|
||||||
from volatility.framework.layers import intel, scanners
|
from volatility.framework.layers import intel, scanners
|
||||||
from volatility.framework.symbols import linux
|
from volatility.framework.symbols import linux
|
||||||
|
|
||||||
@@ -34,16 +35,15 @@ class LinuxSymbolFinder(interfaces.automagic.AutomagicInterface):
|
|||||||
progress_callback: validity.ProgressCallback = None) -> None:
|
progress_callback: validity.ProgressCallback = None) -> None:
|
||||||
"""Searches for LinuxSymbolRequirements and attempt to populate them"""
|
"""Searches for LinuxSymbolRequirements and attempt to populate them"""
|
||||||
self._requirements = self.find_requirements(context, config_path, requirement,
|
self._requirements = self.find_requirements(context, config_path, requirement,
|
||||||
(interfaces.configuration.TranslationLayerRequirement,
|
(requirements.TranslationLayerRequirement,
|
||||||
interfaces.configuration.SymbolRequirement),
|
requirements.SymbolRequirement),
|
||||||
shortcut = False)
|
shortcut = False)
|
||||||
|
|
||||||
for (path, sub_path, requirement) in self._requirements:
|
for (path, sub_path, requirement) in self._requirements:
|
||||||
if (isinstance(requirement, interfaces.configuration.SymbolRequirement) and
|
if (isinstance(requirement, requirements.SymbolRequirement) and requirement.unsatisfied(context, path)):
|
||||||
requirement.unsatisfied(context, path)):
|
|
||||||
for (tl_path, tl_sub_path, tl_requirement) in self._requirements:
|
for (tl_path, tl_sub_path, tl_requirement) in self._requirements:
|
||||||
# Find the TranslationLayer sibling to the SymbolRequirement
|
# Find the TranslationLayer sibling to the SymbolRequirement
|
||||||
if (isinstance(tl_requirement, interfaces.configuration.TranslationLayerRequirement) and
|
if (isinstance(tl_requirement, requirements.TranslationLayerRequirement) and
|
||||||
tl_path == path):
|
tl_path == path):
|
||||||
if context.config.get(tl_sub_path, None):
|
if context.config.get(tl_sub_path, None):
|
||||||
self._banner_scan(context, path, requirement, context.config[tl_sub_path],
|
self._banner_scan(context, path, requirement, context.config[tl_sub_path],
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ import struct
|
|||||||
import typing
|
import typing
|
||||||
|
|
||||||
from volatility.framework import exceptions, layers, validity, constants
|
from volatility.framework import exceptions, layers, validity, constants
|
||||||
|
from volatility.framework.configuration import requirements
|
||||||
from volatility.framework.layers import scanners, intel
|
from volatility.framework.layers import scanners, intel
|
||||||
from volatility.framework.symbols import intermed, native
|
from volatility.framework.symbols import intermed, native
|
||||||
|
|
||||||
@@ -154,7 +155,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
|
|||||||
"""
|
"""
|
||||||
sub_config_path = interfaces.configuration.path_join(config_path, requirement.name)
|
sub_config_path = interfaces.configuration.path_join(config_path, requirement.name)
|
||||||
results = {} # type: typing.Dict[str, typing.Iterable]
|
results = {} # type: typing.Dict[str, typing.Iterable]
|
||||||
if isinstance(requirement, interfaces.configuration.TranslationLayerRequirement):
|
if isinstance(requirement, requirements.TranslationLayerRequirement):
|
||||||
# Check for symbols in this layer
|
# Check for symbols in this layer
|
||||||
# FIXME: optionally allow a full (slow) scan
|
# FIXME: optionally allow a full (slow) scan
|
||||||
# FIXME: Determine the physical layer no matter the virtual layer
|
# FIXME: Determine the physical layer no matter the virtual layer
|
||||||
@@ -325,8 +326,10 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
|
|||||||
if "pdbscan" not in context.symbol_space:
|
if "pdbscan" not in context.symbol_space:
|
||||||
context.symbol_space.append(native.NativeTable("pdbscan", native.std_ctypes))
|
context.symbol_space.append(native.NativeTable("pdbscan", native.std_ctypes))
|
||||||
# TODO: check if this is a windows symbol requirement, otherwise ignore it
|
# TODO: check if this is a windows symbol requirement, otherwise ignore it
|
||||||
self._symbol_requirements = self.find_requirements(context, config_path, requirement,
|
self._symbol_requirements = self.find_requirements(context,
|
||||||
interfaces.configuration.SymbolRequirement)
|
config_path,
|
||||||
|
requirement,
|
||||||
|
requirements.SymbolRequirement)
|
||||||
for symbol_req_config_path, _, symbol_req in self._symbol_requirements:
|
for symbol_req_config_path, _, symbol_req in self._symbol_requirements:
|
||||||
if symbol_req.unsatisfied(context, symbol_req_config_path):
|
if symbol_req.unsatisfied(context, symbol_req_config_path):
|
||||||
potential_kernels = self.recurse_pdb_finder(context, config_path, requirement, progress_callback)
|
potential_kernels = self.recurse_pdb_finder(context, config_path, requirement, progress_callback)
|
||||||
|
|||||||
@@ -11,8 +11,8 @@ import logging
|
|||||||
import traceback
|
import traceback
|
||||||
import typing
|
import typing
|
||||||
|
|
||||||
import volatility
|
from volatility import framework
|
||||||
from volatility.framework import configuration, interfaces, constants, validity
|
from volatility.framework import interfaces, constants, validity
|
||||||
from volatility.framework.automagic import construct_layers
|
from volatility.framework.automagic import construct_layers
|
||||||
from volatility.framework.configuration import requirements
|
from volatility.framework.configuration import requirements
|
||||||
from volatility.framework.layers import physical
|
from volatility.framework.layers import physical
|
||||||
@@ -105,7 +105,7 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
|
|||||||
# Repeatedly apply "determine what this is" code and build as much up as possible
|
# Repeatedly apply "determine what this is" code and build as much up as possible
|
||||||
stacked = True
|
stacked = True
|
||||||
stacked_layers = [current_layer_name]
|
stacked_layers = [current_layer_name]
|
||||||
stack_set = sorted(volatility.framework.class_subclasses(interfaces.automagic.StackerLayerInterface),
|
stack_set = sorted(framework.class_subclasses(interfaces.automagic.StackerLayerInterface),
|
||||||
key = lambda x: x.stack_order)
|
key = lambda x: x.stack_order)
|
||||||
while stacked:
|
while stacked:
|
||||||
stacked = False
|
stacked = False
|
||||||
@@ -164,7 +164,7 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
|
|||||||
or None if suitable requirements are not found
|
or None if suitable requirements are not found
|
||||||
"""
|
"""
|
||||||
child_config_path = interfaces.configuration.path_join(config_path, requirement.name)
|
child_config_path = interfaces.configuration.path_join(config_path, requirement.name)
|
||||||
if isinstance(requirement, interfaces.configuration.TranslationLayerRequirement):
|
if isinstance(requirement, requirements.TranslationLayerRequirement):
|
||||||
if requirement.unsatisfied(context, config_path):
|
if requirement.unsatisfied(context, config_path):
|
||||||
original_setting = context.config.get(child_config_path, None)
|
original_setting = context.config.get(child_config_path, None)
|
||||||
for layer_name in stacked_layers:
|
for layer_name in stacked_layers:
|
||||||
|
|||||||
@@ -380,10 +380,10 @@ class WinSwapLayers(interfaces.automagic.AutomagicInterface):
|
|||||||
"""Finds translation layers that can have swap layers added"""
|
"""Finds translation layers that can have swap layers added"""
|
||||||
path_join = interfaces.configuration.path_join
|
path_join = interfaces.configuration.path_join
|
||||||
self._translation_requirement = self.find_requirements(context, config_path, requirement,
|
self._translation_requirement = self.find_requirements(context, config_path, requirement,
|
||||||
interfaces.configuration.TranslationLayerRequirement,
|
requirements.TranslationLayerRequirement,
|
||||||
shortcut = False)
|
shortcut = False)
|
||||||
for trans_config, trans_sub_config, trans_req in self._translation_requirement:
|
for trans_config, trans_sub_config, trans_req in self._translation_requirement:
|
||||||
if not isinstance(trans_req, interfaces.configuration.TranslationLayerRequirement):
|
if not isinstance(trans_req, requirements.TranslationLayerRequirement):
|
||||||
# We need this so the type-checker knows we're a TranslationLayerRequirement
|
# We need this so the type-checker knows we're a TranslationLayerRequirement
|
||||||
continue
|
continue
|
||||||
swap_config, swap_sub_config, swap_req = self.find_swap_requirement(trans_config, trans_sub_config,
|
swap_config, swap_sub_config, swap_req = self.find_swap_requirement(trans_config, trans_sub_config,
|
||||||
@@ -404,9 +404,9 @@ class WinSwapLayers(interfaces.automagic.AutomagicInterface):
|
|||||||
counter += 1
|
counter += 1
|
||||||
|
|
||||||
# Add the requirement
|
# Add the requirement
|
||||||
new_req = interfaces.configuration.TranslationLayerRequirement(name = current_layer_name,
|
new_req = requirements.TranslationLayerRequirement(name = current_layer_name,
|
||||||
description = "Swap Layer",
|
description = "Swap Layer",
|
||||||
optional = False)
|
optional = False)
|
||||||
swap_req.add_requirement(new_req)
|
swap_req.add_requirement(new_req)
|
||||||
|
|
||||||
context.config[path_join(swap_sub_config, 'number_of_elements')] = counter
|
context.config[path_join(swap_sub_config, 'number_of_elements')] = counter
|
||||||
@@ -417,7 +417,7 @@ class WinSwapLayers(interfaces.automagic.AutomagicInterface):
|
|||||||
def find_swap_requirement(self,
|
def find_swap_requirement(self,
|
||||||
config: str,
|
config: str,
|
||||||
sub_config: str,
|
sub_config: str,
|
||||||
requirement: interfaces.configuration.TranslationLayerRequirement) \
|
requirement: requirements.TranslationLayerRequirement) \
|
||||||
-> typing.Tuple[str, str, typing.Optional[requirements.LayerListRequirement]]:
|
-> typing.Tuple[str, str, typing.Optional[requirements.LayerListRequirement]]:
|
||||||
"""Takes a Translation layer and returns its swap_layer requirement"""
|
"""Takes a Translation layer and returns its swap_layer requirement"""
|
||||||
swap_req = None
|
swap_req = None
|
||||||
|
|||||||
@@ -9,39 +9,34 @@ import logging
|
|||||||
import typing
|
import typing
|
||||||
|
|
||||||
from volatility.framework import interfaces, constants
|
from volatility.framework import interfaces, constants
|
||||||
from volatility.framework.interfaces import configuration as interfaces_configuration
|
from volatility.framework.interfaces import configuration
|
||||||
|
|
||||||
vollog = logging.getLogger(__name__)
|
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(configuration.RequirementInterface):
|
||||||
class MultiRequirement(interfaces_configuration.RequirementInterface):
|
|
||||||
"""Class to hold multiple requirements.
|
"""Class to hold multiple requirements.
|
||||||
|
|
||||||
Technically the Interface could handle this, but it's an interface, so this is a concrete implementation.
|
Technically the Interface could handle this, but it's an interface, so this is a concrete implementation.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def unsatisfied(self,
|
def unsatisfied(self,
|
||||||
context: interfaces_configuration.ContextInterface,
|
context: configuration.ContextInterface,
|
||||||
config_path: str) -> typing.List[str]:
|
config_path: str) -> typing.List[str]:
|
||||||
return self.unsatisfied_children(context, config_path)
|
return self.unsatisfied_children(context, config_path)
|
||||||
|
|
||||||
|
|
||||||
class BooleanRequirement(interfaces_configuration.SimpleTypeRequirement):
|
class BooleanRequirement(configuration.SimpleTypeRequirement):
|
||||||
"""A requirement type that contains a boolean value"""
|
"""A requirement type that contains a boolean value"""
|
||||||
# Note, this must be a separate class in order to differentiate between Booleans and other instance requirements
|
# Note, this must be a separate class in order to differentiate between Booleans and other instance requirements
|
||||||
|
|
||||||
|
|
||||||
class IntRequirement(interfaces_configuration.SimpleTypeRequirement):
|
class IntRequirement(configuration.SimpleTypeRequirement):
|
||||||
"""A requirement type that contains a single integer"""
|
"""A requirement type that contains a single integer"""
|
||||||
instance_type = int # type: typing.ClassVar[typing.Type]
|
instance_type = int # type: typing.ClassVar[typing.Type]
|
||||||
|
|
||||||
|
|
||||||
class StringRequirement(interfaces_configuration.SimpleTypeRequirement):
|
class StringRequirement(configuration.SimpleTypeRequirement):
|
||||||
"""A requirement type that contains a single unicode string"""
|
"""A requirement type that contains a single unicode string"""
|
||||||
# TODO: Maybe add string length limits?
|
# TODO: Maybe add string length limits?
|
||||||
instance_type = str # type: typing.ClassVar[typing.Type]
|
instance_type = str # type: typing.ClassVar[typing.Type]
|
||||||
@@ -52,12 +47,12 @@ class URIRequirement(StringRequirement):
|
|||||||
# TODO: Maybe a a check that to unsatisfied that the path really is a URL?
|
# TODO: Maybe a a check that to unsatisfied that the path really is a URL?
|
||||||
|
|
||||||
|
|
||||||
class BytesRequirement(interfaces_configuration.SimpleTypeRequirement):
|
class BytesRequirement(configuration.SimpleTypeRequirement):
|
||||||
"""A requirement type that contains a byte string"""
|
"""A requirement type that contains a byte string"""
|
||||||
instance_type = bytes # type: typing.ClassVar[typing.Type]
|
instance_type = bytes # type: typing.ClassVar[typing.Type]
|
||||||
|
|
||||||
|
|
||||||
class ListRequirement(interfaces_configuration.RequirementInterface):
|
class ListRequirement(configuration.RequirementInterface):
|
||||||
"""Allows for a list of a specific type of requirement (all of which must be met for this requirement to be met) to be specified
|
"""Allows for a list of a specific type of requirement (all of which must be met for this requirement to be met) to be specified
|
||||||
|
|
||||||
This roughly correlates to allowing a number of arguments to follow a command line parameter,
|
This roughly correlates to allowing a number of arguments to follow a command line parameter,
|
||||||
@@ -68,7 +63,7 @@ class ListRequirement(interfaces_configuration.RequirementInterface):
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self,
|
def __init__(self,
|
||||||
element_type: typing.Type[interfaces_configuration.SimpleTypes] = str,
|
element_type: typing.Type[configuration.SimpleTypes] = str,
|
||||||
max_elements: typing.Optional[int] = 0,
|
max_elements: typing.Optional[int] = 0,
|
||||||
min_elements: typing.Optional[int] = None, *args, **kwargs) -> None:
|
min_elements: typing.Optional[int] = None, *args, **kwargs) -> None:
|
||||||
"""Constructs the object
|
"""Constructs the object
|
||||||
@@ -79,7 +74,7 @@ class ListRequirement(interfaces_configuration.RequirementInterface):
|
|||||||
min_elements: The minimum number of acceptable elements this list can contain
|
min_elements: The minimum number of acceptable elements this list can contain
|
||||||
"""
|
"""
|
||||||
super().__init__(*args, **kwargs)
|
super().__init__(*args, **kwargs)
|
||||||
if not issubclass(element_type, interfaces_configuration.BasicTypes):
|
if not issubclass(element_type, configuration.BasicTypes):
|
||||||
raise TypeError("ListRequirements can only be populated with simple InstanceRequirements")
|
raise TypeError("ListRequirements can only be populated with simple InstanceRequirements")
|
||||||
self.element_type = element_type # type: typing.Type
|
self.element_type = element_type # type: typing.Type
|
||||||
self.min_elements = min_elements or 0 # type: int
|
self.min_elements = min_elements or 0 # type: int
|
||||||
@@ -91,28 +86,28 @@ class ListRequirement(interfaces_configuration.RequirementInterface):
|
|||||||
value = self.config_value(context, config_path, default)
|
value = self.config_value(context, config_path, default)
|
||||||
if not value and self.min_elements > 0:
|
if not value and self.min_elements > 0:
|
||||||
vollog.log(constants.LOGLEVEL_V, "ListRequirement Unsatisfied - ListRequirement has non-zero min_elements")
|
vollog.log(constants.LOGLEVEL_V, "ListRequirement Unsatisfied - ListRequirement has non-zero min_elements")
|
||||||
return [interfaces_configuration.path_join(config_path, self.name)]
|
return [configuration.path_join(config_path, self.name)]
|
||||||
if value == default:
|
if value == default:
|
||||||
# We need to differentiate between no value and an empty list
|
# We need to differentiate between no value and an empty list
|
||||||
vollog.log(constants.LOGLEVEL_V, "ListRequirement Unsatisfied - Value was not specified")
|
vollog.log(constants.LOGLEVEL_V, "ListRequirement Unsatisfied - Value was not specified")
|
||||||
return [interfaces_configuration.path_join(config_path, self.name)]
|
return [configuration.path_join(config_path, self.name)]
|
||||||
if not isinstance(value, list):
|
if not isinstance(value, list):
|
||||||
# TODO: Check this is the correct response for an error
|
# TODO: Check this is the correct response for an error
|
||||||
raise ValueError("Unexpected config value found: {}".format(repr(value)))
|
raise ValueError("Unexpected config value found: {}".format(repr(value)))
|
||||||
if not (self.min_elements <= len(value)):
|
if not (self.min_elements <= len(value)):
|
||||||
vollog.log(constants.LOGLEVEL_V, "TypeError - Too few values provided to list option.")
|
vollog.log(constants.LOGLEVEL_V, "TypeError - Too few values provided to list option.")
|
||||||
return [interfaces_configuration.path_join(config_path, self.name)]
|
return [configuration.path_join(config_path, self.name)]
|
||||||
if self.max_elements and not (len(value) < self.max_elements):
|
if self.max_elements and not (len(value) < self.max_elements):
|
||||||
vollog.log(constants.LOGLEVEL_V, "TypeError - Too many values provided to list option.")
|
vollog.log(constants.LOGLEVEL_V, "TypeError - Too many values provided to list option.")
|
||||||
return [interfaces_configuration.path_join(config_path, self.name)]
|
return [configuration.path_join(config_path, self.name)]
|
||||||
if not all([self._check_type(element, self.element_type) for element in value]):
|
if not all([self._check_type(element, self.element_type) for element in value]):
|
||||||
vollog.log(constants.LOGLEVEL_V,
|
vollog.log(constants.LOGLEVEL_V,
|
||||||
"TypeError - At least one element in the list is not of the correct type.")
|
"TypeError - At least one element in the list is not of the correct type.")
|
||||||
return [interfaces_configuration.path_join(config_path, self.name)]
|
return [configuration.path_join(config_path, self.name)]
|
||||||
return []
|
return []
|
||||||
|
|
||||||
|
|
||||||
class ChoiceRequirement(interfaces_configuration.RequirementInterface):
|
class ChoiceRequirement(configuration.RequirementInterface):
|
||||||
"""Allows one from a choice of strings"""
|
"""Allows one from a choice of strings"""
|
||||||
|
|
||||||
def __init__(self, choices: typing.List[str], *args, **kwargs) -> None:
|
def __init__(self, choices: typing.List[str], *args, **kwargs) -> None:
|
||||||
@@ -131,11 +126,11 @@ class ChoiceRequirement(interfaces_configuration.RequirementInterface):
|
|||||||
value = self.config_value(context, config_path)
|
value = self.config_value(context, config_path)
|
||||||
if value not in self.choices:
|
if value not in self.choices:
|
||||||
vollog.log(constants.LOGLEVEL_V, "ValueError - Value is not within the set of available choices")
|
vollog.log(constants.LOGLEVEL_V, "ValueError - Value is not within the set of available choices")
|
||||||
return [interfaces_configuration.path_join(config_path, self.name)]
|
return [configuration.path_join(config_path, self.name)]
|
||||||
return []
|
return []
|
||||||
|
|
||||||
|
|
||||||
class LayerListRequirement(MultiRequirement, interfaces_configuration.ConfigurableRequirementInterface):
|
class LayerListRequirement(MultiRequirement, configuration.ConfigurableRequirementInterface):
|
||||||
"""Allows a variable length list of layers that must exist """
|
"""Allows a variable length list of layers that must exist """
|
||||||
|
|
||||||
def unsatisfied(self, context: interfaces.context.ContextInterface, config_path: str) -> typing.List[str]:
|
def unsatisfied(self, context: interfaces.context.ContextInterface, config_path: str) -> typing.List[str]:
|
||||||
@@ -144,14 +139,14 @@ class LayerListRequirement(MultiRequirement, interfaces_configuration.Configurab
|
|||||||
if ret_list:
|
if ret_list:
|
||||||
return ret_list
|
return ret_list
|
||||||
if (self.config_value(context, config_path, None) is None or
|
if (self.config_value(context, config_path, None) is None or
|
||||||
self.config_value(context, interfaces_configuration.path_join(config_path, 'number_of_elements'))):
|
self.config_value(context, configuration.path_join(config_path, 'number_of_elements'))):
|
||||||
return [interfaces_configuration.path_join(config_path, self.name)]
|
return [configuration.path_join(config_path, self.name)]
|
||||||
return []
|
return []
|
||||||
|
|
||||||
def construct(self, context: interfaces.context.ContextInterface, config_path: str) -> None:
|
def construct(self, context: interfaces.context.ContextInterface, config_path: str) -> None:
|
||||||
"""Method for constructing within the context any required elements from subrequirements"""
|
"""Method for constructing within the context any required elements from subrequirements"""
|
||||||
new_config_path = interfaces_configuration.path_join(config_path, self.name)
|
new_config_path = configuration.path_join(config_path, self.name)
|
||||||
num_layers_path = interfaces_configuration.path_join(new_config_path, "number_of_elements")
|
num_layers_path = configuration.path_join(new_config_path, "number_of_elements")
|
||||||
number_of_layers = context.config[num_layers_path]
|
number_of_layers = context.config[num_layers_path]
|
||||||
|
|
||||||
# Build all the layers that can be built
|
# Build all the layers that can be built
|
||||||
@@ -170,20 +165,171 @@ class LayerListRequirement(MultiRequirement, interfaces_configuration.Configurab
|
|||||||
def build_configuration(self,
|
def build_configuration(self,
|
||||||
context: interfaces.context.ContextInterface,
|
context: interfaces.context.ContextInterface,
|
||||||
config_path: str,
|
config_path: str,
|
||||||
_: typing.Any) -> interfaces_configuration.HierarchicalDict:
|
_: typing.Any) -> configuration.HierarchicalDict:
|
||||||
result = interfaces_configuration.HierarchicalDict()
|
result = configuration.HierarchicalDict()
|
||||||
num_elem_config_path = interfaces_configuration.path_join(config_path, self.name, 'number_of_elements')
|
num_elem_config_path = configuration.path_join(config_path, self.name, 'number_of_elements')
|
||||||
num_elements = context.config.get(num_elem_config_path, None)
|
num_elements = context.config.get(num_elem_config_path, None)
|
||||||
if num_elements is not None:
|
if num_elements is not None:
|
||||||
result["number_of_elements"] = num_elements
|
result["number_of_elements"] = num_elements
|
||||||
for i in range(num_elements):
|
for i in range(num_elements):
|
||||||
req = interfaces_configuration.TranslationLayerRequirement(name = self.name + str(i),
|
req = TranslationLayerRequirement(name = self.name + str(i),
|
||||||
description = "Swap Layer",
|
description = "Swap Layer",
|
||||||
optional = False)
|
optional = False)
|
||||||
self.add_requirement(req)
|
self.add_requirement(req)
|
||||||
value_path = interfaces_configuration.path_join(config_path, self.name, req.name)
|
value_path = configuration.path_join(config_path, self.name, req.name)
|
||||||
value = context.config.get(value_path, None)
|
value = context.config.get(value_path, None)
|
||||||
if value is not None:
|
if value is not None:
|
||||||
result.splice(req.name, context.memory[value].build_configuration())
|
result.splice(req.name, context.memory[value].build_configuration())
|
||||||
result[req.name] = value
|
result[req.name] = value
|
||||||
return result
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
# 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
|
||||||
|
class TranslationLayerRequirement(configuration.ConstructableRequirementInterface,
|
||||||
|
configuration.ConfigurableRequirementInterface):
|
||||||
|
"""Class maintaining the limitations on what sort of translation layers are acceptable"""
|
||||||
|
|
||||||
|
def __init__(self,
|
||||||
|
name: str,
|
||||||
|
description: str = None,
|
||||||
|
default: configuration.ConfigSimpleType = None,
|
||||||
|
optional: bool = False,
|
||||||
|
oses: typing.List = None,
|
||||||
|
architectures: typing.List = None) -> None:
|
||||||
|
"""Constructs a Translation Layer Requirement
|
||||||
|
|
||||||
|
The configuration option's value will be the name of the layer once it exists in the store
|
||||||
|
|
||||||
|
Args:
|
||||||
|
name: Name of the configuration requirement
|
||||||
|
layer_name: String detailing the expected name of the required layer, this can be None if it is to be randomly generated
|
||||||
|
"""
|
||||||
|
if oses is None:
|
||||||
|
oses = []
|
||||||
|
if architectures is None:
|
||||||
|
architectures = []
|
||||||
|
self.oses = oses
|
||||||
|
self.architectures = architectures
|
||||||
|
super().__init__(name, description, default, optional)
|
||||||
|
|
||||||
|
def unsatisfied(self,
|
||||||
|
context: interfaces.context.ContextInterface,
|
||||||
|
config_path: str) -> typing.List[str]:
|
||||||
|
"""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 [configuration.path_join(config_path, self.name)]
|
||||||
|
if self.oses and context.memory[value].metadata.get('os', None) not in self.oses:
|
||||||
|
vollog.log(9, "TypeError - Layer is not the required OS: {}".format(value))
|
||||||
|
return [configuration.path_join(config_path, self.name)]
|
||||||
|
if (self.architectures and
|
||||||
|
context.memory[value].metadata.get('architecture', None) not in self.architectures):
|
||||||
|
vollog.log(9, "TypeError - Layer is not the required Architecture: {}".format(value))
|
||||||
|
return [configuration.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 [configuration.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._validate_class(context, config_path)
|
||||||
|
vollog.log(constants.LOGLEVEL_V,
|
||||||
|
"IndexError - No configuration provided: {}".format(
|
||||||
|
config_path + configuration.CONFIG_SEPARATOR + self.name))
|
||||||
|
return [configuration.path_join(config_path, self.name)]
|
||||||
|
|
||||||
|
def construct(self, context: interfaces.context.ContextInterface, config_path: str) -> None:
|
||||||
|
"""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 = configuration.path_join(config_path, self.name)
|
||||||
|
|
||||||
|
args = {"context": context,
|
||||||
|
"config_path": config_path,
|
||||||
|
"name": name}
|
||||||
|
|
||||||
|
if any([subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if
|
||||||
|
not subreq.optional]):
|
||||||
|
return None
|
||||||
|
|
||||||
|
obj = self._construct_class(context, config_path, args)
|
||||||
|
if obj is not None and isinstance(obj, interfaces.layers.DataLayerInterface):
|
||||||
|
context.add_layer(obj)
|
||||||
|
# This should already be done by the _construct_class method
|
||||||
|
# context.config[config_path] = obj.name
|
||||||
|
return None
|
||||||
|
|
||||||
|
def build_configuration(self,
|
||||||
|
context: interfaces.context.ContextInterface,
|
||||||
|
_: str,
|
||||||
|
value: typing.Any) -> configuration.HierarchicalDict:
|
||||||
|
"""Builds the appropriate configuration for the specified requirement"""
|
||||||
|
return context.memory[value].build_configuration()
|
||||||
|
|
||||||
|
|
||||||
|
class SymbolRequirement(configuration.ConstructableRequirementInterface,
|
||||||
|
configuration.ConfigurableRequirementInterface):
|
||||||
|
"""Class maintaining the limitations on what sort of symbol spaces are acceptable"""
|
||||||
|
|
||||||
|
def unsatisfied(self, context: interfaces.context.ContextInterface, config_path: str) -> typing.List[str]:
|
||||||
|
"""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 [configuration.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 [configuration.path_join(config_path, self.name)]
|
||||||
|
return []
|
||||||
|
|
||||||
|
def construct(self, context: interfaces.context.ContextInterface, config_path: str) -> None:
|
||||||
|
"""Constructs the symbol space within the context based on the subrequirements"""
|
||||||
|
# Determine the space name
|
||||||
|
name = context.symbol_space.free_table_name(self.name)
|
||||||
|
|
||||||
|
config_path = configuration.path_join(config_path, self.name)
|
||||||
|
args = {"context": context,
|
||||||
|
"config_path": config_path,
|
||||||
|
"name": name}
|
||||||
|
|
||||||
|
if any([subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if
|
||||||
|
not subreq.optional]):
|
||||||
|
return None
|
||||||
|
|
||||||
|
# Fill out the parameter for class creation
|
||||||
|
if not isinstance(self.requirements["class"], configuration.ClassRequirement):
|
||||||
|
raise ValueError(
|
||||||
|
"Class requirement is not of type ClassRequirement: {}".format(repr(self.requirements["class"])))
|
||||||
|
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 not None and isinstance(obj, interfaces.symbols.SymbolTableInterface):
|
||||||
|
context.symbol_space.append(obj)
|
||||||
|
return None
|
||||||
|
|
||||||
|
def build_configuration(self,
|
||||||
|
context: interfaces.context.ContextInterface,
|
||||||
|
_: str,
|
||||||
|
value: typing.Any) -> configuration.HierarchicalDict:
|
||||||
|
"""Builds the appropriate configuration for the specified requirement"""
|
||||||
|
return context.symbol_space[value].build_configuration()
|
||||||
|
|||||||
@@ -5,14 +5,13 @@ Automagic objects attempt to automatically fill configuration values that a user
|
|||||||
import typing
|
import typing
|
||||||
from abc import ABCMeta
|
from abc import ABCMeta
|
||||||
|
|
||||||
import volatility.framework.configuration.requirements
|
|
||||||
from volatility.framework import validity, interfaces
|
from volatility.framework import validity, interfaces
|
||||||
from volatility.framework.interfaces import configuration as interfaces_configuration
|
from volatility.framework.configuration import requirements
|
||||||
|
|
||||||
R = typing.TypeVar('R', bound = interfaces.configuration.RequirementInterface)
|
R = typing.TypeVar('R', bound = interfaces.configuration.RequirementInterface)
|
||||||
|
|
||||||
|
|
||||||
class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metaclass = ABCMeta):
|
class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metaclass = ABCMeta):
|
||||||
"""Class that defines an automagic component that can help fulfill a Requirement
|
"""Class that defines an automagic component that can help fulfill a Requirement
|
||||||
|
|
||||||
These classes are callable with the following parameters:
|
These classes are callable with the following parameters:
|
||||||
@@ -39,11 +38,11 @@ class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metacla
|
|||||||
config_path: str, *args, **kwargs) -> None:
|
config_path: str, *args, **kwargs) -> None:
|
||||||
super().__init__(context, config_path)
|
super().__init__(context, config_path)
|
||||||
for requirement in self.get_requirements():
|
for requirement in self.get_requirements():
|
||||||
if not isinstance(requirement, (interfaces_configuration.SimpleTypeRequirement,
|
if not isinstance(requirement, (interfaces.configuration.SimpleTypeRequirement,
|
||||||
volatility.framework.configuration.requirements.ChoiceRequirement,
|
requirements.ChoiceRequirement,
|
||||||
volatility.framework.configuration.requirements.ListRequirement)):
|
requirements.ListRequirement)):
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
"Automagic requirements must be an SimpleTypeRequirement, ChoiceRequirement or ListRequirement")
|
"Automagic requirements must be a SimpleTypeRequirement, ChoiceRequirement or ListRequirement")
|
||||||
|
|
||||||
def __call__(self,
|
def __call__(self,
|
||||||
context: interfaces.context.ContextInterface,
|
context: interfaces.context.ContextInterface,
|
||||||
@@ -77,7 +76,7 @@ class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metacla
|
|||||||
Returns:
|
Returns:
|
||||||
A list of tuples containing the config_path, sub_config_path and requirement identifying the SymbolRequirements
|
A list of tuples containing the config_path, sub_config_path and requirement identifying the SymbolRequirements
|
||||||
"""
|
"""
|
||||||
sub_config_path = interfaces_configuration.path_join(config_path, requirement_root.name)
|
sub_config_path = interfaces.configuration.path_join(config_path, requirement_root.name)
|
||||||
results = [] # type: typing.List[typing.Tuple[str, str, R]]
|
results = [] # type: typing.List[typing.Tuple[str, str, R]]
|
||||||
recurse = not shortcut
|
recurse = not shortcut
|
||||||
if isinstance(requirement_root, requirement_type):
|
if isinstance(requirement_root, requirement_type):
|
||||||
|
|||||||
@@ -553,147 +553,3 @@ class ConfigurableInterface(validity.ValidityRoutines, metaclass = ABCMeta):
|
|||||||
return new_config_path
|
return new_config_path
|
||||||
|
|
||||||
|
|
||||||
class TranslationLayerRequirement(ConstructableRequirementInterface, ConfigurableRequirementInterface):
|
|
||||||
"""Class maintaining the limitations on what sort of translation layers are acceptable"""
|
|
||||||
|
|
||||||
def __init__(self,
|
|
||||||
name: str,
|
|
||||||
description: str = None,
|
|
||||||
default: ConfigSimpleType = None,
|
|
||||||
optional: bool = False,
|
|
||||||
oses: typing.List = None,
|
|
||||||
architectures: typing.List = None) -> None:
|
|
||||||
"""Constructs a Translation Layer Requirement
|
|
||||||
|
|
||||||
The configuration option's value will be the name of the layer once it exists in the store
|
|
||||||
|
|
||||||
Args:
|
|
||||||
name: Name of the configuration requirement
|
|
||||||
layer_name: String detailing the expected name of the required layer, this can be None if it is to be randomly generated
|
|
||||||
"""
|
|
||||||
if oses is None:
|
|
||||||
oses = []
|
|
||||||
if architectures is None:
|
|
||||||
architectures = []
|
|
||||||
self.oses = oses
|
|
||||||
self.architectures = architectures
|
|
||||||
super().__init__(name, description, default, optional)
|
|
||||||
|
|
||||||
def unsatisfied(self,
|
|
||||||
context: interfaces.context.ContextInterface,
|
|
||||||
config_path: str) -> typing.List[str]:
|
|
||||||
"""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 [path_join(config_path, self.name)]
|
|
||||||
if self.oses and context.memory[value].metadata.get('os', None) not in self.oses:
|
|
||||||
vollog.log(9, "TypeError - Layer is not the required OS: {}".format(value))
|
|
||||||
return [path_join(config_path, self.name)]
|
|
||||||
if (self.architectures and
|
|
||||||
context.memory[value].metadata.get('architecture', None) not in self.architectures):
|
|
||||||
vollog.log(9, "TypeError - Layer is not the required Architecture: {}".format(value))
|
|
||||||
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 [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._validate_class(context, config_path)
|
|
||||||
vollog.log(constants.LOGLEVEL_V,
|
|
||||||
"IndexError - No configuration provided: {}".format(config_path + CONFIG_SEPARATOR + self.name))
|
|
||||||
return [path_join(config_path, self.name)]
|
|
||||||
|
|
||||||
def construct(self, context: interfaces.context.ContextInterface, config_path: str) -> None:
|
|
||||||
"""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 any([subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if
|
|
||||||
not subreq.optional]):
|
|
||||||
return None
|
|
||||||
|
|
||||||
obj = self._construct_class(context, config_path, args)
|
|
||||||
if obj is not None and isinstance(obj, interfaces.layers.DataLayerInterface):
|
|
||||||
context.add_layer(obj)
|
|
||||||
# This should already be done by the _construct_class method
|
|
||||||
# context.config[config_path] = obj.name
|
|
||||||
return None
|
|
||||||
|
|
||||||
def build_configuration(self,
|
|
||||||
context: interfaces.context.ContextInterface,
|
|
||||||
_: str,
|
|
||||||
value: typing.Any) -> HierarchicalDict:
|
|
||||||
"""Builds the appropriate configuration for the specified requirement"""
|
|
||||||
return context.memory[value].build_configuration()
|
|
||||||
|
|
||||||
|
|
||||||
class SymbolRequirement(ConstructableRequirementInterface, ConfigurableRequirementInterface):
|
|
||||||
"""Class maintaining the limitations on what sort of symbol spaces are acceptable"""
|
|
||||||
|
|
||||||
def unsatisfied(self, context: interfaces.context.ContextInterface, config_path: str) -> typing.List[str]:
|
|
||||||
"""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 [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 [path_join(config_path, self.name)]
|
|
||||||
return []
|
|
||||||
|
|
||||||
def construct(self, context: interfaces.context.ContextInterface, config_path: str) -> None:
|
|
||||||
"""Constructs the symbol space within the context based on the subrequirements"""
|
|
||||||
# Determine the space name
|
|
||||||
name = context.symbol_space.free_table_name(self.name)
|
|
||||||
|
|
||||||
config_path = path_join(config_path, self.name)
|
|
||||||
args = {"context": context,
|
|
||||||
"config_path": config_path,
|
|
||||||
"name": name}
|
|
||||||
|
|
||||||
if any([subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if
|
|
||||||
not subreq.optional]):
|
|
||||||
return None
|
|
||||||
|
|
||||||
# Fill out the parameter for class creation
|
|
||||||
if not isinstance(self.requirements["class"], ClassRequirement):
|
|
||||||
raise ValueError(
|
|
||||||
"Class requirement is not of type ClassRequirement: {}".format(repr(self.requirements["class"])))
|
|
||||||
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 not None and isinstance(obj, interfaces.symbols.SymbolTableInterface):
|
|
||||||
context.symbol_space.append(obj)
|
|
||||||
return None
|
|
||||||
|
|
||||||
def build_configuration(self,
|
|
||||||
context: interfaces.context.ContextInterface,
|
|
||||||
_: str,
|
|
||||||
value: typing.Any) -> HierarchicalDict:
|
|
||||||
"""Builds the appropriate configuration for the specified requirement"""
|
|
||||||
return context.symbol_space[value].build_configuration()
|
|
||||||
|
|||||||
@@ -3,8 +3,7 @@ import typing
|
|||||||
|
|
||||||
from volatility.framework import constants, exceptions, interfaces, objects
|
from volatility.framework import constants, exceptions, interfaces, objects
|
||||||
from volatility.framework.configuration import requirements
|
from volatility.framework.configuration import requirements
|
||||||
from volatility.framework.configuration.requirements import IntRequirement
|
from volatility.framework.configuration.requirements import IntRequirement, TranslationLayerRequirement
|
||||||
from volatility.framework.interfaces.configuration import TranslationLayerRequirement
|
|
||||||
from volatility.framework.exceptions import InvalidAddressException
|
from volatility.framework.exceptions import InvalidAddressException
|
||||||
from volatility.framework.symbols import intermed
|
from volatility.framework.symbols import intermed
|
||||||
from volatility.plugins.windows import pslist
|
from volatility.plugins.windows import pslist
|
||||||
|
|||||||
Reference in New Issue
Block a user