Merge branch 'develop' into linux_boottime_support

volatility3/cli/volshell/generic.py

@@ -58,7 +58,7 @@ class Volshell(interfaces.plugins.PluginInterface):
         ]
 
     def run(
-        self, additional_locals: Dict[str, Any] = None
+        self, additional_locals: Dict[str, Any] = {}
     ) -> interfaces.renderers.TreeGrid:
         """Runs the interactive volshell plugin.
 
@@ -94,7 +94,10 @@ class Volshell(interfaces.plugins.PluginInterface):
 """
 
         sys.ps1 = f"({self.current_layer}) >>> "
-        self.__console = code.InteractiveConsole(locals=self._construct_locals_dict())
+        # Dict self._construct_locals_dict() will have priority on keys
+        combined_locals = additional_locals.copy()
+        combined_locals.update(self._construct_locals_dict())
+        self.__console = code.InteractiveConsole(locals=combined_locals)
         # Since we have to do work to add the option only once for all different modes of volshell, we can't
         # rely on the default having been set
         if self.config.get("script", None) is not None:

volatility3/framework/constants/linux/__init__.py

@@ -5,7 +5,7 @@
 
 Linux-specific values that aren't found in debug symbols
 """
-from enum import IntEnum
+from enum import IntEnum, Flag
 
 KERNEL_NAME = "__kernel__"
 
@@ -304,4 +304,41 @@ class ELF_CLASS(IntEnum):
     ELFCLASS64 = 2
 
 
+PT_OPT_FLAG_SHIFT = 3
+
+PTRACE_EVENT_FORK = 1
+PTRACE_EVENT_VFORK = 2
+PTRACE_EVENT_CLONE = 3
+PTRACE_EVENT_EXEC = 4
+PTRACE_EVENT_VFORK_DONE = 5
+PTRACE_EVENT_EXIT = 6
+PTRACE_EVENT_SECCOMP = 7
+
+PTRACE_O_EXITKILL = 1 << 20
+PTRACE_O_SUSPEND_SECCOMP = 1 << 21
+
+
+class PT_FLAGS(Flag):
+    "PTrace flags"
+    PT_PTRACED = 0x00001
+    PT_SEIZED = 0x10000
+
+    PT_TRACESYSGOOD = 1 << (PT_OPT_FLAG_SHIFT + 0)
+    PT_TRACE_FORK = 1 << (PT_OPT_FLAG_SHIFT + PTRACE_EVENT_FORK)
+    PT_TRACE_VFORK = 1 << (PT_OPT_FLAG_SHIFT + PTRACE_EVENT_VFORK)
+    PT_TRACE_CLONE = 1 << (PT_OPT_FLAG_SHIFT + PTRACE_EVENT_CLONE)
+    PT_TRACE_EXEC = 1 << (PT_OPT_FLAG_SHIFT + PTRACE_EVENT_EXEC)
+    PT_TRACE_VFORK_DONE = 1 << (PT_OPT_FLAG_SHIFT + PTRACE_EVENT_VFORK_DONE)
+    PT_TRACE_EXIT = 1 << (PT_OPT_FLAG_SHIFT + PTRACE_EVENT_EXIT)
+    PT_TRACE_SECCOMP = 1 << (PT_OPT_FLAG_SHIFT + PTRACE_EVENT_SECCOMP)
+
+    PT_EXITKILL = PTRACE_O_EXITKILL << PT_OPT_FLAG_SHIFT
+    PT_SUSPEND_SECCOMP = PTRACE_O_SUSPEND_SECCOMP << PT_OPT_FLAG_SHIFT
+
+    @property
+    def flags(self) -> str:
+        """Returns the ptrace flags string"""
+        return str(self).replace(self.__class__.__name__ + ".", "")
+
+
 NSEC_PER_SEC = 1e9

volatility3/framework/contexts/__init__.py

@@ -245,7 +245,7 @@ class Module(interfaces.context.ModuleInterface):
         """
         if constants.BANG not in object_type:
             object_type = self.symbol_table_name + constants.BANG + object_type
-        else:
+        elif not object_type.startswith(self.symbol_table_name + constants.BANG):
             raise ValueError(
                 "Cannot reference another module when constructing an object"
             )

volatility3/framework/plugins/linux/ptrace.py

@@ -0,0 +1,97 @@
+# This file is Copyright 2024 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 List, Iterator
+
+from volatility3.framework import renderers, interfaces
+from volatility3.framework.constants import architectures
+from volatility3.framework.objects import utility
+from volatility3.framework.configuration import requirements
+from volatility3.framework.interfaces import plugins
+from volatility3.plugins.linux import pslist
+
+vollog = logging.getLogger(__name__)
+
+
+class Ptrace(plugins.PluginInterface):
+    """Enumerates ptrace's tracer and tracee tasks"""
+
+    _required_framework_version = (2, 10, 0)
+    _version = (1, 0, 0)
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=architectures.LINUX_ARCHS,
+            ),
+            requirements.PluginRequirement(
+                name="pslist", plugin=pslist.PsList, version=(2, 2, 0)
+            ),
+        ]
+
+    @classmethod
+    def enumerate_ptrace_tasks(
+        cls,
+        context: interfaces.context.ContextInterface,
+        vmlinux_module_name: str,
+    ) -> Iterator[interfaces.objects.ObjectInterface]:
+        """Enumerates ptrace's tracer and tracee tasks
+
+        Args:
+            context: The context to retrieve required elements (layers, symbol tables) from
+            vmlinux_module_name: The name of the kernel module on which to operate
+
+        Yields:
+            A task_struct object
+        """
+
+        tasks = pslist.PsList.list_tasks(
+            context,
+            vmlinux_module_name,
+            filter_func=pslist.PsList.create_pid_filter(),
+            include_threads=True,
+        )
+
+        for task in tasks:
+            if task.is_being_ptraced or task.is_ptracing:
+                yield task
+
+    def _generator(self, vmlinux_module_name):
+        for task in self.enumerate_ptrace_tasks(self.context, vmlinux_module_name):
+            task_comm = utility.array_to_string(task.comm)
+            user_pid = task.tgid
+            user_tid = task.pid
+            tracer_tid = task.get_ptrace_tracer_tid() or renderers.NotAvailableValue()
+            tracee_tids = task.get_ptrace_tracee_tids() or [
+                renderers.NotAvailableValue()
+            ]
+            flags = task.get_ptrace_tracee_flags() or renderers.NotAvailableValue()
+
+            for level, tracee_tid in enumerate(tracee_tids):
+                fields = [
+                    task_comm,
+                    user_pid,
+                    user_tid,
+                    tracer_tid,
+                    tracee_tid,
+                    flags,
+                ]
+                yield (level, fields)
+
+    def run(self):
+        vmlinux_module_name = self.config["kernel"]
+
+        headers = [
+            ("Process", str),
+            ("PID", int),
+            ("TID", int),
+            ("Tracer TID", int),
+            ("Tracee TID", int),
+            ("Flags", str),
+        ]
+        return renderers.TreeGrid(headers, self._generator(vmlinux_module_name))

volatility3/framework/plugins/windows/unloadedmodules.py

@@ -7,7 +7,7 @@ import datetime
 from typing import List, Iterable
 
 from volatility3.framework import constants
-from volatility3.framework import interfaces, symbols
+from volatility3.framework import interfaces, symbols, exceptions
 from volatility3.framework import renderers
 from volatility3.framework.configuration import requirements
 from volatility3.framework.interfaces import configuration
@@ -132,10 +132,15 @@ class UnloadedModules(interfaces.plugins.PluginInterface, timeliner.TimeLinerInt
             kernel.symbol_table_name,
             unloadedmodule_table_name,
         ):
+            try:
+                name = mod.Name.String
+            except exceptions.InvalidAddressException:
+                name = renderers.UnreadableValue()
+
             yield (
                 0,
                 (
-                    mod.Name.String,
+                    name,
                     format_hints.Hex(mod.StartAddress),
                     format_hints.Hex(mod.EndAddress),
                     conversion.wintime_to_datetime(mod.CurrentTime),

volatility3/framework/symbols/linux/__init__.py

@@ -26,6 +26,7 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
         # Set-up Linux specific types
         self.set_type_class("file", extensions.struct_file)
         self.set_type_class("list_head", extensions.list_head)
+        self.set_type_class("hlist_head", extensions.hlist_head)
         self.set_type_class("mm_struct", extensions.mm_struct)
         self.set_type_class("super_block", extensions.super_block)
         self.set_type_class("task_struct", extensions.task_struct)
@@ -57,6 +58,7 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
         # Might not exist in older kernels or the current symbols
         self.optional_set_type_class("mount", extensions.mount)
         self.optional_set_type_class("mnt_namespace", extensions.mnt_namespace)
+        self.optional_set_type_class("rb_root", extensions.rb_root)
 
         # Network
         self.set_type_class("net", extensions.net)

volatility3/framework/symbols/linux/extensions/__init__.py

@@ -18,7 +18,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, PT_FLAGS
 from volatility3.framework.layers import linear
 from volatility3.framework.objects import utility
 from volatility3.framework.symbols import generic, linux, intermed
@@ -382,6 +382,41 @@ class task_struct(generic.GenericIntelProcess):
                 threads_seen.add(task.vol.offset)
                 yield task
 
+    @property
+    def is_being_ptraced(self) -> bool:
+        """Returns True if this task is being traced using ptrace"""
+        return self.ptrace != 0
+
+    @property
+    def is_ptracing(self) -> bool:
+        """Returns True if this task is tracing other tasks using ptrace"""
+        is_tracing = (
+            self.ptraced.next.is_readable()
+            and self.ptraced.next.dereference().vol.offset != self.ptraced.vol.offset
+        )
+        return is_tracing
+
+    def get_ptrace_tracer_tid(self) -> Optional[int]:
+        """Returns the tracer's TID tracing this task"""
+        return self.parent.pid if self.is_being_ptraced else None
+
+    def get_ptrace_tracee_tids(self) -> List[int]:
+        """Returns the list of TIDs being traced by this task"""
+        task_symbol_table_name = self.get_symbol_table_name()
+
+        task_struct_symname = f"{task_symbol_table_name}{constants.BANG}task_struct"
+        tracing_tid_list = [
+            task_being_traced.pid
+            for task_being_traced in self.ptraced.to_list(
+                task_struct_symname, "ptrace_entry"
+            )
+        ]
+        return tracing_tid_list
+
+    def get_ptrace_tracee_flags(self) -> Optional[str]:
+        """Returns a string with the ptrace flags"""
+        return PT_FLAGS(self.ptrace).flags if self.is_being_ptraced else None
+
     def _get_task_start_time(self) -> datetime.timedelta:
         """Returns the task's monotonic start_time as a timedelta.
 
@@ -1043,7 +1078,7 @@ class dentry(objects.StructType):
         if self.has_member("d_sib") and self.has_member("d_children"):
             # kernels >= 6.8
             walk_member = "d_sib"
-            list_head_member = self.d_children.first
+            list_head_member = self.d_children
         elif self.has_member("d_child") and self.has_member("d_subdirs"):
             # 2.5.0 <= kernels < 6.8
             walk_member = "d_child"
@@ -1160,6 +1195,40 @@ class list_head(objects.StructType, collections.abc.Iterable):
         return self.to_list(self.vol.parent.vol.type_name, self.vol.member_name)
 
 
+class hlist_head(objects.StructType):
+    def to_list(
+        self,
+        symbol_type: str,
+        member: str,
+    ) -> Iterator[interfaces.objects.ObjectInterface]:
+        """Returns an iterator of the entries in the list.
+
+        This is a doubly linked list; however, it is not circular, so the 'forward' field
+        doesn't make sense. Also, the sentinel concept doesn't make sense here either;
+        unlike list_head, the head and nodes each have their own distinct types. A list_head
+        cannot be a node by itself.
+        - The 'pprev' of the first 'hlist_node' points to the 'hlist_head', not to the last node.
+        - The last element 'next' member is NULL
+
+        Args:
+            symbol_type: Type of the list elements
+            member: Name of the list_head member in the list elements
+
+        Yields:
+            Objects of the type specified via the "symbol_type" argument.
+
+        """
+        vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
+
+        current = self.first
+        while current and current.is_readable():
+            yield linux.LinuxUtilities.container_of(
+                current, symbol_type, member, vmlinux
+            )
+
+            current = current.next
+
+
 class files_struct(objects.StructType):
     def get_fds(self) -> interfaces.objects.ObjectInterface:
         if self.has_member("fdt"):
@@ -1565,14 +1634,42 @@ class mnt_namespace(objects.StructType):
         else:
             raise AttributeError("Unable to find mnt_namespace inode")
 
-    def get_mount_points(self):
+    def get_mount_points(
+        self,
+    ) -> Iterator[interfaces.objects.ObjectInterface]:
+        """Yields the mount points for this mount namespace.
+
+        Yields:
+            mount struct instances
+        """
         table_name = self.vol.type_name.split(constants.BANG)[0]
-        mnt_type = table_name + constants.BANG + "mount"
-        if not self._context.symbol_space.has_type(mnt_type):
-            # Old kernels ~ 2.6
-            mnt_type = table_name + constants.BANG + "vfsmount"
-        for mount in self.list.to_list(mnt_type, "mnt_list"):
-            yield mount
+
+        if self.has_member("list"):
+            # kernels < 6.8
+            mnt_type = table_name + constants.BANG + "mount"
+            if not self._context.symbol_space.has_type(mnt_type):
+                # In kernels < 3.3, the 'mount' struct didn't exist, and the 'mnt_list'
+                # member was part of the 'vfsmount' struct.
+                mnt_type = table_name + constants.BANG + "vfsmount"
+
+            yield from self.list.to_list(mnt_type, "mnt_list")
+        elif (
+            self.has_member("mounts")
+            and self.mounts.vol.type_name == table_name + constants.BANG + "rb_root"
+        ):
+            # kernels >= 6.8
+            vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(
+                self._context, self
+            )
+            for node in self.mounts.get_nodes():
+                mnt = linux.LinuxUtilities.container_of(
+                    node, "mount", "mnt_list", vmlinux
+                )
+                yield mnt
+        else:
+            raise exceptions.VolatilityException(
+                "Unsupported kernel mount namespace implementation"
+            )
 
 
 class net(objects.StructType):
@@ -2449,3 +2546,31 @@ class IDR(objects.StructType):
 
         for page_addr in get_entries_func():
             yield page_addr
+
+
+class rb_root(objects.StructType):
+    def _walk_nodes(self, root_node) -> Iterator[int]:
+        """Traverses the Red-Black tree from the root node and yields a pointer to each
+        node in this tree.
+
+        Args:
+            root_node: A Red-Black tree node from which to start descending
+
+        Yields:
+            A pointer to every node descending from the specified root node
+        """
+        if not root_node:
+            return
+
+        yield root_node
+        yield from self._walk_nodes(root_node.rb_left)
+        yield from self._walk_nodes(root_node.rb_right)
+
+    def get_nodes(self) -> Iterator[int]:
+        """Yields a pointer to each node in the Red-Black tree
+
+        Yields:
+            A pointer to every node in the Red-Black tree
+        """
+
+        yield from self._walk_nodes(root_node=self.rb_node)

volatility3/schemas/__init__.py

@@ -6,8 +6,8 @@ import hashlib
 import json
 import logging
 import os
-from typing import Any, Dict, Optional, Set
-
+import re
+from typing import Any, Dict, Optional, Set, Tuple
 from volatility3.framework import constants
 
 vollog = logging.getLogger(__name__)
@@ -77,6 +77,17 @@ def valid(
     input: Dict[str, Any], schema: Dict[str, Any], use_cache: bool = True
 ) -> bool:
     """Validates a json schema."""
+    producer = input.get("metadata", {}).get("producer", {})
+    if producer and producer.get("name") == "dwarf2json":
+        dwarf2json_version = parse_producer_version(producer.get("version", ""))
+        # No warnings if version couldn't be parsed, as it's not our role here
+        # to validate the schema.
+        if dwarf2json_version:
+            if dwarf2json_check_rust_type_confusion(input, dwarf2json_version):
+                vollog.warning(
+                    "This ISF was generated by dwarf2json < 0.9.0, which is known to produce inaccurate results (see dwarf2json GitHub issue #63)."
+                )
+
     input_hash = create_json_hash(input, schema)
     if input_hash in cached_validations and use_cache:
         return True
@@ -98,3 +109,42 @@ def valid(
 
     record_cached_validations(cached_validations)
     return True
+
+
+def parse_producer_version(version_string: str) -> Optional[Tuple[int]]:
+    """Parses a producer version and returns a tuple of identifiers.
+
+    Args:
+        version_string: string containing dot-separated integers,
+    expected to follow the Volatility3 versioning schema
+
+    Returns:
+        A tuple containing each version identifier
+    """
+    identifiers = re.search("^(\\d+)[.](\\d+)[.](\\d+)$", version_string)
+    if not identifiers:
+        return None
+
+    return tuple(int(d) for d in identifiers.groups())
+
+
+# dwarf2json sanity checks #
+def dwarf2json_check_rust_type_confusion(
+    input: Dict[str, Any], dwarf2json_version: Tuple[int]
+) -> bool:
+    """dwarf2json sanity check for Rust and C types confusion:
+     - dwarf2json #63
+     - volatility3 #1305
+
+    Args:
+        dwarf2json_version: a tuple containing each version identifier
+
+    Returns:
+        True if the issue was detected
+    """
+
+    return "rust_helper_BUG" in input.get("symbols", {}) and dwarf2json_version < (
+        0,
+        9,
+        0,
+    )