From dfe3d255c064b9c78edf4f5f58eff6c15cc56486 Mon Sep 17 00:00:00 2001 From: Odysseas Stavrou Date: Mon, 20 Jan 2025 22:25:01 +0200 Subject: [PATCH 1/2] Volshell: Update Process retrieval methods with virtual/physical offsets --- volatility3/cli/volshell/linux.py | 58 +++++++++++++++++++++++++++++ volatility3/cli/volshell/windows.py | 47 +++++++++++++++++++++++ 2 files changed, 105 insertions(+) diff --git a/volatility3/cli/volshell/linux.py b/volatility3/cli/volshell/linux.py index cc58fa1c2..9ea3ea1f5 100644 --- a/volatility3/cli/volshell/linux.py +++ b/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: diff --git a/volatility3/cli/volshell/windows.py b/volatility3/cli/volshell/windows.py index 303d4d5c3..a77392561 100644 --- a/volatility3/cli/volshell/windows.py +++ b/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: From b59f051353cd58b7d6e4bfda2f07746820f4f32a Mon Sep 17 00:00:00 2001 From: Odysseas Stavrou Date: Wed, 22 Jan 2025 02:39:35 +0200 Subject: [PATCH 2/2] Volshell: Updates to the get_process() methods --- volatility3/cli/volshell/linux.py | 55 +++++++++++++++++++---------- volatility3/cli/volshell/windows.py | 23 ++++++++---- 2 files changed, 53 insertions(+), 25 deletions(-) diff --git a/volatility3/cli/volshell/linux.py b/volatility3/cli/volshell/linux.py index 9ea3ea1f5..b3689c3ae 100644 --- a/volatility3/cli/volshell/linux.py +++ b/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: diff --git a/volatility3/cli/volshell/windows.py b/volatility3/cli/volshell/windows.py index a77392561..9b89a8b81 100644 --- a/volatility3/cli/volshell/windows.py +++ b/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, )