Issue #1761: handle new bin_attribute format for module sections

volatility3/framework/constants/linux/__init__.py

@@ -431,6 +431,17 @@ def flags(self) -> str:
 VMCOREINFO_MAGIC_ALIGNED = VMCOREINFO_MAGIC + b"\x00"
 OSRELEASE_TAG = b"OSRELEASE="
 
+ATTRIBUTE_NAME_MAX_SIZE = 255
+"""
+In 5.9-rc1+, the Linux kernel limits the READ size of a section bin_attribute name to MODULE_SECT_READ_SIZE:
+
+- https://elixir.bootlin.com/linux/v6.15-rc4/source/kernel/module/sysfs.c#L106
+- https://github.com/torvalds/linux/commit/11990a5bd7e558e9203c1070fc52fb6f0488e75b
+
+However, the raw section name loaded from the .ko ELF can in theory be thousands of characters,
+and unless we do a NULL terminated search we can't set a perfect value.
+"""
+
 
 @dataclass
 class TaintFlag:

volatility3/framework/objects/utility.py

@@ -3,11 +3,13 @@
 #
 
 import re
-
+import logging
 from typing import Optional, Union
 
 from volatility3.framework import interfaces, objects, constants, exceptions
 
+vollog = logging.getLogger(__name__)
+
 
 def rol(value: int, count: int, max_bits: int = 64) -> int:
     """A rotate-left instruction in Python"""
@@ -250,3 +252,50 @@ def array_of_pointers(
     ).clone()
     subtype_pointer.update_vol(subtype=subtype)
     return array.cast("array", count=count, subtype=subtype_pointer)
+
+
+def dynamically_sized_array_of_pointers(
+    context: interfaces.context.ContextInterface,
+    array: interfaces.objects.ObjectInterface,
+    iterator_guard_value: int,
+    subtype: Union[str, interfaces.objects.Template],
+    stop_value: int = 0,
+    stop_on_invalid_pointers: bool = True,
+) -> interfaces.objects.ObjectInterface:
+    """Iterates over a dynamically sized array of pointers (e.g. NULL-terminated).
+    Array iteration should always be performed with an arbitrary guard value as maximum size,
+    to prevent running forever in case something unexpected happens.
+
+        Args:
+            context: The context on which to operate.
+            array: The object to cast to an array.
+            iterator_guard_value: Stop iterating when the iterator index is greater than this value. This is an extra-safety against smearing.
+            subtype: The subtype of the array's pointers.
+            stop_value: Stop value used to determine when to terminate iteration once it is encountered. Defaults to 0 (NULL-terminated arrays).
+            stop_on_invalid_pointers: Determines whether to stop iterating or not when an invalid pointer is encountered. This can be useful for arrays
+    that are known to have smeared entries before the end.
+
+        Returns:
+            An array of pointer objects
+    """
+    new_count = 0
+    for entry in array_of_pointers(
+        array=array, count=iterator_guard_value, subtype=subtype, context=context
+    ):
+        # "entry" is naturally represented by the address that the pointer refers to
+        if (entry == stop_value) or (
+            not entry.is_readable() and stop_on_invalid_pointers
+        ):
+            break
+        new_count += 1
+    else:
+        vollog.log(
+            constants.LOGLEVEL_V,
+            f"""Iterator guard value {iterator_guard_value} reached while iterating over array at offset {array.vol.offset:#x}.\
+ This means that there is a bug (e.g. smearing) with this array, or that it may contain valid entries past the iterator guard value.""",
+        )
+
+    # Leverage the "Array" object instead of returning a Python list
+    return array_of_pointers(
+        array=array, count=new_count, subtype=subtype, context=context
+    )

volatility3/framework/symbols/linux/__init__.py

@@ -52,7 +52,6 @@ def __init__(self, *args, **kwargs) -> None:
         self.set_type_class("idr", extensions.IDR)
         self.set_type_class("address_space", extensions.address_space)
         self.set_type_class("page", extensions.page)
-        self.set_type_class("module_sect_attr", extensions.module_sect_attr)
 
         # Might not exist in the current symbols
         self.optional_set_type_class("module", extensions.module)
@@ -61,6 +60,8 @@ def __init__(self, *args, **kwargs) -> None:
         self.optional_set_type_class("kernel_cap_struct", extensions.kernel_cap_struct)
         self.optional_set_type_class("kernel_cap_t", extensions.kernel_cap_t)
         self.optional_set_type_class("scatterlist", extensions.scatterlist)
+        self.optional_set_type_class("module_sect_attr", extensions.module_sect_attr)
+        self.optional_set_type_class("bin_attribute", extensions.bin_attribute)
 
         # kernels >= 4.18
         self.optional_set_type_class("timespec64", extensions.timespec64)

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

@@ -180,41 +180,64 @@ def get_name(self) -> Optional[str]:
             return None
 
     def _get_sect_count(self, grp: interfaces.objects.ObjectInterface) -> int:
-        """Try to determine the number of valid sections"""
-        symbol_table_name = self.get_symbol_table_name()
-        arr = self._context.object(
-            symbol_table_name + constants.BANG + "array",
-            layer_name=self.vol.layer_name,
-            offset=grp.attrs,
-            subtype=self._context.symbol_space.get_type(
-                symbol_table_name + constants.BANG + "pointer"
-            ),
-            count=25,
-        )
+        """Try to determine the number of valid sections. Support for kernels > 6.14-rc1.
+
+        Resources:
+            - https://github.com/torvalds/linux/commit/d8959b947a8dfab1047c6fd5e982808f65717bfe
+            - https://github.com/torvalds/linux/commit/e0349c46cb4fbbb507fa34476bd70f9c82bad359
+        """
+
+        if grp.has_member("bin_attrs"):
+            arr_offset_ptr = grp.bin_attrs
+            arr_subtype = "bin_attribute"
+        else:
+            arr_offset_ptr = grp.attrs
+            arr_subtype = "attribute"
 
-        idx = 0
-        while arr[idx] and arr[idx].is_readable():
-            idx = idx + 1
-        return idx
+        if not arr_offset_ptr.is_readable():
+            vollog.log(
+                constants.LOGLEVEL_V,
+                f"Cannot dereference the pointer to the NULL-terminated list of binary attributes for module at offset {self.vol.offset:#x}",
+            )
+            return 0
+
+        # We chose 100 as an arbitrary guard value to prevent
+        # looping forever in extreme cases, and because 100 is not expected
+        # to be a valid number of sections. If that still happens,
+        # Vol3 module processing will indicate that it is missing information
+        # with the following message:
+        # "Unable to reconstruct the ELF for module struct at"
+        # See PR #1773 for more information.
+        bin_attrs_list = utility.dynamically_sized_array_of_pointers(
+            context=self._context,
+            array=arr_offset_ptr.dereference(),
+            iterator_guard_value=100,
+            subtype=self.get_symbol_table_name() + constants.BANG + arr_subtype,
+        )
+        return len(bin_attrs_list)
 
     @functools.cached_property
     def number_of_sections(self) -> int:
+        # Dropped in 6.14-rc1: d8959b947a8dfab1047c6fd5e982808f65717bfe
         if self.sect_attrs.has_member("nsections"):
             return self.sect_attrs.nsections
 
         return self._get_sect_count(self.sect_attrs.grp)
 
     def get_sections(self) -> Iterable[interfaces.objects.ObjectInterface]:
         """Get a list of section attributes for the given module."""
+        if self.number_of_sections == 0:
+            vollog.debug(
+                f"Invalid number of sections ({self.number_of_sections}) for module at offset {self.vol.offset:#x}"
+            )
+            return []
 
         symbol_table_name = self.get_symbol_table_name()
         arr = self._context.object(
             symbol_table_name + constants.BANG + "array",
             layer_name=self.vol.layer_name,
             offset=self.sect_attrs.attrs.vol.offset,
-            subtype=self._context.symbol_space.get_type(
-                symbol_table_name + constants.BANG + "module_sect_attr"
-            ),
+            subtype=self.sect_attrs.attrs.vol.subtype,
             count=self.number_of_sections,
         )
 
@@ -3133,22 +3156,28 @@ def get_name(self) -> Optional[str]:
         """
         if hasattr(self, "battr"):
             try:
-                return utility.pointer_to_string(self.battr.attr.name, count=32)
+                return utility.pointer_to_string(
+                    self.battr.attr.name, count=linux_constants.ATTRIBUTE_NAME_MAX_SIZE
+                )
             except exceptions.InvalidAddressException:
                 # if battr is present then its name attribute needs to be valid
                 vollog.debug(f"Invalid battr name for section at {self.vol.offset:#x}")
                 return None
 
         elif self.name.vol.type_name == "array":
             try:
-                return utility.array_to_string(self.name, count=32)
+                return utility.array_to_string(
+                    self.name, count=linux_constants.ATTRIBUTE_NAME_MAX_SIZE
+                )
             except exceptions.InvalidAddressException:
                 # specifically do not return here to give `mattr` a chance
                 vollog.debug(f"Invalid direct name for section at {self.vol.offset:#x}")
 
         elif self.name.vol.type_name == "pointer":
             try:
-                return utility.pointer_to_string(self.name, count=32)
+                return utility.pointer_to_string(
+                    self.name, count=linux_constants.ATTRIBUTE_NAME_MAX_SIZE
+                )
             except exceptions.InvalidAddressException:
                 # specifically do not return here to give `mattr` a chance
                 vollog.debug(
@@ -3158,10 +3187,38 @@ def get_name(self) -> Optional[str]:
         # if everything else failed...
         if hasattr(self, "mattr"):
             try:
-                return utility.pointer_to_string(self.mattr.attr.name, count=32)
+                return utility.pointer_to_string(
+                    self.mattr.attr.name, count=linux_constants.ATTRIBUTE_NAME_MAX_SIZE
+                )
             except exceptions.InvalidAddressException:
                 vollog.debug(
                     f"Unresolvable name for for section at {self.vol.offset:#x}"
                 )
 
         return None
+
+
+class bin_attribute(objects.StructType):
+    def get_name(self) -> Optional[str]:
+        """
+        Performs extraction of the bin_attribute name
+        """
+        if hasattr(self, "attr"):
+            try:
+                return utility.pointer_to_string(
+                    self.attr.name, count=linux_constants.ATTRIBUTE_NAME_MAX_SIZE
+                )
+            except exceptions.InvalidAddressException:
+                vollog.debug(
+                    f"Invalid attr name for bin_attribute at {self.vol.offset:#x}"
+                )
+                return None
+
+        return None
+
+    @property
+    def address(self) -> int:
+        """Equivalent to module_sect_attr.address:
+        - https://github.com/torvalds/linux/commit/4b2c11e4aaf7e3d7fd9ce8e5995a32ff5e27d74f
+        """
+        return self.private

volatility3/framework/symbols/linux/utilities/module_extract.py

@@ -38,51 +38,11 @@
 class ModuleExtract(interfaces.configuration.VersionableInterface):
     """Extracts Linux kernel module structures into an analyzable ELF file"""
 
-    _version = (1, 0, 0)
+    _version = (1, 0, 1)
     _required_framework_version = (2, 25, 0)
 
     framework.require_interface_version(*_required_framework_version)
 
-    @classmethod
-    def _get_module_section_count(
-        cls,
-        context: interfaces.context.ContextInterface,
-        vmlinux_name: str,
-        module: extensions.module,
-        grp: interfaces.objects.ObjectInterface,
-    ) -> int:
-        """
-        Used to manually determine the section count for kernels that do not track
-        this count directly within the attribute structures
-        """
-        kernel = context.modules[vmlinux_name]
-
-        count = 0
-
-        try:
-            array = kernel.object(
-                object_type="array",
-                offset=grp.attrs,
-                sub_type=kernel.get_type("pointer"),
-                count=50,
-                absolute=True,
-            )
-
-            # Walk up to 50 sections counting until we reach the end or a page fault
-            for sect in array:
-                if sect.vol.offset == 0:
-                    break
-
-                count += 1
-
-        except exceptions.InvalidAddressException:
-            # Use whatever count we reached before the error
-            vollog.debug(
-                f"Exception hit counting sections for module at {module.vol.offset:#x}"
-            )
-
-        return count
-
     @classmethod
     def _find_section(
         cls, section_lookups: List[Tuple[str, int, int, int]], sym_address: int
@@ -261,54 +221,6 @@ def _fix_sym_table(
 
         return sym_table_data
 
-    @classmethod
-    def _enumerate_original_sections(
-        cls,
-        context: interfaces.context.ContextInterface,
-        vmlinux_name: str,
-        module: extensions.module,
-    ) -> Optional[Dict[int, str]]:
-        """
-        Enumerates the module's sections as maintained by the kernel after load time
-        'Early' sections like .init.text and .init.data are discarded after module
-        initialization, so they are not expected to be in memory during extraction
-        """
-        if hasattr(module.sect_attrs, "nsections"):
-            num_sections = module.sect_attrs.nsections
-        else:
-            num_sections = cls._get_module_section_count(
-                context, vmlinux_name, module.sect_attrs.grp
-            )
-
-        if num_sections > 1024 or num_sections == 0:
-            vollog.debug(
-                f"Invalid number of sections ({num_sections}) for module at offset {module.vol.offset:#x}"
-            )
-            return None
-
-        vmlinux = context.modules[vmlinux_name]
-
-        # This is declared as a zero sized array, so we create ourselves
-        attribute_type = module.sect_attrs.attrs.vol.subtype
-
-        sect_array = vmlinux.object(
-            object_type="array",
-            subtype=attribute_type,
-            offset=module.sect_attrs.attrs.vol.offset,
-            count=num_sections,
-            absolute=True,
-        )
-
-        sections: Dict[int, str] = {}
-
-        # for each section, gather its name and address
-        for index, section in enumerate(sect_array):
-            name = section.get_name()
-
-            sections[section.address] = name
-
-        return sections
-
     @classmethod
     def _parse_sections(
         cls,
@@ -325,10 +237,12 @@ def _parse_sections(
         The data of .strtab is read directly off the module structure and not its section
         as the section from the original module has no meaning after loading as the kernel does not reference it.
         """
-        original_sections = cls._enumerate_original_sections(
-            context, vmlinux_name, module
-        )
-        if original_sections is None:
+        original_sections = {}
+        for index, section in enumerate(module.get_sections()):
+            name = section.get_name()
+            original_sections[section.address] = name
+
+        if not original_sections:
             return None
 
         kernel = context.modules[vmlinux_name]
@@ -702,9 +616,10 @@ def extract_module(
             return None
 
         # Gather sections
-        updated_sections, strtab_index, symtab_index = cls._parse_sections(
-            context, vmlinux_name, module
-        )
+        parse_sections_result = cls._parse_sections(context, vmlinux_name, module)
+        if parse_sections_result is None:
+            return None
+        updated_sections, strtab_index, symtab_index = parse_sections_result
 
         kernel = context.modules[vmlinux_name]