Merge pull request #1557 from gcmoreira/linux_parity_release_harden_pointer_validation

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

volatility3/framework/plugins/linux/check_syscall.py

@@ -82,7 +82,7 @@ class Check_syscall(plugins.PluginInterface):
 
         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 @@ class Check_syscall(plugins.PluginInterface):
             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

@@ -5,7 +5,7 @@ import re
 import logging
 from abc import ABC, abstractmethod
 from enum import Enum
-from typing import Generator, Iterator, List, Tuple
+from typing import Generator, Iterator, List, Tuple, Union
 
 from volatility3.framework import (
     class_subclasses,
@@ -135,8 +135,14 @@ class ABCKmsg(ABC):
             bool: True if the kernel being analyzed fulfill the class requirements.
         """
 
-    def get_string(self, addr: int, length: int) -> str:
-        txt = self._context.layers[self.layer_name].read(addr, length)  # type: ignore
+    def get_string(self, addr: int, length: int) -> Union[str, None]:
+        layer = self._context.layers[self.layer_name]
+        if not layer.is_valid(addr, length):
+            vollog.warning("Failed to read log record at address 0x%x", addr)
+            return None
+
+        txt = layer.read(addr, length)
+
         return txt.decode(encoding="utf8", errors="replace")
 
     def nsec_to_sec_str(self, nsec: int) -> str:
@@ -263,7 +269,7 @@ class Kmsg_3_5_to_3_11(ABCKmsg):
     def _get_log_struct_name(self):
         return "log"
 
-    def get_text_from_log(self, msg) -> str:
+    def get_text_from_log(self, msg) -> Union[str, None]:
         log_struct_name = self._get_log_struct_name()
         log_struct_size = self.vmlinux.get_type(log_struct_name).size
         msg_offset = msg.vol.offset + log_struct_size
@@ -272,7 +278,8 @@ class Kmsg_3_5_to_3_11(ABCKmsg):
     def get_log_lines(self, msg) -> Generator[str, None, None]:
         if msg.text_len > 0:
             text = self.get_text_from_log(msg)
-            yield from text.splitlines()
+            if text:
+                yield from text.splitlines()
 
     def get_dict_lines(self, msg) -> Generator[str, None, None]:
         if msg.dict_len == 0:
@@ -281,9 +288,13 @@ class Kmsg_3_5_to_3_11(ABCKmsg):
         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()
 
@@ -403,7 +414,7 @@ class Kmsg_5_10_to_(ABCKmsg):
     def symtab_checks(cls, vmlinux) -> bool:
         return vmlinux.has_symbol("prb")
 
-    def get_text_from_data_ring(self, text_data_ring, desc, info) -> str:
+    def get_text_from_data_ring(self, text_data_ring, desc, info) -> Union[str, None]:
         text_data_sz = text_data_ring.size_bits
         text_data_mask = 1 << text_data_sz
 
@@ -431,7 +442,8 @@ class Kmsg_5_10_to_(ABCKmsg):
 
     def get_log_lines(self, text_data_ring, desc, info) -> Generator[str, None, None]:
         text = self.get_text_from_data_ring(text_data_ring, desc, info)
-        yield from text.splitlines()
+        if text:
+            yield from text.splitlines()
 
     def get_dict_lines(self, info) -> Generator[str, None, None]:
         dict_text = utility.array_to_string(info.dev_info.subsystem)

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 @@ class Maps(plugins.PluginInterface):
         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 @@ class Maps(plugins.PluginInterface):
                 ]
 
                 # 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 @@ class Maps(plugins.PluginInterface):
                         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 @@ class Sockstat(plugins.PluginInterface):
             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
             )
-            socket = socket_alloc.socket
-
-            if not (socket and socket.sk):
+            if not socket_alloc:
+                continue
+            socket = socket_alloc.socket
+            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

@@ -2088,8 +2088,11 @@ class bpf_prog(objects.StructType):
 
         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,