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: