From dea0e156890e620ef78aeeadce8139a81e8e9042 Mon Sep 17 00:00:00 2001 From: atcuno Date: Mon, 22 Jul 2024 16:27:05 -0500 Subject: [PATCH] Address feedback --- .../framework/plugins/windows/mftscan.py | 338 +++++++++++------- 1 file changed, 215 insertions(+), 123 deletions(-) diff --git a/volatility3/framework/plugins/windows/mftscan.py b/volatility3/framework/plugins/windows/mftscan.py index ec8dd5fe8..0d858cec1 100644 --- a/volatility3/framework/plugins/windows/mftscan.py +++ b/volatility3/framework/plugins/windows/mftscan.py @@ -5,7 +5,7 @@ import contextlib import datetime import logging -from typing import Generator, Iterable +from typing import Generator, Iterable, Dict, Tuple from volatility3.framework import constants, exceptions, interfaces, renderers from volatility3.framework.configuration import requirements @@ -22,13 +22,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): _required_framework_version = (2, 0, 0) - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self._record_map = {} - self.mft_object = None - self.attribute_object = None - self.si_object = None - self.fn_object = None + _version = (2, 0, 0) @classmethod def get_requirements(cls): @@ -43,9 +37,15 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): ), ] - def enumerate_mft_records(self, attr_callback): + @staticmethod + def enumerate_mft_records( + context: interfaces.context.ContextInterface, + config: interfaces.configuration.HierarchicalDict, + config_path: str, + attr_callback + ) -> interfaces.objects.ObjectInterface: try: - primary = self.context.layers[self.config["primary"]] + primary = context.layers[config["primary"]] except KeyError: vollog.error( "Unable to obtain primary layer for scanning. Please file a bug on GitHub about this issue." @@ -60,7 +60,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): ) return - layer = self.context.layers[phys_layer] + layer = context.layers[phys_layer] # Yara Rule to scan for MFT Header Signatures rules = yarascan.YaraScan.process_yara_options( @@ -69,8 +69,8 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): # Read in the Symbol File symbol_table = intermed.IntermediateSymbolTable.create( - context=self.context, - config_path=self.config_path, + context=context, + config_path=config_path, sub_path="windows", filename="mft", class_types={ @@ -81,23 +81,21 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): ) # get each of the individual Field Sets - self.mft_object = symbol_table + constants.BANG + "MFT_ENTRY" - self.attribute_object = symbol_table + constants.BANG + "ATTRIBUTE" - self.si_object = symbol_table + constants.BANG + "STANDARD_INFORMATION_ENTRY" - self.fn_object = symbol_table + constants.BANG + "FILE_NAME_ENTRY" + mft_object = symbol_table + constants.BANG + "MFT_ENTRY" + attribute_object = symbol_table + constants.BANG + "ATTRIBUTE" # Scan the layer for Raw MFT records and parse the fields for offset, _, _, _ in layer.scan( - context=self.context, scanner=yarascan.YaraScanner(rules=rules) + context=context, scanner=yarascan.YaraScanner(rules=rules) ): with contextlib.suppress(exceptions.InvalidAddressException): - mft_record = self.context.object( - self.mft_object, offset=offset, layer_name=layer.name + mft_record = context.object( + mft_object, offset=offset, layer_name=layer.name ) # We will update this on each pass in the next loop and use it as the new offset. attr_base_offset = mft_record.FirstAttrOffset - attr = self.context.object( - self.attribute_object, + attr = context.object( + attribute_object, offset=offset + attr_base_offset, layer_name=layer.name, ) @@ -105,7 +103,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): # There is no field that has a count of Attributes # Keep Attempting to read attributes until we get an invalid attr_header.AttrType while attr.Attr_Header.AttrType.is_valid_choice: - for record in attr_callback(mft_record, attr): + for record in attr_callback(mft_record, attr, symbol_table): yield record # If there's no advancement the loop will never end, so break it now @@ -115,13 +113,14 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): # Update the base offset to point to the next attribute attr_base_offset += attr.Attr_Header.Length # Get the next attribute - attr = self.context.object( - self.attribute_object, + attr = context.object( + attribute_object, offset=offset + attr_base_offset, layer_name=layer.name, ) - def parse_mft_records(self, mft_record, attr): + @staticmethod + def parse_mft_records(mft_record, attr, symbol_table): # MFT Flags determine the file type or dir # If we don't have a valid enum, coerce to hex so we can keep the record try: @@ -131,7 +130,8 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): # Standard Information Attribute if attr.Attr_Header.AttrType.lookup() == "STANDARD_INFORMATION": - attr_data = attr.Attr_Data.cast(self.si_object) + si_object = symbol_table + constants.BANG + "STANDARD_INFORMATION_ENTRY" + attr_data = attr.Attr_Data.cast(si_object) yield 0, ( format_hints.Hex(attr_data.vol.offset), mft_record.get_signature(), @@ -149,7 +149,9 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): # File Name Attribute elif attr.Attr_Header.AttrType.lookup() == "FILE_NAME": - attr_data = attr.Attr_Data.cast(self.fn_object) + fn_object = symbol_table + constants.BANG + "FILE_NAME_ENTRY" + + attr_data = attr.Attr_Data.cast(fn_object) file_name = attr_data.get_full_name() # If we don't have a valid enum, coerce to hex so we can keep the record @@ -173,8 +175,114 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): file_name, ) + @staticmethod + def parse_data_record( + mft_record: interfaces.objects.ObjectInterface, + attr: interfaces.objects.ObjectInterface, + record_map: Dict[int, Tuple[str, int, int]], + return_first_record: bool, + ) -> Generator[Iterable, None, None]: + """ + Returns the parsed data from a MFT record + """ + # we only care about resident data + if attr.Attr_Header.NonResidentFlag: + return + + content = attr.get_resident_filecontent() + if content: + content = format_hints.HexBytes(content) + else: + content = renderers.NotAvailableValue() + + # past the first $DATA record, attempt to get the ADS name + # NotApplicableValue = 1st Data + # NotAvailableValue = > 1st Data, but name was not parsable + ads_name = renderers.NotApplicableValue() + if not return_first_record and attr.Attr_Header.NameLength > 0: + ads_name = attr.get_resident_filename() + if not ads_name: + ads_name = renderers.NotAvailableValue() + + yield ( + format_hints.Hex(record_map[mft_record.RecordNumber][2]), + mft_record.get_signature(), + mft_record.RecordNumber, + attr.Attr_Header.AttrType.lookup(), + record_map[mft_record.RecordNumber][0], + ads_name, + content, + ) + + @classmethod + def _do_parse_data_records( + cls, + mft_record: interfaces.objects.ObjectInterface, + attr: interfaces.objects.ObjectInterface, + symbol_table, + record_map: Dict[int, Tuple[str, int, int]], + return_first_record: bool, + ) -> Generator[Iterable, None, None]: + """ + Parses DATA records while maintaining the FILE_NAME association + from previous parsing of the record + Suports returning the first/main $DATA as well as however many + ADS records a file might have + """ + rec_num = mft_record.RecordNumber + if rec_num not in record_map: + # file name, DATA count, offset + record_map[rec_num] = [renderers.NotAvailableValue(), 0, None] + + if attr.Attr_Header.AttrType.lookup() == "FILE_NAME": + fn_object = symbol_table + constants.BANG + "FILE_NAME_ENTRY" + attr_data = attr.Attr_Data.cast(fn_object) + rec_name = attr_data.get_full_name() + record_map[rec_num][0] = rec_name + elif attr.Attr_Header.AttrType.lookup() == "DATA": + # first data + record_map[rec_num][2] = attr.Attr_Data.vol.offset + + display_data = False + + # first DATA attribute of this record + if record_map[rec_num][1] == 0 and return_first_record: + if return_first_record: + display_data = True + else: + record_map[rec_num][1] = 1 + + # at the second DATA attribute of this record + elif not return_first_record: + display_data = True + + if display_data: + for record in cls.parse_data_record( + mft_record, attr, record_map, return_first_record + ): + yield record + + @classmethod + def parse_data_records( + cls, + mft_record: interfaces.objects.ObjectInterface, + attr: interfaces.objects.ObjectInterface, + symbol_table, + return_first_record: bool, + ): + """ + Callback for parsing data records through enumerate_mft_records + """ + record_map = {} + for record in cls._do_parse_data_records( + mft_record, attr, symbol_table, record_map, return_first_record + ): + yield record + def _generator(self): - for record in self.enumerate_mft_records(self.parse_mft_records): + for record in self.enumerate_mft_records( + self.context, self.config, self.config_path, self.parse_mft_records + ): yield record def generate_timeline(self): @@ -210,103 +318,53 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): ) -class ADS(MFTScan): +class ADS(interfaces.plugins.PluginInterface): """Scans for Alternate Data Stream""" - _required_framework_version = (2, 0, 0) + _required_framework_version = (2, 7, 0) - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) + _version = (1, 0, 0) - # which DATA attribute should be displayed - self._display_first_data = False + @classmethod + def get_requirements(cls): + return [ + requirements.PluginRequirement( + name="MFTScan", plugin=MFTScan, version=(2, 0, 0) + ), + requirements.TranslationLayerRequirement( + name="primary", + description="Memory layer for the kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="yarascanner", component=yarascan.YaraScanner, version=(2, 0, 0) + ), + ] - def _parse_data_record( - self, + @staticmethod + def parse_ads_data_records( mft_record: interfaces.objects.ObjectInterface, attr: interfaces.objects.ObjectInterface, - ) -> Generator[Iterable, None, None]: - # we only care about resident data - if attr.Attr_Header.NonResidentFlag: - return - - # regular $DATA - elif self._display_first_data: - content = attr.get_resident_filecontent() - if content: - content = format_hints.HexBytes(content) - else: - content = renderers.NotAvailableValue() - - yield 0, ( - format_hints.Hex(self._record_map[mft_record.RecordNumber][2]), - mft_record.get_signature(), - mft_record.RecordNumber, - attr.Attr_Header.AttrType.lookup(), - self._record_map[mft_record.RecordNumber][0], - content, - ) - - # ADS $DATA - elif attr.Attr_Header.NameLength > 0: - ads_name = attr.get_resident_filename() - if not ads_name: - ads_name = renderers.NotAvailableValue() - - content = attr.get_resident_filecontent() - if content: - content = format_hints.HexBytes(content) - else: - content = renderers.NotAvailableValue() - - yield 0, ( - format_hints.Hex(self._record_map[mft_record.RecordNumber][2]), - mft_record.get_signature(), - mft_record.RecordNumber, - attr.Attr_Header.AttrType.lookup(), - self._record_map[mft_record.RecordNumber][0], - ads_name, - content, - ) - - def parse_data_records( - self, - mft_record: interfaces.objects.ObjectInterface, - attr: interfaces.objects.ObjectInterface, - ) -> Generator[Iterable, None, None]: - rec_num = mft_record.RecordNumber - if rec_num not in self._record_map: - # file name, DATA count, offset - self._record_map[rec_num] = [renderers.NotAvailableValue(), 0, None] - - if attr.Attr_Header.AttrType.lookup() == "FILE_NAME": - attr_data = attr.Attr_Data.cast(self.fn_object) - rec_name = attr_data.get_full_name() - self._record_map[rec_num][0] = rec_name - elif attr.Attr_Header.AttrType.lookup() == "DATA": - # first data - self._record_map[rec_num][2] = attr.Attr_Data.vol.offset - - display_data = False - - # first DATA attribute of this record - if self._record_map[rec_num][1] == 0: - if self._display_first_data: - display_data = True - else: - self._record_map[rec_num][1] = 1 - - # at the second DATA attribute of this record - elif not self._display_first_data: - display_data = True - - if display_data: - for record in self._parse_data_record(mft_record, attr): - yield record + symbol_table, + ): + return MFTScan.parse_data_records(mft_record, attr, symbol_table, False) def _generator(self): - for record in self.enumerate_mft_records(self.parse_data_records): - yield record + for ( + offset, + rec_type, + rec_num, + attr_type, + file_name, + ads_name, + content, + ) in MFTScan.enumerate_mft_records( + self.context, self.config, self.config_path, self.parse_ads_data_records + ): + yield ( + 0, + (offset, rec_type, rec_num, attr_type, file_name, ads_name, content), + ) def run(self): return renderers.TreeGrid( @@ -323,16 +381,50 @@ class ADS(MFTScan): ) -class ResidentData(ADS): +class ResidentData(interfaces.plugins.PluginInterface): """Scans for Alternate Data Stream""" - _required_framework_version = (2, 0, 0) + _required_framework_version = (2, 7, 0) - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) + _version = (1, 0, 0) - # which DATA attribute should be displayed - self._display_first_data = True + @classmethod + def get_requirements(cls): + return [ + requirements.PluginRequirement( + name="MFTScan", plugin=MFTScan, version=(2, 0, 0) + ), + requirements.TranslationLayerRequirement( + name="primary", + description="Memory layer for the kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="yarascanner", component=yarascan.YaraScanner, version=(2, 0, 0) + ), + ] + + @staticmethod + def parse_first_data_records( + mft_record: interfaces.objects.ObjectInterface, + attr: interfaces.objects.ObjectInterface, + symbol_table, + ): + return MFTScan.parse_data_records(mft_record, attr, symbol_table, True) + + def _generator(self): + for ( + offset, + rec_type, + rec_num, + attr_type, + file_name, + _, + content, + ) in MFTScan.enumerate_mft_records( + self.context, self.config, self.config_path, self.parse_first_data_records + ): + yield (0, (offset, rec_type, rec_num, attr_type, file_name, content)) def run(self): return renderers.TreeGrid(