Volshell: Updates to the get_process() methods

volatility3/cli/volshell/linux.py

@@ -51,42 +51,61 @@ 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"""
+    def get_process(self, pid=None, virtaddr=None, physaddr=None):
+        """Return the task_struct object that matches the pid. If a physical or a virtual address is provided, construct the task_struct object at said address. Only one parameter is allowed.
 
-        if pid is not None and offset is not None:
+        Args:
+            pid (int, optional): PID to search for
+            virtaddr (int, optional): Virtual address to construct object at
+            physaddr (int, optional): Physical address to construct object at
+
+        Returns:
+            ObjectInterface: task_struct Object
+        """
+
+        if sum(1 if x is not None else 0 for x in [pid, virtaddr, physaddr]) != 1:
             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]
+        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]
+        kernel_layer_name = vmlinux.layer_name
+        kernel_layer = self.context.layers[kernel_layer_name]
 
-            memory_layer_name = kernel_layer.dependencies[0]
+        memory_layer_name = kernel_layer.dependencies[0]
 
-            ptask = self.context.object(
-                vmlinux.symbol_table_name + constants.BANG + "task_struct",
+        task_struct_symbol = vmlinux.symbol_table_name + constants.BANG + "task_struct"
+
+        if virtaddr is not None:
+            task = self.context.object(
+                task_struct_symbol,
+                layer_name=kernel_layer_name,
+                offset=virtaddr,
+            )
+
+        if physaddr is not None:
+            task = self.context.object(
+                task_struct_symbol,
                 layer_name=memory_layer_name,
-                offset=offset,
+                offset=physaddr,
                 native_layer_name=kernel_layer_name,
             )
 
+        if physaddr is not None or virtaddr is not None:
             try:
-                DescExitStateEnum(ptask.exit_state)
+                DescExitStateEnum(task.exit_state)
             except ValueError:
                 print(
-                    f"task_struct @ {hex(ptask.vol.offset)} as exit_state {ptask.exit_state} is likely not valid"
+                    f"task_struct @ {hex(task.vol.offset)} as exit_state {task.exit_state} is likely not valid"
                 )
 
-            if not (0 < ptask.pid < 65535):
+            if not (0 < task.pid < 65535):
                 print(
-                    f"task_struct @ {hex(ptask.vol.offset)} as pid {ptask.pid} is likely not valid"
+                    f"task_struct @ {hex(task.vol.offset)} as pid {task.pid} is likely not valid"
                 )
 
-            return ptask
+            return task
 
         if pid is not None:
             tasks = self.list_tasks()
@@ -107,7 +126,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),
+            (["gp", "get_process", "get_task"], 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,10 +44,19 @@ 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."""
+    def get_process(self, pid=None, virtaddr=None, physaddr=None):
+        """Returns the _EPROCESS object that matches the pid. If a physical or a virtual address is provided, construct the _EPROCESS object at said 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:
+        Args:
+            pid (int, optional): PID / UniqueProcessId to search for.
+            virtaddr (int, optional): Virtual address to construct object at
+            physaddr (int, optional): Physical address to construct object at
+
+        Returns:
+            ObjectInterface: _EPROCESS Object
+        """
+
+        if sum(1 if x is not None else 0 for x in [pid, virtaddr, physaddr]) != 1:
             print("Only one parameter is accepted")
             return None
 
@@ -61,20 +70,20 @@ class Volshell(generic.Volshell):
 
         eprocess_symbol = kernel.symbol_table_name + constants.BANG + "_EPROCESS"
 
-        if v_offset is not None:
+        if virtaddr is not None:
             eproc = self.context.object(
                 eprocess_symbol,
                 layer_name=kernel_layer_name,
-                offset=v_offset,
+                offset=virtaddr,
             )
 
             return eproc
 
-        if p_offset is not None:
+        if physaddr is not None:
             eproc = self.context.object(
                 eprocess_symbol,
                 layer_name=memory_layer_name,
-                offset=p_offset,
+                offset=physaddr,
                 native_layer_name=kernel_layer_name,
             )