Merge pull request #1786 from volatilityfoundation/dgmcdona/handles_interface

Windows Handles: Refactor plugin

Author: ikelos

volatility3/framework/plugins/windows/callbacks.py

@@ -48,7 +48,7 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
                 name="driverirp", component=driverirp.DriverIrp, version=(1, 0, 0)
             ),
             requirements.VersionRequirement(
-                name="handles", component=handles.Handles, version=(3, 0, 0)
+                name="handles", component=handles.Handles, version=(4, 0, 0)
             ),
         ]
 

volatility3/framework/plugins/windows/dumpfiles.py

@@ -5,18 +5,12 @@
 import logging
 import ntpath
 import re
-from typing import List, Tuple, Type, Optional, Generator
-
-from volatility3.framework import (
-    interfaces,
-    exceptions,
-    constants,
-    renderers,
-)
+from typing import Generator, List, Optional, Tuple, Type
+
+from volatility3.framework import constants, exceptions, interfaces, renderers
 from volatility3.framework.configuration import requirements
 from volatility3.framework.renderers import format_hints
-from volatility3.plugins.windows import handles
-from volatility3.plugins.windows import pslist
+from volatility3.plugins.windows import handles, pslist
 
 vollog = logging.getLogger(__name__)
 
@@ -76,7 +70,7 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
                 name="pslist", component=pslist.PsList, version=(3, 0, 0)
             ),
             requirements.VersionRequirement(
-                name="handles", component=handles.Handles, version=(3, 0, 0)
+                name="handles", component=handles.Handles, version=(4, 0, 0)
             ),
         ]
 
@@ -231,14 +225,11 @@ def _generator(self, procs: List, offsets: List):
             # private variables, so we need an instance (for now, anyway). We _could_ call Handles._generator()
             # to do some of the other work that is duplicated here, but then we'd need to parse the TreeGrid
             # results instead of just dealing with them as direct objects here.
-            handles_plugin = handles.Handles(
-                context=self.context, config_path=self._config_path
-            )
-            type_map = handles_plugin.get_type_map(
+            type_map = handles.Handles.get_type_map(
                 context=self.context,
                 kernel_module_name=self.config["kernel"],
             )
-            cookie = handles_plugin.find_cookie(
+            cookie = handles.Handles.find_cookie(
                 context=self.context,
                 kernel_module_name=self.config["kernel"],
             )
@@ -255,7 +246,11 @@ def _generator(self, procs: List, offsets: List):
                     )
                     continue
 
-                for entry in handles_plugin.handles(object_table):
+                for entry in handles.Handles.handles(
+                    context=self.context,
+                    kernel_module_name=self.config["kernel"],
+                    handle_table=object_table,
+                ):
                     try:
                         obj_type = entry.get_object_type(type_map, cookie)
                         if obj_type == "File":

volatility3/framework/plugins/windows/handles.py

@@ -3,7 +3,7 @@
 #
 
 import logging
-from typing import Dict, List, Optional
+from typing import Dict, Iterator, List, Optional
 
 from volatility3.framework import constants, exceptions, interfaces, renderers, symbols
 from volatility3.framework.configuration import requirements
@@ -18,13 +18,9 @@ class Handles(interfaces.plugins.PluginInterface):
     """Lists process open handles."""
 
     _required_framework_version = (2, 0, 0)
-    _version = (3, 0, 0)
+    _version = (4, 0, 0)
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self._type_map = None
-        self._cookie = None
-        self._level_mask = 7
+    LEVEL_MASK = 7
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -54,18 +50,27 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
             ),
         ]
 
-    def _get_item(self, handle_table_entry, handle_value):
-        """Given  a handle table entry (_HANDLE_TABLE_ENTRY) structure from a
+    @classmethod
+    def _get_item(
+        cls,
+        context: interfaces.context.ContextInterface,
+        kernel_module_name: str,
+        handle_table_entry: interfaces.objects.ObjectInterface,
+        handle_value: int,
+    ) -> Optional[interfaces.objects.ObjectInterface]:
+        """
+        Given  a handle table entry (_HANDLE_TABLE_ENTRY) structure from a
         process' handle table, determine where the corresponding object's
-        _OBJECT_HEADER can be found."""
+        _OBJECT_HEADER can be found, and construct and return the _OBJECT_HEADER
+        """
 
-        kernel = self.context.modules[self.config["kernel"]]
+        kernel = context.modules[kernel_module_name]
 
         virtual = kernel.layer_name
 
         try:
             # before windows 7
-            if not self.context.layers[virtual].is_valid(handle_table_entry.Object):
+            if not context.layers[virtual].is_valid(handle_table_entry.Object):
                 return None
             fast_ref = handle_table_entry.Object.cast("_EX_FAST_REF")
 
@@ -78,7 +83,7 @@ def _get_item(self, handle_table_entry, handle_value):
         except AttributeError:
             # starting with windows 8
             is_64bit = symbols.symbol_table_is_64bit(
-                context=self.context, symbol_table_name=kernel.symbol_table_name
+                context=context, symbol_table_name=kernel.symbol_table_name
             )
 
             if is_64bit:
@@ -104,7 +109,7 @@ def _get_item(self, handle_table_entry, handle_value):
                 offset = info_table & ~7
 
             # print("LowValue: {0:#x} Magic: {1:#x} Offset: {2:#x}".format(handle_table_entry.InfoTable, magic, offset))
-            object_header = self.context.object(
+            object_header = context.object(
                 kernel.symbol_table_name + constants.BANG + "_OBJECT_HEADER",
                 virtual,
                 offset=offset,
@@ -205,11 +210,23 @@ def find_cookie(
             offset=symbol_offset,
         )
 
-    def _make_handle_array(self, offset, level, depth=0):
-        """Parse a process' handle table and yield valid handle table entries,
-        going as deep into the table "levels" as necessary."""
+    @classmethod
+    def _make_handle_array(
+        cls,
+        context: interfaces.context.ContextInterface,
+        kernel_module_name: str,
+        offset: int,
+        level: int,
+        depth: int = 0,
+    ) -> Iterator[interfaces.objects.ObjectInterface]:
+        """
+        Parses a process' handle table by constructing an array of
+        `_HANDLE_TABLE_ENTRY` structures at the given offset, and yields valid
+        handle table entries, going as deep into the table "levels" as
+        necessary.
+        """
 
-        kernel = self.context.modules[self.config["kernel"]]
+        kernel = context.modules[kernel_module_name]
 
         if level > 0:
             subtype = kernel.get_type("pointer")
@@ -218,7 +235,7 @@ def _make_handle_array(self, offset, level, depth=0):
             subtype = kernel.get_type("_HANDLE_TABLE_ENTRY")
             count = 0x1000 / subtype.size
 
-        if not self.context.layers[kernel.layer_name].is_valid(offset):
+        if not context.layers[kernel.layer_name].is_valid(offset):
             return None
 
         table = kernel.object(
@@ -229,7 +246,7 @@ def _make_handle_array(self, offset, level, depth=0):
             absolute=True,
         )
 
-        layer_object = self.context.layers[kernel.layer_name]
+        layer_object = context.layers[kernel.layer_name]
         masked_offset = offset & layer_object.maximum_address
 
         for i in range(len(table)):
@@ -243,11 +260,13 @@ def _make_handle_array(self, offset, level, depth=0):
             # The code above this calls `is_valid` on the `offset`
             # It is sent but then does not validate `entry` before
             # sending it to `_get_item`
-            if not self.context.layers[kernel.layer_name].is_valid(entry.vol.offset):
+            if not context.layers[kernel.layer_name].is_valid(entry.vol.offset):
                 continue
 
             if level > 0:
-                yield from self._make_handle_array(entry, level - 1, depth)
+                yield from cls._make_handle_array(
+                    context, kernel_module_name, entry, level - 1, depth
+                )
                 depth += 1
             else:
                 handle_multiplier = 4
@@ -258,7 +277,7 @@ def _make_handle_array(self, offset, level, depth=0):
                     / (subtype.size / handle_multiplier)
                 ) + handle_level_base
 
-                item = self._get_item(entry, handle_value)
+                item = cls._get_item(context, kernel_module_name, entry, handle_value)
 
                 if item is None:
                     continue
@@ -272,18 +291,31 @@ def _make_handle_array(self, offset, level, depth=0):
                 except exceptions.InvalidAddressException:
                     continue
 
-    def handles(self, handle_table):
+    @classmethod
+    def handles(
+        cls,
+        context: interfaces.context.ContextInterface,
+        kernel_module_name: str,
+        handle_table: interfaces.objects.ObjectInterface,
+    ) -> Iterator[interfaces.objects.ObjectInterface]:
+        """
+        Takes a context, kernel module name, and handle table structure
+        (_HANDLE_TABLE), and yields _HANDLE_TABLE_ENTRY structures from the
+        handle table.
+        """
         try:
-            TableCode = handle_table.TableCode & ~self._level_mask
-            table_levels = handle_table.TableCode & self._level_mask
+            TableCode = handle_table.TableCode & ~cls.LEVEL_MASK
+            table_levels = handle_table.TableCode & cls.LEVEL_MASK
         except exceptions.InvalidAddressException:
             vollog.log(
                 constants.LOGLEVEL_VVV,
                 "Handle table parsing was aborted due to an invalid address exception",
             )
             return None
 
-        yield from self._make_handle_array(TableCode, table_levels)
+        yield from cls._make_handle_array(
+            context, kernel_module_name, TableCode, table_levels
+        )
 
     def _generator(self, procs):
         type_map = self.get_type_map(
@@ -306,7 +338,9 @@ def _generator(self, procs):
 
             process_name = utility.array_to_string(proc.ImageFileName)
 
-            for entry in self.handles(object_table):
+            for entry in self.handles(
+                self.context, self.config["kernel"], object_table
+            ):
                 try:
                     obj_type = entry.get_object_type(type_map, cookie)
                     if obj_type is None:

volatility3/framework/plugins/windows/poolscanner.py

@@ -142,7 +142,7 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
                 architectures=["Intel32", "Intel64"],
             ),
             requirements.VersionRequirement(
-                name="handles", component=handles.Handles, version=(3, 0, 0)
+                name="handles", component=handles.Handles, version=(4, 0, 0)
             ),
             requirements.VersionRequirement(
                 name="pool_header_scanner",

volatility3/framework/plugins/windows/psxview.py

@@ -9,12 +9,7 @@
 from volatility3.framework.interfaces import plugins
 from volatility3.framework.renderers import format_hints
 from volatility3.framework.symbols.windows import extensions
-from volatility3.plugins.windows import (
-    handles,
-    pslist,
-    psscan,
-    thrdscan,
-)
+from volatility3.plugins.windows import handles, pslist, psscan, thrdscan
 
 vollog = logging.getLogger(__name__)
 
@@ -58,7 +53,7 @@ def get_requirements(cls):
                 name="thrdscan", component=thrdscan.ThrdScan, version=(2, 0, 0)
             ),
             requirements.VersionRequirement(
-                name="handles", component=handles.Handles, version=(3, 0, 0)
+                name="handles", component=handles.Handles, version=(4, 0, 0)
             ),
             requirements.BooleanRequirement(
                 name="physical-offsets",
@@ -144,15 +139,11 @@ def _check_csrss_handles(
     ) -> Dict[int, extensions.EPROCESS]:
         ret: List[extensions.EPROCESS] = []
 
-        handles_plugin = handles.Handles(
-            context=self.context, config_path=self.config_path
-        )
-
-        type_map = handles_plugin.get_type_map(
+        type_map = handles.Handles.get_type_map(
             context=self.context, kernel_module_name=self.config["kernel"]
         )
 
-        cookie = handles_plugin.find_cookie(
+        cookie = handles.Handles.find_cookie(
             context=self.context, kernel_module_name=self.config["kernel"]
         )
 
@@ -164,7 +155,11 @@ def _check_csrss_handles(
             try:
                 ret += [
                     handle.Body.cast("_EPROCESS")
-                    for handle in handles_plugin.handles(p.ObjectTable)
+                    for handle in handles.Handles.handles(
+                        context=self.context,
+                        kernel_module_name=self.config["kernel"],
+                        handle_table=p.ObjectTable,
+                    )
                     if handle.get_object_type(type_map, cookie) == "Process"
                 ]
             except exceptions.InvalidAddressException: