Linux: pagecache: Harden Page Cache API to consistently yield valid entries

volatility3/framework/plugins/linux/pagecache.py

@@ -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]:
@@ -253,6 +253,10 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
             if not root_inode.is_valid():
                 continue
 
+            if not (root_inode.i_mapping and root_inode.i_mapping.is_readable()):
+                # Retrieving data from the page cache requires a valid address space
+                continue
+
             # Inode already processed?
             if root_inode_ptr in seen_inodes:
                 continue
@@ -284,6 +288,10 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
                 if not file_inode.is_valid():
                     continue
 
+                if not (file_inode.i_mapping and file_inode.i_mapping.is_readable()):
+                    # Retrieving data from the page cache requires a valid address space
+                    continue
+
                 # Inode already processed?
                 if file_inode_ptr in seen_inodes:
                     continue
@@ -316,10 +324,12 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
             if self.config["find"]:
                 if inode_in.path == self.config["find"]:
                     inode_out = inode_in.to_user(vmlinux_layer)
+
                     yield (0, astuple(inode_out))
                     break  # Only the first match
             else:
                 inode_out = inode_in.to_user(vmlinux_layer)
+
                 yield (0, astuple(inode_out))
 
     def generate_timeline(self):
@@ -389,7 +399,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]:

volatility3/framework/symbols/linux/__init__.py

@@ -838,11 +838,12 @@ class PageCache:
         Yields:
             Page objects
         """
-
+        layer = self.vmlinux.context.layers[self.vmlinux.layer_name]
         for page_addr in self._idstorage.get_entries(self._page_cache.i_pages):
             if not page_addr:
                 continue
 
-            page = self.vmlinux.object("page", offset=page_addr, absolute=True)
-            if page:
-                yield page
+            if not layer.is_valid(page_addr):
+                continue
+
+            yield self.vmlinux.object("page", offset=page_addr, absolute=True)

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

@@ -2489,7 +2489,12 @@ class inode(objects.StructType):
         """
         if not self.i_size:
             return
-        elif not (self.i_mapping and self.i_mapping.nrpages > 0):
+
+        if not (
+            self.i_mapping
+            and self.i_mapping.is_readable()
+            and self.i_mapping.nrpages > 0
+        ):
             return
 
         page_cache = linux.PageCache(
@@ -2497,19 +2502,21 @@ class inode(objects.StructType):
             kernel_module_name="kernel",
             page_cache=self.i_mapping.dereference(),
         )
+
         yield from page_cache.get_cached_pages()
 
-    def get_contents(self):
+    def get_contents(self) -> Iterable[Tuple[int, bytes]]:
         """Get the inode cached pages from the page cache
 
         Yields:
             page_index (int): The page index in the Tree. File offset is page_index * PAGE_SIZE.
-            page_content (str): The page content
+            page_content (bytes): The page content
         """
         for page_obj in self.get_pages():
             page_index = int(page_obj.index)
             page_content = page_obj.get_content()
-            yield page_index, page_content
+            if page_content:
+                yield page_index, page_content
 
 
 class address_space(objects.StructType):
@@ -2625,7 +2632,7 @@ class page(objects.StructType):
 
         return page_paddr
 
-    def get_content(self) -> Union[str, None]:
+    def get_content(self) -> Union[bytes, None]:
         """Returns the page content
 
         Returns:
@@ -2641,8 +2648,13 @@ class page(objects.StructType):
         if not page_paddr:
             return None
 
-        page_data = physical_layer.read(page_paddr, vmlinux_layer.page_size)
-        return page_data
+        if not physical_layer.is_valid(page_paddr, length=vmlinux_layer.page_size):
+            vollog.debug(
+                "Unable to read page 0x%x content at 0x%x", self.vol.offset, page_paddr
+            )
+            return None
+
+        return physical_layer.read(page_paddr, vmlinux_layer.page_size)
 
     def get_flags_list(self) -> List[str]:
         """Returns a list of page flags
@@ -2755,17 +2767,17 @@ class IDR(objects.StructType):
 
 
 class rb_root(objects.StructType):
-    def _walk_nodes(self, root_node) -> Iterator[int]:
+    def _walk_nodes(self, root_node: int) -> Iterator[int]:
         """Traverses the Red-Black tree from the root node and yields a pointer to each
         node in this tree.
 
         Args:
-            root_node: A Red-Black tree node from which to start descending
+            root_node: A Red-Black tree node pointer from which to start descending
 
         Yields:
             A pointer to every node descending from the specified root node
         """
-        if not root_node:
+        if not (root_node and root_node.is_readable()):
             return
 
         yield root_node