Merge branch 'develop' into 816-port-cmdscan-and-console-plugins-from-vol2-to-vol3-please

.github/workflows/build-pypi.yml

@@ -18,7 +18,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        python-version: ["3.7"]
+        python-version: ["3.8"]
     steps:
     - uses: actions/checkout@v4
     - name: Set up Python ${{ matrix.python-version }}

.github/workflows/install.yml

@@ -8,7 +8,7 @@ jobs:
       fail-fast: false
       matrix:
         host: [ ubuntu-latest, windows-latest ]
-        python-version: [ "3.7", "3.8", "3.9", "3.10", "3.11" ]
+        python-version: [ "3.8", "3.9", "3.10", "3.11" ]
     steps:
     - uses: actions/checkout@v4
 

.github/workflows/test.yaml

@@ -6,7 +6,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        python-version: ["3.7"]
+        python-version: ["3.8"]
     steps:
     - uses: actions/checkout@v4
     - name: Set up Python ${{ matrix.python-version }}
@@ -46,7 +46,7 @@ jobs:
 
     - name: Clean up post-test
       run: |
-        rm -rf *.lime
+        rm -rf *.bin
         rm -rf *.img
         cd volatility3/symbols
         rm -rf linux

README.md

@@ -20,7 +20,7 @@ more details.
 
 ## Requirements
 
-Volatility 3 requires Python 3.7.3 or later. To install the most minimal set of dependencies (some plugins will not work) use a command such as:
+Volatility 3 requires Python 3.8.0 or later. To install the most minimal set of dependencies (some plugins will not work) use a command such as:
 
 ```shell
 pip3 install -r requirements-minimal.txt

doc/requirements.txt

@@ -4,5 +4,6 @@ sphinx_autodoc_typehints>=1.4.0
 sphinx-rtd-theme>=0.4.3
 
 yara-python
+yara-x
 pycryptodome
 pefile

pyproject.toml

@@ -6,7 +6,7 @@ readme = "README.md"
 authors = [
     { name = "Volatility Foundation", email = "volatility@volatilityfoundation.org" },
 ]
-requires-python = ">=3.7.3"
+requires-python = ">=3.8.0"
 license = { text = "VSL" }
 dynamic = ["dependencies", "optional-dependencies", "version"]
 

test/requirements-testing.txt

@@ -6,5 +6,6 @@ pefile>=2017.8.1 #foo
 
 # This is required for the yara plugins
 yara-python>=3.8.0
+yara-x>=0.5.0
 
 pytest>=7.0.0

volatility3/cli/text_renderer.py

@@ -179,7 +179,14 @@ class QuickTextRenderer(CLIRenderer):
         outfd.write("\n{}\n".format("\t".join(line)))
 
         def visitor(node: interfaces.renderers.TreeNode, accumulator):
-            if self.filter and self.filter.filter(node.values):
+            line = []
+            for column_index, column in enumerate(grid.columns):
+                renderer = self._type_renderers.get(
+                    column.type, self._type_renderers["default"]
+                )
+                line.append(renderer(node.values[column_index]))
+
+            if self.filter and self.filter.filter(line):
                 return accumulator
 
             accumulator.write("\n")
@@ -188,13 +195,6 @@ class QuickTextRenderer(CLIRenderer):
                 "*" * max(0, node.path_depth - 1)
                 + ("" if (node.path_depth <= 1) else " ")
             )
-            line = []
-            for column_index in range(len(grid.columns)):
-                column = grid.columns[column_index]
-                renderer = self._type_renderers.get(
-                    column.type, self._type_renderers["default"]
-                )
-                line.append(renderer(node.values[column_index]))
             accumulator.write("{}".format("\t".join(line)))
             accumulator.flush()
             return accumulator
@@ -259,12 +259,17 @@ class CSVRenderer(CLIRenderer):
         def visitor(node: interfaces.renderers.TreeNode, accumulator):
             # Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case
             row = {"TreeDepth": str(max(0, node.path_depth - 1))}
-            for column_index in range(len(grid.columns)):
-                column = grid.columns[column_index]
+            line = []
+            for column_index, column in enumerate(grid.columns):
                 renderer = self._type_renderers.get(
                     column.type, self._type_renderers["default"]
                 )
                 row[f"{column.name}"] = renderer(node.values[column_index])
+                line.append(row[f"{column.name}"])
+
+            if self.filter and self.filter.filter(line):
+                return accumulator
+
             accumulator.writerow(row)
             return accumulator
 
@@ -317,12 +322,9 @@ class PrettyTextRenderer(CLIRenderer):
                 max_column_widths.get(tree_indent_column, 0), node.path_depth
             )
 
-            if self.filter and self.filter.filter(node.values):
-                return accumulator
-
             line = {}
-            for column_index in range(len(grid.columns)):
-                column = grid.columns[column_index]
+            rendered_line = []
+            for column_index, column in enumerate(grid.columns):
                 renderer = self._type_renderers.get(
                     column.type, self._type_renderers["default"]
                 )
@@ -334,6 +336,11 @@ class PrettyTextRenderer(CLIRenderer):
                     max_column_widths.get(column.name, len(column.name)), field_width
                 )
                 line[column] = data.split("\n")
+                rendered_line.append(data)
+
+            if self.filter and self.filter.filter(rendered_line):
+                return accumulator
+
             accumulator.append((node.path_depth, line))
             return accumulator
 
@@ -347,8 +354,7 @@ class PrettyTextRenderer(CLIRenderer):
         format_string_list = [
             "{0:<" + str(max_column_widths.get(tree_indent_column, 0)) + "s}"
         ]
-        for column_index in range(len(grid.columns)):
-            column = grid.columns[column_index]
+        for column_index, column in enumerate(grid.columns):
             format_string_list.append(
                 "{"
                 + str(column_index + 1)
@@ -437,8 +443,8 @@ class JsonRenderer(CLIRenderer):
             # Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case
             acc_map, final_tree = accumulator
             node_dict: Dict[str, Any] = {"__children": []}
-            for column_index in range(len(grid.columns)):
-                column = grid.columns[column_index]
+            line = []
+            for column_index, column in enumerate(grid.columns):
                 renderer = self._type_renderers.get(
                     column.type, self._type_renderers["default"]
                 )
@@ -446,6 +452,11 @@ class JsonRenderer(CLIRenderer):
                 if isinstance(data, interfaces.renderers.BaseAbsentValue):
                     data = None
                 node_dict[column.name] = data
+                line.append(data)
+
+            if self.filter and self.filter.filter(line):
+                return accumulator
+
             if node.parent:
                 acc_map[node.parent.path]["__children"].append(node_dict)
             else:

volatility3/framework/__init__.py

@@ -7,7 +7,7 @@ import glob
 import sys
 import zipfile
 
-required_python_version = (3, 7, 3)
+required_python_version = (3, 8, 0)
 if (
     sys.version_info.major != required_python_version[0]
     or sys.version_info.minor < required_python_version[1]

volatility3/framework/configuration/requirements.py

@@ -527,12 +527,14 @@ class VersionRequirement(interfaces.configuration.RequirementInterface):
     def __init__(
         self,
         name: str,
-        description: str = None,
+        description: Optional[str] = None,
         default: bool = False,
         optional: bool = False,
         component: Type[interfaces.configuration.VersionableInterface] = None,
         version: Optional[Tuple[int, ...]] = None,
     ) -> None:
+        if description is None:
+            description = f"Version {'.'.join([str(x) for x in version])} dependency on {component.__module__}.{component.__name__} unmet"
         super().__init__(
             name=name, description=description, default=default, optional=optional
         )
@@ -544,15 +546,51 @@ class VersionRequirement(interfaces.configuration.RequirementInterface):
         self._version = version
 
     def unsatisfied(
-        self, context: interfaces.context.ContextInterface, config_path: str
+        self,
+        context: interfaces.context.ContextInterface,
+        config_path: str,
+        accumulator: Optional[
+            List[interfaces.configuration.VersionableInterface]
+        ] = None,
     ) -> Dict[str, interfaces.configuration.RequirementInterface]:
         # Mypy doesn't appreciate our classproperty implementation, self._plugin.version has no type
         config_path = interfaces.configuration.path_join(config_path, self.name)
         if not self.matches_required(self._version, self._component.version):
             return {config_path: self}
+
+        recurse = True
+        if accumulator is None:
+            accumulator = set([self._component])
+        else:
+            if self._component in accumulator:
+                recurse = False
+            else:
+                accumulator.add(self._component)
+
+        # Check for child requirements
+        if (
+            issubclass(self._component, interfaces.configuration.ConfigurableInterface)
+            and recurse
+        ):
+            result = {}
+            for requirement in self._component.get_requirements():
+                if not requirement.optional and isinstance(
+                    requirement, VersionRequirement
+                ):
+                    result.update(
+                        requirement.unsatisfied(
+                            context, config_path, accumulator.copy()
+                        )
+                    )
+
+            if result:
+                result.update({config_path: self})
+                return result
+
         context.config[interfaces.configuration.path_join(config_path, self.name)] = (
             True
         )
+
         return {}
 
     @classmethod

volatility3/framework/constants/__init__.py

@@ -134,4 +134,5 @@ def __getattr__(name):
     ]:
         warnings.warn(f"{name} is deprecated", FutureWarning)
         return globals()[f"{deprecated_tag}{name}"]
-    return None
+
+    return getattr(__import__(__name__), name)

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 = 8  # Number of changes that only add to the interface
+VERSION_MINOR = 10  # 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/architectures.py

@@ -0,0 +1,21 @@
+from volatility3.framework.layers import intel
+
+WIN_ARCHS = ["Intel32", "Intel64"]
+"""Windows supported architectures"""
+WIN_ARCHS_LAYERS = [intel.Intel]
+"""Windows supported architectures layers"""
+
+LINUX_ARCHS = ["Intel32", "Intel64"]
+"""Linux supported architectures"""
+LINUX_ARCHS_LAYERS = [intel.Intel]
+"""Linux supported architectures layers"""
+
+MAC_ARCHS = ["Intel32", "Intel64"]
+"""Mac supported architectures"""
+MAC_ARCHS_LAYERS = [intel.Intel]
+"""Mac supported architectures layers"""
+
+FRAMEWORK_ARCHS = ["Intel32", "Intel64"]
+"""Framework supported architectures"""
+FRAMEWORK_ARCHS_LAYERS = [intel.Intel]
+"""Framework supported architectures layers"""

volatility3/framework/interfaces/configuration.py

@@ -494,8 +494,7 @@ class SimpleTypeRequirement(RequirementInterface):
         """Validates the instance requirement based upon its
         `instance_type`."""
         config_path = path_join(config_path, self.name)
-
-        value = self.config_value(context, config_path, None)
+        value = self.config_value(context, config_path, self.default)
         if not isinstance(value, self.instance_type):
             vollog.log(
                 constants.LOGLEVEL_V,
@@ -536,7 +535,7 @@ class ClassRequirement(RequirementInterface):
         """Checks to see if a class can be recovered."""
         config_path = path_join(config_path, self.name)
 
-        value = self.config_value(context, config_path, None)
+        value = self.config_value(context, config_path, self.default)
         self._cls = None
         if value is not None and isinstance(value, str):
             if "." in value:

volatility3/framework/plugins/linux/check_creds.py

@@ -2,20 +2,19 @@
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
 
-import logging
-
 from volatility3.framework import interfaces, renderers
+from volatility3.framework.renderers import format_hints
 from volatility3.framework.configuration import requirements
 from volatility3.plugins.linux import pslist
 
-vollog = logging.getLogger(__name__)
-
 
 class Check_creds(interfaces.plugins.PluginInterface):
     """Checks if any processes are sharing credential structures"""
 
     _required_framework_version = (2, 0, 0)
 
+    _version = (2, 0, 0)
+
     @classmethod
     def get_requirements(cls):
         return [
@@ -46,20 +45,28 @@ class Check_creds(interfaces.plugins.PluginInterface):
         tasks = pslist.PsList.list_tasks(self.context, vmlinux.name)
 
         for task in tasks:
-            cred_addr = task.cred.dereference().vol.offset
+            task_cred_ptr = task.cred
+            if not (task_cred_ptr and task_cred_ptr.is_readable()):
+                continue
 
-            if cred_addr not in creds:
-                creds[cred_addr] = []
+            cred_addr = task_cred_ptr.dereference().vol.offset
 
+            creds.setdefault(cred_addr, [])
             creds[cred_addr].append(task.pid)
 
-        for _, pids in creds.items():
+        for cred_addr, pids in creds.items():
             if len(pids) > 1:
-                pid_str = ""
-                for pid in pids:
-                    pid_str = pid_str + f"{pid:d}, "
-                pid_str = pid_str[:-2]
-                yield (0, [str(pid_str)])
+                pid_str = ", ".join([str(pid) for pid in pids])
+
+                fields = [
+                    format_hints.Hex(cred_addr),
+                    pid_str,
+                ]
+                yield (0, fields)
 
     def run(self):
-        return renderers.TreeGrid([("PIDs", str)], self._generator())
+        headers = [
+            ("CredVAddr", format_hints.Hex),
+            ("PIDs", str),
+        ]
+        return renderers.TreeGrid(headers, self._generator())

volatility3/framework/plugins/linux/ebpf.py

@@ -0,0 +1,73 @@
+# 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
+
+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
+
+vollog = logging.getLogger(__name__)
+
+
+class EBPF(plugins.PluginInterface):
+    """Enumerate eBPF programs"""
+
+    _required_framework_version = (2, 0, 0)
+
+    _version = (1, 0, 0)
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+        ]
+
+    def get_ebpf_programs(
+        self,
+        context: interfaces.context.ContextInterface,
+        vmlinux_module_name: str,
+    ) -> interfaces.objects.ObjectInterface:
+        """Enumerate eBPF programs walking its IDR.
+
+        Args:
+            context: The context to retrieve required elements (layers, symbol tables) from
+            vmlinux_module_name: The name of the kernel module on which to operate
+        Yields:
+            eBPF program objects
+        """
+        vmlinux = context.modules[vmlinux_module_name]
+
+        if not vmlinux.has_symbol("prog_idr"):
+            raise exceptions.VolatilityException(
+                "Cannot find the eBPF prog idr. Unsupported kernel"
+            )
+
+        prog_idr = vmlinux.object_from_symbol("prog_idr")
+        for page_addr in prog_idr.get_entries():
+            bpf_prog = vmlinux.object("bpf_prog", offset=page_addr, absolute=True)
+            yield bpf_prog
+
+    def _generator(self):
+        for prog in self.get_ebpf_programs(self.context, self.config["kernel"]):
+            prog_addr = prog.vol.offset
+            prog_type = prog.get_type() or renderers.NotAvailableValue()
+            prog_tag = prog.get_tag() or renderers.NotAvailableValue()
+            prog_name = prog.get_name() or renderers.NotAvailableValue()
+            fields = (format_hints.Hex(prog_addr), prog_name, prog_tag, prog_type)
+            yield (0, fields)
+
+    def run(self):
+        headers = [
+            ("Address", format_hints.Hex),
+            ("Name", str),
+            ("Tag", str),
+            ("Type", str),
+        ]
+        return renderers.TreeGrid(headers, self._generator())

volatility3/framework/plugins/linux/lsof.py

@@ -1,27 +1,27 @@
-# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
+# 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
 #
 """A module containing a collection of plugins that produce data typically
 found in Linux's /proc file system."""
-import logging
+import logging, datetime
 from typing import List, Callable
 
-from volatility3.framework import renderers, interfaces, constants
+from volatility3.framework import renderers, interfaces, constants, exceptions
 from volatility3.framework.configuration import requirements
 from volatility3.framework.interfaces import plugins
 from volatility3.framework.objects import utility
 from volatility3.framework.symbols import linux
 from volatility3.plugins.linux import pslist
+from volatility3.plugins import timeliner
 
 vollog = logging.getLogger(__name__)
 
 
-class Lsof(plugins.PluginInterface):
-    """Lists all memory maps for all processes."""
+class Lsof(plugins.PluginInterface, timeliner.TimeLinerInterface):
+    """Lists open files for each processes."""
 
     _required_framework_version = (2, 0, 0)
-
-    _version = (1, 1, 0)
+    _version = (1, 2, 0)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -45,6 +45,29 @@ class Lsof(plugins.PluginInterface):
             ),
         ]
 
+    @classmethod
+    def get_inode_metadata(cls, filp: interfaces.objects.ObjectInterface):
+        try:
+            dentry = filp.get_dentry()
+            if dentry:
+                inode_object = dentry.d_inode
+                if inode_object and inode_object.is_valid():
+                    itype = (
+                        inode_object.get_inode_type() or renderers.NotAvailableValue()
+                    )
+                    return (
+                        inode_object.i_ino,
+                        itype,
+                        inode_object.i_size,
+                        inode_object.get_file_mode(),
+                        inode_object.get_change_time(),
+                        inode_object.get_modification_time(),
+                        inode_object.get_access_time(),
+                    )
+        except (exceptions.InvalidAddressException, AttributeError) as e:
+            vollog.warning(f"Can't get inode metadata: {e}")
+        return None
+
     @classmethod
     def list_fds(
         cls,
@@ -52,7 +75,7 @@ class Lsof(plugins.PluginInterface):
         symbol_table: str,
         filter_func: Callable[[int], bool] = lambda _: False,
     ):
-        linuxutils_symbol_table = None  # type: ignore
+        linuxutils_symbol_table = None
         for task in pslist.PsList.list_tasks(context, symbol_table, filter_func):
             if linuxutils_symbol_table is None:
                 if constants.BANG not in task.vol.type_name:
@@ -69,21 +92,79 @@ class Lsof(plugins.PluginInterface):
             for fd_fields in fd_generator:
                 yield pid, task_comm, task, fd_fields
 
+    @classmethod
+    def list_fds_and_inodes(
+        cls,
+        context: interfaces.context.ContextInterface,
+        symbol_table: str,
+        filter_func: Callable[[int], bool] = lambda _: False,
+    ):
+        for pid, task_comm, task, (fd_num, filp, full_path) in cls.list_fds(
+            context, symbol_table, filter_func
+        ):
+            inode_metadata = cls.get_inode_metadata(filp)
+            if inode_metadata is None:
+                inode_metadata = tuple(
+                    interfaces.renderers.BaseAbsentValue() for _ in range(7)
+                )
+            yield pid, task_comm, task, fd_num, filp, full_path, inode_metadata
+
     def _generator(self, pids, symbol_table):
         filter_func = pslist.PsList.create_pid_filter(pids)
-        fds_generator = self.list_fds(
+        fds_generator = self.list_fds_and_inodes(
             self.context, symbol_table, filter_func=filter_func
         )
 
-        for pid, task_comm, _task, fd_fields in fds_generator:
-            fd_num, _filp, full_path = fd_fields
-
-            fields = (pid, task_comm, fd_num, full_path)
+        for (
+            pid,
+            task_comm,
+            task,
+            fd_num,
+            filp,
+            full_path,
+            inode_metadata,
+        ) in fds_generator:
+            inode_num, itype, file_size, imode, ctime, mtime, atime = inode_metadata
+            fields = (
+                pid,
+                task_comm,
+                fd_num,
+                full_path,
+                inode_num,
+                itype,
+                imode,
+                ctime,
+                mtime,
+                atime,
+                file_size,
+            )
             yield (0, fields)
 
     def run(self):
         pids = self.config.get("pid", None)
         symbol_table = self.config["kernel"]
 
-        tree_grid_args = [("PID", int), ("Process", str), ("FD", int), ("Path", str)]
+        tree_grid_args = [
+            ("PID", int),
+            ("Process", str),
+            ("FD", int),
+            ("Path", str),
+            ("Inode", int),
+            ("Type", str),
+            ("Mode", str),
+            ("Changed", datetime.datetime),
+            ("Modified", datetime.datetime),
+            ("Accessed", datetime.datetime),
+            ("Size", int),
+        ]
         return renderers.TreeGrid(tree_grid_args, self._generator(pids, symbol_table))
+
+    def generate_timeline(self):
+        pids = self.config.get("pid", None)
+        symbol_table = self.config["kernel"]
+        for row in self._generator(pids, symbol_table):
+            _depth, row_data = row
+            description = f'Process {row_data[1]} ({row_data[0]}) Open "{row_data[3]}"'
+            yield description, timeliner.TimeLinerType.CHANGED, row_data[7]
+            yield description, timeliner.TimeLinerType.MODIFIED, row_data[8]
+            yield description, timeliner.TimeLinerType.ACCESSED, row_data[9]

volatility3/framework/plugins/linux/mountinfo.py

@@ -37,7 +37,7 @@ class MountInfo(plugins.PluginInterface):
 
     _required_framework_version = (2, 2, 0)
 
-    _version = (1, 0, 0)
+    _version = (1, 2, 0)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -143,10 +143,10 @@ class MountInfo(plugins.PluginInterface):
             sb_opts,
         )
 
+    @staticmethod
     def _get_tasks_mountpoints(
-        self,
         tasks: Iterable[interfaces.objects.ObjectInterface],
-        filtered_by_pids: bool,
+        filtered_by_pids: bool = False,
     ):
         seen_mountpoints = set()
         for task in tasks:
@@ -184,8 +184,8 @@ class MountInfo(plugins.PluginInterface):
         self,
         tasks: Iterable[interfaces.objects.ObjectInterface],
         mnt_ns_ids: List[int],
-        mount_format: bool,
-        filtered_by_pids: bool,
+        mount_format: bool = False,
+        filtered_by_pids: bool = False,
     ) -> Iterable[Tuple[int, Tuple]]:
         show_filter_warning = False
         for task, mnt, mnt_ns_id in self._get_tasks_mountpoints(
@@ -247,6 +247,37 @@ class MountInfo(plugins.PluginInterface):
                 "Could not filter by mount namespace id. This field is not available in this kernel."
             )
 
+    @classmethod
+    def get_superblocks(
+        cls,
+        context: interfaces.context.ContextInterface,
+        vmlinux_module_name: str,
+    ) -> Iterable[interfaces.objects.ObjectInterface]:
+        """Yield file system superblocks based on the task's mounted filesystems.
+
+        Args:
+            context: The context to retrieve required elements (layers, symbol tables) from
+            vmlinux_module_name: The name of the kernel module on which to operate
+
+        Yields:
+            super_block: Kernel's struct super_block object
+        """
+        # No filter so that we get all the mount namespaces from all tasks
+        tasks = pslist.PsList.list_tasks(context, vmlinux_module_name)
+
+        seen_sb_ptr = set()
+        for task, mnt, _mnt_ns_id in cls._get_tasks_mountpoints(tasks):
+            path_root = linux.LinuxUtilities.get_path_mnt(task, mnt)
+            if not path_root:
+                continue
+
+            sb_ptr = mnt.get_mnt_sb()
+            if not sb_ptr or sb_ptr in seen_sb_ptr:
+                continue
+            seen_sb_ptr.add(sb_ptr)
+
+            yield sb_ptr.dereference(), path_root
+
     def run(self):
         pids = self.config.get("pids")
         mount_ns_ids = self.config.get("mntns")

volatility3/framework/plugins/linux/pagecache.py

@@ -0,0 +1,531 @@
+# 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 math
+import logging
+import datetime
+from dataclasses import dataclass, astuple
+from typing import List, Set, Type, Iterable
+
+from volatility3.framework import renderers, interfaces
+from volatility3.framework.renderers import format_hints
+from volatility3.framework.interfaces import plugins
+from volatility3.framework.configuration import requirements
+from volatility3.plugins import timeliner
+from volatility3.plugins.linux import mountinfo
+
+vollog = logging.getLogger(__name__)
+
+
+@dataclass
+class InodeUser:
+    """Inode user representation, featuring augmented information and formatted fields.
+    This is the data the plugin will eventually display.
+    """
+
+    superblock_addr: int
+    mountpoint: str
+    device: str
+    inode_num: int
+    inode_addr: int
+    type: str
+    inode_pages: int
+    cached_pages: int
+    file_mode: str
+    access_time: str
+    modification_time: str
+    change_time: str
+    path: str
+
+
+@dataclass
+class InodeInternal:
+    """Inode internal representation containing only the core objects
+
+    Fields:
+        superblock: 'super_block' struct
+        mountpoint: Superblock mountpoint path
+        inode: 'inode' struct
+        path: Dentry full path
+    """
+
+    superblock: interfaces.objects.ObjectInterface
+    mountpoint: str
+    inode: interfaces.objects.ObjectInterface
+    path: str
+
+    def to_user(
+        self, kernel_layer: interfaces.layers.TranslationLayerInterface
+    ) -> InodeUser:
+        """Augment the inode information to be presented to the user
+
+        Args:
+            kernel_layer: The kernel layer to obtain the page size
+
+        Returns:
+            An InodeUser dataclass
+        """
+        # Ensure all types are atomic immutable. Otherwise, astuple() will take a long
+        # time doing a deepcopy of the Volatility objects.
+        superblock_addr = self.superblock.vol.offset
+        device = f"{self.superblock.major}:{self.superblock.minor}"
+        inode_num = int(self.inode.i_ino)
+        inode_addr = self.inode.vol.offset
+        inode_type = self.inode.get_inode_type() or renderers.UnparsableValue()
+        # Round up the number of pages to fit the inode's size
+        inode_pages = int(math.ceil(self.inode.i_size / float(kernel_layer.page_size)))
+        cached_pages = int(self.inode.i_mapping.nrpages)
+        file_mode = self.inode.get_file_mode()
+        access_time_dt = self.inode.get_access_time()
+        modification_time_str = self.inode.get_modification_time()
+        change_time_str = self.inode.get_change_time()
+
+        inode_user = InodeUser(
+            superblock_addr=superblock_addr,
+            mountpoint=self.mountpoint,
+            device=device,
+            inode_num=inode_num,
+            inode_addr=inode_addr,
+            type=inode_type,
+            inode_pages=inode_pages,
+            cached_pages=cached_pages,
+            file_mode=file_mode,
+            access_time=access_time_dt,
+            modification_time=modification_time_str,
+            change_time=change_time_str,
+            path=self.path,
+        )
+        return inode_user
+
+
+class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
+    """Lists files from memory"""
+
+    _required_framework_version = (2, 0, 0)
+
+    _version = (1, 0, 0)
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+            requirements.PluginRequirement(
+                name="mountinfo", plugin=mountinfo.MountInfo, version=(1, 2, 0)
+            ),
+            requirements.ListRequirement(
+                name="type",
+                description="List of space-separated file type filters i.e. --type REG DIR",
+                element_type=str,
+                optional=True,
+            ),
+            requirements.StringRequirement(
+                name="find",
+                description="Filename (full path) to find",
+                optional=True,
+            ),
+        ]
+
+    @staticmethod
+    def _follow_symlink(
+        inode: interfaces.objects.ObjectInterface,
+        symlink_path: str,
+    ) -> str:
+        """Follows (fast) symlinks (kernels >= 4.2.x).
+        Fast symlinks are filesystem agnostic.
+
+        Args:
+            inode: The inode (or pointer) to dump
+            symlink_path: The symlink name
+
+        Returns:
+            If it can resolve the symlink, it returns a string "symlink_path -> target_path"
+            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:
+            i_link_str = inode.i_link.dereference().cast(
+                "string", max_length=255, encoding="utf-8", errors="replace"
+            )
+            symlink_path = f"{symlink_path} -> {i_link_str}"
+
+        return symlink_path
+
+    @classmethod
+    def _walk_dentry(
+        cls,
+        seen_dentries: Set[int],
+        root_dentry: interfaces.objects.ObjectInterface,
+        parent_dir: str,
+    ):
+        """Walks dentries recursively
+
+        Args:
+            seen_dentries: A set to ensure each dentry is processed only once
+            root_dentry: Root dentry object
+            parent_dir: Parent directory path
+
+        Yields:
+           file_path: Filename including path
+           dentry: Dentry object
+        """
+
+        for dentry in root_dentry.get_subdirs():
+            dentry_addr = dentry.vol.offset
+
+            # corruption
+            if dentry_addr == root_dentry.vol.offset:
+                continue
+
+            if dentry_addr in seen_dentries:
+                continue
+
+            seen_dentries.add(dentry_addr)
+
+            inode = dentry.d_inode
+            if not (inode and inode.is_valid()):
+                continue
+
+            # This allows us to have consistent paths
+            if dentry.d_name.name:
+                basename = dentry.d_name.name_as_str()
+                # Do NOT use os.path.join() below
+                file_path = parent_dir + "/" + basename
+            else:
+                continue
+
+            yield file_path, dentry
+
+            if inode.is_dir:
+                yield from cls._walk_dentry(seen_dentries, dentry, parent_dir=file_path)
+
+    @classmethod
+    def get_inodes(
+        cls,
+        context: interfaces.context.ContextInterface,
+        vmlinux_module_name: str,
+    ) -> Iterable[InodeInternal]:
+        """Retrieves the inodes from the superblocks
+
+        Args:
+            context: The context that the plugin will operate within
+            vmlinux_module_name: The name of the kernel module on which to operate
+
+        Yields:
+            An InodeInternal object
+        """
+
+        superblocks_iter = mountinfo.MountInfo.get_superblocks(
+            context=context,
+            vmlinux_module_name=vmlinux_module_name,
+        )
+
+        seen_inodes = set()
+        seen_dentries = set()
+        for superblock, mountpoint in superblocks_iter:
+            parent_dir = "" if mountpoint == "/" else mountpoint
+
+            # Superblock root dentry
+            root_dentry_ptr = superblock.s_root
+            if not root_dentry_ptr:
+                continue
+
+            root_dentry = root_dentry_ptr.dereference()
+
+            # Dentry sanity check
+            if not root_dentry.is_root():
+                continue
+
+            # More dentry/inode sanity checks
+            root_inode_ptr = root_dentry.d_inode
+            if not root_inode_ptr:
+                continue
+            root_inode = root_inode_ptr.dereference()
+            if not root_inode.is_valid():
+                continue
+
+            # Inode already processed?
+            if root_inode_ptr in seen_inodes:
+                continue
+            seen_inodes.add(root_inode_ptr)
+
+            root_path = mountpoint
+
+            inode_in = InodeInternal(
+                superblock=superblock,
+                mountpoint=mountpoint,
+                inode=root_inode,
+                path=root_path,
+            )
+            yield inode_in
+
+            # Children
+            for file_path, file_dentry in cls._walk_dentry(
+                seen_dentries, root_dentry, parent_dir
+            ):
+                if not file_dentry:
+                    continue
+                # Dentry/inode sanity checks
+                file_inode_ptr = file_dentry.d_inode
+                if not file_inode_ptr:
+                    continue
+                file_inode = file_inode_ptr.dereference()
+                if not file_inode.is_valid():
+                    continue
+
+                # Inode already processed?
+                if file_inode_ptr in seen_inodes:
+                    continue
+                seen_inodes.add(file_inode_ptr)
+
+                file_path = cls._follow_symlink(file_inode_ptr, file_path)
+                inode_in = InodeInternal(
+                    superblock=superblock,
+                    mountpoint=mountpoint,
+                    inode=file_inode,
+                    path=file_path,
+                )
+                yield inode_in
+
+    def _generator(self):
+        vmlinux_module_name = self.config["kernel"]
+        vmlinux = self.context.modules[vmlinux_module_name]
+        vmlinux_layer = self.context.layers[vmlinux.layer_name]
+
+        inodes_iter = self.get_inodes(
+            context=self.context,
+            vmlinux_module_name=vmlinux_module_name,
+        )
+
+        types_filter = self.config["type"]
+        for inode_in in inodes_iter:
+            if types_filter and inode_in.inode.get_inode_type() not in types_filter:
+                continue
+
+            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):
+        """Generates tuples of (description, timestamp_type, timestamp)
+
+        These need not be generated in any particular order, sorting
+        will be done later
+        """
+        vmlinux_module_name = self.config["kernel"]
+        vmlinux = self.context.modules[vmlinux_module_name]
+        vmlinux_layer = self.context.layers[vmlinux.layer_name]
+
+        inodes_iter = self.get_inodes(
+            context=self.context,
+            vmlinux_module_name=vmlinux_module_name,
+        )
+
+        for inode_in in inodes_iter:
+            inode_out = inode_in.to_user(vmlinux_layer)
+            description = f"Cached Inode for {inode_out.path}"
+            yield description, timeliner.TimeLinerType.ACCESSED, inode_out.access_time
+            yield description, timeliner.TimeLinerType.MODIFIED, inode_out.modification_time
+            yield description, timeliner.TimeLinerType.CHANGE, inode_out.change_time
+
+    @staticmethod
+    def format_fields_with_headers(headers, generator):
+        """Uses the headers type to cast the fields obtained from the generator"""
+        for level, fields in generator:
+            formatted_fields = []
+            for header, field in zip(headers, fields):
+                header_type = header[1]
+
+                if isinstance(
+                    field, (header_type, interfaces.renderers.BaseAbsentValue)
+                ):
+                    formatted_field = field
+                else:
+                    formatted_field = header_type(field)
+
+                formatted_fields.append(formatted_field)
+            yield level, formatted_fields
+
+    def run(self):
+        headers = [
+            ("SuperblockAddr", format_hints.Hex),
+            ("MountPoint", str),
+            ("Device", str),
+            ("InodeNum", int),
+            ("InodeAddr", format_hints.Hex),
+            ("FileType", str),
+            ("InodePages", int),
+            ("CachedPages", int),
+            ("FileMode", str),
+            ("AccessTime", datetime.datetime),
+            ("ModificationTime", datetime.datetime),
+            ("ChangeTime", datetime.datetime),
+            ("FilePath", str),
+        ]
+
+        return renderers.TreeGrid(
+            headers, self.format_fields_with_headers(headers, self._generator())
+        )
+
+
+class InodePages(plugins.PluginInterface):
+    """Lists and recovers cached inode pages"""
+
+    _required_framework_version = (2, 0, 0)
+
+    _version = (1, 0, 0)
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+            requirements.PluginRequirement(
+                name="files", plugin=Files, version=(1, 0, 0)
+            ),
+            requirements.StringRequirement(
+                name="find",
+                description="Filename (full path) to find ",
+                optional=True,
+            ),
+            requirements.IntRequirement(
+                name="inode",
+                description="Inode address",
+                optional=True,
+            ),
+            requirements.StringRequirement(
+                name="dump",
+                description="Output file path",
+                optional=True,
+            ),
+        ]
+
+    @staticmethod
+    def write_inode_content_to_file(
+        inode: interfaces.objects.ObjectInterface,
+        filename: str,
+        open_method: Type[interfaces.plugins.FileHandlerInterface],
+        vmlinux_layer: interfaces.layers.TranslationLayerInterface,
+    ) -> None:
+        """Extracts the inode's contents from the page cache and saves them to a file
+
+        Args:
+            inode: The inode to dump
+            filename: Filename for writing the inode content
+            open_method: class for constructing output files
+            vmlinux_layer: The kernel layer to obtain the page size
+        """
+        if not inode.is_reg:
+            vollog.error("The inode is not a regular file")
+            return
+
+        # By using truncate/seek, provided the filesystem supports it, a sparse file will be
+        # 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.
+        try:
+            with open_method(filename) as f:
+                inode_size = inode.i_size
+                f.truncate(inode_size)
+
+                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:
+                        vollog.error(
+                            "Page out of file bounds: inode 0x%x, inode size %d, page index %d",
+                            inode.vol.object,
+                            inode_size,
+                            page_idx,
+                        )
+                    f.seek(current_fp)
+                    f.write(page_bytes)
+
+        except IOError as e:
+            vollog.error("Unable to write to file (%s): %s", filename, e)
+
+    def _generator(self):
+        vmlinux_module_name = self.config["kernel"]
+        vmlinux = self.context.modules[vmlinux_module_name]
+        vmlinux_layer = self.context.layers[vmlinux.layer_name]
+
+        if self.config["inode"] and self.config["find"]:
+            vollog.error("Cannot use --inode and --find simultaneously")
+            return
+
+        if self.config["find"]:
+            inodes_iter = Files.get_inodes(
+                context=self.context,
+                vmlinux_module_name=vmlinux_module_name,
+            )
+            for inode_in in inodes_iter:
+                if inode_in.path == self.config["find"]:
+                    inode = inode_in.inode
+                    break  # Only the first match
+
+        elif self.config["inode"]:
+            inode = vmlinux.object("inode", self.config["inode"], absolute=True)
+        else:
+            vollog.error("You must use either --inode or --find")
+            return
+
+        if not inode.is_reg:
+            vollog.error("The inode is not a regular file")
+            return
+
+        inode_size = inode.i_size
+        if not inode.is_valid():
+            vollog.error("Invalid inode at 0x%x", self.config["inode"])
+            return
+
+        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"]:
+            filename = self.config["dump"]
+            vollog.info("[*] Writing inode at 0x%x to '%s'", inode.vol.offset, filename)
+            self.write_inode_content_to_file(inode, filename, self.open, vmlinux_layer)
+
+    def run(self):
+        headers = [
+            ("PageVAddr", format_hints.Hex),
+            ("PagePAddr", format_hints.Hex),
+            ("MappingAddr", format_hints.Hex),
+            ("Index", int),
+            ("DumpSafe", bool),
+            ("Flags", str),
+        ]
+
+        return renderers.TreeGrid(
+            headers, Files.format_fields_with_headers(headers, self._generator())
+        )

volatility3/framework/plugins/linux/pidhashtable.py

@@ -0,0 +1,255 @@
+# 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
+
+from volatility3.framework import renderers, interfaces, constants
+from volatility3.framework.symbols import linux
+from volatility3.framework.renderers import format_hints
+from volatility3.framework.interfaces import plugins
+from volatility3.framework.configuration import requirements
+from volatility3.plugins.linux import pslist
+
+vollog = logging.getLogger(__name__)
+
+
+class PIDHashTable(plugins.PluginInterface):
+    """Enumerates processes through the PID hash table"""
+
+    _required_framework_version = (2, 0, 0)
+
+    _version = (1, 0, 1)
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+            requirements.PluginRequirement(
+                name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
+            ),
+            requirements.VersionRequirement(
+                name="linuxutils", component=linux.LinuxUtilities, version=(2, 1, 0)
+            ),
+            requirements.BooleanRequirement(
+                name="decorate_comm",
+                description="Show `user threads` comm in curly brackets, and `kernel threads` comm in square brackets",
+                optional=True,
+                default=False,
+            ),
+        ]
+
+    def _is_valid_task(self, task) -> bool:
+        return bool(task and task.pid > 0 and task.parent.is_readable())
+
+    def _get_pidtype_pid(self):
+        vmlinux = self.context.modules[self.config["kernel"]]
+
+        # The pid_type enumeration is present since 2.5.37, just in case
+        pid_type_enum = vmlinux.get_enumeration("pid_type")
+        if not pid_type_enum:
+            vollog.error("Cannot find pid_type enum. Unsupported kernel")
+            return None
+
+        pidtype_pid = pid_type_enum.choices.get("PIDTYPE_PID")
+        if pidtype_pid is None:
+            vollog.error("Cannot find PIDTYPE_PID. Unsupported kernel")
+            return None
+
+        # Typically PIDTYPE_PID = 0
+        return pidtype_pid
+
+    def _get_pidhash_array(self):
+        vmlinux = self.context.modules[self.config["kernel"]]
+
+        pidhash_shift = vmlinux.object_from_symbol("pidhash_shift")
+        pidhash_size = 1 << pidhash_shift
+
+        array_type_name = vmlinux.symbol_table_name + constants.BANG + "array"
+
+        pidhash_ptr = vmlinux.object_from_symbol("pid_hash")
+        # pidhash is an array of hlist_heads
+        pidhash = self._context.object(
+            array_type_name,
+            offset=pidhash_ptr,
+            subtype=vmlinux.get_type("hlist_head"),
+            count=pidhash_size,
+            layer_name=vmlinux.layer_name,
+        )
+
+        return pidhash
+
+    def _walk_upid(self, seen_upids, upid):
+        vmlinux = self.context.modules[self.config["kernel"]]
+        vmlinux_layer = self.context.layers[vmlinux.layer_name]
+
+        while upid and vmlinux_layer.is_valid(upid.vol.offset):
+            if upid.vol.offset in seen_upids:
+                break
+            seen_upids.add(upid.vol.offset)
+
+            pid_chain = upid.pid_chain
+            if not (pid_chain.next and pid_chain.next.is_readable()):
+                break
+
+            upid = linux.LinuxUtilities.container_of(
+                pid_chain.next, "upid", "pid_chain", vmlinux
+            )
+
+    def _get_upids(self):
+        vmlinux = self.context.modules[self.config["kernel"]]
+
+        # 2.6.24 <= kernels < 4.15
+        pidhash = self._get_pidhash_array()
+
+        seen_upids = set()
+        for hlist in pidhash:
+            # each entry in the hlist is a upid which is wrapped in a pid
+            ent = hlist.first
+
+            while ent and ent.is_readable():
+                # upid->pid_chain exists 2.6.24 <= kernel < 4.15
+                upid = linux.LinuxUtilities.container_of(
+                    ent.vol.offset, "upid", "pid_chain", vmlinux
+                )
+
+                if upid.vol.offset in seen_upids:
+                    break
+
+                self._walk_upid(seen_upids, upid)
+
+                ent = ent.next
+
+        return seen_upids
+
+    def _pid_hash_implementation(self):
+        vmlinux = self.context.modules[self.config["kernel"]]
+
+        # 2.6.24 <= kernels < 4.15
+        task_pids_off = vmlinux.get_type("task_struct").relative_child_offset("pids")
+        pidtype_pid = self._get_pidtype_pid()
+
+        for upid in self._get_upids():
+            pid = linux.LinuxUtilities.container_of(upid, "pid", "numbers", vmlinux)
+            if not pid:
+                continue
+
+            pid_tasks_0 = pid.tasks[pidtype_pid].first
+            if not (pid_tasks_0 and pid_tasks_0.is_readable()):
+                continue
+
+            task = vmlinux.object(
+                "task_struct", offset=pid_tasks_0 - task_pids_off, absolute=True
+            )
+            if self._is_valid_task(task):
+                yield task
+
+    def _task_for_radix_pid_node(self, nodep):
+        vmlinux = self.context.modules[self.config["kernel"]]
+
+        # kernels >= 4.15
+        pid = vmlinux.object("pid", offset=nodep, absolute=True)
+        pidtype_pid = self._get_pidtype_pid()
+
+        pid_tasks_0 = pid.tasks[pidtype_pid].first
+        if not (pid_tasks_0 and pid_tasks_0.is_readable()):
+            return None
+
+        task_struct_type = vmlinux.get_type("task_struct")
+        if task_struct_type.has_member("pids"):
+            member = "pids"
+        elif task_struct_type.has_member("pid_links"):
+            member = "pid_links"
+        else:
+            return None
+
+        task_pids_off = task_struct_type.relative_child_offset(member)
+        task = vmlinux.object(
+            "task_struct", offset=pid_tasks_0 - task_pids_off, absolute=True
+        )
+        return task
+
+    def _pid_namespace_idr(self):
+        vmlinux = self.context.modules[self.config["kernel"]]
+
+        # kernels >= 4.15
+        ns_addr = vmlinux.get_symbol("init_pid_ns").address
+        ns = vmlinux.object("pid_namespace", offset=ns_addr)
+
+        for page_addr in ns.idr.get_entries():
+            task = self._task_for_radix_pid_node(page_addr)
+            if self._is_valid_task(task):
+                yield task
+
+    def _determine_pid_func(self):
+        vmlinux = self.context.modules[self.config["kernel"]]
+
+        pid_hash = vmlinux.has_symbol("pid_hash") and vmlinux.has_symbol(
+            "pidhash_shift"
+        )  # 2.5.55 <= kernels < 4.15
+
+        has_pid_numbers = vmlinux.has_type("pid") and vmlinux.get_type(
+            "pid"
+        ).has_member(
+            "numbers"
+        )  # kernels >= 2.6.24
+
+        has_pid_chain = vmlinux.has_type("upid") and vmlinux.get_type(
+            "upid"
+        ).has_member(
+            "pid_chain"
+        )  # 2.6.24 <= kernels < 4.15
+
+        # kernels >= 4.15
+        pid_idr = vmlinux.has_type("pid_namespace") and vmlinux.get_type(
+            "pid_namespace"
+        ).has_member("idr")
+
+        if pid_idr:
+            # kernels >= 4.15
+            return self._pid_namespace_idr
+        elif pid_hash and has_pid_numbers and has_pid_numbers and has_pid_chain:
+            # 2.6.24 <= kernels < 4.15
+            return self._pid_hash_implementation
+
+        return None
+
+    def get_tasks(self) -> interfaces.objects.ObjectInterface:
+        """Enumerates processes through the PID hash table
+
+        Yields:
+            task_struct objects
+        """
+        pid_func = self._determine_pid_func()
+        if not pid_func:
+            vollog.error("Cannot determine which PID hash table this kernel is using")
+            return
+
+        yield from sorted(pid_func(), key=lambda t: (t.tgid, t.pid))
+
+    def _generator(
+        self, decorate_comm: bool = False
+    ) -> interfaces.objects.ObjectInterface:
+        for task in self.get_tasks():
+            offset, pid, tid, ppid, name = pslist.PsList.get_task_fields(
+                task, decorate_comm
+            )
+            fields = format_hints.Hex(offset), pid, tid, ppid, name
+            yield 0, fields
+
+    def run(self):
+        decorate_comm = self.config.get("decorate_comm")
+
+        headers = [
+            ("OFFSET", format_hints.Hex),
+            ("PID", int),
+            ("TID", int),
+            ("PPID", int),
+            ("COMM", str),
+        ]
+        return renderers.TreeGrid(headers, self._generator(decorate_comm=decorate_comm))

volatility3/framework/plugins/linux/sockstat.py

@@ -22,7 +22,7 @@ class SockHandlers(interfaces.configuration.VersionableInterface):
 
     _required_framework_version = (2, 0, 0)
 
-    _version = (1, 0, 0)
+    _version = (2, 0, 0)
 
     def __init__(self, vmlinux, task):
         self._vmlinux = vmlinux
@@ -151,17 +151,15 @@ class SockHandlers(interfaces.configuration.VersionableInterface):
 
         bpfprog = sock_filter.prog
 
-        BPF_PROG_TYPE_UNSPEC = 0  # cBPF filter
-        try:
-            bpfprog_type = bpfprog.get_type()
-            if bpfprog_type == BPF_PROG_TYPE_UNSPEC:
-                return  # cBPF filter
-        except AttributeError:
+        bpfprog_type = bpfprog.get_type()
+        if not bpfprog_type:
             # kernel < 3.18.140, it's a cBPF filter
             return None
 
-        BPF_PROG_TYPE_SOCKET_FILTER = 1  # eBPF filter
-        if bpfprog_type != BPF_PROG_TYPE_SOCKET_FILTER:
+        if bpfprog_type == "BPF_PROG_TYPE_UNSPEC":
+            return None  # cBPF filter
+
+        if bpfprog_type != "BPF_PROG_TYPE_SOCKET_FILTER":
             socket_filter["bpf_filter_type"] = f"UNK({bpfprog_type})"
             vollog.warning(f"Unexpected BPF type {bpfprog_type} for a socket")
             return None
@@ -509,7 +507,7 @@ class Sockstat(plugins.PluginInterface):
         dfop_addr = vmlinux.object_from_symbol("sockfs_dentry_operations").vol.offset
 
         fd_generator = lsof.Lsof.list_fds(context, vmlinux.name, filter_func)
-        for _pid, _task_comm, task, fd_fields in fd_generator:
+        for _pid, task_comm, task, fd_fields in fd_generator:
             fd_num, filp, _full_path = fd_fields
 
             if filp.f_op not in (sfop_addr, dfop_addr):
@@ -550,7 +548,7 @@ class Sockstat(plugins.PluginInterface):
             except AttributeError:
                 netns_id = NotAvailableValue()
 
-            yield task, netns_id, fd_num, family, sock_type, protocol, sock_fields
+            yield task_comm, task, netns_id, fd_num, family, sock_type, protocol, sock_fields
 
     def _format_fields(self, sock_stat, protocol):
         """Prepare the socket fields to be rendered
@@ -597,6 +595,7 @@ class Sockstat(plugins.PluginInterface):
         )
 
         for (
+            task_comm,
             task,
             netns_id,
             fd_num,
@@ -619,6 +618,7 @@ class Sockstat(plugins.PluginInterface):
 
             fields = (
                 netns_id,
+                task_comm,
                 task.pid,
                 fd_num,
                 format_hints.Hex(sock.vol.offset),
@@ -638,6 +638,7 @@ class Sockstat(plugins.PluginInterface):
 
         tree_grid_args = [
             ("NetNS", int),
+            ("Process Name", str),
             ("Pid", int),
             ("FD", int),
             ("Sock Offset", format_hints.Hex),

volatility3/framework/plugins/linux/vmayarascan.py

@@ -31,10 +31,7 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
                 name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
             ),
             requirements.PluginRequirement(
-                name="yarascan", plugin=yarascan.YaraScan, version=(1, 2, 0)
-            ),
-            requirements.VersionRequirement(
-                name="yarascanner", component=yarascan.YaraScanner, version=(2, 0, 0)
+                name="yarascan", plugin=yarascan.YaraScan, version=(2, 0, 0)
             ),
             requirements.ModuleRequirement(
                 name="kernel",
@@ -69,19 +66,29 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
             # get the proc_layer object from the context
             proc_layer = self.context.layers[proc_layer_name]
 
-            # scan the process layer with the yarascanner
-            for offset, rule_name, name, value in proc_layer.scan(
-                context=self.context,
-                scanner=yarascan.YaraScanner(rules=rules),
-                sections=self.get_vma_maps(task),
-            ):
-                yield 0, (
-                    format_hints.Hex(offset),
-                    task.tgid,
-                    rule_name,
-                    name,
-                    value,
-                )
+            for start, end in self.get_vma_maps(task):
+                for match in rules.match(
+                    data=proc_layer.read(start, end - start, True)
+                ):
+                    if yarascan.YaraScan.yara_returns_instances():
+                        for match_string in match.strings:
+                            for instance in match_string.instances:
+                                yield 0, (
+                                    format_hints.Hex(instance.offset + start),
+                                    task.UniqueProcessId,
+                                    match.rule,
+                                    match_string.identifier,
+                                    instance.matched_data,
+                                )
+                    else:
+                        for offset, name, value in match.strings:
+                            yield 0, (
+                                format_hints.Hex(offset + start),
+                                task.tgid,
+                                match.rule,
+                                name,
+                                value,
+                            )
 
     @staticmethod
     def get_vma_maps(

volatility3/framework/plugins/timeliner.py

@@ -45,6 +45,7 @@ class Timeliner(interfaces.plugins.PluginInterface):
     orders the results by time."""
 
     _required_framework_version = (2, 0, 0)
+    _version = (1, 1, 0)
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -198,9 +199,10 @@ class Timeliner(interfaces.plugins.PluginInterface):
                                     ),
                                 )
                             )
-            except Exception:
+            except Exception as e:
                 vollog.log(
-                    logging.INFO, f"Exception occurred running plugin: {plugin_name}"
+                    logging.INFO,
+                    f"Exception occurred running plugin: {plugin_name}: {e}",
                 )
                 vollog.log(logging.DEBUG, traceback.format_exc())
 
@@ -245,6 +247,18 @@ class Timeliner(interfaces.plugins.PluginInterface):
         filter_list = self.config["plugin-filter"]
         # Identify plugins that we can run which output datetimes
         for plugin_class in self.usable_plugins:
+            if not issubclass(plugin_class, TimeLinerInterface):
+                # get_usable_plugins() should filter this, but adding a safeguard just in case
+                continue
+
+            if filter_list and not any(
+                [
+                    filter in plugin_class.__module__ + "." + plugin_class.__name__
+                    for filter in filter_list
+                ]
+            ):
+                continue
+
             try:
                 automagics = automagic.choose_automagic(self.automagics, plugin_class)
 
@@ -276,15 +290,8 @@ class Timeliner(interfaces.plugins.PluginInterface):
                                 config_value,
                             )
 
-                if isinstance(plugin, TimeLinerInterface):
-                    if not len(filter_list) or any(
-                        [
-                            filter
-                            in plugin.__module__ + "." + plugin.__class__.__name__
-                            for filter in filter_list
-                        ]
-                    ):
-                        plugins_to_run.append(plugin)
+                plugins_to_run.append(plugin)
+
             except exceptions.UnsatisfiedException as excp:
                 # Remove the failed plugin from the list and continue
                 vollog.debug(

volatility3/framework/plugins/windows/debugregisters.py

@@ -0,0 +1,211 @@
+# 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
+
+# Full details on the techniques used in these plugins to detect EDR-evading malware
+# can be found in our 20 page whitepaper submitted to DEFCON along with the presentation
+# https://www.volexity.com/wp-content/uploads/2024/08/Defcon24_EDR_Evasion_Detection_White-Paper_Andrew-Case.pdf
+
+import logging
+
+from typing import Tuple, Optional, Generator, List, Dict
+
+from functools import partial
+
+from volatility3.framework import renderers, interfaces, exceptions
+from volatility3.framework.configuration import requirements
+from volatility3.framework.renderers import format_hints
+import volatility3.plugins.windows.pslist as pslist
+import volatility3.plugins.windows.threads as threads
+import volatility3.plugins.windows.pe_symbols as pe_symbols
+
+vollog = logging.getLogger(__name__)
+
+
+class DebugRegisters(interfaces.plugins.PluginInterface):
+    # version 2.6.0 adds support for scanning for 'Ethread' structures by pool tags
+    _required_framework_version = (2, 6, 0)
+    _version = (1, 0, 0)
+
+    @classmethod
+    def get_requirements(cls) -> List:
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Windows kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+            requirements.VersionRequirement(
+                name="pslist", component=pslist.PsList, version=(2, 0, 0)
+            ),
+            requirements.VersionRequirement(
+                name="pe_symbols", component=pe_symbols.PESymbols, version=(1, 0, 0)
+            ),
+        ]
+
+    @staticmethod
+    def _get_debug_info(
+        ethread: interfaces.objects.ObjectInterface,
+    ) -> Optional[Tuple[interfaces.objects.ObjectInterface, int, int, int, int, int]]:
+        """
+        Gathers information related to the debug registers for the given thread
+        Args:
+            ethread: the thread (_ETHREAD) to examine
+        Returns:
+            Tuple[interfaces.objects.ObjectInterface, int, int, int, int, int]: The owner process of the thread and the values for dr7, dr0, dr1, dr2, dr3
+        """
+        try:
+            dr7 = ethread.Tcb.TrapFrame.Dr7
+            state = ethread.Tcb.State
+        except exceptions.InvalidAddressException:
+            return None
+
+        # 0 = debug registers not active
+        # 4 = terminated
+        if dr7 == 0 or state == 4:
+            return None
+
+        try:
+            owner_proc = ethread.owning_process()
+        except (AttributeError, exceptions.InvalidAddressException):
+            return None
+
+        dr0 = ethread.Tcb.TrapFrame.Dr0
+        dr1 = ethread.Tcb.TrapFrame.Dr1
+        dr2 = ethread.Tcb.TrapFrame.Dr2
+        dr3 = ethread.Tcb.TrapFrame.Dr3
+
+        # bail if all are 0
+        if not (dr0 or dr1 or dr2 or dr3):
+            return None
+
+        return owner_proc, dr7, dr0, dr1, dr2, dr3
+
+    def _generator(
+        self,
+    ) -> Generator[
+        Tuple[
+            int,
+            Tuple[
+                str,
+                int,
+                int,
+                int,
+                int,
+                format_hints.Hex,
+                str,
+                str,
+                format_hints.Hex,
+                str,
+                str,
+                format_hints.Hex,
+                str,
+                str,
+                format_hints.Hex,
+                str,
+                str,
+            ],
+        ],
+        None,
+        None,
+    ]:
+        kernel = self.context.modules[self.config["kernel"]]
+
+        vads_cache: Dict[int, pe_symbols.ranges_type] = {}
+
+        proc_modules = None
+
+        procs = pslist.PsList.list_processes(
+            context=self.context,
+            layer_name=kernel.layer_name,
+            symbol_table=kernel.symbol_table_name,
+        )
+
+        for proc in procs:
+            for thread in threads.Threads.list_threads(kernel, proc):
+                debug_info = self._get_debug_info(thread)
+                if not debug_info:
+                    continue
+
+                owner_proc, dr7, dr0, dr1, dr2, dr3 = debug_info
+
+                vads = pe_symbols.PESymbols.get_vads_for_process_cache(
+                    vads_cache, owner_proc
+                )
+                if not vads:
+                    continue
+
+                # this lookup takes a while, so only perform if we need to
+                if not proc_modules:
+                    proc_modules = pe_symbols.PESymbols.get_process_modules(
+                        self.context, kernel.layer_name, kernel.symbol_table_name, None
+                    )
+                    path_and_symbol = partial(
+                        pe_symbols.PESymbols.path_and_symbol_for_address,
+                        self.context,
+                        self.config_path,
+                        proc_modules,
+                    )
+
+                file0, sym0 = path_and_symbol(vads, dr0)
+                file1, sym1 = path_and_symbol(vads, dr1)
+                file2, sym2 = path_and_symbol(vads, dr2)
+                file3, sym3 = path_and_symbol(vads, dr3)
+
+                # if none map to an actual file VAD then bail
+                if not (file0 or file1 or file2 or file3):
+                    continue
+
+                process_name = owner_proc.ImageFileName.cast(
+                    "string",
+                    max_length=owner_proc.ImageFileName.vol.count,
+                    errors="replace",
+                )
+
+                thread_tid = thread.Cid.UniqueThread
+
+                yield (
+                    0,
+                    (
+                        process_name,
+                        owner_proc.UniqueProcessId,
+                        thread_tid,
+                        thread.Tcb.State,
+                        dr7,
+                        format_hints.Hex(dr0),
+                        file0 or renderers.NotApplicableValue(),
+                        sym0 or renderers.NotApplicableValue(),
+                        format_hints.Hex(dr1),
+                        file1 or renderers.NotApplicableValue(),
+                        sym1 or renderers.NotApplicableValue(),
+                        format_hints.Hex(dr2),
+                        file2 or renderers.NotApplicableValue(),
+                        sym2 or renderers.NotApplicableValue(),
+                        format_hints.Hex(dr3),
+                        file3 or renderers.NotApplicableValue(),
+                        sym3 or renderers.NotApplicableValue(),
+                    ),
+                )
+
+    def run(self) -> renderers.TreeGrid:
+        return renderers.TreeGrid(
+            [
+                ("Process", str),
+                ("PID", int),
+                ("TID", int),
+                ("State", int),
+                ("Dr7", int),
+                ("Dr0", format_hints.Hex),
+                ("Range0", str),
+                ("Symbol0", str),
+                ("Dr1", format_hints.Hex),
+                ("Range1", str),
+                ("Symbol1", str),
+                ("Dr2", format_hints.Hex),
+                ("Range2", str),
+                ("Symbol2", str),
+                ("Dr3", format_hints.Hex),
+                ("Range3", str),
+                ("Symbol3", str),
+            ],
+            self._generator(),
+        )

volatility3/framework/plugins/windows/mftscan.py

@@ -29,7 +29,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                 architectures=["Intel32", "Intel64"],
             ),
             requirements.VersionRequirement(
-                name="yarascanner", component=yarascan.YaraScanner, version=(2, 0, 0)
+                name="yarascanner", component=yarascan.YaraScanner, version=(2, 1, 0)
             ),
         ]
 
@@ -38,7 +38,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
 
         # Yara Rule to scan for MFT Header Signatures
         rules = yarascan.YaraScan.process_yara_options(
-            {"yara_rules": "/FILE0|FILE\\*|BAAD/"}
+            {"yara_string": "/FILE0|FILE\\*|BAAD/"}
         )
 
         # Read in the Symbol File
@@ -197,7 +197,7 @@ class ADS(interfaces.plugins.PluginInterface):
 
         # Yara Rule to scan for MFT Header Signatures
         rules = yarascan.YaraScan.process_yara_options(
-            {"yara_rules": "/FILE0|FILE\\*|BAAD/"}
+            {"yara_string": "/FILE0|FILE\\*|BAAD/"}
         )
 
         # Read in the Symbol File

volatility3/framework/plugins/windows/orphan_kernel_threads.py

@@ -0,0 +1,96 @@
+# 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, Generator
+
+from volatility3.framework import interfaces, symbols
+from volatility3.framework.configuration import requirements
+from volatility3.plugins.windows import thrdscan, ssdt
+
+vollog = logging.getLogger(__name__)
+
+
+class Threads(thrdscan.ThrdScan):
+    """Lists process threads"""
+
+    _required_framework_version = (2, 4, 0)
+    _version = (1, 0, 0)
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.implementation = self.list_orphan_kernel_threads
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        # Since we're calling the plugin, make sure we have the plugin's requirements
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Windows kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+            requirements.PluginRequirement(
+                name="thrdscan", plugin=thrdscan.ThrdScan, version=(1, 1, 0)
+            ),
+            requirements.PluginRequirement(
+                name="ssdt", plugin=ssdt.SSDT, version=(1, 0, 0)
+            ),
+        ]
+
+    @classmethod
+    def list_orphan_kernel_threads(
+        cls,
+        context: interfaces.context.ContextInterface,
+        module_name: str,
+    ) -> Generator[interfaces.objects.ObjectInterface, None, None]:
+        """Yields thread objects of kernel threads that do not map to a module
+
+        Args:
+            cls
+            context: the context to operate upon
+            module_name: name of the module to use for scanning
+        Returns:
+            A generator of thread objects of orphaned threads
+        """
+        module = context.modules[module_name]
+        layer_name = module.layer_name
+        symbol_table = module.symbol_table_name
+
+        collection = ssdt.SSDT.build_module_collection(
+            context, layer_name, symbol_table
+        )
+
+        # FIXME - use a proper constant once established
+        # used to filter out smeared pointers
+        if symbols.symbol_table_is_64bit(context, symbol_table):
+            kernel_start = 0xFFFFF80000000000
+        else:
+            kernel_start = 0x80000000
+
+        for thread in thrdscan.ThrdScan.scan_threads(context, module_name):
+            # we don't want smeared or terminated threads
+            try:
+                proc = thread.owning_process()
+            except AttributeError:
+                continue
+
+            # we only care about kernel threads, 4 = System
+            # previous methods for determining if a thread was a kernel thread
+            # such as bit fields and flags are not stable in Win10+
+            # so we check if the thread is from the kernel itself or one its child
+            # kernel processes (MemCompression, Regsitry, ...)
+            if proc.UniqueProcessId != 4 and proc.InheritedFromUniqueProcessId != 4:
+                continue
+
+            if thread.StartAddress < kernel_start:
+                continue
+
+            module_symbols = list(
+                collection.get_module_symbols_by_absolute_location(thread.StartAddress)
+            )
+
+            # alert on threads that do not map to a module
+            if not module_symbols:
+                yield thread

volatility3/framework/plugins/windows/ssdt.py

@@ -30,7 +30,7 @@ class SSDT(plugins.PluginInterface):
                 architectures=["Intel32", "Intel64"],
             ),
             requirements.PluginRequirement(
-                name="modules", plugin=modules.Modules, version=(1, 0, 0)
+                name="modules", plugin=modules.Modules, version=(2, 0, 0)
             ),
         ]
 

volatility3/framework/plugins/windows/thrdscan.py

@@ -22,8 +22,8 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface)
     _version = (1, 1, 0)
 
     def __init__(self, *args, **kwargs):
-        self.implementation = self.scan_threads
         super().__init__(*args, **kwargs)
+        self.implementation = self.scan_threads
 
     @classmethod
     def get_requirements(cls):
@@ -48,8 +48,7 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface)
 
         Args:
             context: The context to retrieve required elements (layers, symbol tables) from
-            layer_name: The name of the layer on which to operate
-            symbol_table: The name of the table containing the kernel symbols
+            module_name: Name of the module to use for scanning
 
         Returns:
               A list of _ETHREAD objects found by scanning memory for the "Thre" / "Thr\\xE5" pool signatures

volatility3/framework/plugins/windows/threads.py

@@ -19,8 +19,8 @@ class Threads(thrdscan.ThrdScan):
     _version = (1, 0, 0)
 
     def __init__(self, *args, **kwargs):
-        self.implementation = self.list_process_threads
         super().__init__(*args, **kwargs)
+        self.implementation = self.list_process_threads
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -50,7 +50,6 @@ class Threads(thrdscan.ThrdScan):
 
         Args:
             proc: _EPROCESS object from which to list the VADs
-            filter_func: Function to take a virtual address descriptor value and return True if it should be filtered out
 
         Returns:
             A list of threads based on the process and filtered based on the filter function
@@ -64,22 +63,19 @@ class Threads(thrdscan.ThrdScan):
             seen.add(thread.vol.offset)
             yield thread
 
-    @classmethod
-    def filter_func(cls, config: interfaces.configuration.HierarchicalDict) -> Callable:
-        return pslist.PsList.create_pid_filter(config.get("pid", None))
-
     @classmethod
     def list_process_threads(
         cls,
         context: interfaces.context.ContextInterface,
         module_name: str,
-        filter_func: Callable,
     ) -> Iterable[interfaces.objects.ObjectInterface]:
         """Runs through all processes and lists threads for each process"""
         module = context.modules[module_name]
         layer_name = module.layer_name
         symbol_table_name = module.symbol_table_name
 
+        filter_func = pslist.PsList.create_pid_filter(context.config.get("pid", None))
+
         for proc in pslist.PsList.list_processes(
             context=context,
             layer_name=layer_name,

volatility3/framework/plugins/windows/truecrypt.py

@@ -33,7 +33,7 @@ class Passphrase(interfaces.plugins.PluginInterface):
                 architectures=["Intel32", "Intel64"],
             ),
             requirements.VersionRequirement(
-                name="modules", component=modules.Modules, version=(1, 1, 0)
+                name="modules", component=modules.Modules, version=(2, 0, 0)
             ),
             requirements.IntRequirement(
                 name="min-length",

volatility3/framework/plugins/windows/unhooked_system_calls.py

@@ -0,0 +1,208 @@
+# 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
+
+# Full details on the techniques used in these plugins to detect EDR-evading malware
+# can be found in our 20 page whitepaper submitted to DEFCON along with the presentation
+# https://www.volexity.com/wp-content/uploads/2024/08/Defcon24_EDR_Evasion_Detection_White-Paper_Andrew-Case.pdf
+
+import logging
+
+from typing import Dict, Tuple, List, Generator
+
+from volatility3.framework import interfaces, exceptions
+from volatility3.framework import renderers
+from volatility3.framework.configuration import requirements
+from volatility3.framework.objects import utility
+from volatility3.plugins.windows import pslist, pe_symbols
+
+vollog = logging.getLogger(__name__)
+
+
+class unhooked_system_calls(interfaces.plugins.PluginInterface):
+    """Looks for signs of Skeleton Key malware"""
+
+    _required_framework_version = (2, 4, 0)
+
+    system_calls = {
+        "ntdll.dll": {
+            pe_symbols.wanted_names_identifier: [
+                "NtCreateThread",
+                "NtProtectVirtualMemory",
+                "NtReadVirtualMemory",
+                "NtOpenProcess",
+                "NtWriteFile",
+                "NtQueryVirtualMemory",
+                "NtAllocateVirtualMemory",
+                "NtWorkerFactoryWorkerReady",
+                "NtAcceptConnectPort",
+                "NtAddDriverEntry",
+                "NtAdjustPrivilegesToken",
+                "NtAlpcCreatePort",
+                "NtClose",
+                "NtCreateFile",
+                "NtCreateMutant",
+                "NtOpenFile",
+                "NtOpenIoCompletion",
+                "NtOpenJobObject",
+                "NtOpenKey",
+                "NtOpenKeyEx",
+                "NtOpenThread",
+                "NtOpenThreadToken",
+                "NtOpenThreadTokenEx",
+                "NtWriteVirtualMemory",
+                "NtTraceEvent",
+                "NtTranslateFilePath",
+                "NtUmsThreadYield",
+                "NtUnloadDriver",
+                "NtUnloadKey",
+                "NtUnloadKey2",
+                "NtUnloadKeyEx",
+                "NtCreateKey",
+                "NtCreateSection",
+                "NtDeleteKey",
+                "NtDeleteValueKey",
+                "NtDuplicateObject",
+                "NtQueryValueKey",
+                "NtReplaceKey",
+                "NtRequestWaitReplyPort",
+                "NtRestoreKey",
+                "NtSetContextThread",
+                "NtSetSecurityObject",
+                "NtSetValueKey",
+                "NtSystemDebugControl",
+                "NtTerminateProcess",
+            ]
+        }
+    }
+
+    # This data structure is used to track unique implementations of functions across processes
+    # The outer dictionary holds the module name (e.g., ntdll.dll)
+    # The next dictionary holds the function names (NtTerminateProcess, NtSetValueKey, etc.) inside a module
+    # The innermost dictionary holds the unique implementation (bytes) of a function across processes
+    # Each implementation is tracked along with the process(es) that host it
+    # For systems without malware, all functions should have the same implementation
+    # When API hooking/module unhooking is done, the victim (infected) processes will have unique implementations
+    _code_bytes_type = Dict[str, Dict[str, Dict[bytes, List[Tuple[int, str]]]]]
+
+    @classmethod
+    def get_requirements(cls) -> List:
+        # Since we're calling the plugin, make sure we have the plugin's requirements
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Windows kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+            requirements.VersionRequirement(
+                name="pslist", component=pslist.PsList, version=(2, 0, 0)
+            ),
+            requirements.PluginRequirement(
+                name="pe_symbols", plugin=pe_symbols.PESymbols, version=(1, 0, 0)
+            ),
+        ]
+
+    def _gather_code_bytes(
+        self,
+        kernel: interfaces.context.ModuleInterface,
+        found_symbols: pe_symbols.found_symbols_type,
+    ) -> _code_bytes_type:
+        """
+        Enumerates the desired DLLs and function implementations in each process
+        Groups based on unique implementations of each DLLs' functions
+        The purpose is to detect when a function has different implementations (code)
+        in different processes.
+        This very effectively detects code injection.
+        """
+        code_bytes: unhooked_system_calls._code_bytes_type = {}
+
+        procs = pslist.PsList.list_processes(
+            context=self.context,
+            layer_name=kernel.layer_name,
+            symbol_table=kernel.symbol_table_name,
+        )
+
+        for proc in procs:
+            try:
+                proc_id = proc.UniqueProcessId
+                proc_name = utility.array_to_string(proc.ImageFileName)
+                proc_layer_name = proc.add_process_layer()
+            except exceptions.InvalidAddressException:
+                continue
+
+            for dll_name, functions in found_symbols.items():
+                for func_name, func_addr in functions:
+                    try:
+                        fbytes = self.context.layers[proc_layer_name].read(
+                            func_addr, 0x20
+                        )
+                    except exceptions.InvalidAddressException:
+                        continue
+
+                    # see the definition of _code_bytes_type for details of this data structure
+                    if dll_name not in code_bytes:
+                        code_bytes[dll_name] = {}
+
+                    if func_name not in code_bytes[dll_name]:
+                        code_bytes[dll_name][func_name] = {}
+
+                    if fbytes not in code_bytes[dll_name][func_name]:
+                        code_bytes[dll_name][func_name][fbytes] = []
+
+                    code_bytes[dll_name][func_name][fbytes].append((proc_id, proc_name))
+
+        return code_bytes
+
+    def _generator(self) -> Generator[Tuple[int, Tuple[str, str, int]], None, None]:
+        kernel = self.context.modules[self.config["kernel"]]
+
+        found_symbols = pe_symbols.PESymbols.addresses_for_process_symbols(
+            self.context,
+            self.config_path,
+            kernel.layer_name,
+            kernel.symbol_table_name,
+            unhooked_system_calls.system_calls,
+        )
+
+        # code_bytes[dll_name][func_name][func_bytes]
+        code_bytes = self._gather_code_bytes(kernel, found_symbols)
+
+        # walk the functions that were evaluated
+        for functions in code_bytes.values():
+            # cbb is the distinct groups of bytes (instructions)
+            # for this function across processes
+            for func_name, cbb in functions.items():
+                # the dict key here is the raw instructions, which is not helpful to look at
+                # the values are the list of tuples for the (proc_id, proc_name) pairs for this set of bytes (instructions)
+                cb = list(cbb.values())
+
+                # if all processes map to the same implementation, then no malware is present
+                if len(cb) == 1:
+                    yield 0, (func_name, "", len(cb[0]))
+                else:
+                    # if there are differing implementations then it means
+                    # that malware has overwritten system call(s) in infected processes
+                    # max_idx and small_idx find which implementation of a system call has the least processes
+                    # as all observed malware and open source projects only infected a few targets, leaving the
+                    # rest with the original EDR hooks in place
+                    max_idx = 0 if len(cb[0]) > len(cb[1]) else 1
+                    small_idx = (~max_idx) & 1
+
+                    ps = []
+
+                    # gather processes on small_idx since these are the malware infected ones
+                    for pid, pname in cb[small_idx]:
+                        ps.append("{:d}:{}".format(pid, pname))
+
+                    proc_names = ", ".join(ps)
+
+                    yield 0, (func_name, proc_names, len(cb[max_idx]))
+
+    def run(self) -> renderers.TreeGrid:
+        return renderers.TreeGrid(
+            [
+                ("Function", str),
+                ("Distinct Implementations", str),
+                ("Total Implementations", int),
+            ],
+            self._generator(),
+        )

volatility3/framework/plugins/windows/vadinfo.py

@@ -3,7 +3,7 @@
 #
 
 import logging
-from typing import Callable, List, Generator, Iterable, Type, Optional
+from typing import Callable, List, Generator, Iterable, Type, Optional, Tuple
 
 from volatility3.framework import renderers, interfaces, exceptions
 from volatility3.framework.configuration import requirements
@@ -196,11 +196,31 @@ class VadInfo(interfaces.plugins.PluginInterface):
 
         return file_handle
 
-    def _generator(self, procs):
+    def _generator(self, procs: List[interfaces.objects.ObjectInterface]) -> Generator[
+        Tuple[
+            int,
+            Tuple[
+                int,
+                str,
+                format_hints.Hex,
+                format_hints.Hex,
+                format_hints.Hex,
+                str,
+                str,
+                int,
+                int,
+                format_hints.Hex,
+                str,
+                str,
+            ],
+        ],
+        None,
+        None,
+    ]:
         kernel = self.context.modules[self.config["kernel"]]
         kernel_layer = self.context.layers[kernel.layer_name]
 
-        def passthrough(_: interfaces.objects.ObjectInterface) -> bool:
+        def passthrough(x: interfaces.objects.ObjectInterface) -> bool:
             return False
 
         filter_func = passthrough
@@ -250,7 +270,7 @@ class VadInfo(interfaces.plugins.PluginInterface):
                     ),
                 )
 
-    def run(self):
+    def run(self) -> renderers.TreeGrid:
         kernel = self.context.modules[self.config["kernel"]]
 
         filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))

volatility3/framework/plugins/windows/vadyarascan.py

@@ -18,7 +18,7 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
     """Scans all the Virtual Address Descriptor memory maps using yara."""
 
     _required_framework_version = (2, 4, 0)
-    _version = (1, 1, 0)
+    _version = (1, 1, 1)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -33,7 +33,7 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
                 name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
             ),
             requirements.PluginRequirement(
-                name="yarascan", plugin=yarascan.YaraScan, version=(1, 3, 0)
+                name="yarascan", plugin=yarascan.YaraScan, version=(2, 0, 0)
             ),
             requirements.ListRequirement(
                 name="pid",
@@ -68,31 +68,47 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
             layer = self.context.layers[layer_name]
             for start, size in self.get_vad_maps(task):
                 if size > sanity_check:
-                    vollog.warn(
+                    vollog.debug(
                         f"VAD at 0x{start:x} over sanity-check size, not scanning"
                     )
                     continue
 
-                for match in rules.match(data=layer.read(start, size, True)):
-                    if yarascan.YaraScan.yara_returns_instances():
-                        for match_string in match.strings:
-                            for instance in match_string.instances:
+                data = layer.read(start, size, True)
+                if not yarascan.YaraScan._yara_x:
+                    for match in rules.match(data=data):
+                        if yarascan.YaraScan.yara_returns_instances():
+                            for match_string in match.strings:
+                                for instance in match_string.instances:
+                                    yield 0, (
+                                        format_hints.Hex(instance.offset + start),
+                                        task.UniqueProcessId,
+                                        match.rule,
+                                        match_string.identifier,
+                                        instance.matched_data,
+                                    )
+                        else:
+                            for offset, name, value in match.strings:
+                                yield 0, (
+                                    format_hints.Hex(offset + start),
+                                    task.UniqueProcessId,
+                                    match.rule,
+                                    name,
+                                    value,
+                                )
+                else:
+                    for match in rules.scan(data).matching_rules:
+                        for match_string in match.patterns:
+                            for instance in match_string.matches:
                                 yield 0, (
                                     format_hints.Hex(instance.offset + start),
                                     task.UniqueProcessId,
-                                    match.rule,
+                                    f"{match.namespace}.{match.identifier}",
                                     match_string.identifier,
-                                    instance.matched_data,
+                                    data[
+                                        instance.offset : instance.offset
+                                        + instance.length
+                                    ],
                                 )
-                    else:
-                        for offset, name, value in match.strings:
-                            yield 0, (
-                                format_hints.Hex(offset + start),
-                                task.UniqueProcessId,
-                                match.rule,
-                                name,
-                                value,
-                            )
 
     @staticmethod
     def get_vad_maps(

volatility3/framework/plugins/windows/verinfo.py

@@ -46,10 +46,7 @@ class VerInfo(interfaces.plugins.PluginInterface):
                 name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
             ),
             requirements.PluginRequirement(
-                name="modules", plugin=modules.Modules, version=(1, 0, 0)
-            ),
-            requirements.VersionRequirement(
-                name="dlllist", component=dlllist.DllList, version=(2, 0, 0)
+                name="modules", plugin=modules.Modules, version=(2, 0, 0)
             ),
             requirements.BooleanRequirement(
                 name="extensive",

volatility3/framework/plugins/yarascan.py

@@ -13,20 +13,31 @@ from volatility3.framework.renderers import format_hints
 
 vollog = logging.getLogger(__name__)
 
-try:
-    import yara
+USE_YARA_X = False
+
+try:
+    import yara_x
+
+    USE_YARA_X = True
 
-    if tuple([int(x) for x in yara.__version__.split(".")]) < (3, 8):
-        raise ImportError
 except ImportError:
-    vollog.info(
-        "Python Yara (>3.8.0) module not found, plugin (and dependent plugins) not available"
-    )
-    raise
+    try:
+        import yara
+
+        if tuple(int(x) for x in yara.__version__.split(".")) < (3, 8):
+            raise ImportError
+
+        vollog.debug("Using yara-python module")
+
+    except ImportError:
+        vollog.info(
+            "Neither yara-x nor yara-python (>3.8.0) module not found, plugin (and dependent plugins) not available"
+        )
+        raise
 
 
 class YaraScanner(interfaces.layers.ScannerInterface):
-    _version = (2, 0, 0)
+    _version = (2, 1, 0)
 
     # yara.Rules isn't exposed, so we can't type this properly
     def __init__(self, rules) -> None:
@@ -34,37 +45,69 @@ class YaraScanner(interfaces.layers.ScannerInterface):
         if rules is None:
             raise ValueError("No rules provided to YaraScanner")
         self._rules = rules
-        self.st_object = not tuple([int(x) for x in yara.__version__.split(".")]) < (
-            4,
-            3,
+        self.st_object = (
+            None
+            if USE_YARA_X
+            else not tuple(int(x) for x in yara.__version__.split(".")) < (4, 3)
         )
 
     def __call__(
         self, data: bytes, data_offset: int
     ) -> Iterable[Tuple[int, str, str, bytes]]:
-        for match in self._rules.match(data=data):
-            if YaraScan.yara_returns_instances():
-                for match_string in match.strings:
-                    for instance in match_string.instances:
+        if USE_YARA_X:
+            for match in self._rules.scan(data).matching_rules:
+                for match_string in match.patterns:
+                    for instance in match_string.matches:
                         yield (
                             instance.offset + data_offset,
-                            match.rule,
+                            f"{match.namespace}.{match.identifier}",
                             match_string.identifier,
-                            instance.matched_data,
+                            data[instance.offset : instance.offset + instance.length],
                         )
-            else:
-                for offset, name, value in match.strings:
-                    yield (offset + data_offset, match.rule, name, value)
+        else:
+            for match in self._rules.match(data=data):
+                if YaraScan.yara_returns_instances():
+                    for match_string in match.strings:
+                        for instance in match_string.instances:
+                            yield (
+                                instance.offset + data_offset,
+                                match.rule,
+                                match_string.identifier,
+                                instance.matched_data,
+                            )
+                else:
+                    for offset, name, value in match.strings:
+                        yield (offset + data_offset, match.rule, name, value)
+
+    @staticmethod
+    def get_rule(rule):
+        if USE_YARA_X:
+            return yara_x.compile(f"rule r1 {{strings: $a = {rule} condition: $a}}")
+        return yara.compile(
+            sources={"n": f"rule r1 {{strings: $a = {rule} condition: $a}}"}
+        )
+
+    @staticmethod
+    def from_compiled_file(filepath):
+        with resources.ResourceAccessor().open(filepath, "rb") as fp:
+            if USE_YARA_X:
+                return yara_x.Rules.deserialize_from(file=fp)
+            return yara.load(file=fp)
+
+    @staticmethod
+    def from_file(filepath):
+        with resources.ResourceAccessor().open(filepath, "rb") as fp:
+            if USE_YARA_X:
+                return yara_x.compile(fp.read().decode())
+            return yara.compile(file=fp)
 
 
 class YaraScan(plugins.PluginInterface):
     """Scans kernel memory using yara rules (string or file)."""
 
     _required_framework_version = (2, 0, 0)
-    _version = (1, 3, 0)
-
-    # TODO: When the major version is bumped, take the opportunity to rename the yara_rules config to yara_string
-    # or something that makes more sense
+    _version = (2, 0, 0)
+    _yara_x = USE_YARA_X
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -99,10 +142,14 @@ class YaraScan(plugins.PluginInterface):
                 optional=True,
             ),
             requirements.StringRequirement(
-                name="yara_rules", description="Yara rules (as a string)", optional=True
+                name="yara_string",
+                description="Yara rules (as a string)",
+                optional=True,
             ),
             requirements.URIRequirement(
-                name="yara_file", description="Yara rules (as a file)", optional=True
+                name="yara_file",
+                description="Yara rules (as a file)",
+                optional=True,
             ),
             # This additional requirement is to follow suit with upstream, who feel that compiled rules could potentially be used to execute malicious code
             # As such, there's a separate option to run compiled files, as happened with yara-3.9 and later
@@ -121,38 +168,28 @@ class YaraScan(plugins.PluginInterface):
 
     @classmethod
     def yara_returns_instances(cls) -> bool:
-        st_object = not tuple([int(x) for x in yara.__version__.split(".")]) < (
-            4,
-            3,
-        )
-        return st_object
+        return not tuple(int(x) for x in yara.__version__.split(".")) < (4, 3)
 
     @classmethod
     def process_yara_options(cls, config: Dict[str, Any]):
         rules = None
-        if config.get("yara_rules", None) is not None:
-            rule = config["yara_rules"]
+        if config.get("yara_string") is not None:
+            rule = config["yara_string"]
             if rule[0] not in ["{", "/"]:
                 rule = f'"{rule}"'
             if config.get("case", False):
                 rule += " nocase"
             if config.get("wide", False):
                 rule += " wide ascii"
-            rules = yara.compile(
-                sources={"n": f"rule r1 {{strings: $a = {rule} condition: $a}}"}
-            )
-        elif config.get("yara_source", None) is not None:
-            rules = yara.compile(source=config["yara_source"])
-        elif config.get("yara_file", None) is not None:
-            rules = yara.compile(
-                file=resources.ResourceAccessor().open(config["yara_file"], "rb")
-            )
-        elif config.get("yara_compiled_file", None) is not None:
-            rules = yara.load(
-                file=resources.ResourceAccessor().open(
-                    config["yara_compiled_file"], "rb"
-                )
+            rules = YaraScanner.get_rule(rule)
+        elif config.get("yara_file") is not None:
+            vollog.debug(f"Plain file: {config['yara_file']} - yara-x: {USE_YARA_X}")
+            rules = YaraScanner.from_file(config["yara_file"])
+        elif config.get("yara_compiled_file") is not None:
+            vollog.debug(
+                f"Compiled file: {config['yara_compiled_file']} - yara-x: {USE_YARA_X}"
             )
+            rules = YaraScanner.from_compiled_file(config["yara_compiled_file"])
         else:
             vollog.error("No yara rules, nor yara rules file were specified")
         return rules

volatility3/framework/renderers/conversion.py

@@ -11,6 +11,7 @@ from typing import Union
 from volatility3.framework import interfaces, renderers
 
 
+# FIXME: Move wintime_to_datetime() and unixtime_to_datetime() out of renderers, possibly framework.objects.utility
 def wintime_to_datetime(
     wintime: int,
 ) -> Union[interfaces.renderers.BaseAbsentValue, datetime.datetime]:

volatility3/framework/symbols/linux/__init__.py

@@ -1,6 +1,9 @@
 # This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
+import math
+import contextlib
+from abc import ABC, abstractmethod
 from typing import Iterator, List, Tuple, Optional, Union
 
 from volatility3 import framework
@@ -30,9 +33,13 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
         self.set_type_class("kobject", extensions.kobject)
         self.set_type_class("cred", extensions.cred)
         self.set_type_class("inode", extensions.inode)
+        self.set_type_class("idr", extensions.IDR)
+        self.set_type_class("address_space", extensions.address_space)
+        self.set_type_class("page", extensions.page)
         # Might not exist in the current symbols
         self.optional_set_type_class("module", extensions.module)
         self.optional_set_type_class("bpf_prog", extensions.bpf_prog)
+        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)
 
@@ -67,7 +74,7 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
 class LinuxUtilities(interfaces.configuration.VersionableInterface):
     """Class with multiple useful linux functions."""
 
-    _version = (2, 1, 0)
+    _version = (2, 1, 1)
     _required_framework_version = (2, 0, 0)
 
     framework.require_interface_version(*_required_framework_version)
@@ -162,13 +169,30 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
         Returns:
             str: Sock pipe pathname relative to the task's root directory.
         """
+        # FIXME: This function must be moved to the 'dentry' object extension
+        # Also, the scope of this function went beyond the sock pipe path, so we need to rename this.
+        # Once https://github.com/volatilityfoundation/volatility3/pull/1263 is merged, replace the
+        # dentry inode getters
+
+        if not (filp and filp.is_readable()):
+            return f"<invalid file pointer> {filp:x}"
+
         dentry = filp.get_dentry()
+        if not (dentry and dentry.is_readable()):
+            return f"<invalid dentry pointer> {dentry:x}"
 
         kernel_module = cls.get_module_from_volobj_type(context, dentry)
 
         sym_addr = dentry.d_op.d_dname
+        if not (sym_addr and sym_addr.is_readable()):
+            return f"<invalid d_dname pointer> {sym_addr:x}"
+
         symbs = list(kernel_module.get_symbols_by_absolute_location(sym_addr))
 
+        inode = dentry.d_inode
+        if not (inode and inode.is_readable() and inode.is_valid()):
+            return f"<invalid dentry inode> {inode:x}"
+
         if len(symbs) == 1:
             sym = symbs[0].split(constants.BANG)[1]
 
@@ -184,15 +208,41 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
             elif sym == "simple_dname":
                 pre_name = cls._get_path_file(task, filp)
 
+            elif sym == "ns_dname":
+                # From Kernels 3.19
+
+                # In Kernels >= 6.9, see Linux kernel commit 1fa08aece42512be072351f482096d5796edf7ca
+                # ns_common->stashed change from 'atomic64_t' to 'dentry*'
+                try:
+                    ns_common_type = kernel_module.get_type("ns_common")
+                    stashed_template = ns_common_type.child_template("stashed")
+                    stashed_type_full_name = stashed_template.vol.type_name
+                    stashed_type_name = stashed_type_full_name.split(constants.BANG)[1]
+                    if stashed_type_name == "atomic64_t":
+                        # 3.19 <= Kernels < 6.9
+                        fsdata_ptr = dentry.d_fsdata
+                        if not (fsdata_ptr and fsdata_ptr.is_readable()):
+                            raise IndexError
+
+                        ns_ops = fsdata_ptr.dereference().cast("proc_ns_operations")
+                    else:
+                        # Kernels >= 6.9
+                        private_ptr = inode.i_private
+                        if not (private_ptr and private_ptr.is_readable()):
+                            raise IndexError
+
+                        ns_common = private_ptr.dereference().cast("ns_common")
+                        ns_ops = ns_common.ops
+
+                    pre_name = utility.pointer_to_string(ns_ops.name, 255)
+                except IndexError:
+                    pre_name = "<unsupported ns_dname implementation>"
             else:
-                pre_name = f"<unsupported d_op symbol: {sym}>"
-
-            ret = f"{pre_name}:[{dentry.d_inode.i_ino:d}]"
-
+                pre_name = f"<unsupported d_op symbol> {sym}"
         else:
-            ret = f"<invalid d_dname pointer> {sym_addr:x}"
+            pre_name = f"<unknown d_dname pointer> {sym_addr:x}"
 
-        return ret
+        return f"{pre_name}:[{inode.i_ino:d}]"
 
     @classmethod
     def path_for_file(cls, context, task, filp) -> str:
@@ -412,9 +462,7 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
         Returns:
             A kernel object (vmlinux)
         """
-        symbol_table_arr = volobj.vol.type_name.split("!", 1)
-        symbol_table = symbol_table_arr[0] if len(symbol_table_arr) == 2 else None
-
+        symbol_table = volobj.get_symbol_table_name()
         module_names = context.modules.get_modules_by_symbol_tables(symbol_table)
         module_names = list(module_names)
 
@@ -425,3 +473,353 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
         kernel = context.modules[kernel_module_name]
 
         return kernel
+
+
+class IDStorage(ABC):
+    """Abstraction to support both XArray and RadixTree"""
+
+    # Dynamic values, these will be initialized later
+    CHUNK_SHIFT = None
+    CHUNK_SIZE = None
+    CHUNK_MASK = None
+
+    def __init__(
+        self,
+        context: interfaces.context.ContextInterface,
+        kernel_module_name: str,
+    ):
+        self.vmlinux = context.modules[kernel_module_name]
+        self.vmlinux_layer = self.vmlinux.context.layers[self.vmlinux.layer_name]
+
+        self.pointer_size = self.vmlinux.get_type("pointer").size
+        # Dynamically work out the (XA_CHUNK|RADIX_TREE_MAP)_SHIFT values based on
+        # the node.slots[] array size
+        node_type = self.vmlinux.get_type(self.node_type_name)
+        slots_array_size = node_type.child_template("slots").count
+
+        # Calculate the LSB index - 1
+        self.CHUNK_SHIFT = slots_array_size.bit_length() - 1
+        self.CHUNK_SIZE = 1 << self.CHUNK_SHIFT
+        self.CHUNK_MASK = self.CHUNK_SIZE - 1
+
+    @classmethod
+    def choose_id_storage(
+        cls,
+        context: interfaces.context.ContextInterface,
+        kernel_module_name: str,
+    ) -> "IDStorage":
+        """Returns the appropriate ID storage data structure instance for the current kernel implementation.
+        This is used by the IDR and the PageCache to choose between the XArray and RadixTree.
+
+        Args:
+            context: The context to retrieve required elements (layers, symbol tables) from
+            kernel_module_name: The name of the kernel module on which to operate
+
+        Returns:
+            The appropriate ID storage instance for the current kernel
+        """
+        vmlinux = context.modules[kernel_module_name]
+        address_space_type = vmlinux.get_type("address_space")
+        address_space_has_i_pages = address_space_type.has_member("i_pages")
+        i_pages_type_name = (
+            address_space_type.child_template("i_pages").vol.type_name
+            if address_space_has_i_pages
+            else ""
+        )
+        i_pages_is_xarray = i_pages_type_name.endswith(constants.BANG + "xarray")
+        i_pages_is_radix_tree_root = i_pages_type_name.endswith(
+            constants.BANG + "radix_tree_root"
+        ) and vmlinux.get_type("radix_tree_root").has_member("xa_head")
+
+        if i_pages_is_xarray or i_pages_is_radix_tree_root:
+            return XArray(context, kernel_module_name)
+        else:
+            return RadixTree(context, kernel_module_name)
+
+    @property
+    @abstractmethod
+    def node_type_name(self) -> str:
+        """Returns the Tree implementation node type name
+
+        Returns:
+            A string with the node type name
+        """
+        raise NotImplementedError()
+
+    @property
+    def tag_internal_value(self) -> int:
+        """Returns the internal node flag for the tree"""
+        raise NotImplementedError()
+
+    @abstractmethod
+    def node_is_internal(self, nodep) -> bool:
+        """Checks if the node is internal"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def is_node_tagged(self, nodep) -> bool:
+        """Checks if the node pointer is tagged"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def untag_node(self, nodep) -> int:
+        """Untags a node pointer"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_tree_height(self, treep) -> int:
+        """Returns the tree height"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_node_height(self, nodep) -> int:
+        """Returns the node height"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_head_node(self, tree) -> int:
+        """Returns a pointer to the tree's head"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def is_valid_node(self, nodep) -> bool:
+        """Validates a node pointer"""
+        raise NotImplementedError
+
+    def nodep_to_node(self, nodep) -> interfaces.objects.ObjectInterface:
+        """Instanciates a tree node from its pointer
+
+        Args:
+            nodep: Pointer to the XArray/RadixTree node
+
+        Returns:
+            A XArray/RadixTree node instance
+        """
+        node = self.vmlinux.object(self.node_type_name, offset=nodep, absolute=True)
+        return node
+
+    def _slot_to_nodep(self, slot) -> int:
+        if self.node_is_internal(slot):
+            nodep = slot & ~self.tag_internal_value
+        else:
+            nodep = slot
+
+        return nodep
+
+    def _iter_node(self, nodep, height) -> int:
+        node = self.nodep_to_node(nodep)
+        node_slots = node.slots
+        for off in range(self.CHUNK_SIZE):
+            slot = node_slots[off]
+            if slot == 0:
+                continue
+
+            nodep = self._slot_to_nodep(slot)
+
+            if height == 1:
+                if self.is_valid_node(nodep):
+                    yield nodep
+            else:
+                for child_node in self._iter_node(nodep, height - 1):
+                    yield child_node
+
+    def get_entries(self, root: interfaces.objects.ObjectInterface) -> int:
+        """Walks the tree data structure
+
+        Args:
+            root: The tree root object
+
+        Yields:
+            A tree node pointer
+        """
+        height = self.get_tree_height(root.vol.offset)
+
+        nodep = self.get_head_node(root)
+        if not nodep:
+            return
+
+        # Keep the internal flag before untagging it
+        is_internal = self.node_is_internal(nodep)
+        if self.is_node_tagged(nodep):
+            nodep = self.untag_node(nodep)
+
+        if is_internal:
+            height = self.get_node_height(nodep)
+
+        if height == 0:
+            if self.is_valid_node(nodep):
+                yield nodep
+        else:
+            for child_node in self._iter_node(nodep, height):
+                yield child_node
+
+
+class XArray(IDStorage):
+    XARRAY_TAG_MASK = 3
+    XARRAY_TAG_INTERNAL = 2
+
+    def get_tree_height(self, treep) -> int:
+        return 0
+
+    @property
+    def node_type_name(self) -> str:
+        return "xa_node"
+
+    @property
+    def tag_internal_value(self) -> int:
+        return self.XARRAY_TAG_INTERNAL
+
+    def get_node_height(self, nodep) -> int:
+        node = self.nodep_to_node(nodep)
+        return (node.shift / self.CHUNK_SHIFT) + 1
+
+    def get_head_node(self, tree) -> int:
+        return tree.xa_head
+
+    def node_is_internal(self, nodep) -> bool:
+        return (nodep & self.XARRAY_TAG_MASK) == self.XARRAY_TAG_INTERNAL
+
+    def is_node_tagged(self, nodep) -> bool:
+        return (nodep & self.XARRAY_TAG_MASK) != 0
+
+    def untag_node(self, nodep) -> int:
+        return nodep & (~self.XARRAY_TAG_MASK)
+
+    def is_valid_node(self, nodep) -> bool:
+        # It should have the tag mask clear
+        return not self.is_node_tagged(nodep)
+
+
+class RadixTree(IDStorage):
+    RADIX_TREE_INTERNAL_NODE = 1
+    RADIX_TREE_EXCEPTIONAL_ENTRY = 2
+    RADIX_TREE_ENTRY_MASK = 3
+
+    # Dynamic values. These will be initialized later
+    RADIX_TREE_INDEX_BITS = None
+    RADIX_TREE_MAX_PATH = None
+    RADIX_TREE_HEIGHT_SHIFT = None
+    RADIX_TREE_HEIGHT_MASK = None
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+
+        char_bits = 8
+        self.RADIX_TREE_INDEX_BITS = char_bits * self.pointer_size
+        self.RADIX_TREE_MAX_PATH = int(
+            math.ceil(self.RADIX_TREE_INDEX_BITS / float(self.CHUNK_SHIFT))
+        )
+        self.RADIX_TREE_HEIGHT_SHIFT = self.RADIX_TREE_MAX_PATH + 1
+        self.RADIX_TREE_HEIGHT_MASK = (1 << self.RADIX_TREE_HEIGHT_SHIFT) - 1
+
+        if not self.vmlinux.has_type("radix_tree_root"):
+            # In kernels 4.20, RADIX_TREE_INTERNAL_NODE flag took RADIX_TREE_EXCEPTIONAL_ENTRY's
+            # value. RADIX_TREE_EXCEPTIONAL_ENTRY was removed but that's managed in is_valid_node()
+            # Note that the Radix Tree is still in use for IDR, even after kernels 4.20 when XArray
+            # mostly replace it
+            self.RADIX_TREE_INTERNAL_NODE = 2
+
+    @property
+    def node_type_name(self) -> str:
+        return "radix_tree_node"
+
+    @property
+    def tag_internal_value(self) -> int:
+        return self.RADIX_TREE_INTERNAL_NODE
+
+    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
+                radix_tree_root = self.vmlinux.object(
+                    "radix_tree_root", offset=treep, absolute=True
+                )
+                return radix_tree_root.height
+
+        # kernels >= 4.7.10
+        return 0
+
+    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
+        else:
+            # Kernels < 4.7
+            height_to_maxindex_array = self.vmlinux.object_from_symbol(
+                "height_to_maxindex"
+            )
+            maxindex = height_to_maxindex_array[height]
+            return maxindex
+
+    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
+        elif hasattr(node, "path"):
+            # 3.15 <= Kernels < 4.7
+            return node.path & self.RADIX_TREE_HEIGHT_MASK
+        elif hasattr(node, "height"):
+            # Kernels < 3.15
+            return node.height
+        else:
+            raise exceptions.VolatilityException("Cannot find radix-tree node height")
+
+    def get_head_node(self, tree) -> int:
+        return tree.rnode
+
+    def node_is_internal(self, nodep) -> bool:
+        return (nodep & self.RADIX_TREE_INTERNAL_NODE) != 0
+
+    def is_node_tagged(self, nodep) -> bool:
+        return self.node_is_internal(nodep)
+
+    def untag_node(self, nodep) -> int:
+        return nodep & (~self.RADIX_TREE_ENTRY_MASK)
+
+    def is_valid_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 True
+
+
+class PageCache(object):
+    """Linux Page Cache abstraction"""
+
+    def __init__(
+        self,
+        context: interfaces.context.ContextInterface,
+        kernel_module_name: str,
+        page_cache: interfaces.objects.ObjectInterface,
+    ):
+        """
+        Args:
+            context: interfaces.context.ContextInterface,
+            kernel_module_name: The name of the kernel module on which to operate
+            page_cache: Page cache address space
+        """
+        self.vmlinux = context.modules[kernel_module_name]
+
+        self._page_cache = page_cache
+        self._idstorage = IDStorage.choose_id_storage(context, kernel_module_name)
+
+    def get_cached_pages(self) -> interfaces.objects.ObjectInterface:
+        """Returns all page cache contents
+
+        Yields:
+            Page objects
+        """
+
+        for page_addr in self._idstorage.get_entries(self._page_cache.i_pages):
+            if not page_addr:
+                continue
+
+            page = self.vmlinux.object("page", offset=page_addr, absolute=True)
+            if page:
+                yield page

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

@@ -4,13 +4,16 @@
 
 import collections.abc
 import logging
+import functools
+import binascii
 import stat
 from datetime import datetime
 import socket as socket_module
-from typing import Generator, Iterable, Iterator, Optional, Tuple, List, Union
+from typing import Generator, Iterable, Iterator, Optional, Tuple, List, Union, Dict
 
 from volatility3.framework import constants, exceptions, objects, interfaces, symbols
 from volatility3.framework.renderers import conversion
+from volatility3.framework.configuration import requirements
 from volatility3.framework.constants.linux import SOCK_TYPES, SOCK_FAMILY
 from volatility3.framework.constants.linux import IP_PROTOCOLS, IPV6_PROTOCOLS
 from volatility3.framework.constants.linux import TCP_STATES, NETLINK_PROTOCOLS
@@ -47,7 +50,7 @@ class module(generic.GenericIntelProcess):
                 ).choices
             except exceptions.SymbolError:
                 vollog.debug(
-                    f"Unable to find mod_mem_type enum. This message can be ignored for kernels < 6.4"
+                    "Unable to find mod_mem_type enum. This message can be ignored for kernels < 6.4"
                 )
                 # set to empty dict to show that the enum was not found, and so shouldn't be searched for again
                 self._mod_mem_type = {}
@@ -820,6 +823,26 @@ class dentry(objects.StructType):
             current_dentry = current_dentry.d_parent
         return None
 
+    def get_subdirs(self) -> interfaces.objects.ObjectInterface:
+        """Walks dentry subdirs
+
+        Yields:
+            A dentry object
+        """
+        if self.has_member("d_sib") and self.has_member("d_children"):
+            # kernels >= 6.8
+            walk_member = "d_sib"
+            list_head_member = self.d_children.first
+        elif self.has_member("d_child") and self.has_member("d_subdirs"):
+            # 2.5.0 <= kernels < 6.8
+            walk_member = "d_child"
+            list_head_member = self.d_subdirs
+        else:
+            raise exceptions.VolatilityException("Unsupported dentry type")
+
+        dentry_type_name = self.get_symbol_table_name() + constants.BANG + "dentry"
+        yield from list_head_member.to_list(dentry_type_name, walk_member)
+
 
 class struct_file(objects.StructType):
     def get_dentry(self) -> interfaces.objects.ObjectInterface:
@@ -936,7 +959,8 @@ class mount(objects.StructType):
         MNT_RELATIME: "relatime",
     }
 
-    def get_mnt_sb(self):
+    def get_mnt_sb(self) -> int:
+        """Returns a pointer to the super_block"""
         if self.has_member("mnt"):
             return self.mnt.mnt_sb
         elif self.has_member("mnt_sb"):
@@ -1251,6 +1275,7 @@ class vfsmount(objects.StructType):
             return self._get_real_mnt().has_parent()
 
     def get_mnt_sb(self):
+        """Returns a pointer to the super_block"""
         return self.mnt_sb
 
     def get_flags_access(self) -> str:
@@ -1584,20 +1609,59 @@ class xdp_sock(objects.StructType):
 
 
 class bpf_prog(objects.StructType):
-    def get_type(self):
+    def get_type(self) -> Union[str, None]:
+        """Returns a string with the eBPF program type"""
+
         # The program type was in `bpf_prog_aux::prog_type` from 3.18.140 to
         # 4.1.52 before it was moved to `bpf_prog::type`
         if self.has_member("type"):
             # kernel >= 4.1.52
-            return self.type
+            return self.type.description
 
         if self.has_member("aux") and self.aux:
             if self.aux.has_member("prog_type"):
                 # 3.18.140 <= kernel < 4.1.52
-                return self.aux.prog_type
+                return self.aux.prog_type.description
 
         # kernel < 3.18.140
-        raise AttributeError("Unable to find the BPF type")
+        return None
+
+    def get_tag(self) -> Union[str, None]:
+        """Returns a string with the eBPF program tag"""
+        # 'tag' was added in kernels 4.10
+        if not self.has_member("tag"):
+            return None
+
+        vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
+        vmlinux_layer = vmlinux.context.layers[vmlinux.layer_name]
+
+        prog_tag_addr = self.tag.vol.offset
+        prog_tag_size = self.tag.count
+        prog_tag_bytes = vmlinux_layer.read(prog_tag_addr, prog_tag_size)
+
+        prog_tag = binascii.hexlify(prog_tag_bytes).decode()
+        return prog_tag
+
+    def get_name(self) -> Union[str, None]:
+        """Returns a string with the eBPF program name"""
+        if not self.has_member("aux"):
+            # 'prog_aux' was added in kernels 3.18
+            return None
+
+        return self.aux.get_name()
+
+
+class bpf_prog_aux(objects.StructType):
+    def get_name(self) -> Union[str, None]:
+        """Returns a string with the eBPF program name"""
+        if not self.has_member("name"):
+            # 'name' was added in kernels 4.15
+            return None
+
+        if not self.name:
+            return None
+
+        return utility.array_to_string(self.name)
 
 
 class cred(objects.StructType):
@@ -1897,3 +1961,250 @@ class inode(objects.StructType):
             The inode's file mode string
         """
         return stat.filemode(self.i_mode)
+
+    def get_pages(self) -> interfaces.objects.ObjectInterface:
+        """Gets the inode's cached pages
+
+        Yields:
+            The inode's cached pages
+        """
+        if not self.i_size:
+            return
+        elif not (self.i_mapping and self.i_mapping.nrpages > 0):
+            return
+
+        page_cache = linux.PageCache(
+            context=self._context,
+            kernel_module_name="kernel",
+            page_cache=self.i_mapping.dereference(),
+        )
+        yield from page_cache.get_cached_pages()
+
+    def get_contents(self):
+        """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
+        """
+        for page_obj in self.get_pages():
+            page_index = int(page_obj.index)
+            page_content = page_obj.get_content()
+            yield page_index, page_content
+
+
+class address_space(objects.StructType):
+    @property
+    def i_pages(self):
+        """Returns the appropriate member containing the page cache tree"""
+        if self.has_member("i_pages"):
+            # Kernel >= 4.17
+            return self.member("i_pages")
+        elif self.has_member("page_tree"):
+            # Kernel < 4.17
+            return self.member("page_tree")
+
+        raise exceptions.VolatilityException("Unsupported page cache tree")
+
+
+class page(objects.StructType):
+    @property
+    @functools.lru_cache()
+    def pageflags_enum(self) -> Dict:
+        """Returns 'pageflags' enumeration key/values
+
+        Returns:
+            A dictionary with the pageflags enumeration key/values
+        """
+        # FIXME: It would be even better to use @functools.cached_property instead,
+        # however, this requires Python +3.8
+        try:
+            pageflags_enum = self._context.symbol_space.get_enumeration(
+                self.get_symbol_table_name() + constants.BANG + "pageflags"
+            ).choices
+        except exceptions.SymbolError:
+            vollog.debug(
+                "Unable to find pageflags enum. This can happen in kernels < 2.6.26 or wrong ISF"
+            )
+            # set to empty dict to show that the enum was not found, and so shouldn't be searched for again
+            pageflags_enum = {}
+
+        return pageflags_enum
+
+    def get_flags_list(self) -> List[str]:
+        """Returns a list of page flags
+
+        Returns:
+            List of page flags
+        """
+        flags = []
+        for name, value in self.pageflags_enum.items():
+            if self.flags & (1 << value) != 0:
+                flags.append(name)
+
+        return flags
+
+    def to_paddr(self) -> int:
+        """Converts a page's virtual address to its physical address using the current physical memory model.
+
+        Returns:
+            int: page physical address
+        """
+        vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
+        vmlinux_layer = vmlinux.context.layers[vmlinux.layer_name]
+
+        vmemmap_start = None
+        if vmlinux.has_symbol("mem_section"):
+            # SPARSEMEM_VMEMMAP physical memory model: memmap is virtually contiguous
+            if vmlinux.has_symbol("vmemmap_base"):
+                # CONFIG_DYNAMIC_MEMORY_LAYOUT - KASLR kernels >= 4.9
+                vmemmap_start = vmlinux.object_from_symbol("vmemmap_base")
+            else:
+                # !CONFIG_DYNAMIC_MEMORY_LAYOUT
+                if vmlinux_layer._maxvirtaddr < 57:
+                    # 4-Level paging -> VMEMMAP_START = __VMEMMAP_BASE_L4
+                    vmemmap_base_l4 = 0xFFFFEA0000000000
+                    vmemmap_start = vmemmap_base_l4
+                else:
+                    # 5-Level paging -> VMEMMAP_START = __VMEMMAP_BASE_L5
+                    # FIXME: Once 5-level paging is supported, uncomment the following lines and remove the exception
+                    # vmemmap_base_l5 = 0xFFD4000000000000
+                    # vmemmap_start = vmemmap_base_l5
+                    raise exceptions.VolatilityException(
+                        "5-level paging is not yet supported"
+                    )
+
+        elif vmlinux.has_symbol("mem_map"):
+            # FLATMEM physical memory model, typically 32bit
+            vmemmap_start = vmlinux.object_from_symbol("mem_map")
+
+        elif vmlinux.has_symbol("node_data"):
+            raise exceptions.VolatilityException("NUMA systems are not yet supported")
+        else:
+            raise exceptions.VolatilityException("Unsupported Linux memory model")
+
+        if not vmemmap_start:
+            raise exceptions.VolatilityException(
+                "Something went wrong, we shouldn't be here"
+            )
+
+        page_type_size = vmlinux.get_type("page").size
+        pagec = vmlinux_layer.canonicalize(self.vol.offset)
+        pfn = (pagec - vmemmap_start) // page_type_size
+        page_paddr = pfn * vmlinux_layer.page_size
+
+        return page_paddr
+
+    def get_content(self) -> Union[str, None]:
+        """Returns the page content
+
+        Returns:
+            The page content
+        """
+        vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
+        vmlinux_layer = vmlinux.context.layers[vmlinux.layer_name]
+        physical_layer = vmlinux.context.layers["memory_layer"]
+        page_paddr = self.to_paddr()
+        if not page_paddr:
+            return None
+
+        page_data = physical_layer.read(page_paddr, vmlinux_layer.page_size)
+        return page_data
+
+
+class IDR(objects.StructType):
+    IDR_BITS = 8
+    IDR_MASK = (1 << IDR_BITS) - 1
+    INT_SIZE = 4
+    MAX_IDR_SHIFT = INT_SIZE * 8 - 1
+    MAX_IDR_BIT = 1 << MAX_IDR_SHIFT
+
+    def idr_max(self, num_layers: int) -> int:
+        """Returns the maximum ID which can be allocated given idr::layers
+
+        Args:
+            num_layers: Number of layers
+
+        Returns:
+            Maximum ID for a given number of layers
+        """
+        # Kernel < 4.17
+        bits = min([self.INT_SIZE, num_layers * self.IDR_BITS, self.MAX_IDR_SHIFT])
+
+        return (1 << bits) - 1
+
+    def idr_find(self, idr_id: int) -> int:
+        """Finds an ID within the IDR data structure.
+        Based on idr_find_slowpath(), 3.9 <= Kernel < 4.11
+        Args:
+            idr_id: The IDR lookup ID
+
+        Returns:
+            A pointer to the given ID element
+        """
+        vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
+        if not vmlinux.get_type("idr_layer").has_member("layer"):
+            vollog.info(
+                "Unsupported IDR implementation, it should be a very very old kernel, probabably < 2.6"
+            )
+            return None
+
+        if idr_id < 0:
+            return None
+
+        idr_layer = self.top
+        if not idr_layer:
+            return None
+
+        n = (idr_layer.layer + 1) * self.IDR_BITS
+
+        if idr_id > self.idr_max(idr_layer.layer + 1):
+            return None
+
+        assert n != 0
+
+        while n > 0 and idr_layer:
+            n -= self.IDR_BITS
+            assert n == idr_layer.layer * self.IDR_BITS
+            idr_layer = idr_layer.ary[(idr_id >> n) & self.IDR_MASK]
+
+        return idr_layer
+
+    def _old_kernel_get_entries(self) -> int:
+        # Kernels < 4.11
+        cur = self.cur
+        total = next_id = 0
+        while next_id < cur:
+            entry = self.idr_find(next_id)
+            if entry:
+                yield entry
+                total += 1
+
+            next_id += 1
+
+    def _new_kernel_get_entries(self) -> int:
+        # Kernels >= 4.11
+        id_storage = linux.IDStorage.choose_id_storage(
+            self._context, kernel_module_name="kernel"
+        )
+        for page_addr in id_storage.get_entries(root=self.idr_rt):
+            yield page_addr
+
+    def get_entries(self) -> int:
+        """Walks the IDR and yield a pointer associated with each element.
+
+        Args:
+            in_use (int, optional): _description_. Defaults to 0.
+
+        Yields:
+            A pointer associated with each element.
+        """
+        if self.has_member("idr_rt"):
+            # Kernels >= 4.11
+            get_entries_func = self._new_kernel_get_entries
+        else:
+            # Kernels < 4.11
+            get_entries_func = self._old_kernel_get_entries
+
+        for page_addr in get_entries_func():
+            yield page_addr

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

@@ -2,6 +2,8 @@
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
 
+from typing import Optional
+
 from volatility3.framework import objects, constants, exceptions
 
 
@@ -26,7 +28,15 @@ class MFTFileName(objects.StructType):
 class MFTAttribute(objects.StructType):
     """This represents an MFT ATTRIBUTE"""
 
-    def get_resident_filename(self) -> str:
+    def get_resident_filename(self) -> Optional[str]:
+        # 4MB chosen as cutoff instead of 4KB to allow for recovery from format /L created file systems
+        # Length as 512 as its 256*2, which is the maximum size for an entire file path, so this is even generous
+        if (
+            self.Attr_Header.ContentOffset > 0x400000
+            or self.Attr_Header.NameLength > 512
+        ):
+            return None
+
         # To get the resident name, we jump to relative name offset and read name length * 2 bytes of data
         try:
             name = self._context.object(
@@ -41,7 +51,15 @@ class MFTAttribute(objects.StructType):
         except exceptions.InvalidAddressException:
             return None
 
-    def get_resident_filecontent(self) -> bytes:
+    def get_resident_filecontent(self) -> Optional[bytes]:
+        # smear observed in mass testing of samples
+        # 4MB chosen as cutoff instead of 4KB to allow for recovery from format /L created file systems
+        if (
+            self.Attr_Header.ContentOffset > 0x400000
+            or self.Attr_Header.ContentLength > 0x400000
+        ):
+            return None
+
         # To get the resident content, we jump to relative content offset and read name length * 2 bytes of data
         try:
             bytesobj = self._context.object(