From 55fe4ba47aece0882a0b5c690710cba1fa438989 Mon Sep 17 00:00:00 2001 From: Dave Lassalle Date: Mon, 15 Jul 2024 15:20:10 -0500 Subject: [PATCH] #118 - MR feedback --- .../framework/plugins/windows/kpcrs.py | 106 ++++++++++++++++++ .../framework/plugins/windows/timers.py | 86 +++----------- .../symbols/windows/extensions/__init__.py | 6 +- 3 files changed, 125 insertions(+), 73 deletions(-) create mode 100644 volatility3/framework/plugins/windows/kpcrs.py diff --git a/volatility3/framework/plugins/windows/kpcrs.py b/volatility3/framework/plugins/windows/kpcrs.py new file mode 100644 index 000000000..558ea844c --- /dev/null +++ b/volatility3/framework/plugins/windows/kpcrs.py @@ -0,0 +1,106 @@ +# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# + +import logging + +from typing import Iterator, List, Tuple + +from volatility3.framework import ( + renderers, + interfaces, + constants, +) +from volatility3.framework.configuration import requirements +from volatility3.framework.objects import utility +from volatility3.framework.renderers import format_hints + +vollog = logging.getLogger(__name__) + + +class KPCRs(interfaces.plugins.PluginInterface): + """Print KPCR structure for each processor""" + + _required_framework_version = (2, 0, 0) + _version = (1, 0, 0) + + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + return [ + requirements.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], + ), + ] + + @classmethod + def list_kpcrs( + cls, + context: interfaces.context.ContextInterface, + kernel_module_name: str, + layer_name: str, + symbol_table: str, + ) -> interfaces.objects.ObjectInterface: + """Returns the KPCR structure for each processor + + Args: + context: The context to retrieve required elements (layers, symbol tables) from + kernel_module_name: The name of the kernel module on which to operate + layer_name: The name of the layer on which to operate + symbol_table: The name of the table containing the kernel symbols + + Returns: + The _KPCR structure for each processor + """ + + kernel = context.modules[kernel_module_name] + cpu_count_offset = kernel.get_symbol("KeNumberProcessors").address + cpu_count = kernel.object( + object_type="unsigned int", layer_name=layer_name, offset=cpu_count_offset + ) + processor_block = kernel.object( + object_type="pointer", + layer_name=layer_name, + offset=kernel.get_symbol("KiProcessorBlock").address, + ) + processor_pointers = utility.array_of_pointers( + context=context, + array=processor_block, + count=cpu_count, + subtype=symbol_table + constants.BANG + "_KPRCB", + ) + for pointer in processor_pointers: + kprcb = pointer.dereference() + reloff = kernel.get_type("_KPCR").relative_child_offset("Prcb") + kpcr = context.object( + symbol_table + constants.BANG + "_KPCR", + offset=kprcb.vol.offset - reloff, + layer_name=layer_name, + ) + yield kpcr + + def _generator(self) -> Iterator[Tuple]: + kernel = self.context.modules[self.config["kernel"]] + layer_name = kernel.layer_name + symbol_table = kernel.symbol_table_name + + for kpcr in self.list_kpcrs( + self.context, self.config["kernel"], layer_name, symbol_table + ): + yield ( + 0, + ( + format_hints.Hex(kpcr.vol.offset), + format_hints.Hex(kpcr.CurrentPrcb), + ), + ) + + def run(self): + return renderers.TreeGrid( + [ + ("Offset", format_hints.Hex), + ("PRCB Offset", format_hints.Hex), + ], + self._generator(), + ) diff --git a/volatility3/framework/plugins/windows/timers.py b/volatility3/framework/plugins/windows/timers.py index ec8bc17e9..d49c28784 100644 --- a/volatility3/framework/plugins/windows/timers.py +++ b/volatility3/framework/plugins/windows/timers.py @@ -7,17 +7,15 @@ import logging from typing import Iterator, List, Tuple, Iterable from volatility3.framework import ( - layers, renderers, interfaces, constants, symbols, ) from volatility3.framework.configuration import requirements -from volatility3.framework.objects import utility from volatility3.framework.renderers import format_hints from volatility3.framework.symbols.windows import versions -from volatility3.plugins.windows import ssdt +from volatility3.plugins.windows import ssdt, kpcrs vollog = logging.getLogger(__name__) @@ -39,73 +37,16 @@ class Timers(interfaces.plugins.PluginInterface): requirements.PluginRequirement( name="ssdt", plugin=ssdt.SSDT, version=(1, 0, 0) ), + requirements.PluginRequirement( + name="kpcrs", plugin=kpcrs.KPCRs, version=(1, 0, 0) + ), ] - @classmethod - def get_kernel_module( - cls, - context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, - ): - """Returns the kernel module based on the layer and symbol_table""" - virtual_layer = context.layers[layer_name] - if not isinstance(virtual_layer, layers.intel.Intel): - raise TypeError("Virtual Layer is not an intel layer") - - kvo = virtual_layer.config["kernel_virtual_offset"] - - ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo) - return ntkrnlmp - - @classmethod - def get_kpcrs( - cls, - context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table: str, - ) -> interfaces.objects.ObjectInterface: - """Returns the KPCR structure for each processor - - Args: - context: The context to retrieve required elements (layers, symbol tables) from - symbol_table: The name of an existing symbol table containing the kernel symbols - config_path: The configuration path within the context of the symbol table to create - - Returns: - The _KPCR structure for each processor - """ - - ntkrnlmp = cls.get_kernel_module(context, layer_name, symbol_table) - cpu_count_offset = ntkrnlmp.get_symbol("KeNumberProcessors").address - cpu_count = ntkrnlmp.object( - object_type="unsigned int", layer_name=layer_name, offset=cpu_count_offset - ) - processor_block = ntkrnlmp.object( - object_type="pointer", - layer_name=layer_name, - offset=ntkrnlmp.get_symbol("KiProcessorBlock").address, - ) - processor_pointers = utility.array_of_pointers( - context=context, - array=processor_block, - count=cpu_count, - subtype=symbol_table + constants.BANG + "_KPRCB", - ) - for pointer in processor_pointers: - kprcb = pointer.dereference() - reloff = ntkrnlmp.get_type("_KPCR").relative_child_offset("Prcb") - kpcr = context.object( - symbol_table + constants.BANG + "_KPCR", - offset=kprcb.vol.offset - reloff, - layer_name=layer_name, - ) - yield kpcr - @classmethod def list_timers( cls, context: interfaces.context.ContextInterface, + kernel_module_name: str, layer_name: str, symbol_table: str, ) -> Iterable[Tuple[str, int, str]]: @@ -113,21 +54,24 @@ class Timers(interfaces.plugins.PluginInterface): Args: context: The context to retrieve required elements (layers, symbol tables) from + kernel_module_name: The name of the kernel module on which to operate layer_name: The name of the layer on which to operate symbol_table: The name of the table containing the kernel symbols Yields: A _KTIMER entry """ - ntkrnlmp = cls.get_kernel_module(context, layer_name, symbol_table) + kernel = context.modules[kernel_module_name] if versions.is_windows_7( context=context, symbol_table=symbol_table ) or versions.is_windows_8_or_later(context=context, symbol_table=symbol_table): # Starting with Windows 7, there is no more KiTimerTableListHead. The list is # at _KPCR.PrcbData.TimerTable.TimerEntries # See http://pastebin.com/FiRsGW3f - for kpcr in cls.get_kpcrs(context, layer_name, symbol_table): + for kpcr in kpcrs.KPCRs.list_kpcrs( + context, kernel_module_name, layer_name, symbol_table + ): if hasattr(kpcr.Prcb.TimerTable, "TableState"): for timer_entries in kpcr.Prcb.TimerTable.TimerEntries: for timer_entry in timer_entries: @@ -160,10 +104,10 @@ class Timers(interfaces.plugins.PluginInterface): # is an array of 256 _LIST_ENTRY for _KTIMERs. array_size = 256 - timer_table_list_head = ntkrnlmp.object( + timer_table_list_head = kernel.object( object_type="array", - offset=ntkrnlmp.get_symbol("KiTimerTableListHead").address, - subtype=ntkrnlmp.get_type("_LIST_ENTRY"), + offset=kernel.get_symbol("KiTimerTableListHead").address, + subtype=kernel.get_type("_LIST_ENTRY"), count=array_size, ) for table in timer_table_list_head: @@ -185,7 +129,9 @@ class Timers(interfaces.plugins.PluginInterface): self.context, kernel.layer_name, kernel.symbol_table_name ) - for timer in self.list_timers(self.context, layer_name, symbol_table): + for timer in self.list_timers( + self.context, self.config["kernel"], layer_name, symbol_table + ): if not timer.valid_type(): continue try: diff --git a/volatility3/framework/symbols/windows/extensions/__init__.py b/volatility3/framework/symbols/windows/extensions/__init__.py index b9eabcda6..86ea2febb 100755 --- a/volatility3/framework/symbols/windows/extensions/__init__.py +++ b/volatility3/framework/symbols/windows/extensions/__init__.py @@ -1011,12 +1011,12 @@ class KTIMER(objects.StructType): def get_raw_dpc(self): """Returns the encoded DPC since it may not look like a pointer after encoding""" symbol_table_name = self.get_symbol_table_name() - ulonglong_type = self._context.symbol_space.get_type( - symbol_table_name + constants.BANG + "unsigned long long" + pointer_type = self._context.symbol_space.get_type( + symbol_table_name + constants.BANG + "pointer" ) return self._context.object( - object_type=ulonglong_type, + object_type=pointer_type, layer_name=self.vol.layer_name, offset=self.Dpc.vol.offset, )