Windows: Resolve file_handler type confusion

This commit is contained in:
Mike Auty
2020-10-29 09:43:16 +00:00
committed by ikelos
parent eaa0d2cd9d
commit d554ae0218
9 changed files with 44 additions and 43 deletions
+3 -3
View File
@@ -64,15 +64,15 @@ class LayerWriter(plugins.PluginInterface):
if chunk_size is None:
chunk_size = cls.default_block_size
filehandler = file_handler(preferred_name)
with filehandler as file_data:
file_handle = file_handler(preferred_name)
with file_handle as file_data:
for i in range(0, layer.maximum_address, chunk_size):
current_chunk_size = min(chunk_size, layer.maximum_address - i)
data = layer.read(i, current_chunk_size, pad = True)
file_data.write(data)
if progress_callback:
progress_callback((i / layer.maximum_address) * 100, 'Writing layer {}'.format(layer_name))
return filehandler
return file_handle
def _generator(self):
if self.config['primary'] not in self.context.layers:
+10 -10
View File
@@ -71,23 +71,23 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
if layer_name is None:
layer_name = dll_entry.vol.layer_name
file_handler = file_handler("{}{}.{:#x}.{:#x}.dmp".format(prefix,
ntpath.basename(name),
dll_entry.vol.offset,
dll_entry.DllBase))
file_handle = file_handler("{}{}.{:#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,
layer_name = layer_name)
with file_handler as file_data:
with file_handle as file_data:
for offset, data in dos_header.reconstruct():
file_data.seek(offset)
file_data.write(data)
except (IOError, exceptions.VolatilityException, OverflowError, ValueError) as excp:
vollog.debug("Unable to dump dll at offset {}: {}".format(dll_entry.DllBase, excp))
return None
return file_handler
return file_handle
def _generator(self, procs):
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
@@ -128,11 +128,11 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
file_output = "Disabled"
if self.config['dump']:
file_handler = self.dump_pe(self.context, pe_table_name, entry, self._file_handler,
proc_layer_name, prefix = "pid.{}.".format(proc_id))
file_handle = self.dump_pe(self.context, pe_table_name, entry, self._file_handler,
proc_layer_name, prefix = "pid.{}.".format(proc_id))
file_output = "Error outputting file"
if file_handler:
file_output = file_handler.preferred_filename
if file_handle:
file_output = file_handle.preferred_filename
yield (0, (proc.UniqueProcessId,
proc.ImageFileName.cast("string",
@@ -135,8 +135,8 @@ class Malfind(interfaces.plugins.PluginInterface):
if self.config['dump']:
file_output = "Error outputting to file"
try:
file_handler = vadinfo.VadInfo.vad_dump(self.context, proc, vad, self._file_handler)
file_output = file_handler.preferred_filename
file_handle = vadinfo.VadInfo.vad_dump(self.context, proc, vad, self._file_handler)
file_output = file_handle.preferred_filename
except (exceptions.InvalidAddressException, OverflowError) as excp:
vollog.debug("Unable to dump PE with pid {0}.{1:#x}: {2}".format(proc.UniqueProcessId,
vad.get_start(), excp))
@@ -48,8 +48,8 @@ class Memmap(interfaces.plugins.PluginInterface):
excp.layer_name))
continue
file_handler = self.open("pid.{}.dmp".format(pid))
with file_handler as file_data:
file_handle = self.open("pid.{}.dmp".format(pid))
with file_handle as file_data:
for mapval in proc_layer.mapping(0x0, proc_layer.maximum_address, ignore_errors = True):
offset, size, mapped_offset, mapped_size, maplayer = mapval
@@ -59,11 +59,11 @@ class Memmap(interfaces.plugins.PluginInterface):
try:
data = proc_layer.read(offset, size, pad = True)
file_data.write(data)
file_output = file_handler.preferred_filename
file_output = file_handle.preferred_filename
except exceptions.InvalidAddressException:
file_output = "Error outputting to file"
vollog.debug("Unable to write {}'s address {} to {}".format(proc_layer_name, offset,
file_handler.preferred_filename))
file_handle.preferred_filename))
yield (0, (
format_hints.Hex(offset),
@@ -161,11 +161,11 @@ class ModScan(interfaces.plugins.PluginInterface):
session_layer_name = self.find_session_layer(self.context, session_layers, mod.DllBase)
file_output = "Cannot find a viable session layer for {0:#x}".format(mod.DllBase)
if session_layer_name:
file_handler = dlllist.DllList.dump_pe(self.context, pe_table_name, mod, self.open,
layer_name = session_layer_name)
file_handle = dlllist.DllList.dump_pe(self.context, pe_table_name, mod, self.open,
layer_name = session_layer_name)
file_output = "Error outputting file"
if file_handler:
file_output = file_handler.preferred_filename
if file_handle:
file_output = file_handle.preferred_filename
yield (0, (
format_hints.Hex(mod.vol.offset),
@@ -58,10 +58,10 @@ class Modules(interfaces.plugins.PluginInterface):
file_output = "Disabled"
if self.config['dump']:
file_handler = dlllist.DllList.dump_pe(self.context, pe_table_name, mod, self._file_handler)
file_handle = dlllist.DllList.dump_pe(self.context, pe_table_name, mod, self._file_handler)
file_output = "Error outputting file"
if file_handler:
file_output = file_handler.preferred_filename
if file_handle:
file_output = file_handle.preferred_filename
yield (0, (
format_hints.Hex(mod.vol.offset),
@@ -59,12 +59,13 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
kernel_table_name: the name for the symbol table containing the kernel's symbols
pe_table_name: the name for the symbol table containing the PE format symbols
proc: the process object whose memory should be output
file_handler: class to write construct for writing the file
file_handler: class to provide context manager for opening the file
Returns:
A FileHandlerInterface object containing the complete data for the process or None in the case of failure
"""
file_handle = None
try:
proc_layer_name = proc.add_process_layer()
peb = context.object(kernel_table_name + constants.BANG + "_PEB",
@@ -74,15 +75,15 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
dos_header = context.object(pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
offset = peb.ImageBaseAddress,
layer_name = proc_layer_name)
file_handler = file_handler("pid.{0}.{1:#x}.dmp".format(proc.UniqueProcessId, peb.ImageBaseAddress))
with file_handler as file_data:
file_handle = file_handler("pid.{0}.{1:#x}.dmp".format(proc.UniqueProcessId, peb.ImageBaseAddress))
with file_handle as file_data:
for offset, data in dos_header.reconstruct():
file_data.seek(offset)
file_data.write(data)
except Exception as excp:
vollog.debug("Unable to dump PE with pid {}: {}".format(proc.UniqueProcessId, excp))
return file_handler
return file_handle
@classmethod
def create_pid_filter(cls, pid_list: List[int] = None) -> Callable[[interfaces.objects.ObjectInterface], bool]:
@@ -190,11 +191,11 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
file_output = "Disabled"
if self.config['dump']:
file_handler = self.process_dump(self.context, self.config['nt_symbols'], pe_table_name, proc,
self._file_handler)
file_handle = self.process_dump(self.context, self.config['nt_symbols'], pe_table_name, proc,
self._file_handler)
file_output = "Error outputting file"
if file_handler:
file_output = file_handler.preferred_filename
if file_handle:
file_output = str(file_handle.preferred_filename)
yield (0, (proc.UniqueProcessId, proc.InheritedFromUniqueProcessId,
proc.ImageFileName.cast("string", max_length = proc.ImageFileName.vol.count, errors = 'replace'),
@@ -82,8 +82,8 @@ class HiveList(interfaces.plugins.PluginInterface):
maxaddr = hive.hive.Storage[0].Length
hive_name = self._sanitize_hive_name(hive.get_name())
file_handler = self.open('registry.{}.{}.hive'.format(hive_name, hex(hive.hive_offset)))
with file_handler as file_data:
file_handle = self.open('registry.{}.{}.hive'.format(hive_name, hex(hive.hive_offset)))
with file_handle as file_data:
if hive._base_block:
hive_data = self.context.layers[hive.dependencies[0]].read(hive.hive.BaseBlock, 1 << 12)
else:
@@ -96,7 +96,7 @@ class HiveList(interfaces.plugins.PluginInterface):
file_data.write(data)
# if self._progress_callback:
# self._progress_callback((i / maxaddr) * 100, 'Writing layer {}'.format(hive_name))
file_output = file_handler.preferred_filename
file_output = file_handle.preferred_filename
yield (0, (format_hints.Hex(hive_object.vol.offset), hive_object.get_name() or "", file_output))
@@ -149,8 +149,8 @@ class VadInfo(interfaces.plugins.PluginInterface):
proc_layer = context.layers[proc_layer_name]
file_name = "pid.{0}.vad.{1:#x}-{2:#x}.dmp".format(proc_id, vad_start, vad_end)
try:
file_handler = file_handler(file_name)
with file_handler as file_data:
file_handle = file_handler(file_name)
with file_handle as file_data:
chunk_size = 1024 * 1024 * 10
offset = vad_start
while offset < vad_end:
@@ -165,7 +165,7 @@ class VadInfo(interfaces.plugins.PluginInterface):
vollog.debug("Unable to dump VAD {}: {}".format(file_name, excp))
return
return file_handler
return file_handle
def _generator(self, procs):
@@ -188,10 +188,10 @@ class VadInfo(interfaces.plugins.PluginInterface):
file_output = "Disabled"
if self.config['dump']:
file_handler = self.vad_dump(self.context, proc, vad, self._file_handler)
file_handle = self.vad_dump(self.context, proc, vad, self._file_handler)
file_output = "Error outputting file"
if file_handler:
file_output = file_handler.preferred_filename
if file_handle:
file_output = file_handle.preferred_filename
yield (0, (proc.UniqueProcessId, process_name, format_hints.Hex(vad.vol.offset),
format_hints.Hex(vad.get_start()), format_hints.Hex(vad.get_end()), vad.get_tag(),