From 79cb7a28f6e64ff9db1de0d9482139ee91f5cdab Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Sun, 25 Aug 2019 13:27:23 +0100 Subject: [PATCH] Add in support for non-linear mapping. --- volatility/framework/interfaces/layers.py | 61 +++++++++++---------- volatility/framework/layers/segmented.py | 64 ++++++++++++++++++++++- 2 files changed, 96 insertions(+), 29 deletions(-) diff --git a/volatility/framework/interfaces/layers.py b/volatility/framework/interfaces/layers.py index 4d3b80d23..33bf45c06 100644 --- a/volatility/framework/interfaces/layers.py +++ b/volatility/framework/interfaces/layers.py @@ -358,19 +358,12 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta): """Returns a list of layer names that this layer translates onto""" return [] - ### Translation layer convenience function + def _decode(self, data: bytes, mapped_offset: int, offset: int) -> bytes: + """Decodes any necessary data""" + return data - def translate(self, offset: int, ignore_errors: bool = False) -> Tuple[Optional[int], Optional[str]]: - mapping = self.mapping(offset, 0, ignore_errors) - if mapping: - _, mapped_offset, _, layer = list(mapping)[0] - else: - if ignore_errors: - # We should only hit this if we ignored errors, but check anyway - return None, None - raise exceptions.InvalidAddressException(self.name, offset, - "Cannot translate {} in layer {}".format(offset, self.name)) - return mapped_offset, layer + def _encode(self, data: bytes, mapped_offset: int, offset: int) -> bytes: + """Encodes any necessary data""" # ## Read/Write functions for mapped pages @@ -379,18 +372,24 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta): """Reads an offset for length bytes and returns 'bytes' (not 'str') of length size""" current_offset = offset output = [] # type: List[bytes] - for (offset, mapped_offset, mapped_length, layer) in self.mapping(offset, length, ignore_errors = pad): - if not pad and offset > current_offset: + for (layer_offset, mapped_offset, mapped_length, layer) in self.mapping(offset, length, ignore_errors = pad): + if not pad and layer_offset > current_offset: raise exceptions.InvalidAddressException( self.name, current_offset, "Layer {} cannot map offset: {}".format(self.name, current_offset)) - elif offset > current_offset: - output += [b"\x00" * (offset - current_offset)] - current_offset = offset - elif offset < current_offset: - raise exceptions.LayerException(self.name, "Mapping returned an overlapping element") + elif layer_offset > current_offset: + output += [b"\x00" * (layer_offset - current_offset)] + current_offset = layer_offset + # The layer_offset can be less than the current_offset in non-linearly mapped layers + # it does not suggest an overlap, but that the data is in an encoded block if mapped_length > 0: - output += [self._context.layers.read(layer, mapped_offset, mapped_length, pad)] - current_offset += mapped_length + processed_data = self._decode( + self._context.layers.read(layer, mapped_offset, mapped_length, pad), mapped_offset, layer_offset) + # Chop off anything unnecessary at the start + processed_data = processed_data[current_offset - layer_offset:] + # Chop off anything unnecessary at the end + processed_data = processed_data[:length - (current_offset - offset)] + output += [processed_data] + current_offset += len(processed_data) recovered_data = b"".join(output) return recovered_data + b"\x00" * (length - len(recovered_data)) @@ -398,15 +397,21 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta): """Writes a value at offset, distributing the writing across any underlying mapping""" current_offset = offset length = len(value) - for (offset, mapped_offset, length, layer) in self.mapping(offset, length): - if offset > current_offset: + for (layer_offset, mapped_offset, mapped_length, layer) in self.mapping(offset, length): + if layer_offset > current_offset: raise exceptions.InvalidAddressException( self.name, current_offset, "Layer {} cannot map offset: {}".format(self.name, current_offset)) - elif offset < current_offset: - raise exceptions.LayerException(self.name, "Mapping returned an overlapping element") - self._context.layers.write(layer, mapped_offset, value[:length]) - value = value[length:] - current_offset += length + original_data = self._context.layers.read(layer, mapped_offset, mapped_length) + # Always chunk the value based on the mapping + value_to_write = original_data[:current_offset - layer_offset] + value[:mapped_length - + (current_offset - layer_offset)] + value = value[mapped_length - (current_offset - layer_offset):] + encoded_value = self._encode(value_to_write, mapped_offset, layer_offset) + if len(encoded_value) != mapped_length: + raise exceptions.LayerException(self.name, + "Unable to write new value, does not map to the same dimensions") + self._context.layers.write(layer, mapped_offset, encoded_value) + current_offset += len(value_to_write) # ## Scan implementation with knowledge of pages diff --git a/volatility/framework/layers/segmented.py b/volatility/framework/layers/segmented.py index afad5dbb4..1278a54a9 100644 --- a/volatility/framework/layers/segmented.py +++ b/volatility/framework/layers/segmented.py @@ -17,7 +17,7 @@ # WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for the # specific language governing rights and limitations under the License. # - +import functools from abc import ABCMeta, abstractmethod from bisect import bisect_right from typing import Any, Dict, Iterable, List, Optional, Tuple @@ -26,6 +26,68 @@ from volatility.framework import exceptions, interfaces from volatility.framework.configuration import requirements +class LinearlyMappedLayer(interfaces.layers.TranslationLayerInterface): + """Class to differentiate Linearly Mapped layers (where a => b implies that a + c => b + c)""" + + ### Translation layer convenience function + + def translate(self, offset: int, ignore_errors: bool = False) -> Tuple[Optional[int], Optional[str]]: + mapping = list(self.mapping(offset, 0, ignore_errors)) + if len(mapping) == 1: + original_offset, mapped_offset, _, layer = mapping[0] + if original_offset != offset: + raise exceptions.LayerException(self.name, + "Layer {} claims to map linearly but does not".format(self.name)) + else: + if ignore_errors: + # We should only hit this if we ignored errors, but check anyway + return None, None + raise exceptions.InvalidAddressException(self.name, offset, + "Cannot translate {} in layer {}".format(offset, self.name)) + return mapped_offset, layer + + # ## Read/Write functions for mapped pages + # Redefine read here for speed reasons (so we don't call a processing method + + @functools.lru_cache(maxsize = 512) + def read(self, offset: int, length: int, pad: bool = False) -> bytes: + """Reads an offset for length bytes and returns 'bytes' (not 'str') of length size""" + current_offset = offset + output = [] # type: List[bytes] + for (offset, mapped_offset, mapped_length, layer) in self.mapping(offset, length, ignore_errors = pad): + if not pad and offset > current_offset: + raise exceptions.InvalidAddressException( + self.name, current_offset, "Layer {} cannot map offset: {}".format(self.name, current_offset)) + elif offset > current_offset: + output += [b"\x00" * (offset - current_offset)] + current_offset = offset + elif offset < current_offset: + raise exceptions.LayerException(self.name, "Mapping returned an overlapping element") + if mapped_length > 0: + output += [self._context.layers.read(layer, mapped_offset, mapped_length, pad)] + current_offset += mapped_length + recovered_data = b"".join(output) + return recovered_data + b"\x00" * (length - len(recovered_data)) + + def write(self, offset: int, value: bytes) -> None: + """Writes a value at offset, distributing the writing across any underlying mapping""" + current_offset = offset + length = len(value) + for (offset, mapped_offset, length, layer) in self.mapping(offset, length): + if offset > current_offset: + raise exceptions.InvalidAddressException( + self.name, current_offset, "Layer {} cannot map offset: {}".format(self.name, current_offset)) + elif offset < current_offset: + raise exceptions.LayerException(self.name, "Mapping returned an overlapping element") + self._context.layers.write(layer, mapped_offset, value[:length]) + value = value[length:] + current_offset += length + + +class NonLinearlyMappedLayer(interfaces.layers.TranslationLayerInterface): + """Class to allow layers which don't map linearly to exist""" + + class SegmentedLayer(interfaces.layers.TranslationLayerInterface, metaclass = ABCMeta): """A class to handle a single run-based layer-to-layer mapping