volatility3/volatility3/framework/plugins/linux/malware/process_spoofing.py

# This file is Copyright 2025 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#

import logging
from pathlib import PurePosixPath
from typing import Optional, Tuple, Iterator

from volatility3.framework import exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.objects import utility
from volatility3.framework.symbols import linux
from volatility3.plugins.linux import pslist

vollog = logging.getLogger(__name__)


# https://github.com/SolitudePy/linux-mal
class ProcessSpoofing(plugins.PluginInterface):
    """Detects process spoofing by comparing executable path to cmdline & comm fields"""

    _required_framework_version = (2, 27, 0)
    _version = (1, 1, 0)

    @classmethod
    def get_requirements(cls):
        return [
            requirements.ModuleRequirement(
                name="kernel",
                description="Linux kernel",
                architectures=["Intel32", "Intel64"],
            ),
            requirements.VersionRequirement(
                name="pslist", component=pslist.PsList, version=(4, 0, 0)
            ),
            requirements.VersionRequirement(
                name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0)
            ),
            requirements.ListRequirement(
                name="pid",
                description="Filter on specific process IDs",
                element_type=int,
                optional=True,
            ),
        ]

    @classmethod
    def get_executable_path(
        cls,
        context: interfaces.context.ContextInterface,
        task: interfaces.objects.ObjectInterface,
    ) -> Optional[str]:
        """
        Extract the executable path from task_struct.mm.exe_file

        Args:
            context: The context to operate on
            task: task_struct object of the process

        Returns:
            Returns executable path or None if not available
        """

        try:
            mm = task.mm
        except (exceptions.InvalidAddressException, AttributeError) as e:
            vollog.debug(f"Unable to access mm for task at {task.vol.offset:#x}: {e}")
            return None

        if not mm or not mm.is_readable():
            # Kernel threads don't have mm struct
            return None

        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

        if not exe_file or not exe_file.is_readable():
            return None

        try:
            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

        return exe_path

    @classmethod
    def get_cmdline_basename(
        cls,
        context: interfaces.context.ContextInterface,
        task: interfaces.objects.ObjectInterface,
    ) -> Optional[str]:
        """
        Extract the command line arguments and return the basename of the first argument

        Args:
            context: The context to operate on
            task: task_struct object of the process

        Returns:
            Basename of the first command line argument or None if not available
        """
        mm = task.mm
        if not mm or not mm.is_readable():
            return None

        proc_layer_name = task.add_process_layer()
        if proc_layer_name is None:
            return None

        proc_layer = context.layers[proc_layer_name]
        start = task.mm.arg_start
        size_to_read = task.mm.arg_end - task.mm.arg_start

        if not (0 < size_to_read <= 4096):
            return None

        try:
            argv = proc_layer.read(start, size_to_read)
        except exceptions.InvalidAddressException as e:
            vollog.debug(
                f"Unable to read cmdline for task at {task.vol.offset:#x}: {e}"
            )
            return None

        # Parse the arguments - they are null byte terminated
        args_str = argv.decode(encoding="utf8", errors="replace")
        args_list = args_str.split("\x00")
        if args_list and args_list[0]:
            basename = PurePosixPath(args_list[0]).name
            return basename
        else:
            return None

    @classmethod
    def get_comm(cls, task: interfaces.objects.ObjectInterface) -> Optional[str]:
        """
        Extract the comm field from task_struct

        Args:
            task: task_struct object of the process

        Returns:
            Process name from comm field or None if not available
        """
        try:
            return utility.array_to_string(task.comm)
        except (exceptions.InvalidAddressException, AttributeError) as e:
            vollog.debug(f"Unable to read comm for task at {task.vol.offset:#x}: {e}")
            return None

    @classmethod
    def extract_process_names(
        cls,
        context: interfaces.context.ContextInterface,
        task: interfaces.objects.ObjectInterface,
    ) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[str], bool]:
        """
        Extract all process name sources for comparison

        Returns:
            Tuple of (exe_path, exe_basename, cmdline_basename, comm)
        """
        exe_path = cls.get_executable_path(context, task)
        exe_basename = PurePosixPath(exe_path).name if exe_path else None
        if exe_basename and exe_basename.endswith(" (deleted)"):
            exe_basename = exe_basename[: -len(" (deleted)")]
        cmdline_basename = cls.get_cmdline_basename(context, task)
        comm = cls.get_comm(task)

        return exe_path, exe_basename, cmdline_basename, comm

    def _detect_spoofing(
        self,
        exe_basename: Optional[str],
        cmdline_basename: Optional[str],
        comm: Optional[str],
    ) -> Tuple[bool, bool]:
        """
        Analyze the three name sources to detect potential spoofing

        Args:
            exe_basename: Basename from exe_file path
            cmdline_basename: Basename from command line
            comm: Name from comm field

        Returns:
            Tuple of (cmdline_spoofed, comm_spoofed) boolean flags
        """
        # Skip kernel threads - need at least 2 sources for comparison
        available_sources = sum(
            1 for name in [exe_basename, cmdline_basename, comm] if name
        )
        if available_sources < 2:
            return False, False

        # Check for cmdline spoofing
        cmdline_spoofed = False
        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 exe_basename and comm:
            comm_spoofed = exe_basename[:15] != comm

        return cmdline_spoofed, comm_spoofed

    def _generator(self, tasks) -> Iterator[Tuple[int, Tuple]]:
        """
        Generate process spoofing detection results

        Args:
            tasks: Iterator of task_struct objects

        Yields:
            Tuple containing process information and spoofing analysis
        """
        for task in tasks:
            try:
                pid = task.pid
                ppid = task.get_parent_pid()

                exe_path, exe_basename, cmdline_basename, comm = (
                    self.extract_process_names(self.context, task)
                )

                cmdline_spoofed, comm_spoofed = self._detect_spoofing(
                    exe_basename, cmdline_basename, comm
                )

                is_deleted = exe_path.endswith(" (deleted)") if exe_path else False

                # Convert None values to strings for TreeGrid compatibility
                exe_path_render = exe_path if exe_path else "N/A"
                cmdline_render = cmdline_basename if cmdline_basename else "N/A"
                comm_render = comm if comm else "N/A"

                yield (
                    0,
                    (
                        pid,
                        ppid,
                        exe_path_render,
                        cmdline_render,
                        comm_render,
                        cmdline_spoofed,
                        comm_spoofed,
                        is_deleted,
                    ),
                )

            except (exceptions.InvalidAddressException, AttributeError) as e:
                vollog.warning(
                    f"Unable to process task PID {getattr(task, 'pid', 'unknown')} at {task.vol.offset:#x}: {e}"
                )
                continue

    def run(self):
        filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))

        return renderers.TreeGrid(
            [
                ("PID", int),
                ("PPID", int),
                ("Exe_Path", str),
                ("Cmdline_Basename", str),
                ("Comm", str),
                ("Cmdline_Spoofed", bool),
                ("Comm_Spoofed", bool),
                ("Exe_Deleted", bool),
            ],
            self._generator(
                pslist.PsList.list_tasks(
                    self.context, self.config["kernel"], filter_func=filter_func
                )
            ),
        )