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

import collections.abc
import random
import string

from volatility.framework import interfaces
from volatility.framework import objects
from volatility.framework import exceptions

# Keep these in a basic module, to prevent import cycles when symbol providers require them

class _ETHREAD(objects.Struct):
    def owning_process(self, kernel_layer = None):
        """Return the EPROCESS that owns this thread"""
        return self.ThreadsProcess.dereference(kernel_layer)

class _CMHIVE(objects.Struct):
    @property
    def name(self):
        """Determine a name for the hive. Note that some attributes are
        unpredictably blank across different OS versions while others are populated,
        so we check all possibilities and take the first one that's not empty"""
        
        for attr in ["FileFullPath", "FileUserName", "HiveRootPath"]:
            try:
                return getattr(self, attr).String
            except (AttributeError, exceptions.InvalidAddressException):
                pass

        return None

class _UNICODE_STRING(objects.Struct):
    @property
    def String(self):
        # We explicitly do *not* catch errors here, we allow an exception to be thrown
        # (otherwise there's no way to determine anything went wrong)
        # It's up to the user of this method to catch exceptions
        return self.Buffer.dereference().cast("string", max_length = self.Length, errors = "replace",
                                              encoding = "utf16")


class _EPROCESS(objects.Struct):
    def add_process_layer(self, context, config_prefix = None, preferred_name = None):
        """Constructs a new layer based on the process's DirectoryTableBase"""

        if config_prefix is None:
            # TODO: Ensure collisions can't happen by verifying the config_prefix is empty
            random_prefix = ''.join(
                random.SystemRandom().choice(string.ascii_uppercase + string.digits) for _ in range(8))
            config_prefix = interfaces.configuration.path_join("temporary", "_" + random_prefix)

        # Figure out a suitable name we can use for the new layer
        if preferred_name is None:
            preferred_name = context.memory.free_layer_name(
                prefix = self.vol.layer_name + "_PID" + str(self.UniqueProcessId) + "_")
        else:
            if preferred_name in context.memory:
                preferred_name = context.memory.free_layer_name(prefix = preferred_name)

        # Copy the parent's config and then make suitable changes
        parent_layer = context.memory[self.vol.layer_name]
        parent_config = parent_layer.build_configuration()
        parent_config['memory_layer'] = parent_layer.config['memory_layer']
        # Presumably for 64-bit systems, the DTB is defined as an array, rather than an unsigned long long
        if isinstance(self.Pcb.DirectoryTableBase, objects.Array):
            parent_config['page_map_offset'] = self.Pcb.DirectoryTableBase.cast("unsigned long long")
        else:
            parent_config['page_map_offset'] = self.Pcb.DirectoryTableBase
        parent_config['page_map_offset'] = parent_config['page_map_offset'] & (
            (1 << parent_layer.bits_per_register) - 1)

        # Set the new configuration and construct the layer
        config_path = interfaces.configuration.path_join(config_prefix, preferred_name)
        context.config.splice(config_path, parent_config)
        new_layer = parent_layer.__class__(context, config_path = config_path, name = preferred_name)

        # Add the constructed layer and return the name
        context.memory.add_layer(new_layer)
        return preferred_name

    def load_order_modules(self):
        """Generator for DLLs in the order that they were loaded"""

        proc_layer_name = self.add_process_layer(self._context)

        proc_layer = self._context.memory[proc_layer_name]
        if not proc_layer.is_valid(self.Peb):
            raise StopIteration

        sym_table = self.vol.type_name.split("!")[0]
        peb = self._context.object("{}!_PEB".format(sym_table), layer_name = proc_layer_name, offset = self.Peb)

        for entry in peb.Ldr.InLoadOrderModuleList.to_list("{}!_LDR_DATA_TABLE_ENTRY".format(sym_table),
                                                           "InLoadOrderLinks"):
            yield entry


class _LIST_ENTRY(objects.Struct, collections.abc.Iterable):
    def to_list(self, symbol_type, member, forward = True, sentinel = True, layer = None):
        """Returns an iterator of the entries in the list"""

        if layer is None:
            layer = self.vol.layer_name

        relative_offset = self._context.symbol_space.get_type(symbol_type).relative_child_offset(member)

        direction = 'Blink'
        if forward:
            direction = 'Flink'
        link = getattr(self, direction).dereference()

        if not sentinel:
            yield self._context.object(symbol_type, layer, offset = self.vol.offset - relative_offset)

        seen = {self.vol.offset}
        while link.vol.offset not in seen:

            obj = self._context.object(symbol_type, layer, offset = link.vol.offset - relative_offset)
            yield obj

            seen.add(link.vol.offset)
            link = getattr(link, direction).dereference()

    def __iter__(self):
        return self.to_list(self.vol.parent.vol.type_name, self.vol.member_name)