mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-09 19:27:39 +02:00
Remove NLP DTB finder, since it seems most other scanners cover all the bases.
This commit is contained in:
Mike Auty
@@ -1,216 +0,0 @@
|
||||
import logging
|
||||
import math
|
||||
import struct
|
||||
import typing
|
||||
|
||||
from volatility.framework import exceptions, interfaces, validity
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.layers import intel
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
validity_tests = {} # type: typing.Dict[typing.Type[intel.Intel], typing.List[typing.Tuple[int, int]]]
|
||||
validity_tests.update({intel.Intel: [],
|
||||
intel.IntelPAE: [],
|
||||
intel.Intel32e: [
|
||||
(0b10111011, 0b00100011),
|
||||
(0b1, 0b1),
|
||||
(0b1111011, 0b1100011),
|
||||
(0b1111011, 0b1100011)]})
|
||||
|
||||
|
||||
class NlpDtbScanner(interfaces.layers.ScannerInterface):
|
||||
"""A scanner that checks for a mapping (and no invalid mappings) to quickly identify DTBs
|
||||
|
||||
A secondary method to test the initial returns should be used to reduce false positives.
|
||||
|
||||
Some of the masks used are heuristics, but have been tested against recent
|
||||
images from OSX, Windows, and Linux.
|
||||
"""
|
||||
|
||||
# Overlap must be a multiple of the page size
|
||||
overlap = 0x4000
|
||||
|
||||
def __init__(self,
|
||||
layer_class: typing.Type[intel.Intel],
|
||||
physical_layer: interfaces.layers.DataLayerInterface) -> None:
|
||||
super().__init__()
|
||||
self._layer_class = layer_class
|
||||
self._physical_layer = physical_layer
|
||||
|
||||
def test_entries(self, valid_entries: typing.Iterable[typing.Tuple[int, int]]) -> bool:
|
||||
"""Scans through valid_entries, descending to see whether one can be successfully mapped to completion
|
||||
|
||||
Returns the first valid DTB or None is no valid DTBs could be found
|
||||
"""
|
||||
for _, entry in valid_entries:
|
||||
if self.test_entry(entry, 1):
|
||||
return True
|
||||
return False
|
||||
|
||||
def _get_mask(self, value: int) -> int:
|
||||
"""Returns a value that correctly masks all numbers less than that of value"""
|
||||
bits = int(math.ceil(math.log(value, 2)))
|
||||
return (1 << (bits + 1)) - 1
|
||||
|
||||
def test_entry(self, entry: int, level: int = 0) -> bool:
|
||||
"""Tests an individual entry at a particular level in the structure"""
|
||||
name, size, large_page = self._layer_class.structure[level]
|
||||
|
||||
# Test the page for validity
|
||||
validity_mask, validity_success = validity_tests[self._layer_class][level]
|
||||
if not (entry & validity_mask == validity_success):
|
||||
return False
|
||||
|
||||
# If it's a large page, we accept that it maps appropriately.
|
||||
if large_page and (entry & (1 << 7)):
|
||||
return True
|
||||
|
||||
# Short circuit out if the entry is the last level and we're valid
|
||||
if level == len(self._layer_class.structure) - 1:
|
||||
return True
|
||||
|
||||
# Somewhat hacky means of determining which format string to use
|
||||
format_str = 'Q'
|
||||
if size == 10:
|
||||
format_str = 'I'
|
||||
|
||||
# We're going to check all entries, so we don't need to figure out indexes
|
||||
# or cut up any addresses
|
||||
page_bits = math.ceil(math.log2(self._layer_class.page_size))
|
||||
next_table_offset = (entry & self._get_mask(self._layer_class.maximum_address)) >> page_bits
|
||||
|
||||
try:
|
||||
next_table = self._physical_layer.read(next_table_offset, self._layer_class.page_size)
|
||||
except exceptions.InvalidAddressException:
|
||||
return False
|
||||
|
||||
calcsize = struct.calcsize('<' + str(2 ** size) + format_str)
|
||||
if calcsize > self._layer_class.page_size:
|
||||
raise ValueError("Size of number of entries cannot exceed a single page")
|
||||
new_entries = struct.unpack('<' + str(2 ** size) + format_str, next_table)
|
||||
for new_entry in new_entries:
|
||||
if self.test_entry(new_entry, level + 1):
|
||||
return True
|
||||
return False
|
||||
|
||||
def __call__(self, data: bytes, data_offset: int) \
|
||||
-> typing.Generator[typing.Tuple[int, typing.List[typing.Tuple[int, int]]], None, None]:
|
||||
structure = self._layer_class.structure # type: typing.List[typing.Tuple[str, int, bool]]
|
||||
name, size, large_page = structure[0]
|
||||
|
||||
# Somewhat hacky means of determining which format string to use
|
||||
format_str = 'Q'
|
||||
if size == 10:
|
||||
format_str = 'I'
|
||||
|
||||
for page_offset in range(0, len(data), self._layer_class.page_size):
|
||||
calcsize = struct.calcsize('<' + str(2 ** size) + format_str)
|
||||
if calcsize > self._layer_class.page_size:
|
||||
raise ValueError("Size of number of entries cannot exceed a single page")
|
||||
if len(data[page_offset:page_offset + calcsize]) < calcsize:
|
||||
continue
|
||||
entries = struct.unpack('<' + str(2 ** size) + format_str, data[page_offset:page_offset + calcsize])
|
||||
valid_entries = [] # type: typing.List[typing.Tuple[int, int]]
|
||||
invalid_count = 0
|
||||
user_count = 0
|
||||
supervisor_count = 0
|
||||
|
||||
entry_num = 0
|
||||
if len(validity_tests[self._layer_class]) < 1:
|
||||
return
|
||||
validity_mask, validity_success = validity_tests[self._layer_class][0]
|
||||
for entry in entries:
|
||||
if (entry & validity_mask) == validity_success:
|
||||
valid_entries.append((entry_num, entry))
|
||||
|
||||
if (entry & (1 << 2)):
|
||||
user_count = user_count + 1
|
||||
else:
|
||||
supervisor_count = supervisor_count + 1
|
||||
|
||||
elif entry != 0:
|
||||
invalid_count = invalid_count + 1
|
||||
|
||||
entry_num = entry_num + 1
|
||||
|
||||
if invalid_count == 0 and len(valid_entries) > 3 and user_count != 0 and supervisor_count != 0:
|
||||
if page_offset < self.chunk_size:
|
||||
yield (data_offset + page_offset, valid_entries)
|
||||
|
||||
|
||||
class NlpDtbfinder(interfaces.automagic.AutomagicInterface):
|
||||
"""Operating-system agnostic DirectoryTableBase scanner
|
||||
|
||||
Checks a page for references off to other pages.
|
||||
"""
|
||||
priority = 11
|
||||
|
||||
def __call__(self,
|
||||
context: interfaces.context.ContextInterface,
|
||||
config_path: str,
|
||||
requirement: interfaces.configuration.RequirementInterface,
|
||||
progress_callback: validity.ProgressCallback = None) \
|
||||
-> None:
|
||||
results = {}
|
||||
sub_config_path = interfaces.configuration.path_join(config_path, requirement.name)
|
||||
if (not interfaces.configuration.path_join(sub_config_path, "page_map_offset") in context.config and
|
||||
isinstance(requirement, requirements.TranslationLayerRequirement) and
|
||||
requirement.requirements.get("class", False)):
|
||||
class_req = requirement.requirements["class"]
|
||||
for layer_class in validity_tests:
|
||||
if (layer_class.__module__ + "." + layer_class.__name__ == class_req.config_value(context,
|
||||
sub_config_path)):
|
||||
|
||||
if ("memory_layer" in requirement.requirements and
|
||||
not requirement.requirements["memory_layer"].unsatisfied(context, sub_config_path)):
|
||||
physical_layer = requirement.requirements["memory_layer"].config_value(context, sub_config_path)
|
||||
if not isinstance(physical_layer, str):
|
||||
raise ValueError("Physical Layer configuration must be a string")
|
||||
layer = context.memory[physical_layer]
|
||||
valid_entries = layer.scan(context,
|
||||
NlpDtbScanner(layer_class, layer),
|
||||
progress_callback)
|
||||
results[layer_class] = physical_layer, valid_entries
|
||||
|
||||
for layer_class in results:
|
||||
physical_layer, sub_results = results[layer_class]
|
||||
for dtb, valid_entries in sub_results:
|
||||
if NlpDtbScanner(layer_class, self.context.memory[physical_layer]).test_entries(valid_entries):
|
||||
context.config[
|
||||
interfaces.configuration.path_join(sub_config_path, "page_map_offset")] = dtb
|
||||
requirement.construct(context, config_path)
|
||||
break
|
||||
else:
|
||||
for subreq in requirement.requirements.values():
|
||||
self(context, sub_config_path, subreq)
|
||||
|
||||
|
||||
class NlpDtbStacker(interfaces.automagic.StackerLayerInterface):
|
||||
stack_order = 13
|
||||
|
||||
@classmethod
|
||||
def stack(cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
layer_name: str,
|
||||
progress_callback: validity.ProgressCallback = None):
|
||||
"""Attempts to determine and stack an intel layer on a physical layer where possible"""
|
||||
if isinstance(context.memory[layer_name], intel.Intel):
|
||||
return None
|
||||
layer, dtb = None, 0
|
||||
for layer_class in validity_tests:
|
||||
scanner = NlpDtbScanner(layer_class, context.memory[layer_name])
|
||||
results = context.memory[layer_name].scan(context, scanner, progress_callback)
|
||||
for dtb, valid_entries in results:
|
||||
if scanner.test_entries(valid_entries):
|
||||
new_layer_name = context.memory.free_layer_name("IntelLayer")
|
||||
config_path = interfaces.configuration.path_join("IntelHelper", new_layer_name)
|
||||
context.config[interfaces.configuration.path_join(config_path, "memory_layer")] = layer_name
|
||||
context.config[interfaces.configuration.path_join(config_path, "page_map_offset")] = dtb
|
||||
layer = layer_class(context, config_path = config_path, name = new_layer_name)
|
||||
break
|
||||
if layer is not None:
|
||||
break
|
||||
if layer is not None:
|
||||
vollog.debug("DTB was found at: 0x{:0x}".format(dtb))
|
||||
return layer
|
||||
Reference in New Issue
Block a user