Plugins: Make open method clearer to use

This highlights that the FileHandler class can also be seen as a method
similar to open, and it removes unnecessary context managers, allowing
plugins to close files as they wish (they must, however, remember to
close the file for it to be committed).
This commit is contained in:
Mike Auty
2020-10-29 09:43:16 +00:00
committed by ikelos
parent 0e596c2112
commit ea629591ef
11 changed files with 75 additions and 71 deletions
+11 -10
View File
@@ -48,7 +48,7 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
context: interfaces.context.ContextInterface,
pe_table_name: str,
dll_entry: interfaces.objects.ObjectInterface,
file_handler: Type[interfaces.plugins.FileHandlerInterface],
open_method: Type[interfaces.plugins.FileHandlerInterface],
layer_name: str = None,
prefix: str = '') -> Optional[interfaces.plugins.FileHandlerInterface]:
"""Extracts the complete data for a process as a FileInterface
@@ -58,10 +58,11 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
pe_table_name: the name for the symbol table containing the PE format symbols
dll_entry: the object representing the module
layer_name: the layer that the DLL lives within
file_handler: class for constructing output files
open_method: class for constructing output files
Returns:
A FileInterface object containing the complete data for the DLL or None in the case of failure"""
An open FileHandlerInterface object containing the complete data for the DLL or None in the case of failure
"""
try:
try:
name = dll_entry.FullDllName.get_string()
@@ -71,17 +72,16 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
if layer_name is None:
layer_name = dll_entry.vol.layer_name
file_handle = file_handler("{}{}.{:#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,
layer_name = layer_name)
with file_handle as file_data:
for offset, data in dos_header.reconstruct():
file_data.seek(offset)
file_data.write(data)
for offset, data in dos_header.reconstruct():
file_handle.seek(offset)
file_handle.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
@@ -129,11 +129,12 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
file_handle = self.dump_pe(self.context,
pe_table_name,
entry,
self._file_handler,
self.open,
proc_layer_name,
prefix = "pid.{}.".format(proc_id))
file_output = "Error outputting file"
if file_handle:
file_handle.close()
file_output = file_handle.preferred_filename
yield (0, (proc.UniqueProcessId,
@@ -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
@@ -135,7 +135,8 @@ class Malfind(interfaces.plugins.PluginInterface):
if self.config['dump']:
file_output = "Error outputting to file"
try:
file_handle = vadinfo.VadInfo.vad_dump(self.context, proc, vad, self._file_handler)
file_handle = vadinfo.VadInfo.vad_dump(self.context, proc, vad, self.open)
file_handle.close()
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(
@@ -58,9 +58,10 @@ class Modules(interfaces.plugins.PluginInterface):
file_output = "Disabled"
if self.config['dump']:
file_handle = 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.open)
file_output = "Error outputting file"
if file_handle:
file_handle.close()
file_output = file_handle.preferred_filename
yield (0, (format_hints.Hex(mod.vol.offset), format_hints.Hex(mod.DllBase),
@@ -49,7 +49,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def process_dump(
cls, context: interfaces.context.ContextInterface, kernel_table_name: str, pe_table_name: str,
proc: interfaces.objects.ObjectInterface,
file_handler: Type[interfaces.plugins.FileHandlerInterface]) -> interfaces.plugins.FileHandlerInterface:
open_method: Type[interfaces.plugins.FileHandlerInterface]) -> interfaces.plugins.FileHandlerInterface:
"""Extracts the complete data for a process as a FileHandlerInterface
Args:
@@ -57,10 +57,10 @@ 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 provide context manager for opening the file
open_method: 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
An open FileHandlerInterface object containing the complete data for the process or None in the case of failure
"""
file_handle = None
@@ -73,11 +73,10 @@ 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_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)
file_handle = open_method("pid.{0}.{1:#x}.dmp".format(proc.UniqueProcessId, peb.ImageBaseAddress))
for offset, data in dos_header.reconstruct():
file_handle.seek(offset)
file_handle.write(data)
except Exception as excp:
vollog.debug("Unable to dump PE with pid {}: {}".format(proc.UniqueProcessId, excp))
@@ -190,9 +189,10 @@ 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._file_handler)
self.open)
file_output = "Error outputting file"
if file_handle:
file_handle.close()
file_output = str(file_handle.preferred_filename)
yield (0, (proc.UniqueProcessId, proc.InheritedFromUniqueProcessId,
@@ -143,7 +143,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
self.config['nt_symbols'], proc)
file_handle = pslist.PsList.process_dump(self.context, self.config['nt_symbols'], pe_table_name,
vproc, self._file_handler)
vproc, self.open)
file_output = "Error outputting file"
if file_handle:
file_output = file_handle.preferred_filename
+17 -15
View File
@@ -112,18 +112,20 @@ class VadInfo(interfaces.plugins.PluginInterface):
context: interfaces.context.ContextInterface,
proc: interfaces.objects.ObjectInterface,
vad: interfaces.objects.ObjectInterface,
file_handler: Type[
interfaces.plugins.FileHandlerInterface]) -> Optional[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.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
proc: an _EPROCESS instance
vad: The suspected VAD to extract (ObjectInterface)
open_method: class to provide context manager for opening the file
maxsize: Max size of VAD section (default MAXSIZE_DEFAULT)
Returns:
A FileInterface object containing the complete data for the process or None in the case of failure
An open FileInterface object containing the complete data for the process or None in the case of failure
"""
try:
@@ -149,17 +151,16 @@ 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_handle = file_handler(file_name)
with file_handle as file_data:
chunk_size = 1024 * 1024 * 10
offset = vad_start
while offset < vad_end:
to_read = min(chunk_size, vad_end - offset)
data = proc_layer.read(offset, to_read, pad = True)
if not data:
break
file_data.write(data)
offset += to_read
file_handle = open_method(file_name)
chunk_size = 1024 * 1024 * 10
offset = vad_start
while offset < vad_end:
to_read = min(chunk_size, vad_end - offset)
data = proc_layer.read(offset, to_read, pad = True)
if not data:
break
file_handle.write(data)
offset += to_read
except Exception as excp:
vollog.debug("Unable to dump VAD {}: {}".format(file_name, excp))
@@ -188,9 +189,10 @@ class VadInfo(interfaces.plugins.PluginInterface):
file_output = "Disabled"
if self.config['dump']:
file_handle = self.vad_dump(self.context, proc, vad, self._file_handler)
file_handle = self.vad_dump(self.context, proc, vad, self.open)
file_output = "Error outputting file"
if file_handle:
file_handle.close()
file_output = file_handle.preferred_filename
yield (0, (proc.UniqueProcessId, process_name, format_hints.Hex(vad.vol.offset),