mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-06 17:57:38 +02:00
111 lines
4.7 KiB
Python
111 lines
4.7 KiB
Python
import collections.abc
|
|
import random
|
|
import string
|
|
|
|
from volatility.framework import interfaces
|
|
from volatility.framework import objects
|
|
|
|
|
|
# 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 _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)
|