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).
This commit is contained in:
Mike Auty
2018-04-26 12:48:14 +01:00
parent 556fa29ada
commit 9512cbe9eb
11 changed files with 98 additions and 62 deletions
+1 -1
View File
@@ -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))
@@ -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)
+59 -20
View File
@@ -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")]))
+1 -1
View File
@@ -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
@@ -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 []
+13 -23
View File
@@ -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
+8 -5
View File
@@ -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 = "<I"
_page_size_in_bits = 12
_bits_per_register = 32
@@ -36,12 +36,15 @@ class Intel(interfaces.layers.TranslationLayerInterface):
_maxvirtaddr = _maxphyaddr
_structure = [('page directory', 10, False),
('page table', 10, True)]
_direct_metadata = collections.ChainMap({'architecture': 'Intel32'},
interfaces.layers.TranslationLayerInterface._direct_metadata)
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._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 = "<Q"
_bits_per_register = 32
_maxphyaddr = 40
@@ -267,7 +269,7 @@ class Intel32e(Intel):
"""Class for handling 64-bit (32-bit extensions) for Intel architectures"""
priority = 30
_architecture = "Intel64"
_direct_metadata = collections.ChainMap({'architecture': 'Intel64'}, Intel._direct_metadata)
_entry_format = "<Q"
_bits_per_register = 64
_maxphyaddr = 52
@@ -279,6 +281,7 @@ class Intel32e(Intel):
class WindowsMixin(Intel):
@staticmethod
def _page_is_valid(entry: int) -> bool:
"""Returns whether a particular page is valid based on its entry
+6 -4
View File
@@ -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()
+2 -2
View File
@@ -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"]
+3 -2
View File
@@ -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]]
+3 -2
View File
@@ -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"""