diff --git a/volatility3/framework/plugins/windows/psxview.py b/volatility3/framework/plugins/windows/psxview.py index 3bf6c27f5..1fde5b12d 100644 --- a/volatility3/framework/plugins/windows/psxview.py +++ b/volatility3/framework/plugins/windows/psxview.py @@ -4,19 +4,28 @@ 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.plugins.windows import handles, info, pslist, psscan, sessions, thrdscan +from volatility3.plugins.windows import ( + handles, + info, + pslist, + psscan, + sessions, + thrdscan, +) vollog = logging.getLogger(__name__) + class PsXView(plugins.PluginInterface): - """Lists all processes found via 6 of the methods described in \"The Art of Memory Forensics,\" which may help + """Lists all processes found via 6 of the methods described in \"The Art of Memory Forensics,\" which may help identify processes that are trying to hide themselves. I recommend using -r pretty if you are looking at this plugin's output in a terminal.""" - # I've omitted the desktop thread scanning method because Volatility3 doesn't appear to have the funcitonality - # which the original plugin to do it. - # I don't think it's worth including the sessions method either because both the original psxview plugin - # and Volatility3's sessions plugin begin with the list of processes found by PsList. + # I've omitted the desktop thread scanning method because Volatility3 doesn't appear to have the funcitonality + # which the original plugin to do it. + + # I don't think it's worth including the sessions method either because both the original psxview plugin + # and Volatility3's sessions plugin begin with the list of processes found by PsList. # The original psxview plugin's session code essentially just filters the pslist for processes # whose session ID is not None. I've matched this in my code, but again, it doesn't seem worth including. @@ -28,52 +37,88 @@ class PsXView(plugins.PluginInterface): @classmethod def get_requirements(cls): - return [requirements.ModuleRequirement(name="kernel", description="Windows kernel", architectures=["Intel32", "Intel64"]), - requirements.VersionRequirement(name="info", component=info.Info, version=(1, 0, 0)), - requirements.VersionRequirement(name="pslist", component=pslist.PsList, version=(2, 0, 0)), - requirements.VersionRequirement(name="psscan", component=psscan.PsScan, version=(1, 0, 0)), - requirements.VersionRequirement(name="thrdscan", component=thrdscan.ThrdScan, version=(1, 0, 0)), - requirements.VersionRequirement(name="handles", component=handles.Handles, version=(1, 0, 0)), - requirements.VersionRequirement(name="sessions", component=sessions.Sessions, version=(0, 0, 0)), - requirements.BooleanRequirement(name="identify-expected", description="In the plugin's output, replace false with \ + return [ + requirements.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="info", component=info.Info, version=(1, 0, 0) + ), + requirements.VersionRequirement( + name="pslist", component=pslist.PsList, version=(2, 0, 0) + ), + requirements.VersionRequirement( + name="psscan", component=psscan.PsScan, version=(1, 0, 0) + ), + requirements.VersionRequirement( + name="thrdscan", component=thrdscan.ThrdScan, version=(1, 0, 0) + ), + requirements.VersionRequirement( + name="handles", component=handles.Handles, version=(1, 0, 0) + ), + requirements.VersionRequirement( + name="sessions", component=sessions.Sessions, version=(0, 0, 0) + ), + requirements.BooleanRequirement( + name="identify-expected", + description='In the plugin\'s output, replace false with \ normal where false is the expected result for a Windows machine running normally. \ - Keep in mind that this plugin uses simple checks to identify \"normal\" behavior, \ - so you may want to double-check the legitimacy of these processes yourself.", optional=True), - requirements.BooleanRequirement(name="physical-offsets", description="List processes with phyiscall offsets instead of virtual offsets.", optional=True)] - + Keep in mind that this plugin uses simple checks to identify "normal" behavior, \ + so you may want to double-check the legitimacy of these processes yourself.', + optional=True, + ), + requirements.BooleanRequirement( + name="physical-offsets", + description="List processes with phyiscall offsets instead of virtual offsets.", + optional=True, + ), + ] + def proc_name_to_string(self, proc): - return proc.ImageFileName.cast("string", max_length=proc.ImageFileName.vol.count, errors="replace") + return proc.ImageFileName.cast( + "string", max_length=proc.ImageFileName.vol.count, errors="replace" + ) def is_ascii(self, str): - return str.split('.')[0].isalnum() - + return str.split(".")[0].isalnum() + def filter_garbage_procs(self, proc_list): - return [p for p in proc_list if p.is_valid() and self.is_ascii(self.proc_name_to_string(p))] - + return [ + p + for p in proc_list + if p.is_valid() and self.is_ascii(self.proc_name_to_string(p)) + ] + def translate_offset(self, offset): if self.config["physical-offsets"]: return offset - + kernel = self.context.modules[self.config["kernel"]] layer_name = kernel.layer_name try: - offset = list(self.context.layers[layer_name].mapping(offset=offset, length=0))[0][2] + offset = list( + self.context.layers[layer_name].mapping(offset=offset, length=0) + )[0][2] except: # already have physical address pass return offset - + def proc_list_to_dict(self, tasks): - return {self.translate_offset(proc.vol.offset):proc for proc in tasks} - + return {self.translate_offset(proc.vol.offset): proc for proc in tasks} + def check_pslist(self, tasks): res = self.filter_garbage_procs(tasks) return self.proc_list_to_dict(tasks) - + def check_psscan(self, layer_name, symbol_table): - res = psscan.PsScan.scan_processes(context=self.context, layer_name=layer_name, symbol_table=symbol_table) + res = psscan.PsScan.scan_processes( + context=self.context, layer_name=layer_name, symbol_table=symbol_table + ) res = self.filter_garbage_procs(res) return self.proc_list_to_dict(res) @@ -81,7 +126,9 @@ class PsXView(plugins.PluginInterface): def check_thrdscan(self): ret = [] - for ethread in thrdscan.ThrdScan.scan_threads(self.context, module_name='kernel'): + for ethread in thrdscan.ThrdScan.scan_threads( + self.context, module_name="kernel" + ): process = None try: process = ethread.owning_process() @@ -90,50 +137,70 @@ class PsXView(plugins.PluginInterface): ret.append(process) except AttributeError: - vollog.log(constants.LOGLEVEL_VVV, "Unable to find the owning process of ethread") + vollog.log( + constants.LOGLEVEL_VVV, + "Unable to find the owning process of ethread", + ) return self.proc_list_to_dict(ret) - + def check_csrss_handles(self, tasks, layer_name, symbol_table): ret = [] for p in tasks: name = self.proc_name_to_string(p) - if name == 'csrss.exe': + if name == "csrss.exe": try: if p.has_member("ObjectTable"): - handles_plugin = handles.Handles(context=self.context, config_path=self.config_path) + 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"): + if ( + h.get_object_type( + handles_plugin.get_type_map( + self.context, layer_name, symbol_table + ) + ) + == "Process" + ): ret.append(h.Body.cast("_EPROCESS")) except exceptions.InvalidAddressException: - vollog.log(constants.LOGLEVEL_VVV, "Cannot access eprocess object table") + vollog.log( + constants.LOGLEVEL_VVV, "Cannot access eprocess object table" + ) ret = self.filter_garbage_procs(ret) return self.proc_list_to_dict(ret) def check_session(self, pslist_procs): procs = [p for p in pslist_procs if p.get_session_id() != None] - + return self.proc_list_to_dict(procs) def _generator(self): kernel = self.context.modules[self.config["kernel"]] layer_name = kernel.layer_name - symbol_table = kernel.symbol_table_name + symbol_table = kernel.symbol_table_name + + kdbg_list_processes = list( + pslist.PsList.list_processes( + context=self.context, layer_name=layer_name, symbol_table=symbol_table + ) + ) - kdbg_list_processes = list(pslist.PsList.list_processes(context=self.context, layer_name=layer_name, symbol_table=symbol_table)) - processes = {} - processes['pslist'] = self.check_pslist(kdbg_list_processes) - processes['psscan'] = self.check_psscan(layer_name, symbol_table) - processes['thrdscan'] = self.check_thrdscan() - processes['csrss'] = self.check_csrss_handles(kdbg_list_processes, layer_name, symbol_table) - processes['sessions'] = self.check_session(kdbg_list_processes) + processes["pslist"] = self.check_pslist(kdbg_list_processes) + processes["psscan"] = self.check_psscan(layer_name, symbol_table) + processes["thrdscan"] = self.check_thrdscan() + processes["csrss"] = self.check_csrss_handles( + kdbg_list_processes, layer_name, symbol_table + ) + processes["sessions"] = self.check_session(kdbg_list_processes) seen_offsets = set() for source in processes: @@ -146,12 +213,14 @@ class PsXView(plugins.PluginInterface): name = self.proc_name_to_string(proc) exit_time = proc.get_exit_time() - if (type(exit_time) != datetime.datetime): + if type(exit_time) != datetime.datetime: exit_time = "" else: exit_time = str(exit_time) - in_sources = {src:str(offset in processes[src]) for src in processes} + in_sources = { + src: str(offset in processes[src]) for src in processes + } if self.config["identify-expected"]: f = "False" @@ -173,16 +242,38 @@ class PsXView(plugins.PluginInterface): if in_sources["sessions"] == f: if name.lower() in ["system", "smss.exe"]: - in_sources["sessions"] = n + in_sources["sessions"] = n + + yield ( + 0, + ( + format_hints.Hex(offset), + name, + pid, + in_sources["pslist"], + in_sources["psscan"], + in_sources["thrdscan"], + in_sources["csrss"], + in_sources["sessions"], + exit_time, + ), + ) - yield (0, (format_hints.Hex(offset), name, pid, in_sources["pslist"], - in_sources["psscan"], in_sources["thrdscan"], in_sources["csrss"], - in_sources["sessions"], exit_time)) - - def run(self): offset_type = "(Physical)" if self.config["physical-offsets"] else "(Virtual)" offset_str = "Offset" + offset_type - return TreeGrid([(offset_str, format_hints.Hex), ("Name", str), ("PID", int), ("pslist", str), ("psscan", str), - ("thrdscan", str), ("csrss", str), ("sessions", str), ("Exit Time", str) ], self._generator()) \ No newline at end of file + return TreeGrid( + [ + (offset_str, format_hints.Hex), + ("Name", str), + ("PID", int), + ("pslist", str), + ("psscan", str), + ("thrdscan", str), + ("csrss", str), + ("sessions", str), + ("Exit Time", str), + ], + self._generator(), + )