WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-13 13:17:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #957 from gcmoreira/linux_capabilities

Browse Source 
Linux capabilities plugin
This commit is contained in:
ikelos
2023-07-10 07:49:44 +01:00
committed by GitHub
parent 239e164ce6 42fe37ed52
commit 74e8582360
4 changed files with 367 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/framework/constants/linux/__init__.py
+47
View File
@@ -234,3 +234,50 @@ BLUETOOTH_PROTOCOLS = (
"HIDP",
"AVDTP",
)
# Ref: include/uapi/linux/capability.h
CAPABILITIES = (
"chown",
"dac_override",
"dac_read_search",
"fowner",
"fsetid",
"kill",
"setgid",
"setuid",
"setpcap",
"linux_immutable",
"net_bind_service",
"net_broadcast",
"net_admin",
"net_raw",
"ipc_lock",
"ipc_owner",
"sys_module",
"sys_rawio",
"sys_chroot",
"sys_ptrace",
"sys_pacct",
"sys_admin",
"sys_boot",
"sys_nice",
"sys_resource",
"sys_time",
"sys_tty_config",
"mknod",
"lease",
"audit_write",
"audit_control",
"setfcap",
"mac_override",
"mac_admin",
"syslog",
"wake_alarm",
"block_suspend",
"audit_read",
"perfmon",
"bpf",
"checkpoint_restore",
)
CAP_FULL = 0xFFFFFFFF
volatility3/framework/plugins/linux/capabilities.py
+209
View File
@@ -0,0 +1,209 @@
# This file is Copyright 2023 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
from dataclasses import dataclass, astuple, fields
from typing import Iterable, List, Tuple
from volatility3.framework import interfaces, renderers, exceptions
from volatility3.framework.constants.linux import CAP_FULL
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.objects import utility
from volatility3.framework.symbols.linux import extensions
from volatility3.plugins.linux import pslist
vollog = logging.getLogger(__name__)
@dataclass
class TaskData:
"""Stores basic information about a task"""
comm: str
pid: int
tgid: int
ppid: int
euid: int
@dataclass
class CapabilitiesData:
"""Stores each set of capabilties for a task"""
cap_inheritable: interfaces.objects.ObjectInterface
cap_permitted: interfaces.objects.ObjectInterface
cap_effective: interfaces.objects.ObjectInterface
cap_bset: interfaces.objects.ObjectInterface
cap_ambient: interfaces.objects.ObjectInterface
def astuple(self) -> Tuple:
"""Returns a shallow copy of the capability sets in a tuple.
Otherwise, when dataclasses.astuple() performs a deep-copy recursion on
ObjectInterface will take a substantial amount of time.
"""
return tuple(getattr(self, field.name) for field in fields(self))
class Capabilities(plugins.PluginInterface):
"""Lists process capabilities"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
),
requirements.ListRequirement(
name="pids",
description="Filter on specific process IDs.",
element_type=int,
optional=True,
),
]
def _check_capabilities_support(
self,
context: interfaces.context.ContextInterface,
vmlinux_module_name: str,
):
"""Checks that the framework supports at least as much capabilities as
the kernel being analysed. Otherwise, it shows a warning for the
developers.
"""
vmlinux = context.modules[vmlinux_module_name]
try:
kernel_cap_last_cap = vmlinux.object_from_symbol(symbol_name="cap_last_cap")
except exceptions.SymbolError:
# It should be a kernel < 3.2
return
vol2_last_cap = extensions.kernel_cap_struct.get_last_cap_value()
if kernel_cap_last_cap > vol2_last_cap:
vollog.warning(
"Developers: The supported Linux capabilities of this plugin are outdated for this kernel"
)
@staticmethod
def _decode_cap(cap: interfaces.objects.ObjectInterface) -> str:
"""Returns a textual representation of the capability set.
The format is a comma-separated list of capabilitites. In order to
summarize the output and if all the capabilities are enabled, instead of
the individual capabilities, the special name "all" will be shown.
Args:
cap: Kernel capability object. Usually a 'kernel_cap_struct' struct
Returns:
str: A string with a comma separated list of decoded capabilities
"""
if isinstance(cap, renderers.NotAvailableValue):
return cap
cap_value = cap.get_capabilities()
if not cap_value:
return ""
if cap_value == CAP_FULL:
return "all"
return ", ".join(cap.enumerate_capabilities())
@classmethod
def get_task_capabilities(
cls, task: interfaces.objects.ObjectInterface
) -> Tuple[TaskData, CapabilitiesData]:
"""Returns a tuple with the task basic information along with its capabilities
Args:
task: A task object from where to get the fields.
Returns:
A tuple with the task basic information and its capabilities
"""
task_data = TaskData(
comm=utility.array_to_string(task.comm),
pid=int(task.pid),
tgid=int(task.tgid),
ppid=int(task.parent.pid),
euid=int(task.cred.euid),
)
task_cred = task.real_cred
capabilities_data = CapabilitiesData(
cap_inheritable=task_cred.cap_inheritable,
cap_permitted=task_cred.cap_permitted,
cap_effective=task_cred.cap_effective,
cap_bset=task_cred.cap_bset,
cap_ambient=renderers.NotAvailableValue(),
)
# Ambient capabilities were added in kernels 4.3.6
if task_cred.has_member("cap_ambient"):
capabilities_data.cap_ambient = task_cred.cap_ambient
return task_data, capabilities_data
@classmethod
def get_tasks_capabilities(
cls, tasks: List[interfaces.objects.ObjectInterface]
) -> Iterable[Tuple[TaskData, CapabilitiesData]]:
"""Yields a tuple for each task containing the task's basic information along with its capabilities
Args:
tasks: An iterable with the tasks to process.
Yields:
A tuple for each task containing the task's basic information and its capabilities
"""
for task in tasks:
yield cls.get_task_capabilities(task)
def _generator(
self, tasks: Iterable[interfaces.objects.ObjectInterface]
) -> Iterable[Tuple[int, Tuple]]:
for task_fields, capabilities_fields in self.get_tasks_capabilities(tasks):
task_fields = astuple(task_fields)
capabilities_text = tuple(
self._decode_cap(cap) for cap in capabilities_fields.astuple()
)
yield 0, task_fields + capabilities_text
def run(self):
self._check_capabilities_support(self.context, self.config["kernel"])
pids = self.config.get("pids")
pid_filter = pslist.PsList.create_pid_filter(pids)
tasks = pslist.PsList.list_tasks(
self.context, self.config["kernel"], filter_func=pid_filter
)
columns = [
("Name", str),
("Tid", int),
("Pid", int),
("PPid", int),
("EUID", int),
("cap_inheritable", str),
("cap_permitted", str),
("cap_effective", str),
("cap_bounding", str),
("cap_ambient", str),
]
return renderers.TreeGrid(columns, self._generator(tasks))
volatility3/framework/symbols/linux/__init__.py
+2
View File
@@ -28,6 +28,8 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
self.set_type_class("fs_struct", extensions.fs_struct)
self.set_type_class("files_struct", extensions.files_struct)
self.set_type_class("kobject", extensions.kobject)
self.set_type_class("cred", extensions.cred)
self.set_type_class("kernel_cap_struct", extensions.kernel_cap_struct)
# Might not exist in the current symbols
self.optional_set_type_class("module", extensions.module)
self.optional_set_type_class("bpf_prog", extensions.bpf_prog)
volatility3/framework/symbols/linux/extensions/__init__.py
+109 -1
View File
@@ -5,7 +5,7 @@
import collections.abc
import logging
import socket as socket_module
from typing import Generator, Iterable, Iterator, Optional, Tuple
from typing import Generator, Iterable, Iterator, Optional, Tuple, List
from volatility3.framework import constants
from volatility3.framework.constants.linux import SOCK_TYPES, SOCK_FAMILY
@@ -13,6 +13,7 @@ from volatility3.framework.constants.linux import IP_PROTOCOLS, IPV6_PROTOCOLS
from volatility3.framework.constants.linux import TCP_STATES, NETLINK_PROTOCOLS
from volatility3.framework.constants.linux import ETH_PROTOCOLS, BLUETOOTH_STATES
from volatility3.framework.constants.linux import BLUETOOTH_PROTOCOLS, SOCKET_STATES
from volatility3.framework.constants.linux import CAPABILITIES, CAP_FULL
from volatility3.framework import exceptions, objects, interfaces, symbols
from volatility3.framework.layers import linear
from volatility3.framework.objects import utility
@@ -1428,3 +1429,110 @@ class bpf_prog(objects.StructType):
# kernel < 3.18.140
raise AttributeError("Unable to find the BPF type")
class cred(objects.StructType):
# struct cred was added in kernels 2.6.29
def _get_cred_int_value(self, member: str) -> int:
"""Helper to obtain the right cred member value for the current kernel.
Args:
member (str): The requested cred member name to obtain its value
Raises:
AttributeError: When the requested cred member doesn't exist
AttributeError: When the cred implementation is not supported.
Returns:
int: The cred member value
"""
if not self.has_member(member):
raise AttributeError(f"struct cred doesn't have a '{member}' member")
cred_val = self.member(member)
if hasattr(cred_val, "val"):
# From kernels 3.5.7 on it is a 'kuid_t' type
value = cred_val.val
elif isinstance(cred_val, objects.Integer):
# From at least 2.6.30 and until 3.5.7 it was a 'uid_t' type which was an 'unsigned int'
value = cred_val
else:
raise AttributeError("Kernel struct cred is not supported")
return int(value)
@property
def euid(self):
"""Returns the effective user ID
Returns:
int: the effective user ID value
"""
return self._get_cred_int_value("euid")
class kernel_cap_struct(objects.StructType):
# struct kernel_cap_struct was added in kernels 2.5.0
@classmethod
def get_last_cap_value(cls) -> int:
"""Returns the latest capability ID supported by the framework.
Returns:
int: The latest supported capability ID supported by the framework.
"""
return len(CAPABILITIES) - 1
@classmethod
def capabilities_to_string(cls, capabilities_bitfield: int) -> List[str]:
"""Translates a capability bitfield to a list of capability strings.
Args:
capabilities_bitfield (int): The capability bitfield value.
Returns:
List[str]: A list of capability strings.
"""
capabilities = []
for bit, name in enumerate(CAPABILITIES):
if capabilities_bitfield & (1 << bit) != 0:
capabilities.append(name)
return capabilities
def get_capabilities(self) -> int:
"""Returns the capability bitfield value
Returns:
int: The capability bitfield value.
"""
# In kernels 2.6.25.20 the kernel_cap_struct::cap became and array
cap_value = self.cap[0] if isinstance(self.cap, objects.Array) else self.cap
return cap_value & CAP_FULL
def enumerate_capabilities(self) -> List[str]:
"""Returns the list of capability strings.
Returns:
List[str]: The list of capability strings.
"""
capabilities_value = self.get_capabilities()
return self.capabilities_to_string(capabilities_value)
def has_capability(self, capability: str) -> bool:
"""Checks if the given capability string is enabled.
Args:
capability (str): A string representing the capability i.e. dac_read_search
Raises:
AttributeError: If the given capability is unknown to the framework.
Returns:
bool: "True" if the given capability is enabled.
"""
if capability not in CAPABILITIES:
raise AttributeError(f"Unknown capability with name '{capability}'")
cap_value = 1 << CAPABILITIES.index(capability)
return cap_value & self.get_capabilities() != 0