Merge pull request #1621 from volatilityfoundation/issue_1487_driver_…

…handling

Stop reporting terminated/smeared drivers. Enforce names present. Add…

volatility3/framework/plugins/windows/devicetree.py

@@ -90,16 +90,15 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
                 architectures=["Intel32", "Intel64"],
             ),
             requirements.PluginRequirement(
-                name="driverscan", plugin=driverscan.DriverScan, version=(1, 0, 0)
+                name="driverscan", plugin=driverscan.DriverScan, version=(2, 0, 0)
             ),
         ]
 
     def _generator(self) -> Iterator[Tuple]:
-        kernel = self.context.modules[self.config["kernel"]]
-
         # Scan the Layer for drivers
         for driver in driverscan.DriverScan.scan_drivers(
-            self.context, kernel.layer_name, kernel.symbol_table_name
+            self.context,
+            self.config["kernel"],
         ):
             try:
                 try:

volatility3/framework/plugins/windows/driverirp.py

@@ -2,11 +2,15 @@
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
 
+import logging
+
 from volatility3.framework import constants
 from volatility3.framework import renderers, exceptions, interfaces
 from volatility3.framework.configuration import requirements
 from volatility3.framework.renderers import format_hints
-from volatility3.plugins.windows import ssdt, driverscan
+from volatility3.plugins.windows import ssdt, driverscan, modules
+
+vollog = logging.getLogger(__name__)
 
 MAJOR_FUNCTIONS = [
     "IRP_MJ_CREATE",
@@ -58,7 +62,10 @@ def get_requirements(cls):
                 name="ssdt", plugin=ssdt.SSDT, version=(1, 0, 0)
             ),
             requirements.PluginRequirement(
-                name="driverscan", plugin=driverscan.DriverScan, version=(1, 0, 0)
+                name="driverscan", plugin=driverscan.DriverScan, version=(2, 0, 0)
+            ),
+            requirements.PluginRequirement(
+                name="modules", plugin=modules.Modules, version=(2, 1, 0)
             ),
         ]
 
@@ -69,17 +76,34 @@ def _generator(self):
             self.context, kernel.layer_name, kernel.symbol_table_name
         )
 
+        kernel_space_start = modules.Modules.get_kernel_space_start(
+            self.context, self.config["kernel"]
+        )
+
         for driver in driverscan.DriverScan.scan_drivers(
-            self.context, kernel.layer_name, kernel.symbol_table_name
+            self.context,
+            self.config["kernel"],
         ):
             try:
                 driver_name = driver.get_driver_name()
             except (ValueError, exceptions.InvalidAddressException):
                 driver_name = renderers.NotApplicableValue()
 
-            for i, address in enumerate(driver.MajorFunction):
+            for i in range(len(driver.MajorFunction)):
+                try:
+                    irp_handler = driver.MajorFunction[i]
+                except exceptions.InvalidAddressException:
+                    vollog.debug(
+                        f"Failed to get IRP handler entry at index {i} for driver at {driver.vol.offset:#x}"
+                    )
+                    continue
+
+                # smear
+                if irp_handler < kernel_space_start:
+                    continue
+
                 module_symbols = collection.get_module_symbols_by_absolute_location(
-                    address
+                    irp_handler
                 )
 
                 module_found = False
@@ -96,7 +120,7 @@ def _generator(self):
                                 format_hints.Hex(driver.vol.offset),
                                 driver_name,
                                 MAJOR_FUNCTIONS[i],
-                                format_hints.Hex(address),
+                                format_hints.Hex(irp_handler),
                                 module_name,
                                 symbol.split(constants.BANG)[1],
                             ),
@@ -109,7 +133,7 @@ def _generator(self):
                                 format_hints.Hex(driver.vol.offset),
                                 driver_name,
                                 MAJOR_FUNCTIONS[i],
-                                format_hints.Hex(address),
+                                format_hints.Hex(irp_handler),
                                 module_name,
                                 renderers.NotAvailableValue(),
                             ),
@@ -122,7 +146,7 @@ def _generator(self):
                             format_hints.Hex(driver.vol.offset),
                             driver_name,
                             MAJOR_FUNCTIONS[i],
-                            format_hints.Hex(address),
+                            format_hints.Hex(irp_handler),
                             renderers.NotAvailableValue(),
                             renderers.NotAvailableValue(),
                         ),

volatility3/framework/plugins/windows/drivermodule.py

@@ -5,7 +5,7 @@
 from volatility3.framework import renderers, interfaces
 from volatility3.framework.configuration import requirements
 from volatility3.framework.renderers import format_hints
-from volatility3.plugins.windows import ssdt, driverscan
+from volatility3.plugins.windows import ssdt, driverscan, modules
 
 # built in Windows-components that trigger false positives
 KNOWN_DRIVERS = ["ACPI_HAL", "PnpManager", "RAW", "WMIxWDM", "Win32k", "Fs_Rec"]
@@ -29,7 +29,10 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
                 name="ssdt", plugin=ssdt.SSDT, version=(1, 0, 0)
             ),
             requirements.PluginRequirement(
-                name="driverscan", plugin=driverscan.DriverScan, version=(1, 0, 0)
+                name="driverscan", plugin=driverscan.DriverScan, version=(2, 0, 0)
+            ),
+            requirements.PluginRequirement(
+                name="modules", plugin=modules.Modules, version=(2, 1, 0)
             ),
         ]
 
@@ -45,9 +48,19 @@ def _generator(self) -> Iterator[Tuple]:
             self.context, kernel.layer_name, kernel.symbol_table_name
         )
 
+        kernel_space_start = modules.Modules.get_kernel_space_start(
+            self.context, self.config["kernel"]
+        )
+
         for driver in driverscan.DriverScan.scan_drivers(
-            self.context, kernel.layer_name, kernel.symbol_table_name
+            self.context,
+            self.config["kernel"],
         ):
+            # We want starts of 0 as rootkits often set this value
+            # greater than 0 but less than the kernel space start is smear/terminated though
+            if 0 < driver.DriverStart < kernel_space_start:
+                continue
+
             # we do not care about actual symbol names, we just want to know if the driver points to a known module
             module_symbols = list(
                 collection.get_module_symbols_by_absolute_location(driver.DriverStart)
@@ -59,16 +72,20 @@ def _generator(self) -> Iterator[Tuple]:
                     name,
                 ) = driverscan.DriverScan.get_names_for_driver(driver)
 
+                # drivers without any names will not produce useful output
+                if not driver_name and not service_key and not name:
+                    continue
+
                 known_exception = driver_name in KNOWN_DRIVERS
 
                 yield (
                     0,
                     (
                         format_hints.Hex(driver.vol.offset),
                         known_exception,
-                        driver_name,
-                        service_key,
-                        name,
+                        driver_name or renderers.NotAvailableValue(),
+                        service_key or renderers.NotAvailableValue(),
+                        name or renderers.NotAvailableValue(),
                     ),
                 )
 

volatility3/framework/plugins/windows/driverscan.py

@@ -2,19 +2,19 @@
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
 
-from typing import Iterable
+from typing import Iterable, Optional, Tuple
 
 from volatility3.framework import renderers, interfaces, exceptions
 from volatility3.framework.configuration import requirements
 from volatility3.framework.renderers import format_hints
-from volatility3.plugins.windows import poolscanner
+from volatility3.plugins.windows import poolscanner, modules
 
 
 class DriverScan(interfaces.plugins.PluginInterface):
     """Scans for drivers present in a particular windows memory image."""
 
     _required_framework_version = (2, 0, 0)
-    _version = (1, 0, 0)
+    _version = (2, 0, 0)
 
     @classmethod
     def get_requirements(cls):
@@ -33,8 +33,7 @@ def get_requirements(cls):
     def scan_drivers(
         cls,
         context: interfaces.context.ContextInterface,
-        layer_name: str,
-        symbol_table: str,
+        kernel_module_name: str,
     ) -> Iterable[interfaces.objects.ObjectInterface]:
         """Scans for drivers using the poolscanner module and constraints.
 
@@ -47,18 +46,51 @@ def scan_drivers(
             A list of Driver objects as found from the `layer_name` layer based on Driver pool signatures
         """
 
+        kernel = context.modules[kernel_module_name]
+
+        symbol_table_name = kernel.symbol_table_name
+        layer_name = kernel.layer_name
+
         constraints = poolscanner.PoolScanner.builtin_constraints(
-            symbol_table, [b"Dri\xf6", b"Driv"]
+            symbol_table_name, [b"Dri\xf6", b"Driv"]
+        )
+
+        module = context.module(symbol_table_name, layer_name, 0)
+        driver_start_offset = module.get_type("_DRIVER_OBJECT").relative_child_offset(
+            "DriverStart"
+        )
+
+        kernel_space_start = modules.Modules.get_kernel_space_start(
+            context, kernel_module_name
         )
 
         for result in poolscanner.PoolScanner.generate_pool_scan(
-            context, layer_name, symbol_table, constraints
+            context, layer_name, symbol_table_name, constraints
         ):
             _constraint, mem_object, _header = result
-            yield mem_object
+
+            scanned_layer = context.layers[mem_object.vol.layer_name]
+
+            # *Many* _DRIVER_OBJECT instances were found at the end of a page
+            # leading to member access causing backtraces across several plugins
+            # when members were accessed as the next page was paged out.
+            # `DriverStart` is the first member from the beginning of the structure
+            #  of interest to plugins, so if it is not accessible then this instance
+            # is not useful or usable during analysis
+            # 8 covers this value 32 and 64 bit systems
+            if scanned_layer.is_valid(mem_object.vol.offset + driver_start_offset, 8):
+                # Many/most rootkits zero out their DriverStart member for anti-forensics
+                # so we accept a driver start that is either 0 or is points into kernel memory (the current layer)
+                if (
+                    mem_object.DriverStart == 0
+                    or mem_object.DriverStart > kernel_space_start
+                ):
+                    yield mem_object
 
     @classmethod
-    def get_names_for_driver(cls, driver):
+    def get_names_for_driver(
+        cls, driver
+    ) -> Tuple[Optional[str], Optional[str], Optional[str]]:
         """
         Convenience method for getting the commonly used
         names associated with a driver
@@ -72,37 +104,40 @@ def get_names_for_driver(cls, driver):
         try:
             driver_name = driver.get_driver_name()
         except (ValueError, exceptions.InvalidAddressException):
-            driver_name = renderers.NotApplicableValue()
+            driver_name = None
 
         try:
             service_key = driver.DriverExtension.ServiceKeyName.String
         except exceptions.InvalidAddressException:
-            service_key = renderers.NotApplicableValue()
+            service_key = None
 
         try:
             name = driver.DriverName.String
         except exceptions.InvalidAddressException:
-            name = renderers.NotApplicableValue()
+            name = None
 
         return driver_name, service_key, name
 
     def _generator(self):
-        kernel = self.context.modules[self.config["kernel"]]
-
         for driver in self.scan_drivers(
-            self.context, kernel.layer_name, kernel.symbol_table_name
+            self.context,
+            self.config["kernel"],
         ):
             driver_name, service_key, name = self.get_names_for_driver(driver)
 
+            # Prior to #1481, this plugin reported dozens to hundreds of junk drivers per sample
+            if not driver_name and not service_key and not name:
+                continue
+
             yield (
                 0,
                 (
                     format_hints.Hex(driver.vol.offset),
                     format_hints.Hex(driver.DriverStart),
                     format_hints.Hex(driver.DriverSize),
-                    service_key,
-                    driver_name,
-                    name,
+                    service_key or renderers.NotAvailableValue(),
+                    driver_name or renderers.NotAvailableValue(),
+                    name or renderers.NotAvailableValue(),
                 ),
             )
 

volatility3/framework/plugins/windows/modules.py

@@ -138,16 +138,25 @@ def get_kernel_space_start(cls, context, module_name: str) -> int:
         """
         module = context.modules[module_name]
 
+        # default is used if/when MmSystemRangeStart is paged out
         if symbols.symbol_table_is_64bit(context, module.symbol_table_name):
             object_type = "unsigned long long"
+            default_start = 0xFFFF800000000000
         else:
             object_type = "unsigned long"
+            default_start = 0x80000000
 
         range_start_offset = module.get_symbol("MmSystemRangeStart").address
 
-        kernel_space_start = module.object(
-            object_type=object_type, offset=range_start_offset
-        )
+        try:
+            kernel_space_start = module.object(
+                object_type=object_type, offset=range_start_offset
+            )
+        except exceptions.InvalidAddressException:
+            vollog.debug(
+                f"Unable to read MmSystemRangeStart. Defaulting to {default_start:#x} for the kernel space start."
+            )
+            kernel_space_start = default_start
 
         layer = context.layers[module.layer_name]