mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-08-26 16:12:23 +02:00
Core: Apply black 24.1.0 across the whole codebase
This commit is contained in:
@@ -389,9 +389,11 @@ class JsonRenderer(CLIRenderer):
|
||||
interfaces.renderers.Disassembly: quoted_optional(display_disassembly),
|
||||
format_hints.MultiTypeData: quoted_optional(multitypedata_as_text),
|
||||
bytes: optional(lambda x: " ".join([f"{b:02x}" for b in x])),
|
||||
datetime.datetime: lambda x: x.isoformat()
|
||||
if not isinstance(x, interfaces.renderers.BaseAbsentValue)
|
||||
else None,
|
||||
datetime.datetime: lambda x: (
|
||||
x.isoformat()
|
||||
if not isinstance(x, interfaces.renderers.BaseAbsentValue)
|
||||
else None
|
||||
),
|
||||
"default": lambda x: x,
|
||||
}
|
||||
|
||||
|
||||
@@ -206,9 +206,9 @@ def _zipwalk(path: str):
|
||||
if not file.is_dir():
|
||||
dirlist = zip_results.get(os.path.dirname(file.filename), [])
|
||||
dirlist.append(os.path.basename(file.filename))
|
||||
zip_results[
|
||||
os.path.join(path, os.path.dirname(file.filename))
|
||||
] = dirlist
|
||||
zip_results[os.path.join(path, os.path.dirname(file.filename))] = (
|
||||
dirlist
|
||||
)
|
||||
for value in zip_results:
|
||||
yield value, zip_results[value]
|
||||
|
||||
|
||||
@@ -138,9 +138,9 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
|
||||
config_path = join("automagic", "MacIntelHelper", new_layer_name)
|
||||
context.config[join(config_path, "memory_layer")] = layer_name
|
||||
context.config[join(config_path, "page_map_offset")] = dtb
|
||||
context.config[
|
||||
join(config_path, MacSymbolFinder.banner_config_key)
|
||||
] = str(banner, "latin-1")
|
||||
context.config[join(config_path, MacSymbolFinder.banner_config_key)] = (
|
||||
str(banner, "latin-1")
|
||||
)
|
||||
|
||||
new_layer = intel.Intel32e(
|
||||
context,
|
||||
|
||||
@@ -34,9 +34,9 @@ class KernelModule(interfaces.automagic.AutomagicInterface):
|
||||
return None
|
||||
# The requirement is unfulfilled and is a ModuleRequirement
|
||||
|
||||
context.config[
|
||||
interfaces.configuration.path_join(new_config_path, "class")
|
||||
] = "volatility3.framework.contexts.Module"
|
||||
context.config[interfaces.configuration.path_join(new_config_path, "class")] = (
|
||||
"volatility3.framework.contexts.Module"
|
||||
)
|
||||
|
||||
for req in requirement.requirements:
|
||||
if (
|
||||
|
||||
@@ -150,12 +150,12 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
|
||||
clazz = self.symbol_class
|
||||
# Set the discovered options
|
||||
path_join = interfaces.configuration.path_join
|
||||
context.config[
|
||||
path_join(config_path, requirement.name, "class")
|
||||
] = clazz
|
||||
context.config[
|
||||
path_join(config_path, requirement.name, "isf_url")
|
||||
] = isf_path
|
||||
context.config[path_join(config_path, requirement.name, "class")] = (
|
||||
clazz
|
||||
)
|
||||
context.config[path_join(config_path, requirement.name, "isf_url")] = (
|
||||
isf_path
|
||||
)
|
||||
context.config[
|
||||
path_join(config_path, requirement.name, "symbol_mask")
|
||||
] = layer.address_mask
|
||||
|
||||
@@ -402,19 +402,19 @@ class WinSwapLayers(interfaces.automagic.AutomagicInterface):
|
||||
if swap_location:
|
||||
context.config[current_layer_path] = current_layer_name
|
||||
try:
|
||||
context.config[
|
||||
layer_loc_path
|
||||
] = requirements.URIRequirement.location_from_file(
|
||||
swap_location
|
||||
context.config[layer_loc_path] = (
|
||||
requirements.URIRequirement.location_from_file(
|
||||
swap_location
|
||||
)
|
||||
)
|
||||
except ValueError:
|
||||
vollog.warning(
|
||||
f"Volatility swap_location {swap_location} could not be validated - swap layer disabled"
|
||||
)
|
||||
continue
|
||||
context.config[
|
||||
layer_class_path
|
||||
] = "volatility3.framework.layers.physical.FileLayer"
|
||||
context.config[layer_class_path] = (
|
||||
"volatility3.framework.layers.physical.FileLayer"
|
||||
)
|
||||
|
||||
# Add the requirement
|
||||
new_req = requirements.TranslationLayerRequirement(
|
||||
@@ -424,9 +424,9 @@ class WinSwapLayers(interfaces.automagic.AutomagicInterface):
|
||||
)
|
||||
swap_req.add_requirement(new_req)
|
||||
|
||||
context.config[
|
||||
path_join(swap_sub_config, "number_of_elements")
|
||||
] = counter
|
||||
context.config[path_join(swap_sub_config, "number_of_elements")] = (
|
||||
counter
|
||||
)
|
||||
context.config[swap_sub_config] = True
|
||||
|
||||
swap_req.construct(context, swap_config)
|
||||
|
||||
@@ -550,9 +550,9 @@ class VersionRequirement(interfaces.configuration.RequirementInterface):
|
||||
config_path = interfaces.configuration.path_join(config_path, self.name)
|
||||
if not self.matches_required(self._version, self._component.version):
|
||||
return {config_path: self}
|
||||
context.config[
|
||||
interfaces.configuration.path_join(config_path, self.name)
|
||||
] = True
|
||||
context.config[interfaces.configuration.path_join(config_path, self.name)] = (
|
||||
True
|
||||
)
|
||||
return {}
|
||||
|
||||
@classmethod
|
||||
|
||||
@@ -224,7 +224,7 @@ class AVMLStacker(interfaces.automagic.StackerLayerInterface):
|
||||
except exceptions.LayerException:
|
||||
return None
|
||||
new_name = context.layers.free_layer_name("AVMLLayer")
|
||||
context.config[
|
||||
interfaces.configuration.path_join(new_name, "base_layer")
|
||||
] = layer_name
|
||||
context.config[interfaces.configuration.path_join(new_name, "base_layer")] = (
|
||||
layer_name
|
||||
)
|
||||
return AVMLLayer(context, new_name, new_name)
|
||||
|
||||
@@ -115,9 +115,9 @@ class Elf64Stacker(interfaces.automagic.StackerLayerInterface):
|
||||
vollog.log(constants.LOGLEVEL_VVVV, f"Exception: {excp}")
|
||||
return None
|
||||
new_name = context.layers.free_layer_name("Elf64Layer")
|
||||
context.config[
|
||||
interfaces.configuration.path_join(new_name, "base_layer")
|
||||
] = layer_name
|
||||
context.config[interfaces.configuration.path_join(new_name, "base_layer")] = (
|
||||
layer_name
|
||||
)
|
||||
|
||||
try:
|
||||
return Elf64Layer(context, new_name, new_name)
|
||||
|
||||
@@ -277,9 +277,9 @@ class Intel(linear.LinearlyMappedLayer):
|
||||
This allows translation layers to provide maps of contiguous
|
||||
regions in one layer
|
||||
"""
|
||||
stashed_offset = (
|
||||
stashed_mapped_offset
|
||||
) = stashed_size = stashed_mapped_size = stashed_map_layer = None
|
||||
stashed_offset = stashed_mapped_offset = stashed_size = stashed_mapped_size = (
|
||||
stashed_map_layer
|
||||
) = None
|
||||
for offset, size, mapped_offset, mapped_size, map_layer in self._mapping(
|
||||
offset, length, ignore_errors
|
||||
):
|
||||
|
||||
@@ -104,7 +104,7 @@ class LimeStacker(interfaces.automagic.StackerLayerInterface):
|
||||
except LimeFormatException:
|
||||
return None
|
||||
new_name = context.layers.free_layer_name("LimeLayer")
|
||||
context.config[
|
||||
interfaces.configuration.path_join(new_name, "base_layer")
|
||||
] = layer_name
|
||||
context.config[interfaces.configuration.path_join(new_name, "base_layer")] = (
|
||||
layer_name
|
||||
)
|
||||
return LimeLayer(context, new_name, new_name)
|
||||
|
||||
@@ -486,9 +486,9 @@ class QemuStacker(interfaces.automagic.StackerLayerInterface):
|
||||
except exceptions.LayerException:
|
||||
return None
|
||||
new_name = context.layers.free_layer_name("QemuSuspendLayer")
|
||||
context.config[
|
||||
interfaces.configuration.path_join(new_name, "base_layer")
|
||||
] = layer_name
|
||||
context.config[interfaces.configuration.path_join(new_name, "base_layer")] = (
|
||||
layer_name
|
||||
)
|
||||
layer = QemuSuspendLayer(context, new_name, new_name)
|
||||
cls.stacker_slow_warning()
|
||||
return layer
|
||||
|
||||
@@ -173,8 +173,8 @@ class XenCoreDumpStacker(elf.Elf64Stacker):
|
||||
vollog.log(constants.LOGLEVEL_VVVV, f"Exception: {excp}")
|
||||
return None
|
||||
new_name = context.layers.free_layer_name("XenCoreDumpLayer")
|
||||
context.config[
|
||||
interfaces.configuration.path_join(new_name, "base_layer")
|
||||
] = layer_name
|
||||
context.config[interfaces.configuration.path_join(new_name, "base_layer")] = (
|
||||
layer_name
|
||||
)
|
||||
|
||||
return XenCoreDumpLayer(context, new_name, new_name)
|
||||
|
||||
@@ -768,12 +768,10 @@ class Array(interfaces.objects.ObjectInterface, collections.abc.Sequence):
|
||||
raise IndexError(f"Member not present in array template: {child}")
|
||||
|
||||
@overload
|
||||
def __getitem__(self, i: int) -> interfaces.objects.Template:
|
||||
...
|
||||
def __getitem__(self, i: int) -> interfaces.objects.Template: ...
|
||||
|
||||
@overload
|
||||
def __getitem__(self, s: slice) -> List[interfaces.objects.Template]:
|
||||
...
|
||||
def __getitem__(self, s: slice) -> List[interfaces.objects.Template]: ...
|
||||
|
||||
def __getitem__(self, i):
|
||||
"""Returns the i-th item from the array."""
|
||||
|
||||
@@ -49,9 +49,7 @@ class PsList(interfaces.plugins.PluginInterface):
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def get_list_tasks(
|
||||
cls, method: str
|
||||
) -> Callable[
|
||||
def get_list_tasks(cls, method: str) -> Callable[
|
||||
[interfaces.context.ContextInterface, str, Callable[[int], bool]],
|
||||
Iterable[interfaces.objects.ObjectInterface],
|
||||
]:
|
||||
|
||||
@@ -46,9 +46,9 @@ class Crashinfo(interfaces.plugins.PluginInterface):
|
||||
bitmap_size = format_hints.Hex(summary_header.BitmapSize)
|
||||
bitmap_pages = format_hints.Hex(summary_header.Pages)
|
||||
else:
|
||||
bitmap_header_size = (
|
||||
bitmap_size
|
||||
) = bitmap_pages = renderers.NotApplicableValue()
|
||||
bitmap_header_size = bitmap_size = bitmap_pages = (
|
||||
renderers.NotApplicableValue()
|
||||
)
|
||||
|
||||
yield (
|
||||
0,
|
||||
|
||||
@@ -176,7 +176,6 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
|
||||
|
||||
class ADS(interfaces.plugins.PluginInterface):
|
||||
|
||||
"""Scans for Alternate Data Stream"""
|
||||
|
||||
_required_framework_version = (2, 0, 0)
|
||||
|
||||
@@ -487,10 +487,12 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
if not isinstance(row_data[9], datetime.datetime):
|
||||
continue
|
||||
row_data = [
|
||||
"N/A"
|
||||
if isinstance(i, renderers.UnreadableValue)
|
||||
or isinstance(i, renderers.UnparsableValue)
|
||||
else i
|
||||
(
|
||||
"N/A"
|
||||
if isinstance(i, renderers.UnreadableValue)
|
||||
or isinstance(i, renderers.UnparsableValue)
|
||||
else i
|
||||
)
|
||||
for i in row_data
|
||||
]
|
||||
description = (
|
||||
|
||||
@@ -193,9 +193,9 @@ class PrintKey(interfaces.plugins.PluginInterface):
|
||||
vollog.debug(
|
||||
"Couldn't read registry value type, so data is unreadable"
|
||||
)
|
||||
value_data: Union[
|
||||
interfaces.renderers.BaseAbsentValue, bytes
|
||||
] = renderers.UnreadableValue()
|
||||
value_data: Union[interfaces.renderers.BaseAbsentValue, bytes] = (
|
||||
renderers.UnreadableValue()
|
||||
)
|
||||
else:
|
||||
try:
|
||||
value_data = node.decode_data()
|
||||
|
||||
@@ -28,9 +28,9 @@ def wintime_to_datetime(
|
||||
def unixtime_to_datetime(
|
||||
unixtime: int,
|
||||
) -> Union[interfaces.renderers.BaseAbsentValue, datetime.datetime]:
|
||||
ret: Union[
|
||||
interfaces.renderers.BaseAbsentValue, datetime.datetime
|
||||
] = renderers.UnparsableValue()
|
||||
ret: Union[interfaces.renderers.BaseAbsentValue, datetime.datetime] = (
|
||||
renderers.UnparsableValue()
|
||||
)
|
||||
|
||||
if unixtime > 0:
|
||||
with contextlib.suppress(ValueError):
|
||||
|
||||
@@ -35,9 +35,9 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self._dict: Dict[
|
||||
str, interfaces.symbols.BaseSymbolTableInterface
|
||||
] = collections.OrderedDict()
|
||||
self._dict: Dict[str, interfaces.symbols.BaseSymbolTableInterface] = (
|
||||
collections.OrderedDict()
|
||||
)
|
||||
# Permanently cache all resolved symbols
|
||||
self._resolved: Dict[str, interfaces.objects.Template] = {}
|
||||
self._resolved_symbols: Dict[str, interfaces.objects.Template] = {}
|
||||
@@ -73,9 +73,9 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
|
||||
self, offset: int, size: int = 0, table_name: str = None
|
||||
) -> Iterable[str]:
|
||||
"""Returns all symbols that exist at a specific relative address."""
|
||||
table_list: Iterable[
|
||||
interfaces.symbols.BaseSymbolTableInterface
|
||||
] = self._dict.values()
|
||||
table_list: Iterable[interfaces.symbols.BaseSymbolTableInterface] = (
|
||||
self._dict.values()
|
||||
)
|
||||
if table_name is not None:
|
||||
if table_name in self._dict:
|
||||
table_list = [self._dict[table_name]]
|
||||
@@ -179,15 +179,15 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
|
||||
if child.vol.type_name not in self._resolved:
|
||||
traverse_list.append(child.vol.type_name)
|
||||
try:
|
||||
self._resolved[
|
||||
child.vol.type_name
|
||||
] = self._weak_resolve(
|
||||
SymbolType.TYPE, child.vol.type_name
|
||||
self._resolved[child.vol.type_name] = (
|
||||
self._weak_resolve(
|
||||
SymbolType.TYPE, child.vol.type_name
|
||||
)
|
||||
)
|
||||
except exceptions.SymbolError:
|
||||
self._resolved[
|
||||
child.vol.type_name
|
||||
] = self.UnresolvedTemplate(child.vol.type_name)
|
||||
self._resolved[child.vol.type_name] = (
|
||||
self.UnresolvedTemplate(child.vol.type_name)
|
||||
)
|
||||
# Stash the replacement
|
||||
replacements.add((traverser, child))
|
||||
elif child.children:
|
||||
|
||||
@@ -452,9 +452,9 @@ class FILE_OBJECT(objects.StructType, pool.ExecutiveObject):
|
||||
].is_valid(self.FileName.Buffer)
|
||||
|
||||
def file_name_with_device(self) -> Union[str, interfaces.renderers.BaseAbsentValue]:
|
||||
name: Union[
|
||||
str, interfaces.renderers.BaseAbsentValue
|
||||
] = renderers.UnreadableValue()
|
||||
name: Union[str, interfaces.renderers.BaseAbsentValue] = (
|
||||
renderers.UnreadableValue()
|
||||
)
|
||||
|
||||
# this pointer needs to be checked against native_layer_name because the object may
|
||||
# be instantiated from a primary (virtual) layer or a memory (physical) layer.
|
||||
|
||||
@@ -31,13 +31,9 @@ class Statistics(plugins.PluginInterface):
|
||||
# Do mass mapping and determine the number of different layers and how many pages go to each one
|
||||
layer = self.context.layers[self.config["primary"]]
|
||||
|
||||
page_count = (
|
||||
swap_count
|
||||
) = (
|
||||
invalid_page_count
|
||||
) = (
|
||||
large_page_count
|
||||
) = large_swap_count = large_invalid_count = other_invalid = 0
|
||||
page_count = swap_count = invalid_page_count = large_page_count = (
|
||||
large_swap_count
|
||||
) = large_invalid_count = other_invalid = 0
|
||||
|
||||
if isinstance(layer, intel.Intel):
|
||||
page_addr = 0
|
||||
|
||||
Reference in New Issue
Block a user