diff --git a/volatility3/framework/plugins/windows/pslist.py b/volatility3/framework/plugins/windows/pslist.py index 8234b210f..478cc8b1b 100644 --- a/volatility3/framework/plugins/windows/pslist.py +++ b/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. diff --git a/volatility3/framework/plugins/windows/psxview.py b/volatility3/framework/plugins/windows/psxview.py index 71919c410..a8d185a2c 100644 --- a/volatility3/framework/plugins/windows/psxview.py +++ b/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)