From 3d9efc3b3466a91dcbaf5aacfbc302b816bba316 Mon Sep 17 00:00:00 2001 From: Gustavo Moreira Date: Wed, 17 May 2023 14:14:33 +0100 Subject: [PATCH] Added linux capabilities plugin --- .../framework/constants/linux/__init__.py | 45 ++++ .../framework/plugins/linux/capabilities.py | 218 ++++++++++++++++++ .../framework/symbols/linux/__init__.py | 2 + .../symbols/linux/extensions/__init__.py | 108 ++++++++- 4 files changed, 372 insertions(+), 1 deletion(-) create mode 100644 volatility3/framework/plugins/linux/capabilities.py diff --git a/volatility3/framework/constants/linux/__init__.py b/volatility3/framework/constants/linux/__init__.py index 1b133eb42..a802e0ada 100644 --- a/volatility3/framework/constants/linux/__init__.py +++ b/volatility3/framework/constants/linux/__init__.py @@ -234,3 +234,48 @@ 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", +) diff --git a/volatility3/framework/plugins/linux/capabilities.py b/volatility3/framework/plugins/linux/capabilities.py new file mode 100644 index 000000000..8bcd80eef --- /dev/null +++ b/volatility3/framework/plugins/linux/capabilities.py @@ -0,0 +1,218 @@ +# 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 typing import Iterable, List, Tuple, Dict + +from volatility3.framework import interfaces, renderers +from volatility3.framework.configuration import requirements +from volatility3.framework.interfaces import plugins +from volatility3.framework.objects import utility +from volatility3.framework.renderers import format_hints +from volatility3.framework.symbols.linux import extensions +from volatility3.plugins.linux import pslist + +vollog = logging.getLogger(__name__) + + +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, + ), + requirements.BooleanRequirement( + name="inheritable", + description="Show only inheritable capabilities in human-readable strings.", + optional=True, + ), + requirements.BooleanRequirement( + name="permitted", + description="Show only permitted capabilities in human-readable strings.", + optional=True, + ), + requirements.BooleanRequirement( + name="effective", + description="Show only effective capabilities in human-readable strings.", + optional=True, + ), + requirements.BooleanRequirement( + name="bounding", + description="Show only bounding capabilities in human-readable strings.", + optional=True, + ), + requirements.BooleanRequirement( + name="ambient", + description="Show only ambient capabilities in human-readable strings.", + optional=True, + ), + ] + + def _check_capabilities_support(self): + """Checks that the framework supports at least as much capabilities as + the kernel being analysed. Otherwise, it shows a warning for the + developers. + """ + vmlinux = self.context.modules[self.config["kernel"]] + + kernel_cap_last_cap = vmlinux.object(object_type="int", offset=kernel_cap_last_cap) + 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 cap_value == 0: + return "-" + + CAP_FULL = 0xffffffff + if cap_value == CAP_FULL: + return "all" + + return ", ".join(cap.enumerate_capabilities()) + + @classmethod + def get_task_capabilities(cls, task: interfaces.objects.ObjectInterface) -> Dict: + """Returns a dict with the task basic information along with its capabilities + + Args: + task: A task object from where to get the fields. + + Returns: + dict: A dict with the task basic information along with its capabilities + """ + task_cred = task.real_cred + fields = { + "common": [ + utility.array_to_string(task.comm), + int(task.pid), + int(task.tgid), + int(task.parent.pid), + int(task.cred.euid), + ], + "capabilities": [ + task_cred.cap_inheritable, + task_cred.cap_permitted, + task_cred.cap_effective, + task_cred.cap_bset, + ] + } + + # Ambient capabilities were added in kernels 4.3.6 + if task_cred.has_member("cap_ambient"): + fields["capabilities"].append(task_cred.cap_ambient) + else: + fields["capabilities"].append(renderers.NotAvailableValue()) + + return fields + + def get_tasks_capabilities(self, tasks: List[interfaces.objects.ObjectInterface]) -> Iterable[Dict]: + """Yields a dict for each task containing the task's basic information along with its capabilities + + Args: + tasks: An iterable with the tasks to process. + + Yields: + Iterable[Dict]: A dict for each task containing the task's basic information along with its capabilities + """ + for task in tasks: + if task.is_kernel_thread: + continue + + yield self.get_task_capabilities(task) + + def _generator(self, tasks: Iterable[interfaces.objects.ObjectInterface]) -> Iterable[Tuple[int, Tuple]]: + for fields in self.get_tasks_capabilities(tasks): + selected_fields = fields["common"] + cap_inh, cap_prm, cap_eff, cap_bnd, cap_amb = fields["capabilities"] + + if self.config.get("inheritable"): + selected_fields.append(self._decode_cap(cap_inh)) + elif self.config.get("permitted"): + selected_fields.append(self._decode_cap(cap_prm)) + elif self.config.get("effective"): + selected_fields.append(self._decode_cap(cap_eff)) + elif self.config.get("bounding"): + selected_fields.append(self._decode_cap(cap_bnd)) + elif self.config.get("ambient"): + selected_fields.append(self._decode_cap(cap_amb)) + else: + # Raw values + selected_fields.append(format_hints.Hex(cap_inh.get_capabilities())) + selected_fields.append(format_hints.Hex(cap_prm.get_capabilities())) + selected_fields.append(format_hints.Hex(cap_eff.get_capabilities())) + selected_fields.append(format_hints.Hex(cap_bnd.get_capabilities())) + + # Ambient capabilities were added in kernels 4.3.6 + if isinstance(cap_amb, renderers.NotAvailableValue): + selected_fields.append(cap_amb) + else: + selected_fields.append(format_hints.Hex(cap_amb.get_capabilities())) + + yield 0, selected_fields + + def run(self): + 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), + ] + + if self.config.get("inheritable"): + columns.append(("cap_inheritable", str)) + elif self.config.get("permitted"): + columns.append(("cap_permitted", str)) + elif self.config.get("effective"): + columns.append(("cap_effective", str)) + elif self.config.get("bounding"): + columns.append(("cap_bounding", str)) + elif self.config.get("ambient"): + columns.append(("cap_ambient", str)) + else: + columns.append(("cap_inheritable", format_hints.Hex)) + columns.append(("cap_permitted", format_hints.Hex)) + columns.append(("cap_effective", format_hints.Hex)) + columns.append(("cap_bounding", format_hints.Hex)) + columns.append(("cap_ambient", format_hints.Hex)) + + return renderers.TreeGrid(columns, self._generator(tasks)) diff --git a/volatility3/framework/symbols/linux/__init__.py b/volatility3/framework/symbols/linux/__init__.py index 36314b2c6..0bab9dedf 100644 --- a/volatility3/framework/symbols/linux/__init__.py +++ b/volatility3/framework/symbols/linux/__init__.py @@ -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) diff --git a/volatility3/framework/symbols/linux/extensions/__init__.py b/volatility3/framework/symbols/linux/extensions/__init__.py index 9ac98b5da..8a8785bd0 100644 --- a/volatility3/framework/symbols/linux/extensions/__init__.py +++ b/volatility3/framework/symbols/linux/extensions/__init__.py @@ -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 @@ -1428,3 +1428,109 @@ 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(constants.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(constants.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 int(cap_value & 0xffffffff) + + 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 fiven capability is unknown to the framework. + + Returns: + bool: "True" if the given capability is enabled. + """ + if capability not in constants.CAPABILITIES: + raise AttributeError(f"Unknown capability with name '{capability}'") + + cap_value = 1 << constants.CAPABILITIES.index(capability) + return cap_value & self.get_capabilities() != 0