From 41f964cfe12d3958c284e00a08860de94cbb6f52 Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Wed, 25 Jun 2025 20:16:59 +0300 Subject: [PATCH] Plugins: determine process exe deletion structurally --- .../plugins/linux/malware/process_spoofing.py | 109 ++++++++++-------- 1 file changed, 63 insertions(+), 46 deletions(-) diff --git a/volatility3/framework/plugins/linux/malware/process_spoofing.py b/volatility3/framework/plugins/linux/malware/process_spoofing.py index a38367bd1..1c2b1a478 100644 --- a/volatility3/framework/plugins/linux/malware/process_spoofing.py +++ b/volatility3/framework/plugins/linux/malware/process_spoofing.py @@ -22,7 +22,6 @@ class ProcessSpoofing(plugins.PluginInterface): _required_framework_version = (2, 0, 0) _version = (1, 1, 0) - deleted = " (deleted)" @classmethod def get_requirements(cls): @@ -51,7 +50,7 @@ class ProcessSpoofing(plugins.PluginInterface): cls, context: interfaces.context.ContextInterface, task: interfaces.objects.ObjectInterface, - ) -> Optional[str]: + ) -> Tuple[Optional[str], bool]: """ Extract the executable path from task_struct.mm.exe_file @@ -60,33 +59,54 @@ class ProcessSpoofing(plugins.PluginInterface): task: task_struct object of the process Returns: - Executable path or None if not available + Tuple of (basename, is_deleted) or (None, False) if not available """ + is_deleted = False + try: mm = task.mm - if not mm or not mm.is_readable(): - # Kernel threads don't have mm struct - return None + except (exceptions.InvalidAddressException, AttributeError) as e: + vollog.debug(f"Unable to access mm for task at {task.vol.offset:#x}: {e}") + return None, is_deleted + if not mm or not mm.is_readable(): + # Kernel threads don't have mm struct + return None, is_deleted + + try: exe_file = mm.exe_file - - if not exe_file or not exe_file.is_readable(): - return None - - exe_inode = exe_file.f_path.dentry.d_inode - exe_path = linux.LinuxUtilities.path_for_file(context, task, exe_file) - - # If the inode link count is 0, the process image has been deleted - if exe_inode.i_nlink == 0: - exe_path += cls.deleted - - return exe_path if exe_path else None - except (exceptions.InvalidAddressException, AttributeError) as e: vollog.debug( - f"Unable to read executable path for task at {task.vol.offset:#x}: {e}" + f"Unable to access exe_file for task at {task.vol.offset:#x}: {e}" ) - return None + return None, is_deleted + + if not exe_file or not exe_file.is_readable(): + return None, is_deleted + + try: + exe_inode = exe_file.f_path.dentry.d_inode + exe_path = linux.LinuxUtilities.path_for_file(context, task, exe_file) + except (exceptions.InvalidAddressException, AttributeError) as e: + vollog.debug( + f"Unable to read exe_file path for task at {task.vol.offset:#x}: {e}" + ) + return None, is_deleted + + if not exe_path: + return None, is_deleted + + try: + # Check if the inode link count is 0 (process image has been deleted) + is_deleted = exe_inode.i_nlink == 0 + except (exceptions.InvalidAddressException, AttributeError) as e: + vollog.debug( + f"Unable to check inode link count for task at {task.vol.offset:#x}: {e}" + ) + # Continue without deletion info - we still have the path + + basename = PurePosixPath(exe_path).name + return basename, is_deleted @classmethod def get_cmdline_basename( @@ -155,7 +175,7 @@ class ProcessSpoofing(plugins.PluginInterface): def _extract_process_names( self, task: interfaces.objects.ObjectInterface - ) -> Tuple[Optional[str], Optional[str], Optional[str]]: + ) -> Tuple[Optional[str], Optional[str], Optional[str], bool]: """ Extract all three process name sources for comparison @@ -163,21 +183,20 @@ class ProcessSpoofing(plugins.PluginInterface): task: task_struct object of the process Returns: - Tuple of (exe_path_basename, cmdline_basename, comm) + Tuple of (exe_basename, cmdline_basename, comm, is_deleted) """ - exe_path = self.get_executable_path(self.context, task) - exe_basename = PurePosixPath(exe_path).name if exe_path else None + exe_basename, is_deleted = self.get_executable_path(self.context, task) cmdline_basename = self.get_cmdline_basename(self.context, task) comm = self.get_comm(task) - return exe_basename, cmdline_basename, comm + return exe_basename, cmdline_basename, comm, is_deleted def _detect_spoofing( self, exe_basename: Optional[str], cmdline_basename: Optional[str], comm: Optional[str], - ) -> Tuple[bool, bool, bool]: + ) -> Tuple[bool, bool]: """ Analyze the three name sources to detect potential spoofing @@ -187,34 +206,26 @@ class ProcessSpoofing(plugins.PluginInterface): comm: Name from comm field Returns: - Tuple of (is_deleted, cmdline_spoofed, comm_spoofed) boolean flags + Tuple of (cmdline_spoofed, comm_spoofed) boolean flags """ - # Check if process image has been deleted - is_deleted = exe_basename and exe_basename.endswith(self.deleted) - - # Get clean basename for comparison (without " (deleted)" suffix) - clean_exe_basename = exe_basename - if is_deleted: - clean_exe_basename = exe_basename[: len(self.deleted) * -1] - # Skip kernel threads - need at least 2 sources for comparison available_sources = sum( - 1 for name in [clean_exe_basename, cmdline_basename, comm] if name + 1 for name in [exe_basename, cmdline_basename, comm] if name ) if available_sources < 2: - return False, False, False + return False, False # Check for cmdline spoofing cmdline_spoofed = False - if clean_exe_basename and cmdline_basename: - cmdline_spoofed = clean_exe_basename != cmdline_basename + if exe_basename and cmdline_basename: + cmdline_spoofed = exe_basename != cmdline_basename # Check for comm spoofing (comm is truncated to 15 characters) comm_spoofed = False - if clean_exe_basename and comm: - comm_spoofed = clean_exe_basename[:15] != comm + if exe_basename and comm: + comm_spoofed = exe_basename[:15] != comm - return is_deleted, cmdline_spoofed, comm_spoofed + return cmdline_spoofed, comm_spoofed def _generator(self, tasks) -> Iterator[Tuple[int, Tuple]]: """ @@ -231,13 +242,19 @@ class ProcessSpoofing(plugins.PluginInterface): pid = task.pid ppid = task.get_parent_pid() - exe_basename, cmdline_basename, comm = self._extract_process_names(task) + exe_basename, cmdline_basename, comm, is_deleted = ( + self._extract_process_names(task) + ) - is_deleted, cmdline_spoofed, comm_spoofed = self._detect_spoofing( + cmdline_spoofed, comm_spoofed = self._detect_spoofing( exe_basename, cmdline_basename, comm ) + # Prepare display values exe_render = exe_basename if exe_basename else "N/A" + if is_deleted and exe_basename: + exe_render += " (deleted)" + cmdline_render = cmdline_basename if cmdline_basename else "N/A" comm_render = comm if comm else "N/A" @@ -273,7 +290,7 @@ class ProcessSpoofing(plugins.PluginInterface): ("Comm", str), ("Cmdline_Spoofed", bool), ("Comm_Spoofed", bool), - ("Deleted", bool), + ("Exe_Deleted", bool), ], self._generator( pslist.PsList.list_tasks(