From e172676d3caaecd3592211cd561aa612d934bcbc Mon Sep 17 00:00:00 2001 From: Dave Lassalle Date: Fri, 21 Jun 2024 15:07:18 -0500 Subject: [PATCH 1/8] #118 - initial timers plugin --- volatility3/framework/objects/utility.py | 21 ++ .../framework/plugins/windows/timers.py | 264 ++++++++++++++++++ .../framework/symbols/windows/__init__.py | 1 + .../symbols/windows/extensions/__init__.py | 80 ++++++ 4 files changed, 366 insertions(+) create mode 100644 volatility3/framework/plugins/windows/timers.py diff --git a/volatility3/framework/objects/utility.py b/volatility3/framework/objects/utility.py index 0292608c1..8aa527cdb 100644 --- a/volatility3/framework/objects/utility.py +++ b/volatility3/framework/objects/utility.py @@ -7,6 +7,27 @@ from typing import Optional, Union from volatility3.framework import interfaces, objects, constants +def rol(value: int, count: int, max_bits: int = 64) -> int: + """A rotate-left instruction in Python""" + max_bits_mask = (1 << max_bits) - 1 + return (value << count % max_bits) & max_bits_mask | ( + (value & max_bits_mask) >> (max_bits - (count % max_bits)) + ) + + +def bswap_32(value: int) -> int: + value = ((value << 8) & 0xFF00FF00) | ((value >> 8) & 0x00FF00FF) + + return ((value << 16) | (value >> 16)) & 0xFFFFFFFF + + +def bswap_64(value: int) -> int: + low = bswap_32((value >> 32)) + high = bswap_32((value & 0xFFFFFFFF)) + + return ((high << 32) | low) & 0xFFFFFFFFFFFFFFFF + + def array_to_string( array: "objects.Array", count: Optional[int] = None, errors: str = "replace" ) -> interfaces.objects.ObjectInterface: diff --git a/volatility3/framework/plugins/windows/timers.py b/volatility3/framework/plugins/windows/timers.py new file mode 100644 index 000000000..19d52d9bf --- /dev/null +++ b/volatility3/framework/plugins/windows/timers.py @@ -0,0 +1,264 @@ +# 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, Iterable + +from volatility3.framework import exceptions, 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 + +vollog = logging.getLogger(__name__) + + +class Timers(interfaces.plugins.PluginInterface): + """Print kernel timers and associated module DPCs""" + + _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"], + ), + requirements.PluginRequirement( + name="ssdt", plugin=ssdt.SSDT, 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, + layer_name: str, + symbol_table: str, + ) -> Iterable[Tuple[str, int, str]]: + """Lists all kernel timers. + + Args: + context: The context to retrieve required elements (layers, symbol tables) from + 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) + + 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): + if hasattr(kpcr.Prcb.TimerTable, "TableState"): + for timer_entries in kpcr.Prcb.TimerTable.TimerEntries: + for timer_entry in timer_entries: + for timer in timer_entry.Entry.to_list( + symbol_table + constants.BANG + "_KTIMER", + "TimerListEntry", + ): + yield timer + + else: + for timer_entries in kpcr.Prcb.TimerTable.TimerEntries: + for timer in timer_entries.Entry.to_list( + symbol_table + constants.BANG + "_KTIMER", + "TimerListEntry", + ): + yield timer + + elif versions.is_xp_or_2003( + context=context, symbol_table=symbol_table + ) or versions.is_vista_or_later( + context=context, symbol_table=symbol_table + ): + is_64bit = symbols.symbol_table_is_64bit(context, symbol_table) + if is_64bit or versions.is_vista_or_later(context=context, symbol_table=symbol_table): + # On XP x64, Windows 2003 SP1-SP2, and Vista SP0-SP2, KiTimerTableListHead + # is an array of 512 _KTIMER_TABLE_ENTRY structs. + array_size = 512 + else: + # On XP SP0-SP3 x86 and Windows 2003 SP0, KiTimerTableListHead + # is an array of 256 _LIST_ENTRY for _KTIMERs. + array_size = 256 + + timer_table_list_head = ntkrnlmp.object( + object_type="array", + offset=ntkrnlmp.get_symbol("KiTimerTableListHead").address, + subtype=ntkrnlmp.get_type("_LIST_ENTRY"), + count=array_size, + ) + for table in timer_table_list_head: + for timer in table.to_list( + symbol_table + constants.BANG + "_KTIMER", + "TimerListEntry", + ): + yield timer + + else: + raise NotImplementedError("This version of Windows is not supported!") + + + def _generator(self) -> Iterator[Tuple]: + kernel = self.context.modules[self.config["kernel"]] + layer_name = kernel.layer_name + symbol_table = kernel.symbol_table_name + + collection = ssdt.SSDT.build_module_collection( + self.context, kernel.layer_name, kernel.symbol_table_name + ) + + for timer in self.list_timers(self.context, layer_name, symbol_table): + if not timer.valid_type(): + continue + try: + dpc = timer.get_dpc() + if dpc == 0: + continue + if dpc.DeferredRoutine == 0: + continue + deferred_routine = dpc.DeferredRoutine + except Exception as e: + continue + + module_symbols = list( + collection.get_module_symbols_by_absolute_location(deferred_routine) + ) + + if module_symbols: + for module_name, symbol_generator in module_symbols: + symbols_found = False + + # we might have multiple symbols pointing to the same location + for symbol in symbol_generator: + symbols_found = True + yield ( + 0, + ( + format_hints.Hex(timer.vol.offset), + timer.get_due_time(), + timer.Period, + timer.get_signaled(), + format_hints.Hex(deferred_routine), + module_name, + symbol.split(constants.BANG)[1], + ), + ) + + # no symbols, but we at least can report the module name + if not symbols_found: + yield ( + 0, + ( + format_hints.Hex(timer.vol.offset), + timer.get_due_time(), + timer.Period, + timer.get_signaled(), + format_hints.Hex(deferred_routine), + module_name, + renderers.NotAvailableValue(), + ), + ) + else: + # no module was found at the absolute location + yield ( + 0, + ( + format_hints.Hex(timer.vol.offset), + timer.get_due_time(), + timer.Period, + timer.get_signaled(), + format_hints.Hex(deferred_routine), + renderers.NotAvailableValue(), + renderers.NotAvailableValue(), + ), + ) + + + def run(self): + return renderers.TreeGrid( + [ + ("Offset", format_hints.Hex), + ("DueTime", str), + ("Period(ms)", int), + ("Signaled", str), + ("Routine", format_hints.Hex), + ("Module", str), + ("Symbol", str), + ], + self._generator(), + ) diff --git a/volatility3/framework/symbols/windows/__init__.py b/volatility3/framework/symbols/windows/__init__.py index abf9f6da3..4aeb22dcf 100755 --- a/volatility3/framework/symbols/windows/__init__.py +++ b/volatility3/framework/symbols/windows/__init__.py @@ -39,6 +39,7 @@ class WindowsKernelIntermedSymbols(intermed.IntermediateSymbolTable): self.set_type_class("_VACB", extensions.VACB) self.set_type_class("_POOL_TRACKER_BIG_PAGES", pool.POOL_TRACKER_BIG_PAGES) self.set_type_class("_IMAGE_DOS_HEADER", pe.IMAGE_DOS_HEADER) + self.set_type_class("_KTIMER", extensions.KTIMER) # Might not necessarily defined in every version of windows self.optional_set_type_class("_IMAGE_NT_HEADERS", pe.IMAGE_NT_HEADERS) diff --git a/volatility3/framework/symbols/windows/extensions/__init__.py b/volatility3/framework/symbols/windows/extensions/__init__.py index a8fa7b2ff..040c57bae 100755 --- a/volatility3/framework/symbols/windows/extensions/__init__.py +++ b/volatility3/framework/symbols/windows/extensions/__init__.py @@ -20,6 +20,7 @@ from volatility3.framework import ( ) from volatility3.framework.interfaces.objects import ObjectInterface from volatility3.framework.layers import intel +from volatility3.framework.objects import utility from volatility3.framework.renderers import conversion from volatility3.framework.symbols import generic from volatility3.framework.symbols.windows.extensions import kdbg, pe, pool @@ -994,6 +995,85 @@ class TOKEN(objects.StructType): vollog.log(constants.LOGLEVEL_VVVV, "Broken Token Privileges.") +class KTIMER(objects.StructType): + """A class for Kernel Timers""" + + VALID_TYPES = { + 8: "TimerNotificationObject", + 9: "TimerSynchronizationObject", + } + + def get_signaled(self): + if self.Header.SignalState: + return "Yes" + return "-" + + 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" + ) + + return self._context.object( + object_type=ulonglong_type, + layer_name=self.vol.layer_name, + offset=self.Dpc.vol.offset, + ) + def valid_type(self): + return self.Header.Type in self.VALID_TYPES + + def get_due_time(self): + return "{0:#010x}:{1:#010x}".format(self.DueTime.HighPart, self.DueTime.LowPart) + + def get_dpc(self): + """Return Dpc, and if Windows 7 or later, decode it""" + symbol_table_name = self.get_symbol_table_name() + kvo = self._context.layers[self.vol.native_layer_name].config[ + "kernel_virtual_offset" + ] + ntkrnlmp = self._context.module( + symbol_table_name, + layer_name=self.vol.native_layer_name, + offset=kvo, + native_layer_name=self.vol.native_layer_name, + ) + + try: + wait_never = ntkrnlmp.object( + object_type="unsigned long long", + offset=ntkrnlmp.get_symbol("KiWaitNever").address, + ) + + wait_always = ntkrnlmp.object( + object_type="unsigned long long", + offset=ntkrnlmp.get_symbol("KiWaitAlways").address, + ) + except exceptions.SymbolError: + wait_never = None + wait_always = None + + if wait_never is None or wait_always is None: + return self.Dpc + else: + low_byte = (wait_never) & 0xFF + entry = utility.rol(self.get_raw_dpc() ^ wait_never, low_byte) + swap_xor = self.vol.offset | 0xFFFF000000000000 + entry = utility.bswap_64(entry ^ swap_xor) + dpc = entry ^ wait_always + + symbol_table_name = self.get_symbol_table_name() + kdpc_type = self._context.symbol_space.get_type( + symbol_table_name + constants.BANG + "_KDPC" + ) + + return self._context.object( + object_type=kdpc_type, + layer_name=self.vol.layer_name, + offset=dpc, + ) + + class KTHREAD(objects.StructType): """A class for thread control block objects.""" From c79dc52a854d2a5be41283a720081f63471c0df8 Mon Sep 17 00:00:00 2001 From: Dave Lassalle Date: Fri, 21 Jun 2024 15:09:21 -0500 Subject: [PATCH 2/8] #118 - black formatted --- .../framework/plugins/windows/timers.py | 23 +++++++++++-------- .../symbols/windows/extensions/__init__.py | 7 +++--- 2 files changed, 17 insertions(+), 13 deletions(-) diff --git a/volatility3/framework/plugins/windows/timers.py b/volatility3/framework/plugins/windows/timers.py index 19d52d9bf..e54b2a13f 100644 --- a/volatility3/framework/plugins/windows/timers.py +++ b/volatility3/framework/plugins/windows/timers.py @@ -6,7 +6,14 @@ import logging from typing import Iterator, List, Tuple, Iterable -from volatility3.framework import exceptions, layers, renderers, interfaces, constants, symbols +from volatility3.framework import ( + exceptions, + layers, + renderers, + interfaces, + constants, + symbols, +) from volatility3.framework.configuration import requirements from volatility3.framework.objects import utility from volatility3.framework.renderers import format_hints @@ -117,9 +124,7 @@ class Timers(interfaces.plugins.PluginInterface): if versions.is_windows_7( context=context, symbol_table=symbol_table - ) or versions.is_windows_8_or_later( - 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 @@ -143,11 +148,11 @@ class Timers(interfaces.plugins.PluginInterface): elif versions.is_xp_or_2003( context=context, symbol_table=symbol_table - ) or versions.is_vista_or_later( - context=context, symbol_table=symbol_table - ): + ) or versions.is_vista_or_later(context=context, symbol_table=symbol_table): is_64bit = symbols.symbol_table_is_64bit(context, symbol_table) - if is_64bit or versions.is_vista_or_later(context=context, symbol_table=symbol_table): + if is_64bit or versions.is_vista_or_later( + context=context, symbol_table=symbol_table + ): # On XP x64, Windows 2003 SP1-SP2, and Vista SP0-SP2, KiTimerTableListHead # is an array of 512 _KTIMER_TABLE_ENTRY structs. array_size = 512 @@ -172,7 +177,6 @@ class Timers(interfaces.plugins.PluginInterface): else: raise NotImplementedError("This version of Windows is not supported!") - def _generator(self) -> Iterator[Tuple]: kernel = self.context.modules[self.config["kernel"]] layer_name = kernel.layer_name @@ -248,7 +252,6 @@ class Timers(interfaces.plugins.PluginInterface): ), ) - def run(self): return renderers.TreeGrid( [ diff --git a/volatility3/framework/symbols/windows/extensions/__init__.py b/volatility3/framework/symbols/windows/extensions/__init__.py index 040c57bae..b830a9d9f 100755 --- a/volatility3/framework/symbols/windows/extensions/__init__.py +++ b/volatility3/framework/symbols/windows/extensions/__init__.py @@ -453,9 +453,9 @@ class FILE_OBJECT(objects.StructType, pool.ExecutiveObject): ].is_valid(self.FileName.Buffer) def file_name_with_device(self) -> Union[str, interfaces.renderers.BaseAbsentValue]: - name: Union[str, interfaces.renderers.BaseAbsentValue] = ( - renderers.UnreadableValue() - ) + name: Union[ + str, interfaces.renderers.BaseAbsentValue + ] = renderers.UnreadableValue() # this pointer needs to be checked against native_layer_name because the object may # be instantiated from a primary (virtual) layer or a memory (physical) layer. @@ -1020,6 +1020,7 @@ class KTIMER(objects.StructType): layer_name=self.vol.layer_name, offset=self.Dpc.vol.offset, ) + def valid_type(self): return self.Header.Type in self.VALID_TYPES From 7065446e8ab9efb02254e4415235072d30422596 Mon Sep 17 00:00:00 2001 From: Dave Lassalle Date: Fri, 21 Jun 2024 15:17:40 -0500 Subject: [PATCH 3/8] #118 - fix black issues --- volatility3/framework/plugins/windows/timers.py | 1 - volatility3/framework/symbols/windows/extensions/__init__.py | 1 - 2 files changed, 2 deletions(-) diff --git a/volatility3/framework/plugins/windows/timers.py b/volatility3/framework/plugins/windows/timers.py index e54b2a13f..ec8bc17e9 100644 --- a/volatility3/framework/plugins/windows/timers.py +++ b/volatility3/framework/plugins/windows/timers.py @@ -7,7 +7,6 @@ import logging from typing import Iterator, List, Tuple, Iterable from volatility3.framework import ( - exceptions, layers, renderers, interfaces, diff --git a/volatility3/framework/symbols/windows/extensions/__init__.py b/volatility3/framework/symbols/windows/extensions/__init__.py index b830a9d9f..895a8c094 100755 --- a/volatility3/framework/symbols/windows/extensions/__init__.py +++ b/volatility3/framework/symbols/windows/extensions/__init__.py @@ -1045,7 +1045,6 @@ class KTIMER(objects.StructType): object_type="unsigned long long", offset=ntkrnlmp.get_symbol("KiWaitNever").address, ) - wait_always = ntkrnlmp.object( object_type="unsigned long long", offset=ntkrnlmp.get_symbol("KiWaitAlways").address, From 2d7789f8509664f3930705bf83524d6b5ec8de33 Mon Sep 17 00:00:00 2001 From: Dave Lassalle Date: Fri, 21 Jun 2024 15:49:03 -0500 Subject: [PATCH 4/8] #118 - fix black issues --- .../framework/symbols/windows/extensions/__init__.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/volatility3/framework/symbols/windows/extensions/__init__.py b/volatility3/framework/symbols/windows/extensions/__init__.py index 895a8c094..4d4ffc055 100755 --- a/volatility3/framework/symbols/windows/extensions/__init__.py +++ b/volatility3/framework/symbols/windows/extensions/__init__.py @@ -453,9 +453,9 @@ class FILE_OBJECT(objects.StructType, pool.ExecutiveObject): ].is_valid(self.FileName.Buffer) def file_name_with_device(self) -> Union[str, interfaces.renderers.BaseAbsentValue]: - name: Union[ - str, interfaces.renderers.BaseAbsentValue - ] = renderers.UnreadableValue() + name: Union[str, interfaces.renderers.BaseAbsentValue] = ( + renderers.UnreadableValue() + ) # this pointer needs to be checked against native_layer_name because the object may # be instantiated from a primary (virtual) layer or a memory (physical) layer. From 146baab14c4a17bdcbc35715b30a4d190ed82214 Mon Sep 17 00:00:00 2001 From: Dave Lassalle Date: Fri, 21 Jun 2024 15:54:16 -0500 Subject: [PATCH 5/8] #118 - refactor get_dpc --- .../framework/symbols/windows/extensions/__init__.py | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/volatility3/framework/symbols/windows/extensions/__init__.py b/volatility3/framework/symbols/windows/extensions/__init__.py index 4d4ffc055..b9eabcda6 100755 --- a/volatility3/framework/symbols/windows/extensions/__init__.py +++ b/volatility3/framework/symbols/windows/extensions/__init__.py @@ -1040,7 +1040,7 @@ class KTIMER(objects.StructType): native_layer_name=self.vol.native_layer_name, ) - try: + if ntkrnlmp.has_symbol("KiWaitNever") and ntkrnlmp.has_symbol("KiWaitAlways"): wait_never = ntkrnlmp.object( object_type="unsigned long long", offset=ntkrnlmp.get_symbol("KiWaitNever").address, @@ -1049,13 +1049,7 @@ class KTIMER(objects.StructType): object_type="unsigned long long", offset=ntkrnlmp.get_symbol("KiWaitAlways").address, ) - except exceptions.SymbolError: - wait_never = None - wait_always = None - if wait_never is None or wait_always is None: - return self.Dpc - else: low_byte = (wait_never) & 0xFF entry = utility.rol(self.get_raw_dpc() ^ wait_never, low_byte) swap_xor = self.vol.offset | 0xFFFF000000000000 @@ -1072,6 +1066,8 @@ class KTIMER(objects.StructType): layer_name=self.vol.layer_name, offset=dpc, ) + else: + return self.Dpc class KTHREAD(objects.StructType): From 55fe4ba47aece0882a0b5c690710cba1fa438989 Mon Sep 17 00:00:00 2001 From: Dave Lassalle Date: Mon, 15 Jul 2024 15:20:10 -0500 Subject: [PATCH 6/8] #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, ) From 99cf48597abbc1126ef06731ed7836c04614514d Mon Sep 17 00:00:00 2001 From: Dave Lassalle Date: Mon, 15 Jul 2024 15:46:27 -0500 Subject: [PATCH 7/8] #118 - use canonicalize for offset --- volatility3/framework/symbols/windows/extensions/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/volatility3/framework/symbols/windows/extensions/__init__.py b/volatility3/framework/symbols/windows/extensions/__init__.py index 86ea2febb..d5c3d3f96 100755 --- a/volatility3/framework/symbols/windows/extensions/__init__.py +++ b/volatility3/framework/symbols/windows/extensions/__init__.py @@ -1052,7 +1052,7 @@ class KTIMER(objects.StructType): low_byte = (wait_never) & 0xFF entry = utility.rol(self.get_raw_dpc() ^ wait_never, low_byte) - swap_xor = self.vol.offset | 0xFFFF000000000000 + swap_xor = self._context.layers[self.vol.native_layer_name].canonicalize(self.vol.offset) entry = utility.bswap_64(entry ^ swap_xor) dpc = entry ^ wait_always From 43e22d72bdaa2cbb6366b1911a6ff38ced83394f Mon Sep 17 00:00:00 2001 From: Dave Lassalle Date: Tue, 16 Jul 2024 09:23:06 -0500 Subject: [PATCH 8/8] #118 - black formatting --- volatility3/framework/symbols/windows/extensions/__init__.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/volatility3/framework/symbols/windows/extensions/__init__.py b/volatility3/framework/symbols/windows/extensions/__init__.py index d5c3d3f96..b333755f7 100755 --- a/volatility3/framework/symbols/windows/extensions/__init__.py +++ b/volatility3/framework/symbols/windows/extensions/__init__.py @@ -1052,7 +1052,9 @@ class KTIMER(objects.StructType): low_byte = (wait_never) & 0xFF entry = utility.rol(self.get_raw_dpc() ^ wait_never, low_byte) - swap_xor = self._context.layers[self.vol.native_layer_name].canonicalize(self.vol.offset) + swap_xor = self._context.layers[self.vol.native_layer_name].canonicalize( + self.vol.offset + ) entry = utility.bswap_64(entry ^ swap_xor) dpc = entry ^ wait_always