Linux - Parity release harden Page Cache

test/test_volatility.py

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

volatility3/framework/exceptions.py

@@ -130,3 +130,7 @@ def __str__(self):
 
 class RenderException(VolatilityException):
     """Thrown if there is an error during rendering"""
+
+
+class LinuxPageCacheException(VolatilityException):
+    """Thrown if there is an error during Linux Page Cache processing"""

volatility3/framework/plugins/linux/pagecache.py

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