Improve DTB tables validation (check user space entries). Add index file support for a process dump.

Author: psviderski

volatility/plugins/linux/auto_dump_map.py

@@ -1,4 +1,5 @@
 import os
+from struct import pack
 from volatility import debug, utils
 from volatility.plugins.linux.auto_dtblist import linux_auto_dtblist
 
@@ -14,10 +15,10 @@ def __init__(self, config, *args, **kwargs):
                                 help = "Starting virtual address to dump")
         self._config.add_option('VA-END', short_option='e', type='int', default = None,
                                 help = "Ending virtual address to dump")
-        self._config.add_option('OUTPUTFILE', short_option='O', default=None,
+        self._config.add_option('DUMP-FILE', short_option='O', default=None,
                                 help="Output file to write a dump of virtual address space to")
         self._config.add_option('DUMP-DIR', short_option='D', default = None,
-                                help = "Directory to write the output files to")
+                                help = "Directory to write the dump files to")
 
     def _parse_address(self, addr):
         if addr[:2].lower() == '0x':  # hexadecimal
@@ -28,6 +29,27 @@ def _parse_address(self, addr):
             radix = 10
         return int(addr, radix)
 
+    def _process_name(self, dtb):
+        """Returns the name of a process associated with the given dtb."""
+        return "DTB_{0:#010x}".format(dtb)
+
+    def _process_dump_filepath(self, dtb):
+        dump_dir = self._config.DUMP_DIR
+        if not dump_dir:
+            debug.error("Dump directory is not specified.")
+        process_name = self._process_name(dtb)
+        dump_filepath = os.path.join(dump_dir, '{0}.bin'.format(process_name))
+        if not os.path.exists(dump_filepath):
+            return dump_filepath
+        # The dump file path is already exist. It means that there are more
+        # than one process named 'process_name'.
+        name_id = 0
+        while True:
+            dump_filepath = os.path.join(dump_dir, '{0}_{1}.bin'.format(process_name, name_id))
+            if not os.path.exists(dump_filepath):
+                return dump_filepath
+            name_id += 1
+
     def calculate(self):
         if self._config.PROC_DTB:
             process_dtblist = []
@@ -61,10 +83,10 @@ def calculate(self):
                     yield process_dtb, vaddr, page
 
     def render_text(self, outfd, data):
-        output_file = self._config.OUTPUTFILE
+        dump_file = self._config.DUMP_FILE
         dump_dir = self._config.DUMP_DIR
-        if not output_file and not dump_dir:
-            debug.error("Please specify an output file (use option --outputfile)"
+        if not dump_file and not dump_dir:
+            debug.error("Please specify an output file (use option --dump-file)"
                         " or a dump directory (use option --dump-dir).")
         if dump_dir and not os.path.isdir(dump_dir):
             debug.error("'{0}' is not a directory.".format(self._config.DUMP_DIR))
@@ -74,13 +96,15 @@ def render_text(self, outfd, data):
         vaddr_start = None
         vaddr_end = None
         cur_process_dtb = None
-        # Dump all processes to a single file (OUTPUTFILE) if DUMP_DIR is not specified.
-        dump_fd = open(output_file, 'wb') if not dump_dir else None
+        # Dump all processes to a single file (DUMP_FILE) if DUMP_DIR is not specified.
+        dump_fd = open(dump_file, 'wb') if not dump_dir else None
+        index_fd = IndexFile('{0}.index'.format(dump_file)) if not dump_dir else None
         for process_dtb, vaddr, page in data:
             if process_dtb != cur_process_dtb:
-                # Print the last range of virtual addresses of the previous process
+                # Write the last range of virtual addresses of the previous process
                 if vaddr_start is not None:
                     self.table_row(outfd, cur_process_dtb, vaddr_start, vaddr_end)
+                    index_fd.write_range(vaddr_start, vaddr_end)
                 vaddr_start = None
                 vaddr_end = None
                 cur_process_dtb = process_dtb
@@ -89,18 +113,46 @@ def render_text(self, outfd, data):
                     # next process.
                     if dump_fd:
                         dump_fd.close()
-                    output_file = os.path.join(dump_dir, "{0:#010x}".format(process_dtb))
-                    dump_fd = open(output_file, 'wb')
+                        index_fd.close()
+                    dump_file = os.path.join(self._process_dump_filepath(process_dtb))
+                    index_file = '{0}.index'.format(dump_file)
+                    dump_fd = open(dump_file, 'wb')
+                    index_fd = IndexFile(index_file)
             if vaddr == vaddr_end:
                 vaddr_end += len(page)
             else:
                 if vaddr_start is not None:
                     self.table_row(outfd, process_dtb, vaddr_start, vaddr_end)
+                    index_fd.write_range(vaddr_start, vaddr_end)
                 vaddr_start = vaddr
                 vaddr_end = vaddr + len(page)
             dump_fd.write(page)
-        # Print the last range of virtual addresses
+        # Write the last range of virtual addresses
         if vaddr_start is not None:
             self.table_row(outfd, process_dtb, vaddr_start, vaddr_end)
+            index_fd.write_range(vaddr_start, vaddr_end)
         if dump_fd:
-            dump_fd.close()
+            dump_fd.close()
+            index_fd.close()
+
+
+class IndexFile(object):
+    """Used to describe virtual address ranges that are stored in a dump file."""
+    def __init__(self, filepath):
+        self.fd = open(filepath, 'wb')
+        self.offset = 0
+
+    def write_range(self, vaddr_start, vaddr_end):
+        """Appends index record.
+
+        A record is a packed structure (12 bytes):
+            0x0: virtual address start
+            0x4: virtual address end
+            0x8: offset of the range within a dump file
+
+        """
+        self.fd.write(pack('III', vaddr_start, vaddr_end, self.offset))
+        self.offset += vaddr_end - vaddr_start
+
+    def close(self):
+        self.fd.close()

volatility/plugins/overlays/linux/linux_auto.py

@@ -6,6 +6,8 @@
 ARM_PGD_SIZE = 0x4000  # 16 KB
 ARM_PGD_ENTRIES = 4096
 ARM_PGD_ENTRY_SIZE = 4
+# Entry offsets
+ARM_PGD_DOMAIN_OFFSET = 5
 
 linux_auto_overlay = {
     'VOLATILITY_MAGIC': [None, {
@@ -47,8 +49,8 @@ def generate_suggestions(self):
 class VolatilityDTBARM(obj.VolatilityMagic):
     """A scanner for DTB values in an ARM image."""
 
-    def _is_valid_pgd(self, pgd_addr):
-        """Checks if PGD table starting at pgd_addr is valid.
+    def _is_valid_pgd_kernel_space(self, pgd_addr):
+        """Checks if kernel space entries in PGD table are valid.
 
         In case of 3G/1G User Space/Kernel Space mapping of virtual memory,
         the last quarter (entries 3072-4095) of a PGD table maintains Kernel's
@@ -69,8 +71,12 @@ def _is_valid_pgd(self, pgd_addr):
         for entry_num in xrange(first_kernel_entry, ARM_PGD_ENTRIES):
             entry_offset = entry_num * ARM_PGD_ENTRY_SIZE
             (pgd_entry, ) = unpack('<I', pgd_page[entry_offset:entry_offset + ARM_PGD_ENTRY_SIZE])
-            if (pgd_entry & 0x7ff) == 0x40e:
+            #if (pgd_entry & 0xfff) == 0b010001001110:  # TODO domain 0
+            if (pgd_entry & 0b110000001111) == 0b010000001110:
+                # bits[1:0] == 0b10: 'pgd_entry' is a section or a supersection descriptor
+                # AP == b01, C == 1, B == 1
                 valid_kernel_entries += 1
+        #debug.debug("Valid kernel entrie: {0}".format(valid_kernel_entries))
         valid_kernel_entries_ratio = 1.0 * valid_kernel_entries / min(as_size, 896)  # FIXME 896?
         if valid_kernel_entries_ratio > VALID_KERNEL_ENTRIES_THRESHOLD:
             debug.debug("Found {0} valid kernel entries in a PGD table at physical address {1:#x}. "
@@ -80,6 +86,33 @@ def _is_valid_pgd(self, pgd_addr):
             return True
         return False
 
+    def _is_valid_pgd_user_space(self, pgd_addr):
+        """Checks if user space entries in PGD table are valid.
+
+        Linux kernel doesn't use fine page tables to map memory, so there
+        shouldn't be any fine page table descriptor in a PGD table.
+
+        """
+        pgd_page = self.obj_vm.read(pgd_addr, ARM_PGD_SIZE)
+        #define TASK_SIZE  (UL(CONFIG_PAGE_OFFSET) - UL(0x01000000))
+        user_entries = ARM_PGD_ENTRIES / 4 * 3 - 16  # the last 16M is kernel module space
+        for entry_num in xrange(user_entries):
+            entry_offset = entry_num * ARM_PGD_ENTRY_SIZE
+            (pgd_entry, ) = unpack('<I', pgd_page[entry_offset:entry_offset + ARM_PGD_ENTRY_SIZE])
+            # TODO: domain user (1)
+            if (pgd_entry & 0b11) == 0b11:  # reserved
+                debug.debug("Found incorrect page table descriptor (entry #{0}) in a PGD table "
+                            "at physical address {1:#x}.".format(entry_num, pgd_addr))
+                return False
+        return True
+
+    def _is_valid_pgd(self, pgd_addr):
+        """Checks if PGD table starting at pgd_addr is valid."""
+        if self._is_valid_pgd_kernel_space(pgd_addr) and self._is_valid_pgd_user_space(pgd_addr):
+            debug.debug("Found a valid PGD table at physical address {0:#x}.".format(pgd_addr))
+            return True
+        return False
+
     def generate_suggestions(self):
         """Tries to locate DTBs.
 
@@ -115,4 +148,4 @@ def modification(self, profile):
         profile.object_classes.update({
             'VolatilityDTB': VolatilityDTB,
             'VolatilityLinuxAutoARMValidAS': VolatilityLinuxAutoARMValidAS,
-        })
+        })