Merge remote-tracking branch 'origin/develop' into linux_tracing_ftrace

pyproject.toml

@@ -37,7 +37,7 @@ cloud = ["gcsfs>=2024.10.0", "s3fs>=2024.10.0"]
 dev = [
     "volatility3[full,cloud]",
     "jsonschema>=4.23.0,<5",
-    "pyinstaller>=6.11.0,<7",
+    "pyinstaller>=6.5.0,<7",
     "pyinstaller-hooks-contrib>=2024.9",
     "types-jsonschema>=4.23.0,<5",
 ]
@@ -51,8 +51,8 @@ test = [
 
 docs = [
     "volatility3[dev]",
-    "sphinx>=8.0.0,<7",
-    "sphinx-autodoc-typehints>=2.5.0,<3",
+    "sphinx>=4.0.0,<9",
+    "sphinx-autodoc-typehints>=2.0.0,<3",
     "sphinx-rtd-theme>=3.0.1,<4",
 ]
 

volatility3/cli/__init__.py

@@ -573,6 +573,8 @@ class CommandLine:
         fulltrace = traceback.TracebackException.from_exception(excp).format(chain=True)
         vollog.debug("".join(fulltrace))
 
+        file_a_bug_msg = f"Please re-run with -vvv and file a bug with the output at {constants.BUG_URL}"
+
         if isinstance(excp, exceptions.InvalidAddressException):
             general = "Volatility was unable to read a requested page:"
             if isinstance(excp, exceptions.SwappedInvalidAddressException):
@@ -617,9 +619,7 @@ class CommandLine:
         elif isinstance(excp, exceptions.LayerException):
             general = f"Volatility experienced a layer-related issue: {excp.layer_name}"
             detail = f"{excp}"
-            caused_by = [
-                "A faulty layer implementation (re-run with -vvv and file a bug)"
-            ]
+            caused_by = [f"A faulty layer implementation. {file_a_bug_msg}"]
         elif isinstance(excp, exceptions.MissingModuleException):
             general = f"Volatility could not import a necessary module: {excp.module}"
             detail = f"{excp}"
@@ -630,13 +630,17 @@ class CommandLine:
             general = "Volatility experienced an issue when rendering the output:"
             detail = f"{excp}"
             caused_by = ["An invalid renderer option, such as no visible columns"]
+        elif isinstance(excp, exceptions.VersionMismatchException):
+            general = "A version mismatch was detected between two components:"
+            detail = f"{excp}"
+            caused_by = [
+                excp.failure_reason or "An outdated API caller, such as a method.",
+                file_a_bug_msg,
+            ]
         else:
             general = "Volatility encountered an unexpected situation."
             detail = ""
-            caused_by = [
-                "Please re-run using with -vvv and file a bug with the output",
-                f"at {constants.BUG_URL}",
-            ]
+            caused_by = [file_a_bug_msg]
 
         # Code that actually renders the exception
         output = sys.stderr

volatility3/cli/volshell/linux.py

@@ -3,6 +3,7 @@
 #
 
 from typing import Any, List, Optional, Tuple, Union
+from enum import Enum
 
 from volatility3.cli.volshell import generic
 from volatility3.framework import constants, interfaces
@@ -10,6 +11,16 @@ from volatility3.framework.configuration import requirements
 from volatility3.plugins.linux import pslist
 
 
+# Could import the enum from psscan.py to avoid code duplication
+class DescExitStateEnum(Enum):
+    """Enum for linux task exit_state as defined in include/linux/sched.h"""
+
+    TASK_RUNNING = 0x00000000
+    EXIT_DEAD = 0x00000010
+    EXIT_ZOMBIE = 0x00000020
+    EXIT_TRACE = EXIT_ZOMBIE | EXIT_DEAD
+
+
 class Volshell(generic.Volshell):
     """Shell environment to directly interact with a linux memory image."""
 
@@ -40,6 +51,71 @@ class Volshell(generic.Volshell):
                 return None
         print(f"No task with task ID {pid} found")
 
+    def get_process(self, pid=None, virtaddr=None, physaddr=None):
+        """Return the task_struct object that matches the pid. If a physical or a virtual address is provided, construct the task_struct object at said address. Only one parameter is allowed.
+
+        Args:
+            pid (int, optional): PID to search for
+            virtaddr (int, optional): Virtual address to construct object at
+            physaddr (int, optional): Physical address to construct object at
+
+        Returns:
+            ObjectInterface: task_struct Object
+        """
+
+        if sum(1 if x is not None else 0 for x in [pid, virtaddr, physaddr]) != 1:
+            print("Only one parameter is accepted")
+            return None
+
+        vmlinux_module_name = self.config["kernel"]
+        vmlinux = self.context.modules[vmlinux_module_name]
+
+        kernel_layer_name = vmlinux.layer_name
+        kernel_layer = self.context.layers[kernel_layer_name]
+
+        memory_layer_name = kernel_layer.dependencies[0]
+
+        task_struct_symbol = vmlinux.symbol_table_name + constants.BANG + "task_struct"
+
+        if virtaddr is not None:
+            task = self.context.object(
+                task_struct_symbol,
+                layer_name=kernel_layer_name,
+                offset=virtaddr,
+            )
+
+        if physaddr is not None:
+            task = self.context.object(
+                task_struct_symbol,
+                layer_name=memory_layer_name,
+                offset=physaddr,
+                native_layer_name=kernel_layer_name,
+            )
+
+        if physaddr is not None or virtaddr is not None:
+            try:
+                DescExitStateEnum(task.exit_state)
+            except ValueError:
+                print(
+                    f"task_struct @ {hex(task.vol.offset)} as exit_state {task.exit_state} is likely not valid"
+                )
+
+            if not (0 < task.pid < 65535):
+                print(
+                    f"task_struct @ {hex(task.vol.offset)} as pid {task.pid} is likely not valid"
+                )
+
+            return task
+
+        if pid is not None:
+            tasks = self.list_tasks()
+            for task in tasks:
+                if task.pid == pid:
+                    return task
+            print(f"No task with task ID {pid} found")
+
+        return None
+
     def list_tasks(self):
         """Returns a list of task objects from the primary layer"""
         # We always use the main kernel memory and associated symbols
@@ -50,6 +126,7 @@ class Volshell(generic.Volshell):
         result += [
             (["ct", "change_task", "cp"], self.change_task),
             (["lt", "list_tasks", "ps"], self.list_tasks),
+            (["gp", "get_process", "get_task"], self.get_process),
             (["symbols"], self.context.symbol_space[self.current_symbol_table]),
         ]
         if self.config.get("pid", None) is not None:

volatility3/cli/volshell/windows.py

@@ -44,11 +44,67 @@ class Volshell(generic.Volshell):
             )
         )
 
+    def get_process(self, pid=None, virtaddr=None, physaddr=None):
+        """Returns the _EPROCESS object that matches the pid. If a physical or a virtual address is provided, construct the _EPROCESS object at said address. Only one parameter is allowed.
+
+        Args:
+            pid (int, optional): PID / UniqueProcessId to search for.
+            virtaddr (int, optional): Virtual address to construct object at
+            physaddr (int, optional): Physical address to construct object at
+
+        Returns:
+            ObjectInterface: _EPROCESS Object
+        """
+
+        if sum(1 if x is not None else 0 for x in [pid, virtaddr, physaddr]) != 1:
+            print("Only one parameter is accepted")
+            return None
+
+        kernel_name = self.config["kernel"]
+        kernel = self.context.modules[kernel_name]
+
+        kernel_layer_name = kernel.layer_name
+
+        kernel_layer = self.context.layers[kernel_layer_name]
+        memory_layer_name = kernel_layer.dependencies[0]
+
+        eprocess_symbol = kernel.symbol_table_name + constants.BANG + "_EPROCESS"
+
+        if virtaddr is not None:
+            eproc = self.context.object(
+                eprocess_symbol,
+                layer_name=kernel_layer_name,
+                offset=virtaddr,
+            )
+
+            return eproc
+
+        if physaddr is not None:
+            eproc = self.context.object(
+                eprocess_symbol,
+                layer_name=memory_layer_name,
+                offset=physaddr,
+                native_layer_name=kernel_layer_name,
+            )
+
+            return eproc
+
+        if pid is not None:
+            processes = self.list_processes()
+            for process in processes:
+                if process.UniqueProcessId == pid:
+                    return process
+            print(f"No process with process ID {pid} found")
+            return None
+
+        return None
+
     def construct_locals(self) -> List[Tuple[List[str], Any]]:
         result = super().construct_locals()
         result += [
             (["cp", "change_process"], self.change_process),
             (["lp", "list_processes", "ps"], self.list_processes),
+            (["gp", "get_process"], self.get_process),
             (["symbols"], self.context.symbol_space[self.current_symbol_table]),
         ]
         if self.config.get("pid", None) is not None:

volatility3/framework/__init__.py

@@ -11,10 +11,24 @@ import inspect
 import logging
 import os
 import traceback
-from typing import Any, Dict, Generator, List, Optional, Tuple, Type, TypeVar
+import functools
+import warnings
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, TypeVar
 
-from volatility3.framework import constants, interfaces
+from volatility3.framework import constants, exceptions, interfaces
+from volatility3.framework.configuration import requirements
 
+if (
+    sys.version_info.major != constants.REQUIRED_PYTHON_VERSION[0]
+    or sys.version_info.minor < constants.REQUIRED_PYTHON_VERSION[1]
+    or (
+        sys.version_info.minor == constants.REQUIRED_PYTHON_VERSION[1]
+        and sys.version_info.micro < constants.REQUIRED_PYTHON_VERSION[2]
+    )
+):
+    raise RuntimeError(
+        f"Volatility framework requires python version {'.'.join(str(x) for x in constants.REQUIRED_PYTHON_VERSION)} or greater"
+    )
 
 # ##
 #
@@ -52,12 +66,67 @@ def require_interface_version(*args) -> None:
                 )
 
 
+class Deprecation:
+    """Deprecation related methods."""
+
+    @staticmethod
+    def deprecated_method(
+        replacement: Callable,
+        replacement_version: Tuple[int, int, int] = None,
+        additional_information: str = "",
+    ):
+        """A decorator for marking functions as deprecated.
+
+        Args:
+            replacement: The replacement function overriding the deprecated API, in the form of a Callable (typically a method)
+            replacement_version: The "replacement" base class version that the deprecated method expects before proxying to it. This implies that "replacement" is a method from a class that inherits from VersionableInterface.
+            additional_information: Information appended at the end of the deprecation message
+        """
+
+        def decorator(deprecated_func):
+            @functools.wraps(deprecated_func)
+            def wrapper(*args, **kwargs):
+                nonlocal replacement, replacement_version, additional_information
+                # Prevent version mismatches between deprecated (proxy) methods and the ones they proxy
+                if (
+                    replacement_version is not None
+                    and callable(replacement)
+                    and hasattr(replacement, "__self__")
+                ):
+                    replacement_base_class = replacement.__self__
+
+                    # Verify that the base class inherits from VersionableInterface
+                    if inspect.isclass(replacement_base_class) and issubclass(
+                        replacement_base_class,
+                        interfaces.configuration.VersionableInterface,
+                    ):
+                        # SemVer check
+                        if not requirements.VersionRequirement.matches_required(
+                            replacement_version, replacement_base_class.version
+                        ):
+                            raise exceptions.VersionMismatchException(
+                                deprecated_func,
+                                replacement_base_class,
+                                replacement_version,
+                                "This is a bug, the deprecated call needs to be removed and the caller needs to update their code to use the new method.",
+                            )
+
+                deprecation_msg = f"Method \"{deprecated_func.__module__ + '.' + deprecated_func.__qualname__}\" is deprecated, use \"{replacement.__module__ + '.' + replacement.__qualname__}\" instead. {additional_information}"
+                warnings.warn(deprecation_msg, FutureWarning)
+                # Return the wrapped function with its original arguments
+                return deprecated_func(*args, **kwargs)
+
+            return wrapper
+
+        return decorator
+
+
 class NonInheritable:
     def __init__(self, value: Any, cls: Type) -> None:
         self.default_value = value
         self.cls = cls
 
-    def __get__(self, obj: Any, get_type: Optional[Type] = None) -> Any:
+    def __get__(self, obj: Any, get_type: Type = Optional[None]) -> Any:
         if type is self.cls:
             if hasattr(self.default_value, "__get__"):
                 return self.default_value.__get__(obj, get_type)

volatility3/framework/automagic/pdbscan.py

@@ -376,8 +376,70 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
                     valid_kernel = (virtual_layer_name, address, res[0])
         return valid_kernel
 
+    def method_low_stub_offset(
+        self,
+        context: interfaces.context.ContextInterface,
+        vlayer: layers.intel.Intel,
+        progress_callback: constants.ProgressCallback = None,
+    ) -> Optional[ValidKernelType]:
+        # This method is only valid for x64 systems
+        if not isinstance(vlayer, intel.Intel32e):
+            return None
+        kernel_hint = 0
+        kernel_base = 0
+        physical_layer = context.layers.get("memory_layer")
+
+        # Try locating kernel base via x64 Low Stub in lower 1MB starting from second page (4KB)
+        # If "Discard Low Memory" setting is disabled in BIOS, the Low Stub may be at the third/fourth or further pages
+        for offset in range(0x1000, 0x100000, 0x1000):
+            jmp_and_completion_values = int.from_bytes(
+                physical_layer.read(offset, 0x8), "little"
+            )
+            if (
+                0xFFFFFFFFFFFF00FF & jmp_and_completion_values
+                != constants.windows.JMP_AND_COMPLETION_SIGNATURE
+            ):
+                continue
+            cr3_value = int.from_bytes(
+                physical_layer.read(
+                    offset + constants.windows.PROCESSOR_START_BLOCK_CR3_OFFSET, 0x8
+                ),
+                "little",
+            )
+
+            # Compare previously observed valid page table address that's stored in vlayer._initial_entry
+            # with PROCESSOR_START_BLOCK->ProcessorState->SpecialRegisters->Cr3
+            # which was observed to be an invalid page address, so add 1 (to make it valid too)
+            if (cr3_value + 1) != vlayer._initial_entry:
+                continue
+            potential_kernel_hint = int.from_bytes(
+                physical_layer.read(
+                    offset + constants.windows.PROCESSOR_START_BLOCK_LM_TARGET_OFFSET,
+                    0x8,
+                ),
+                "little",
+            )
+            if 0x3 & potential_kernel_hint:
+                continue
+            kernel_hint = potential_kernel_hint & 0xFFFFFFFFFFFF
+            kernel_base = kernel_hint & (~0x1FFFFF) & 0xFFFFFFFFFFFF
+            break
+
+        if kernel_base:
+            # Scanning 32mb in 2mb chunks for the 'ntoskrnl' base address
+            while (kernel_base + 0x2000000) > kernel_hint:
+                for i in range(0, 0x200000, 0x1000):
+                    valid_kernel = self.check_kernel_offset(
+                        context, vlayer, kernel_base, progress_callback
+                    )
+                    if valid_kernel:
+                        return valid_kernel
+                kernel_base -= 0x200000
+        return None
+
     # List of methods to be run, in order, to determine the valid kernels
     methods = [
+        method_low_stub_offset,
         method_kdbg_offset,
         method_module_offset,
         method_fixed_mapping,

volatility3/framework/check_python_version.py

@@ -1,14 +0,0 @@
-import sys
-
-required_python_version = (3, 8, 0)
-if (
-    sys.version_info.major != required_python_version[0]
-    or sys.version_info.minor < required_python_version[1]
-    or (
-        sys.version_info.minor == required_python_version[1]
-        and sys.version_info.micro < required_python_version[2]
-    )
-):
-    raise RuntimeError(
-        f"Volatility framework requires python version {required_python_version[0]}.{required_python_version[1]}.{required_python_version[2]} or greater"
-    )

volatility3/framework/constants/__init__.py

@@ -23,6 +23,8 @@ from volatility3.framework.constants._version import (
     VERSION_SUFFIX as VERSION_SUFFIX,
 )
 
+REQUIRED_PYTHON_VERSION = (3, 8, 0)
+
 PLUGINS_PATH = [
     os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "plugins")),
     os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "plugins")),

volatility3/framework/constants/windows/__init__.py

@@ -10,3 +10,21 @@ KERNEL_MODULE_NAMES = ["ntkrnlmp", "ntkrnlpa", "ntkrpamp", "ntoskrnl"]
 """The list of names that kernel modules can have within the windows OS"""
 
 PE_MAX_EXTRACTION_SIZE = 1024 * 1024 * 256
+
+"""
+The following constants represent the layout of the Low Stub which exists only on x64 machines with no virtualization/emulation,
+responsible for transitioning from Real Mode(16 bit) to Protected Mode(32 bit) and Long Mode(64 bit) on boot/return from sleep.
+Contains offsets to fields and structures within the undocumented structure _PROCESSOR_START_BLOCK.
+Here's a reference: https://github.com/mic101/windows/blob/master/WRK-v1.2/base/ntos/inc/amd64.h#L3334
+"""
+# Expected signature for validation, constructed from:
+# PROCESSOR_START_BLOCK->Jmp->OpCode | PROCESSOR_START_BLOCK->Jmp->Offset | PROCESSOR_START_BLOCK->CompletionFlag
+JMP_AND_COMPLETION_SIGNATURE = 0x00000001000600E9
+
+# Address of LmTarget (Long Mode target)
+PROCESSOR_START_BLOCK_LM_TARGET_OFFSET = (
+    0x70  # PROCESSOR_START_BLOCK->LmTarget, PVOID 8 bytes
+)
+
+# CR3 register within structures describing initial processor state to be started
+PROCESSOR_START_BLOCK_CR3_OFFSET = 0xA0  # PROCESSOR_START_BLOCK->ProcessorState->SpecialRegisters->Cr3, ULONG64 8 bytes

volatility3/framework/contexts/__init__.py

@@ -11,7 +11,8 @@ without them interfering with each other.
 import functools
 import hashlib
 import logging
-from typing import Callable, Iterable, List, Optional, Set, Tuple, Union
+import re
+from typing import Callable, Dict, Iterable, List, Optional, Set, Tuple, Union
 
 from volatility3.framework import constants, interfaces, symbols, exceptions
 from volatility3.framework.objects import templates
@@ -386,10 +387,8 @@ class ModuleCollection(interfaces.context.ModuleContainer):
     """Class to contain a collection of SizedModules and reason about their
     contents."""
 
-    def __init__(
-        self, modules: Optional[List[interfaces.context.ModuleInterface]] = None
-    ) -> None:
-        self._prefix_count = {}
+    def __init__(self, modules: Optional[List[SizedModule]] = None) -> None:
+        self._modules: Dict[str, SizedModule] = {}
         super().__init__(modules)
 
     def deduplicate(self) -> "ModuleCollection":
@@ -402,20 +401,19 @@ class ModuleCollection(interfaces.context.ModuleContainer):
         new_modules = []
         seen: Set[str] = set()
         for mod in self._modules:
-            if mod.hash not in seen or mod.size == 0:
+            if self._modules[mod].hash not in seen or self._modules[mod].size == 0:
                 new_modules.append(mod)
-                seen.add(mod.hash)  # type: ignore # FIXME: mypy #5107
+                seen.add(self._modules[mod].hash)
         return ModuleCollection(new_modules)
 
     def free_module_name(self, prefix: str = "module") -> str:
         """Returns an unused module name"""
-        if prefix not in self._prefix_count:
-            self._prefix_count[prefix] = 1
+        existing_names = [name for name in self if re.match(rf"^{prefix}[0-9]*$", name)]
+        if not existing_names:
             return prefix
-        count = self._prefix_count[prefix]
+        count = len(existing_names)
         while prefix + str(count) in self:
             count += 1
-        self._prefix_count[prefix] = count
         return prefix + str(count)
 
     @property

volatility3/framework/exceptions.py

@@ -8,9 +8,10 @@ space or symbol tables, and by layers when an address is invalid.  The
 :class:`PagedInvalidAddressException` contains information about the
 size of the invalid page.
 """
-from typing import Dict, Optional
+from typing import Callable, Dict, Optional, Tuple
 
 from volatility3.framework import interfaces
+from volatility3.framework.interfaces.configuration import VersionableInterface
 
 
 class VolatilityException(Exception):
@@ -134,3 +135,31 @@ class RenderException(VolatilityException):
 
 class LinuxPageCacheException(VolatilityException):
     """Thrown if there is an error during Linux Page Cache processing"""
+
+
+class VersionMismatchException(VolatilityException):
+    """Thrown if a version mismatch has been encountered between two components."""
+
+    def __init__(
+        self,
+        source_component: Callable,
+        target_component: VersionableInterface,
+        target_version: Tuple[int, int, int],
+        failure_reason: str = None,
+        *args,
+    ):
+        """
+        Args:
+            source_component: The component that required the target component
+            target_component: The component that is required. Must inherit from VersionableInterface
+            target_version: The version of the target component that was required, and ultimately was not satisfied
+            failure_reason: A detailed failure reason to enhance debugging and bug tracking
+        """
+        super().__init__(*args)
+        self.source_component = source_component
+        self.target_component = target_component
+        self.target_version = target_version
+        self.failure_reason = failure_reason
+
+    def __str__(self):
+        return f"{self.source_component.__module__+ '.' + self.source_component.__qualname__}: Version {self.target_version} dependency on {self.target_component.__module__+ '.' + self.target_component.__name__} {self.target_component.version} unmet."

volatility3/framework/interfaces/context.py

@@ -302,9 +302,11 @@ class ModuleInterface(interfaces.configuration.ConfigurableInterface):
     def has_enumeration(self, name: str) -> bool:
         """Determines whether an enumeration is present in the module's symbol table."""
 
+    @property
     @abstractmethod
     def symbols(self) -> Iterable[str]:
         """Returns an iterable of the symbols contained in the symbol table for this module"""
+        raise NotImplementedError("Symbols property has not been implemented.")
 
     @abstractmethod
     def get_symbols_by_absolute_location(self, offset: int, size: int = 0) -> List[str]:

volatility3/framework/layers/leechcore.py

@@ -129,6 +129,8 @@ if HAS_LEECHCORE:
 
         def readline(self, __size: Optional[int] = ...) -> bytes:
             data = b""
+            if not __size:
+                __size = 0
             while __size > self._chunk_size or __size < 0:
                 data += self.read(self._chunk_size)
                 index = data.find(b"\n")

volatility3/framework/plugins/linux/check_idt.py

@@ -5,6 +5,7 @@
 import logging
 from typing import List
 
+import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
 from volatility3.framework import interfaces, renderers, symbols
 from volatility3.framework.configuration import requirements
 from volatility3.framework.renderers import format_hints
@@ -27,6 +28,11 @@ class Check_idt(interfaces.plugins.PluginInterface):
                 description="Linux kernel",
                 architectures=["Intel32", "Intel64"],
             ),
+            requirements.VersionRequirement(
+                name="linux_utilities_modules",
+                component=linux_utilities_modules.Modules,
+                version=(1, 0, 0),
+            ),
             requirements.VersionRequirement(
                 name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0)
             ),
@@ -99,8 +105,10 @@ class Check_idt(interfaces.plugins.PluginInterface):
 
                 idt_addr = idt_addr & address_mask
 
-            module_name, symbol_name = linux.LinuxUtilities.lookup_module_address(
-                vmlinux, handlers, idt_addr
+            module_name, symbol_name = (
+                linux_utilities_modules.Modules.lookup_module_address(
+                    self.context, vmlinux.name, handlers, idt_addr
+                )
             )
 
             yield (

volatility3/framework/plugins/linux/keyboard_notifiers.py

@@ -4,6 +4,7 @@
 
 import logging
 
+import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
 from volatility3.framework import interfaces, renderers, exceptions
 from volatility3.framework.configuration import requirements
 from volatility3.framework.renderers import format_hints
@@ -26,6 +27,11 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface):
                 description="Linux kernel",
                 architectures=["Intel32", "Intel64"],
             ),
+            requirements.VersionRequirement(
+                name="linux_utilities_modules",
+                component=linux_utilities_modules.Modules,
+                version=(1, 0, 0),
+            ),
             requirements.PluginRequirement(
                 name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
             ),
@@ -66,8 +72,10 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface):
         ):
             call_addr = call_back.notifier_call
 
-            module_name, symbol_name = linux.LinuxUtilities.lookup_module_address(
-                vmlinux, handlers, call_addr
+            module_name, symbol_name = (
+                linux_utilities_modules.Modules.lookup_module_address(
+                    self.context, vmlinux.name, handlers, call_addr
+                )
             )
 
             yield (0, [format_hints.Hex(call_addr), module_name, symbol_name])

volatility3/framework/plugins/linux/kthreads.py

@@ -4,6 +4,7 @@
 import logging
 from typing import List
 
+import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
 from volatility3.framework import constants, exceptions, interfaces, renderers
 from volatility3.framework.configuration import requirements
 from volatility3.framework.interfaces import plugins
@@ -20,7 +21,7 @@ class Kthreads(plugins.PluginInterface):
     """Enumerates kthread functions"""
 
     _required_framework_version = (2, 11, 0)
-    _version = (1, 0, 2)
+    _version = (1, 0, 3)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -30,6 +31,11 @@ class Kthreads(plugins.PluginInterface):
                 description="Linux kernel",
                 architectures=architectures.LINUX_ARCHS,
             ),
+            requirements.VersionRequirement(
+                name="linux_utilities_modules",
+                component=linux_utilities_modules.Modules,
+                version=(1, 0, 0),
+            ),
             requirements.VersionRequirement(
                 name="linuxutils", component=linux.LinuxUtilities, version=(2, 1, 0)
             ),
@@ -88,8 +94,10 @@ class Kthreads(plugins.PluginInterface):
                 if kthread.has_member("full_name")
                 else task_name
             )
-            module_name, symbol_name = linux.LinuxUtilities.lookup_module_address(
-                vmlinux, handlers, threadfn
+            module_name, symbol_name = (
+                linux_utilities_modules.Modules.lookup_module_address(
+                    self.context, vmlinux.name, handlers, threadfn
+                )
             )
 
             fields = [

volatility3/framework/plugins/linux/netfilter.py

@@ -5,6 +5,7 @@ from dataclasses import dataclass, field
 from abc import ABC, abstractmethod
 import logging
 
+import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
 from typing import Iterator, List, Tuple
 from volatility3 import framework
 from volatility3.framework import (
@@ -98,6 +99,20 @@ class AbstractNetfilter(ABC):
                 f"linux.LinuxUtilities version not suitable: required {linuxutils_required_version} found {linuxutils_current_version}"
             )
 
+        linux_utilities_modules_required_version = (
+            Netfilter._required_linux_utilities_modules_version
+        )
+        linux_utilities_modules_current_version = (
+            linux_utilities_modules.Modules._version
+        )
+        if not requirements.VersionRequirement.matches_required(
+            linux_utilities_modules_required_version,
+            linux_utilities_modules_current_version,
+        ):
+            raise exceptions.PluginRequirementException(
+                f"linux_utilities_modules.Modules version not suitable: required {linux_utilities_modules_required_version} found {linux_utilities_modules_current_version}"
+            )
+
         modules = lsmod.Lsmod.list_modules(context, kernel_module_name)
         self.handlers = linux.LinuxUtilities.generate_kernel_handler_info(
             context, kernel_module_name, modules
@@ -263,8 +278,10 @@ class AbstractNetfilter(ABC):
         """Helper to obtain the module and symbol name in the format needed for the
         output of this plugin.
         """
-        module_name, symbol_name = linux.LinuxUtilities.lookup_module_address(
-            self.vmlinux, self.handlers, addr
+        module_name, symbol_name = (
+            linux_utilities_modules.Modules.lookup_module_address(
+                self._context, self.vmlinux.name, self.handlers, addr
+            )
         )
 
         if module_name == "UNKNOWN":
@@ -677,6 +694,7 @@ class Netfilter(interfaces.plugins.PluginInterface):
 
     _version = (1, 1, 0)
 
+    _required_linux_utilities_modules_version = (1, 0, 0)
     _required_linuxutils_version = (2, 1, 0)
     _required_lsmod_version = (2, 0, 0)
 
@@ -688,6 +706,11 @@ class Netfilter(interfaces.plugins.PluginInterface):
                 description="Linux kernel",
                 architectures=["Intel32", "Intel64"],
             ),
+            requirements.VersionRequirement(
+                name="linux_utilities_modules",
+                component=linux_utilities_modules.Modules,
+                version=cls._required_linux_utilities_modules_version,
+            ),
             requirements.PluginRequirement(
                 name="lsmod", plugin=lsmod.Lsmod, version=cls._required_lsmod_version
             ),

volatility3/framework/plugins/linux/tty_check.py

@@ -5,6 +5,7 @@
 import logging
 from typing import List
 
+import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
 from volatility3.framework import interfaces, renderers, exceptions, constants
 from volatility3.framework.configuration import requirements
 from volatility3.framework.interfaces import plugins
@@ -29,6 +30,11 @@ class tty_check(plugins.PluginInterface):
                 description="Linux kernel",
                 architectures=["Intel32", "Intel64"],
             ),
+            requirements.VersionRequirement(
+                name="linux_utilities_modules",
+                component=linux_utilities_modules.Modules,
+                version=(1, 0, 0),
+            ),
             requirements.PluginRequirement(
                 name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
             ),
@@ -79,8 +85,10 @@ class tty_check(plugins.PluginInterface):
 
                 recv_buf = tty_dev.ldisc.ops.receive_buf
 
-                module_name, symbol_name = linux.LinuxUtilities.lookup_module_address(
-                    vmlinux, handlers, recv_buf
+                module_name, symbol_name = (
+                    linux_utilities_modules.Modules.lookup_module_address(
+                        self.context, vmlinux.name, handlers, recv_buf
+                    )
                 )
 
                 yield (0, (name, format_hints.Hex(recv_buf), module_name, symbol_name))

volatility3/framework/plugins/windows/envars.py

@@ -76,14 +76,14 @@ class Envars(interfaces.plugins.PluginInterface):
                     "CurrentControlSet\\Control\\Session Manager\\Environment"
                 )
                 sys = True
-            except KeyError:
-                with contextlib.suppress(KeyError):
+            except (KeyError, registry.RegistryFormatException):
+                with contextlib.suppress(KeyError, registry.RegistryFormatException):
                     key = hive.get_key(
                         "ControlSet001\\Control\\Session Manager\\Environment"
                     )
                     sys = True
             if sys:
-                with contextlib.suppress(KeyError):
+                with contextlib.suppress(KeyError, registry.RegistryFormatException):
                     for node in key.get_values():
                         try:
                             value_node_name = node.get_name()
@@ -100,11 +100,11 @@ class Envars(interfaces.plugins.PluginInterface):
                             continue
 
             ## The user-specific variables
-            with contextlib.suppress(KeyError):
+            with contextlib.suppress(KeyError, registry.RegistryFormatException):
                 key = hive.get_key("Environment")
                 ntuser = True
             if ntuser:
-                with contextlib.suppress(KeyError):
+                with contextlib.suppress(KeyError, registry.RegistryFormatException):
                     for node in key.get_values():
                         try:
                             value_node_name = node.get_name()
@@ -123,7 +123,7 @@ class Envars(interfaces.plugins.PluginInterface):
             ## The volatile user variables
             try:
                 key = hive.get_key("Volatile Environment")
-            except KeyError:
+            except (KeyError, registry.RegistryFormatException):
                 continue
             try:
                 for node in key.get_values():

volatility3/framework/plugins/windows/getservicesids.py

@@ -10,6 +10,7 @@ from typing import List
 
 from volatility3.framework import renderers, interfaces, constants, exceptions
 from volatility3.framework.configuration import requirements
+from volatility3.framework.layers import registry
 from volatility3.plugins.windows.registry import hivelist
 
 vollog = logging.getLogger(__name__)
@@ -86,10 +87,18 @@ class GetServiceSIDs(interfaces.plugins.PluginInterface):
             # Get ControlSet\Services.
             try:
                 services = hive.get_key(r"CurrentControlSet\Services")
-            except (KeyError, exceptions.InvalidAddressException):
+            except (
+                KeyError,
+                exceptions.InvalidAddressException,
+                registry.RegistryFormatException,
+            ):
                 try:
                     services = hive.get_key(r"ControlSet001\Services")
-                except (KeyError, exceptions.InvalidAddressException):
+                except (
+                    KeyError,
+                    exceptions.InvalidAddressException,
+                    registry.RegistryFormatException,
+                ):
                     continue
 
             if services:

volatility3/framework/plugins/windows/getsids.py

@@ -158,7 +158,11 @@ class GetSIDs(interfaces.plugins.PluginInterface):
                             layers.registry.RegistryFormatException,
                         ):
                             continue
-            except (KeyError, exceptions.InvalidAddressException):
+            except (
+                KeyError,
+                exceptions.InvalidAddressException,
+                layers.registry.RegistryFormatException,
+            ):
                 continue
 
         return sids

volatility3/framework/plugins/windows/handles.py

@@ -341,7 +341,7 @@ class Handles(interfaces.plugins.PluginInterface):
                         try:
                             obj_name = entry.NameInfo.Name.String
                         except (ValueError, exceptions.InvalidAddressException):
-                            obj_name = ""
+                            obj_name = None
 
                 except exceptions.InvalidAddressException:
                     vollog.log(
@@ -359,7 +359,7 @@ class Handles(interfaces.plugins.PluginInterface):
                         format_hints.Hex(entry.HandleValue),
                         obj_type,
                         format_hints.Hex(entry.GrantedAccess),
-                        obj_name,
+                        obj_name or renderers.NotAvailableValue(),
                     ),
                 )
 

volatility3/framework/plugins/windows/registry/userassist.py

@@ -13,7 +13,7 @@ from typing import Any, Generator, List, Tuple
 from volatility3.framework import constants, exceptions, interfaces, renderers
 from volatility3.framework.configuration import requirements
 from volatility3.framework.layers.physical import BufferDataLayer
-from volatility3.framework.layers.registry import RegistryHive
+from volatility3.framework.layers.registry import RegistryHive, RegistryFormatException
 from volatility3.framework.renderers import conversion, format_hints
 from volatility3.framework.symbols import intermed
 from volatility3.plugins.windows.registry import hivelist
@@ -167,10 +167,21 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
 
         self._determine_userassist_type()
 
-        userassist_node_path = hive.get_key(
-            "software\\microsoft\\windows\\currentversion\\explorer\\userassist",
-            return_list=True,
-        )
+        try:
+            userassist_node_path = hive.get_key(
+                "software\\microsoft\\windows\\currentversion\\explorer\\userassist",
+                return_list=True,
+            )
+        except RegistryFormatException as e:
+            vollog.warning(
+                f"Error accessing UserAssist key in {hive_name} at {hive.hive_offset:#x}: {e}"
+            )
+            return None
+        except KeyError:
+            vollog.warning(
+                f"UserAssist key not found in {hive_name} at {hive.hive_offset:#x}"
+            )
+            return None
 
         if not userassist_node_path:
             vollog.warning("list_userassist did not find a valid node_path (or None)")

volatility3/framework/plugins/windows/svcscan.py

@@ -15,7 +15,7 @@ from volatility3.framework import (
     symbols,
 )
 from volatility3.framework.configuration import requirements
-from volatility3.framework.layers import scanners
+from volatility3.framework.layers import scanners, registry
 from volatility3.framework.renderers import format_hints
 from volatility3.framework.symbols import intermed
 from volatility3.framework.symbols.windows import versions
@@ -159,12 +159,20 @@ class SvcScan(interfaces.plugins.PluginInterface):
                 return cast(
                     objects.StructType, hive.get_key(r"CurrentControlSet\Services")
                 )
-            except (KeyError, exceptions.InvalidAddressException):
+            except (
+                KeyError,
+                exceptions.InvalidAddressException,
+                registry.RegistryFormatException,
+            ):
                 try:
                     return cast(
                         objects.StructType, hive.get_key(r"ControlSet001\Services")
                     )
-                except (KeyError, exceptions.InvalidAddressException):
+                except (
+                    KeyError,
+                    exceptions.InvalidAddressException,
+                    registry.RegistryFormatException,
+                ):
                     vollog.log(
                         constants.LOGLEVEL_VVVV,
                         "Could not retrieve any control set from SYSTEM hive",

volatility3/framework/symbols/linux/__init__.py

@@ -8,8 +8,15 @@ import logging
 from abc import ABC, abstractmethod
 from typing import Iterator, List, Tuple, Optional, Union
 
+import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
 from volatility3 import framework
-from volatility3.framework import constants, exceptions, interfaces, objects
+from volatility3.framework import (
+    constants,
+    exceptions,
+    interfaces,
+    objects,
+    Deprecation,
+)
 from volatility3.framework.objects import utility
 from volatility3.framework.symbols import intermed
 from volatility3.framework.symbols.linux import extensions
@@ -81,7 +88,7 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
 class LinuxUtilities(interfaces.configuration.VersionableInterface):
     """Class with multiple useful linux functions."""
 
-    _version = (2, 2, 0)
+    _version = (2, 2, 1)
     _required_framework_version = (2, 0, 0)
 
     framework.require_interface_version(*_required_framework_version)
@@ -339,6 +346,10 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
                 yield fd_num, filp, full_path
 
     @classmethod
+    @Deprecation.deprecated_method(
+        replacement=linux_utilities_modules.Modules.mask_mods_list,
+        replacement_version=(1, 0, 0),
+    )
     def mask_mods_list(
         cls,
         context: interfaces.context.ContextInterface,
@@ -346,18 +357,11 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
         mods: Iterator[interfaces.objects.ObjectInterface],
     ) -> List[Tuple[str, int, int]]:
         """
+        DEPRECATED: use "volatility3.framework.symbols.linux.utilities.modules.Modules.mask_mods_list" instead.
+
         A helper function to mask the starting and end address of kernel modules
         """
-        mask = context.layers[layer_name].address_mask
-
-        return [
-            (
-                utility.array_to_string(mod.name),
-                mod.get_module_base() & mask,
-                (mod.get_module_base() & mask) + mod.get_core_size(),
-            )
-            for mod in mods
-        ]
+        return linux_utilities_modules.Modules.mask_mods_list(context, layer_name, mods)
 
     @classmethod
     def generate_kernel_handler_info(
@@ -382,41 +386,30 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
 
         return [
             (constants.linux.KERNEL_NAME, start_addr, end_addr)
-        ] + LinuxUtilities.mask_mods_list(context, kernel.layer_name, mods_list)
+        ] + linux_utilities_modules.Modules.mask_mods_list(
+            context, kernel.layer_name, mods_list
+        )
 
     @classmethod
+    @Deprecation.deprecated_method(
+        replacement=linux_utilities_modules.Modules.lookup_module_address,
+        replacement_version=(1, 0, 0),
+    )
     def lookup_module_address(
         cls,
         kernel_module: interfaces.context.ModuleInterface,
         handlers: List[Tuple[str, int, int]],
         target_address: int,
-    ):
+    ) -> Tuple[str, str]:
         """
+        DEPRECATED: use "volatility3.framework.symbols.linux.utilities.modules.Modules.lookup_module_address" instead.
+
         Searches between the start and end address of the kernel module using target_address.
         Returns the module and symbol name of the address provided.
         """
-
-        mod_name = "UNKNOWN"
-        symbol_name = "N/A"
-
-        for name, start, end in handlers:
-            if start <= target_address <= end:
-                mod_name = name
-                if name == constants.linux.KERNEL_NAME:
-                    symbols = list(
-                        kernel_module.get_symbols_by_absolute_location(target_address)
-                    )
-
-                    if len(symbols):
-                        symbol_name = (
-                            symbols[0].split(constants.BANG)[1]
-                            if constants.BANG in symbols[0]
-                            else symbols[0]
-                        )
-
-                break
-
-        return mod_name, symbol_name
+        return linux_utilities_modules.Modules.lookup_module_address(
+            kernel_module.context, kernel_module.name, handlers, target_address
+        )
 
     @classmethod
     def walk_internal_list(cls, vmlinux, struct_name, list_member, list_start):

volatility3/framework/symbols/linux/utilities/modules.py

@@ -1,7 +1,9 @@
+from typing import Iterable, Iterator, List, Optional, Tuple
+
 from volatility3 import framework
-from volatility3.framework import interfaces
-from volatility3.framework.symbols.linux import extensions, LinuxUtilities
-from typing import Iterable, Optional
+from volatility3.framework import constants, interfaces
+from volatility3.framework.objects import utility
+from volatility3.framework.symbols.linux import extensions
 
 
 class Modules(interfaces.configuration.VersionableInterface):
@@ -12,8 +14,9 @@ class Modules(interfaces.configuration.VersionableInterface):
 
     framework.require_interface_version(*_required_framework_version)
 
-    @staticmethod
+    @classmethod
     def module_lookup_by_address(
+        cls,
         context: interfaces.context.ContextInterface,
         layer_name: str,
         modules: Iterable[extensions.module],
@@ -31,10 +34,64 @@ class Modules(interfaces.configuration.VersionableInterface):
         """
 
         for module in modules:
-            _, start, end = LinuxUtilities.mask_mods_list(
-                context, layer_name, [module]
-            )[0]
+            _, start, end = cls.mask_mods_list(context, layer_name, [module])[0]
             if start <= target_address <= end:
                 return module
 
         return None
+
+    @classmethod
+    def mask_mods_list(
+        cls,
+        context: interfaces.context.ContextInterface,
+        layer_name: str,
+        mods: Iterator[interfaces.objects.ObjectInterface],
+    ) -> List[Tuple[str, int, int]]:
+        """
+        A helper function to mask the starting and end address of kernel modules
+        """
+        mask = context.layers[layer_name].address_mask
+
+        return [
+            (
+                utility.array_to_string(mod.name),
+                mod.get_module_base() & mask,
+                (mod.get_module_base() & mask) + mod.get_core_size(),
+            )
+            for mod in mods
+        ]
+
+    @classmethod
+    def lookup_module_address(
+        cls,
+        context: interfaces.context.ContextInterface,
+        kernel_module_name: str,
+        handlers: List[Tuple[str, int, int]],
+        target_address: int,
+    ) -> Tuple[str, str]:
+        """
+        Searches between the start and end address of the kernel module using target_address.
+        Returns the module and symbol name of the address provided.
+        """
+        kernel_module = context.modules[kernel_module_name]
+        mod_name = "UNKNOWN"
+        symbol_name = "N/A"
+
+        for name, start, end in handlers:
+            if start <= target_address <= end:
+                mod_name = name
+                if name == constants.linux.KERNEL_NAME:
+                    symbols = list(
+                        kernel_module.get_symbols_by_absolute_location(target_address)
+                    )
+
+                    if len(symbols):
+                        symbol_name = (
+                            symbols[0].split(constants.BANG)[1]
+                            if constants.BANG in symbols[0]
+                            else symbols[0]
+                        )
+
+                break
+
+        return mod_name, symbol_name

volatility3/framework/symbols/windows/extensions/registry.py

@@ -133,8 +133,17 @@ class CM_KEY_BODY(objects.StructType):
 
     def get_full_key_name(self) -> str:
         output = []
+        seen = set()
+
         kcb = self.KeyControlBlock
         while kcb.ParentKcb:
+            if kcb.ParentKcb.vol.offset in seen:
+                return None
+            seen.add(kcb.ParentKcb.vol.offset)
+
+            if len(output) > 128:
+                return None
+
             if kcb.NameBlock.Name is None:
                 break