diff --git a/volatility3/framework/plugins/windows/mftscan.py b/volatility3/framework/plugins/windows/mftscan.py index 7f9095c6a..f0be3cf7f 100644 --- a/volatility3/framework/plugins/windows/mftscan.py +++ b/volatility3/framework/plugins/windows/mftscan.py @@ -1,13 +1,11 @@ # This file is Copyright 2022 Volatility Foundation and licensed under the Volatility Software License 1.0 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 # -import contextlib import datetime import logging +from typing import Iterator, NamedTuple, Optional, Tuple, Union -from typing import Generator, Iterable, Dict, Tuple, Callable - -from volatility3.framework import constants, exceptions, interfaces, renderers +from volatility3.framework import constants, exceptions, interfaces, objects, renderers from volatility3.framework.configuration import requirements from volatility3.framework.renderers import conversion, format_hints from volatility3.framework.symbols import intermed @@ -20,9 +18,23 @@ vollog = logging.getLogger(__name__) class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): """Scans for MFT FILE objects present in a particular windows memory image.""" - _required_framework_version = (2, 0, 0) + _required_framework_version = (2, 26, 0) - _version = (2, 0, 1) + _version = (3, 0, 0) + + class MFTScanResult(NamedTuple): + offset: format_hints.Hex + record_type: str + record_number: objects.Integer + link_count: objects.Integer + mft_type: str + permissions: Union[str, interfaces.renderers.BaseAbsentValue] + attribute_type: str + created: Union[interfaces.renderers.BaseAbsentValue, datetime.datetime] + modified: Union[interfaces.renderers.BaseAbsentValue, datetime.datetime] + updated: Union[interfaces.renderers.BaseAbsentValue, datetime.datetime] + accessed: Union[interfaces.renderers.BaseAbsentValue, datetime.datetime] + filename: Union[interfaces.renderers.BaseAbsentValue, objects.String] @classmethod def get_requirements(cls): @@ -51,16 +63,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): context: interfaces.context.ContextInterface, config_path: str, primary_layer_name: str, - attr_callback: Callable[ - [ - Dict[int, Tuple[str, int, int]], - interfaces.objects.ObjectInterface, - interfaces.objects.ObjectInterface, - str, - ], - Generator, - ], - ) -> interfaces.objects.ObjectInterface: + ) -> Iterator[mft.MFTEntry]: try: primary = context.layers[primary_layer_name] except KeyError: @@ -70,14 +73,14 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): return try: - phys_layer = primary.config["memory_layer"] + memory_layer_name = primary.config["memory_layer"] except KeyError: vollog.error( "Unable to obtain memory layer from primary layer. Please file a bug on GitHub about this issue." ) return - layer = context.layers[phys_layer] + layer = context.layers[memory_layer_name] # Yara Rule to scan for MFT Header Signatures rules = yarascan.YaraScan.process_yara_options( @@ -85,7 +88,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): ) # Read in the Symbol File - symbol_table = intermed.IntermediateSymbolTable.create( + symbol_table_name = intermed.IntermediateSymbolTable.create( context=context, config_path=config_path, sub_path="windows", @@ -98,53 +101,25 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): ) # get each of the individual Field Sets - mft_object = symbol_table + constants.BANG + "MFT_ENTRY" - attribute_object = symbol_table + constants.BANG + "ATTRIBUTE" - record_map = {} + mft_object_type_name = symbol_table_name + constants.BANG + "MFT_ENTRY" # Scan the layer for Raw MFT records and parse the fields for offset, _rule_name, _name, _value in layer.scan( context=context, scanner=yarascan.YaraScanner(rules=rules) ): - with contextlib.suppress(exceptions.InvalidAddressException): - 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 = context.object( - attribute_object, - offset=offset + attr_base_offset, - layer_name=layer.name, - ) + mft_record: mft.MFTEntry = context.object( + mft_object_type_name, + offset=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: - yield from attr_callback(record_map, mft_record, attr, symbol_table) - - # 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 = context.object( - attribute_object, - offset=offset + attr_base_offset, - layer_name=layer.name, - ) + yield mft_record @classmethod - def parse_mft_records( - cls, - record_map: Dict[int, Tuple[str, int, int]], - mft_record: interfaces.objects.ObjectInterface, - attr: interfaces.objects.ObjectInterface, - symbol_table_name: str, - ): + def parse_standard_information_records( + cls, mft_record: mft.MFTEntry + ) -> Iterator[Tuple[int, MFTScanResult]]: # 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: @@ -153,163 +128,130 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): mft_flag = hex(mft_record.Flags) # Standard Information Attribute - if attr.Attr_Header.AttrType.lookup() == "STANDARD_INFORMATION": - si_object = ( - symbol_table_name + 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(), - 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(), - ) + try: + # There should only be one STANDARD_INFORMATION attribute, but we + # do this just in case. + for std_information in mft_record.standard_information_entries(): + yield 0, cls.MFTScanResult( + format_hints.Hex(std_information.vol.offset), + str(mft_record.get_signature()), + mft_record.RecordNumber, + mft_record.LinkCount, + mft_flag, + renderers.NotApplicableValue(), + "STANDARD_INFORMATION", + conversion.wintime_to_datetime(std_information.CreationTime), + conversion.wintime_to_datetime(std_information.ModifiedTime), + conversion.wintime_to_datetime(std_information.UpdatedTime), + conversion.wintime_to_datetime(std_information.AccessedTime), + renderers.NotApplicableValue(), + ) + except exceptions.InvalidAddressException: + pass + + @classmethod + def parse_filename_records( + cls, mft_record: mft.MFTEntry + ) -> Iterator[Tuple[int, MFTScanResult]]: + # 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) # File Name Attribute - elif attr.Attr_Header.AttrType.lookup() == "FILE_NAME": - fn_object = symbol_table_name + constants.BANG + "FILE_NAME_ENTRY" + try: + for filename_info in mft_record.filename_entries(): - 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 = filename_info.Flags.lookup() + except ValueError: + permissions = hex(filename_info.Flags) - # 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, - ) - - @classmethod - def parse_data_record( - cls, - 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 - - # we aren't looking ADS when we want the first data record - if return_first_record: - ads_name = renderers.NotApplicableValue() - - # skip records without a name if we want ADS entries - elif attr.Attr_Header.NameLength == 0: - return - - else: - # past the first $DATA record, attempt to get the ADS name - # NotAvailableValue = > 1st Data, but name was not parsable - ads_name = attr.get_resident_filename() or renderers.NotAvailableValue() - - content = attr.get_resident_filecontent() - if content: - content = renderers.LayerData.from_object(content) - else: - content = renderers.NotAvailableValue() - - yield ( - format_hints.Hex(record_map[mft_record.vol.offset][2]), - mft_record.get_signature(), - mft_record.RecordNumber, - attr.Attr_Header.AttrType.lookup(), - record_map[mft_record.vol.offset][0], - ads_name, - content, - ) - - @classmethod - def parse_data_records( - cls, - record_map: Dict[int, Tuple[str, int, int]], - mft_record: interfaces.objects.ObjectInterface, - attr: interfaces.objects.ObjectInterface, - symbol_table_name: str, - 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 - """ - if mft_record.vol.offset not in record_map: - # file name, DATA count, offset - record_map[mft_record.vol.offset] = [renderers.NotAvailableValue(), 0, None] - if attr.Attr_Header.AttrType.lookup() == "FILE_NAME": - fn_object = symbol_table_name + constants.BANG + "FILE_NAME_ENTRY" - attr_data = attr.Attr_Data.cast(fn_object) - rec_name = attr_data.get_full_name() - record_map[mft_record.vol.offset][0] = rec_name - elif attr.Attr_Header.AttrType.lookup() == "DATA": - # first data - record_map[mft_record.vol.offset][2] = attr.Attr_Data.vol.offset - - display_data = False - - # first DATA attribute of this record - if record_map[mft_record.vol.offset][1] == 0: - if return_first_record: - display_data = True - - record_map[mft_record.vol.offset][1] = 1 - - # at the second DATA attribute of this record - elif record_map[mft_record.vol.offset][1] == 1 and not return_first_record: - display_data = True - - if display_data: - yield from cls.parse_data_record( - mft_record, attr, record_map, return_first_record + yield 1, cls.MFTScanResult( + format_hints.Hex(filename_info.vol.offset), + str(mft_record.get_signature()), + mft_record.RecordNumber, + mft_record.LinkCount, + mft_flag, + permissions, + "FILE_NAME", + conversion.wintime_to_datetime(filename_info.CreationTime), + conversion.wintime_to_datetime(filename_info.ModifiedTime), + conversion.wintime_to_datetime(filename_info.UpdatedTime), + conversion.wintime_to_datetime(filename_info.AccessedTime), + filename_info.get_full_name(), ) + except exceptions.InvalidAddressException: + return + + @classmethod + def parse_mft_records( + cls, + context: interfaces.context.ContextInterface, + config_path: str, + primary_layer_name: str, + ) -> Iterator[Tuple[int, MFTScanResult]]: + for mft_record in cls.enumerate_mft_records( + context=context, + config_path=config_path, + primary_layer_name=primary_layer_name, + ): + yield from cls.parse_standard_information_records(mft_record) + yield from cls.parse_filename_records(mft_record) def _generator(self): - yield from self.enumerate_mft_records( + for level, record in self.parse_mft_records( self.context, self.config_path, self.config["primary"], - self.parse_mft_records, - ) + ): + # Convert all `objects.PrimitiveObject` to their simpler Python + # types. This is normally not something we would do, since it's + # lossy and prevents users from getting back to the data source, + # but in this case memory usage is so extreme due to the number of + # records that it becomes necessary. The rich types are still + # exposed through classmethods. + yield level, ( + record.offset, + record.record_type, + int(record.record_number), + int(record.link_count), + record.mft_type, + record.permissions, + record.attribute_type, + record.created, + record.modified, + record.updated, + record.accessed, + ( + str(record.filename) + if isinstance(record.filename, objects.String) + else record.filename + ), + ) def generate_timeline(self): - for row in self._generator(): - _depth, row_data = row + for record in self.enumerate_mft_records( + self.context, self.config_path, self.config["primary"] + ): + fname = record.longest_filename() - # Only Output FN Records - if row_data[6] == "FILE_NAME": - filename = row_data[-1] - description = f"MFT FILE_NAME entry for {filename}" - yield (description, timeliner.TimeLinerType.CREATED, row_data[7]) - yield (description, timeliner.TimeLinerType.MODIFIED, row_data[8]) - yield (description, timeliner.TimeLinerType.CHANGED, row_data[9]) - yield (description, timeliner.TimeLinerType.ACCESSED, row_data[10]) + for _, item in self.parse_standard_information_records(record): + description = f"MFT {item.attribute_type} entry for {fname}" + yield (description, timeliner.TimeLinerType.CREATED, item.created) + yield (description, timeliner.TimeLinerType.MODIFIED, item.modified) + yield (description, timeliner.TimeLinerType.CHANGED, item.updated) + yield (description, timeliner.TimeLinerType.ACCESSED, item.accessed) + + for _, item in self.parse_filename_records(record): + description = f"MFT {item.attribute_type} entry for {item.filename}" + yield (description, timeliner.TimeLinerType.CREATED, item.created) + yield (description, timeliner.TimeLinerType.MODIFIED, item.modified) + yield (description, timeliner.TimeLinerType.CHANGED, item.updated) + yield (description, timeliner.TimeLinerType.ACCESSED, item.accessed) def run(self): return renderers.TreeGrid( @@ -334,15 +276,24 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): class ADS(interfaces.plugins.PluginInterface): """Scans for Alternate Data Stream""" - _required_framework_version = (2, 22, 0) + _required_framework_version = (2, 26, 0) - _version = (1, 0, 2) + _version = (2, 0, 0) + + class ADSResult(NamedTuple): + offset: format_hints.Hex + signature: objects.String + record_number: objects.Integer + attribute_type: str + filename: Union[objects.String, interfaces.renderers.BaseAbsentValue] + stream_name: Union[objects.String, interfaces.renderers.BaseAbsentValue] + content: Union[renderers.LayerData, interfaces.renderers.BaseAbsentValue] @classmethod def get_requirements(cls): return [ requirements.VersionRequirement( - name="MFTScan", component=MFTScan, version=(2, 0, 0) + name="MFTScan", component=MFTScan, version=(3, 0, 0) ), requirements.TranslationLayerRequirement( name="primary", @@ -355,36 +306,61 @@ class ADS(interfaces.plugins.PluginInterface): ] @classmethod - def parse_ads_data_records( - cls, - record_map: Dict[int, Tuple[str, int, int]], - mft_record: interfaces.objects.ObjectInterface, - attr: interfaces.objects.ObjectInterface, - symbol_table_name: str, - ): - return MFTScan.parse_data_records( - record_map, mft_record, attr, symbol_table_name, False - ) + def parse_ads_data_records(cls, mft_record: mft.MFTEntry) -> Iterator[ADSResult]: + for data_attr in mft_record.alternate_data_streams(): + record_filename = ( + mft_record.longest_filename() or renderers.NotAvailableValue() + ) + content_obj = data_attr.get_resident_filecontent() + content = ( + renderers.LayerData.from_object(content_obj) + if content_obj + else renderers.NotAvailableValue() + ) + ads_filename = ( + data_attr.get_resident_filename() or renderers.NotAvailableValue() + ) + + yield cls.ADSResult( + format_hints.Hex(data_attr.Attr_Data.vol.offset), + mft_record.get_signature(), + mft_record.RecordNumber, + data_attr.Attr_Header.AttrType.lookup(), + record_filename, + ads_filename, + content, + ) def _generator(self): - for ( - offset, - rec_type, - rec_num, - attr_type, - file_name, - ads_name, - content, - ) in MFTScan.enumerate_mft_records( + for mft_entry in MFTScan.enumerate_mft_records( self.context, self.config_path, self.config["primary"], - self.parse_ads_data_records, ): - yield ( - 0, - (offset, rec_type, rec_num, attr_type, file_name, ads_name, content), - ) + for record in self.parse_ads_data_records(mft_entry): + # Convert all `objects.PrimitiveObject` to their simpler Python + # types. This is normally not something we would do, since it's + # lossy and prevents users from getting back to the data source, + # but in this case memory usage is so extreme due to the number of + # records that it becomes necessary. The rich types are still + # exposed through classmethods. + yield 0, ( + record.offset, + str(record.signature), + int(record.record_number), + record.attribute_type, + ( + str(record.filename) + if isinstance(record.filename, objects.String) + else record.filename + ), + ( + str(record.stream_name) + if isinstance(record.stream_name, objects.String) + else record.stream_name + ), + record.content, + ) def run(self): return renderers.TreeGrid( @@ -404,15 +380,23 @@ class ADS(interfaces.plugins.PluginInterface): class ResidentData(interfaces.plugins.PluginInterface): """Scans for MFT Records with Resident Data""" - _required_framework_version = (2, 22, 0) + _required_framework_version = (2, 26, 0) - _version = (1, 0, 2) + _version = (2, 0, 0) + + class ResidentDataResult(NamedTuple): + offset: format_hints.Hex + signature: objects.String + record_number: int + attribute_type: str + filename: Union[objects.String, interfaces.renderers.BaseAbsentValue] + content: Union[renderers.LayerData, interfaces.renderers.BaseAbsentValue] @classmethod def get_requirements(cls): return [ requirements.VersionRequirement( - name="MFTScan", component=MFTScan, version=(2, 0, 0) + name="MFTScan", component=MFTScan, version=(3, 0, 0) ), requirements.TranslationLayerRequirement( name="primary", @@ -425,33 +409,59 @@ class ResidentData(interfaces.plugins.PluginInterface): ] @classmethod - def parse_first_data_records( + def parse_resident_data( cls, - record_map: Dict[int, Tuple[str, int, int]], - mft_record: interfaces.objects.ObjectInterface, - attr: interfaces.objects.ObjectInterface, - symbol_table_name: str, - ): - return MFTScan.parse_data_records( - record_map, mft_record, attr, symbol_table_name, True + mft_record: mft.MFTEntry, + ) -> Optional[ResidentDataResult]: + """ + Returns the parsed data from a MFT record + """ + + try: + attr = next(mft_record.resident_data_attributes()) + except StopIteration: + return None + + content = attr.get_resident_filecontent() + if content: + content = renderers.LayerData.from_object(content) + else: + content = renderers.NotAvailableValue() + + # Choose the longest of the two, since it often includes a DOS 8.3 name + filename = mft_record.longest_filename() or renderers.NotAvailableValue() + + return cls.ResidentDataResult( + format_hints.Hex(attr.Attr_Data.vol.offset), + mft_record.get_signature(), + mft_record.RecordNumber, + attr.Attr_Header.AttrType.lookup(), + filename, + content, ) def _generator(self): - for ( - offset, - rec_type, - rec_num, - attr_type, - file_name, - _, - content, - ) in MFTScan.enumerate_mft_records( + for mft_record in MFTScan.enumerate_mft_records( self.context, self.config_path, self.config["primary"], - self.parse_first_data_records, ): - yield (0, (offset, rec_type, rec_num, attr_type, file_name, content)) + resident_data_entry = self.parse_resident_data(mft_record) + if resident_data_entry: + # Convert all `objects.PrimitiveObject` to their simpler Python + # types. This is normally not something we would do, since it's + # lossy and prevents users from getting back to the data source, + # but in this case memory usage is so extreme due to the number of + # records that it becomes necessary. The rich types are still + # exposed through classmethods. + yield 0, ( + resident_data_entry.offset, + str(resident_data_entry.signature), + int(resident_data_entry.record_number), + resident_data_entry.attribute_type, + str(resident_data_entry.filename), + resident_data_entry.content, + ) def run(self): return renderers.TreeGrid( diff --git a/volatility3/framework/symbols/windows/extensions/mft.py b/volatility3/framework/symbols/windows/extensions/mft.py index ebba882c0..bf20c1ffc 100644 --- a/volatility3/framework/symbols/windows/extensions/mft.py +++ b/volatility3/framework/symbols/windows/extensions/mft.py @@ -2,23 +2,170 @@ # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 # -from typing import Optional +import logging +from typing import Dict, Iterator, List, Optional, Tuple -from volatility3.framework import objects, constants, exceptions +from volatility3.framework import constants, exceptions, interfaces, objects + +vollog = logging.getLogger(__name__) class MFTEntry(objects.StructType): """This represents the base MFT Record""" - def get_signature(self) -> str: + def __init__( + self, + context: interfaces.context.ContextInterface, + type_name: str, + object_info: interfaces.objects.ObjectInformation, + size: int, + members: Dict[str, Tuple[int, interfaces.objects.Template]], + ) -> None: + super().__init__(context, type_name, object_info, size, members) + + self._attrs_loaded = False + self._attrs: List[MFTAttribute] = [] + + @property + def symbol_table_name(self) -> str: + return self.vol.type_name.split(constants.BANG)[0] + + def get_signature(self) -> objects.String: signature = self.Signature.cast("string", max_length=4, encoding="latin-1") return signature + @property + def attributes(self) -> Iterator["MFTAttribute"]: + """ + Lazily evaluate and yield attributes, caching them in an internal list + for re-retrieval. + """ + if not self._attrs_loaded: + self._attrs = list(self._attributes()) + self._attrs_loaded = True + + yield from self._attrs + + def longest_filename(self) -> Optional[objects.String]: + names = [name.get_full_name() for name in self.filename_entries()] + if not names: + return None + + return max(names, key=lambda x: len(str(x))) + + def _attributes(self) -> Iterator["MFTAttribute"]: + + # We will update this on each pass in the next loop and use it as the new offset. + attr_base_offset = self.FirstAttrOffset + attribute_object_type_name = ( + self.symbol_table_name + constants.BANG + "ATTRIBUTE" + ) + + attr: MFTAttribute = self._context.object( + attribute_object_type_name, + offset=self.vol.offset + attr_base_offset, + layer_name=self.vol.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 + try: + while attr.Attr_Header.AttrType.is_valid_choice: + yield attr + + # 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: MFTAttribute = self._context.object( + attribute_object_type_name, + offset=self.vol.offset + attr_base_offset, + layer_name=self.vol.layer_name, + ) + except exceptions.InvalidAddressException as e: + vollog.debug( + f"Failed to read attribute at {attr.vol.offset:#x}: {e.__class__.__name__}" + ) + return + + def standard_information_entries( + self, + ) -> Iterator[objects.StructType]: + """ + Yields a STANDARD_INFORMATION struct for each of the + STANDARD_INFORMATION attributes in this MFT record (although there + should only be one per record). + """ + for attr in self.attributes: + attr_type = attr.Attr_Header.AttrType.lookup() + if attr_type != "STANDARD_INFORMATION": + continue + + si_object = ( + self.symbol_table_name + constants.BANG + "STANDARD_INFORMATION_ENTRY" + ) + + yield attr.Attr_Data.cast(si_object) + + def filename_entries(self) -> Iterator["MFTFileName"]: + """ + Yields an MFT Filename for each of the FILE_NAME attributes contained + in this MFT record. There are often two - one for the long filename, + and the other with the DOS 8.3 short name. + """ + for attr in self.attributes: + try: + attr_type = attr.Attr_Header.AttrType.lookup() + if attr_type != "FILE_NAME": + continue + + fn_object = self.symbol_table_name + constants.BANG + "FILE_NAME_ENTRY" + attr_data = attr.Attr_Data.cast(fn_object) + except exceptions.InvalidAddressException as e: + vollog.debug( + f"Failed to read attr at {attr.vol.offset:#x}: {e.__class__.__name__}" + ) + continue + yield attr_data + + def _data_attributes(self): + for attr in self.attributes: + if not ( + attr.Attr_Header.AttrType.lookup() == "DATA" + and attr.Attr_Header.NonResidentFlag == 0 + ): + continue + + yield attr + + def resident_data_attributes(self) -> Iterator["MFTAttribute"]: + """ + Yields all MFT attributes that contain resident data for the primary + stream. + """ + for attr in self._data_attributes(): + if attr.Attr_Header.NameLength == 0: + yield attr + + def alternate_data_streams(self) -> Iterator["MFTAttribute"]: + """ + Yields all MFT attributes that contain alternate data streams (ADS). + """ + for attr in self._data_attributes(): + if attr.Attr_Header.NameLength != 0: + yield attr + class MFTFileName(objects.StructType): """This represents an MFT $FILE_NAME Attribute""" - def get_full_name(self) -> str: + def get_full_name(self) -> objects.String: + """ + Returns the UTF-16 decoded filename. + """ output = self.Name.cast( "string", encoding="utf16", max_length=self.NameLength * 2, errors="replace" ) @@ -28,7 +175,10 @@ class MFTFileName(objects.StructType): class MFTAttribute(objects.StructType): """This represents an MFT ATTRIBUTE""" - def get_resident_filename(self) -> Optional[str]: + def get_resident_filename(self) -> Optional[objects.String]: + """ + Returns the resident filename (typically for an Alternate Data Stream (ADS)). + """ # 4MB chosen as cutoff instead of 4KB to allow for recovery from format /L created file systems # Length as 512 as its 256*2, which is the maximum size for an entire file path, so this is even generous if ( @@ -48,10 +198,17 @@ class MFTAttribute(objects.StructType): encoding="utf16", ) return name - except exceptions.InvalidAddressException: + except exceptions.InvalidAddressException as e: + vollog.debug( + f"Failed to get resident file content due to {e.__class__.__name__}" + ) return None - def get_resident_filecontent(self) -> Optional[bytes]: + def get_resident_filecontent(self) -> Optional[objects.Bytes]: + """ + Returns the file content that is resident within this MFT attribute, + for either the primary or an alternate data stream. + """ # smear observed in mass testing of samples # 4MB chosen as cutoff instead of 4KB to allow for recovery from format /L created file systems if ( @@ -70,5 +227,8 @@ class MFTAttribute(objects.StructType): length=self.Attr_Header.ContentLength, ) return bytesobj - except exceptions.InvalidAddressException: + except exceptions.InvalidAddressException as e: + vollog.debug( + f"Failed to get resident file content due to {e.__class__.__name__}" + ) return None