From b9e0670e90de541c492653e1bd4f0da129f4a088 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Tue, 28 Jul 2020 23:01:54 +0100 Subject: [PATCH] PDBUtility: Refactor this to a more central place --- volatility/framework/automagic/pdbscan.py | 281 +------------------ volatility/framework/symbols/windows/pdb.py | 288 ++++++++++++++++++++ 2 files changed, 291 insertions(+), 278 deletions(-) create mode 100644 volatility/framework/symbols/windows/pdb.py diff --git a/volatility/framework/automagic/pdbscan.py b/volatility/framework/automagic/pdbscan.py index df083db85..baed4f562 100644 --- a/volatility/framework/automagic/pdbscan.py +++ b/volatility/framework/automagic/pdbscan.py @@ -7,23 +7,16 @@ from loaded PE files. This module contains a standalone scanner, and also a :class:`~volatility.framework.interfaces.layers.ScannerInterface` based scanner for use within the framework by calling :func:`~volatility.framework.interfaces.layers.DataLayerInterface.scan`. """ -import binascii -import json import logging -import lzma import math import os -import struct -from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Tuple, Union -from urllib import request +from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Union -from volatility import symbols from volatility.framework import constants, exceptions, interfaces, layers from volatility.framework.configuration import requirements -from volatility.framework.configuration.requirements import SymbolTableRequirement from volatility.framework.layers import intel, scanners -from volatility.framework.symbols import intermed, native -from volatility.framework.symbols.windows import pdbconv +from volatility.framework.symbols import native +from volatility.framework.symbols.windows.pdb import PDBUtility if __name__ == "__main__": import sys @@ -36,274 +29,6 @@ ValidKernelsType = Dict[str, Tuple[int, Dict[str, Optional[Union[bytes, str, int KernelsType = Iterable[Dict[str, Any]] -class PdbSignatureScanner(interfaces.layers.ScannerInterface): - """A :class:`~volatility.framework.interfaces.layers.ScannerInterface` - based scanner use to identify Windows PDB records. - - Args: - pdb_names: A list of bytestrings, used to match pdb signatures against the pdb names within the records. - - .. note:: The pdb_names must be a list of byte strings, unicode strs will not match against the data scanned - """ - overlap = 0x4000 - """The size of overlap needed for the signature to ensure data cannot hide between two scanned chunks""" - thread_safe = True - """Determines whether the scanner accesses global variables in a thread safe manner (for use with :mod:`multiprocessing`)""" - - _RSDS_format = struct.Struct("<16BI") - - def __init__(self, pdb_names: List[bytes]) -> None: - super().__init__() - self._pdb_names = pdb_names - - def __call__(self, data: bytes, data_offset: int) -> Generator[Tuple[str, Any, bytes, int], None, None]: - sig = data.find(b"RSDS") - while sig >= 0: - null = data.find(b'\0', sig + 4 + self._RSDS_format.size) - if null > -1: - if (null - sig - self._RSDS_format.size) <= 100: - name_offset = sig + 4 + self._RSDS_format.size - pdb_name = data[name_offset:null] - if pdb_name in self._pdb_names: - - ## this ordering is intentional due to mixed endianness in the GUID - (g3, g2, g1, g0, g5, g4, g7, g6, g8, g9, ga, gb, gc, gd, ge, gf, a) = \ - self._RSDS_format.unpack(data[sig + 4:name_offset]) - - guid = (16 * '{:02X}').format(g0, g1, g2, g3, g4, g5, g6, g7, g8, g9, ga, gb, gc, gd, ge, gf) - if sig < self.chunk_size: - yield (guid, a, pdb_name, data_offset + sig) - sig = data.find(b"RSDS", sig + 1) - - -class PDBUtility: - """Class to handle and manage all getting symbols based on MZ header""" - - @classmethod - def symbol_table_from_offset(cls, - context: interfaces.context.ContextInterface, - layer_name: str, - offset: int, - symbol_table_class: str, - config_path: str = None, - progress_callback: constants.ProgressCallback = None) -> str: - """Produces the name of a symbol table loaded from the offset for an MZ header - - Args: - context: The context on which to operate - layer_name: The name of the (contiguous) layer within the context that contains the MZ file - offset: The offset in the layer at which the MZ file begins - symbol_table_class: The type of symbol class to construct the SymbolTable - config_path: New path for the produced symbol table configuration with the config tree - progress_callback: Callable called to update ongoing progress - - Returns: - None if no pdb information can be determined, else returned the name of the loaded symbols for the MZ - """ - guid, age, pdb_name = cls.get_guid_from_mz(context, layer_name, offset) - if config_path is None: - config_path = interfaces.configuration.path_join('pdbutility', pdb_name.strip('\x00').replace('.', '_')) - - return cls.load_windows_symbol_table(context, guid, age, pdb_name, symbol_table_class, config_path, - progress_callback) - - @classmethod - def load_windows_symbol_table(cls, - context: interfaces.context.ContextInterface, - guid: str, - age: int, - pdb_name: str, - symbol_table_class: str, - config_path: str = 'pdbutility', - progress_callback: constants.ProgressCallback = None): - """Loads (downlading if necessary) a windows symbol table""" - - filter_string = os.path.join(pdb_name.strip('\x00'), guid.upper() + "-" + str(age)) - - isf_path = False - # Take the first result of search for the intermediate file - for value in intermed.IntermediateSymbolTable.file_symbol_url("windows", filter_string): - isf_path = value - break - else: - # If none are found, attempt to download the pdb, convert it and try again - cls.download_pdb_isf(context, guid.upper(), age, pdb_name, progress_callback) - # Try again - for value in intermed.IntermediateSymbolTable.file_symbol_url("windows", filter_string): - isf_path = value - break - - if not isf_path: - vollog.debug("Required symbol library path not found: {}".format(filter_string)) - return None - - vollog.debug("Using symbol library: {}".format(filter_string)) - - # Set the discovered options - join = interfaces.configuration.path_join - context.config[join(config_path, "class")] = symbol_table_class - context.config[join(config_path, "isf_url")] = isf_path - parent_config_path = interfaces.configuration.parent_path(config_path) - requirement_name = interfaces.configuration.path_head(config_path) - - # Construct the appropriate symbol table - requirement = SymbolTableRequirement(name = requirement_name, description = "PDBUtility generated symbol table") - requirement.construct(context, parent_config_path) - return context.config[config_path] - - @classmethod - def get_guid_from_mz(cls, context: interfaces.context.ContextInterface, layer_name: str, - offset: int) -> Optional[Tuple[str, int, str]]: - """Takes the offset to an MZ header, locates any available pdb headers, and extracts the guid, age and pdb_name from them - - Args: - context: The context on which to operate - layer_name: The name of the (contiguous) layer within the context that contains the MZ file - offset: The offset in the layer at which the MZ file begins - - Returns: - A tuple of the guid, age and pdb_name, or None if no PDB record can be found - """ - try: - import pefile - except ImportError: - vollog.error("Get_guid_from_mz requires the following python module: pefile") - return None - - layer = context.layers[layer_name] - mz_sig = layer.read(offset, 2) - - # Check it is actually the MZ header - if mz_sig != b"MZ": - return None - - nt_header_start = ord(layer.read(offset + 0x3C, 1)) - optional_header_size = struct.unpack(' None: - """Attempts to download the PDB file, convert it to an ISF file and - save it to one of the symbol locations.""" - # Check for writability - filter_string = os.path.join(pdb_name, guid + "-" + str(age)) - for path in symbols.__path__: - - # Store any temporary files created by downloading PDB files - tmp_files = [] - potential_output_filename = os.path.join(path, "windows", filter_string + ".json.xz") - data_written = False - try: - os.makedirs(os.path.dirname(potential_output_filename), exist_ok = True) - with lzma.open(potential_output_filename, "w") as of: - # Once we haven't thrown an error, do the computation - filename = pdbconv.PdbRetreiver().retreive_pdb(guid + str(age), - file_name = pdb_name, - progress_callback = progress_callback) - if filename: - tmp_files.append(filename) - location = "file:" + request.pathname2url(tmp_files[-1]) - json_output = pdbconv.PdbReader(context, location, progress_callback).get_json() - of.write(bytes(json.dumps(json_output, indent = 2, sort_keys = True), 'utf-8')) - # After we've successfully written it out, record the fact so we don't clear it out - data_written = True - else: - vollog.warning("Symbol file could not be found on remote server" + (" " * 100)) - break - except PermissionError: - vollog.warning("Cannot write necessary symbol file, please check permissions on {}".format( - potential_output_filename)) - continue - finally: - # If something else failed, removed the symbol file so we don't pick it up in the future - if not data_written and os.path.exists(potential_output_filename): - os.remove(potential_output_filename) - # Clear out all the temporary file if we constructed one - for filename in tmp_files: - try: - os.remove(filename) - except PermissionError: - vollog.warning("Temporary file could not be removed: {}".format(filename)) - else: - vollog.warning("Cannot write downloaded symbols, please add the appropriate symbols" - " or add/modify a symbols directory that is writable") - - @classmethod - def pdbname_scan(cls, - ctx: interfaces.context.ContextInterface, - layer_name: str, - page_size: int, - pdb_names: List[bytes], - progress_callback: constants.ProgressCallback = None, - start: Optional[int] = None, - end: Optional[int] = None) -> Generator[Dict[str, Optional[Union[bytes, str, int]]], None, None]: - """Scans through `layer_name` at `ctx` looking for RSDS headers that - indicate one of four common pdb kernel names (as listed in - `self.pdb_names`) and returns the tuple (GUID, age, pdb_name, - signature_offset, mz_offset) - - .. note:: This is automagical and therefore not guaranteed to provide correct results. - - The UI should always provide the user an opportunity to specify the - appropriate types and PDB values themselves - """ - min_pfn = 0 - - if start is None: - start = ctx.layers[layer_name].minimum_address - if end is None: - end = ctx.layers[layer_name].maximum_address - - for (GUID, age, pdb_name, - signature_offset) in ctx.layers[layer_name].scan(ctx, - PdbSignatureScanner(pdb_names), - progress_callback = progress_callback, - sections = [(start, end - start)]): - mz_offset = None - sig_pfn = signature_offset // page_size - - for i in range(sig_pfn, min_pfn, -1): - if not ctx.layers[layer_name].is_valid(i * page_size, 2): - break - - data = ctx.layers[layer_name].read(i * page_size, 2) - if data == b'MZ': - mz_offset = i * page_size - break - min_pfn = sig_pfn - - yield { - 'GUID': GUID, - 'age': age, - 'pdb_name': str(pdb_name, "utf-8"), - 'signature_offset': signature_offset, - 'mz_offset': mz_offset - } - - class KernelPDBScanner(interfaces.automagic.AutomagicInterface): """Windows symbol loader based on PDB signatures. diff --git a/volatility/framework/symbols/windows/pdb.py b/volatility/framework/symbols/windows/pdb.py new file mode 100644 index 000000000..a66ab4c8c --- /dev/null +++ b/volatility/framework/symbols/windows/pdb.py @@ -0,0 +1,288 @@ +# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# + +import binascii +import json +import logging +import lzma +import os +import struct +from typing import Any, Dict, Generator, List, Optional, Tuple, Union +from urllib import request + +from volatility import symbols +from volatility.framework import constants, interfaces +from volatility.framework.configuration.requirements import SymbolTableRequirement +from volatility.framework.symbols import intermed +from volatility.framework.symbols.windows import pdbconv + +vollog = logging.getLogger(__name__) + + +class PDBUtility: + """Class to handle and manage all getting symbols based on MZ header""" + + @classmethod + def symbol_table_from_offset(cls, + context: interfaces.context.ContextInterface, + layer_name: str, + offset: int, + symbol_table_class: str, + config_path: str = None, + progress_callback: constants.ProgressCallback = None) -> str: + """Produces the name of a symbol table loaded from the offset for an MZ header + + Args: + context: The context on which to operate + layer_name: The name of the (contiguous) layer within the context that contains the MZ file + offset: The offset in the layer at which the MZ file begins + symbol_table_class: The type of symbol class to construct the SymbolTable + config_path: New path for the produced symbol table configuration with the config tree + progress_callback: Callable called to update ongoing progress + + Returns: + None if no pdb information can be determined, else returned the name of the loaded symbols for the MZ + """ + guid, age, pdb_name = cls.get_guid_from_mz(context, layer_name, offset) + if config_path is None: + config_path = interfaces.configuration.path_join('pdbutility', pdb_name.strip('\x00').replace('.', '_')) + + return cls.load_windows_symbol_table(context, guid, age, pdb_name, symbol_table_class, config_path, + progress_callback) + + @classmethod + def load_windows_symbol_table(cls, + context: interfaces.context.ContextInterface, + guid: str, + age: int, + pdb_name: str, + symbol_table_class: str, + config_path: str = 'pdbutility', + progress_callback: constants.ProgressCallback = None): + """Loads (downlading if necessary) a windows symbol table""" + + filter_string = os.path.join(pdb_name.strip('\x00'), guid.upper() + "-" + str(age)) + + isf_path = False + # Take the first result of search for the intermediate file + for value in intermed.IntermediateSymbolTable.file_symbol_url("windows", filter_string): + isf_path = value + break + else: + # If none are found, attempt to download the pdb, convert it and try again + cls.download_pdb_isf(context, guid.upper(), age, pdb_name, progress_callback) + # Try again + for value in intermed.IntermediateSymbolTable.file_symbol_url("windows", filter_string): + isf_path = value + break + + if not isf_path: + vollog.debug("Required symbol library path not found: {}".format(filter_string)) + return None + + vollog.debug("Using symbol library: {}".format(filter_string)) + + # Set the discovered options + join = interfaces.configuration.path_join + context.config[join(config_path, "class")] = symbol_table_class + context.config[join(config_path, "isf_url")] = isf_path + parent_config_path = interfaces.configuration.parent_path(config_path) + requirement_name = interfaces.configuration.path_head(config_path) + + # Construct the appropriate symbol table + requirement = SymbolTableRequirement(name = requirement_name, description = "PDBUtility generated symbol table") + requirement.construct(context, parent_config_path) + return context.config[config_path] + + @classmethod + def get_guid_from_mz(cls, context: interfaces.context.ContextInterface, layer_name: str, + offset: int) -> Optional[Tuple[str, int, str]]: + """Takes the offset to an MZ header, locates any available pdb headers, and extracts the guid, age and pdb_name from them + + Args: + context: The context on which to operate + layer_name: The name of the (contiguous) layer within the context that contains the MZ file + offset: The offset in the layer at which the MZ file begins + + Returns: + A tuple of the guid, age and pdb_name, or None if no PDB record can be found + """ + try: + import pefile + except ImportError: + vollog.error("Get_guid_from_mz requires the following python module: pefile") + return None + + layer = context.layers[layer_name] + mz_sig = layer.read(offset, 2) + + # Check it is actually the MZ header + if mz_sig != b"MZ": + return None + + nt_header_start = ord(layer.read(offset + 0x3C, 1)) + optional_header_size = struct.unpack(' None: + """Attempts to download the PDB file, convert it to an ISF file and + save it to one of the symbol locations.""" + # Check for writability + filter_string = os.path.join(pdb_name, guid + "-" + str(age)) + for path in symbols.__path__: + + # Store any temporary files created by downloading PDB files + tmp_files = [] + potential_output_filename = os.path.join(path, "windows", filter_string + ".json.xz") + data_written = False + try: + os.makedirs(os.path.dirname(potential_output_filename), exist_ok = True) + with lzma.open(potential_output_filename, "w") as of: + # Once we haven't thrown an error, do the computation + filename = pdbconv.PdbRetreiver().retreive_pdb(guid + str(age), + file_name = pdb_name, + progress_callback = progress_callback) + if filename: + tmp_files.append(filename) + location = "file:" + request.pathname2url(tmp_files[-1]) + json_output = pdbconv.PdbReader(context, location, progress_callback).get_json() + of.write(bytes(json.dumps(json_output, indent = 2, sort_keys = True), 'utf-8')) + # After we've successfully written it out, record the fact so we don't clear it out + data_written = True + else: + vollog.warning("Symbol file could not be found on remote server" + (" " * 100)) + break + except PermissionError: + vollog.warning("Cannot write necessary symbol file, please check permissions on {}".format( + potential_output_filename)) + continue + finally: + # If something else failed, removed the symbol file so we don't pick it up in the future + if not data_written and os.path.exists(potential_output_filename): + os.remove(potential_output_filename) + # Clear out all the temporary file if we constructed one + for filename in tmp_files: + try: + os.remove(filename) + except PermissionError: + vollog.warning("Temporary file could not be removed: {}".format(filename)) + else: + vollog.warning("Cannot write downloaded symbols, please add the appropriate symbols" + " or add/modify a symbols directory that is writable") + + @classmethod + def pdbname_scan(cls, + ctx: interfaces.context.ContextInterface, + layer_name: str, + page_size: int, + pdb_names: List[bytes], + progress_callback: constants.ProgressCallback = None, + start: Optional[int] = None, + end: Optional[int] = None) -> Generator[Dict[str, Optional[Union[bytes, str, int]]], None, None]: + """Scans through `layer_name` at `ctx` looking for RSDS headers that + indicate one of four common pdb kernel names (as listed in + `self.pdb_names`) and returns the tuple (GUID, age, pdb_name, + signature_offset, mz_offset) + + .. note:: This is automagical and therefore not guaranteed to provide correct results. + + The UI should always provide the user an opportunity to specify the + appropriate types and PDB values themselves + """ + min_pfn = 0 + + if start is None: + start = ctx.layers[layer_name].minimum_address + if end is None: + end = ctx.layers[layer_name].maximum_address + + for (GUID, age, pdb_name, + signature_offset) in ctx.layers[layer_name].scan(ctx, + PdbSignatureScanner(pdb_names), + progress_callback = progress_callback, + sections = [(start, end - start)]): + mz_offset = None + sig_pfn = signature_offset // page_size + + for i in range(sig_pfn, min_pfn, -1): + if not ctx.layers[layer_name].is_valid(i * page_size, 2): + break + + data = ctx.layers[layer_name].read(i * page_size, 2) + if data == b'MZ': + mz_offset = i * page_size + break + min_pfn = sig_pfn + + yield { + 'GUID': GUID, + 'age': age, + 'pdb_name': str(pdb_name, "utf-8"), + 'signature_offset': signature_offset, + 'mz_offset': mz_offset + } + + +class PdbSignatureScanner(interfaces.layers.ScannerInterface): + """A :class:`~volatility.framework.interfaces.layers.ScannerInterface` + based scanner use to identify Windows PDB records. + + Args: + pdb_names: A list of bytestrings, used to match pdb signatures against the pdb names within the records. + + .. note:: The pdb_names must be a list of byte strings, unicode strs will not match against the data scanned + """ + overlap = 0x4000 + """The size of overlap needed for the signature to ensure data cannot hide between two scanned chunks""" + thread_safe = True + """Determines whether the scanner accesses global variables in a thread safe manner (for use with :mod:`multiprocessing`)""" + + _RSDS_format = struct.Struct("<16BI") + + def __init__(self, pdb_names: List[bytes]) -> None: + super().__init__() + self._pdb_names = pdb_names + + def __call__(self, data: bytes, data_offset: int) -> Generator[Tuple[str, Any, bytes, int], None, None]: + sig = data.find(b"RSDS") + while sig >= 0: + null = data.find(b'\0', sig + 4 + self._RSDS_format.size) + if null > -1: + if (null - sig - self._RSDS_format.size) <= 100: + name_offset = sig + 4 + self._RSDS_format.size + pdb_name = data[name_offset:null] + if pdb_name in self._pdb_names: + + ## this ordering is intentional due to mixed endianness in the GUID + (g3, g2, g1, g0, g5, g4, g7, g6, g8, g9, ga, gb, gc, gd, ge, gf, a) = \ + self._RSDS_format.unpack(data[sig + 4:name_offset]) + + guid = (16 * '{:02X}').format(g0, g1, g2, g3, g4, g5, g6, g7, g8, g9, ga, gb, gc, gd, ge, gf) + if sig < self.chunk_size: + yield (guid, a, pdb_name, data_offset + sig) + sig = data.find(b"RSDS", sig + 1)