Merge pull request #1233 from gcmoreira/linux_page_cache

Linux Page Cache and IDR

.github/workflows/test.yaml

@@ -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

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/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 @@ def get_mountinfo(
             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 @@ def _generator(
         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 @@ def _generator(
                 "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")