Fix timers and kpcrs bugs, missing smear checks, and API. Closes #1642

Author: atcuno

volatility3/framework/plugins/windows/kpcrs.py

@@ -4,7 +4,7 @@
 
 import logging
 
-from typing import Iterator, List, Tuple
+from typing import Iterator, Generator, List, Tuple
 
 from volatility3.framework import (
     renderers,
@@ -22,7 +22,7 @@ class KPCRs(interfaces.plugins.PluginInterface):
     """Print KPCR structure for each processor"""
 
     _required_framework_version = (2, 0, 0)
-    _version = (1, 0, 0)
+    _version = (2, 0, 0)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -39,60 +39,70 @@ def list_kpcrs(
         cls,
         context: interfaces.context.ContextInterface,
         kernel_module_name: str,
-        layer_name: str,
-        symbol_table: str,
-    ) -> interfaces.objects.ObjectInterface:
+    ) -> Generator[Tuple[interfaces.objects.ObjectInterface, int], None, None]:
         """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]
+        kernel_layer = context.layers[kernel.layer_name]
+
+        kpcr_type = kernel.get_type("_KPCR")
+
+        reloff = kpcr_type.relative_child_offset("Prcb")
+
+        if kpcr_type.has_member("CurrentPrcb"):
+            kpcr_member = "CurrentPrcb"
+        else:
+            kpcr_member = "Prcb"
+
         cpu_count_offset = kernel.get_symbol("KeNumberProcessors").address
+
         cpu_count = kernel.object(
-            object_type="unsigned int", layer_name=layer_name, offset=cpu_count_offset
+            object_type="unsigned int",
+            layer_name=kernel_layer.name,
+            offset=cpu_count_offset,
         )
+
         processor_block = kernel.object(
             object_type="pointer",
-            layer_name=layer_name,
+            layer_name=kernel_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",
+            subtype=kernel.symbol_table_name + 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
+
+            object_address = kprcb.vol.offset - reloff
+
+            if not kernel_layer.is_valid(kprcb.vol.offset):
+                continue
+
+            kpcr = kernel.object("_KPCR", offset=object_address, absolute=True)
+
+            yield kpcr, kpcr.member(kpcr_member)
 
     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
-        ):
+        for kpcr, current_prcb in self.list_kpcrs(self.context, self.config["kernel"]):
             yield (
                 0,
                 (
                     format_hints.Hex(kpcr.vol.offset),
-                    format_hints.Hex(kpcr.CurrentPrcb),
+                    format_hints.Hex(current_prcb),
                 ),
             )
 

volatility3/framework/plugins/windows/timers.py

@@ -38,7 +38,7 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
                 name="ssdt", plugin=ssdt.SSDT, version=(2, 0, 0)
             ),
             requirements.PluginRequirement(
-                name="kpcrs", plugin=kpcrs.KPCRs, version=(1, 0, 0)
+                name="kpcrs", plugin=kpcrs.KPCRs, version=(2, 0, 0)
             ),
         ]
 
@@ -47,54 +47,52 @@ def list_timers(
         cls,
         context: interfaces.context.ContextInterface,
         kernel_module_name: str,
-        layer_name: str,
-        symbol_table: str,
     ) -> Iterable[extensions.KTIMER]:
         """Lists all kernel timers.
 
         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
         """
 
         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):
+            context=context, symbol_table=kernel.symbol_table_name
+        ) or versions.is_windows_8_or_later(
+            context=context, symbol_table=kernel.symbol_table_name
+        ):
             # 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 kpcrs.KPCRs.list_kpcrs(
-                context, kernel_module_name, layer_name, symbol_table
-            ):
+            for kpcr, _ in kpcrs.KPCRs.list_kpcrs(context, kernel_module_name):
                 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",
+                                kernel.symbol_table_name + 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",
+                            kernel.symbol_table_name + 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)
+            context=context, symbol_table=kernel.symbol_table_name
+        ) or versions.is_vista_or_later(
+            context=context, symbol_table=kernel.symbol_table_name
+        ):
+            is_64bit = symbols.symbol_table_is_64bit(context, kernel.symbol_table_name)
             if is_64bit or versions.is_vista_or_later(
-                context=context, symbol_table=symbol_table
+                context=context, symbol_table=kernel.symbol_table_name
             ):
                 # On XP x64, Windows 2003 SP1-SP2, and Vista SP0-SP2, KiTimerTableListHead
                 # is an array of 512 _KTIMER_TABLE_ENTRY structs.
@@ -112,7 +110,7 @@ def list_timers(
             )
             for table in timer_table_list_head:
                 for timer in table.to_list(
-                    symbol_table + constants.BANG + "_KTIMER",
+                    kernel.symbol_table_name + constants.BANG + "_KTIMER",
                     "TimerListEntry",
                 ):
                     yield timer
@@ -121,8 +119,6 @@ def list_timers(
             raise NotImplementedError("This version of Windows is not supported!")
 
     def _generator(self) -> Iterator[Tuple]:
-        kernel = self.context.modules[self.config["kernel"]]
-
         collection = ssdt.SSDT.build_module_collection(
             context=self.context,
             kernel_module_name=self.config["kernel"],
@@ -132,8 +128,6 @@ def _generator(self) -> Iterator[Tuple]:
         for timer in self.list_timers(
             self.context,
             self.config["kernel"],
-            kernel.layer_name,
-            kernel.symbol_table_name,
         ):
             if not timer.valid_type():
                 continue