mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-25 02:54:51 +02:00
Codebase: Yapf the codebase again
This commit is contained in:
Mike Auty
@@ -91,9 +91,9 @@ class VmwareLayer(segmented.SegmentedLayer):
|
||||
if tags[("regionsCount", ())][1] == 0:
|
||||
raise VmwareFormatException(self.name, "VMware VMEM is not split into regions")
|
||||
for region in range(tags[("regionsCount", ())][1]):
|
||||
offset = tags[("regionPPN", (region,))][1] * self._page_size
|
||||
mapped_offset = tags[("regionPageNum", (region,))][1] * self._page_size
|
||||
length = tags[("regionSize", (region,))][1] * self._page_size
|
||||
offset = tags[("regionPPN", (region, ))][1] * self._page_size
|
||||
mapped_offset = tags[("regionPageNum", (region, ))][1] * self._page_size
|
||||
length = tags[("regionSize", (region, ))][1] * self._page_size
|
||||
self._segments.append((offset, mapped_offset, length, length))
|
||||
|
||||
@property
|
||||
|
||||
@@ -80,7 +80,7 @@ class LayerWriter(plugins.PluginInterface):
|
||||
def _generator(self):
|
||||
if self.config['list']:
|
||||
for name in self.context.layers:
|
||||
yield 0, (name,)
|
||||
yield 0, (name, )
|
||||
else:
|
||||
import pdb
|
||||
pdb.set_trace()
|
||||
@@ -94,7 +94,7 @@ class LayerWriter(plugins.PluginInterface):
|
||||
for name in self.config['layers']:
|
||||
# Check the layer exists and validate the output file
|
||||
if name not in self.context.layers:
|
||||
yield 0, ('Layer Name {} does not exist'.format(name),)
|
||||
yield 0, ('Layer Name {} does not exist'.format(name), )
|
||||
else:
|
||||
output_name = self.config.get('output', ".".join([name, "raw"]))
|
||||
try:
|
||||
@@ -106,14 +106,14 @@ class LayerWriter(plugins.PluginInterface):
|
||||
progress_callback = self._progress_callback)
|
||||
file_handle.close()
|
||||
except IOError as excp:
|
||||
yield 0, ('Layer cannot be written to {}: {}'.format(self.config['output_name'], excp),)
|
||||
yield 0, ('Layer cannot be written to {}: {}'.format(self.config['output_name'], excp), )
|
||||
|
||||
yield 0, ('Layer has been written to {}'.format(output_name),)
|
||||
yield 0, ('Layer has been written to {}'.format(output_name), )
|
||||
|
||||
def _generate_layers(self):
|
||||
"""List layer names from this run"""
|
||||
for name in self.context.layers:
|
||||
yield (0, (name,))
|
||||
yield (0, (name, ))
|
||||
|
||||
def run(self):
|
||||
if self.config['list']:
|
||||
|
||||
@@ -39,10 +39,11 @@ class Socket_filters(plugins.PluginInterface):
|
||||
|
||||
handlers = mac.MacUtilities.generate_kernel_handler_info(self.context, self.config['primary'], kernel, mods)
|
||||
|
||||
members_to_check = ["sf_unregistered", "sf_attach", "sf_detach", "sf_notify", "sf_getpeername",
|
||||
"sf_getsockname",
|
||||
"sf_data_in", "sf_data_out", "sf_connect_in", "sf_connect_out", "sf_bind", "sf_setoption",
|
||||
"sf_getoption", "sf_listen", "sf_ioctl"]
|
||||
members_to_check = [
|
||||
"sf_unregistered", "sf_attach", "sf_detach", "sf_notify", "sf_getpeername", "sf_getsockname", "sf_data_in",
|
||||
"sf_data_out", "sf_connect_in", "sf_connect_out", "sf_bind", "sf_setoption", "sf_getoption", "sf_listen",
|
||||
"sf_ioctl"
|
||||
]
|
||||
|
||||
filter_list = kernel.object_from_symbol(symbol_name = "sock_filter_head")
|
||||
|
||||
|
||||
@@ -188,7 +188,7 @@ class Timeliner(interfaces.plugins.PluginInterface):
|
||||
|
||||
if isinstance(plugin, TimeLinerInterface):
|
||||
if not len(filter_list) or any(
|
||||
[filter in plugin.__module__ + '.' + plugin.__class__.__name__ for filter in filter_list]):
|
||||
[filter in plugin.__module__ + '.' + plugin.__class__.__name__ for filter in filter_list]):
|
||||
plugins_to_run.append(plugin)
|
||||
except exceptions.UnsatisfiedException as excp:
|
||||
# Remove the failed plugin from the list and continue
|
||||
|
||||
@@ -55,7 +55,7 @@ class Cachedump(interfaces.plugins.PluginInterface):
|
||||
(uname_len, domain_len) = unpack("<HH", cache_data[:4])
|
||||
if len(cache_data[60:62]) == 0:
|
||||
return (uname_len, domain_len, 0, '', '')
|
||||
(domain_name_len,) = unpack("<H", cache_data[60:62])
|
||||
(domain_name_len, ) = unpack("<H", cache_data[60:62])
|
||||
ch = cache_data[64:80]
|
||||
enc_data = cache_data[96:]
|
||||
return (uname_len, domain_len, domain_name_len, enc_data, ch)
|
||||
|
||||
@@ -72,8 +72,8 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
if layer_name is None:
|
||||
layer_name = dll_entry.vol.layer_name
|
||||
|
||||
file_handle = open_method("{}{}.{:#x}.{:#x}.dmp".format(prefix, ntpath.basename(name),
|
||||
dll_entry.vol.offset, dll_entry.DllBase))
|
||||
file_handle = open_method("{}{}.{:#x}.{:#x}.dmp".format(prefix, ntpath.basename(name), dll_entry.vol.offset,
|
||||
dll_entry.DllBase))
|
||||
|
||||
dos_header = context.object(pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
|
||||
offset = dll_entry.DllBase,
|
||||
|
||||
@@ -21,6 +21,7 @@ EXTENSION_CACHE_MAP = {
|
||||
"vacb": "SharedCacheMap",
|
||||
}
|
||||
|
||||
|
||||
class DumpFiles(interfaces.plugins.PluginInterface):
|
||||
"""Dumps cached file contents from Windows memory samples."""
|
||||
|
||||
@@ -31,26 +32,25 @@ class DumpFiles(interfaces.plugins.PluginInterface):
|
||||
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='virtaddr',
|
||||
description="Dump a single _FILE_OBJECT at this virtual address",
|
||||
optional=True),
|
||||
requirements.IntRequirement(name='physaddr',
|
||||
description="Dump a single _FILE_OBJECT at this physical address",
|
||||
optional=True),
|
||||
requirements.VersionRequirement(name='pslist', component=pslist.PsList, version=(2, 0, 0)),
|
||||
requirements.VersionRequirement(name='handles', component=handles.Handles, version=(1, 0, 0))
|
||||
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 = 'virtaddr',
|
||||
description = "Dump a single _FILE_OBJECT at this virtual address",
|
||||
optional = True),
|
||||
requirements.IntRequirement(name = 'physaddr',
|
||||
description = "Dump a single _FILE_OBJECT at this physical address",
|
||||
optional = True),
|
||||
requirements.VersionRequirement(name = 'pslist', component = pslist.PsList, version = (2, 0, 0)),
|
||||
requirements.VersionRequirement(name = 'handles', component = handles.Handles, version = (1, 0, 0))
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def dump_file_producer(cls,
|
||||
file_object: interfaces.objects.ObjectInterface,
|
||||
def dump_file_producer(cls, file_object: interfaces.objects.ObjectInterface,
|
||||
memory_object: interfaces.objects.ObjectInterface,
|
||||
open_method: Type[interfaces.plugins.FileHandlerInterface],
|
||||
layer: interfaces.layers.DataLayerInterface,
|
||||
@@ -86,14 +86,11 @@ class DumpFiles(interfaces.plugins.PluginInterface):
|
||||
vollog.debug("Stored {}".format(filedata.preferred_filename))
|
||||
return filedata
|
||||
except exceptions.InvalidAddressException:
|
||||
vollog.debug("Unable to dump file at {0:#x}".format(
|
||||
file_object.vol.offset))
|
||||
vollog.debug("Unable to dump file at {0:#x}".format(file_object.vol.offset))
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def process_file_object(cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
primary_layer_name: str,
|
||||
def process_file_object(cls, context: interfaces.context.ContextInterface, primary_layer_name: str,
|
||||
open_method: Type[interfaces.plugins.FileHandlerInterface],
|
||||
file_obj: interfaces.objects.ObjectInterface) -> Tuple:
|
||||
"""Given a FILE_OBJECT, dump data to separate files for each of the three file caches.
|
||||
@@ -153,10 +150,8 @@ class DumpFiles(interfaces.plugins.PluginInterface):
|
||||
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)
|
||||
memory_object.vol.offset, cache_name,
|
||||
ntpath.basename(obj_name), extension)
|
||||
|
||||
file_handle = DumpFiles.dump_file_producer(file_obj, memory_object, open_method, layer, desired_file_name)
|
||||
|
||||
@@ -165,8 +160,10 @@ class DumpFiles(interfaces.plugins.PluginInterface):
|
||||
file_handle.close()
|
||||
file_output = file_handle.preferred_filename
|
||||
|
||||
yield (cache_name, format_hints.Hex(file_obj.vol.offset),
|
||||
ntpath.basename(obj_name), # temporary, so its easier to visualize output
|
||||
yield (
|
||||
cache_name,
|
||||
format_hints.Hex(file_obj.vol.offset),
|
||||
ntpath.basename(obj_name), # temporary, so its easier to visualize output
|
||||
file_output)
|
||||
|
||||
def _generator(self, procs: List, offsets: List):
|
||||
@@ -176,13 +173,13 @@ class DumpFiles(interfaces.plugins.PluginInterface):
|
||||
# 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.
|
||||
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"])
|
||||
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:
|
||||
|
||||
@@ -198,7 +195,8 @@ class DumpFiles(interfaces.plugins.PluginInterface):
|
||||
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(self.context, self.config["primary"], self.open, file_obj):
|
||||
for result in self.process_file_object(self.context, self.config["primary"], self.open,
|
||||
file_obj):
|
||||
yield (0, result)
|
||||
except exceptions.InvalidAddressException:
|
||||
vollog.log(constants.LOGLEVEL_VVV,
|
||||
@@ -221,7 +219,8 @@ class DumpFiles(interfaces.plugins.PluginInterface):
|
||||
if not file_obj.is_valid():
|
||||
continue
|
||||
|
||||
for result in self.process_file_object(self.context, self.config["primary"], self.open, file_obj):
|
||||
for result in self.process_file_object(self.context, self.config["primary"], self.open,
|
||||
file_obj):
|
||||
yield (0, result)
|
||||
except exceptions.InvalidAddressException:
|
||||
vollog.log(constants.LOGLEVEL_VVV,
|
||||
@@ -237,14 +236,13 @@ class DumpFiles(interfaces.plugins.PluginInterface):
|
||||
layer_name = self.context.layers[layer_name].config["memory_layer"]
|
||||
|
||||
file_obj = self.context.object(self.config["nt_symbols"] + constants.BANG + "_FILE_OBJECT",
|
||||
layer_name=layer_name,
|
||||
native_layer_name=self.config["primary"],
|
||||
offset=offset)
|
||||
layer_name = layer_name,
|
||||
native_layer_name = self.config["primary"],
|
||||
offset = offset)
|
||||
for result in self.process_file_object(self.context, self.config["primary"], self.open, file_obj):
|
||||
yield (0, result)
|
||||
except exceptions.InvalidAddressException:
|
||||
vollog.log(constants.LOGLEVEL_VVV,
|
||||
"Cannot extract file at {0:#x}".format(offset))
|
||||
vollog.log(constants.LOGLEVEL_VVV, "Cannot extract file at {0:#x}".format(offset))
|
||||
|
||||
def run(self):
|
||||
# a list of tuples (<int>, <bool>) where <int> is the address and <bool> is True for virtual.
|
||||
@@ -261,8 +259,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
|
||||
procs = pslist.PsList.list_processes(self.context,
|
||||
self.config["primary"],
|
||||
self.config["nt_symbols"],
|
||||
filter_func=filter_func)
|
||||
filter_func = filter_func)
|
||||
|
||||
return renderers.TreeGrid(
|
||||
[("Cache", str), ("FileObject", format_hints.Hex), ("FileName", str), ("Result", str)],
|
||||
self._generator(procs, offsets))
|
||||
return renderers.TreeGrid([("Cache", str), ("FileObject", format_hints.Hex), ("FileName", str),
|
||||
("Result", str)], self._generator(procs, offsets))
|
||||
|
||||
@@ -90,8 +90,8 @@ class Handles(interfaces.plugins.PluginInterface):
|
||||
if not has_capstone:
|
||||
raise AttributeError("Unable to find the SAR value for decoding handle table pointers")
|
||||
else:
|
||||
raise exceptions.MissingModuleException("capstone",
|
||||
"Unable to find the SAR value for decoding handle table pointers")
|
||||
raise exceptions.MissingModuleException(
|
||||
"capstone", "Unable to find the SAR value for decoding handle table pointers")
|
||||
|
||||
offset = self._decode_pointer(handle_table_entry.LowValue, magic)
|
||||
# print("LowValue: {0:#x} Magic: {1:#x} Offset: {2:#x}".format(handle_table_entry.InfoTable, magic, offset))
|
||||
|
||||
@@ -105,8 +105,8 @@ class Malfind(interfaces.plugins.PluginInterface):
|
||||
continue
|
||||
|
||||
if (vad.get_private_memory() == 1
|
||||
and vad.get_tag() == "VadS") or (vad.get_private_memory() == 0
|
||||
and protection_string != "PAGE_EXECUTE_WRITECOPY"):
|
||||
and vad.get_tag() == "VadS") or (vad.get_private_memory() == 0
|
||||
and protection_string != "PAGE_EXECUTE_WRITECOPY"):
|
||||
if cls.is_vad_empty(proc_layer, vad):
|
||||
continue
|
||||
|
||||
|
||||
@@ -185,7 +185,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
}
|
||||
|
||||
# special use case: Win10_18363 is not recognized by windows.info as 18363
|
||||
# because all kernel file headers and debug structures report 18363 as
|
||||
# because all kernel file headers and debug structures report 18363 as
|
||||
# "10.0.18362.1198" with the last part being incremented. However, we can use
|
||||
# os_distinguisher to differentiate between 18362 and 18363
|
||||
if vers_minor_version == 18362 and is_18363_or_later:
|
||||
@@ -202,9 +202,11 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
# no match on filename means that we possibly have a version newer than those listed here.
|
||||
# try to grab the latest supported version of the current image NT version. If that symbol
|
||||
# version does not work, support has to be added manually.
|
||||
current_versions = [key for key in list(version_dict.keys()) if key[0] == nt_major_version and key[1] == nt_minor_version]
|
||||
current_versions = [
|
||||
key for key in list(version_dict.keys()) if key[0] == nt_major_version and key[1] == nt_minor_version
|
||||
]
|
||||
current_versions.sort()
|
||||
|
||||
|
||||
if current_versions:
|
||||
latest_version = current_versions[-1]
|
||||
|
||||
|
||||
@@ -188,8 +188,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
|
||||
file_output = "Disabled"
|
||||
if self.config['dump']:
|
||||
file_handle = self.process_dump(self.context, self.config['nt_symbols'], pe_table_name, proc,
|
||||
self.open)
|
||||
file_handle = self.process_dump(self.context, self.config['nt_symbols'], pe_table_name, proc, self.open)
|
||||
file_output = "Error outputting file"
|
||||
if file_handle:
|
||||
file_handle.close()
|
||||
|
||||
@@ -123,11 +123,8 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
proc.vol.offset == ph_offset:
|
||||
return virtual_process
|
||||
|
||||
|
||||
@classmethod
|
||||
def get_osversion(cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
layer_name: str,
|
||||
def get_osversion(cls, context: interfaces.context.ContextInterface, layer_name: str,
|
||||
symbol_table: str) -> Tuple[int, int, int]:
|
||||
"""Returns the complete OS version (MAJ,MIN,BUILD)
|
||||
|
||||
@@ -166,8 +163,8 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
vproc = self.virtual_process_from_physical(self.context, self.config['primary'],
|
||||
self.config['nt_symbols'], proc)
|
||||
|
||||
file_handle = pslist.PsList.process_dump(self.context, self.config['nt_symbols'], pe_table_name,
|
||||
vproc, self.open)
|
||||
file_handle = pslist.PsList.process_dump(self.context, self.config['nt_symbols'], pe_table_name, vproc,
|
||||
self.open)
|
||||
file_output = "Error outputting file"
|
||||
if file_handle:
|
||||
file_output = file_handle.preferred_filename
|
||||
|
||||
@@ -33,9 +33,7 @@ class PsTree(interfaces.plugins.PluginInterface):
|
||||
description = 'Display physical offsets instead of virtual',
|
||||
default = pslist.PsList.PHYSICAL_DEFAULT,
|
||||
optional = True),
|
||||
requirements.VersionRequirement(name = 'pslist',
|
||||
component = pslist.PsList,
|
||||
version = (2, 0, 0)),
|
||||
requirements.VersionRequirement(name = 'pslist', component = pslist.PsList, version = (2, 0, 0)),
|
||||
requirements.ListRequirement(name = 'pid',
|
||||
element_type = int,
|
||||
description = "Process ID to include (all other processes are excluded)",
|
||||
|
||||
@@ -43,10 +43,10 @@ class PrintKey(interfaces.plugins.PluginInterface):
|
||||
|
||||
@classmethod
|
||||
def key_iterator(
|
||||
cls,
|
||||
hive: RegistryHive,
|
||||
node_path: Sequence[objects.StructType] = None,
|
||||
recurse: bool = False
|
||||
cls,
|
||||
hive: RegistryHive,
|
||||
node_path: Sequence[objects.StructType] = None,
|
||||
recurse: bool = False
|
||||
) -> Iterable[Tuple[int, bool, datetime.datetime, str, bool, interfaces.objects.ObjectInterface]]:
|
||||
"""Walks through a set of nodes from a given node (last one in
|
||||
node_path). Avoids loops by not traversing into nodes already present
|
||||
|
||||
@@ -112,8 +112,7 @@ class VadInfo(interfaces.plugins.PluginInterface):
|
||||
context: interfaces.context.ContextInterface,
|
||||
proc: interfaces.objects.ObjectInterface,
|
||||
vad: interfaces.objects.ObjectInterface,
|
||||
open_method: Type[
|
||||
interfaces.plugins.FileHandlerInterface],
|
||||
open_method: Type[interfaces.plugins.FileHandlerInterface],
|
||||
maxsize: int = MAXSIZE_DEFAULT) -> Optional[interfaces.plugins.FileHandlerInterface]:
|
||||
"""Extracts the complete data for Vad as a FileInterface.
|
||||
|
||||
|
||||
@@ -405,6 +405,7 @@ class FILE_OBJECT(objects.StructType, pool.ExecutiveObject):
|
||||
pass
|
||||
|
||||
return name
|
||||
|
||||
def access_string(self):
|
||||
## Make a nicely formatted ACL string
|
||||
return (('R' if self.ReadAccess else '-') + ('W' if self.WriteAccess else '-') +
|
||||
@@ -431,6 +432,7 @@ class ETHREAD(objects.StructType):
|
||||
def owning_process(self, kernel_layer: str = None) -> interfaces.objects.ObjectInterface:
|
||||
"""Return the EPROCESS that owns this thread."""
|
||||
return self.ThreadsProcess.dereference(kernel_layer)
|
||||
|
||||
def get_cross_thread_flags(self) -> str:
|
||||
dictCrossThreadFlags = {
|
||||
'PS_CROSS_THREAD_FLAGS_TERMINATED': 0,
|
||||
@@ -665,6 +667,7 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
|
||||
else:
|
||||
# windows xp and 2003
|
||||
return self.VadRoot.dereference().cast("_MMVAD")
|
||||
|
||||
def environment_variables(self):
|
||||
"""Generator for environment variables.
|
||||
|
||||
@@ -754,6 +757,7 @@ 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 TOKEN(objects.StructType):
|
||||
"""A class for process etoken object."""
|
||||
|
||||
@@ -873,6 +877,7 @@ class KTHREAD(objects.StructType):
|
||||
}
|
||||
return dictWaitReason.get(self.WaitReason, renderers.NotApplicableValue())
|
||||
|
||||
|
||||
class CONTROL_AREA(objects.StructType):
|
||||
"""A class for _CONTROL_AREA structures"""
|
||||
|
||||
@@ -904,17 +909,17 @@ class CONTROL_AREA(objects.StructType):
|
||||
"""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)
|
||||
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)
|
||||
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.
|
||||
@@ -963,8 +968,8 @@ class CONTROL_AREA(objects.StructType):
|
||||
|
||||
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))
|
||||
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:
|
||||
@@ -977,6 +982,7 @@ class CONTROL_AREA(objects.StructType):
|
||||
# Go to the next Subsection in the single-linked list
|
||||
subsection = subsection.NextSubsection
|
||||
|
||||
|
||||
class VACB(objects.StructType):
|
||||
"""A class for _VACB structures"""
|
||||
|
||||
@@ -988,6 +994,7 @@ class VACB(objects.StructType):
|
||||
# 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"""
|
||||
|
||||
@@ -1003,13 +1010,16 @@ class SHARED_CACHE_MAP(objects.StructType):
|
||||
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)):
|
||||
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,
|
||||
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.
|
||||
|
||||
@@ -1029,11 +1039,11 @@ class SHARED_CACHE_MAP(objects.StructType):
|
||||
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)
|
||||
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):
|
||||
@@ -1042,7 +1052,7 @@ class SHARED_CACHE_MAP(objects.StructType):
|
||||
continue
|
||||
|
||||
vacb_obj = vacb_array[counter].dereference().cast(symbol_table_name + constants.BANG + "_VACB")
|
||||
if vacb_obj.is_valid(shared_cache_map=self):
|
||||
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
|
||||
@@ -1112,8 +1122,8 @@ class SHARED_CACHE_MAP(objects.StructType):
|
||||
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))
|
||||
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
|
||||
@@ -1127,8 +1137,8 @@ class SHARED_CACHE_MAP(objects.StructType):
|
||||
|
||||
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))
|
||||
layer_name = self.vol.layer_name,
|
||||
offset = array_head + ((counter + 1) * size_of_pointer))
|
||||
|
||||
if not vacb_entry:
|
||||
return vacb_list
|
||||
@@ -1152,11 +1162,11 @@ class SHARED_CACHE_MAP(objects.StructType):
|
||||
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)
|
||||
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.
|
||||
|
||||
@@ -159,7 +159,8 @@ class _TCP_LISTENER(objects.StructType):
|
||||
|
||||
try:
|
||||
if not self.get_address_family() in (AF_INET, AF_INET6):
|
||||
vollog.debug("netw obj 0x{:x} invalid due to invalid address_family {}".format(self.vol.offset, self.get_address_family()))
|
||||
vollog.debug("netw obj 0x{:x} invalid due to invalid address_family {}".format(
|
||||
self.vol.offset, self.get_address_family()))
|
||||
return False
|
||||
|
||||
except exceptions.InvalidAddressException:
|
||||
|
||||
Reference in New Issue
Block a user