Merge pull request #1259 from gcmoreira/linux_pagecache_pointer_verification_improvements

Linux Page Cache smear memory protection improvements

Author: ikelos

volatility3/framework/plugins/linux/mountinfo.py

@@ -37,7 +37,7 @@ class MountInfo(plugins.PluginInterface):
 
     _required_framework_version = (2, 2, 0)
 
-    _version = (1, 2, 0)
+    _version = (1, 2, 1)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -272,11 +272,16 @@ def get_superblocks(
                 continue
 
             sb_ptr = mnt.get_mnt_sb()
-            if not sb_ptr or sb_ptr in seen_sb_ptr:
+            if not (sb_ptr and sb_ptr.is_readable()):
+                continue
+
+            if sb_ptr in seen_sb_ptr:
                 continue
             seen_sb_ptr.add(sb_ptr)
 
-            yield sb_ptr.dereference(), path_root
+            superblock = sb_ptr.dereference()
+
+            yield superblock, path_root
 
     def run(self):
         pids = self.config.get("pids")

volatility3/framework/plugins/linux/pagecache.py

@@ -78,8 +78,8 @@ def to_user(
         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()
+        modification_time_dt = self.inode.get_modification_time()
+        change_time_dt = self.inode.get_change_time()
 
         inode_user = InodeUser(
             superblock_addr=superblock_addr,
@@ -92,8 +92,8 @@ def to_user(
             cached_pages=cached_pages,
             file_mode=file_mode,
             access_time=access_time_dt,
-            modification_time=modification_time_str,
-            change_time=change_time_str,
+            modification_time=modification_time_dt,
+            change_time=change_time_dt,
             path=self.path,
         )
         return inode_user
@@ -104,7 +104,7 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
 
     _required_framework_version = (2, 0, 0)
 
-    _version = (1, 0, 0)
+    _version = (1, 0, 1)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -186,8 +186,12 @@ def _walk_dentry(
 
             seen_dentries.add(dentry_addr)
 
-            inode = dentry.d_inode
-            if not (inode and inode.is_valid()):
+            inode_ptr = dentry.d_inode
+            if not (inode_ptr and inode_ptr.is_readable()):
+                continue
+
+            inode = inode_ptr.dereference()
+            if not inode.is_valid():
                 continue
 
             # This allows us to have consistent paths
@@ -242,8 +246,9 @@ def get_inodes(
 
             # More dentry/inode sanity checks
             root_inode_ptr = root_dentry.d_inode
-            if not root_inode_ptr:
+            if not (root_inode_ptr and root_inode_ptr.is_readable()):
                 continue
+
             root_inode = root_inode_ptr.dereference()
             if not root_inode.is_valid():
                 continue
@@ -269,10 +274,12 @@ def get_inodes(
             ):
                 if not file_dentry:
                     continue
+
                 # Dentry/inode sanity checks
                 file_inode_ptr = file_dentry.d_inode
-                if not file_inode_ptr:
+                if not (file_inode_ptr and file_inode_ptr.is_readable()):
                     continue
+
                 file_inode = file_inode_ptr.dereference()
                 if not file_inode.is_valid():
                     continue
@@ -335,7 +342,7 @@ def generate_timeline(self):
             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
+            yield description, timeliner.TimeLinerType.CHANGED, inode_out.change_time
 
     @staticmethod
     def format_fields_with_headers(headers, generator):
@@ -382,7 +389,7 @@ class InodePages(plugins.PluginInterface):
 
     _required_framework_version = (2, 0, 0)
 
-    _version = (1, 0, 0)
+    _version = (1, 0, 1)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -482,15 +489,15 @@ def _generator(self):
             vollog.error("You must use either --inode or --find")
             return
 
+        if not inode.is_valid():
+            vollog.error("Invalid inode at 0x%x", inode.vol.offset)
+            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()

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

@@ -638,11 +638,23 @@ def get_flags_opts(self) -> Iterable[str]:
         ]
         return sb_opts
 
-    def get_type(self):
-        mnt_sb_type = utility.pointer_to_string(self.s_type.name, count=255)
-        if self.s_subtype:
-            mnt_sb_subtype = utility.pointer_to_string(self.s_subtype, count=255)
+    def get_type(self) -> Optional[str]:
+        """Gets the superblock filesystem type string"""
+
+        s_type_ptr = self.s_type
+        if not (s_type_ptr and s_type_ptr.is_readable()):
+            return None
+
+        s_type_name_ptr = s_type_ptr.name
+        if not (s_type_name_ptr and s_type_name_ptr.is_readable()):
+            return None
+
+        mnt_sb_type = utility.pointer_to_string(s_type_name_ptr, count=255)
+        s_subtype_ptr = self.s_subtype
+        if s_subtype_ptr and s_subtype_ptr.is_readable():
+            mnt_sb_subtype = utility.pointer_to_string(s_subtype_ptr, count=255)
             mnt_sb_type += "." + mnt_sb_subtype
+
         return mnt_sb_type
 
 
@@ -843,25 +855,55 @@ def get_subdirs(self) -> interfaces.objects.ObjectInterface:
         dentry_type_name = self.get_symbol_table_name() + constants.BANG + "dentry"
         yield from list_head_member.to_list(dentry_type_name, walk_member)
 
+    def get_inode(self) -> interfaces.objects.ObjectInterface:
+        """Returns the inode associated with this dentry"""
+
+        inode_ptr = self.d_inode
+        if not (inode_ptr and inode_ptr.is_readable() and inode_ptr.is_valid()):
+            return None
+
+        return inode_ptr.dereference()
+
 
 class struct_file(objects.StructType):
     def get_dentry(self) -> interfaces.objects.ObjectInterface:
-        if self.has_member("f_dentry"):
-            return self.f_dentry
-        elif self.has_member("f_path"):
+        """Returns a pointer to the dentry associated with this file"""
+        if self.has_member("f_path"):
             return self.f_path.dentry
+        elif self.has_member("f_dentry"):
+            return self.f_dentry
         else:
             raise AttributeError("Unable to find file -> dentry")
 
     def get_vfsmnt(self) -> interfaces.objects.ObjectInterface:
         """Returns the fs (vfsmount) where this file is mounted"""
-        if self.has_member("f_vfsmnt"):
-            return self.f_vfsmnt
-        elif self.has_member("f_path"):
+        if self.has_member("f_path"):
             return self.f_path.mnt
+        elif self.has_member("f_vfsmnt"):
+            return self.f_vfsmnt
         else:
             raise AttributeError("Unable to find file -> vfs mount")
 
+    def get_inode(self) -> interfaces.objects.ObjectInterface:
+        """Returns an inode associated with this file"""
+
+        inode_ptr = None
+        if self.has_member("f_inode") and self.f_inode and self.f_inode.is_readable():
+            # Try first the cached value, kernels +3.9
+            inode_ptr = self.f_inode
+
+        if not (inode_ptr and inode_ptr.is_readable() and inode_ptr.is_valid()):
+            dentry_ptr = self.get_dentry()
+            if not (dentry_ptr and dentry_ptr.is_readable()):
+                return None
+
+            inode_ptr = dentry_ptr.d_inode
+
+        if not (inode_ptr and inode_ptr.is_readable() and inode_ptr.is_valid()):
+            return None
+
+        return inode_ptr.dereference()
+
 
 class list_head(objects.StructType, collections.abc.Iterable):
     def to_list(