Merge branch 'develop' into linux_parity_release_harden_mountinfo_api_and_related_fixes

.github/workflows/build-pyinstaller.yml

@@ -0,0 +1,50 @@
+name: build-pyinstaller
+on:
+  push:
+    branches:
+      - stable
+      - develop
+      - 'release/**'
+  pull_request:
+    branches:
+      - stable
+      - 'release/**'
+
+jobs:
+
+  exe:
+    runs-on: windows-latest
+    strategy:
+      matrix:
+        python-version: ["3.11"]
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install pyinstaller
+    
+    - name: Pyinstall executable
+      run: |
+        pyinstaller --clean -y vol.spec
+        pyinstaller --clean -y volshell.spec
+
+    - name: Move files
+      run: |
+        mv dist/vol.exe vol.exe
+        mv dist/volshell.exe volshell.exe
+
+    - name: Archive 
+      uses: actions/upload-artifact@v4
+      with:
+        name: volatility3-pyinstaller
+        path: |
+          vol.exe
+          volshell.exe
+          README.md
+          LICENSE.txt

README.md

@@ -88,7 +88,7 @@ The latest generated copy of the documentation can be found at: <https://volatil
 
 ## Licensing and Copyright
 
-Copyright (C) 2007-2024 Volatility Foundation
+Copyright (C) 2007-2025 Volatility Foundation
 
 All Rights Reserved
 

doc/source/conf.py

@@ -167,7 +167,7 @@ master_doc = "index"
 
 # General information about the project.
 project = "Volatility 3"
-copyright = "2012-2024, Volatility Foundation"
+copyright = "2012-2025, Volatility Foundation"
 
 # The version info for the project you're documenting, acts as replacement for
 # |version| and |release|, also used in various other places throughout the

pyproject.toml

@@ -1,7 +1,15 @@
 [project]
 name = "volatility3"
 description = "Memory forensics framework"
-keywords = ["volatility", "memory", "forensics", "framework", "windows", "linux", "volshell"]
+keywords = [
+    "volatility",
+    "memory",
+    "forensics",
+    "framework",
+    "windows",
+    "linux",
+    "volshell",
+]
 readme = "README.md"
 authors = [
     { name = "Volatility Foundation", email = "volatility@volatilityfoundation.org" },
@@ -10,9 +18,7 @@ requires-python = ">=3.8.0"
 license = { text = "VSL" }
 dynamic = ["version"]
 
-dependencies = [
-    "pefile>=2024.8.26",
-]
+dependencies = ["pefile>=2024.8.26"]
 
 [project.optional-dependencies]
 full = [
@@ -26,15 +32,12 @@ full = [
     "pillow>=10.0.0,<11.0.0",
 ]
 
-cloud = [
-    "gcsfs>=2024.10.0",
-    "s3fs>=2024.10.0",
-]
+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",
 ]
@@ -48,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",
 ]
 
@@ -79,16 +82,16 @@ target-version = "py38"
 
 [tool.ruff.lint]
 select = [
-	"F",       # pyflakes
-	"E",       # pycodestyle errors
-	"W",       # pycodestyle warnings
-	"G",       # flake8-logging-format
-	"PIE",     # flake8-pie
-	"UP",      # pyupgrade
+    "F",   # pyflakes
+    "E",   # pycodestyle errors
+    "W",   # pycodestyle warnings
+    "G",   # flake8-logging-format
+    "PIE", # flake8-pie
+    "UP",  # pyupgrade
 ]
 
 ignore = [
-    "E501",    # ignore due to conflict with formatter
+    "E501", # ignore due to conflict with formatter
 ]
 
 [build-system]

test/test_volatility.py

@@ -708,6 +708,24 @@ def test_linux_page_cache_inodepages(image, volatility, python):
 
     inode_address = hex(0x88001AB5C270)
     inode_dump_filename = f"inode_{inode_address}.dmp"
+
+    rc, out, _err = runvol_plugin(
+        "linux.pagecache.InodePages",
+        image,
+        volatility,
+        python,
+        pluginargs=["--inode", inode_address],
+    )
+
+    assert rc == 0
+    assert out.count(b"\n") > 4
+
+    # PageVAddr PagePAddr MappingAddr .. DumpSafe
+    assert re.search(
+        rb"0xea000054c5f8\s0x18389000\s0x88001ab5c3b0.*?True",
+        out,
+    )
+
     try:
         rc, out, _err = runvol_plugin(
             "linux.pagecache.InodePages",
@@ -718,13 +736,8 @@ def test_linux_page_cache_inodepages(image, volatility, python):
         )
 
         assert rc == 0
-        assert out.count(b"\n") > 4
+        assert out.count(b"\n") >= 4
 
-        # PageVAddr PagePAddr MappingAddr .. DumpSafe
-        assert re.search(
-            rb"0xea000054c5f8\s0x18389000\s0x88001ab5c3b0.*?True",
-            out,
-        )
         assert os.path.exists(inode_dump_filename)
         with open(inode_dump_filename, "rb") as fp:
             inode_contents = fp.read()

volatility3/cli/__init__.py

@@ -363,11 +363,21 @@ class CommandLine:
             metavar="PLUGIN",
         )
         for plugin in sorted(plugin_list):
+            # First line of a plugin docstring will be the short description for -h.
+            # Text after the first two consecutive new lines will be
+            # the additional description (argparse epilog).
+            short_help = additional_help = None
+            if plugin_list[plugin].__doc__ is not None:
+                doc_split = plugin_list[plugin].__doc__.split("\n\n", 1)
+                short_help = doc_split[0].strip()
+                if len(doc_split) > 1:
+                    additional_help = doc_split[1].strip()
+
             plugin_parser = subparser.add_parser(
                 plugin,
-                help=plugin_list[plugin].__doc__,
-                description=plugin_list[plugin].__doc__,
-                epilog=plugin_list[plugin].additional_description,
+                help=short_help,
+                description=short_help,
+                epilog=additional_help,
             )
             self.populate_requirements_argparse(plugin_parser, plugin_list[plugin])
 
@@ -573,6 +583,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 +629,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 +640,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/linux.py

@@ -71,6 +71,11 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
                 elif "init_level4_pgt" in table.symbols:
                     layer_class = intel.LinuxIntel32e
                     dtb_symbol_name = "init_level4_pgt"
+                elif "pkmap_count" in table.symbols and table.get_symbol(
+                    "pkmap_count"
+                ).type.count in (512, 2048):
+                    layer_class = intel.LinuxIntelPAE
+                    dtb_symbol_name = "swapper_pg_dir"
                 else:
                     layer_class = intel.LinuxIntel
                     dtb_symbol_name = "swapper_pg_dir"

volatility3/framework/automagic/pdbscan.py

@@ -376,8 +376,74 @@ 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):
+            try:
+                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
+            except exceptions.InvalidAddressException:
+                continue
+
+        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/_version.py

@@ -1,6 +1,6 @@
 # We use the SemVer 2.0.0 versioning scheme
 VERSION_MAJOR = 2  # Number of releases of the library with a breaking change
-VERSION_MINOR = 16  # Number of changes that only add to the interface
+VERSION_MINOR = 19  # Number of changes that only add to the interface
 VERSION_PATCH = 0  # Number of changes that do not change the interface
 VERSION_SUFFIX = ""
 

volatility3/framework/constants/linux/__init__.py

@@ -6,6 +6,7 @@
 Linux-specific values that aren't found in debug symbols
 """
 from enum import IntEnum, Flag
+from dataclasses import dataclass
 
 KERNEL_NAME = "__kernel__"
 
@@ -352,3 +353,57 @@ NSEC_PER_SEC = 1e9
 MODULE_MAXIMUM_CORE_SIZE = 20000000
 MODULE_MAXIMUM_CORE_TEXT_SIZE = 20000000
 MODULE_MINIMUM_SIZE = 4096
+
+
+@dataclass
+class TaintFlag:
+    shift: int
+    desc: str
+    when_present: bool
+    module: bool
+
+
+TAINT_FLAGS = {
+    "P": TaintFlag(
+        shift=1 << 0, desc="PROPRIETARY_MODULE", when_present=True, module=True
+    ),
+    "G": TaintFlag(
+        shift=1 << 0, desc="PROPRIETARY_MODULE", when_present=False, module=True
+    ),
+    "F": TaintFlag(shift=1 << 1, desc="FORCED_MODULE", when_present=True, module=False),
+    "S": TaintFlag(
+        shift=1 << 2, desc="CPU_OUT_OF_SPEC", when_present=True, module=False
+    ),
+    "R": TaintFlag(shift=1 << 3, desc="FORCED_RMMOD", when_present=True, module=False),
+    "M": TaintFlag(shift=1 << 4, desc="MACHINE_CHECK", when_present=True, module=False),
+    "B": TaintFlag(shift=1 << 5, desc="BAD_PAGE", when_present=True, module=False),
+    "U": TaintFlag(shift=1 << 6, desc="USER", when_present=True, module=False),
+    "D": TaintFlag(shift=1 << 7, desc="DIE", when_present=True, module=False),
+    "A": TaintFlag(
+        shift=1 << 8, desc="OVERRIDDEN_ACPI_TABLE", when_present=True, module=False
+    ),
+    "W": TaintFlag(shift=1 << 9, desc="WARN", when_present=True, module=False),
+    "C": TaintFlag(shift=1 << 10, desc="CRAP", when_present=True, module=True),
+    "I": TaintFlag(
+        shift=1 << 11, desc="FIRMWARE_WORKAROUND", when_present=True, module=False
+    ),
+    "O": TaintFlag(shift=1 << 12, desc="OOT_MODULE", when_present=True, module=True),
+    "E": TaintFlag(
+        shift=1 << 13, desc="UNSIGNED_MODULE", when_present=True, module=True
+    ),
+    "L": TaintFlag(shift=1 << 14, desc="SOFTLOCKUP", when_present=True, module=False),
+    "K": TaintFlag(shift=1 << 15, desc="LIVEPATCH", when_present=True, module=True),
+    "X": TaintFlag(shift=1 << 16, desc="AUX", when_present=True, module=True),
+    "T": TaintFlag(shift=1 << 17, desc="RANDSTRUCT", when_present=True, module=True),
+    "N": TaintFlag(shift=1 << 18, desc="TEST", when_present=True, module=True),
+}
+"""Flags used to taint kernel and modules, for debugging purposes.
+
+Map based on 6.12-rc5.
+
+Documentation :
+    - https://www.kernel.org/doc/Documentation/admin-guide/sysctl/kernel.rst#:~:text=guide/sysrq.rst.-,tainted,-%3D%3D%3D%3D%3D%3D%3D%0A%0ANon%2Dzero%20if
+    - https://www.kernel.org/doc/Documentation/admin-guide/tainted-kernels.rst#:~:text=More%20detailed%20explanation%20for%20tainting
+    - taint_flag kernel struct
+    - taint_flags kernel constant
+"""

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
@@ -337,7 +338,7 @@ class Module(interfaces.context.ModuleInterface):
         )
 
     @property
-    def symbols(self):
+    def symbols(self) -> Iterable[str]:
         return self.context.symbol_space[self.symbol_table_name].symbols
 
     get_symbol = get_module_wrapper("get_symbol")
@@ -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):
@@ -130,3 +131,35 @@ class OfflineException(VolatilityException):
 
 class RenderException(VolatilityException):
     """Thrown if there is an error during rendering"""
+
+
+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) -> List:
-        """Lists the symbols contained in the symbol table for this module"""
+    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/interfaces/plugins.py

@@ -112,8 +112,6 @@ class PluginInterface(
     # Be careful with inheritance around this (We default to requiring a version which doesn't exist, so it must be set)
     _required_framework_version: Tuple[int, int, int] = (0, 0, 0)
     """The _version variable is a quick way for plugins to define their current interface, it should follow SemVer rules"""
-    additional_description: str = None
-    """Display additional description of the plugin after the description of the arguments. See: https://docs.python.org/3/library/argparse.html#epilog"""
 
     def __init__(
         self,

volatility3/framework/interfaces/symbols.py

@@ -122,7 +122,7 @@ class BaseSymbolTableInterface:
 
     @property
     def symbols(self) -> Iterable[str]:
-        """Returns an iterator of the Symbol names."""
+        """Returns an iterable of the available symbol names."""
         raise NotImplementedError(
             "Abstract property symbols not implemented by subclass."
         )
@@ -131,7 +131,7 @@ class BaseSymbolTableInterface:
 
     @property
     def types(self) -> Iterable[str]:
-        """Returns an iterator of the Symbol type names."""
+        """Returns an iterable of the available symbol type names."""
         raise NotImplementedError(
             "Abstract property types not implemented by subclass."
         )
@@ -149,7 +149,7 @@ class BaseSymbolTableInterface:
 
     @property
     def enumerations(self) -> Iterable[Any]:
-        """Returns an iterator of the Enumeration names."""
+        """Returns an iterable of the available enumerations."""
         raise NotImplementedError(
             "Abstract property enumerations not implemented by subclass."
         )
@@ -366,6 +366,7 @@ class NativeTableInterface(BaseSymbolTableInterface):
 
     @property
     def symbols(self) -> Iterable[str]:
+        """Returns an iterable of the available symbol names."""
         return []
 
     def get_enumeration(self, name: str) -> objects.Template:
@@ -374,7 +375,13 @@ class NativeTableInterface(BaseSymbolTableInterface):
         )
 
     @property
-    def enumerations(self) -> Iterable[str]:
+    def enumerations(self) -> Iterable[Any]:
+        """Returns an iterable of the available enumerations."""
+        return []
+
+    @property
+    def types(self) -> Iterable[str]:
+        """Returns an iterable of the available symbol type names."""
         return []
 
 

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/configwriter.py

@@ -14,8 +14,8 @@ vollog = logging.getLogger(__name__)
 
 
 class ConfigWriter(plugins.PluginInterface):
-    """Runs the automagics and both prints and outputs configuration in the
-    output directory."""
+    """Runs the automagics and both prints and outputs configuration in the \
+output directory."""
 
     _required_framework_version = (2, 0, 0)
 

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/check_modules.py

@@ -18,6 +18,7 @@ vollog = logging.getLogger(__name__)
 class Check_modules(plugins.PluginInterface):
     """Compares module list to sysfs info, if available"""
 
+    _version = (1, 0, 0)
     _required_framework_version = (2, 0, 0)
 
     @classmethod

volatility3/framework/plugins/linux/envars.py

@@ -5,7 +5,7 @@
 import logging
 from typing import Iterable, Tuple
 
-from volatility3.framework import renderers, interfaces, exceptions
+from volatility3.framework import renderers, interfaces
 from volatility3.framework.configuration import requirements
 from volatility3.framework.interfaces import plugins
 from volatility3.framework.objects import utility
@@ -58,16 +58,10 @@ class Envars(plugins.PluginInterface):
             Tuples of (key, value) representing each environment variable.
         """
 
-        # This ensures the `task` is valid as well as its
-        # memory mapping structures
-        try:
-            task_name = utility.array_to_string(task.comm)
-            env_start = task.mm.env_start
-            env_end = task.mm.env_end
-        except exceptions.InvalidAddressException:
-            return None
-
+        task_name = utility.array_to_string(task.comm)
         task_pid = task.pid
+        env_start = task.mm.env_start
+        env_end = task.mm.env_end
         env_area_size = env_end - env_start
         if not (0 < env_area_size <= env_area_max_size):
             vollog.debug(

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/modxview.py

@@ -0,0 +1,189 @@
+# 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, Dict, Iterator
+from volatility3.plugins.linux import lsmod, check_modules, hidden_modules
+from volatility3.framework import interfaces
+from volatility3.framework.configuration import requirements
+from volatility3.framework.renderers import format_hints, TreeGrid, NotAvailableValue
+from volatility3.framework.symbols.linux import extensions
+from volatility3.framework.constants import architectures
+from volatility3.framework.symbols.linux.utilities import tainting
+
+vollog = logging.getLogger(__name__)
+
+
+class Modxview(interfaces.plugins.PluginInterface):
+    """Centralize lsmod, check_modules and hidden_modules results to efficiently \
+spot modules presence and taints."""
+
+    _version = (1, 0, 0)
+    _required_framework_version = (2, 17, 0)
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=architectures.LINUX_ARCHS,
+            ),
+            requirements.VersionRequirement(
+                name="linux-tainting", component=tainting.Tainting, version=(1, 0, 0)
+            ),
+            requirements.PluginRequirement(
+                name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
+            ),
+            requirements.PluginRequirement(
+                name="check_modules",
+                plugin=check_modules.Check_modules,
+                version=(1, 0, 0),
+            ),
+            requirements.PluginRequirement(
+                name="hidden_modules",
+                plugin=hidden_modules.Hidden_modules,
+                version=(1, 0, 0),
+            ),
+            requirements.BooleanRequirement(
+                name="plain_taints",
+                description="Display the plain taints string for each module.",
+                optional=True,
+                default=False,
+            ),
+        ]
+
+    @classmethod
+    def flatten_run_modules_results(
+        cls, run_results: Dict[str, List[extensions.module]], deduplicate: bool = True
+    ) -> Iterator[extensions.module]:
+        """Flatten a dictionary mapping plugin names and modules list, to a single merged list.
+        This is useful to get a generic lookup list of all the detected modules.
+
+        Args:
+            run_results: dictionary of plugin names mapping a list of detected modules
+            deduplicate: remove duplicate modules, based on their offsets
+
+        Returns:
+            Iterator of modules objects
+        """
+        seen_addresses = set()
+        for modules in run_results.values():
+            for module in modules:
+                if deduplicate and module.vol.offset in seen_addresses:
+                    continue
+                seen_addresses.add(module.vol.offset)
+                yield module
+
+    @classmethod
+    def run_modules_scanners(
+        cls,
+        context: interfaces.context.ContextInterface,
+        kernel_name: str,
+        run_hidden_modules: bool = True,
+    ) -> Dict[str, List[extensions.module]]:
+        """Run module scanning plugins and aggregate the results. It is designed
+        to not operate any inter-plugin results triage.
+
+        Args:
+            run_hidden_modules: specify if the hidden_modules plugin should be run
+        Returns:
+            Dictionary mapping each plugin to its corresponding result
+        """
+
+        kernel = context.modules[kernel_name]
+        run_results = {}
+        # lsmod
+        run_results["lsmod"] = list(lsmod.Lsmod.list_modules(context, kernel_name))
+        # check_modules
+        sysfs_modules: dict = check_modules.Check_modules.get_kset_modules(
+            context, kernel_name
+        )
+        ## Convert get_kset_modules() offsets back to module objects
+        run_results["check_modules"] = [
+            kernel.object(object_type="module", offset=m_offset, absolute=True)
+            for m_offset in sysfs_modules.values()
+        ]
+        # hidden_modules
+        if run_hidden_modules:
+            known_modules_addresses = set(
+                context.layers[kernel.layer_name].canonicalize(module.vol.offset)
+                for module in run_results["lsmod"] + run_results["check_modules"]
+            )
+            modules_memory_boundaries = (
+                hidden_modules.Hidden_modules.get_modules_memory_boundaries(
+                    context, kernel_name
+                )
+            )
+            run_results["hidden_modules"] = list(
+                hidden_modules.Hidden_modules.get_hidden_modules(
+                    context,
+                    kernel_name,
+                    known_modules_addresses,
+                    modules_memory_boundaries,
+                )
+            )
+
+        return run_results
+
+    def _generator(self):
+        kernel_name = self.config["kernel"]
+        run_results = self.run_modules_scanners(self.context, kernel_name)
+        aggregated_modules = {}
+        # We want to be explicit on the plugins results we are interested in
+        for plugin_name in ["lsmod", "check_modules", "hidden_modules"]:
+            # Iterate over each recovered module
+            for module in run_results[plugin_name]:
+                # Use offsets as unique keys, whether a module
+                # appears in many plugin runs or not
+                if aggregated_modules.get(module.vol.offset, None) is not None:
+                    # Append the plugin to the list of originating plugins
+                    aggregated_modules[module.vol.offset][1].append(plugin_name)
+                else:
+                    aggregated_modules[module.vol.offset] = (module, [plugin_name])
+
+        for module_offset, (module, originating_plugins) in aggregated_modules.items():
+            # Tainting parsing capabilities applied to the module
+            if self.config.get("plain_taints"):
+                taints = tainting.Tainting.get_taints_as_plain_string(
+                    self.context,
+                    kernel_name,
+                    module.taints,
+                    True,
+                )
+            else:
+                taints = ",".join(
+                    tainting.Tainting.get_taints_parsed(
+                        self.context,
+                        kernel_name,
+                        module.taints,
+                        True,
+                    )
+                )
+
+            yield (
+                0,
+                (
+                    module.get_name() or NotAvailableValue(),
+                    format_hints.Hex(module_offset),
+                    "lsmod" in originating_plugins,
+                    "check_modules" in originating_plugins,
+                    "hidden_modules" in originating_plugins,
+                    taints or NotAvailableValue(),
+                ),
+            )
+
+    def run(self):
+        columns = [
+            ("Name", str),
+            ("Address", format_hints.Hex),
+            ("In procfs", bool),
+            ("In sysfs", bool),
+            ("Hidden", bool),
+            ("Taints", str),
+        ]
+
+        return TreeGrid(
+            columns,
+            self._generator(),
+        )

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/pagecache.py

@@ -6,9 +6,9 @@ import math
 import logging
 import datetime
 from dataclasses import dataclass, astuple
-from typing import List, Set, Type, Iterable
+from typing import List, Set, Type, Iterable, Tuple
 
-from volatility3.framework import renderers, interfaces
+from volatility3.framework import renderers, interfaces, exceptions
 from volatility3.framework.renderers import format_hints
 from volatility3.framework.interfaces import plugins
 from volatility3.framework.configuration import requirements
@@ -104,7 +104,7 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
 
     _required_framework_version = (2, 0, 0)
 
-    _version = (1, 0, 1)
+    _version = (1, 0, 2)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -147,7 +147,13 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
             Otherwise, it returns the same symlink_path
         """
         # i_link (fast symlinks) were introduced in 4.2
-        if inode and inode.is_link and inode.has_member("i_link") and inode.i_link:
+        if (
+            inode
+            and inode.is_link
+            and inode.has_member("i_link")
+            and inode.i_link
+            and inode.i_link.is_readable()
+        ):
             i_link_str = inode.i_link.dereference().cast(
                 "string", max_length=255, encoding="utf-8", errors="replace"
             )
@@ -253,6 +259,10 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
             if not root_inode.is_valid():
                 continue
 
+            if not (root_inode.i_mapping and root_inode.i_mapping.is_readable()):
+                # Retrieving data from the page cache requires a valid address space
+                continue
+
             # Inode already processed?
             if root_inode_ptr in seen_inodes:
                 continue
@@ -284,6 +294,10 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
                 if not file_inode.is_valid():
                     continue
 
+                if not (file_inode.i_mapping and file_inode.i_mapping.is_readable()):
+                    # Retrieving data from the page cache requires a valid address space
+                    continue
+
                 # Inode already processed?
                 if file_inode_ptr in seen_inodes:
                     continue
@@ -316,10 +330,12 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
             if self.config["find"]:
                 if inode_in.path == self.config["find"]:
                     inode_out = inode_in.to_user(vmlinux_layer)
+
                     yield (0, astuple(inode_out))
                     break  # Only the first match
             else:
                 inode_out = inode_in.to_user(vmlinux_layer)
+
                 yield (0, astuple(inode_out))
 
     def generate_timeline(self):
@@ -389,7 +405,7 @@ class InodePages(plugins.PluginInterface):
 
     _required_framework_version = (2, 0, 0)
 
-    _version = (2, 0, 0)
+    _version = (2, 0, 1)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -443,28 +459,80 @@ class InodePages(plugins.PluginInterface):
         # created, saving both disk space and I/O time.
         # Additionally, using the page index will guarantee that each page is written at the
         # appropriate file position.
+        inode_size = inode.i_size
         try:
-            with open_method(filename) as f:
-                inode_size = inode.i_size
-                f.truncate(inode_size)
-
+            file_initialized = False
+            with open_method(filename) as file_obj:
                 for page_idx, page_content in inode.get_contents():
                     current_fp = page_idx * vmlinux_layer.page_size
                     max_length = inode_size - current_fp
-                    page_bytes = page_content[:max_length]
-                    if current_fp + len(page_bytes) > inode_size:
+                    page_bytes_len = min(max_length, len(page_content))
+                    if (
+                        current_fp >= inode_size
+                        or current_fp + page_bytes_len > inode_size
+                    ):
                         vollog.error(
                             "Page out of file bounds: inode 0x%x, inode size %d, page index %d",
                             inode.vol.offset,
                             inode_size,
                             page_idx,
                         )
-                    f.seek(current_fp)
-                    f.write(page_bytes)
+                        continue
+                    page_bytes = page_content[:page_bytes_len]
 
+                    if not file_initialized:
+                        # Lazy initialization to avoid truncating the file until we are
+                        # certain there is something to write
+                        file_obj.truncate(inode_size)
+                        file_initialized = True
+
+                    file_obj.seek(current_fp)
+                    file_obj.write(page_bytes)
+        except exceptions.LinuxPageCacheException:
+            vollog.error(
+                f"Error dumping cached pages for inode at {inode.vol.offset:#x}"
+            )
         except OSError as e:
             vollog.error("Unable to write to file (%s): %s", filename, e)
 
+    def _generate_inode_fields(
+        self,
+        inode: interfaces.objects.ObjectInterface,
+        vmlinux_layer: interfaces.layers.TranslationLayerInterface,
+    ) -> Iterable[Tuple[int, int, int, int, bool, str]]:
+        inode_size = inode.i_size
+        try:
+            for page_obj in inode.get_pages():
+                if page_obj.mapping != inode.i_mapping:
+                    vollog.warning(
+                        f"Cached page at {page_obj.vol.offset:#x} has a mismatched address space with the inode. Skipping page"
+                    )
+                    continue
+                page_vaddr = page_obj.vol.offset
+                page_paddr = page_obj.to_paddr()
+                page_mapping_addr = page_obj.mapping
+                page_index = page_obj.index
+                page_file_offset = page_index * vmlinux_layer.page_size
+                dump_safe = (
+                    page_file_offset < inode_size
+                    and page_mapping_addr
+                    and page_mapping_addr.is_readable()
+                )
+                page_flags_list = page_obj.get_flags_list()
+                page_flags = ",".join([x.replace("PG_", "") for x in page_flags_list])
+                fields = (
+                    page_vaddr,
+                    page_paddr,
+                    page_mapping_addr,
+                    page_index,
+                    dump_safe,
+                    page_flags,
+                )
+
+                yield 0, fields
+        except exceptions.LinuxPageCacheException:
+            vollog.warning(f"Page cache for inode at {inode.vol.offset:#x} is corrupt")
+
     def _generator(self):
         vmlinux_module_name = self.config["kernel"]
         vmlinux = self.context.modules[vmlinux_module_name]
@@ -486,7 +554,6 @@ class InodePages(plugins.PluginInterface):
             else:
                 vollog.error("Unable to find inode with path %s", self.config["find"])
                 return None
-
         elif self.config["inode"]:
             inode = vmlinux.object("inode", self.config["inode"], absolute=True)
         else:
@@ -501,27 +568,6 @@ class InodePages(plugins.PluginInterface):
             vollog.error("The inode is not a regular file")
             return None
 
-        inode_size = inode.i_size
-        for page_obj in inode.get_pages():
-            page_vaddr = page_obj.vol.offset
-            page_paddr = page_obj.to_paddr()
-            page_mapping_addr = page_obj.mapping
-            page_index = int(page_obj.index)
-            page_file_offset = page_index * vmlinux_layer.page_size
-            dump_safe = page_file_offset < inode_size
-            page_flags_list = page_obj.get_flags_list()
-            page_flags = ",".join([x.replace("PG_", "") for x in page_flags_list])
-            fields = (
-                page_vaddr,
-                page_paddr,
-                page_mapping_addr,
-                page_index,
-                dump_safe,
-                page_flags,
-            )
-
-            yield 0, fields
-
         if self.config["dump"]:
             open_method = self.open
             inode_address = inode.vol.offset
@@ -530,6 +576,8 @@ class InodePages(plugins.PluginInterface):
             self.write_inode_content_to_file(
                 inode, filename, open_method, vmlinux_layer
             )
+        else:
+            yield from self._generate_inode_fields(inode, vmlinux_layer)
 
     def run(self):
         headers = [

volatility3/framework/plugins/linux/pslist.py

@@ -34,7 +34,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
     """Lists the processes present in a particular linux memory image."""
 
     _required_framework_version = (2, 13, 0)
-    _version = (4, 0, 0)
+    _version = (4, 1, 0)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -179,6 +179,10 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                 file_output = "VMA start matching task start_code not found"
         return file_output
 
+    @staticmethod
+    def _format_cred(cred):
+        return renderers.NotAvailableValue() if cred is None else cred
+
     def _generator(
         self,
         pid_filter: Callable[[Any], bool],
@@ -212,16 +216,21 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
 
             task_fields = self.get_task_fields(task, decorate_comm)
 
+            task_uid = self._format_cred(task_fields.uid)
+            task_gid = self._format_cred(task_fields.gid)
+            task_euid = self._format_cred(task_fields.euid)
+            task_egid = self._format_cred(task_fields.egid)
+
             yield 0, (
                 format_hints.Hex(task_fields.offset),
                 task_fields.user_pid,
                 task_fields.user_tid,
                 task_fields.user_ppid,
                 task_fields.name,
-                task_fields.uid or renderers.NotAvailableValue(),
-                task_fields.gid or renderers.NotAvailableValue(),
-                task_fields.euid or renderers.NotAvailableValue(),
-                task_fields.egid or renderers.NotAvailableValue(),
+                task_uid,
+                task_gid,
+                task_euid,
+                task_egid,
                 task_fields.creation_time or renderers.NotAvailableValue(),
                 file_output,
             )
@@ -250,6 +259,9 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
 
         # Note that the init_task itself is not yielded, since "ps" also never shows it.
         for task in init_task.tasks:
+            if not task.is_valid():
+                continue
+
             if filter_func(task):
                 continue
 

volatility3/framework/plugins/linux/pstree.py

@@ -9,8 +9,7 @@ from volatility3.plugins.linux import pslist
 
 
 class PsTree(interfaces.plugins.PluginInterface):
-    """Plugin for listing processes in a tree based on their parent process
-    ID."""
+    """Plugin for listing processes in a tree based on their parent process ID."""
 
     _required_framework_version = (2, 13, 0)
     _version = (1, 1, 1)

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/mac/mount.py

@@ -11,8 +11,8 @@ from volatility3.framework.symbols import mac
 
 
 class Mount(plugins.PluginInterface):
-    """A module containing a collection of plugins that produce data typically
-    found in Mac's mount command"""
+    """A module containing a collection of plugins that produce data typically \
+found in Mac's mount command"""
 
     _required_framework_version = (2, 0, 0)
 

volatility3/framework/plugins/mac/pstree.py

@@ -10,8 +10,7 @@ from volatility3.plugins.mac import pslist
 
 
 class PsTree(plugins.PluginInterface):
-    """Plugin for listing processes in a tree based on their parent process
-    ID."""
+    """Plugin for listing processes in a tree based on their parent process ID."""
 
     _required_framework_version = (2, 0, 0)
 

volatility3/framework/plugins/timeliner.py

@@ -41,8 +41,8 @@ class TimeLinerInterface(metaclass=abc.ABCMeta):
 
 
 class Timeliner(interfaces.plugins.PluginInterface):
-    """Runs all relevant plugins that provide time related information and
-    orders the results by time."""
+    """Runs all relevant plugins that provide time related information and \
+orders the results by time."""
 
     _required_framework_version = (2, 0, 0)
     _version = (1, 1, 0)

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/pstree.py

@@ -14,8 +14,7 @@ vollog = logging.getLogger(__name__)
 
 
 class PsTree(interfaces.plugins.PluginInterface):
-    """Plugin for listing processes in a tree based on their parent process
-    ID."""
+    """Plugin for listing processes in a tree based on their parent process ID."""
 
     _required_framework_version = (2, 0, 0)
 

volatility3/framework/plugins/windows/psxview.py

@@ -21,9 +21,10 @@ vollog = logging.getLogger(__name__)
 
 
 class PsXView(plugins.PluginInterface):
-    """Lists all processes found via four of the methods described in \"The Art of Memory Forensics,\" which may help
-    identify processes that are trying to hide themselves. I recommend using -r pretty if you are looking at this
-    plugin's output in a terminal."""
+    """Lists all processes found via four of the methods described in \"The Art of Memory Forensics\" which may help \
+identify processes that are trying to hide themselves.
+
+We recommend using -r pretty if you are looking at this plugin's output in a terminal."""
 
     # I've omitted the desktop thread scanning method because Volatility3 doesn't appear to have the functionality
     # which the original plugin used to do it.

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

@@ -12,8 +12,7 @@ from volatility3.plugins.windows import poolscanner, bigpools
 
 
 class HiveScan(interfaces.plugins.PluginInterface):
-    """Scans for registry hives present in a particular windows memory
-    image."""
+    """Scans for registry hives present in a particular windows memory image."""
 
     _required_framework_version = (2, 0, 0)
     _version = (1, 0, 0)

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/scheduled_tasks.py

@@ -1099,9 +1099,8 @@ class DynamicInfo:
 
 
 class ScheduledTasks(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
-    """Decodes scheduled task information from the Windows registry, including
-    information about triggers, actions, run times, and creation times.
-    """
+    """Decodes scheduled task information from the Windows registry, including \
+information about triggers, actions, run times, and creation times."""
 
     _required_framework_version = (2, 11, 0)
     _version = (1, 0, 0)

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/plugins/windows/vadyarascan.py

@@ -84,7 +84,7 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
 
             if not vad_maps_to_scan:
                 vollog.warning(
-                    f"No VADs were found for task {task.UniqueProcessID}, not scanning"
+                    f"No VADs were found for task {task.UniqueProcessId}, not scanning"
                 )
                 continue
 

volatility3/framework/symbols/intermed.py

@@ -411,18 +411,27 @@ class Version1Format(ISFormatTable):
 
     @property
     def symbols(self) -> Iterable[str]:
-        """Returns an iterator of the symbol names."""
-        return list(self._json_object.get("symbols", {}))
+        """Returns an iterable (KeysView) of the available symbol names."""
+        return self._json_object.get("symbols", {}).keys()
 
     @property
-    def enumerations(self) -> Iterable[str]:
-        """Returns an iterator of the available enumerations."""
-        return list(self._json_object.get("enums", {}))
+    def enumerations(self) -> Iterable[Any]:
+        """Returns an iterable (KeysView) of the available enumerations."""
+        return self._json_object.get("enums", {}).keys()
 
     @property
     def types(self) -> Iterable[str]:
-        """Returns an iterator of the symbol type names."""
-        return list(self._json_object.get("user_types", {})) + list(self.natives.types)
+        """Returns an iterable (KeysView) of the available symbol type names."""
+        # We use ** instead of
+        # `set(self._json_object.get("user_types", {}).keys()).union(self.natives.types)`
+        # because converting user_types dict to a set is costly.
+        # It is more efficient to convert the (very small) self.natives.types set to a dict.
+        # FIXME: On Python3.8 support drop, merge the two dicts using the merge operator:
+        # (self._json_object.get("user_types", {}) | dict.fromkeys(self.natives.types)).keys()
+        return {
+            **self._json_object.get("user_types", {}),
+            **dict.fromkeys(self.natives.types),
+        }.keys()
 
     def get_type_class(self, name: str) -> Type[interfaces.objects.ObjectInterface]:
         return self._overrides.get(name, objects.AggregateType)

volatility3/framework/symbols/linux/__init__.py

@@ -3,15 +3,26 @@
 #
 import math
 import contextlib
+import functools
+import logging
 from abc import ABC, abstractmethod
 from typing import Iterator, List, Tuple, Optional
 
+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
 
+vollog = logging.getLogger(__name__)
+
 
 class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
     provides = {"type": "interface"}
@@ -43,6 +54,7 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
         self.optional_set_type_class("bpf_prog_aux", extensions.bpf_prog_aux)
         self.optional_set_type_class("kernel_cap_struct", extensions.kernel_cap_struct)
         self.optional_set_type_class("kernel_cap_t", extensions.kernel_cap_t)
+        self.optional_set_type_class("scatterlist", extensions.scatterlist)
 
         # kernels >= 4.18
         self.optional_set_type_class("timespec64", extensions.timespec64)
@@ -348,6 +360,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,
@@ -355,18 +371,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(
@@ -391,41 +400,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):
@@ -630,7 +628,7 @@ class IDStorage(ABC):
         raise NotImplementedError
 
     def nodep_to_node(self, nodep) -> interfaces.objects.ObjectInterface:
-        """Instanciates a tree node from its pointer
+        """Instantiates a tree node from its pointer
 
         Args:
             nodep: Pointer to the XArray/RadixTree node
@@ -677,7 +675,7 @@ class IDStorage(ABC):
         height = self.get_tree_height(root.vol.offset)
 
         nodep = self.get_head_node(root)
-        if not nodep:
+        if not (nodep and nodep.is_readable()):
             return
 
         # Keep the internal flag before untagging it
@@ -712,7 +710,7 @@ class XArray(IDStorage):
 
     def get_node_height(self, nodep) -> int:
         node = self.nodep_to_node(nodep)
-        return (node.shift / self.CHUNK_SHIFT) + 1
+        return (node.shift // self.CHUNK_SHIFT) + 1
 
     def get_head_node(self, tree) -> int:
         return tree.xa_head
@@ -735,6 +733,7 @@ class RadixTree(IDStorage):
     RADIX_TREE_INTERNAL_NODE = 1
     RADIX_TREE_EXCEPTIONAL_ENTRY = 2
     RADIX_TREE_ENTRY_MASK = 3
+    RADIX_TREE_MAP_SHIFT = 6  # CONFIG_BASE_FULL
 
     # Dynamic values. These will be initialized later
     RADIX_TREE_INDEX_BITS = None
@@ -771,43 +770,57 @@ class RadixTree(IDStorage):
     def get_tree_height(self, treep) -> int:
         with contextlib.suppress(exceptions.SymbolError):
             if self.vmlinux.get_type("radix_tree_root").has_member("height"):
-                # kernels < 4.7.10
+                # kernels < 4.7 d0891265bbc988dc91ed8580b38eb3dac128581b
                 radix_tree_root = self.vmlinux.object(
                     "radix_tree_root", offset=treep, absolute=True
                 )
                 return radix_tree_root.height
 
-        # kernels >= 4.7.10
+        # kernels >= 4.7
         return 0
 
+    @functools.cached_property
+    def _max_height_array(self):
+        if self.vmlinux.has_symbol("height_to_maxindex"):
+            # 2.6.24 26fb1589cb0aaec3a0b4418c54f30c1a2b1781f6 <= Kernels < 4.7 d0891265bbc988dc91ed8580b38eb3dac128581b
+            return self.vmlinux.object_from_symbol("height_to_maxindex")
+        elif self.vmlinux.has_symbol("height_to_maxnodes"):
+            # 4.8 c78c66d1ddfdbd2353f3fcfeba0268524537b096 <= kernels < 4.20 8cf2f98411e3a0865026a1061af637161b16d32b
+            return self.vmlinux.object_from_symbol("height_to_maxnodes")
+
+        return None
+
     def _radix_tree_maxindex(self, node, height) -> int:
         """Return the maximum key which can be store into a radix tree with this height."""
 
-        if not self.vmlinux.has_symbol("height_to_maxindex"):
-            # Kernels >= 4.7
-            return (self.CHUNK_SIZE << node.shift) - 1
+        if self._max_height_array:
+            # 2.6.24 <= kernels <= 4.20 See _max_height_array()
+            return self._max_height_array[height]
         else:
-            # Kernels < 4.7
-            height_to_maxindex_array = self.vmlinux.object_from_symbol(
-                "height_to_maxindex"
-            )
-            maxindex = height_to_maxindex_array[height]
-            return maxindex
+            # Kernels >= 4.20
+            return (self.CHUNK_SIZE << node.shift) - 1
 
     def get_node_height(self, nodep) -> int:
         node = self.nodep_to_node(nodep)
         if hasattr(node, "shift"):
             # 4.7 <= Kernels < 4.20
-            return (node.shift / self.CHUNK_SHIFT) + 1
+            height = (node.shift // self.CHUNK_SHIFT) + 1
         elif hasattr(node, "path"):
             # 3.15 <= Kernels < 4.7
-            return node.path & self.RADIX_TREE_HEIGHT_MASK
+            height = node.path & self.RADIX_TREE_HEIGHT_MASK
         elif hasattr(node, "height"):
             # Kernels < 3.15
-            return node.height
+            height = node.height
         else:
             raise exceptions.VolatilityException("Cannot find radix-tree node height")
 
+        if self._max_height_array and not (0 <= height < self._max_height_array.count):
+            error_msg = f"Radix Tree node {node.vol.offset:#x} height {height} exceeds max height of {self._max_height_array.count}"
+            vollog.error(error_msg)
+            raise exceptions.LinuxPageCacheException(error_msg)
+
+        return height
+
     def get_head_node(self, tree) -> int:
         return tree.rnode
 
@@ -820,14 +833,16 @@ class RadixTree(IDStorage):
     def untag_node(self, nodep) -> int:
         return nodep & (~self.RADIX_TREE_ENTRY_MASK)
 
-    def is_valid_node(self, nodep) -> bool:
+    def _is_exceptional_node(self, nodep) -> bool:
         # In kernels 4.20, exceptional nodes were removed and internal entries took their bitmask
-        if self.vmlinux.has_type("radix_tree_root"):
-            return (
-                nodep & self.RADIX_TREE_ENTRY_MASK
-            ) != self.RADIX_TREE_EXCEPTIONAL_ENTRY
+        return (
+            self.vmlinux.has_type("radix_tree_root")
+            and (nodep & self.RADIX_TREE_ENTRY_MASK)
+            == self.RADIX_TREE_EXCEPTIONAL_ENTRY
+        )
 
-        return True
+    def is_valid_node(self, nodep) -> bool:
+        return not self._is_exceptional_node(nodep)
 
 
 class PageCache:
@@ -856,11 +871,17 @@ class PageCache:
         Yields:
             Page objects
         """
-
+        layer = self.vmlinux.context.layers[self.vmlinux.layer_name]
         for page_addr in self._idstorage.get_entries(self._page_cache.i_pages):
-            if not page_addr:
-                continue
+            if not layer.is_valid(page_addr):
+                error_msg = f"Invalid cached page address at {page_addr:#x}, aborting"
+                vollog.error(error_msg)
+                raise exceptions.LinuxPageCacheException(error_msg)
 
             page = self.vmlinux.object("page", offset=page_addr, absolute=True)
-            if page:
-                yield page
+            if not page.is_valid():
+                error_msg = f"Invalid cached page at {page_addr:#x}, aborting"
+                vollog.error(error_msg)
+                raise exceptions.LinuxPageCacheException(error_msg)
+
+            yield page

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

@@ -20,7 +20,6 @@ from volatility3.framework.objects import utility
 from volatility3.framework.symbols import generic, linux, intermed
 from volatility3.framework.symbols.linux.extensions import elf
 
-
 vollog = logging.getLogger(__name__)
 
 # Keep these in a basic module, to prevent import cycles when symbol providers require them
@@ -307,6 +306,46 @@ class module(generic.GenericIntelProcess):
 
 
 class task_struct(generic.GenericIntelProcess):
+    def is_valid(self) -> bool:
+        layer = self._context.layers[self.vol.layer_name]
+        # Make sure the entire task content is readable
+        if not layer.is_valid(self.vol.offset, self.vol.size):
+            return False
+
+        if self.pid < 0 or self.tgid < 0:
+            return False
+
+        if self.has_member("signal") and not (
+            self.signal and self.signal.is_readable()
+        ):
+            return False
+
+        if self.has_member("nsproxy") and not (
+            self.nsproxy and self.nsproxy.is_readable()
+        ):
+            return False
+
+        if self.has_member("real_parent") and not (
+            self.real_parent and self.real_parent.is_readable()
+        ):
+            return False
+
+        if (
+            self.has_member("active_mm")
+            and self.active_mm
+            and not self.active_mm.is_readable()
+        ):
+            return False
+
+        if self.mm:
+            if not self.mm.is_readable():
+                return False
+
+            if self.mm != self.active_mm:
+                return False
+
+        return True
+
     def add_process_layer(
         self, config_prefix: Optional[str] = None, preferred_name: Optional[str] = None
     ) -> Optional[str]:
@@ -401,6 +440,8 @@ class task_struct(generic.GenericIntelProcess):
         tasks_iterable = self._get_tasks_iterable()
         threads_seen = set([self.vol.offset])
         for task in tasks_iterable:
+            if not task.is_valid():
+                continue
             if task.vol.offset not in threads_seen:
                 threads_seen.add(task.vol.offset)
                 yield task
@@ -811,23 +852,30 @@ class mm_struct(objects.StructType):
     def _get_mmap_iter(self) -> Iterable[interfaces.objects.ObjectInterface]:
         """Returns an iterator for the mmap list member of an mm_struct. Use this only if
         required, get_vma_iter() will choose the correct _get_maple_tree_iter() or
-        _get_mmap_iter() automatically as required."""
+        _get_mmap_iter() automatically as required.
+
+        Yields:
+            vm_area_struct objects
+        """
 
         if not self.has_member("mmap"):
             raise AttributeError(
                 "_get_mmap_iter called on mm_struct where no mmap member exists."
             )
-        if not self.mmap:
+        vma_pointer = self.mmap
+        if not (vma_pointer and vma_pointer.is_readable()):
             return None
-        yield self.mmap
+        vma_object = vma_pointer.dereference()
+        yield vma_object
 
-        seen = {self.mmap.vol.offset}
-        link = self.mmap.vm_next
+        seen = {vma_pointer}
+        vma_pointer = vma_pointer.vm_next
 
-        while link != 0 and link.vol.offset not in seen:
-            yield link
-            seen.add(link.vol.offset)
-            link = link.vm_next
+        while vma_pointer and vma_pointer.is_readable() and vma_pointer not in seen:
+            vma_object = vma_pointer.dereference()
+            yield vma_object
+            seen.add(vma_pointer)
+            vma_pointer = vma_pointer.vm_next
 
     # TODO: As of version 3.0.0 this method should be removed
     def get_maple_tree_iter(self) -> Iterable[interfaces.objects.ObjectInterface]:
@@ -842,7 +890,11 @@ class mm_struct(objects.StructType):
     def _get_maple_tree_iter(self) -> Iterable[interfaces.objects.ObjectInterface]:
         """Returns an iterator for the mm_mt member of an mm_struct. Use this only if
         required, get_vma_iter() will choose the correct _get_maple_tree_iter() or
-        get_mmap_iter() automatically as required."""
+        get_mmap_iter() automatically as required.
+
+        Yields:
+            vm_area_struct objects
+        """
 
         if not self.has_member("mm_mt"):
             raise AttributeError(
@@ -850,20 +902,27 @@ class mm_struct(objects.StructType):
             )
         symbol_table_name = self.get_symbol_table_name()
         for vma_pointer in self.mm_mt.get_slot_iter():
-            # convert pointer to vm_area_struct and yield
-            vma = self._context.object(
+            # Convert pointer to vm_area_struct and yield
+            vma_object = self._context.object(
                 symbol_table_name + constants.BANG + "vm_area_struct",
                 layer_name=self.vol.native_layer_name,
                 offset=vma_pointer,
             )
-            yield vma
+            yield vma_object
 
     def get_vma_iter(self) -> Iterable[interfaces.objects.ObjectInterface]:
-        """Returns an iterator for the VMAs in an mm_struct. Automatically choosing the mmap or mm_mt as required."""
+        """Returns an iterator for the VMAs in an mm_struct.
+        Automatically choosing the mmap or mm_mt as required.
+
+        Yields:
+            vm_area_struct objects
+        """
 
         if self.has_member("mmap"):
+            # kernels < 6.1
             yield from self._get_mmap_iter()
         elif self.has_member("mm_mt"):
+            # kernels >= 6.1 d4af56c5c7c6781ca6ca8075e2cf5bc119ed33d1
             yield from self._get_maple_tree_iter()
         else:
             raise AttributeError("Unable to find mmap or mm_mt in mm_struct")
@@ -1205,35 +1264,43 @@ class list_head(objects.StructType, collections.abc.Iterable):
             Objects of the type specified via the "symbol_type" argument.
 
         """
-        layer = layer or self.vol.layer_name
+        layer_name = layer or self.vol.layer_name
+
+        trans_layer = self._context.layers[layer_name]
+        if not trans_layer.is_valid(self.vol.offset):
+            return None
 
         relative_offset = self._context.symbol_space.get_type(
             symbol_type
         ).relative_child_offset(member)
 
-        direction = "prev"
-        if forward:
-            direction = "next"
-        try:
-            link = getattr(self, direction).dereference()
-        except exceptions.InvalidAddressException:
+        direction = "next" if forward else "prev"
+
+        link_ptr = getattr(self, direction)
+        if not (link_ptr and link_ptr.is_readable()):
             return None
+        link = link_ptr.dereference()
+
         if not sentinel:
-            yield self._context.object(
-                symbol_type, layer, offset=self.vol.offset - relative_offset
-            )
+            obj_offset = self.vol.offset - relative_offset
+            if not trans_layer.is_valid(obj_offset):
+                return None
+
+            yield self._context.object(symbol_type, layer_name, offset=obj_offset)
+
         seen = {self.vol.offset}
         while link.vol.offset not in seen:
-            obj = self._context.object(
-                symbol_type, layer, offset=link.vol.offset - relative_offset
-            )
-            yield obj
+            obj_offset = link.vol.offset - relative_offset
+            if not trans_layer.is_valid(obj_offset):
+                return None
+
+            yield self._context.object(symbol_type, layer_name, offset=obj_offset)
 
             seen.add(link.vol.offset)
-            try:
-                link = getattr(link, direction).dereference()
-            except exceptions.InvalidAddressException:
+            link_ptr = getattr(link, direction)
+            if not (link_ptr and link_ptr.is_readable()):
                 break
+            link = link_ptr.dereference()
 
     def __iter__(self) -> Iterator[interfaces.objects.ObjectInterface]:
         return self.to_list(self.vol.parent.vol.type_name, self.vol.member_name)
@@ -2489,7 +2556,12 @@ class inode(objects.StructType):
         """
         if not self.i_size:
             return
-        elif not (self.i_mapping and self.i_mapping.nrpages > 0):
+
+        if not (
+            self.i_mapping
+            and self.i_mapping.is_readable()
+            and self.i_mapping.nrpages > 0
+        ):
             return
 
         page_cache = linux.PageCache(
@@ -2497,19 +2569,26 @@ class inode(objects.StructType):
             kernel_module_name="kernel",
             page_cache=self.i_mapping.dereference(),
         )
+
         yield from page_cache.get_cached_pages()
 
-    def get_contents(self):
+    def get_contents(self) -> Iterable[Tuple[int, bytes]]:
         """Get the inode cached pages from the page cache
 
         Yields:
             page_index (int): The page index in the Tree. File offset is page_index * PAGE_SIZE.
-            page_content (str): The page content
+            page_content (bytes): The page content
         """
         for page_obj in self.get_pages():
+            if page_obj.mapping != self.i_mapping:
+                vollog.warning(
+                    f"Cached page at {page_obj.vol.offset:#x} has a mismatched address space with the inode. Skipping page"
+                )
+                continue
             page_index = int(page_obj.index)
             page_content = page_obj.get_content()
-            yield page_index, page_content
+            if page_content:
+                yield page_index, page_content
 
 
 class address_space(objects.StructType):
@@ -2517,7 +2596,7 @@ class address_space(objects.StructType):
     def i_pages(self):
         """Returns the appropriate member containing the page cache tree"""
         if self.has_member("i_pages"):
-            # Kernel >= 4.17
+            # Kernel >= 4.17 b93b016313b3ba8003c3b8bb71f569af91f19fc7
             return self.member("i_pages")
         elif self.has_member("page_tree"):
             # Kernel < 4.17
@@ -2527,6 +2606,15 @@ class address_space(objects.StructType):
 
 
 class page(objects.StructType):
+    def is_valid(self) -> bool:
+        if self.mapping and not self.mapping.is_readable():
+            return False
+
+        if self.to_paddr() < 0:
+            return False
+
+        return True
+
     @functools.cached_property
     def pageflags_enum(self) -> Dict:
         """Returns 'pageflags' enumeration key/values
@@ -2625,7 +2713,7 @@ class page(objects.StructType):
 
         return page_paddr
 
-    def get_content(self) -> Union[str, None]:
+    def get_content(self) -> Union[bytes, None]:
         """Returns the page content
 
         Returns:
@@ -2641,8 +2729,13 @@ class page(objects.StructType):
         if not page_paddr:
             return None
 
-        page_data = physical_layer.read(page_paddr, vmlinux_layer.page_size)
-        return page_data
+        if not physical_layer.is_valid(page_paddr, length=vmlinux_layer.page_size):
+            vollog.debug(
+                "Unable to read page 0x%x content at 0x%x", self.vol.offset, page_paddr
+            )
+            return None
+
+        return physical_layer.read(page_paddr, vmlinux_layer.page_size)
 
     def get_flags_list(self) -> List[str]:
         """Returns a list of page flags
@@ -2755,17 +2848,17 @@ class IDR(objects.StructType):
 
 
 class rb_root(objects.StructType):
-    def _walk_nodes(self, root_node) -> Iterator[int]:
+    def _walk_nodes(self, root_node: int) -> 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
+            root_node: A Red-Black tree node pointer from which to start descending
 
         Yields:
             A pointer to every node descending from the specified root node
         """
-        if not root_node:
+        if not (root_node and root_node.is_readable()):
             return
 
         yield root_node
@@ -2780,3 +2873,111 @@ class rb_root(objects.StructType):
         """
 
         yield from self._walk_nodes(root_node=self.rb_node)
+
+
+class scatterlist(objects.StructType):
+    SG_CHAIN = 0x01
+    SG_END = 0x02
+    SG_PAGE_LINK_MASK = SG_CHAIN | SG_END
+
+    def _sg_flags(self) -> int:
+        return self.page_link & self.SG_PAGE_LINK_MASK
+
+    def _sg_is_chain(self) -> int:
+        return self._sg_flags() & self.SG_CHAIN
+
+    def _sg_is_last(self) -> int:
+        return self._sg_flags() & self.SG_END
+
+    def _sg_chain_ptr(self) -> int:
+        """Clears the last two bits basically."""
+        return self.page_link & ~self.SG_PAGE_LINK_MASK
+
+    def _sg_dma_len(self) -> int:
+        # Depends on CONFIG_NEED_SG_DMA_LENGTH
+        if self.has_member("dma_length"):
+            return self.dma_length
+        return self.length
+
+    def _get_sg_max_single_alloc(self) -> int:
+        """Based on kernel's SG_MAX_SINGLE_ALLOC.
+
+        Doc. from kernel source :
+            * Maximum number of entries that will be allocated in one piece, if
+            * a list larger than this is required then chaining will be utilized.
+        """
+        return self._context.layers[self.vol.layer_name].page_size // self.vol.size
+
+    def _sg_next(self) -> Optional[interfaces.objects.ObjectInterface]:
+        """Get the next scatterlist struct from the list.
+        Based on kernel's sg_next.
+
+        Doc. from kernel source :
+            * Notes on SG table design.
+            *
+            * We use the unsigned long page_link field in the scatterlist struct to place
+            * the page pointer AND encode information about the sg table as well. The two
+            * lower bits are reserved for this information.
+            *
+            * If bit 0 is set, then the page_link contains a pointer to the next sg
+            * table list. Otherwise the next entry is at sg + 1.
+            *
+            * If bit 1 is set, then this sg entry is the last element in a list.
+        """
+        if self._sg_is_last():
+            return None
+
+        if self._sg_is_chain():
+            next_address = self._sg_chain_ptr()
+        else:
+            next_address = self.vol.offset + self.vol.size
+
+        sg = self._context.object(
+            self.get_symbol_table_name() + constants.BANG + "scatterlist",
+            self.vol.layer_name,
+            next_address,
+        )
+        return sg
+
+    def for_each_sg(self) -> Optional[Iterator[interfaces.objects.ObjectInterface]]:
+        """Iterate over each struct in the scatterlist."""
+        sg = self
+        sg_max_single_alloc = self._get_sg_max_single_alloc()
+
+        # Empty scatterlists protection
+        if sg.page_link == 0 and sg._sg_dma_len() == 0 and sg.dma_address == 0:
+            return None
+        else:
+            # Yield itself first
+            yield sg
+
+        entries_count = 1
+        # entries_count <= sg_max_single_alloc should always be true if the
+        # scatterlists were correctly chained.
+        while entries_count <= sg_max_single_alloc:
+            sg = sg._sg_next()
+            if sg is None:
+                break
+            # Points to a new scatterlist
+            elif sg._sg_is_chain():
+                entries_count = 0
+            else:
+                entries_count += 1
+                yield sg
+
+    def get_content(
+        self,
+    ) -> Optional[Iterator[bytes]]:
+        """Traverse a scatterlist to gather content located at each
+        dma_address position.
+
+        Returns:
+            An iterator of bytes
+        """
+        # Either "physical" is layer-1 because this is a module layer, or "physical" is the current layer
+        physical_layer_name = self._context.layers[self.vol.layer_name].config.get(
+            "memory_layer", self.vol.layer_name
+        )
+        physical_layer = self._context.layers[physical_layer_name]
+        for sg in self.for_each_sg():
+            yield from physical_layer.read(sg.dma_address, sg._sg_dma_len())

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

@@ -0,0 +1,70 @@
+from typing import Iterator, List, Tuple
+
+from volatility3 import framework
+from volatility3.framework import constants, interfaces
+from volatility3.framework.objects import utility
+
+
+class Modules(interfaces.configuration.VersionableInterface):
+    """Kernel modules related utilities."""
+
+    _version = (1, 0, 0)
+    _required_framework_version = (2, 0, 0)
+
+    framework.require_interface_version(*_required_framework_version)
+
+    @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/linux/utilities/tainting.py

@@ -0,0 +1,161 @@
+import functools
+
+from volatility3 import framework
+from volatility3.framework import interfaces
+from volatility3.framework.constants import linux as linux_constants
+from typing import List, Optional
+
+
+class Tainting(interfaces.configuration.VersionableInterface):
+    """Tainted kernel and modules parsing capabilities.
+
+    Relevant Linux kernel functions:
+        - modules: module_flags_taint
+        - kernel: print_tainted
+    """
+
+    _version = (1, 0, 0)
+    _required_framework_version = (2, 0, 0)
+
+    framework.require_interface_version(*_required_framework_version)
+
+    @classmethod
+    @functools.lru_cache
+    def _get_kernel_taint_flags_list(
+        cls,
+        context: interfaces.context.ContextInterface,
+        kernel_module_name: str,
+    ) -> Optional[List[interfaces.objects.ObjectInterface]]:
+        """Determine whether the kernel embeds taint flags definition
+        in-memory or not.
+
+        Returns:
+            A list of "taint_flag" kernel objects if taint_flags symbol exists
+        """
+        kernel = context.modules[kernel_module_name]
+        if kernel.has_symbol("taint_flags"):
+            return list(kernel.object_from_symbol("taint_flags"))
+        return None
+
+    @classmethod
+    def _module_flags_taint_pre_4_10_rc1(
+        cls,
+        taints: int,
+        is_module: bool = False,
+    ) -> str:
+        """Convert the module's taints value to a 1-1 character mapping.
+        Relies on statically defined taints mappings in the framework.
+
+        Args:
+            taints: The taints value, represented by an integer
+            is_module: Indicates if the taints value is associated with a built-in/LKM module
+
+        Returns:
+            The raw taints string.
+        """
+        taints_string = ""
+        for char, taint_flag in linux_constants.TAINT_FLAGS.items():
+            if is_module and not taint_flag.module:
+                continue
+
+            if taints & taint_flag.shift:
+                taints_string += char
+
+        return taints_string
+
+    @classmethod
+    def _module_flags_taint_post_4_10_rc1(
+        cls,
+        context: interfaces.context.ContextInterface,
+        kernel_module_name: str,
+        taints: int,
+        is_module: bool = False,
+    ) -> str:
+        """Convert the module's taints value to a 1-1 character mapping.
+        Relies on kernel symbol embedded taints definitions.
+
+            struct taint_flag {
+                char c_true;		/* character printed when tainted */
+                char c_false;		/* character printed when not tainted */
+                bool module;		/* also show as a per-module taint flag */
+            };
+
+        Args:
+            taints: The taints value, represented by an integer
+            is_module: Indicates if the taints value is associated with a built-in/LKM module
+
+        Returns:
+            The raw taints string.
+        """
+        taints_string = ""
+        for taint_bit, taint_flag in enumerate(
+            cls._get_kernel_taint_flags_list(context, kernel_module_name)
+        ):
+            if is_module and not taint_flag.module:
+                continue
+            c_true = chr(taint_flag.c_true)
+            c_false = chr(taint_flag.c_false)
+            if taints & (1 << taint_bit):
+                taints_string += c_true
+            elif c_false != " ":
+                taints_string += c_false
+
+        return taints_string
+
+    @classmethod
+    def get_taints_as_plain_string(
+        cls,
+        context: interfaces.context.ContextInterface,
+        kernel_module_name: str,
+        taints: int,
+        is_module: bool = False,
+    ) -> str:
+        """Convert the taints value to a 1-1 character mapping.
+
+        Args:
+            taints: The taints value, represented by an integer
+            is_module: Indicates if the taints value is associated with a built-in/LKM module
+        Returns:
+            The raw taints string.
+
+        Documentation:
+            - module_flags_taint kernel function
+        """
+
+        if cls._get_kernel_taint_flags_list(context, kernel_module_name):
+            return cls._module_flags_taint_post_4_10_rc1(
+                context, kernel_module_name, taints, is_module
+            )
+        return cls._module_flags_taint_pre_4_10_rc1(taints, is_module)
+
+    @classmethod
+    def get_taints_parsed(
+        cls,
+        context: interfaces.context.ContextInterface,
+        kernel_module_name: str,
+        taints: int,
+        is_module: bool = False,
+    ) -> List[str]:
+        """Convert the taints string to a 1-1 descriptor mapping.
+
+        Args:
+            taints: The taints value, represented by an integer
+            is_module: Indicates if the taints value is associated with a built-in/LKM module
+
+        Returns:
+            A comprehensive (user-friendly) taint descriptor list.
+
+        Documentation:
+            - module_flags_taint kernel function
+        """
+        comprehensive_taints = []
+        for character in cls.get_taints_as_plain_string(
+            context, kernel_module_name, taints, is_module
+        ):
+            taint_flag = linux_constants.TAINT_FLAGS.get(character)
+            if not taint_flag:
+                comprehensive_taints.append(f"<UNKNOWN_TAINT_CHAR_{character}>")
+            elif taint_flag.when_present:
+                comprehensive_taints.append(taint_flag.desc)
+
+        return comprehensive_taints

volatility3/framework/symbols/native.py

@@ -30,7 +30,7 @@ class NativeTable(interfaces.symbols.NativeTableInterface):
 
     @property
     def types(self) -> Iterable[str]:
-        """Returns an iterator of the symbol type names."""
+        """Returns an iterable (set) of the available symbol type names."""
         return self._types
 
     def get_type(self, type_name: str) -> interfaces.objects.Template:

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

@@ -962,56 +962,55 @@ class LIST_ENTRY(objects.StructType, collections.abc.Iterable):
     ) -> Iterator[interfaces.objects.ObjectInterface]:
         """Returns an iterator of the entries in the list."""
 
-        layer = layer or self.vol.layer_name
+        layer_name = layer or self.vol.layer_name
+        native_layer_name = layer_name or self.vol.native_layer_name
+
+        trans_layer = self._context.layers[layer_name]
+        if not trans_layer.is_valid(self.vol.offset):
+            return None
 
         relative_offset = self._context.symbol_space.get_type(
             symbol_type
         ).relative_child_offset(member)
 
-        direction = "Blink"
-        if forward:
-            direction = "Flink"
+        direction = "Flink" if forward else "Blink"
 
-        trans_layer = self._context.layers[layer]
-
-        try:
-            is_valid = trans_layer.is_valid(self.vol.offset)
-            if not is_valid:
-                return None
-
-            link = getattr(self, direction).dereference()
-        except exceptions.InvalidAddressException:
+        link_ptr = getattr(self, direction)
+        if not (link_ptr and link_ptr.is_readable()):
             return None
+        link = link_ptr.dereference()
 
         if not sentinel:
+            obj_offset = self.vol.offset - relative_offset
+            if not trans_layer.is_valid(obj_offset):
+                return None
+
             yield self._context.object(
                 symbol_type,
-                layer,
-                offset=self.vol.offset - relative_offset,
-                native_layer_name=layer or self.vol.native_layer_name,
+                layer_name,
+                offset=obj_offset,
+                native_layer_name=native_layer_name,
             )
 
         seen = {self.vol.offset}
         while link.vol.offset not in seen:
             obj_offset = link.vol.offset - relative_offset
-
             if not trans_layer.is_valid(obj_offset):
                 return None
 
-            obj = self._context.object(
+            yield self._context.object(
                 symbol_type,
-                layer,
+                layer_name,
                 offset=obj_offset,
-                native_layer_name=layer or self.vol.native_layer_name,
+                native_layer_name=native_layer_name,
             )
-            yield obj
 
             seen.add(link.vol.offset)
 
-            try:
-                link = getattr(link, direction).dereference()
-            except exceptions.InvalidAddressException:
+            link_ptr = getattr(link, direction)
+            if not (link_ptr and link_ptr.is_readable()):
                 return None
+            link = link_ptr.dereference()
 
     def __iter__(self) -> Iterator[interfaces.objects.ObjectInterface]:
         return self.to_list(self.vol.parent.vol.type_name, self.vol.member_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
 

volatility3/schemas/__init__.py

@@ -14,6 +14,8 @@ vollog = logging.getLogger(__name__)
 
 cached_validation_filepath = os.path.join(constants.CACHE_PATH, "valid_isf.hashcache")
 
+validators = {}
+
 
 def load_cached_validations() -> Set[str]:
     """Loads up the list of successfully cached json objects, so we don't need
@@ -93,6 +95,13 @@ def valid(
         return True
     try:
         import jsonschema
+
+        schema_key = json.dumps(schema, sort_keys=True)
+        if schema_key not in validators:
+            validator_class = jsonschema.validators.validator_for(schema)
+            validator_class.check_schema(schema)
+            validator = validator_class(schema)
+            validators[schema_key] = validator
     except ImportError:
         vollog.info("Dependency for validation unavailable: jsonschema")
         vollog.debug("All validations will report success, even with malformed input")
@@ -100,7 +109,7 @@ def valid(
 
     try:
         vollog.debug("Validating JSON against schema...")
-        jsonschema.validate(input, schema)
+        validators[schema_key].validate(input)
         cached_validations.add(input_hash)
         vollog.debug("JSON validated against schema (result cached)")
     except jsonschema.exceptions.SchemaError: