Merge pull request #1273 from dgmcdona/dgmcdona/windows-psxview-win10-bugfix

Windows: psxview win10 bugfix

volatility3/framework/plugins/windows/pslist.py

@@ -4,7 +4,7 @@
 
 import datetime
 import logging
-from typing import Callable, Iterable, List, Type
+from typing import Callable, Iterator, List, Type
 
 from volatility3.framework import renderers, interfaces, layers, exceptions, constants
 from volatility3.framework.configuration import requirements
@@ -12,6 +12,7 @@ from volatility3.framework.objects import utility
 from volatility3.framework.renderers import format_hints
 from volatility3.framework.symbols import intermed
 from volatility3.framework.symbols.windows.extensions import pe
+from volatility3.framework.symbols.windows import extensions
 from volatility3.plugins import timeliner
 
 vollog = logging.getLogger(__name__)
@@ -197,7 +198,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
         filter_func: Callable[
             [interfaces.objects.ObjectInterface], bool
         ] = lambda _: False,
-    ) -> Iterable[interfaces.objects.ObjectInterface]:
+    ) -> Iterator["extensions.EPROCESS"]:
         """Lists all the processes in the primary layer that are in the pid
         config option.
 

volatility3/framework/plugins/windows/psxview.py

@@ -1,9 +1,14 @@
-import datetime, logging, string
+import datetime
+import logging
+import string
+from itertools import chain
+from typing import Dict, Iterable, List
 
 from volatility3.framework import constants, exceptions
-from volatility3.framework.interfaces import plugins
 from volatility3.framework.configuration import requirements
-from volatility3.framework.renderers import format_hints, TreeGrid
+from volatility3.framework.interfaces import plugins
+from volatility3.framework.renderers import TreeGrid, format_hints
+from volatility3.framework.symbols.windows import extensions
 from volatility3.plugins.windows import (
     handles,
     info,
@@ -71,20 +76,19 @@ class PsXView(plugins.PluginInterface):
             "string", max_length=proc.ImageFileName.vol.count, errors="replace"
         )
 
-    def _is_valid_proc_name(self, str):
-        for c in str:
-            if not c in self.valid_proc_name_chars:
-                return False
-        return True
+    def _is_valid_proc_name(self, string: str) -> bool:
+        return all(c in self.valid_proc_name_chars for c in string)
 
-    def _filter_garbage_procs(self, proc_list):
+    def _filter_garbage_procs(
+        self, proc_list: Iterable[extensions.EPROCESS]
+    ) -> List[extensions.EPROCESS]:
         return [
             p
             for p in proc_list
             if p.is_valid() and self._is_valid_proc_name(self._proc_name_to_string(p))
         ]
 
-    def _translate_offset(self, offset):
+    def _translate_offset(self, offset: int) -> int:
         if not self.config["physical-offsets"]:
             return offset
 
@@ -100,21 +104,25 @@ class PsXView(plugins.PluginInterface):
 
         return offset
 
-    def _proc_list_to_dict(self, tasks):
+    def _proc_list_to_dict(
+        self, tasks: Iterable[extensions.EPROCESS]
+    ) -> Dict[int, extensions.EPROCESS]:
         tasks = self._filter_garbage_procs(tasks)
         return {self._translate_offset(proc.vol.offset): proc for proc in tasks}
 
     def _check_pslist(self, tasks):
         return self._proc_list_to_dict(tasks)
 
-    def _check_psscan(self, layer_name, symbol_table):
+    def _check_psscan(
+        self, layer_name: str, symbol_table: str
+    ) -> Dict[int, extensions.EPROCESS]:
         res = psscan.PsScan.scan_processes(
             context=self.context, layer_name=layer_name, symbol_table=symbol_table
         )
 
         return self._proc_list_to_dict(res)
 
-    def _check_thrdscan(self):
+    def _check_thrdscan(self) -> Dict[int, extensions.EPROCESS]:
         ret = []
 
         for ethread in thrdscan.ThrdScan.scan_threads(
@@ -135,33 +143,38 @@ class PsXView(plugins.PluginInterface):
 
         return self._proc_list_to_dict(ret)
 
-    def _check_csrss_handles(self, tasks, layer_name, symbol_table):
-        ret = []
+    def _check_csrss_handles(
+        self, tasks: Iterable[extensions.EPROCESS], layer_name: str, symbol_table: str
+    ) -> 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(self.context, layer_name, symbol_table)
+
+        cookie = handles_plugin.find_cookie(
+            context=self.context,
+            layer_name=layer_name,
+            symbol_table=symbol_table,
+        )
 
         for p in tasks:
             name = self._proc_name_to_string(p)
-            if name == "csrss.exe":
-                try:
-                    if p.has_member("ObjectTable"):
-                        handles_plugin = handles.Handles(
-                            context=self.context, config_path=self.config_path
-                        )
-                        hndls = list(handles_plugin.handles(p.ObjectTable))
-                        for h in hndls:
-                            if (
-                                h.get_object_type(
-                                    handles_plugin.get_type_map(
-                                        self.context, layer_name, symbol_table
-                                    )
-                                )
-                                == "Process"
-                            ):
-                                ret.append(h.Body.cast("_EPROCESS"))
+            if name != "csrss.exe":
+                continue
 
-                except exceptions.InvalidAddressException:
-                    vollog.log(
-                        constants.LOGLEVEL_VVV, "Cannot access eprocess object table"
-                    )
+            try:
+                ret += [
+                    handle.Body.cast("_EPROCESS")
+                    for handle in handles_plugin.handles(p.ObjectTable)
+                    if handle.get_object_type(type_map, cookie) == "Process"
+                ]
+            except exceptions.InvalidAddressException:
+                vollog.log(
+                    constants.LOGLEVEL_VVV, "Cannot access eprocess object table"
+                )
 
         return self._proc_list_to_dict(ret)
 
@@ -178,7 +191,7 @@ class PsXView(plugins.PluginInterface):
         )
 
         # get processes from each source
-        processes = {}
+        processes: Dict[str, Dict[int, extensions.EPROCESS]] = {}
 
         processes["pslist"] = self._check_pslist(kdbg_list_processes)
         processes["psscan"] = self._check_psscan(layer_name, symbol_table)
@@ -187,27 +200,20 @@ class PsXView(plugins.PluginInterface):
             kdbg_list_processes, layer_name, symbol_table
         )
 
-        # print results
-
-        # list of lists of offsets
-        offsets = [list(processes[source].keys()) for source in processes]
-
-        # flatten to one list
-        offsets = sum(offsets, [])
-
-        # remove duplicates
-        offsets = set(offsets)
+        # Unique set of all offsets from all sources
+        offsets = set(chain(*(mapping.keys() for mapping in processes.values())))
 
         for offset in offsets:
-            proc = None
+            # We know there will be at least one process mapped to each offset
+            proc: extensions.EPROCESS = next(
+                mapping[offset] for mapping in processes.values() if offset in mapping
+            )
 
             in_sources = {src: False for src in processes}
 
-            for source in processes:
-                if offset in processes[source]:
+            for source, process_mapping in processes.items():
+                if offset in process_mapping:
                     in_sources[source] = True
-                    if not proc:
-                        proc = processes[source][offset]
 
             pid = proc.UniqueProcessId
             name = self._proc_name_to_string(proc)