Linux - Parity release - Harden pointer validation in several plugins and APIs

volatility3/framework/plugins/linux/check_syscall.py

@@ -82,7 +82,7 @@ def _get_table_info_other(self, table_addr, ptr_sz, vmlinux):
 
         return table_size
 
-    def _get_table_info_disassembly(self, ptr_sz, vmlinux):
+    def _get_table_info_disassembly(self, ptr_sz, vmlinux) -> int:
         """Find the size of the system call table by disassembling functions
         that immediately reference it in their first instruction This is in the
         form 'cmp reg,NR_syscalls'."""
@@ -107,9 +107,13 @@ def _get_table_info_disassembly(self, ptr_sz, vmlinux):
             return 0
 
         vmlinux = self.context.modules[self.config["kernel"]]
-        data = self.context.layers.read(vmlinux.layer_name, func_addr, 6)
+        vmlinux_layer = self.context.layers[vmlinux.layer_name]
+        try:
+            data = vmlinux_layer.read(func_addr, 6)
+        except exceptions.InvalidAddressException:
+            return 0
 
-        for address, size, mnemonic, op_str in md.disasm_lite(data, func_addr):
+        for _address, _size, mnemonic, op_str in md.disasm_lite(data, func_addr):
             if mnemonic == "CMP":
                 table_size = int(op_str.split(",")[1].strip()) & 0xFFFF
                 break

volatility3/framework/plugins/linux/kmsg.py

@@ -136,7 +136,12 @@ def symtab_checks(cls, vmlinux: interfaces.context.ModuleInterface) -> bool:
         """
 
     def get_string(self, addr: int, length: int) -> str:
-        txt = self._context.layers[self.layer_name].read(addr, length)  # type: ignore
+        layer = self._context.layers[self.layer_name]
+        if not layer.is_valid(addr, length):
+            return ""
+
+        txt = layer.read(addr, length)
+
         return txt.decode(encoding="utf8", errors="replace")
 
     def nsec_to_sec_str(self, nsec: int) -> str:
@@ -281,9 +286,13 @@ def get_dict_lines(self, msg) -> Generator[str, None, None]:
         log_struct_name = self._get_log_struct_name()
         log_struct_size = self.vmlinux.get_type(log_struct_name).size
         dict_offset = msg.vol.offset + log_struct_size + msg.text_len
-        dict_data = self._context.layers[self.layer_name].read(
-            dict_offset, msg.dict_len
-        )
+        layer = self._context.layers[self.layer_name]
+        try:
+            dict_data = layer.read(dict_offset, msg.dict_len)
+        except exceptions.InvalidAddressException:
+            vollog.debug("Unable to read kmsg dict from 0x%x", dict_offset)
+            return None
+
         for chunk in dict_data.split(b"\x00"):
             yield " " + chunk.decode()
 

volatility3/framework/plugins/linux/proc.py

@@ -21,7 +21,7 @@ class Maps(plugins.PluginInterface):
     """Lists all memory maps for all processes."""
 
     _required_framework_version = (2, 0, 0)
-    _version = (1, 0, 2)
+    _version = (1, 0, 3)
 
     MAXSIZE_DEFAULT = 1024 * 1024 * 1024  # 1 Gb
 
@@ -83,18 +83,24 @@ def list_vmas(
         Returns:
             Yields vmas based on the task and filtered based on the filter function
         """
-        if task.mm:
-            for vma in task.mm.get_vma_iter():
-                if filter_func(vma):
-                    yield vma
-                else:
-                    vollog.debug(
-                        f"Excluded vma at offset {vma.vol.offset:#x} for pid {task.pid} due to filter_func"
-                    )
-        else:
+        mm_pointer = task.mm
+        if not mm_pointer:
             vollog.debug(
-                f"Excluded pid {task.pid} as there is no mm member. It is likely a kernel thread."
+                f"Excluded pid {task.pid} as there is no mm member. It is likely a kernel thread"
             )
+            return
+
+        if not mm_pointer.is_readable():
+            vollog.error(f"Task {task.pid} has an invalid mm member")
+            return
+
+        for vma in mm_pointer.get_vma_iter():
+            if filter_func(vma):
+                yield vma
+            else:
+                vollog.debug(
+                    f"Excluded vma at offset {vma.vol.offset:#x} for pid {task.pid} due to filter_func"
+                )
 
     @classmethod
     def vma_dump(
@@ -174,31 +180,32 @@ def vma_filter_function(x: interfaces.objects.ObjectInterface) -> bool:
                 ]
 
                 # if any of the user supplied addresses would fall within this vma return true
-                if addrs_in_vma:
-                    return True
-                else:
-                    return False
+                return bool(addrs_in_vma)
 
             vma_filter_func = vma_filter_function
+
         for task in tasks:
-            if not task.mm:
+            if not (task.mm and task.mm.is_readable()):
                 continue
             name = utility.array_to_string(task.comm)
 
             for vma in self.list_vmas(task, filter_func=vma_filter_func):
                 flags = vma.get_protection()
                 page_offset = vma.get_page_offset()
-                major = 0
-                minor = 0
-                inode = 0
 
-                if vma.vm_file != 0:
+                inode_num = None
+                try:
                     dentry = vma.vm_file.get_dentry()
-                    if dentry != 0:
-                        inode_object = dentry.d_inode
-                        major = inode_object.i_sb.major
-                        minor = inode_object.i_sb.minor
-                        inode = inode_object.i_ino
+                    inode_ptr = dentry.d_inode
+                    inode_num = inode_ptr.i_ino
+                    major = inode_ptr.i_sb.major
+                    minor = inode_ptr.i_sb.minor
+                except exceptions.InvalidAddressException:
+                    if not inode_num:
+                        inode_num = 0
+                    major = 0
+                    minor = 0
+
                 path = vma.get_name(self.context, task)
 
                 file_output = "Disabled"
@@ -238,7 +245,7 @@ def vma_filter_function(x: interfaces.objects.ObjectInterface) -> bool:
                         format_hints.Hex(page_offset),
                         major,
                         minor,
-                        inode,
+                        inode_num,
                         path,
                         file_output,
                     ),

volatility3/framework/plugins/linux/sockstat.py

@@ -438,7 +438,7 @@ class Sockstat(plugins.PluginInterface):
     """Lists all network connections for all processes."""
 
     _required_framework_version = (2, 0, 0)
-    _version = (3, 0, 2)
+    _version = (3, 0, 3)
 
     @classmethod
     def get_requirements(cls):
@@ -514,25 +514,28 @@ def list_sockets(
             fd_num, filp, _full_path = fd_internal.fd_fields
             task = fd_internal.task
 
+            if not (filp.f_op and filp.f_op.is_readable()):
+                continue
+
             if filp.f_op not in (sfop_addr, dfop_addr):
                 continue
 
             dentry = filp.get_dentry()
-            if not dentry:
+            if not (dentry and dentry.is_readable()):
                 continue
 
             d_inode = dentry.d_inode
-            if not d_inode:
+            if not (d_inode and d_inode.is_readable()):
                 continue
 
             socket_alloc = linux.LinuxUtilities.container_of(
                 d_inode, "socket_alloc", "vfs_inode", vmlinux
             )
+            if not socket_alloc:
+                continue
             socket = socket_alloc.socket
-
-            if not (socket and socket.sk):
+            if not (socket.sk and socket.sk.is_readable()):
                 continue
-
             sock = socket.sk.dereference()
 
             sock_type = sock.get_type()

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

@@ -2023,8 +2023,11 @@ def get_tag(self) -> Union[str, None]:
 
         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)
+        if not vmlinux_layer.is_valid(prog_tag_addr, prog_tag_size):
+            vollog.debug("Unable to read bpf tag string from 0x%x", prog_tag_addr)
+            return None
 
+        prog_tag_bytes = vmlinux_layer.read(prog_tag_addr, prog_tag_size)
         prog_tag = binascii.hexlify(prog_tag_bytes).decode()
         return prog_tag
 

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

@@ -437,7 +437,7 @@ class elf_linkmap(objects.StructType):
     def get_name(self):
         try:
             buf = self._context.layers.read(self.vol.layer_name, self.l_name, 256)
-        except exceptions.PagedInvalidAddressException:
+        except exceptions.InvalidAddressException:
             # Protection against memory smear
             vollog.log(
                 constants.LOGLEVEL_VVVV,