Merge pull request #1678 from volatilityfoundation/fix_unloaded_modules

Fix unloaded modules bugs. Change API to fit current formats

volatility3/framework/plugins/windows/unloadedmodules.py

@@ -4,7 +4,7 @@
 
 import logging
 import datetime
-from typing import List, Iterable
+from typing import List, Generator, Tuple
 
 from volatility3.framework import constants
 from volatility3.framework import interfaces, symbols, exceptions
@@ -14,6 +14,7 @@
 from volatility3.framework.renderers import format_hints, conversion
 from volatility3.framework.symbols import intermed
 from volatility3.plugins import timeliner
+from volatility3.plugins.windows import modules
 
 vollog = logging.getLogger(__name__)
 
@@ -22,7 +23,7 @@ class UnloadedModules(interfaces.plugins.PluginInterface, timeliner.TimeLinerInt
     """Lists the unloaded kernel modules."""
 
     _required_framework_version = (2, 0, 0)
-    _version = (1, 0, 2)
+    _version = (2, 0, 0)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -75,10 +76,9 @@ def create_unloadedmodules_table(
     def list_unloadedmodules(
         cls,
         context: interfaces.context.ContextInterface,
-        layer_name: str,
-        symbol_table: str,
+        kernel_module_name: str,
         unloadedmodule_table_name: str,
-    ) -> Iterable[interfaces.objects.ObjectInterface]:
+    ) -> Generator[Tuple[str, int, int, datetime.datetime], None, None]:
         """Lists all the unloaded modules in the primary layer.
 
         Args:
@@ -90,20 +90,16 @@ def list_unloadedmodules(
             A list of Unloaded Modules as retrieved from MmUnloadedDrivers
         """
 
-        kvo = context.layers[layer_name].config.get("kernel_virtual_offset", None)
-        if not kvo:
-            raise ValueError(
-                "Intel layer does not have an associated kernel virtual offset, failing"
-            )
-        ntkrnlmp = context.module(symbol_table, layer_name=layer_name, offset=kvo)
+        ntkrnlmp = context.modules[kernel_module_name]
+
         unloadedmodules_offset = ntkrnlmp.get_symbol("MmUnloadedDrivers").address
         unloadedmodules = ntkrnlmp.object(
             object_type="pointer",
             offset=unloadedmodules_offset,
             subtype="array",
         )
         is_64bit = symbols.symbol_table_is_64bit(
-            context=context, symbol_table_name=symbol_table
+            context=context, symbol_table_name=ntkrnlmp.symbol_table_name
         )
 
         if is_64bit:
@@ -116,53 +112,86 @@ def list_unloadedmodules(
             object_type=unloaded_count_type, offset=last_unloadedmodule_offset
         )
 
+        # Bring down to default when smear present. Some samples had this completely broken
+        if unloaded_count > 1024:
+            vollog.warning(
+                f"Smeared array count found {unloaded_count}. Defaulting to 1024 elements."
+            )
+            unloaded_count = 1024
+
         unloadedmodules_array = context.object(
             object_type=unloadedmodule_table_name
             + constants.BANG
             + "_UNLOADED_DRIVERS",
-            layer_name=layer_name,
+            layer_name=ntkrnlmp.layer_name,
             offset=unloadedmodules,
         )
         unloadedmodules_array.UnloadedDrivers.count = unloaded_count
 
+        kernel_space_start = modules.Modules.get_kernel_space_start(
+            context, kernel_module_name
+        )
+
+        address_mask = context.layers[ntkrnlmp.layer_name].address_mask
+
         for driver in unloadedmodules_array.UnloadedDrivers:
             # Mass testing led to dozens of samples backtracing on this plugin when
             # accessing members of modules coming out this list
             # Given how often temporary drivers load and unload on Win10+, I
             # assume the chance for smear is very high
             try:
-                driver.StartAddress
-                driver.EndAddress
-                driver.CurrentTime
-                yield driver
+                start_address = driver.StartAddress & address_mask
+                end_address = driver.EndAddress & address_mask
+                current_time = driver.CurrentTime
+                driver_name = driver.Name.String
             except exceptions.InvalidAddressException:
                 continue
 
+            if (
+                current_time > 1024
+                and start_address > kernel_space_start
+                and start_address & 0xFFF == 0x0
+                and end_address & 0xFFF == 0x0
+                and end_address > kernel_space_start
+            ):
+                yield driver_name, start_address, end_address, current_time
+
     def _generator(self):
         kernel = self.context.modules[self.config["kernel"]]
 
+        if not kernel.has_symbol("MmUnloadedDrivers"):
+            vollog.error(
+                "The symbol table for this sample is missing the `MmUnloadedDrivers` symbol. Cannot proceed."
+            )
+            return
+
+        if not kernel.has_symbol("MmLastUnloadedDriver"):
+            vollog.error(
+                "The symbol table for this sample is missing the `MmLastUnloadededDriver` symbol. Cannot proceed."
+            )
+            return
+
         unloadedmodule_table_name = self.create_unloadedmodules_table(
             self.context, kernel.symbol_table_name, self.config_path
         )
 
-        for mod in self.list_unloadedmodules(
+        for (
+            driver_name,
+            start_address,
+            end_address,
+            current_time,
+        ) in self.list_unloadedmodules(
             self.context,
-            kernel.layer_name,
-            kernel.symbol_table_name,
+            self.config["kernel"],
             unloadedmodule_table_name,
         ):
-            try:
-                name = mod.Name.String
-            except exceptions.InvalidAddressException:
-                name = renderers.UnreadableValue()
-
             yield (
                 0,
                 (
-                    name,
-                    format_hints.Hex(mod.StartAddress),
-                    format_hints.Hex(mod.EndAddress),
-                    conversion.wintime_to_datetime(mod.CurrentTime),
+                    driver_name,
+                    format_hints.Hex(start_address),
+                    format_hints.Hex(end_address),
+                    conversion.wintime_to_datetime(current_time),
                 ),
             )