mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-13 21:27:39 +02:00
Codebase: Apply yapf clean-up across the codebase
This commit is contained in:
@@ -459,7 +459,6 @@ class CommandLine:
|
||||
|
||||
def _get_final_filename(self):
|
||||
"""Gets the final filename"""
|
||||
|
||||
if output_dir is None:
|
||||
raise TypeError("Output directory is not a string")
|
||||
os.makedirs(output_dir, exist_ok = True)
|
||||
|
||||
@@ -44,7 +44,7 @@ def available(context: interfaces.context.ContextInterface) -> List[interfaces.a
|
||||
clazz(context, interfaces.configuration.path_join(config_path, clazz.__name__))
|
||||
for clazz in class_subclasses(interfaces.automagic.AutomagicInterface)
|
||||
],
|
||||
key = lambda x: x.priority)
|
||||
key = lambda x: x.priority)
|
||||
|
||||
|
||||
def choose_automagic(
|
||||
|
||||
@@ -26,9 +26,9 @@ class FrameworkInfo(plugins.PluginInterface):
|
||||
}
|
||||
|
||||
for category, module_interface in categories.items():
|
||||
yield (0, (category,))
|
||||
yield (0, (category, ))
|
||||
for clazz in framework.class_subclasses(module_interface):
|
||||
yield (1, (clazz.__name__,))
|
||||
yield (1, (clazz.__name__, ))
|
||||
|
||||
def run(self):
|
||||
return renderers.TreeGrid([("Data", str)], self._generator())
|
||||
|
||||
@@ -41,7 +41,7 @@ class PsList(interfaces.plugins.PluginInterface):
|
||||
|
||||
@classmethod
|
||||
def get_list_tasks(
|
||||
cls, method: str
|
||||
cls, method: str
|
||||
) -> Callable[[interfaces.context.ContextInterface, str, str, Callable[[int], bool]],
|
||||
Iterable[interfaces.objects.ObjectInterface]]:
|
||||
"""Returns the list_tasks method based on the selector
|
||||
|
||||
@@ -145,13 +145,12 @@ class Timeliner(interfaces.plugins.PluginInterface):
|
||||
# Body format is: MD5|name|inode|mode_as_string|UID|GID|size|atime|mtime|ctime|crtime
|
||||
|
||||
if self._any_time_present(times):
|
||||
fp.write(
|
||||
"|{} - {}||||||{}|{}|{}|{}\n".format(
|
||||
plugin_name, self._sanitize_body_format(item),
|
||||
self._text_format(times.get(TimeLinerType.ACCESSED, "")),
|
||||
self._text_format(times.get(TimeLinerType.MODIFIED, "")),
|
||||
self._text_format(times.get(TimeLinerType.CHANGED, "")),
|
||||
self._text_format(times.get(TimeLinerType.CREATED, ""))))
|
||||
fp.write("|{} - {}||||||{}|{}|{}|{}\n".format(
|
||||
plugin_name, self._sanitize_body_format(item),
|
||||
self._text_format(times.get(TimeLinerType.ACCESSED, "")),
|
||||
self._text_format(times.get(TimeLinerType.MODIFIED, "")),
|
||||
self._text_format(times.get(TimeLinerType.CHANGED, "")),
|
||||
self._text_format(times.get(TimeLinerType.CREATED, ""))))
|
||||
|
||||
def _sanitize_body_format(self, value):
|
||||
return value.replace("|", "_")
|
||||
|
||||
@@ -71,10 +71,8 @@ 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 = 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,
|
||||
@@ -128,8 +126,12 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
|
||||
file_output = "Disabled"
|
||||
if self.config['dump']:
|
||||
file_handle = 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_handle:
|
||||
file_output = file_handle.preferred_filename
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -138,16 +138,16 @@ class Malfind(interfaces.plugins.PluginInterface):
|
||||
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))
|
||||
vollog.debug("Unable to dump PE with pid {0}.{1:#x}: {2}".format(
|
||||
proc.UniqueProcessId, vad.get_start(), excp))
|
||||
|
||||
yield (0, (proc.UniqueProcessId, process_name, format_hints.Hex(vad.get_start()),
|
||||
format_hints.Hex(vad.get_end()), vad.get_tag(),
|
||||
vad.get_protection(
|
||||
vadinfo.VadInfo.protect_values(self.context, self.config["primary"],
|
||||
self.config["nt_symbols"]), vadinfo.winnt_protections),
|
||||
vad.get_commit_charge(), vad.get_private_memory(), file_output, format_hints.HexBytes(data),
|
||||
disasm))
|
||||
self.config["nt_symbols"]),
|
||||
vadinfo.winnt_protections), vad.get_commit_charge(), vad.get_private_memory(),
|
||||
file_output, format_hints.HexBytes(data), disasm))
|
||||
|
||||
def run(self):
|
||||
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
|
||||
|
||||
@@ -62,25 +62,20 @@ class Memmap(interfaces.plugins.PluginInterface):
|
||||
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_handle.preferred_filename))
|
||||
vollog.debug("Unable to write {}'s address {} to {}".format(
|
||||
proc_layer_name, offset, file_handle.preferred_filename))
|
||||
|
||||
yield (0, (
|
||||
format_hints.Hex(offset),
|
||||
format_hints.Hex(mapped_offset),
|
||||
format_hints.Hex(mapped_size),
|
||||
format_hints.Hex(offset),
|
||||
file_output))
|
||||
yield (0, (format_hints.Hex(offset), format_hints.Hex(mapped_offset), format_hints.Hex(mapped_size),
|
||||
format_hints.Hex(offset), file_output))
|
||||
offset += mapped_size
|
||||
|
||||
def run(self):
|
||||
filter_func = pslist.PsList.create_pid_filter([self.config.get('pid', None)])
|
||||
|
||||
return renderers.TreeGrid(
|
||||
[("Virtual", format_hints.Hex), ("Physical", format_hints.Hex), ("Size", format_hints.Hex),
|
||||
("Offset", format_hints.Hex), ("File output", str)],
|
||||
self._generator(
|
||||
pslist.PsList.list_processes(context = self.context,
|
||||
layer_name = self.config['primary'],
|
||||
symbol_table = self.config['nt_symbols'],
|
||||
filter_func = filter_func)))
|
||||
return renderers.TreeGrid([("Virtual", format_hints.Hex), ("Physical", format_hints.Hex),
|
||||
("Size", format_hints.Hex), ("Offset", format_hints.Hex), ("File output", str)],
|
||||
self._generator(
|
||||
pslist.PsList.list_processes(context = self.context,
|
||||
layer_name = self.config['primary'],
|
||||
symbol_table = self.config['nt_symbols'],
|
||||
filter_func = filter_func)))
|
||||
|
||||
@@ -63,14 +63,8 @@ class Modules(interfaces.plugins.PluginInterface):
|
||||
if file_handle:
|
||||
file_output = file_handle.preferred_filename
|
||||
|
||||
yield (0, (
|
||||
format_hints.Hex(mod.vol.offset),
|
||||
format_hints.Hex(mod.DllBase),
|
||||
format_hints.Hex(mod.SizeOfImage),
|
||||
BaseDllName,
|
||||
FullDllName,
|
||||
file_output
|
||||
))
|
||||
yield (0, (format_hints.Hex(mod.vol.offset), format_hints.Hex(mod.DllBase),
|
||||
format_hints.Hex(mod.SizeOfImage), BaseDllName, FullDllName, file_output))
|
||||
|
||||
@classmethod
|
||||
def get_session_layers(cls,
|
||||
|
||||
@@ -46,12 +46,10 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
]
|
||||
|
||||
@classmethod
|
||||
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:
|
||||
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:
|
||||
"""Extracts the complete data for a process as a FileHandlerInterface
|
||||
|
||||
Args:
|
||||
|
||||
@@ -57,8 +57,8 @@ class Certificates(interfaces.plugins.PluginInterface):
|
||||
key_hash = key_path[key_path.rindex("\\") + 1:]
|
||||
|
||||
if not isinstance(certificate_data, interfaces.renderers.BaseAbsentValue):
|
||||
with self.open("{} - {} - {}.crt".format(
|
||||
hex(hive.hive_offset), reg_section, key_hash)) as file_data:
|
||||
with self.open("{} - {} - {}.crt".format(hex(hive.hive_offset), reg_section,
|
||||
key_hash)) as file_data:
|
||||
file_data.write(certificate_data)
|
||||
yield (0, (top_key, reg_section, key_hash, name))
|
||||
except KeyError:
|
||||
|
||||
Reference in New Issue
Block a user