mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-06 17:57:38 +02:00
Merge pull request #1758 from volatilityfoundation/bugfix/mftscan_memory_usage
Windows MFTScan: Memory usage performance fixes
This commit is contained in:
ikelos
@@ -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(
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user