diff --git a/volatility3/framework/plugins/windows/prefetch.py b/volatility3/framework/plugins/windows/prefetch.py deleted file mode 100644 index a5f054f15..000000000 --- a/volatility3/framework/plugins/windows/prefetch.py +++ /dev/null @@ -1,535 +0,0 @@ -# References : -# https://winprotocoldoc.blob.core.windows.net/productionwindowsarchives/MS-XCA/%5bMS-XCA%5d.pdf -# https://github.com/libyal/libscca/blob/main/documentation/Windows%20Prefetch%20File%20(PF)%20format.asciidoc -# https://github.com/volatilityfoundation/volatility3/ -# https://github.com/EricZimmerman/Prefetch/tree/master/Prefetch -import logging -import pathlib -import datetime -import io -import struct - -from volatility3.framework import renderers, interfaces, exceptions, constants -from volatility3.framework.configuration import requirements -from volatility3.plugins.windows import filescan -from volatility3.framework.renderers import format_hints, conversion -from typing import Tuple, List, Union - -vollog = logging.getLogger(__name__) - - -class BitStream: - def __init__(self, source: bytes, in_pos: int): - self.source = source - self.index = in_pos + 4 - # read UInt16 little endian - mask = struct.unpack_from(" int: - if n == 0: - return 0 - return self.mask >> (32 - n) - - def skip(self, n: int) -> Union[None, Exception]: - self.mask = (self.mask << n) & 0xFFFFFFFF - self.bits -= n - if self.bits < 16: - if self.index + 2 > len(self.source): - return Exception("EOF Error") - # read UInt16 little endian - self.mask += ( - (struct.unpack_from(" int: - node = treeNodes[0] - i = leafIndex + 1 - childIndex = None - - while bits > 1: - bits -= 1 - childIndex = (mask >> bits) & 1 - if node.child[childIndex] is None: - node.child[childIndex] = treeNodes[i] - treeNodes[i].leaf = False - i += 1 - node = node.child[childIndex] - - node.child[mask & 1] = treeNodes[leafIndex] - - return i - - -def prefix_code_tree_rebuild(input: bytes) -> PREFIX_CODE_NODE: - treeNodes = [PREFIX_CODE_NODE() for _ in range(1024)] - symbolInfo = [PREFIX_CODE_SYMBOL() for _ in range(512)] - - for i in range(256): - value = input[i] - - symbolInfo[2 * i].id = 2 * i - symbolInfo[2 * i].symbol = 2 * i - symbolInfo[2 * i].length = value & 0xF - - value >>= 4 - - symbolInfo[2 * i + 1].id = 2 * i + 1 - symbolInfo[2 * i + 1].symbol = 2 * i + 1 - symbolInfo[2 * i + 1].length = value & 0xF - - symbolInfo = sorted(symbolInfo, key=lambda x: (x.length, x.symbol)) - - i = 0 - while i < 512 and symbolInfo[i].length == 0: - i += 1 - - mask = 0 - bits = 1 - - root = treeNodes[0] - root.leaf = False - - j = 1 - while i < 512: - treeNodes[j].id = j - treeNodes[j].symbol = symbolInfo[i].symbol - treeNodes[j].leaf = True - mask = mask << (symbolInfo[i].length - bits) - bits = symbolInfo[i].length - j = prefix_code_tree_add_leaf(treeNodes, j, mask, bits) - mask += 1 - i += 1 - - return root - - -def prefix_code_tree_decode_symbol( - bstr: BitStream, root: PREFIX_CODE_NODE -) -> Tuple[int, Union[None, Exception]]: - node = root - while True: - bit = bstr.lookup(1) - err = bstr.skip(1) - if err is not None: - return 0, err - - node = node.child[bit] - if node is None: - return 0, Exception("Corruption detected") - - if node.leaf: - break - return node.symbol, None - - -def lz77_huffman_decompress_chunck( - in_idx: int, input: bytes, out_idx: int, output: bytearray, chunk_size: int -) -> Tuple[int, int, Union[None, Exception]]: - # Ensure there are at least 256 bytes available to read - if in_idx + 256 > len(input): - return 0, 0, Exception("EOF Error") - - root = prefix_code_tree_rebuild(input[in_idx:]) - # print_tree(root) - bstr = BitStream(input, in_idx + 256) - - i = out_idx - - while i < out_idx + chunk_size: - symbol, err = prefix_code_tree_decode_symbol(bstr, root) - - if err is not None: - return int(bstr.index), i, err - - if symbol < 256: - output[i] = symbol - i += 1 - else: - symbol -= 256 - length = symbol & 15 - symbol >>= 4 - - offset = 0 - if symbol != 0: - offset = int(bstr.lookup(symbol)) - - offset |= 1 << symbol - offset = -offset - - if length == 15: - length = bstr.source[bstr.index] + 15 - bstr.index += 1 - - if length == 270: - length = struct.unpack_from(" 0: - if i + offset < 0: - print(i + offset) - return int(bstr.index), i, Exception("Decompression Error") - - output[i] = output[i + offset] - i += 1 - length -= 1 - if length == 0: - break - return int(bstr.index), i, None - - -def lz77_huffman_decompress( - input: bytes, output_size: int -) -> Tuple[bytes, Union[None, Exception]]: - output = bytearray(output_size) - err = None - - # Index into the input buffer. - in_idx = 0 - - # Index into the output buffer. - out_idx = 0 - - while True: - # How much data belongs in the current chunk. Chunks - # are split into maximum 65536 bytes. - chunk_size = output_size - out_idx - if chunk_size > 65536: - chunk_size = 65536 - - in_idx, out_idx, err = lz77_huffman_decompress_chunck( - in_idx, input, out_idx, output, chunk_size - ) - if err is not None: - return output, err - if out_idx >= len(output) or in_idx >= len(input): - break - return output, None - - -class Prefetch(interfaces.plugins.PluginInterface): - """Get and parse the prefetch files""" - - _required_framework_version = (2, 0, 0) - _version = (1, 0, 0) - - @classmethod - def get_requirements(cls): - return [ - requirements.ModuleRequirement( - name="kernel", - description="Windows kernel", - architectures=["Intel32", "Intel64"], - ), - requirements.PluginRequirement( - name="filescan", plugin=filescan.FileScan, version=(0, 0, 0) - ), - ] - - @classmethod - def version_17(cls, prefetch_file): - """Extract pf information for Version 17""" - stream = io.BytesIO(prefetch_file) - - stream.seek(0x000C) - file_size = int.from_bytes(stream.read(4), "little") - - stream.seek(0x0010) - executable_raw = stream.read(60).decode("utf-16") - executable_name = executable_raw.split("\u0000")[0] - - stream.seek(0x004C) - prefetch_hash = int.from_bytes(stream.read(4), "little") - - stream.seek(0x0078) - last_execution_filetime = int.from_bytes(stream.read(8), "little") - last_execution_filetime_human = conversion.wintime_to_datetime( - last_execution_filetime - ) - - stream.seek(0x0090) - execution_counter = int.from_bytes(stream.read(4), "little") - - yield ( - executable_name, - file_size, - format_hints.Hex(prefetch_hash), - last_execution_filetime_human, - execution_counter, - ) - - @classmethod - def version_23(cls, prefetch_file): - """Extract pf information for Version 23""" - stream = io.BytesIO(prefetch_file) - - stream.seek(0x000C) - file_size = int.from_bytes(stream.read(4), "little") - - stream.seek(0x0010) - executable_raw = stream.read(60).decode("utf-16") - executable_name = executable_raw.split("\u0000")[0] - - stream.seek(0x004C) - prefetch_hash = int.from_bytes(stream.read(4), "little") - - stream.seek(0x0080) - last_execution_filetime = int.from_bytes(stream.read(8), "little") - last_execution_filetime_human = conversion.wintime_to_datetime( - last_execution_filetime - ) - - stream.seek(0x0098) - execution_counter = int.from_bytes(stream.read(4), "little") - - yield ( - executable_name, - file_size, - format_hints.Hex(prefetch_hash), - last_execution_filetime_human, - execution_counter, - ) - - @classmethod - def version_26(cls, prefetch_file): - """Extract pf information for Version 26""" - stream = io.BytesIO(prefetch_file) - - stream.seek(0x000C) - file_size = int.from_bytes(stream.read(4), "little") - - stream.seek(0x0010) - executable_raw = stream.read(60).decode("utf-16") - executable_name = executable_raw.split("\u0000")[0] - - stream.seek(0x004C) - prefetch_hash = int.from_bytes(stream.read(4), "little") - - stream.seek(0x0080) - last_execution_filetime = int.from_bytes(stream.read(8), "little") - last_execution_filetime_human = conversion.wintime_to_datetime( - last_execution_filetime - ) - - stream.seek(0x00D0) - execution_counter = int.from_bytes(stream.read(4), "little") - - yield ( - executable_name, - file_size, - format_hints.Hex(prefetch_hash), - last_execution_filetime_human, - execution_counter, - ) - - @classmethod - def version_30(cls, prefetch_file): - """Extract pf information for Version 30""" - stream = io.BytesIO(prefetch_file) - - stream.seek(0x000C) - file_size = int.from_bytes(stream.read(4), "little") - - stream.seek(0x0010) - executable_raw = stream.read(60).decode("utf-16") - executable_name = executable_raw.split("\u0000")[0] - - stream.seek(0x004C) - prefetch_hash = int.from_bytes(stream.read(4), "little") - - stream.seek(0x0080) - # The first FILETIME is the most recent run time - last_execution_filetime = int.from_bytes(stream.read(8), "little") - last_execution_filetime_human = conversion.wintime_to_datetime( - last_execution_filetime - ) - - stream.seek(0x00C8) # Variant 1 - execution_counter = int.from_bytes(stream.read(4), "little") - if execution_counter == 0: - stream.seek(0x00D0) # Variant 2 - execution_counter = int.from_bytes(stream.read(4), "little") - - yield ( - executable_name, - file_size, - format_hints.Hex(prefetch_hash), - last_execution_filetime_human, - execution_counter, - ) - - @classmethod - def parse_prefetch(cls, prefetch_file): - WinXpOrWin2K3 = 17 - VistaOrWin7 = 23 - Win8xOrWin2012x = 26 - Win10OrWin11 = 30 - stream = io.BytesIO(prefetch_file) - # First, we need to know if the prefetch is compressed (Win10/11) - signature = prefetch_file[:3].decode() - if signature == "MAM": - vollog.info("Windows 1X prefetch file detected.") - # The size of decompressed data is at offset 4 - stream.seek(0x0004) - decompressed_size = int.from_bytes(stream.read(4), "little") - vollog.info(f"decompressed size : {decompressed_size}") - stream.seek(0x0008) - compressed_bytes = stream.read() - prefetch_file = lz77_huffman_decompress( - bytearray(compressed_bytes), decompressed_size - )[0] - try: - file_version = int.from_bytes(prefetch_file[:4], "little") - signature = prefetch_file[4:8].decode() - vollog.info(f"File version : {file_version}") - vollog.info(f"Signature : {signature}") - except Exception: - # We can not even read the header - pass - - if signature != "SCCA": - vollog.info("Wrong signature, should be SCCA") - return - if file_version == WinXpOrWin2K3: - for result in cls.version_17(prefetch_file): - yield result - elif file_version == VistaOrWin7: - for result in cls.version_23(prefetch_file): - yield result - elif file_version == Win8xOrWin2012x: - for result in cls.version_26(prefetch_file): - yield result - elif file_version == Win10OrWin11: - for result in cls.version_30(prefetch_file): - yield result - - def _generator(self, files): - kernel = self.context.modules[self.config["kernel"]] - for file_obj in files: - """Get the prefetch recovered files from the “filescan” plugin;""" - try: - file_name = file_obj.FileName.String - file_extension = pathlib.Path(file_name).suffix - if file_extension == ".pf": - """If found, try to dump the prefetch file (inspired from the "DumpFiles" plugin)""" - memory_objects = [] - memory_layer_name = self.context.layers[kernel.layer_name].config[ - "memory_layer" - ] - memory_layer = self.context.layers[memory_layer_name] - primary_layer = self.context.layers[kernel.layer_name] - for member_name in ["DataSectionObject", "ImageSectionObject"]: - try: - section_obj = getattr( - file_obj.SectionObjectPointer, member_name - ) - control_area = section_obj.dereference().cast( - "_CONTROL_AREA" - ) - if control_area.is_valid(): - vollog.info(f"Found : {file_obj.FileName.String}") - memory_objects.append((control_area, memory_layer)) - except exceptions.InvalidAddressException: - vollog.log( - constants.LOGLEVEL_VVV, - f"{member_name} is unavailable for file {file_obj.vol.offset:#x}", - ) - try: - scm_pointer = file_obj.SectionObjectPointer.SharedCacheMap - shared_cache_map = scm_pointer.dereference().cast( - "_SHARED_CACHE_MAP" - ) - if shared_cache_map.is_valid(): - memory_objects.append((shared_cache_map, primary_layer)) - except exceptions.InvalidAddressException: - vollog.log( - constants.LOGLEVEL_VVV, - f"SharedCacheMap is unavailable for file {file_obj.vol.offset:#x}", - ) - vollog.info(f"memory_objects : {memory_objects}") - - """Now, read and parse our PF to retrieve our artifacts""" - for memory_object, layer in memory_objects: - bytes_read = 0 - prefetch_raw = b"" - try: - for ( - mem_offset, - _, - datasize, - ) in memory_object.get_available_pages(): - prefetch_raw += layer.read( - mem_offset, datasize, pad=True - ) - bytes_read += len(prefetch_raw) - vollog.info(f"Read {bytes_read}") - if not bytes_read: - vollog.info("Prefetch is empty") - else: - """Prefetch parsing""" - for result in self.parse_prefetch(prefetch_raw): - yield 0, result - - except exceptions.InvalidAddressException: - vollog.debug( - f"Unable to dump file at {file_obj.vol.offset:#x}" - ) - - except exceptions.InvalidAddressException: - continue - - def run(self): - kernel = self.context.modules[self.config["kernel"]] - return renderers.TreeGrid( - [ - ("ExecutableName", str), - ("FileSize", int), - ("PrefetchHash", format_hints.Hex), - ("LastExecution", datetime.datetime), - ("ExecutionCounter", int), - ], - self._generator( - filescan.FileScan.scan_files( - self.context, kernel.layer_name, kernel.symbol_table_name - ) - ), - )