diff --git a/setup.py b/setup.py index 62f5bd2d5..8e9f88217 100644 --- a/setup.py +++ b/setup.py @@ -6,7 +6,7 @@ import setuptools from volatility3.framework import constants -with open("README.md", "r", encoding="utf-8") as fh: +with open("README.md", "r", encoding = "utf-8") as fh: long_description = fh.read() setuptools.setup(name = "volatility3", @@ -45,4 +45,5 @@ setuptools.setup(name = "volatility3", 'crypto': ["pycryptodome>=3"], 'disasm': ["capstone;platform_system=='Linux'", "capstone-windows;platform_system=='Windows'"], 'doc': ["sphinx>=1.8.2", "sphinx_autodoc_typehints>=1.4.0", "sphinx-rtd-theme>=0.4.3"], + 'avml': ["python-snappy==0.6.0"], }) diff --git a/volatility3/framework/layers/avml.py b/volatility3/framework/layers/avml.py new file mode 100644 index 000000000..83c43fcc9 --- /dev/null +++ b/volatility3/framework/layers/avml.py @@ -0,0 +1,136 @@ +"""Functions that read AVML files. + +The user of the file doesn't have to worry about the compression, +but random access is not allowed.""" +import logging +import struct +from typing import Tuple, List, Optional + +from volatility3.framework import exceptions, interfaces, constants +from volatility3.framework.layers import segmented + +vollog = logging.getLogger(__name__) + +try: + import snappy + + HAS_SNAPPY = True +except ImportError: + HAS_SNAPPY = False + + +class AVMLLayer(segmented.NonLinearlySegmentedLayer): + """A Lime format TranslationLayer. + + Lime is generally used to store physical memory images where there + are large holes in the physical layer + """ + + def __init__(self, *args, **kwargs): + self._compressed = {} + super().__init__(*args, **kwargs) + + @classmethod + def _check_header(cls, layer: interfaces.layers.DataLayerInterface): + header_structure = " None: + base_layer = self.context.layers[self._base_layer] + offset = base_layer.minimum_address + while offset + 4 < base_layer.maximum_address: + avml_header_structure = " Tuple[ + List[Tuple[int, int, int, int, bool]], int]: + """ + Reads a framed-format snappy stream + + Args: + data: The stream to read + expected_length: How big the decompressed stream is expected to be (termination limit) + + Returns: + (offset, mapped_offset, length, mapped_length, compressed) relative to the data chunk (ie, not relative to the file start) + """ + segments = [] + decompressed_len = 0 + offset = 0 + crc_len = 4 + frame_header_struct = '> 8 + if frame_type == 0xff: + if data[offset + frame_header_len:offset + frame_header_len + frame_size] != b'sNaPpY': + raise ValueError(f"Snappy header missing at offset: {offset}") + elif frame_type in [0x00, 0x01]: + # CRC + (Un)compressed data + mapped_start = offset + frame_header_len + frame_crc = data[mapped_start: mapped_start + crc_len] + frame_data = data[mapped_start + crc_len: mapped_start + frame_size] + if frame_type == 0x00: + # Compressed data + frame_data = snappy.decompress(frame_data) + # TODO: Verify CRC + segments.append((decompressed_len, mapped_start + crc_len, len(frame_data), frame_size - crc_len, + frame_type == 0x00)) + decompressed_len += len(frame_data) + elif frame_type in range(0x2, 0x80): + # Unskippable + raise exceptions.LayerException(f"Unskippable chunk of type {frame_type} found: {offset}") + offset += frame_header_len + frame_size + return segments, offset + + def _decode_data(self, data: bytes, mapped_offset: int, offset: int, output_length: int) -> bytes: + start_offset, _, _, _ = self._find_segment(offset) + if self._compressed[mapped_offset]: + decoded_data = snappy.decompress(data) + else: + decoded_data = data + decoded_data = decoded_data[offset - start_offset:] + decoded_data = decoded_data[:output_length] + return decoded_data + + +class AVMLStacker(interfaces.automagic.StackerLayerInterface): + stack_order = 10 + + @classmethod + def stack(cls, + context: interfaces.context.ContextInterface, + layer_name: str, + progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: + try: + AVMLLayer._check_header(context.layers[layer_name]) + except exceptions.LayerException: + return None + new_name = context.layers.free_layer_name("AVMLLayer") + context.config[interfaces.configuration.path_join(new_name, "base_layer")] = layer_name + return AVMLLayer(context, new_name, new_name)