From 3d9efc3b3466a91dcbaf5aacfbc302b816bba316 Mon Sep 17 00:00:00 2001 From: Gustavo Moreira Date: Wed, 17 May 2023 14:14:33 +0100 Subject: [PATCH 1/8] 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 From d5d318d3cfd43f4d33fdb9252b120b5ed8666a12 Mon Sep 17 00:00:00 2001 From: Gustavo Moreira Date: Wed, 17 May 2023 15:26:24 +0100 Subject: [PATCH 2/8] Fix wrong constants module --- .../framework/symbols/linux/extensions/__init__.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/volatility3/framework/symbols/linux/extensions/__init__.py b/volatility3/framework/symbols/linux/extensions/__init__.py index 8a8785bd0..f626db434 100644 --- a/volatility3/framework/symbols/linux/extensions/__init__.py +++ b/volatility3/framework/symbols/linux/extensions/__init__.py @@ -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 from volatility3.framework import exceptions, objects, interfaces, symbols from volatility3.framework.layers import linear from volatility3.framework.objects import utility @@ -1478,7 +1479,7 @@ class kernel_cap_struct(objects.StructType): Returns: int: The latest supported capability ID supported by the framework. """ - return len(constants.CAPABILITIES) - 1 + return len(CAPABILITIES) - 1 @classmethod def capabilities_to_string(cls, capabilities_bitfield: int) -> List[str]: @@ -1492,7 +1493,7 @@ class kernel_cap_struct(objects.StructType): """ capabilities = [] - for bit, name in enumerate(constants.CAPABILITIES): + for bit, name in enumerate(CAPABILITIES): if capabilities_bitfield & (1 << bit) != 0: capabilities.append(name) @@ -1529,8 +1530,8 @@ class kernel_cap_struct(objects.StructType): Returns: bool: "True" if the given capability is enabled. """ - if capability not in constants.CAPABILITIES: + if capability not in CAPABILITIES: raise AttributeError(f"Unknown capability with name '{capability}'") - cap_value = 1 << constants.CAPABILITIES.index(capability) + cap_value = 1 << CAPABILITIES.index(capability) return cap_value & self.get_capabilities() != 0 From 4f5ca6116c73c456a9e55c56a41882b5bcccf7ac Mon Sep 17 00:00:00 2001 From: Gustavo Moreira Date: Wed, 17 May 2023 16:07:56 +0100 Subject: [PATCH 3/8] Add black recommendations --- .../framework/plugins/linux/capabilities.py | 24 +++++++++++++------ .../symbols/linux/extensions/__init__.py | 3 ++- 2 files changed, 19 insertions(+), 8 deletions(-) diff --git a/volatility3/framework/plugins/linux/capabilities.py b/volatility3/framework/plugins/linux/capabilities.py index 8bcd80eef..ed08c4c05 100644 --- a/volatility3/framework/plugins/linux/capabilities.py +++ b/volatility3/framework/plugins/linux/capabilities.py @@ -74,10 +74,14 @@ class Capabilities(plugins.PluginInterface): """ vmlinux = self.context.modules[self.config["kernel"]] - kernel_cap_last_cap = vmlinux.object(object_type="int", offset=kernel_cap_last_cap) + 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") + vollog.warning( + "Developers: The supported Linux capabilities of this plugin are outdated for this kernel" + ) @staticmethod def _decode_cap(cap: interfaces.objects.ObjectInterface) -> str: @@ -99,7 +103,7 @@ class Capabilities(plugins.PluginInterface): if cap_value == 0: return "-" - CAP_FULL = 0xffffffff + CAP_FULL = 0xFFFFFFFF if cap_value == CAP_FULL: return "all" @@ -129,7 +133,7 @@ class Capabilities(plugins.PluginInterface): task_cred.cap_permitted, task_cred.cap_effective, task_cred.cap_bset, - ] + ], } # Ambient capabilities were added in kernels 4.3.6 @@ -140,7 +144,9 @@ class Capabilities(plugins.PluginInterface): return fields - def get_tasks_capabilities(self, tasks: List[interfaces.objects.ObjectInterface]) -> Iterable[Dict]: + 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: @@ -155,7 +161,9 @@ class Capabilities(plugins.PluginInterface): yield self.get_task_capabilities(task) - def _generator(self, tasks: Iterable[interfaces.objects.ObjectInterface]) -> Iterable[Tuple[int, Tuple]]: + 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"] @@ -188,7 +196,9 @@ class Capabilities(plugins.PluginInterface): 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) + tasks = pslist.PsList.list_tasks( + self.context, self.config["kernel"], filter_func=pid_filter + ) columns = [ ("Name", str), diff --git a/volatility3/framework/symbols/linux/extensions/__init__.py b/volatility3/framework/symbols/linux/extensions/__init__.py index f626db434..0051cdc2f 100644 --- a/volatility3/framework/symbols/linux/extensions/__init__.py +++ b/volatility3/framework/symbols/linux/extensions/__init__.py @@ -1430,6 +1430,7 @@ 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: @@ -1507,7 +1508,7 @@ class kernel_cap_struct(objects.StructType): """ # 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) + return int(cap_value & 0xFFFFFFFF) def enumerate_capabilities(self) -> List[str]: """Returns the list of capability strings. From f0bd9787160c8bca29e52090dc484dea213004e7 Mon Sep 17 00:00:00 2001 From: Gustavo Moreira Date: Sat, 20 May 2023 12:17:09 +0200 Subject: [PATCH 4/8] Fix typo --- volatility3/framework/symbols/linux/extensions/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/volatility3/framework/symbols/linux/extensions/__init__.py b/volatility3/framework/symbols/linux/extensions/__init__.py index 0051cdc2f..4b1c53683 100644 --- a/volatility3/framework/symbols/linux/extensions/__init__.py +++ b/volatility3/framework/symbols/linux/extensions/__init__.py @@ -1526,7 +1526,7 @@ class kernel_cap_struct(objects.StructType): capability (str): A string representing the capability i.e. dac_read_search Raises: - AttributeError: If the fiven capability is unknown to the framework. + AttributeError: If the given capability is unknown to the framework. Returns: bool: "True" if the given capability is enabled. From 241cf832f9b3676fbcf4e724c646dbfe46c24eb2 Mon Sep 17 00:00:00 2001 From: Gustavo Moreira Date: Mon, 10 Jul 2023 00:07:40 +0200 Subject: [PATCH 5/8] Plugin parameters removed. Instead, it now shows each capability set in their textual representation. The dictionary was replaced by a dataclass. Last but not leas, CAP_FULL moved to constants. --- .../framework/constants/linux/__init__.py | 2 + .../framework/plugins/linux/capabilities.py | 179 ++++++++---------- .../symbols/linux/extensions/__init__.py | 4 +- 3 files changed, 83 insertions(+), 102 deletions(-) diff --git a/volatility3/framework/constants/linux/__init__.py b/volatility3/framework/constants/linux/__init__.py index a802e0ada..e57fa30d4 100644 --- a/volatility3/framework/constants/linux/__init__.py +++ b/volatility3/framework/constants/linux/__init__.py @@ -279,3 +279,5 @@ CAPABILITIES = ( "bpf", "checkpoint_restore", ) + +CAP_FULL = 0xFFFFFFFF diff --git a/volatility3/framework/plugins/linux/capabilities.py b/volatility3/framework/plugins/linux/capabilities.py index ed08c4c05..9a59f31c1 100644 --- a/volatility3/framework/plugins/linux/capabilities.py +++ b/volatility3/framework/plugins/linux/capabilities.py @@ -3,19 +3,50 @@ # import logging +from dataclasses import dataclass, astuple, fields from typing import Iterable, List, Tuple, Dict -from volatility3.framework import interfaces, renderers +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.renderers import format_hints 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""" @@ -40,43 +71,26 @@ class Capabilities(plugins.PluginInterface): 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): + 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 = self.context.modules[self.config["kernel"]] - kernel_cap_last_cap = vmlinux.object( - object_type="int", offset=kernel_cap_last_cap - ) + 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( @@ -103,7 +117,6 @@ class Capabilities(plugins.PluginInterface): if cap_value == 0: return "-" - CAP_FULL = 0xFFFFFFFF if cap_value == CAP_FULL: return "all" @@ -119,33 +132,32 @@ class Capabilities(plugins.PluginInterface): Returns: dict: A dict with the task basic information along with 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 - 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, - ], - } + 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"): - fields["capabilities"].append(task_cred.cap_ambient) - else: - fields["capabilities"].append(renderers.NotAvailableValue()) + capabilities_data.cap_ambient = task_cred.cap_ambient - return fields + return task_data, capabilities_data + @classmethod def get_tasks_capabilities( - self, tasks: List[interfaces.objects.ObjectInterface] + cls, tasks: List[interfaces.objects.ObjectInterface] ) -> Iterable[Dict]: """Yields a dict for each task containing the task's basic information along with its capabilities @@ -156,44 +168,23 @@ class Capabilities(plugins.PluginInterface): 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) + yield cls.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"] + for task_fields, capabilities_fields in self.get_tasks_capabilities(tasks): + task_fields = astuple(task_fields) - 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())) + capabilities_text = tuple( + self._decode_cap(cap) for cap in capabilities_fields.astuple() + ) - # 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 + 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( @@ -206,23 +197,11 @@ class Capabilities(plugins.PluginInterface): ("Pid", int), ("PPid", int), ("EUID", int), + ("cap_inheritable", str), + ("cap_permitted", str), + ("cap_effective", str), + ("cap_bounding", str), + ("cap_ambient", str), ] - 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/extensions/__init__.py b/volatility3/framework/symbols/linux/extensions/__init__.py index 4b1c53683..641f77728 100644 --- a/volatility3/framework/symbols/linux/extensions/__init__.py +++ b/volatility3/framework/symbols/linux/extensions/__init__.py @@ -13,7 +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 +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 @@ -1508,7 +1508,7 @@ class kernel_cap_struct(objects.StructType): """ # 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) + return cap_value & CAP_FULL def enumerate_capabilities(self) -> List[str]: """Returns the list of capability strings. From 63c7719763d8d5c5080840bad740d7fa6be14816 Mon Sep 17 00:00:00 2001 From: Gustavo Moreira Date: Mon, 10 Jul 2023 01:01:30 +0200 Subject: [PATCH 6/8] Fixed returned types. An empty string is now returned when no capability is enabled. --- .../framework/plugins/linux/capabilities.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/volatility3/framework/plugins/linux/capabilities.py b/volatility3/framework/plugins/linux/capabilities.py index 9a59f31c1..aa3edfe4b 100644 --- a/volatility3/framework/plugins/linux/capabilities.py +++ b/volatility3/framework/plugins/linux/capabilities.py @@ -114,8 +114,8 @@ class Capabilities(plugins.PluginInterface): return cap cap_value = cap.get_capabilities() - if cap_value == 0: - return "-" + if not cap_value: + return "" if cap_value == CAP_FULL: return "all" @@ -123,14 +123,16 @@ class Capabilities(plugins.PluginInterface): return ", ".join(cap.enumerate_capabilities()) @classmethod - def get_task_capabilities(cls, task: interfaces.objects.ObjectInterface) -> Dict: + def get_task_capabilities( + cls, task: interfaces.objects.ObjectInterface + ) -> Tuple[TaskData, CapabilitiesData]: """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 + A tuple with the task basic information and its capabilities """ task_data = TaskData( comm=utility.array_to_string(task.comm), @@ -158,14 +160,14 @@ class Capabilities(plugins.PluginInterface): @classmethod def get_tasks_capabilities( cls, tasks: List[interfaces.objects.ObjectInterface] - ) -> Iterable[Dict]: + ) -> Iterable[Tuple[TaskData, CapabilitiesData]]: """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 + A tuple for each task containing the task's basic information and its capabilities """ for task in tasks: yield cls.get_task_capabilities(task) From 38e9b6baa87dbfb9ad8795621aa97a5fd1c61b30 Mon Sep 17 00:00:00 2001 From: Gustavo Moreira Date: Mon, 10 Jul 2023 01:08:46 +0200 Subject: [PATCH 7/8] Removed Dict type from the imports --- volatility3/framework/plugins/linux/capabilities.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/volatility3/framework/plugins/linux/capabilities.py b/volatility3/framework/plugins/linux/capabilities.py index aa3edfe4b..0eb9e3b7a 100644 --- a/volatility3/framework/plugins/linux/capabilities.py +++ b/volatility3/framework/plugins/linux/capabilities.py @@ -4,7 +4,7 @@ import logging from dataclasses import dataclass, astuple, fields -from typing import Iterable, List, Tuple, Dict +from typing import Iterable, List, Tuple from volatility3.framework import interfaces, renderers, exceptions from volatility3.framework.constants.linux import CAP_FULL From 42fe37ed52cf62b5eebe073420e9a6245307eb60 Mon Sep 17 00:00:00 2001 From: Gustavo Moreira Date: Mon, 10 Jul 2023 01:12:20 +0200 Subject: [PATCH 8/8] Adjust docstring to the new retuned type. --- volatility3/framework/plugins/linux/capabilities.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/volatility3/framework/plugins/linux/capabilities.py b/volatility3/framework/plugins/linux/capabilities.py index 0eb9e3b7a..84d1d543f 100644 --- a/volatility3/framework/plugins/linux/capabilities.py +++ b/volatility3/framework/plugins/linux/capabilities.py @@ -126,7 +126,7 @@ class Capabilities(plugins.PluginInterface): def get_task_capabilities( cls, task: interfaces.objects.ObjectInterface ) -> Tuple[TaskData, CapabilitiesData]: - """Returns a dict with the task basic information along with its capabilities + """Returns a tuple with the task basic information along with its capabilities Args: task: A task object from where to get the fields. @@ -161,7 +161,7 @@ class Capabilities(plugins.PluginInterface): def get_tasks_capabilities( cls, tasks: List[interfaces.objects.ObjectInterface] ) -> Iterable[Tuple[TaskData, CapabilitiesData]]: - """Yields a dict for each task containing the task's basic information along with its capabilities + """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.