Plugins: determine process exe deletion structurally

Author: SolitudePy

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 @@ def get_executable_path(
         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 @@ def get_executable_path(
             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
+        except (exceptions.InvalidAddressException, AttributeError) as e:
+            vollog.debug(
+                f"Unable to access exe_file for task at {task.vol.offset:#x}: {e}"
+            )
+            return None, is_deleted
 
-            if not exe_file or not exe_file.is_readable():
-                return None
+        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 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
+        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 read executable path for task at {task.vol.offset:#x}: {e}"
+                f"Unable to check inode link count for task at {task.vol.offset:#x}: {e}"
             )
-            return None
+            # Continue without deletion info - we still have the path
+
+        basename = PurePosixPath(exe_path).name
+        return basename, is_deleted
 
     @classmethod
     def get_cmdline_basename(
@@ -155,29 +175,28 @@ def get_comm(cls, task: interfaces.objects.ObjectInterface) -> Optional[str]:
 
     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
 
         Args:
             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 @@ def _detect_spoofing(
             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 @@ def _generator(self, tasks) -> Iterator[Tuple[int, Tuple]]:
                 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 @@ def run(self):
                 ("Comm", str),
                 ("Cmdline_Spoofed", bool),
                 ("Comm_Spoofed", bool),
-                ("Deleted", bool),
+                ("Exe_Deleted", bool),
             ],
             self._generator(
                 pslist.PsList.list_tasks(