refs #197 initial commit of windows.dumpfiles

volatility/framework/plugins/windows/dumpfiles.py

@@ -0,0 +1,233 @@
+# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
+# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
+#
+
+import logging
+import ntpath
+from volatility.framework import interfaces, renderers, exceptions, constants
+from volatility.plugins.windows import handles
+from volatility.plugins.windows import pslist
+from volatility.framework.configuration import requirements
+from volatility.framework.renderers import format_hints
+from volatility.framework.objects import utility
+from typing import List, Tuple
+vollog = logging.getLogger(__name__)
+
+FILE_DEVICE_DISK = 0x7
+FILE_DEVICE_NETWORK_FILE_SYSTEM = 0x14
+EXTENSION_CACHE_MAP = {
+    "dat": "DataSectionObject",
+    "img": "ImageSectionObject",
+    "vacb": "SharedCacheMap",
+}
+
+class DumpFiles(interfaces.plugins.PluginInterface):
+    """Dumps cached file contents from Windows memory samples."""
+
+    _version = (1, 0, 0)
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        # Since we're calling the plugin, make sure we have the plugin's requirements
+        return [
+            requirements.TranslationLayerRequirement(name='primary',
+                                                     description='Memory layer for the kernel',
+                                                     architectures=["Intel32", "Intel64"]),
+            requirements.SymbolTableRequirement(name="nt_symbols", description="Windows kernel symbols"),
+            requirements.IntRequirement(name='pid',
+                                        description="Process ID to include (all other processes are excluded)",
+                                        optional=True),
+            requirements.IntRequirement(name='fileoffset',
+                                        description="Dump a single _FILE_OBJECT at this offset",
+                                        optional=True),
+            requirements.PluginRequirement(name='pslist', plugin=pslist.PsList, version=(1, 0, 0)),
+            requirements.PluginRequirement(name='handles', plugin=handles.Handles, version=(1, 0, 0))
+        ]
+
+    def dump_file_producer(self, file_object: interfaces.objects.ObjectInterface,
+                           memory_object: interfaces.objects.ObjectInterface,
+                           layer: interfaces.layers.DataLayerInterface,
+                           desired_file_name: str) -> str:
+        """Produce a file from the memory object's get_available_pages() interface.
+
+        :param file_object: the parent _FILE_OBJECT
+        :param memory_object: the _CONTROL_AREA or _SHARED_CACHE_MAP
+        :param layer: the memory layer to read from
+        :param desired_file_name: name of the output file
+        :return: result status
+        """
+        filedata = interfaces.plugins.FileInterface(desired_file_name)
+        try:
+            # Description of these variables:
+            #   memoffset: offset in the specified layer where the page begins
+            #   fileoffset: write to this offset in the destination file
+            #   datasize: size of the page
+            for memoffset, fileoffset, datasize in memory_object.get_available_pages():
+                data = layer.read(memoffset, datasize, pad = True)
+                filedata.data.seek(fileoffset)
+                filedata.data.write(data)
+
+            # Avoid writing files to disk if they are going to be empty or all zeros.
+            cached_length = len(filedata.data.getvalue())
+            if cached_length == 0 or filedata.data.getvalue().count(0) == cached_length:
+                result_text = "No data is cached for the file at {0:#x}".format(file_object.vol.offset)
+            else:
+                self.produce_file(filedata)
+                result_text = "Stored {}".format(filedata.preferred_filename)
+        except exceptions.InvalidAddressException:
+            result_text = "Unable to dump file at {0:#x}".format(file_object.vol.offset)
+
+        return result_text
+
+    def process_file_object(self, file_obj: interfaces.objects.ObjectInterface) -> Tuple:
+        """Given a FILE_OBJECT, dump data to separate files for each of the three file caches.
+
+        :param file_object: the FILE_OBJECT
+        """
+
+        # Filtering by these types of devices prevents us from processing other types of devices that
+        # use the "File" object type, such as \Device\Tcp and \Device\NamedPipe.
+        if file_obj.DeviceObject.DeviceType not in [FILE_DEVICE_DISK, FILE_DEVICE_NETWORK_FILE_SYSTEM]:
+            vollog.log(constants.LOGLEVEL_VVV,
+                       "The file object at {0:#x} is not a file on disk".format(file_obj.vol.offset))
+            return
+
+        # Depending on the type of object (DataSection, ImageSection, SharedCacheMap) we may need to
+        # read from the memory layer or the primary layer.
+        memory_layer = self.context.layers["memory_layer"]
+        primary_layer = self.context.layers[self.config["primary"]]
+
+        obj_name = file_obj.file_name_with_device()
+
+        # This stores a list of tuples, describing what to dump and how to dump it.
+        # Ex: (
+        #     memory_object with get_available_pages() API (either CONTROL_AREA or SHARED_CACHE_MAP),
+        #     layer to read from,
+        #     file extension to apply,
+        #     )
+        dump_parameters = []
+
+        # The DataSectionObject and ImageSectionObject caches are handled in basically the same way.
+        # We carve these "pages" from the memory_layer.
+        for member_name, extension in [("DataSectionObject", "dat"), ("ImageSectionObject", "img")]:
+            try:
+                section_obj = getattr(file_obj.SectionObjectPointer, member_name)
+                control_area = section_obj.dereference().cast("_CONTROL_AREA")
+                if control_area.is_valid():
+                    dump_parameters.append((control_area, memory_layer, extension))
+            except exceptions.InvalidAddressException:
+                vollog.log(constants.LOGLEVEL_VVV,
+                           "{0} is unavailable for file {1:#x}".format(member_name, file_obj.vol.offset))
+
+        # The SharedCacheMap is handled differently than the caches above.
+        # We carve these "pages" from the primary_layer.
+        try:
+            scm_pointer = file_obj.SectionObjectPointer.SharedCacheMap
+            shared_cache_map = scm_pointer.dereference().cast("_SHARED_CACHE_MAP")
+            if shared_cache_map.is_valid():
+                dump_parameters.append((shared_cache_map, primary_layer, "vacb"))
+        except exceptions.InvalidAddressException:
+            vollog.log(constants.LOGLEVEL_VVV,
+                       "SharedCacheMap is unavailable for file {0:#x}".format(file_obj.vol.offset))
+
+        for memory_object, layer, extension in dump_parameters:
+            cache_name = EXTENSION_CACHE_MAP[extension]
+            desired_file_name = "file.{0:#x}.{1:#x}.{2}.{3}.{4}".format(file_obj.vol.offset,
+                                                                        memory_object.vol.offset,
+                                                                        cache_name,
+                                                                        ntpath.basename(obj_name),
+                                                                        extension)
+
+            result_text = self.dump_file_producer(file_obj, memory_object, layer, desired_file_name)
+
+            yield (cache_name, format_hints.Hex(file_obj.vol.offset),
+                ntpath.basename(obj_name), # temporary, so its easier to visualize output
+                result_text)
+
+    def _generator(self, procs: List, offsets: List):
+        # The handles plugin doesn't expose any staticmethod/classmethod, and it also requires stashing
+        # private variables, so we need an instance (for now, anyway). We _could_ call Handles._generator()
+        # to do some of the other work that is duplicated here, but then we'd need to parse the TreeGrid
+        # results instead of just dealing with them as direct objects here.
+
+        if procs:
+            # Standard code for invoking the Handles() plugin from another plugin.
+            handles_plugin = handles.Handles(context=self.context, config_path=self._config_path)
+            type_map = handles_plugin.get_type_map(context=self.context,
+                                                   layer_name=self.config["primary"],
+                                                   symbol_table=self.config["nt_symbols"])
+            cookie = handles_plugin.find_cookie(context=self.context,
+                                                layer_name=self.config["primary"],
+                                                symbol_table=self.config["nt_symbols"])
+
+            for proc in procs:
+                try:
+                    object_table = proc.ObjectTable
+                except exceptions.InvalidAddressException:
+                    vollog.log(constants.LOGLEVEL_VVV,
+                               "Cannot access _EPROCESS.ObjectTable at {0:#x}".format(proc.vol.offset))
+                    continue
+
+                for entry in handles_plugin.handles(object_table):
+                    try:
+                        obj_type = entry.get_object_type(type_map, cookie)
+                        if obj_type == "File":
+                            file_obj = entry.Body.cast("_FILE_OBJECT")
+                            for result in self.process_file_object(file_obj):
+                                yield (0, result)
+                    except exceptions.InvalidAddressException:
+                        vollog.log(constants.LOGLEVEL_VVV,
+                                   "Cannot extract file from _OBJECT_HEADER at {0:#x}".format(entry.vol.offset))
+
+                # Pull file objects from the VADs. This will produce DLLs and EXEs that are
+                # mapped into the process as images, but that the process doesn't have an
+                # explicit handle remaining open to those files on disk.
+                for vad in proc.get_vad_root().traverse():
+                    try:
+                        if vad.has_member("ControlArea"):
+                            # Windows xp and 2003
+                            file_obj = vad.ControlArea.FilePointer.dereference()
+                        elif vad.has_member("Subsection"):
+                            # Vista and beyond
+                            file_obj = vad.Subsection.ControlArea.FilePointer.dereference().cast("_FILE_OBJECT")
+                        else:
+                            continue
+
+                        if not file_obj.is_valid():
+                            continue
+
+                        for result in self.process_file_object(file_obj):
+                            yield (0, result)
+                    except exceptions.InvalidAddressException:
+                        vollog.log(constants.LOGLEVEL_VVV,
+                                   "Cannot extract file from VAD at {0:#x}".format(vad.vol.offset))
+
+        elif offsets:
+            # Now process any offsets explicitly requested by the user.
+            for offset in offsets:
+                try:
+                    file_obj = self.context.object(self.config["nt_symbols"] + constants.BANG + "_FILE_OBJECT",
+                                                   layer_name=self.config["primary"],
+                                                   native_layer_name=self.config["primary"],
+                                                   offset=offset)
+                    for result in self.process_file_object(file_obj):
+                        yield (0, result)
+                except exceptions.InvalidAddressException:
+                    vollog.log(constants.LOGLEVEL_VVV,
+                               "Cannot extract file at {0:#x}".format(offset))
+
+    def run(self):
+        if self.config.get("fileoffset", None) is not None:
+            offsets = [self.config["fileoffset"]]
+            procs = []
+        else:
+            filter_func = pslist.PsList.create_pid_filter([self.config.get("pid", None)])
+            offsets = []
+            procs = pslist.PsList.list_processes(self.context,
+                                                 self.config["primary"],
+                                                 self.config["nt_symbols"],
+                                                 filter_func=filter_func)
+
+        return renderers.TreeGrid(
+            [("Cache", str), ("FileObject", format_hints.Hex), ("FileName", str), ("Result", str)],
+            self._generator(procs, offsets))

volatility/framework/symbols/windows/__init__.py

@@ -32,6 +32,9 @@ class WindowsKernelIntermedSymbols(intermed.IntermediateSymbolTable):
         self.set_type_class('_KMUTANT', extensions.KMUTANT)
         self.set_type_class('_DRIVER_OBJECT', extensions.DRIVER_OBJECT)
         self.set_type_class('_OBJECT_SYMBOLIC_LINK', extensions.OBJECT_SYMBOLIC_LINK)
+        self.set_type_class('_CONTROL_AREA', extensions.CONTROL_AREA)
+        self.set_type_class('_SHARED_CACHE_MAP', extensions.SHARED_CACHE_MAP)
+        self.set_type_class('_VACB', extensions.VACB)
         self.set_type_class('_POOL_TRACKER_BIG_PAGES', pool.POOL_TRACKER_BIG_PAGES)
 
         # This doesn't exist in very specific versions of windows

volatility/framework/symbols/windows/extensions/__init__.py

@@ -7,6 +7,7 @@ import datetime
 import functools
 import logging
 import struct
+import math
 from typing import Iterable, Iterator, Optional, Union, Dict, Tuple, List
 
 from volatility.framework import constants, exceptions, interfaces, objects, renderers, symbols
@@ -667,3 +668,320 @@ class LIST_ENTRY(objects.StructType, collections.abc.Iterable):
 
     def __iter__(self) -> Iterator[interfaces.objects.ObjectInterface]:
         return self.to_list(self.vol.parent.vol.type_name, self.vol.member_name)
+
+class CONTROL_AREA(objects.StructType):
+    """A class for _CONTROL_AREA structures"""
+
+    PAGE_SIZE = 0x1000
+    PAGE_MASK = PAGE_SIZE - 1
+
+    def is_valid(self) -> bool:
+        """Determine if the object is valid."""
+        try:
+            # The Segment.ControlArea should point back to this object
+            if self.Segment.ControlArea != self.vol.offset:
+                return False
+
+            # The SizeOfSegment should match the total PTEs multiplied by a default page size
+            if self.Segment.SizeOfSegment != (self.Segment.TotalNumberOfPtes * self.PAGE_SIZE):
+                return False
+
+            # The first SubsectionBase should not be page aligned
+            #subsection = self.get_subsection()
+            #if subsection.SubsectionBase & self.PAGE_MASK == 0:
+            #    return False
+        except exceptions.InvalidAddressException:
+            return False
+
+        # True if everything else passes
+        return True
+
+    def get_subsection(self) -> interfaces.objects.ObjectInterface:
+        """Get the Subsection object, which is found immediately after the _CONTROL_AREA."""
+
+        return self._context.object(self.get_symbol_table_name() + constants.BANG + "_SUBSECTION",
+                             layer_name=self.vol.layer_name,
+                             offset=self.vol.offset + self.vol.size,
+                             native_layer_name=self.vol.native_layer_name)
+
+    def get_pte(self, offset: int) -> interfaces.objects.ObjectInterface:
+        """Get a PTE object at the requested offset"""
+
+        return self._context.object(self.get_symbol_table_name() + constants.BANG + "_MMPTE",
+                             layer_name=self.vol.layer_name,
+                             offset=offset,
+                             native_layer_name=self.vol.native_layer_name)
+
+    def get_available_pages(self) -> Iterable[Tuple[int, int, int]]:
+        """Get the available pages that correspond to a cached file.
+
+        The tuples generated are (physical_offset, file_offset, page_size).
+        """
+        symbol_table_name = self.get_symbol_table_name()
+        mmpte_type = self._context.symbol_space.get_type(symbol_table_name + constants.BANG + "_MMPTE")
+        mmpte_size = mmpte_type.size
+        subsection = self.get_subsection()
+        is_64bit = symbols.symbol_table_is_64bit(self._context, symbol_table_name)
+        is_pae = self._context.layers[self.vol.layer_name].metadata.get("pae", False)
+
+        # This is a null-terminated single-linked list.
+        while subsection != 0:
+            try:
+                if subsection.ControlArea != self.vol.offset:
+                    break
+            except exceptions.InvalidAddressException:
+                break
+
+            # The offset into the file is stored implicitly based on the PTE location within the Subsection.
+            starting_sector = subsection.StartingSector
+            subsection_offset = starting_sector * 0x200
+
+            # Similar to the check in is_valid(), make sure the SubsectionBase is not page aligned.
+            #if subsection.SubsectionBase & self.PAGE_MASK == 0:
+            #    break
+
+            ptecount = 0
+            while ptecount < subsection.PtesInSubsection:
+                pte_offset = subsection.SubsectionBase + (mmpte_size * ptecount)
+                file_offset = subsection_offset + ptecount * 0x1000
+
+                try:
+                    mmpte = self.get_pte(pte_offset)
+                except exceptions.InvalidAddressException:
+                    ptecount += 1
+                    continue
+
+                # First we check if the entry is valid. If so, then we get the physical offset.
+                # The valid entries are actually handled by the hardware.
+                if mmpte.u.Hard.Valid == 1:
+                    physoffset = mmpte.u.Hard.PageFrameNumber << 12
+                    yield physoffset, file_offset, self.PAGE_SIZE
+
+                elif mmpte.u.Soft.Prototype == 1:
+                    if not is_64bit and not is_pae:
+                        subsection_offset = ((mmpte.u.Subsect.SubsectionAddressHigh << 7) | (
+                                mmpte.u.Subsect.SubsectionAddressLow << 3))
+
+                # If the entry is not a valid physical address then see if it is in transition.
+                elif mmpte.u.Trans.Transition == 1:
+                    physoffset = mmpte.u.Trans.PageFrameNumber << 12
+                    yield physoffset, file_offset, self.PAGE_SIZE
+
+                # Go to the next PTE entry
+                ptecount += 1
+
+            # Go to the next Subsection in the single-linked list
+            subsection = subsection.NextSubsection
+
+class VACB(objects.StructType):
+    """A class for _VACB structures"""
+
+    FILEOFFSET_MASK = 0xFFFFFFFFFFFF0000
+
+    def is_valid(self, shared_cache_map: interfaces.objects.ObjectInterface) -> bool:
+        """Determine if the object is valid."""
+        try:
+            layer = self._context.layers[self.vol.layer_name]
+
+            # Check if the Overlay member of _VACB is resident. The Overlay member stores information
+            # about the FileOffset and the ActiveCount. This is just another proactive sanity check.
+            #if not self.Overlay:
+            #    return False
+
+            if not layer.is_valid(self.SharedCacheMap):
+                return False
+
+            # Make sure that the SharedCacheMap member of the VACB points back to the parent object.
+            return self.SharedCacheMap == shared_cache_map.vol.offset
+        except exceptions.InvalidAddressException:
+            return False
+
+    def get_file_offset(self) -> int:
+        # The FileOffset member of VACB is used to denote the offset within the file where the
+        # view begins. Since all views are 256 KB in size, the bottom 16 bits are used to
+        # store the number of references to the view.
+        return self.Overlay.FileOffset.QuadPart & self.FILEOFFSET_MASK
+
+class SHARED_CACHE_MAP(objects.StructType):
+    """A class for _SHARED_CACHE_MAP structures"""
+
+    VACB_BLOCK = 0x40000
+    VACB_OFFSET_SHIFT = 18
+    VACB_LEVEL_SHIFT = 7
+    VACB_SIZE_OF_FIRST_LEVEL = 1 << (VACB_OFFSET_SHIFT + VACB_LEVEL_SHIFT)
+    VACB_ARRAY = 0x80
+
+    def is_valid(self) -> bool:
+        """Determine if the object is valid."""
+
+        if self.FileSize.QuadPart <= 0 or self.ValidDataLength.QuadPart <= 0:
+            return False
+
+        if self.SectionSize.QuadPart < 0 or ((self.FileSize.QuadPart < self.ValidDataLength.QuadPart) and (
+                   self.ValidDataLength.QuadPart != 0x7fffffffffffffff)):
+            return False
+
+        return True
+
+    def process_index_array(self, array_pointer: interfaces.objects.ObjectInterface, level: int, limit: int,
+                            vacb_list: Optional[List] = None) -> List:
+        """Recursively process the sparse multilevel VACB index array.
+
+        :param array_pointer: The address of a possible index array
+        :param level: The current level
+        :param limit: The level where we abandon all hope. Ideally this is 7
+        :param vacb_list: An array of collected VACBs
+        :return: Collected VACBs
+        """
+        if vacb_list is None:
+            vacb_list = []
+
+        if level > limit:
+            return []
+
+        symbol_table_name = self.get_symbol_table_name()
+        pointer_type = self._context.symbol_space.get_type(symbol_table_name + constants.BANG + "pointer")
+
+        # Create an array of 128 entries for the VACB index array
+        vacb_array = self._context.object(object_type=symbol_table_name + constants.BANG + "array",
+                                          layer_name=self.vol.layer_name,
+                                          offset=array_pointer,
+                                          count=self.VACB_ARRAY,
+                                          subtype=pointer_type)
+
+        # Iterate through the entries
+        for counter in range(0, self.VACB_ARRAY):
+            # Check if the VACB entry is in use
+            if not vacb_array[counter]:
+                continue
+
+            vacb_obj = vacb_array[counter].dereference().cast(symbol_table_name + constants.BANG + "_VACB")
+            if vacb_obj.is_valid(shared_cache_map=self):
+                self.save_vacb(vacb_obj, vacb_list)
+            else:
+                # Process the next level of the multi-level array
+                vacb_list = self.process_index_array(vacb_array[counter], level + 1, limit, vacb_list)
+        return vacb_list
+
+    def save_vacb(self, vacb_obj: interfaces.objects.ObjectInterface, vacb_list: List):
+        data = (int(vacb_obj.BaseAddress), int(vacb_obj.get_file_offset()), self.VACB_BLOCK)
+        vacb_list.append(data)
+
+    def get_available_pages(self) -> List:
+        """Get the available pages that correspond to a cached file.
+
+        The lists generated are (virtual_offset, file_offset, page_size).
+        """
+        vacb_list = []
+        section_size = self.SectionSize.QuadPart
+
+        # Determine the number of VACBs within the cache (nonpaged). each VACB
+        # represents a 256-KB view in the system cache.
+        full_blocks = section_size // self.VACB_BLOCK
+        left_over = section_size % self.VACB_BLOCK
+
+        # As an optimization, the shared cache map object contains a VACB index array of four entries.
+        # The VACB index arrays are arrays of pointers to VACBs, that track which views of a given file
+        # are mapped in the cache. For example, the first entry in the VACB index array refers to the first
+        # 256 KB of the file. The InitialVacbs can describe a file up to 1 MB (4xVACB).
+        iterval = 0
+        while (iterval < full_blocks) and (full_blocks <= 4):
+            vacb_obj = self.InitialVacbs[iterval]
+            if vacb_obj.is_valid(shared_cache_map=self):
+                self.save_vacb(vacb_obj, vacb_list)
+            iterval += 1
+
+        # We also have to account for the spill over data that is not found in the full blocks.
+        # The first case to consider is when the spill over is still in InitialVacbs.
+        if (left_over > 0) and (full_blocks < 4):
+            vacb_obj = self.InitialVacbs[iterval]
+            if vacb_obj.is_valid(shared_cache_map=self):
+                self.save_vacb(vacb_obj, vacb_list)
+
+        # If the file is larger than 1 MB, a seperate VACB index array needs to be allocated.
+        # This is based on how many 256 KB blocks would be required for the size of the file.
+        # This newly allocated VACB index array is found through the Vacbs member of SHARED_CACHE_MAP.
+        vacb_obj = self.Vacbs
+
+        # Note: avoid calling is_valid() here, since self.Vacbs is a pointer to a pointer
+        if not vacb_obj:
+            return vacb_list
+
+        # There are a number of instances where the initial value in InitialVacb will also be the fist
+        # entry in Vacbs. Thus we ignore, since it was already processed. It is possible to just
+        # process again as the file offset is specified for each VACB.
+        if self.InitialVacbs[0].vol.offset == vacb_obj:
+            return vacb_list
+
+        # If the file is less than 32 MB than it can be found in a single level VACB index array.
+        symbol_table_name = self.get_symbol_table_name()
+        pointer_type = self._context.symbol_space.get_type(symbol_table_name + constants.BANG + "pointer")
+        size_of_pointer = pointer_type.size
+
+        if not section_size > self.VACB_SIZE_OF_FIRST_LEVEL:
+            array_head = vacb_obj
+            for counter in range(0, full_blocks):
+                vacb_entry = self._context.object(symbol_table_name + constants.BANG + "pointer",
+                                                  layer_name=self.vol.layer_name,
+                                                  offset=array_head + (counter * size_of_pointer))
+
+                # If we find a zero entry, then we proceed to the next one. If the entry is zero,
+                # then the view is not mapped and we skip. We do not pad because we use the
+                # FileOffset to seek to the correct offset in the file.
+                if not vacb_entry:
+                    continue
+
+                vacb = vacb_entry.dereference().cast(symbol_table_name + constants.BANG + "_VACB")
+                if vacb.is_valid(shared_cache_map=self):
+                    self.save_vacb(vacb, vacb_list)
+
+            if left_over > 0:
+                vacb_entry = self._context.object(symbol_table_name + constants.BANG + "pointer",
+                                                  layer_name=self.vol.layer_name,
+                                                  offset=array_head + ((counter + 1) * size_of_pointer))
+
+                if not vacb_entry:
+                    return vacb_list
+
+                vacb = vacb_entry.dereference().cast(symbol_table_name + constants.BANG + "_VACB")
+                if vacb.is_valid(shared_cache_map=self):
+                    self.save_vacb(vacb, vacb_list)
+
+            # The file is less than 32 MB, so we can stop processing.
+            return vacb_list
+
+        # If we get to this point, then we know that the SectionSize is greater than
+        # VACB_SIZE_OF_FIRST_LEVEL (32 MB). Then we have a "sparse" multilevel index
+        # array where each VACB index array is made up of 128 entries. We no
+        # longer assume the data is sequential. (Log2 (32 MB) - 18)/7
+        level_depth = math.ceil(math.log(section_size, 2))
+        level_depth = (level_depth - self.VACB_OFFSET_SHIFT) / self.VACB_LEVEL_SHIFT
+        level_depth = math.ceil(level_depth)
+        limit_depth = level_depth
+
+        if section_size > self.VACB_SIZE_OF_FIRST_LEVEL:
+
+            # Create an array of 128 entries for the VACB index array.
+            vacb_array = self._context.object(object_type=symbol_table_name + constants.BANG + "array",
+                                              layer_name=self.vol.layer_name,
+                                              offset=vacb_obj,
+                                              count=self.VACB_ARRAY,
+                                              subtype=pointer_type)
+
+            # Walk the array and if any entry points to the shared cache map object then we extract it.
+            # Otherwise, if it is non-zero, then traverse to the next level.
+            for counter in range(0, self.VACB_ARRAY):
+                if not vacb_array[counter]:
+                    continue
+
+                vacb = vacb_array[counter].dereference().cast(symbol_table_name + constants.BANG + "_VACB")
+                if vacb.SharedCacheMap == self.vol.offset:
+                    if vacb.is_valid(shared_cache_map=self):
+                        self.save_vacb(vacb, vacb_list)
+                else:
+                    # Process the next level of the multi-level array. We set the limit_depth to be
+                    # the depth of the tree as determined from the size and we initialize the
+                    # current level to 2.
+                    vacb_list = self.process_index_array(vacb_array[counter], 2, limit_depth, vacb_list)
+
+        return vacb_list