diff --git a/volatility/framework/layers/__init__.py b/volatility/framework/layers/__init__.py index b690dabc1..8c8cac765 100644 --- a/volatility/framework/layers/__init__.py +++ b/volatility/framework/layers/__init__.py @@ -3,7 +3,4 @@ Created on 4 May 2013 @author: mike """ - -from volatility.framework.layers import intel, lime, physical - - +from volatility.framework.layers import intel, lime, physical, segmented diff --git a/volatility/framework/layers/lime.py b/volatility/framework/layers/lime.py index d058c6896..91290f718 100644 --- a/volatility/framework/layers/lime.py +++ b/volatility/framework/layers/lime.py @@ -7,14 +7,14 @@ Created on 6 Apr 2016 import struct from volatility.framework import exceptions, interfaces -from volatility.framework.configuration import requirements +from volatility.framework.layers import segmented class LimeFormatException(exceptions.LayerException): """Thrown when an error occurs with the underlying Lime file format""" -class LimeLayer(interfaces.layers.TranslationLayerInterface): +class LimeLayer(segmented.SegmentedLayer): """A Lime format TranslationLayer. Lime is generally used to store physical memory images where there are large holes in the physical address space""" @@ -32,25 +32,9 @@ class LimeLayer(interfaces.layers.TranslationLayerInterface): def __init__(self, context, config_path, name): super().__init__(context, config_path, name) - self._base_layer = self.config["base_layer"] - - # list of tuples (logical start, base start, size) - # loaded by _load_segments() on first access - self._segments = [] - self._minaddr = 0 - self._maxaddr = 0 - # We must run this on creation in order to get the right min/maxaddr in case scanning is our first action self._load_segments() - @property - def minimum_address(self): - return self._minaddr - - @property - def maximum_address(self): - return self._maxaddr - def _load_segments(self): base_layer = self._context.memory[self._base_layer] base_maxaddr = base_layer.maximum_address @@ -75,8 +59,6 @@ class LimeLayer(interfaces.layers.TranslationLayerInterface): raise LimeFormatException("No LiME segments defined in {}".format(self._base_layer)) self._segments = segments - self._minaddr = segments[0][0] - self._maxaddr = maxaddr @classmethod def _check_header(cls, base_layer, offset = 0): @@ -88,68 +70,6 @@ class LimeLayer(interfaces.layers.TranslationLayerInterface): raise LimeFormatException("unexpected version {:d} at file offset 0x{:x}".format(version, offset)) return start, end - def _find_segment(self, offset): - """Finds the segment containing a given offset - - Returns the segment tuple - """ - - if not self._segments: - self._load_segments() - - for logical_start, base_start, size in self._segments: - if offset >= logical_start and offset < (logical_start + size): - return (logical_start, base_start, size) - - raise exceptions.InvalidAddressException(self.name, offset, "Lime fault at address {:0x}".format(offset)) - - def is_valid(self, offset, length = 1): - """Returns whether the address offset can be translated to a valid address""" - try: - return all([self._context.memory[self._base_layer].is_valid(mapped_offset) for _, mapped_offset, _, _ in - self.mapping(offset, length)]) - except exceptions.InvalidAddressException: - return False - - def mapping(self, offset, length, ignore_errors = False): - """Returns a sorted iterable of (offset, mapped_offset, length, layer) mappings""" - result = [] - if ignore_errors: - for (seg_offset, mapped_seg_offset, seg_length) in self._segments: - if (offset + length < seg_offset) or (offset > seg_offset + seg_length): - continue - if seg_offset <= offset < seg_offset + seg_length: - diff = offset - seg_offset - seg_length -= diff - mapped_seg_offset += diff - seg_offset += diff - if offset + length < seg_offset + seg_length: - seg_length = offset + length - seg_offset - yield (seg_offset, mapped_seg_offset, seg_length, self._base_layer) - raise StopIteration - if length == 0: - logical_start, base_start, size = self._find_segment(offset) - mapped_offset = offset - logical_start + base_start - yield (offset, mapped_offset, 0, self._base_layer) - raise StopIteration - while length > 0: - logical_start, base_start, size = self._find_segment(offset) - chunk_offset = offset - logical_start + base_start - chunk_size = min(size - (offset - logical_start), length) - yield (offset, chunk_offset, chunk_size, self._base_layer) - length -= chunk_size - offset += chunk_size - - @property - def dependencies(self): - """Returns a list of the lower layers that this layer is dependent upon""" - return [self._base_layer] - - @classmethod - def get_requirements(cls): - return [requirements.TranslationLayerRequirement(name = 'base_layer', - optional = False)] - class LimeStacker(interfaces.automagic.StackerLayerInterface): stack_order = 10 diff --git a/volatility/framework/layers/segmented.py b/volatility/framework/layers/segmented.py new file mode 100644 index 000000000..c63c51815 --- /dev/null +++ b/volatility/framework/layers/segmented.py @@ -0,0 +1,123 @@ +from abc import ABCMeta, abstractmethod +from bisect import bisect_right + +from volatility.framework import exceptions +from volatility.framework import interfaces +from volatility.framework.configuration import requirements + + +class SegmentedLayer(interfaces.layers.TranslationLayerInterface, metaclass = ABCMeta): + """A class to handle a single run-based layer-to-layer mapping + + In the documentation "mapped address" or "mapped offset" refers to an offset once it has been mapped to the underlying layer + """ + + def __init__(self, context, config_path, name): + super().__init__(context, config_path = config_path, name = name) + + self._base_layer = self.config["base_layer"] + self._segments = [] + self._minaddr = None + self._maxaddr = None + + self._load_segments() + + @abstractmethod + def _load_segments(self): + """Populates the _segments variable + + Segments must be (address, mapped address, length) and must be sorted by address when this method exits + """ + + def is_valid(self, offset, length = 1): + """Returns whether the address offset can be translated to a valid address""" + try: + return all([self._context.memory[self._base_layer].is_valid(mapped_offset) for _, mapped_offset, _, _ in + self.mapping(offset, length)]) + except exceptions.InvalidAddressException: + return False + + def _find_segment(self, offset, next = False): + """Finds the segment containing a given offset + + Returns the segment tuple (offset, mapped_offset, length) + """ + + if not self._segments: + self._load_segments() + + 'Find rightmost value less than or equal to x' + i = bisect_right(self._segments, (offset, self.context.memory[self._base_layer].maximum_address)) + if i: + if not next: + segment = self._segments[i - 1] + if segment[0] <= offset < segment[0] + segment[2]: + return segment + else: + if i < len(self._segments): + return self._segments[i] + raise exceptions.InvalidAddressException(self.name, offset, "Invalid address at {:0x}".format(offset)) + + def mapping(self, offset, length, ignore_errors = False): + """Returns a sorted iterable of (offset, mapped_offset, length, layer) mappings""" + done = False + current_offset = offset + while not done: + try: + # Search for the appropriate segment that contains the current_offset + logical_offset, mapped_offset, size = self._find_segment(current_offset) + # If it starts before the current_offset, bring the lower edge up to the right place + if current_offset > logical_offset: + difference = current_offset - logical_offset + logical_offset += difference + mapped_offset += difference + size -= difference + except exceptions.InvalidAddressException: + if not ignore_errors: + # If we're not ignoring errors, raise the invalid address exception + raise + try: + # Find the next valid segment after our current_offset + logical_offset, mapped_offset, size = self._find_segment(current_offset, next = True) + # We know that the logical_offset must be greater than current_offset so skip to that value + current_offset = logical_offset + # If it starts too late then we're done + if logical_offset > offset + length: + raise StopIteration + except exceptions.InvalidAddressException: + raise StopIteration + # Crop it to the amount we need left + chunk_size = min(size, length + offset - logical_offset) + yield (logical_offset, mapped_offset, chunk_size, self._base_layer) + current_offset += chunk_size + # Terminate if we've gone (or reached) our required limit + if current_offset >= offset + length: + done = True + + @property + def minimum_address(self): + if not self._segments: + raise ValueError("SegmentedLayer must contain some segments") + if self._minaddr is None: + mapped, _, _ = self._segments[0] + self._minaddr = mapped + return self._minaddr + + @property + def maximum_address(self): + if not self._segments: + raise ValueError("SegmentedLayer must contain some segments") + if self._maxaddr is None: + mapped, _, length = self._segments[-1] + self._maxaddr = mapped + length + return self._maxaddr + + @property + def dependencies(self): + """Returns a list of the lower layers that this layer is dependent upon""" + return [self._base_layer] + + @classmethod + def get_requirements(cls): + return [requirements.TranslationLayerRequirement(name = 'base_layer', + optional = False)]