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.
This commit is contained in:
Gustavo Moreira
2023-07-10 00:07:40 +02:00
parent f0bd978716
commit 241cf832f9
3 changed files with 83 additions and 102 deletions
@@ -279,3 +279,5 @@ CAPABILITIES = (
"bpf",
"checkpoint_restore",
)
CAP_FULL = 0xFFFFFFFF
@@ -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))
@@ -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.