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"""