Merge pull request #1429 from gcmoreira/linux_pslist_user_group_cred_ids

Linux pslist plugin - Add tasks real and effective user and group identifiers
This commit is contained in:
ikelos
2024-12-19 00:15:53 +00:00
committed by GitHub
22 changed files with 164 additions and 108 deletions
+2 -2
View File
@@ -22,7 +22,7 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Recovers bash command history from memory."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -33,7 +33,7 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.ListRequirement(
name="pid",
@@ -15,7 +15,7 @@ class Boottime(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface)
"""Shows the time the system was started"""
_required_framework_version = (2, 11, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -26,7 +26,7 @@ class Boottime(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface)
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
]
@@ -50,7 +50,7 @@ class Capabilities(plugins.PluginInterface):
"""Lists process capabilities"""
_required_framework_version = (2, 13, 0)
_version = (1, 1, 0)
_version = (1, 1, 1)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -61,7 +61,7 @@ class Capabilities(plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.ListRequirement(
name="pids",
@@ -12,7 +12,7 @@ class Check_creds(interfaces.plugins.PluginInterface):
"""Checks if any processes are sharing credential structures"""
_required_framework_version = (2, 0, 0)
_version = (2, 0, 1)
_version = (2, 0, 2)
@classmethod
def get_requirements(cls):
@@ -23,7 +23,7 @@ class Check_creds(interfaces.plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
]
+2 -2
View File
@@ -25,7 +25,7 @@ class Elfs(plugins.PluginInterface):
"""Lists all memory mapped ELF files for all processes."""
_required_framework_version = (2, 0, 0)
_version = (2, 0, 2)
_version = (2, 0, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -36,7 +36,7 @@ class Elfs(plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.ListRequirement(
name="pid",
@@ -30,7 +30,7 @@ class Envars(plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.ListRequirement(
name="pid",
@@ -20,7 +20,7 @@ class Kthreads(plugins.PluginInterface):
"""Enumerates kthread functions"""
_required_framework_version = (2, 11, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -34,7 +34,7 @@ class Kthreads(plugins.PluginInterface):
name="linuxutils", component=linux.LinuxUtilities, version=(2, 1, 0)
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
@@ -21,7 +21,7 @@ class LibraryList(interfaces.plugins.PluginInterface):
"""Enumerate libraries loaded into processes"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls):
@@ -32,7 +32,7 @@ class LibraryList(interfaces.plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.ListRequirement(
name="pids",
+2 -2
View File
@@ -110,7 +110,7 @@ class Lsof(plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Lists open files for each processes."""
_required_framework_version = (2, 0, 0)
_version = (2, 0, 1)
_version = (2, 0, 2)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -121,7 +121,7 @@ class Lsof(plugins.PluginInterface, timeliner.TimeLinerInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.VersionRequirement(
name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0)
@@ -18,7 +18,7 @@ class Malfind(interfaces.plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -29,7 +29,7 @@ class Malfind(interfaces.plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.ListRequirement(
name="pid",
@@ -36,7 +36,7 @@ class MountInfo(plugins.PluginInterface):
"""Lists mount points on processes mount namespaces"""
_required_framework_version = (2, 2, 0)
_version = (1, 2, 2)
_version = (1, 2, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -47,7 +47,7 @@ class MountInfo(plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.VersionRequirement(
name="linuxutils", component=linux.LinuxUtilities, version=(2, 1, 0)
@@ -3,7 +3,7 @@
#
import logging
from typing import List
from typing import List, Iterable
from volatility3.framework import renderers, interfaces, constants
from volatility3.framework.symbols import linux
@@ -19,7 +19,7 @@ class PIDHashTable(plugins.PluginInterface):
"""Enumerates processes through the PID hash table"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 2)
_version = (1, 0, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -30,7 +30,7 @@ class PIDHashTable(plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.VersionRequirement(
name="linuxutils", component=linux.LinuxUtilities, version=(2, 1, 0)
@@ -218,7 +218,7 @@ class PIDHashTable(plugins.PluginInterface):
return None
def get_tasks(self) -> interfaces.objects.ObjectInterface:
def get_tasks(self) -> Iterable[interfaces.objects.ObjectInterface]:
"""Enumerates processes through the PID hash table
Yields:
@@ -231,14 +231,16 @@ class PIDHashTable(plugins.PluginInterface):
yield from sorted(pid_func(), key=lambda t: (t.tgid, t.pid))
def _generator(
self, decorate_comm: bool = False
) -> interfaces.objects.ObjectInterface:
def _generator(self, decorate_comm: bool = False):
for task in self.get_tasks():
offset, pid, tid, ppid, name, _creation_time = (
pslist.PsList.get_task_fields(task, decorate_comm)
task_fields = pslist.PsList.get_task_fields(task, decorate_comm)
fields = (
format_hints.Hex(task_fields.offset),
task_fields.user_pid,
task_fields.user_tid,
task_fields.user_ppid,
task_fields.name,
)
fields = format_hints.Hex(offset), pid, tid, ppid, name
yield 0, fields
def run(self):
+2 -2
View File
@@ -21,7 +21,7 @@ class Maps(plugins.PluginInterface):
"""Lists all memory maps for all processes."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
MAXSIZE_DEFAULT = 1024 * 1024 * 1024 # 1 Gb
@@ -35,7 +35,7 @@ class Maps(plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.ListRequirement(
name="pid",
+2 -2
View File
@@ -15,7 +15,7 @@ class PsAux(plugins.PluginInterface):
"""Lists processes with their command line arguments"""
_required_framework_version = (2, 13, 0)
_version = (1, 1, 0)
_version = (1, 1, 1)
@classmethod
def get_requirements(cls):
@@ -27,7 +27,7 @@ class PsAux(plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.ListRequirement(
name="pid",
+61 -25
View File
@@ -2,7 +2,9 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import datetime
from typing import Any, Callable, Iterable, List, Tuple
import dataclasses
import contextlib
from typing import Any, Callable, Iterable, List, Optional
from volatility3.framework import interfaces, renderers
from volatility3.framework.configuration import requirements
@@ -14,11 +16,25 @@ from volatility3.plugins import timeliner
from volatility3.plugins.linux import elfs
@dataclasses.dataclass
class TaskFields:
offset: int
user_pid: int
user_tid: int
user_ppid: int
name: str
uid: Optional[int]
gid: Optional[int]
euid: Optional[int]
egid: Optional[int]
creation_time: Optional[datetime.datetime]
class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Lists the processes present in a particular linux memory image."""
_required_framework_version = (2, 13, 0)
_version = (3, 1, 0)
_version = (4, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -82,7 +98,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
@classmethod
def get_task_fields(
cls, task: interfaces.objects.ObjectInterface, decorate_comm: bool = False
) -> Tuple[int, int, int, int, str, datetime.datetime]:
) -> TaskFields:
"""Extract the fields needed for the final output
Args:
@@ -91,21 +107,34 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
and of Kernel threads in square brackets.
Defaults to False.
Returns:
A tuple with the fields to show in the plugin output.
A TaskFields object with the fields to show in the plugin output.
"""
pid = task.tgid
tid = task.pid
ppid = task.get_parent_pid()
name = utility.array_to_string(task.comm)
start_time = task.get_create_time()
if decorate_comm:
if task.is_kernel_thread:
name = f"[{name}]"
elif task.is_user_thread:
name = f"{{{name}}}"
task_fields = (task.vol.offset, pid, tid, ppid, name, start_time)
return task_fields
# This function may be called with a partially initialized/uninitialized task.
# Ensure it always returns a valid TaskFields object, ready for use in a plugin.
valid_cred = task.cred and task.cred.is_readable()
creation_time = None
with contextlib.suppress(Exception):
creation_time = task.get_create_time()
return TaskFields(
offset=task.vol.offset,
user_pid=task.tgid,
user_tid=task.pid,
user_ppid=task.get_parent_pid(),
name=name,
uid=task.cred.uid if valid_cred else None,
gid=task.cred.gid if valid_cred else None,
euid=task.cred.euid if valid_cred else None,
egid=task.cred.egid if valid_cred else None,
creation_time=creation_time,
)
def _get_file_output(self, task: interfaces.objects.ObjectInterface) -> str:
"""Extract the elf for the process if requested
@@ -179,17 +208,19 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
else:
file_output = "Disabled"
offset, pid, tid, ppid, name, creation_time = self.get_task_fields(
task, decorate_comm
)
task_fields = self.get_task_fields(task, decorate_comm)
yield 0, (
format_hints.Hex(offset),
pid,
tid,
ppid,
name,
creation_time or renderers.NotAvailableValue(),
format_hints.Hex(task_fields.offset),
task_fields.user_pid,
task_fields.user_tid,
task_fields.user_ppid,
task_fields.name,
task_fields.uid or renderers.NotAvailableValue(),
task_fields.gid or renderers.NotAvailableValue(),
task_fields.euid or renderers.NotAvailableValue(),
task_fields.egid or renderers.NotAvailableValue(),
task_fields.creation_time or renderers.NotAvailableValue(),
file_output,
)
@@ -238,6 +269,10 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
("TID", int),
("PPID", int),
("COMM", str),
("UID", int),
("GID", int),
("EUID", int),
("EGID", int),
("CREATION TIME", datetime.datetime),
("File output", str),
]
@@ -251,10 +286,11 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
for task in self.list_tasks(
self.context, self.config["kernel"], filter_func, include_threads=True
):
offset, user_pid, user_tid, _user_ppid, name, creation_time = (
self.get_task_fields(task)
task_fields = self.get_task_fields(task)
description = f"Process {task_fields.user_pid}/{task_fields.user_tid} {task_fields.name} ({task_fields.offset})"
yield (
description,
timeliner.TimeLinerType.CREATED,
task_fields.creation_time,
)
description = f"Process {user_pid}/{user_tid} {name} ({offset})"
yield (description, timeliner.TimeLinerType.CREATED, creation_time)
+18 -31
View File
@@ -2,15 +2,15 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
from typing import Iterable, List, Tuple
from typing import Iterable, List
import struct
from enum import Enum
from volatility3.framework import renderers, interfaces, symbols, constants, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.objects import utility
from volatility3.framework.layers import scanners
from volatility3.framework.renderers import format_hints
from volatility3.plugins.linux import pslist
vollog = logging.getLogger(__name__)
@@ -28,7 +28,7 @@ class PsScan(interfaces.plugins.PluginInterface):
"""Scans for processes present in a particular linux image."""
_required_framework_version = (2, 13, 0)
_version = (1, 1, 0)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -38,34 +38,11 @@ class PsScan(interfaces.plugins.PluginInterface):
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
]
def _get_task_fields(
self, task: interfaces.objects.ObjectInterface
) -> Tuple[int, int, int, str, str]:
"""Extract the fields needed for the final output
Args:
task: A task object from where to get the fields.
Returns:
A tuple with the fields to show in the plugin output.
"""
pid = task.tgid
tid = task.pid
ppid = task.get_parent_pid()
name = utility.array_to_string(task.comm)
exit_state = DescExitStateEnum(task.exit_state).name
task_fields = (
format_hints.Hex(task.vol.offset),
pid,
tid,
ppid,
name,
exit_state,
)
return task_fields
def _generator(self):
"""Generates the tasks found from scanning."""
@@ -75,8 +52,18 @@ class PsScan(interfaces.plugins.PluginInterface):
for task in self.scan_tasks(
self.context, vmlinux_module_name, vmlinux.layer_name
):
row = self._get_task_fields(task)
yield (0, row)
task_fields = pslist.PsList.get_task_fields(task)
exit_state = DescExitStateEnum(task.exit_state).name
fields = (
format_hints.Hex(task_fields.offset),
task_fields.user_pid,
task_fields.user_tid,
task_fields.user_ppid,
task_fields.name,
exit_state,
)
yield (0, fields)
@classmethod
def scan_tasks(
+11 -7
View File
@@ -13,7 +13,7 @@ class PsTree(interfaces.plugins.PluginInterface):
ID."""
_required_framework_version = (2, 13, 0)
_version = (1, 1, 0)
_version = (1, 1, 1)
@classmethod
def get_requirements(cls):
@@ -25,7 +25,7 @@ class PsTree(interfaces.plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.ListRequirement(
name="pid",
@@ -101,13 +101,17 @@ class PsTree(interfaces.plugins.PluginInterface):
def yield_processes(pid):
task = self._tasks[pid]
offset, pid, tid, ppid, name, _creation_time = (
pslist.PsList.get_task_fields(task, decorate_comm)
task_fields = pslist.PsList.get_task_fields(task, decorate_comm)
fields = (
format_hints.Hex(task_fields.offset),
task_fields.user_pid,
task_fields.user_tid,
task_fields.user_ppid,
task_fields.name,
)
fields = format_hints.Hex(offset), pid, tid, ppid, name
yield (self._levels[tid] - 1, fields)
yield (self._levels[task_fields.user_tid] - 1, fields)
for child_pid in sorted(self._children.get(tid, [])):
for child_pid in sorted(self._children.get(task_fields.user_tid, [])):
yield from yield_processes(child_pid)
for pid, level in self._levels.items():
@@ -19,7 +19,7 @@ class Ptrace(plugins.PluginInterface):
"""Enumerates ptrace's tracer and tracee tasks"""
_required_framework_version = (2, 10, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -30,7 +30,7 @@ class Ptrace(plugins.PluginInterface):
architectures=architectures.LINUX_ARCHS,
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
]
@@ -438,7 +438,7 @@ class Sockstat(plugins.PluginInterface):
"""Lists all network connections for all processes."""
_required_framework_version = (2, 0, 0)
_version = (3, 0, 1)
_version = (3, 0, 2)
@classmethod
def get_requirements(cls):
@@ -455,7 +455,7 @@ class Sockstat(plugins.PluginInterface):
name="lsof", plugin=lsof.Lsof, version=(2, 0, 0)
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.VersionRequirement(
name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0)
@@ -21,7 +21,7 @@ class VmaRegExScan(plugins.PluginInterface):
"""Scans all virtual memory areas for tasks using RegEx."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
MAXSIZE_DEFAULT = 128
@@ -35,7 +35,7 @@ class VmaRegExScan(plugins.PluginInterface):
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.ListRequirement(
name="pid",
@@ -18,7 +18,7 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
"""Scans all virtual memory areas for tasks using yara."""
_required_framework_version = (2, 4, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -31,7 +31,7 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
optional=True,
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.PluginRequirement(
name="yarascan", plugin=yarascan.YaraScan, version=(2, 0, 0)
@@ -2079,13 +2079,40 @@ class cred(objects.StructType):
return int(value)
@property
def euid(self):
def uid(self) -> int:
"""Returns the real user ID
Returns:
The real user ID value
"""
return self._get_cred_int_value("uid")
@property
def gid(self) -> int:
"""Returns the real user ID
Returns:
The real user ID value
"""
return self._get_cred_int_value("gid")
@property
def euid(self) -> int:
"""Returns the effective user ID
Returns:
The effective user ID value
"""
return self._get_cred_int_value("euid")
@property
def egid(self) -> int:
"""Returns the effective group ID
Returns:
int: the effective user ID value
"""
return self._get_cred_int_value("euid")
return self._get_cred_int_value("egid")
class kernel_cap_struct(objects.StructType):