mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-30 13:34:53 +02:00
Mike Auty
d32fc223e2
Note: also remove interfaces_configuration where ever possible (this is not possible in the interfaces directory itself because it would cause an import loop).
103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
import collections.abc
|
|
|
|
from volatility.framework import constants, exceptions, objects
|
|
from volatility.framework.symbols import generic
|
|
|
|
|
|
# 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(generic.GenericIntelProcess):
|
|
def add_process_layer(self, context, config_prefix = None, preferred_name = None):
|
|
"""Constructs a new layer based on the process's DirectoryTableBase"""
|
|
|
|
parent_layer = context.memory[self.vol.layer_name]
|
|
# 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):
|
|
dtb = self.Pcb.DirectoryTableBase.cast("unsigned long long")
|
|
else:
|
|
dtb = self.Pcb.DirectoryTableBase
|
|
dtb = dtb & ((1 << parent_layer.bits_per_register) - 1)
|
|
|
|
# Add the constructed layer and return the name
|
|
return self._add_process_layer(context, dtb, config_prefix, 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(constants.BANG)[0]
|
|
peb = self._context.object("{}{}_PEB".format(sym_table, constants.BANG), layer_name = proc_layer_name,
|
|
offset = self.Peb)
|
|
|
|
for entry in peb.Ldr.InLoadOrderModuleList.to_list(
|
|
"{}{}_LDR_DATA_TABLE_ENTRY".format(sym_table, constants.BANG), "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)
|