Plugins: determine process exe deletion structurally

This commit is contained in:
SolitudePy
2025-12-29 19:00:30 +02:00
parent 5c7cf11152
commit 41f964cfe1
@@ -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(