Volshell: Update Process retrieval methods with virtual/physical offsets

volatility3/cli/volshell/linux.py

@@ -3,6 +3,7 @@
 #
 
 from typing import Any, List, Optional, Tuple, Union
+from enum import Enum
 
 from volatility3.cli.volshell import generic
 from volatility3.framework import constants, interfaces
@@ -10,6 +11,16 @@ from volatility3.framework.configuration import requirements
 from volatility3.plugins.linux import pslist
 
 
+# Could import the enum from psscan.py to avoid code duplication
+class DescExitStateEnum(Enum):
+    """Enum for linux task exit_state as defined in include/linux/sched.h"""
+
+    TASK_RUNNING = 0x00000000
+    EXIT_DEAD = 0x00000010
+    EXIT_ZOMBIE = 0x00000020
+    EXIT_TRACE = EXIT_ZOMBIE | EXIT_DEAD
+
+
 class Volshell(generic.Volshell):
     """Shell environment to directly interact with a linux memory image."""
 
@@ -40,6 +51,52 @@ class Volshell(generic.Volshell):
                 return None
         print(f"No task with task ID {pid} found")
 
+    def get_process(self, pid=None, offset=None):
+        """Get Task based on a process ID. Does not retrieve the layer, to change layer use the .pid attribute. The offset argument can be used both for physical or virtual offsets"""
+
+        if pid is not None and offset is not None:
+            print("Only one parameter is accepted")
+            return None
+
+        if offset is not None:
+            vmlinux_module_name = self.config["kernel"]
+            vmlinux = self.context.modules[vmlinux_module_name]
+
+            kernel_layer_name = vmlinux.layer_name
+            kernel_layer = self.context.layers[kernel_layer_name]
+
+            memory_layer_name = kernel_layer.dependencies[0]
+
+            ptask = self.context.object(
+                vmlinux.symbol_table_name + constants.BANG + "task_struct",
+                layer_name=memory_layer_name,
+                offset=offset,
+                native_layer_name=kernel_layer_name,
+            )
+
+            try:
+                DescExitStateEnum(ptask.exit_state)
+            except ValueError:
+                print(
+                    f"task_struct @ {hex(ptask.vol.offset)} as exit_state {ptask.exit_state} is likely not valid"
+                )
+
+            if not (0 < ptask.pid < 65535):
+                print(
+                    f"task_struct @ {hex(ptask.vol.offset)} as pid {ptask.pid} is likely not valid"
+                )
+
+            return ptask
+
+        if pid is not None:
+            tasks = self.list_tasks()
+            for task in tasks:
+                if task.pid == pid:
+                    return task
+            print(f"No task with task ID {pid} found")
+
+        return None
+
     def list_tasks(self):
         """Returns a list of task objects from the primary layer"""
         # We always use the main kernel memory and associated symbols
@@ -50,6 +107,7 @@ class Volshell(generic.Volshell):
         result += [
             (["ct", "change_task", "cp"], self.change_task),
             (["lt", "list_tasks", "ps"], self.list_tasks),
+            (["gp", "get_process"], self.get_process),
             (["symbols"], self.context.symbol_space[self.current_symbol_table]),
         ]
         if self.config.get("pid", None) is not None:

volatility3/cli/volshell/windows.py

@@ -44,11 +44,58 @@ class Volshell(generic.Volshell):
             )
         )
 
+    def get_process(self, pid=None, v_offset=None, p_offset=None):
+        """Returns the EPROCESS object that matches the pid. If v_offset/p_offset is provided, construct the EPROCESS object at the provided address. Only one parameter is allowed."""
+
+        if sum(1 if x is not None else 0 for x in [pid, v_offset, p_offset]) != 1:
+            print("Only one parameter is accepted")
+            return None
+
+        kernel_name = self.config["kernel"]
+        kernel = self.context.modules[kernel_name]
+
+        kernel_layer_name = kernel.layer_name
+
+        kernel_layer = self.context.layers[kernel_layer_name]
+        memory_layer_name = kernel_layer.dependencies[0]
+
+        eprocess_symbol = kernel.symbol_table_name + constants.BANG + "_EPROCESS"
+
+        if v_offset is not None:
+            eproc = self.context.object(
+                eprocess_symbol,
+                layer_name=kernel_layer_name,
+                offset=v_offset,
+            )
+
+            return eproc
+
+        if p_offset is not None:
+            eproc = self.context.object(
+                eprocess_symbol,
+                layer_name=memory_layer_name,
+                offset=p_offset,
+                native_layer_name=kernel_layer_name,
+            )
+
+            return eproc
+
+        if pid is not None:
+            processes = self.list_processes()
+            for process in processes:
+                if process.UniqueProcessId == pid:
+                    return process
+            print(f"No process with process ID {pid} found")
+            return None
+
+        return None
+
     def construct_locals(self) -> List[Tuple[List[str], Any]]:
         result = super().construct_locals()
         result += [
             (["cp", "change_process"], self.change_process),
             (["lp", "list_processes", "ps"], self.list_processes),
+            (["gp", "get_process"], self.get_process),
             (["symbols"], self.context.symbol_space[self.current_symbol_table]),
         ]
         if self.config.get("pid", None) is not None: