From 9512cbe9eb69b97dbf2c03c46865ab7ae703fc1c Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Thu, 26 Apr 2018 12:48:14 +0100 Subject: [PATCH] Commit metadata changeset. Layers now accept metadata dictionaries (and chain/stack them on top of those from lower layers). Metadata can only be set at construction time, and the metadata dictionary is readonly. The hope is this will make enumerating metadata keys across the codebase simpler. The current metadata items that layers hold is: architecture (Unknown | Intel32 | Intel64) os (Unknown | Windows | Linux) pae (bool) page_map_offset (int) This patchset may develop further to help enumerate all of these (through a registration/reporting system). --- volatility/framework/automagic/linux.py | 2 +- volatility/framework/automagic/stacker.py | 1 - volatility/framework/automagic/windows.py | 79 ++++++++++++++----- volatility/framework/interfaces/automagic.py | 2 +- .../framework/interfaces/configuration.py | 3 +- volatility/framework/interfaces/layers.py | 36 +++------ volatility/framework/layers/intel.py | 13 +-- volatility/framework/layers/physical.py | 10 ++- volatility/framework/layers/registry.py | 4 +- volatility/framework/layers/segmented.py | 5 +- volatility/framework/layers/vmware.py | 5 +- 11 files changed, 98 insertions(+), 62 deletions(-) diff --git a/volatility/framework/automagic/linux.py b/volatility/framework/automagic/linux.py index f75705466..0e7dc65f0 100644 --- a/volatility/framework/automagic/linux.py +++ b/volatility/framework/automagic/linux.py @@ -149,7 +149,7 @@ class LintelStacker(interfaces.automagic.StackerLayerInterface): context.config[join(config_path, "page_map_offset")] = dtb context.config[join(config_path, "linux_banner")] = str(banner, 'latin-1') - layer = layer_class(context, config_path = config_path, name = new_layer_name) + layer = layer_class(context, config_path = config_path, name = new_layer_name, os = 'Linux') if layer: vollog.debug("DTB was found at: 0x{:0x}".format(dtb)) diff --git a/volatility/framework/automagic/stacker.py b/volatility/framework/automagic/stacker.py index 0780e244d..3f5ce85cc 100644 --- a/volatility/framework/automagic/stacker.py +++ b/volatility/framework/automagic/stacker.py @@ -33,7 +33,6 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): """ # Most important automagic, must happen first! priority = 10 - page_map_offset = None def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) diff --git a/volatility/framework/automagic/windows.py b/volatility/framework/automagic/windows.py index 5f0da6ae8..d6ef8b126 100644 --- a/volatility/framework/automagic/windows.py +++ b/volatility/framework/automagic/windows.py @@ -262,16 +262,23 @@ class WintelHelper(interfaces.automagic.AutomagicInterface): if ("memory_layer" in requirement.requirements and not requirement.requirements["memory_layer"].unsatisfied(context, sub_config_path)): # Only bother getting the DTB if we don't already have one - if not context.config.get(interfaces.configuration.path_join(sub_config_path, "page_map_offset"), None): - physical_layer = requirement.requirements["memory_layer"].config_value(context, sub_config_path) - if not isinstance(physical_layer, str): + page_map_offset_path = interfaces.configuration.path_join(sub_config_path, "page_map_offset") + if not context.config.get(page_map_offset_path, None): + physical_layer_name = requirement.requirements["memory_layer"].config_value(context, + sub_config_path) + if not isinstance(physical_layer_name, str): raise TypeError("Physical layer name is not a string: {}".format(sub_config_path)) - hits = context.memory[physical_layer].scan(context, PageMapScanner(useful), progress_callback) - for test, dtb in hits: - context.config[interfaces.configuration.path_join(sub_config_path, "page_map_offset")] = dtb - break + physical_layer = context.memory[physical_layer_name] + # Check lower layer metadata first + if physical_layer.metadata.get('page_map_offset', None): + context.config[page_map_offset_path] = physical_layer.metadata['page_map_offset'] else: - return None + hits = physical_layer.scan(context, PageMapScanner(useful), progress_callback) + for test, dtb in hits: + context.config[page_map_offset_path] = dtb + break + else: + return None if isinstance(requirement, interfaces.configuration.ConstructableRequirementInterface): requirement.construct(context, config_path) else: @@ -294,24 +301,56 @@ class WintelStacker(interfaces.automagic.StackerLayerInterface): that range, and ignore any that contain multiple self-references (since the DTB is very unlikely to point to itself more than once). """ - if isinstance(context.memory[layer_name], intel.Intel): + base_layer = context.memory[layer_name] + if isinstance(base_layer, intel.Intel): return None - hits = context.memory[layer_name].scan(context, PageMapScanner(WintelHelper.tests)) - layer = None - config_path = None - for test, dtb in hits: + if (base_layer.metadata.get('os', None) not in ['Windows', 'Unknown']): + return None + layer = config_path = None + + # Check the metadata + if (base_layer.metadata.get('os', None) == 'Windows' and + base_layer.metadata.get('page_map_offset')): + arch = base_layer.metadata.get('architecture', None) + if arch not in ['Intel32', 'Intel64']: + return None + # Set the layer type + layer_type = intel.WindowsIntel + if arch == 'Intel64': + layer_type = intel.WindowsIntel32e + elif base_layer.metadata.get('pae', False): + layer_type = intel.WindowsIntelPAE + # Construct the layer new_layer_name = context.memory.free_layer_name("IntelLayer") config_path = interfaces.configuration.path_join("IntelHelper", new_layer_name) context.config[interfaces.configuration.path_join(config_path, "memory_layer")] = layer_name - context.config[interfaces.configuration.path_join(config_path, "page_map_offset")] = dtb - layer = test.layer_type(context, - config_path = config_path, - name = new_layer_name) - break + context.config[interfaces.configuration.path_join(config_path, "page_map_offset")] = base_layer.metadata[ + 'page_map_offset'] + layer = layer_type(context, + config_path = config_path, + name = new_layer_name, + metadata = {'os': 'Windows'}) + + # Check for the self-referential pointer + if layer is None: + hits = base_layer.scan(context, PageMapScanner(WintelHelper.tests)) + layer = None + config_path = None + for test, dtb in hits: + new_layer_name = context.memory.free_layer_name("IntelLayer") + config_path = interfaces.configuration.path_join("IntelHelper", new_layer_name) + context.config[interfaces.configuration.path_join(config_path, "memory_layer")] = layer_name + context.config[interfaces.configuration.path_join(config_path, "page_map_offset")] = dtb + layer = test.layer_type(context, + config_path = config_path, + name = new_layer_name, + metadata = {'os': 'Windows'}) + break + + # Fall back to a heuristic for finding the Windows DTB if layer is None: vollog.debug("Self-referential pointer not in well-known location, moving to recent windows heuristic") # There is a very high chance that the DTB will live in this narrow segment, assuming we couldn't find it previously - # TODO: This scan takes time, it might be worth adding a progress callback to it hits = context.memory[layer_name].scan(context, PageMapScanner([DtbSelfRef64bit()]), min_address = 0x1a0000, max_address = 0x1f0000, progress_callback = progress_callback) # Flatten the generator @@ -325,7 +364,7 @@ class WintelStacker(interfaces.automagic.StackerLayerInterface): context.config[interfaces.configuration.path_join(config_path, "page_map_offset")] = page_map_offset # TODO: Need to determine the layer type (chances are high it's x64, hence this default) layer = layers.intel.WindowsIntel32e(context, config_path = config_path, - name = new_layer_name) + name = new_layer_name, metadata = {'os': 'Windows'}) if layer is not None and config_path: vollog.debug("DTB was found at: 0x{:0x}".format( context.config[interfaces.configuration.path_join(config_path, "page_map_offset")])) diff --git a/volatility/framework/interfaces/automagic.py b/volatility/framework/interfaces/automagic.py index 7580084ae..a176857ff 100644 --- a/volatility/framework/interfaces/automagic.py +++ b/volatility/framework/interfaces/automagic.py @@ -62,7 +62,7 @@ class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metacla context: interfaces.context.ContextInterface, config_path: str, requirement_root: interfaces.configuration.RequirementInterface, - requirement_type: typing.Type[R], + requirement_type: typing.Union[typing.Tuple[typing.Type[R], ...], typing.Type[R]], shortcut: bool = True) \ -> typing.List[typing.Tuple[str, str, R]]: """Determines if there is actually an unfulfilled requirement waiting diff --git a/volatility/framework/interfaces/configuration.py b/volatility/framework/interfaces/configuration.py index e593f7ac3..2f2761771 100644 --- a/volatility/framework/interfaces/configuration.py +++ b/volatility/framework/interfaces/configuration.py @@ -591,7 +591,8 @@ class TranslationLayerRequirement(ConstructableRequirementInterface, Configurabl if self.oses and context.memory[value].os 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].architecture not in self.architectures: + 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 [] diff --git a/volatility/framework/interfaces/layers.py b/volatility/framework/interfaces/layers.py index 3e5e29676..e17ae1a91 100644 --- a/volatility/framework/interfaces/layers.py +++ b/volatility/framework/interfaces/layers.py @@ -87,36 +87,18 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR """A Layer that directly holds data (and does not translate it). This is effectively a leaf node in a layer tree. It directly accesses a data source and exposes it within volatility.""" - _architecture = "Unknown" + _direct_metadata = collections.ChainMap({}, {'architecture': 'Unknown', + 'os': 'Unknown'}) def __init__(self, context: 'interfaces.context.ContextInterface', config_path: str, name: str, - os: str = "Unknown") -> None: + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None) -> None: super().__init__(context, config_path) self._name = self._check_type(name, str) - self._os = self._check_type(os, str) - - # Memory specific attributes - - @property - def architecture(self) -> str: - """The architecutre of the TranslationLayer - - This cannot be modified after construction outside of the class - """ - return self._architecture - - @property - def os(self) -> str: - """The operating system related to the TranslationLayer""" - return self._os - - @os.setter - def os(self, value: str) -> None: - """Sets the operating system of the TranslationLayer""" - self._os = self._check_type(value, str) + if metadata: + self._direct_metadata.update(metadata) # Standard attributes @@ -274,6 +256,14 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR config["class"] = self.__class__.__module__ + "." + self.__class__.__name__ return config + # ## Metadata methods + + @property + def metadata(self) -> typing.Mapping: + """Returns a ReadOnly copy of the metadata published by this layer""" + maps = [self.context.memory[layer_name].metadata for layer_name in self.dependencies] + return interfaces.objects.ReadOnlyMapping(collections.ChainMap({}, self._direct_metadata, *maps)) + class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta): """Provides a layer that translates or transforms another layer or layers. Translation layers always depend on diff --git a/volatility/framework/layers/intel.py b/volatility/framework/layers/intel.py index da01a6bfc..c4a543426 100644 --- a/volatility/framework/layers/intel.py +++ b/volatility/framework/layers/intel.py @@ -1,3 +1,4 @@ +import collections import logging import math import struct @@ -27,7 +28,6 @@ class Intel(interfaces.layers.TranslationLayerInterface): """Translation Layer for the Intel IA32 memory mapping""" priority = 40 - _architecture = "Intel32" _entry_format = " None: - super().__init__(context, config_path, name) + name: str, + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None) -> None: + super().__init__(context = context, config_path = config_path, name = name, metadata = metadata) self._base_layer = self._check_type(self.config["memory_layer"], str) self._swap_layers = [] # type: typing.List[str] self._check_type(self.config.get("swap_layers", False), bool) @@ -253,7 +256,6 @@ class IntelPAE(Intel): """Class for handling Physical Address Extensions for Intel architectures""" priority = 35 - _architecture = "Intel32" _entry_format = " bool: """Returns whether a particular page is valid based on its entry diff --git a/volatility/framework/layers/physical.py b/volatility/framework/layers/physical.py index 7246b4682..17ab7eac7 100644 --- a/volatility/framework/layers/physical.py +++ b/volatility/framework/layers/physical.py @@ -13,8 +13,9 @@ class BufferDataLayer(interfaces.layers.DataLayerInterface): context: interfaces.context.ContextInterface, config_path: str, name: str, - buffer: bytes) -> None: - super().__init__(context, config_path, name) + buffer: bytes, + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None) -> None: + super().__init__(context = context, config_path = config_path, name = name, metadata = metadata) self._buffer = self._check_type(buffer, bytes) @property @@ -62,8 +63,9 @@ class FileLayer(interfaces.layers.DataLayerInterface): def __init__(self, context: interfaces.context.ContextInterface, config_path: str, - name: str) -> None: - super().__init__(context, config_path, name) + name: str, + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None) -> None: + super().__init__(context = context, config_path = config_path, name = name, metadata = metadata) self._location = self.config["location"] self._accessor = layers.ResourceAccessor() diff --git a/volatility/framework/layers/registry.py b/volatility/framework/layers/registry.py index a4f0665b8..135dffc42 100644 --- a/volatility/framework/layers/registry.py +++ b/volatility/framework/layers/registry.py @@ -23,8 +23,8 @@ class RegistryHive(interfaces.layers.TranslationLayerInterface): context: interfaces.context.ContextInterface, config_path: str, name: str, - os: str = "Unknown") -> None: - super().__init__(context, config_path, name, os) + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None) -> None: + super().__init__(context = context, config_path = config_path, name = name, metadata = metadata) self._base_layer = self.config["base_layer"] self._hive_offset = self.config["hive_offset"] diff --git a/volatility/framework/layers/segmented.py b/volatility/framework/layers/segmented.py index 9e9df2840..5859c8a6c 100644 --- a/volatility/framework/layers/segmented.py +++ b/volatility/framework/layers/segmented.py @@ -15,8 +15,9 @@ class SegmentedLayer(interfaces.layers.TranslationLayerInterface, metaclass = AB def __init__(self, context: interfaces.configuration.ContextInterface, config_path: str, - name: str) -> None: - super().__init__(context, config_path = config_path, name = name) + name: str, + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None) -> None: + super().__init__(context = context, config_path = config_path, name = name, metadata = metadata) self._base_layer = self.config["base_layer"] self._segments = [] # type: typing.List[typing.Tuple[int, int, int]] diff --git a/volatility/framework/layers/vmware.py b/volatility/framework/layers/vmware.py index e6ee2a7a0..ad99c46b3 100644 --- a/volatility/framework/layers/vmware.py +++ b/volatility/framework/layers/vmware.py @@ -17,14 +17,15 @@ class VmwareLayer(segmented.SegmentedLayer): def __init__(self, context: interfaces.context.ContextInterface, config_path: str, - name: str) -> None: + name: str, + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None) -> None: # Construct these so we can use self.config self._context = context self._config_path = config_path self._page_size = 0x1000 self._base_layer, self._meta_layer = self.config["base_layer"], self.config["meta_layer"] # Then call the super, which will call load_segments (which needs the base_layer before it'll work) - super().__init__(context, config_path = config_path, name = name) + super().__init__(context, config_path = config_path, name = name, metadata = metadata) def _load_segments(self) -> None: """Loads up the segments from the meta_layer"""