[2024] AArch64 support

volatility3/framework/automagic/linux.py

@@ -4,20 +4,21 @@
 
 import logging
 import os
-from typing import Optional, Tuple, Type
+from typing import Optional, Tuple, Type, Union
 
-from volatility3.framework import constants, interfaces
+from volatility3.framework import constants, interfaces, exceptions
 from volatility3.framework.automagic import symbol_cache, symbol_finder
 from volatility3.framework.configuration import requirements
-from volatility3.framework.layers import intel, scanners
+from volatility3.framework.layers import intel, scanners, arm
 from volatility3.framework.symbols import linux
 
 vollog = logging.getLogger(__name__)
 
 
-class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
+class LinuxStacker(interfaces.automagic.StackerLayerInterface):
     stack_order = 35
     exclusion_list = ["mac", "windows"]
+    join = interfaces.configuration.path_join
 
     @classmethod
     def stack(
@@ -39,11 +40,10 @@ def stack(
 
         # Bail out by default unless we can stack properly
         layer = context.layers[layer_name]
-        join = interfaces.configuration.path_join
 
-        # Never stack on top of an intel layer
+        # Never stack on top of a linux layer
         # FIXME: Find a way to improve this check
-        if isinstance(layer, intel.Intel):
+        if isinstance(layer, intel.Intel) or isinstance(layer, arm.AArch64):
             return None
 
         identifiers_path = os.path.join(
@@ -63,66 +63,252 @@ def stack(
         for _, banner in layer.scan(
             context=context, scanner=mss, progress_callback=progress_callback
         ):
-            dtb = None
             vollog.debug(f"Identified banner: {repr(banner)}")
 
             isf_path = linux_banners.get(banner, None)
             if isf_path:
-                table_name = context.symbol_space.free_table_name("LintelStacker")
+                table_name = context.symbol_space.free_table_name("LinuxStacker")
                 table = linux.LinuxKernelIntermedSymbols(
                     context,
                     "temporary." + table_name,
                     name=table_name,
                     isf_url=isf_path,
                 )
                 context.symbol_space.append(table)
-                kaslr_shift, aslr_shift = cls.find_aslr(
-                    context, table_name, layer_name, progress_callback=progress_callback
-                )
+                new_layer_name = context.layers.free_layer_name("LinuxLayer")
+                config_path = cls.join("LinuxHelper", new_layer_name)
+                context.config[cls.join(config_path, "memory_layer")] = layer_name
+                context.config[
+                    cls.join(config_path, LinuxSymbolFinder.banner_config_key)
+                ] = str(banner, "latin-1")
 
-                layer_class: Type = intel.Intel
-                if "init_top_pgt" in table.symbols:
-                    layer_class = intel.Intel32e
-                    dtb_symbol_name = "init_top_pgt"
-                elif "init_level4_pgt" in table.symbols:
-                    layer_class = intel.Intel32e
-                    dtb_symbol_name = "init_level4_pgt"
-                else:
-                    dtb_symbol_name = "swapper_pg_dir"
+                linux_arch_stackers = [cls.intel_stacker, cls.aarch64_stacker]
 
-                dtb = cls.virtual_to_physical_address(
-                    table.get_symbol(dtb_symbol_name).address + kaslr_shift
-                )
+                for linux_arch_stacker in linux_arch_stackers:
+                    try:
+                        layer = linux_arch_stacker(
+                            context=context,
+                            layer_name=layer_name,
+                            table=table,
+                            table_name=table_name,
+                            config_path=config_path,
+                            new_layer_name=new_layer_name,
+                            banner=banner,
+                            progress_callback=progress_callback,
+                        )
+                        if layer:
+                            return layer
+                    except Exception as e:
+                        vollog.log(
+                            constants.LOGLEVEL_VVVV,
+                            f"{linux_arch_stacker.__name__} exception: {e}",
+                        )
 
-                # Build the new layer
-                new_layer_name = context.layers.free_layer_name("IntelLayer")
-                config_path = join("IntelHelper", new_layer_name)
-                context.config[join(config_path, "memory_layer")] = layer_name
-                context.config[join(config_path, "page_map_offset")] = dtb
-                context.config[
-                    join(config_path, LinuxSymbolFinder.banner_config_key)
-                ] = str(banner, "latin-1")
+        vollog.debug("No suitable linux banner could be matched")
+        return None
 
-                layer = layer_class(
-                    context,
-                    config_path=config_path,
-                    name=new_layer_name,
-                    metadata={"os": "Linux"},
+    @classmethod
+    def intel_stacker(
+        cls,
+        context: interfaces.context.ContextInterface,
+        layer_name: str,
+        table: linux.LinuxKernelIntermedSymbols,
+        table_name: str,
+        config_path: str,
+        new_layer_name: str,
+        banner: str,
+        progress_callback: constants.ProgressCallback = None,
+    ) -> Union[intel.Intel, intel.Intel32e, None]:
+        layer_class: Type = intel.Intel
+        if "init_top_pgt" in table.symbols:
+            layer_class = intel.Intel32e
+            dtb_symbol_name = "init_top_pgt"
+        elif "init_level4_pgt" in table.symbols:
+            layer_class = intel.Intel32e
+            dtb_symbol_name = "init_level4_pgt"
+        else:
+            dtb_symbol_name = "swapper_pg_dir"
+
+        kaslr_shift, aslr_shift = cls.find_aslr(
+            context,
+            table_name,
+            layer_name,
+            layer_class,
+            progress_callback=progress_callback,
+        )
+
+        dtb = cls.virtual_to_physical_address(
+            table.get_symbol(dtb_symbol_name).address + kaslr_shift
+        )
+
+        # Build the new layer
+        context.config[cls.join(config_path, "page_map_offset")] = dtb
+
+        layer = layer_class(
+            context,
+            config_path=config_path,
+            name=new_layer_name,
+            metadata={"os": "Linux"},
+        )
+        layer.config["kernel_virtual_offset"] = aslr_shift
+        linux_banner_address = table.get_symbol("linux_banner").address + aslr_shift
+        test_banner_equality = cls.verify_translation_by_banner(
+            context=context,
+            layer=layer,
+            layer_name=layer_name,
+            linux_banner_address=linux_banner_address,
+            target_banner=banner,
+        )
+
+        if layer and dtb and test_banner_equality:
+            vollog.debug(f"DTB was found at: 0x{dtb:0x}")
+            vollog.debug("Intel image found")
+            return layer
+
+        return None
+
+    @classmethod
+    def aarch64_stacker(
+        cls,
+        context: interfaces.context.ContextInterface,
+        layer_name: str,
+        table: linux.LinuxKernelIntermedSymbols,
+        table_name: str,
+        config_path: str,
+        new_layer_name: str,
+        banner: bytes,
+        progress_callback: constants.ProgressCallback = None,
+    ) -> Optional[arm.AArch64]:
+        layer_class = arm.AArch64
+        kaslr_shift, aslr_shift = cls.find_aslr(
+            context,
+            table_name,
+            layer_name,
+            layer_class,
+            progress_callback=progress_callback,
+        )
+        dtb = table.get_symbol("swapper_pg_dir").address + kaslr_shift
+        context.config[cls.join(config_path, "page_map_offset")] = dtb
+        context.config[cls.join(config_path, "page_map_offset_kernel")] = dtb
+
+        # CREDIT : https://github.com/crash-utility/crash/blob/28891d1127542dbb2d5ba16c575e14e741ed73ef/arm64.c#L941
+        kernel_flags = 0
+        if "_kernel_flags_le" in table.symbols:
+            kernel_flags = table.get_symbol("_kernel_flags_le").address
+        if "_kernel_flags_le_hi32" in table.symbols:
+            kernel_flags |= table.get_symbol("_kernel_flags_le_hi32").address << 32
+        if "_kernel_flags_le_lo32" in table.symbols:
+            kernel_flags |= table.get_symbol("_kernel_flags_le_lo32").address
+
+        # https://www.kernel.org/doc/Documentation/arm64/booting.txt
+        page_size_kernel_space_bit = (kernel_flags >> 1) & 3
+        linux_banner_address = table.get_symbol("linux_banner").address + aslr_shift
+
+        # v6.7/source/arch/arm64/include/asm/memory.h#L186 - v5.7/source/arch/arm64/include/asm/memory.h#L160
+        if "vabits_actual" in table.symbols:
+            vabits_actual_phys_addr = (
+                table.get_symbol("vabits_actual").address + kaslr_shift
+            )
+            va_bits = int.from_bytes(
+                context.layers[layer_name].read(vabits_actual_phys_addr, 8),
+                "little",
+            )
+        else:
+            # TODO: If KASAN space is large enough, it *might* push kernel addresses higher and generate inaccurate results ?
+            # We count the number of high bits equal to 1, which gives us the kernel space address mask and ultimately TCR_EL1.T1SZ.
+            va_bits = (linux_banner_address ^ (2**64 - 1)).bit_length() + 1
+
+        tcr_el1_t1sz = 64 - va_bits
+        context.config[cls.join(config_path, "tcr_el1_t1sz")] = tcr_el1_t1sz
+        context.config[cls.join(config_path, "tcr_el1_t0sz")] = tcr_el1_t1sz
+
+        # If "_kernel_flags_le*" aren't in the symbols, we can still do a quick bruteforce on [4,16,64] page sizes
+        # False positives cannot happen, as translation indexes will be off on a wrong page size
+        page_size_kernel_space_candidates = (
+            [4**page_size_kernel_space_bit]
+            if 1 <= page_size_kernel_space_bit <= 3
+            else [4, 16, 64]
+        )
+
+        for i, page_size_kernel_space in enumerate(page_size_kernel_space_candidates):
+            # Kernel space page size is considered equal to the user space page size
+            # T1SZ is considered equal to T0SZ
+            context.config[
+                cls.join(config_path, "page_size_kernel_space")
+            ] = page_size_kernel_space
+            context.config[
+                cls.join(config_path, "page_size_user_space")
+            ] = page_size_kernel_space
+
+            # Build layer
+            layer = layer_class(
+                context,
+                config_path=config_path,
+                name=new_layer_name,
+                metadata={"os": "Linux"},
+            )
+            layer.config["kernel_virtual_offset"] = aslr_shift
+
+            try:
+                test_banner_equality = cls.verify_translation_by_banner(
+                    context=context,
+                    layer=layer,
+                    layer_name=layer_name,
+                    linux_banner_address=linux_banner_address,
+                    target_banner=banner,
                 )
-                layer.config["kernel_virtual_offset"] = aslr_shift
+            except Exception as e:
+                # Only raise the banner translation error if there are no more candidates
+                if i < len(page_size_kernel_space_candidates) - 1:
+                    continue
+                else:
+                    raise e
 
-            if layer and dtb:
-                vollog.debug(f"DTB was found at: 0x{dtb:0x}")
+            if layer and dtb and test_banner_equality:
+                vollog.debug(f"Kernel DTB was found at: 0x{dtb:0x}")
+                vollog.debug("AArch64 image found")
                 return layer
-        vollog.debug("No suitable linux banner could be matched")
+
         return None
 
+    @classmethod
+    def verify_translation_by_banner(
+        cls,
+        context: interfaces.context.ContextInterface,
+        layer,
+        layer_name: str,
+        linux_banner_address: int,
+        target_banner: bytes,
+    ) -> bool:
+        """Determine if a stacked layer is correct or a false positive, by calling the underlying
+        _translate method against the linux_banner symbol virtual address. Then, compare it with
+        the detected banner to verify the correct translation.
+        This will directly raise an exception, as any failed attempt indicates a wrong layer selection.
+        """
+
+        test_banner_equality = True
+        try:
+            banner_phys_address = layer._translate(linux_banner_address)[0]
+            banner_value = context.layers[layer_name].read(
+                banner_phys_address, len(target_banner)
+            )
+        except exceptions.InvalidAddressException:
+            raise Exception('Cannot translate "linux_banner" virtual address')
+
+        if not banner_value == target_banner:
+            raise Exception(
+                f'Translated "linux_banner" virtual address mismatches detected banner : \n{banner_value}\n!=\n{target_banner}'
+            )
+
+        return test_banner_equality
+
     @classmethod
     def find_aslr(
         cls,
         context: interfaces.context.ContextInterface,
         symbol_table: str,
         layer_name: str,
+        layer_class,
         progress_callback: constants.ProgressCallback = None,
     ) -> Tuple[int, int]:
         """Determines the offset of the actual DTB in physical space and its
@@ -162,9 +348,12 @@ def find_aslr(
                 init_task.files.cast("long unsigned int")
                 - module.get_symbol("init_files").address
             )
-            kaslr_shift = init_task_address - cls.virtual_to_physical_address(
-                init_task_json_address
-            )
+            if layer_class == arm.AArch64:
+                kaslr_shift = init_task_address - init_task_json_address
+            else:
+                kaslr_shift = init_task_address - cls.virtual_to_physical_address(
+                    init_task_json_address
+                )
             if address_mask:
                 aslr_shift = aslr_shift & address_mask
 
@@ -196,5 +385,5 @@ class LinuxSymbolFinder(symbol_finder.SymbolFinder):
     banner_config_key = "kernel_banner"
     operating_system = "linux"
     symbol_class = "volatility3.framework.symbols.linux.LinuxKernelIntermedSymbols"
-    find_aslr = lambda cls, *args: LinuxIntelStacker.find_aslr(*args)[1]
+    find_aslr = lambda cls, *args: LinuxStacker.find_aslr(*args)[1]
     exclusion_list = ["mac", "windows"]