diff --git a/requirements-dev.txt b/requirements-dev.txt index 7c372da2a..9db14d441 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -20,7 +20,3 @@ jsonschema>=2.3.0 # This is required for memory acquisition via leechcore/pcileech. leechcorepyc>=2.4.0 - -# This is required for analyzing Linux samples compressed using AVMLs native -# compression format. It is not required for AVML's standard LiME compression. -python-snappy==0.6.0 diff --git a/requirements.txt b/requirements.txt index 1793012f1..99e0786cc 100644 --- a/requirements.txt +++ b/requirements.txt @@ -16,7 +16,3 @@ pycryptodome # This is required for memory acquisition via leechcore/pcileech. leechcorepyc>=2.4.0 - -# This is required for analyzing Linux samples compressed using AVMLs native -# compression format. It is not required for AVML's standard LiME compression. -python-snappy==0.6.0 diff --git a/volatility3/framework/layers/avml.py b/volatility3/framework/layers/avml.py index 66f3f0e4f..3ce25ca6f 100644 --- a/volatility3/framework/layers/avml.py +++ b/volatility3/framework/layers/avml.py @@ -6,6 +6,7 @@ The user of the file doesn't have to worry about the compression, but random access is not allowed.""" +import ctypes import logging import struct from typing import Tuple, List, Optional @@ -16,13 +17,35 @@ from volatility3.framework.layers import segmented vollog = logging.getLogger(__name__) try: - import snappy + from ctypes import cdll + + # TODO: Find library for windows if needed + lib_snappy = cdll.LoadLibrary("libsnappy.so.1") + __snappy_uncompress = lib_snappy.snappy_uncompress + __snappy_uncompressed_length = lib_snappy.snappy_uncompressed_length HAS_SNAPPY = True -except ImportError: +except OSError: HAS_SNAPPY = False +class SnappyException(Exception): + pass + + +def uncompress(s): + """Uncompress a snappy compressed string.""" + ulen = ctypes.c_int(0) + cresult = __snappy_uncompressed_length(s, len(s), ctypes.byref(ulen)) + if cresult != 0: + raise SnappyException(f"Error in snappy_uncompressed_length: {cresult}") + ubuf = ctypes.create_string_buffer(ulen.value) + __snappy_uncompress(s, len(s), ubuf, ctypes.byref(ulen)) + if cresult != 0: + raise SnappyException(f"Error in snappy_uncompress: {cresult}") + return ubuf.raw + + class AVMLLayer(segmented.NonLinearlySegmentedLayer): """A Lime format TranslationLayer. @@ -44,9 +67,7 @@ class AVMLLayer(segmented.NonLinearlySegmentedLayer): if magic not in [0x4C4D5641] or version != 2: raise exceptions.LayerException("File not completely in AVML format") if not HAS_SNAPPY: - vollog.warning( - "AVML file detected, but snappy python library not installed" - ) + vollog.warning("AVML file detected, but snappy library could not be found") raise exceptions.LayerException( "AVML format dependencies not satisfied (snappy)" ) @@ -131,7 +152,7 @@ class AVMLLayer(segmented.NonLinearlySegmentedLayer): ] if frame_type == 0x00: # Compressed data - frame_data = snappy.decompress(frame_data) + frame_data = uncompress(frame_data) # TODO: Verify CRC segments.append( ( @@ -156,7 +177,7 @@ class AVMLLayer(segmented.NonLinearlySegmentedLayer): ) -> bytes: start_offset, _, _, _ = self._find_segment(offset) if self._compressed[mapped_offset]: - decoded_data = snappy.decompress(data) + decoded_data = uncompress(data) else: decoded_data = data decoded_data = decoded_data[offset - start_offset :]