mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-08-25 07:32:23 +02:00
226 lines
10 KiB
Python
226 lines
10 KiB
Python
# This file was contributed to the Volatility Framework Version 3.
|
|
# Copyright (C) 2018 Volatility Foundation.
|
|
#
|
|
# THE LICENSED WORK IS PROVIDED UNDER THE TERMS OF THE Volatility Contributors
|
|
# Public License V1.0("LICENSE") AS FIRST COMPLETED BY: Volatility Foundation,
|
|
# Inc. ANY USE, PUBLIC DISPLAY, PUBLIC PERFORMANCE, REPRODUCTION OR DISTRIBUTION
|
|
# OF, OR PREPARATION OF SUBSEQUENT WORKS, DERIVATIVE WORKS OR DERIVED WORKS BASED
|
|
# ON, THE LICENSED WORK CONSTITUTES RECIPIENT'S ACCEPTANCE OF THIS LICENSE AND ITS
|
|
# TERMS, WHETHER OR NOT SUCH RECIPIENT READS THE TERMS OF THE LICENSE. "LICENSED
|
|
# WORK,” “RECIPIENT" AND “DISTRIBUTOR" ARE DEFINED IN THE LICENSE. A COPY OF THE
|
|
# LICENSE IS LOCATED IN THE TEXT FILE ENTITLED "LICENSE.txt" ACCOMPANYING THE
|
|
# CONTENTS OF THIS FILE. IF A COPY OF THE LICENSE DOES NOT ACCOMPANY THIS FILE, A
|
|
# COPY OF THE LICENSE MAY ALSO BE OBTAINED AT THE FOLLOWING WEB SITE:
|
|
# https://www.volatilityfoundation.org/license/vcpl_v1.0
|
|
#
|
|
# Software distributed under the License is distributed on an "AS IS" basis,
|
|
# WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for the
|
|
# specific language governing rights and limitations under the License.
|
|
#
|
|
import math
|
|
from typing import Optional, Dict, Any, List, Iterable, Tuple
|
|
|
|
from volatility.framework import interfaces, constants
|
|
from volatility.framework.configuration import requirements
|
|
from volatility.framework.objects import utility
|
|
from volatility.framework.symbols import intermed
|
|
|
|
|
|
class PdbMSF(interfaces.layers.TranslationLayerInterface):
|
|
_headers = {
|
|
"MSF_HDR": "Microsoft C/C++ program database 2.00\r\n\x1a\x4a\x47",
|
|
"BIG_MSF_HDR": "Microsoft C/C++ MSF 7.00\r\n\x1a\x44\x53",
|
|
}
|
|
|
|
def __init__(self,
|
|
context: 'interfaces.context.ContextInterface',
|
|
config_path: str,
|
|
name: str,
|
|
metadata: Optional[Dict[str, Any]] = None) -> None:
|
|
super().__init__(context, config_path, name, metadata)
|
|
self._base_layer = self.config["base_layer"]
|
|
|
|
self._pdb_symbol_table = intermed.IntermediateSymbolTable.create(context, self._config_path, 'windows', 'mspdb')
|
|
response = self._check_header()
|
|
if response is None:
|
|
raise ValueError("Could not find a suitable header")
|
|
self._version, self._header = response
|
|
self._streams = {} # type: Dict[int, str]
|
|
|
|
@property
|
|
def pdb_symbol_table(self) -> str:
|
|
return self._pdb_symbol_table
|
|
|
|
def read_streams(self):
|
|
# Shortcut in case they've already been read
|
|
if self._streams:
|
|
return
|
|
|
|
# Recover the root table, by recovering the root table index table...
|
|
module = self.context.module(self.pdb_symbol_table, self._base_layer, offset = 0)
|
|
entry_size = module.get_type("unsigned long").size
|
|
|
|
root_table_num_pages = math.ceil(self._header.StreamInfo.StreamInfoSize / self._header.PageSize)
|
|
root_index_size = math.ceil((root_table_num_pages * entry_size) / self._header.PageSize)
|
|
root_index = module.object(
|
|
type_name = "array",
|
|
layer_name = self._base_layer,
|
|
offset = self._header.vol.size,
|
|
count = root_index_size,
|
|
subtype = module.get_type("unsigned long"))
|
|
root_index_layer_name = self.create_stream_from_pages("root_index", self._header.StreamInfo.StreamInfoSize,
|
|
[x for x in root_index])
|
|
|
|
module = self.context.module(self.pdb_symbol_table, root_index_layer_name, offset = 0)
|
|
root_pages = module.object(
|
|
type_name = "array", offset = 0, count = root_table_num_pages, subtype = module.get_type("unsigned long"))
|
|
root_layer_name = self.create_stream_from_pages("root", self._header.StreamInfo.StreamInfoSize,
|
|
[x for x in root_pages])
|
|
|
|
module = self.context.module(self.pdb_symbol_table, root_layer_name, offset = 0)
|
|
num_streams = module.object(type_name = "unsigned long", offset = 0)
|
|
stream_sizes = module.object(
|
|
type_name = "array", offset = entry_size, count = num_streams, subtype = module.get_type("unsigned long"))
|
|
|
|
current_offset = (num_streams + 1) * entry_size
|
|
|
|
for stream in range(num_streams):
|
|
list_size = math.ceil(stream_sizes[stream] / self.page_size)
|
|
if list_size:
|
|
stream_page_list = module.object(
|
|
type_name = "array",
|
|
offset = current_offset,
|
|
count = list_size,
|
|
subtype = module.get_type("unsigned long"))
|
|
current_offset += (list_size * entry_size)
|
|
self._streams[stream] = self.create_stream_from_pages("stream" + str(stream), stream_sizes[stream],
|
|
[x for x in stream_page_list])
|
|
else:
|
|
self._streams[stream] = None
|
|
|
|
def create_stream_from_pages(self, stream_name: str, maximum_size: int, pages: List[int]) -> str:
|
|
# Construct a root layer based on a number of pages
|
|
layer_name = self.name + "_" + stream_name
|
|
path_join = interfaces.configuration.path_join
|
|
config_path = path_join(self.config_path, stream_name)
|
|
self.context.config[path_join(config_path, 'base_layer')] = self.name
|
|
self.context.config[path_join(config_path, 'pages')] = pages
|
|
self.context.config[path_join(config_path, 'maximum_size')] = maximum_size
|
|
layer = PdbMSFStream(self.context, config_path, layer_name)
|
|
self.context.layers.add_layer(layer)
|
|
return layer_name
|
|
|
|
def _check_header(self) -> Optional[Tuple[str, interfaces.objects.ObjectInterface]]:
|
|
"""Verifies the header of the PDB file and returns the version of the file"""
|
|
for header in self._headers:
|
|
header_type = self.pdb_symbol_table + constants.BANG + header
|
|
current_header = self.context.object(header_type, self._base_layer, 0)
|
|
if utility.array_to_string(current_header.Magic) == self._headers[header]:
|
|
if not (current_header.PageSize < 0x100 or current_header.PageSize > (128 * 0x10000)):
|
|
return header, current_header
|
|
return None
|
|
|
|
@property
|
|
def page_size(self):
|
|
return self._header.PageSize
|
|
|
|
@property
|
|
def dependencies(self) -> List[str]:
|
|
"""Returns a list of the lower layers that this layer is dependent upon"""
|
|
return [self._base_layer]
|
|
|
|
@classmethod
|
|
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
|
return [requirements.TranslationLayerRequirement(name = 'base_layer', optional = False)]
|
|
|
|
@property
|
|
def maximum_address(self) -> int:
|
|
return self.context.layers[self._base_layer].maximum_address
|
|
|
|
@property
|
|
def minimum_address(self) -> int:
|
|
return self.context.layers[self._base_layer].minimum_address
|
|
|
|
def is_valid(self, offset: int, length: int = 1) -> bool:
|
|
return self.context.layers[self._base_layer].is_valid(offset, length)
|
|
|
|
def mapping(self, offset: int, length: int, ignore_errors: bool = False) -> Iterable[Tuple[int, int, int, str]]:
|
|
yield (offset, offset, length, self._base_layer)
|
|
|
|
def get_stream(self, index) -> Optional['PdbMSFStream']:
|
|
self.read_streams()
|
|
if index not in self._streams:
|
|
raise ValueError("Stream not present")
|
|
if self._streams[index]:
|
|
layer = self.context.layers[self._streams[index]]
|
|
if isinstance(layer, PdbMSFStream):
|
|
return layer
|
|
return None
|
|
|
|
|
|
class PdbMSFStream(interfaces.layers.TranslationLayerInterface):
|
|
|
|
def __init__(self,
|
|
context: 'interfaces.context.ContextInterface',
|
|
config_path: str,
|
|
name: str,
|
|
metadata: Optional[Dict[str, Any]] = None) -> None:
|
|
super().__init__(context, config_path, name, metadata)
|
|
self._base_layer = self.config["base_layer"]
|
|
self._pages = self.config.get("pages", None)
|
|
if not self._pages:
|
|
raise ValueError("Invalid/no pages specified")
|
|
if not isinstance(self._pdb_layer, PdbMSF):
|
|
raise TypeError("Base Layer must be a PdbMSF layer")
|
|
|
|
@property
|
|
def pdb_symbol_table(self) -> Optional[str]:
|
|
layer = self._context.layers[self._base_layer]
|
|
if isinstance(layer, PdbMSF):
|
|
return layer.pdb_symbol_table
|
|
else:
|
|
return None
|
|
|
|
@classmethod
|
|
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
|
return [
|
|
requirements.ListRequirement(name = 'pages', element_type = int, min_elements = 1),
|
|
requirements.TranslationLayerRequirement(name = 'base_layer'),
|
|
requirements.IntRequirement(name = 'maximum_size')
|
|
]
|
|
|
|
def mapping(self, offset: int, length: int, ignore_errors: bool = False) -> Iterable[Tuple[int, int, int, str]]:
|
|
returned = 0
|
|
page_size = self._pdb_layer.page_size
|
|
while length > 0:
|
|
page = math.floor((offset + returned) / page_size)
|
|
page_position = ((offset + returned) % page_size)
|
|
chunk_size = min(page_size - page_position, length)
|
|
yield (offset + returned, (self._pages[page] * page_size) + page_position, chunk_size, self._base_layer)
|
|
returned += chunk_size
|
|
length -= chunk_size
|
|
|
|
@property
|
|
def dependencies(self) -> List[str]:
|
|
return [self._base_layer]
|
|
|
|
def is_valid(self, offset: int, length: int = 1) -> bool:
|
|
return self.minimum_address <= offset and offset + length < self.maximum_address
|
|
|
|
@property
|
|
def minimum_address(self) -> int:
|
|
return 0
|
|
|
|
@property
|
|
def maximum_address(self) -> int:
|
|
return self.config.get('maximum_size', len(self._pages) * self._pdb_layer.page_size)
|
|
|
|
@property
|
|
def _pdb_layer(self) -> PdbMSF:
|
|
if self._base_layer not in self._context.layers:
|
|
raise ValueError("No PdbMSF layer found: {}".format(self._base_layer))
|
|
result = self._context.layers[self._base_layer]
|
|
if isinstance(result, PdbMSF):
|
|
return result
|
|
raise ValueError("Base layer is not PdbMSF")
|