From d579f7f583a4a8227221cab4121ebc2fed00f5b8 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Mon, 1 May 2017 16:06:35 +0100 Subject: [PATCH] Add in a condensed version of Nick Petroni's DTB finder/stacker code. --- .../framework/automagic/nlpdtbfinder.py | 189 ++++++++++++++++++ 1 file changed, 189 insertions(+) create mode 100644 volatility/framework/automagic/nlpdtbfinder.py diff --git a/volatility/framework/automagic/nlpdtbfinder.py b/volatility/framework/automagic/nlpdtbfinder.py new file mode 100644 index 000000000..78c96119f --- /dev/null +++ b/volatility/framework/automagic/nlpdtbfinder.py @@ -0,0 +1,189 @@ +import logging +import math +import struct + +from volatility.framework import interfaces, exceptions +from volatility.framework.configuration import requirements +from volatility.framework.layers import intel + +vollog = logging.getLogger(__name__) + +validity_tests = {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, physical_layer): + super().__init__() + self._layer_class = layer_class + self._physical_layer = physical_layer + + def test_entries(self, valid_entries): + """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 test_entry(self, entry, level = 0): + """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 + maxvirtaddr_bits = math.ceil(math.log2(self._layer_class.maximum_address + 1)) + next_table_offset = (entry & ((2 ** (maxvirtaddr_bits + 1)) - 1)) >> self._layer_class.page_size + + try: + next_table = self._physical_layer.read(next_table_offset, self._layer_class.page_size) + except exceptions.InvalidAddressException: + print("Failing, invalid address {}".format(hex(next_table_offset))) + return False + + calcsize = struct.calcsize('<' + str(2 ** size) + format_str) + if calcsize > self._layer_class.page_size: + print(size, format_str) + 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 + print("Dropped out with no success: {}".format(name)) + return False + + def __call__(self, data, data_offset): + structure = self._layer_class.structure + 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 = [] + invalid_count = 0 + user_count = 0 + supervisor_count = 0 + + entry_num = 0 + 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: + yield (data_offset + page_offset, valid_entries) + + +class NlpDtbfinder(interfaces.automagic.AutomagicInterface): + def __call__(self, context, config_path, requirement, progress_callback = None): + results = {} + sub_config_path = interfaces.configuration.path_join(config_path, requirement.name) + if (isinstance(requirement, requirements.TranslationLayerRequirement) and + requirement.requirements.get("class", None)): + 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) + + valid_entries = context.memory[physical_layer].scan(context, + NlpDtbScanner(layer_class, + physical_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 = 10 + + @classmethod + def stack(cls, context, layer_name, progress_callback = 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 = list(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