mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-11 20:27:38 +02:00
Add in an initial draft of the dependency resolver
Some gotchas: * Classes can't update attributes that they inherit, because the inheritance happens after class construction * The dep resolver can go in circle if the constraints in a layer are too permissive
This commit is contained in:
@@ -29,27 +29,22 @@ class CommandLine(object):
|
||||
|
||||
# TODO: Choose a plugin
|
||||
plugin = volatility.plugins.windows.pslist.PsList
|
||||
context, req_mapping = self.collect_plugin_requirements(plugin)
|
||||
parser = argparse.ArgumentParser(prog = 'volatility',
|
||||
description = "An open-source memory forensics framework")
|
||||
argparse_adapter.adapt_config(context.config, parser)
|
||||
# argparse_adapter.adapt_config(context.config, parser)
|
||||
|
||||
# Run the argparser
|
||||
parser.parse_args()
|
||||
|
||||
# Determine the selected plugin
|
||||
# Resolve the dependencies on that plugin
|
||||
dldr = depresolver.DataLayerDependencyResolver(plugin)
|
||||
dldr = depresolver.DataLayerDependencyResolver()
|
||||
print("DEPS RESOLVED", dldr.resolve_dependencies(plugin))
|
||||
|
||||
# Translate the parsed args to a context configuration
|
||||
|
||||
# Generate the layers from the arguments
|
||||
for req in req_mapping:
|
||||
factory = req_mapping[req]
|
||||
req.value = factory(context)
|
||||
|
||||
# Construct and run the plugin
|
||||
plugin(context).run()
|
||||
# plugin(context).run()
|
||||
|
||||
def collect_plugin_requirements(self, plugin):
|
||||
"""Generates the requirements necessary for the plugin"""
|
||||
|
||||
@@ -1,28 +1,68 @@
|
||||
import volatility.framework as framework
|
||||
import volatility.framework.interfaces as interfaces
|
||||
import volatility.framework.validity as validity
|
||||
from volatility.framework.interfaces import layers, configuration
|
||||
|
||||
|
||||
class DataLayerDependencyResolver(validity.ValidityRoutines):
|
||||
def __init__(self):
|
||||
# Maintain a cache of translation layers
|
||||
self.layer_cache = []
|
||||
for layer_class in framework.class_subclasses(interfaces.layers.DataLayerInterface):
|
||||
# TODO: Improve this hard coded list with a way for layers to say they're abstract or not
|
||||
if layer_class.__name__.endswith("Interface"):
|
||||
self.metadata = {}
|
||||
self.populate_metadata()
|
||||
|
||||
def populate_metadata(self):
|
||||
self.metadata = {}
|
||||
for layer_class in framework.class_subclasses(layers.DataLayerInterface):
|
||||
for k, v in layer_class.metadata.items():
|
||||
if not isinstance(v, list):
|
||||
new_v = self.metadata.get(k, set())
|
||||
new_v.add(v)
|
||||
else:
|
||||
new_v = self.metadata.get(k, set()) + v
|
||||
self.metadata[k] = new_v
|
||||
self.layer_cache.append(layer_class)
|
||||
|
||||
def satisfies(self, layer_class, requirement):
|
||||
"""Takes the requirement (which should always be a TranslationLayerRequirement) and determines if the
|
||||
layer_class satisfies it"""
|
||||
satisfied = True
|
||||
for k, v in requirement.constraints.items():
|
||||
print("Constraint", k, v)
|
||||
if k in layer_class.metadata:
|
||||
print(layer_class.__class__.__name__, layer_class.metadata)
|
||||
if isinstance(v, list):
|
||||
satisfied = satisfied and layer_class.metadata[k] not in v
|
||||
else:
|
||||
satisfied = satisfied and (layer_class.metadata[k] == v)
|
||||
return satisfied
|
||||
|
||||
def resolve_dependencies(self, configurable):
|
||||
"""Takes a configurable and produces a priority ordered tree of possible solutions to satisfy the various requirements
|
||||
"""Takes a configurable class and produces a priority ordered tree of possible solutions to satisfy the various requirements
|
||||
|
||||
The return should include each of the potential nodes (and requirements, including optional ones) allowing the UI
|
||||
to decide the layer build-path and get all the necessary variables from the user for that path.
|
||||
"""
|
||||
self._check_type(configurable, interfaces.configuration.Configurable)
|
||||
self._check_class(configurable, configuration.Configurable)
|
||||
|
||||
for requirement in configurable.get_schemas():
|
||||
possible_array = []
|
||||
|
||||
for requirement in configurable.get_schema():
|
||||
print("GET_SCHEMA", requirement)
|
||||
# If the requirement is a layer/configurable
|
||||
# Recurse over it
|
||||
print(requirement)
|
||||
# Add all base-type requirements
|
||||
# Add all optional base-type requirements in order
|
||||
if isinstance(requirement, framework.configuration.TranslationLayerRequirement):
|
||||
possibilities = {}
|
||||
for potential_layer in self.layer_cache:
|
||||
if self.satisfies(potential_layer, requirement):
|
||||
print("Resolving sub-dependencies", potential_layer)
|
||||
possibility = self.resolve_dependencies(potential_layer)
|
||||
# Only add a possibility if there are suitable lower layers for it
|
||||
if possibility:
|
||||
possibilities[potential_layer] = possibility
|
||||
possible_array.append(possibilities)
|
||||
else:
|
||||
possible_array.append(requirement)
|
||||
# Recurse over it
|
||||
print(requirement.name)
|
||||
# Add all base-type requirements
|
||||
# Add all optional base-type requirements in order
|
||||
return possible_array
|
||||
|
||||
Reference in New Issue
Block a user