From 223fdc6dd938fe10cdde44a1d8ec9ef7b304a216 Mon Sep 17 00:00:00 2001 From: atcuno Date: Wed, 19 Jun 2024 17:58:49 -0500 Subject: [PATCH] placeholder --- .../framework/plugins/windows/mftscan.py | 367 +++++++++--------- 1 file changed, 187 insertions(+), 180 deletions(-) diff --git a/volatility3/framework/plugins/windows/mftscan.py b/volatility3/framework/plugins/windows/mftscan.py index 9e6585345..0687c7796 100644 --- a/volatility3/framework/plugins/windows/mftscan.py +++ b/volatility3/framework/plugins/windows/mftscan.py @@ -20,6 +20,10 @@ 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 = {} + @classmethod def get_requirements(cls): return [ @@ -33,8 +37,10 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): ), ] - def _generator(self): - layer = self.context.layers[self.config["primary"]] + def enumerate_mft_records(self, attr_callback): + phys_layer = self.context.layers[self.config["primary"]].config["memory_layer"] + + layer = self.context.layers[phys_layer] # Yara Rule to scan for MFT Header Signatures rules = yarascan.YaraScan.process_yara_options( @@ -47,87 +53,36 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): config_path=self.config_path, sub_path="windows", filename="mft", - class_types={"FILE_NAME_ENTRY": mft.MFTFileName, "MFT_ENTRY": mft.MFTEntry}, + class_types={"FILE_NAME_ENTRY": mft.MFTFileName, "MFT_ENTRY": mft.MFTEntry, "ATTRIBUTE": mft.MFTAttribute}, ) # get each of the individual Field Sets - mft_object = symbol_table + constants.BANG + "MFT_ENTRY" - attribute_object = symbol_table + constants.BANG + "ATTRIBUTE" - si_object = symbol_table + constants.BANG + "STANDARD_INFORMATION_ENTRY" - fn_object = symbol_table + constants.BANG + "FILE_NAME_ENTRY" + 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" # Scan the layer for Raw MFT records and parse the fields - for offset, _rule_name, _name, _value in layer.scan( + for offset, _, _, _ in layer.scan( context=self.context, scanner=yarascan.YaraScanner(rules=rules) ): - with contextlib.suppress(exceptions.PagedInvalidAddressException): + with contextlib.suppress(exceptions.InvalidAddressException): mft_record = self.context.object( - mft_object, offset=offset, layer_name=layer.name + self.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( - attribute_object, + self.attribute_object, offset=offset + attr_base_offset, layer_name=layer.name, ) # 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: - vollog.debug(f"Attr Type: {attr.Attr_Header.AttrType.lookup()}") - - # 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: - mft_flag = mft_record.Flags.lookup() - except ValueError: - mft_flag = hex(mft_record.Flags) - - # Standard Information Attribute - if attr.Attr_Header.AttrType.lookup() == "STANDARD_INFORMATION": - attr_data = attr.Attr_Data.cast(si_object) - yield 0, ( - format_hints.Hex(attr_data.vol.offset), - mft_record.get_signature(), - mft_record.RecordNumber, - mft_record.LinkCount, - mft_flag, - renderers.NotApplicableValue(), - attr.Attr_Header.AttrType.lookup(), - conversion.wintime_to_datetime(attr_data.CreationTime), - conversion.wintime_to_datetime(attr_data.ModifiedTime), - conversion.wintime_to_datetime(attr_data.UpdatedTime), - conversion.wintime_to_datetime(attr_data.AccessedTime), - renderers.NotApplicableValue(), - ) - - # File Name Attribute - if attr.Attr_Header.AttrType.lookup() == "FILE_NAME": - 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 - try: - permissions = attr_data.Flags.lookup() - except ValueError: - permissions = hex(attr_data.Flags) - - yield 1, ( - format_hints.Hex(attr_data.vol.offset), - mft_record.get_signature(), - mft_record.RecordNumber, - mft_record.LinkCount, - mft_flag, - permissions, - attr.Attr_Header.AttrType.lookup(), - conversion.wintime_to_datetime(attr_data.CreationTime), - conversion.wintime_to_datetime(attr_data.ModifiedTime), - conversion.wintime_to_datetime(attr_data.UpdatedTime), - conversion.wintime_to_datetime(attr_data.AccessedTime), - file_name, - ) + for record in attr_callback(mft_record, attr): + yield record # If there's no advancement the loop will never end, so break it now if attr.Attr_Header.Length == 0: @@ -135,12 +90,69 @@ 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( - attribute_object, + self.attribute_object, offset=offset + attr_base_offset, layer_name=layer.name, ) + def parse_mft_records(self, mft_record, attr): + # 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: + mft_flag = mft_record.Flags.lookup() + except ValueError: + mft_flag = hex(mft_record.Flags) + + # Standard Information Attribute + if attr.Attr_Header.AttrType.lookup() == "STANDARD_INFORMATION": + attr_data = attr.Attr_Data.cast(self.si_object) + yield 0, ( + format_hints.Hex(attr_data.vol.offset), + mft_record.get_signature(), + mft_record.RecordNumber, + mft_record.LinkCount, + mft_flag, + renderers.NotApplicableValue(), + attr.Attr_Header.AttrType.lookup(), + conversion.wintime_to_datetime(attr_data.CreationTime), + conversion.wintime_to_datetime(attr_data.ModifiedTime), + conversion.wintime_to_datetime(attr_data.UpdatedTime), + conversion.wintime_to_datetime(attr_data.AccessedTime), + renderers.NotApplicableValue(), + ) + + # File Name Attribute + elif attr.Attr_Header.AttrType.lookup() == "FILE_NAME": + attr_data = attr.Attr_Data.cast(self.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 + try: + permissions = attr_data.Flags.lookup() + except ValueError: + permissions = hex(attr_data.Flags) + + yield 1, ( + format_hints.Hex(attr_data.vol.offset), + mft_record.get_signature(), + mft_record.RecordNumber, + mft_record.LinkCount, + mft_flag, + permissions, + attr.Attr_Header.AttrType.lookup(), + conversion.wintime_to_datetime(attr_data.CreationTime), + conversion.wintime_to_datetime(attr_data.ModifiedTime), + conversion.wintime_to_datetime(attr_data.UpdatedTime), + conversion.wintime_to_datetime(attr_data.AccessedTime), + file_name, + ) + + def _generator(self): + for record in self.enumerate_mft_records(self.parse_mft_records): + yield record + def generate_timeline(self): for row in self._generator(): _depth, row_data = row @@ -173,127 +185,98 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): self._generator(), ) - -class ADS(interfaces.plugins.PluginInterface): +class ADS(MFTScan): """Scans for Alternate Data Stream""" _required_framework_version = (2, 0, 0) - @classmethod - def get_requirements(cls): - return [ - 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 __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + + # which DATA attribute should be displayed + self._display_first_data = False + + def _parse_data_record(self, mft_record, attr): + if attr.Attr_Header.NonResidentFlag: + return + + # regular $DATA + if 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, attr): + 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 def _generator(self): - layer = self.context.layers[self.config["primary"]] - - # Yara Rule to scan for MFT Header Signatures - rules = yarascan.YaraScan.process_yara_options( - {"yara_string": "/FILE0|FILE\\*|BAAD/"} - ) - - # Read in the Symbol File - symbol_table = intermed.IntermediateSymbolTable.create( - context=self.context, - config_path=self.config_path, - sub_path="windows", - filename="mft", - class_types={ - "MFT_ENTRY": mft.MFTEntry, - "FILE_NAME_ENTRY": mft.MFTFileName, - "ATTRIBUTE": mft.MFTAttribute, - }, - ) - - # get each of the individual Field Sets - mft_object = symbol_table + constants.BANG + "MFT_ENTRY" - attribute_object = symbol_table + constants.BANG + "ATTRIBUTE" - fn_object = symbol_table + constants.BANG + "FILE_NAME_ENTRY" - - # Scan the layer for Raw MFT records and parse the fields - for offset, _rule_name, _name, _value in layer.scan( - context=self.context, scanner=yarascan.YaraScanner(rules=rules) + for record in self.enumerate_mft_records( + self.parse_data_records ): - with contextlib.suppress(exceptions.PagedInvalidAddressException): - mft_record = self.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( - attribute_object, - offset=offset + attr_base_offset, - layer_name=layer.name, - ) - - # There is no field that has a count of Attributes - # Keep Attempting to read attributes until we get an invalid attr.AttrType - is_ads = False - file_name = renderers.NotAvailableValue - # The First $DATA Attr is the 'principal' file itself not the ADS - while attr.Attr_Header.AttrType.is_valid_choice: - if attr.Attr_Header.AttrType.lookup() == "FILE_NAME": - attr_data = attr.Attr_Data.cast(fn_object) - file_name = attr_data.get_full_name() - if attr.Attr_Header.AttrType.lookup() == "DATA": - if is_ads: - if not attr.Attr_Header.NonResidentFlag: - # Resident files are the most interesting. - if 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: - # Preparing for Disassembly - disasm = interfaces.renderers.BaseAbsentValue - architecture = layer.metadata.get( - "architecture", None - ) - if architecture: - disasm = interfaces.renderers.Disassembly( - content, 0, architecture.lower() - ) - content = format_hints.HexBytes(content) - else: - content = renderers.NotAvailableValue() - disasm = interfaces.renderers.BaseAbsentValue() - - yield 0, ( - format_hints.Hex(attr_data.vol.offset), - mft_record.get_signature(), - mft_record.RecordNumber, - attr.Attr_Header.AttrType.lookup(), - file_name, - ads_name, - content, - disasm, - ) - else: - is_ads = True - - # If there's no advancement the loop will never end, so break it now - if attr.Attr_Header.Length == 0: - break - - # 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( - attribute_object, - offset=offset + attr_base_offset, - layer_name=layer.name, - ) + yield record def run(self): return renderers.TreeGrid( @@ -305,7 +288,31 @@ class ADS(interfaces.plugins.PluginInterface): ("Filename", str), ("ADS Filename", str), ("Hexdump", format_hints.HexBytes), - ("Disasm", interfaces.renderers.Disassembly), ], self._generator(), ) + +class ResidentData(ADS): + """Scans for Alternate Data Stream""" + + _required_framework_version = (2, 0, 0) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + + # which DATA attribute should be displayed + self._display_first_data = True + + def run(self): + return renderers.TreeGrid( + [ + ("Offset", format_hints.Hex), + ("Record Type", str), + ("Record Number", int), + ("MFT Type", str), + ("Filename", str), + ("Hexdump", format_hints.HexBytes), + ], + self._generator(), + ) +