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

Added linux capabilities plugin

Browse Source
This commit is contained in:
Gustavo Moreira
2023-05-17 14:14:33 +01:00
parent 8e56cb39d1
commit 3d9efc3b34
4 changed files with 372 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/framework/constants/linux/__init__.py
+45
View File
@@ -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",
)
volatility3/framework/plugins/linux/capabilities.py
+218
View File
@@ -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))
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
+107 -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
@@ -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