WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-09 19:27:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

linted with ruff (ruff check . --fix)

Browse Source
This commit is contained in:
Arthur Deierlein
2024-12-17 22:55:38 +01:00
parent 8e9b719a16
commit ad7daafc50
84 changed files with 191 additions and 381 deletions
Download Patch File Download Diff File Expand all files Collapse all files
development/compare-vol.py
+1 -1
View File
@@ -101,7 +101,7 @@ class Volatility2Test(VolatilityTest):
print(f" Tested volatility2 imageinfo with image {image.filepath}: {end_time - start_time}")
with open(os.path.join(self.output_directory, f'vol2_imageinfo_{image_hash}_stdout'), "wb") as f:
f.write(vol2_completed.stdout)
image.vol2_profile = re.search(b"Suggested Profile\(s\) : ([^,]+)", vol2_completed.stdout)[1]
image.vol2_profile = re.search(rb"Suggested Profile\(s\) : ([^,]+)", vol2_completed.stdout)[1]
class RekallTest(VolatilityTest):
development/pdbparse-to-json.py
+2 -3
View File
@@ -145,8 +145,7 @@ class PDBConvertor:
"""Generates the metadata necessary for this object"""
dbg = self._pdb.STREAM_DBI
last_bytes = str(binascii.hexlify(self._pdb.STREAM_PDB.GUID.Data4), 'ascii')[-16:]
guidstr = u'{:08x}{:04x}{:04x}{}'.format(self._pdb.STREAM_PDB.GUID.Data1, self._pdb.STREAM_PDB.GUID.Data2,
self._pdb.STREAM_PDB.GUID.Data3, last_bytes)
guidstr = f'{self._pdb.STREAM_PDB.GUID.Data1:08x}{self._pdb.STREAM_PDB.GUID.Data2:04x}{self._pdb.STREAM_PDB.GUID.Data3:04x}{last_bytes}'
pdb_data = {
"GUID": guidstr.upper(),
"age": self._pdb.STREAM_PDB.Age,
@@ -195,7 +194,7 @@ class PDBConvertor:
try:
sects = self._pdb.STREAM_SECT_HDR_ORIG.sections
omap = self._pdb.STREAM_OMAP_FROM_SRC
except AttributeError as e:
except AttributeError:
# In this case there is no OMAP, so we use the given section
# headers and use the identity function for omap.remap
sects = self._pdb.STREAM_SECT_HDR.sections
development/schema_validate.py
+2 -2
View File
@@ -28,7 +28,7 @@ if __name__ == '__main__':
schema = None
if args.schema:
with open(os.path.abspath(args.schema), 'r') as s:
with open(os.path.abspath(args.schema)) as s:
schema = json.load(s)
failures = []
@@ -36,7 +36,7 @@ if __name__ == '__main__':
try:
if os.path.exists(filename):
print(f"[?] Validating file: {filename}")
with open(filename, 'r') as t:
with open(filename) as t:
test = json.load(t)
if args.schema:
volatility3/cli/__init__.py
+7 -15
View File
@@ -57,7 +57,7 @@ formatter = logging.Formatter("%(levelname)-8s %(name)-12s: %(message)s")
console.setFormatter(formatter)
class PrintedProgress(object):
class PrintedProgress:
"""A progress handler that prints the progress value and the description
onto the command line."""
@@ -126,9 +126,7 @@ class CommandLine:
"--help",
action="help",
default=argparse.SUPPRESS,
help="Show this help message and exit, for specific plugin options use '{} <pluginname> --help'".format(
parser.prog
),
help=f"Show this help message and exit, for specific plugin options use '{parser.prog} <pluginname> --help'",
)
parser.add_argument(
"-c",
@@ -360,9 +358,7 @@ class CommandLine:
subparser = parser.add_subparsers(
title="Plugins",
dest="plugin",
description="For plugin specific options, run '{} <plugin> --help'".format(
self.CLI_NAME
),
description=f"For plugin specific options, run '{self.CLI_NAME} <plugin> --help'",
action=volargparse.HelpfulSubparserAction,
metavar="PLUGIN",
)
@@ -416,7 +412,7 @@ class CommandLine:
# UI fills in the config, here we load it from the config file and do it before we process the CL parameters
if args.config:
with open(args.config, "r") as f:
with open(args.config) as f:
json_val = json.load(f)
ctx.config.splice(
plugin_config_path,
@@ -722,9 +718,7 @@ class CommandLine:
if isinstance(requirement, requirements.ListRequirement):
if not isinstance(value, list):
raise TypeError(
"Configuration for ListRequirement was not a list: {}".format(
requirement.name
)
f"Configuration for ListRequirement was not a list: {requirement.name}"
)
value = [requirement.element_type(x) for x in value]
if not inspect.isclass(configurables_list[configurable]):
@@ -797,7 +791,7 @@ class CommandLine:
fd, self._name = tempfile.mkstemp(
suffix=".vol3", prefix="tmp_", dir=output_dir
)
self._file = io.open(fd, mode="w+b")
self._file = open(fd, mode="w+b")
CLIFileHandler.__init__(self, filename)
for item in dir(self._file):
if not item.startswith("_") and item not in (
@@ -870,9 +864,7 @@ class CommandLine:
requirement, interfaces.configuration.RequirementInterface
):
raise TypeError(
"Plugin contains requirements that are not RequirementInterfaces: {}".format(
configurable.__name__
)
f"Plugin contains requirements that are not RequirementInterfaces: {configurable.__name__}"
)
if isinstance(requirement, interfaces.configuration.SimpleTypeRequirement):
additional["type"] = requirement.instance_type
volatility3/cli/text_filter.py
+1 -1
View File
@@ -76,7 +76,7 @@ class ColumnFilter:
if self.regex:
return re.search(self.pattern, f"{item}")
return self.pattern in f"{item}"
except IOError:
except OSError:
return False
def found(self, row: List[Any]) -> bool:
volatility3/cli/text_renderer.py
+1 -1
View File
@@ -467,7 +467,7 @@ class JsonRenderer(CLIRenderer):
def output_result(self, outfd, result):
"""Outputs the JSON data to a file in a particular format"""
outfd.write("{}\n".format(json.dumps(result, indent=2, sort_keys=True)))
outfd.write(f"{json.dumps(result, indent=2, sort_keys=True)}\n")
def render(self, grid: interfaces.renderers.TreeGrid):
outfd = sys.stdout
volatility3/cli/volargparse.py
+1 -1
View File
@@ -5,7 +5,7 @@
import argparse
import gettext
import re
from typing import List, Optional, Sequence, Any, Union
from typing import Optional, Sequence, Any, Union
# This effectively overrides/monkeypatches the core argparse module to provide more helpful output around choices
volatility3/cli/volshell/__init__.py
+2 -4
View File
@@ -282,9 +282,7 @@ class VolShell(cli.CommandLine):
for plugin in volshell_plugin_list:
subparser = parser.add_argument_group(
title=plugin.capitalize(),
description="Configuration options based on {} options".format(
plugin.capitalize()
),
description=f"Configuration options based on {plugin.capitalize()} options",
)
self.populate_requirements_argparse(subparser, volshell_plugin_list[plugin])
configurables_list[plugin] = volshell_plugin_list[plugin]
@@ -331,7 +329,7 @@ class VolShell(cli.CommandLine):
# UI fills in the config, here we load it from the config file and do it before we process the CL parameters
if args.config:
with open(args.config, "r") as f:
with open(args.config) as f:
json_val = json.load(f)
ctx.config.splice(
plugin_config_path,
volatility3/cli/volshell/generic.py
-1
View File
@@ -585,7 +585,6 @@ class NullFileHandler(io.BytesIO, interfaces.plugins.FileHandlerInterface):
def writelines(self, lines: Iterable[bytes]):
"""Dummy method"""
pass
def write(self, b: bytes):
"""Dummy method"""
volatility3/framework/__init__.py
+3 -7
View File
@@ -56,9 +56,7 @@ def require_interface_version(*args) -> None:
if len(args):
if args[0] != interface_version()[0]:
raise RuntimeError(
"Framework interface version {} is incompatible with required version {}".format(
interface_version()[0], args[0]
)
f"Framework interface version {interface_version()[0]} is incompatible with required version {args[0]}"
)
if len(args) > 1:
if args[1] > interface_version()[1]:
@@ -70,7 +68,7 @@ def require_interface_version(*args) -> None:
)
class NonInheritable(object):
class NonInheritable:
def __init__(self, value: Any, cls: Type) -> None:
self.default_value = value
self.cls = cls
@@ -187,9 +185,7 @@ def import_file(module: str, path: str, ignore_errors: bool = False) -> List[str
traceback.TracebackException.from_exception(e).format(chain=True)
)
)
vollog.debug(
"Failed to import module {} based on file: {}".format(module, path)
)
vollog.debug(f"Failed to import module {module} based on file: {path}")
failures.append(module)
if not ignore_errors:
raise
volatility3/framework/automagic/linux.py
+1 -3
View File
@@ -173,9 +173,7 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
if aslr_shift & 0xFFF != 0 or kaslr_shift & 0xFFF != 0:
continue
vollog.debug(
"Linux ASLR shift values determined: physical {:0x} virtual {:0x}".format(
kaslr_shift, aslr_shift
)
f"Linux ASLR shift values determined: physical {kaslr_shift:0x} virtual {aslr_shift:0x}"
)
return kaslr_shift, aslr_shift
volatility3/framework/automagic/mac.py
+1 -1
View File
@@ -184,7 +184,7 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
aslr_shift = 0
for offset, banner in offset_generator:
banner_major, banner_minor = [int(x) for x in banner[22:].split(b".")[0:2]]
banner_major, banner_minor = (int(x) for x in banner[22:].split(b".")[0:2])
tmp_aslr_shift = offset - cls.virtual_to_physical_address(
version_json_address
volatility3/framework/automagic/pdbscan.py
+1 -3
View File
@@ -215,9 +215,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
return (virtual_layer_name, kvo, kernel)
else:
vollog.debug(
"Potential kernel_virtual_offset did not map to expected location: {}".format(
hex(kvo)
)
f"Potential kernel_virtual_offset did not map to expected location: {hex(kvo)}"
)
except exceptions.InvalidAddressException:
vollog.debug(
volatility3/framework/automagic/symbol_cache.py
+1 -8
View File
@@ -106,7 +106,6 @@ class CacheManagerInterface(interfaces.configuration.VersionableInterface):
def add_identifier(self, location: str, operating_system: str, identifier: str):
"""Adds an identifier to the store"""
pass
def find_location(
self, identifier: bytes, operating_system: Optional[str]
@@ -120,18 +119,15 @@ class CacheManagerInterface(interfaces.configuration.VersionableInterface):
Returns:
The location of the symbols file that matches the identifier
"""
pass
def get_local_locations(self) -> Iterable[str]:
"""Returns a list of all the local locations"""
pass
def update(self):
"""Locates all files under the symbol directories. Updates the cache with additions, modifications and removals.
This also updates remote locations based on a cache timeout.
"""
pass
def get_identifier_dictionary(
self, operating_system: Optional[str] = None, local_only: bool = False
@@ -145,15 +141,12 @@ class CacheManagerInterface(interfaces.configuration.VersionableInterface):
Returns:
A dictionary of identifiers mapped to a location
"""
pass
def get_identifier(self, location: str) -> Optional[bytes]:
"""Returns an identifier based on a specific location or None"""
pass
def get_identifiers(self, operating_system: Optional[str]) -> List[bytes]:
"""Returns all identifiers for a particular operating system"""
pass
def get_location_statistics(
self, location: str
@@ -572,6 +565,6 @@ class RemoteIdentifierFormat:
try:
subrbf = RemoteIdentifierFormat(location)
yield from subrbf.process(identifiers, operating_system)
except IOError:
except OSError:
vollog.debug(f"Remote file not found: {location}")
return identifiers
volatility3/framework/automagic/symbol_finder.py
+1 -1
View File
@@ -4,7 +4,7 @@
import logging
import os
from typing import Any, Callable, Iterable, List, Optional, Tuple
from typing import Callable, List, Optional, Tuple
from volatility3.framework import constants, interfaces, layers
from volatility3.framework.automagic import symbol_cache
volatility3/framework/configuration/requirements.py
+1 -3
View File
@@ -664,9 +664,7 @@ class ModuleRequirement(
if value is not None:
vollog.log(
constants.LOGLEVEL_V,
"TypeError - Module Requirement only accepts string labels: {}".format(
repr(value)
),
f"TypeError - Module Requirement only accepts string labels: {repr(value)}",
)
return {config_path: self}
volatility3/framework/contexts/__init__.py
+1 -1
View File
@@ -356,7 +356,7 @@ class SizedModule(Module):
return size or 0
@property # type: ignore # FIXME: mypy #5107
@functools.lru_cache()
@functools.lru_cache
def hash(self) -> str:
"""Hashes the module for equality checks.
volatility3/framework/interfaces/configuration.py
+3 -7
View File
@@ -94,7 +94,7 @@ class HierarchicalDict(collections.abc.Mapping):
raise TypeError(f"Separator must be a one character string: {separator}")
self._separator = separator
self._data: Dict[str, ConfigSimpleType] = {}
self._subdict: Dict[str, "HierarchicalDict"] = {}
self._subdict: Dict[str, HierarchicalDict] = {}
if isinstance(initial_dict, str):
initial_dict = json.loads(initial_dict)
if isinstance(initial_dict, dict):
@@ -182,9 +182,7 @@ class HierarchicalDict(collections.abc.Mapping):
else:
if not isinstance(value, HierarchicalDict):
raise TypeError(
"HierarchicalDicts can only store HierarchicalDicts within their structure: {}".format(
type(value)
)
f"HierarchicalDicts can only store HierarchicalDicts within their structure: {type(value)}"
)
self._subdict[key] = value
@@ -498,9 +496,7 @@ class SimpleTypeRequirement(RequirementInterface):
if not isinstance(value, self.instance_type):
vollog.log(
constants.LOGLEVEL_V,
"TypeError - {} requirements only accept {} type: {}".format(
self.name, self.instance_type.__name__, repr(value)
),
f"TypeError - {self.name} requirements only accept {self.instance_type.__name__} type: {repr(value)}",
)
return {config_path: self}
return {}
volatility3/framework/interfaces/layers.py
+2 -5
View File
@@ -188,7 +188,6 @@ class DataLayerInterface(
the object unreadable (exceptions will be thrown using a
DataLayer after destruction)
"""
pass
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -361,9 +360,7 @@ class DataLayerInterface(
data += self.context.layers[layer_name].read(address, chunk_size)
except exceptions.InvalidAddressException:
vollog.debug(
"Invalid address in layer {} found scanning {} at address {:x}".format(
layer_name, self.name, address
)
f"Invalid address in layer {layer_name} found scanning {self.name} at address {address:x}"
)
if len(data) > scanner.chunk_size + scanner.overlap:
@@ -721,7 +718,7 @@ class LayerContainer(collections.abc.Mapping):
raise NotImplementedError("Cycle checking has not yet been implemented")
class DummyProgress(object):
class DummyProgress:
"""A class to emulate Multiprocessing/threading Value objects."""
def __init__(self) -> None:
volatility3/framework/interfaces/plugins.py
+1 -1
View File
@@ -46,7 +46,7 @@ class FileHandlerInterface(io.RawIOBase):
def preferred_filename(self, filename: str):
"""Sets the preferred filename"""
if self.closed:
raise IOError("FileHandler name cannot be changed once closed")
raise OSError("FileHandler name cannot be changed once closed")
if not isinstance(filename, str):
raise TypeError("FileHandler preferred filenames must be strings")
if os.path.sep in filename:
volatility3/framework/interfaces/renderers.py
+8 -4
View File
@@ -26,7 +26,11 @@ from typing import (
Union,
)
Column = NamedTuple("Column", [("name", str), ("type", Any)])
class Column(NamedTuple):
name: str
type: Any
RenderOption = Any
@@ -98,11 +102,11 @@ class TreeNode(abc.Sequence, metaclass=ABCMeta):
"""
class BaseAbsentValue(object):
class BaseAbsentValue:
"""Class that represents values which are not present for some reason."""
class Disassembly(object):
class Disassembly:
"""A class to indicate that the bytes provided should be disassembled
(based on the architecture)"""
@@ -137,7 +141,7 @@ ColumnsType = List[Tuple[str, BaseTypes]]
VisitorSignature = Callable[[TreeNode, _Type], _Type]
class TreeGrid(object, metaclass=ABCMeta):
class TreeGrid(metaclass=ABCMeta):
"""Class providing the interface for a TreeGrid (which contains TreeNodes)
The structure of a TreeGrid is designed to maintain the structure of the tree in a single object.
volatility3/framework/interfaces/symbols.py
+1 -2
View File
@@ -250,7 +250,6 @@ class BaseSymbolTableInterface:
def clear_symbol_cache(self) -> None:
"""Clears the symbol cache of this symbol table."""
pass
class SymbolSpaceInterface(collections.abc.Mapping):
@@ -378,7 +377,7 @@ class NativeTableInterface(BaseSymbolTableInterface):
return []
class MetadataInterface(object):
class MetadataInterface:
"""Interface for accessing metadata stored within a symbol table."""
def __init__(self, json_data: Dict) -> None:
volatility3/framework/layers/crash.py
+3 -6
View File
@@ -1,7 +1,6 @@
# This file is Copyright 2021 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import contextlib
import logging
import struct
from typing import Tuple, Optional
@@ -138,7 +137,7 @@ class WindowsCrashDump32Layer(segmented.SegmentedLayer):
ulong_bitmap_array = summary_header.get_buffer_long()
# outer_index points to a 32 bits array inside a list of arrays,
# each bit indicating a page mapping state
for outer_index in range(0, ulong_bitmap_array.vol.count):
for outer_index in range(ulong_bitmap_array.vol.count):
ulong_bitmap = ulong_bitmap_array[outer_index]
# All pages in this 32 bits array are mapped (speedup iteration process)
if ulong_bitmap == 0xFFFFFFFF:
@@ -166,7 +165,7 @@ class WindowsCrashDump32Layer(segmented.SegmentedLayer):
seg_first_bit = None
# Some pages in this 32 bits array are mapped and some aren't
else:
for inner_bit_position in range(0, 32):
for inner_bit_position in range(32):
current_bit = outer_index * 32 + inner_bit_position
page_mapped = ulong_bitmap & (1 << inner_bit_position)
if page_mapped:
@@ -220,9 +219,7 @@ class WindowsCrashDump32Layer(segmented.SegmentedLayer):
for idx, (start_position, mapped_offset, length, _) in enumerate(segments):
vollog.log(
constants.LOGLEVEL_VVVV,
"Segment {}: Position {:#x} Offset {:#x} Length {:#x}".format(
idx, start_position, mapped_offset, length
),
f"Segment {idx}: Position {start_position:#x} Offset {mapped_offset:#x} Length {length:#x}",
)
self._segments = segments
volatility3/framework/layers/intel.py
+7 -12
View File
@@ -76,13 +76,13 @@ class Intel(linear.LinearlyMappedLayer):
self._index_shift = math.ceil(math.log2(struct.calcsize(self._entry_format)))
@classproperty
@functools.lru_cache()
@functools.lru_cache
def page_shift(cls) -> int:
"""Page shift for the intel memory layers."""
return cls._page_size_in_bits
@classproperty
@functools.lru_cache()
@functools.lru_cache
def page_size(cls) -> int:
"""Page size for the intel memory layers.
@@ -91,25 +91,25 @@ class Intel(linear.LinearlyMappedLayer):
return 1 << cls._page_size_in_bits
@classproperty
@functools.lru_cache()
@functools.lru_cache
def page_mask(cls) -> int:
"""Page mask for the intel memory layers."""
return ~(cls.page_size - 1)
@classproperty
@functools.lru_cache()
@functools.lru_cache
def bits_per_register(cls) -> int:
"""Returns the bits_per_register to determine the range of an
IntelTranslationLayer."""
return cls._bits_per_register
@classproperty
@functools.lru_cache()
@functools.lru_cache
def minimum_address(cls) -> int:
return 0
@classproperty
@functools.lru_cache()
@functools.lru_cache
def maximum_address(cls) -> int:
return (1 << cls._maxvirtaddr) - 1
@@ -251,12 +251,7 @@ class Intel(linear.LinearlyMappedLayer):
if INTEL_TRANSLATION_DEBUGGING:
vollog.log(
constants.LOGLEVEL_VVVV,
"Entry {} at index {} gives data {} as {}".format(
hex(entry),
hex(index),
hex(struct.unpack(self._entry_format, entry_data)[0]),
name,
),
f"Entry {hex(entry)} at index {hex(index)} gives data {hex(struct.unpack(self._entry_format, entry_data)[0])} as {name}",
)
# Read out the new entry from memory
volatility3/framework/layers/leechcore.py
+1 -1
View File
@@ -48,7 +48,7 @@ if HAS_LEECHCORE:
try:
self._handle = leechcorepyc.LeechCore(self._device)
except TypeError:
raise IOError(f"Unable to open LeechCore device {self._device}")
raise OSError(f"Unable to open LeechCore device {self._device}")
return self._handle
def fileno(self):
volatility3/framework/layers/qemu.py
+1 -1
View File
@@ -236,7 +236,7 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
if self._architecture is None:
vollog.log(
constants.LOGLEVEL_VV,
f"QEVM architecture could not be determined",
"QEVM architecture could not be determined",
)
# Once all segments have been read, determine the PCI hole if any
volatility3/framework/layers/registry.py
+2 -6
View File
@@ -156,9 +156,7 @@ class RegistryHive(linear.LinearlyMappedLayer):
else:
# It doesn't matter that we use KeyNode, we're just after the first two bytes
vollog.debug(
"Unknown Signature {} (0x{:x}) at offset {}".format(
signature, cell.u.KeyNode.Signature, cell_offset
)
f"Unknown Signature {signature} (0x{cell.u.KeyNode.Signature:x}) at offset {cell_offset}"
)
return cell
@@ -178,9 +176,7 @@ class RegistryHive(linear.LinearlyMappedLayer):
if not root_node.vol.type_name.endswith(constants.BANG + "_CM_KEY_NODE"):
raise RegistryFormatException(
self.name,
"Encountered {} instead of _CM_KEY_NODE".format(
root_node.vol.type_name
),
f"Encountered {root_node.vol.type_name} instead of _CM_KEY_NODE",
)
node_key = [root_node]
if key.endswith("\\"):
volatility3/framework/layers/resources.py
+1 -1
View File
@@ -57,7 +57,7 @@ def cascadeCloseFile(new_fp: IO[bytes], original_fp: IO[bytes]) -> IO[bytes]:
return new_fp
class ResourceAccessor(object):
class ResourceAccessor:
"""Object for opening URLs as files (downloading locally first if
necessary)"""
volatility3/framework/layers/scanners/multiregexp.py
+1 -1
View File
@@ -6,7 +6,7 @@ import re
from typing import Generator, List, Tuple
class MultiRegexp(object):
class MultiRegexp:
"""Algorithm for multi-string matching."""
def __init__(self) -> None:
volatility3/framework/objects/utility.py
+2 -2
View File
@@ -22,8 +22,8 @@ def bswap_32(value: int) -> int:
def bswap_64(value: int) -> int:
low = bswap_32((value >> 32))
high = bswap_32((value & 0xFFFFFFFF))
low = bswap_32(value >> 32)
high = bswap_32(value & 0xFFFFFFFF)
return ((high << 32) | low) & 0xFFFFFFFFFFFFFFFF
volatility3/framework/plugins/layerwriter.py
+1 -1
View File
@@ -119,7 +119,7 @@ class LayerWriter(plugins.PluginInterface):
# Update the filename, which may have changed if a file
# with the same name already existed.
output_name = file_handle.preferred_filename
except IOError as excp:
except OSError as excp:
yield 0, (f"Layer cannot be written to {output_name}: {excp}",)
yield 0, (f"Layer has been written to {output_name}",)
volatility3/framework/plugins/linux/check_idt.py
+1 -1
View File
@@ -53,7 +53,7 @@ class Check_idt(interfaces.plugins.PluginInterface):
address_mask = self.context.layers[vmlinux.layer_name].address_mask
# hw handlers + system call
check_idxs = list(range(0, 20)) + [128]
check_idxs = list(range(20)) + [128]
if is_32bit:
if vmlinux.has_type("gate_struct"):
volatility3/framework/plugins/linux/check_syscall.py
+1 -1
View File
@@ -103,7 +103,7 @@ class Check_syscall(plugins.PluginInterface):
try:
func_addr = vmlinux.get_symbol(syscall_entry_func).address
except exceptions.SymbolError as e:
except exceptions.SymbolError:
# if we can't find the disassemble function then bail and rely on a different method
return 0
volatility3/framework/plugins/linux/pagecache.py
+1 -1
View File
@@ -462,7 +462,7 @@ class InodePages(plugins.PluginInterface):
f.seek(current_fp)
f.write(page_bytes)
except IOError as e:
except OSError as e:
vollog.error("Unable to write to file (%s): %s", filename, e)
def _generator(self):
volatility3/framework/plugins/linux/proc.py
+1 -3
View File
@@ -125,9 +125,7 @@ class Maps(plugins.PluginInterface):
proc_layer_name = task.add_process_layer()
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
pid, excp.invalid_address, excp.layer_name
)
f"Process {pid}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
return None
vm_size = vm_end - vm_start
volatility3/framework/plugins/linux/psscan.py
+1 -1
View File
@@ -133,7 +133,7 @@ class PsScan(interfaces.plugins.PluginInterface):
)
elif len(kernel_layer.dependencies) == 0:
vollog.error(
f"Kernel layer has no dependencies, meaning there is no memory layer for this plugin to scan."
"Kernel layer has no dependencies, meaning there is no memory layer for this plugin to scan."
)
raise exceptions.LayerException(
kernel_layer_name, f"Layer {kernel_layer_name} has no dependencies"
volatility3/framework/plugins/mac/proc_maps.py
+1 -3
View File
@@ -115,9 +115,7 @@ class Maps(interfaces.plugins.PluginInterface):
proc_layer_name = task.add_process_layer()
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
pid, excp.invalid_address, excp.layer_name
)
f"Process {pid}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
return None
vm_size = vm_end - vm_start
volatility3/framework/plugins/timeliner.py
+1 -3
View File
@@ -143,9 +143,7 @@ class Timeliner(interfaces.plugins.PluginInterface):
times = self.timeline.get((plugin_name, item), {})
if times.get(timestamp_type, None) is not None:
vollog.debug(
"Multiple timestamps for the same plugin/file combination found: {} {}".format(
plugin_name, item
)
f"Multiple timestamps for the same plugin/file combination found: {plugin_name} {item}"
)
times[timestamp_type] = timestamp
self.timeline[(plugin_name, item)] = times
volatility3/framework/plugins/windows/cmdline.py
+1 -3
View File
@@ -84,9 +84,7 @@ class CmdLine(interfaces.plugins.PluginInterface):
result_text = f"Required memory at {exp.invalid_address:#x} is not valid (process exited?)"
except exceptions.InvalidAddressException as exp:
result_text = "Process {}: Required memory at {:#x} is not valid (incomplete layer {}?)".format(
proc_id, exp.invalid_address, exp.layer_name
)
result_text = f"Process {proc_id}: Required memory at {exp.invalid_address:#x} is not valid (incomplete layer {exp.layer_name}?)"
yield (0, (proc.UniqueProcessId, process_name, result_text))
volatility3/framework/plugins/windows/consoles.py
+4 -18
View File
@@ -95,9 +95,7 @@ class Consoles(interfaces.plugins.PluginInterface):
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
proc_id, excp.invalid_address, excp.layer_name
)
f"Process {proc_id}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
@classmethod
@@ -176,12 +174,7 @@ class Consoles(interfaces.plugins.PluginInterface):
)
vollog.debug(
"Determined OS Version: {}.{} {}.{}".format(
kuser.NtMajorVersion,
kuser.NtMinorVersion,
vers.MajorVersion,
vers.MinorVersion,
)
f"Determined OS Version: {kuser.NtMajorVersion}.{kuser.NtMinorVersion} {vers.MajorVersion}.{vers.MinorVersion}"
)
if nt_major_version == 10 and arch == "x64":
@@ -260,9 +253,7 @@ class Consoles(interfaces.plugins.PluginInterface):
if ver:
conhost_mod_version = ver[3]
vollog.debug(
"Determined conhost.exe's FileVersion: {}".format(
conhost_mod_version
)
f"Determined conhost.exe's FileVersion: {conhost_mod_version}"
)
else:
vollog.debug("Could not determine conhost.exe's FileVersion.")
@@ -311,12 +302,7 @@ class Consoles(interfaces.plugins.PluginInterface):
else:
raise NotImplementedError(
"This version of Windows is not supported: {}.{} {}.{}!".format(
nt_major_version,
nt_minor_version,
vers.MajorVersion,
vers_minor_version,
)
f"This version of Windows is not supported: {nt_major_version}.{nt_minor_version} {vers.MajorVersion}.{vers_minor_version}!"
)
vollog.debug(f"Determined symbol filename: {filename}")
volatility3/framework/plugins/windows/dlllist.py
+3 -12
View File
@@ -5,9 +5,9 @@ import contextlib
import datetime
import logging
import re
from typing import List, Optional, Type
from typing import List
from volatility3.framework import constants, exceptions, interfaces, renderers
from volatility3.framework import exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import conversion, format_hints
from volatility3.framework.symbols import intermed
@@ -199,16 +199,7 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
_depth, row_data = row
if not isinstance(row_data[6], datetime.datetime):
continue
description = (
"DLL Load: Process {} {} Loaded {} ({}) Size {} Offset {}".format(
row_data[0],
row_data[1],
row_data[4],
row_data[5],
row_data[3],
row_data[2],
)
)
description = f"DLL Load: Process {row_data[0]} {row_data[1]} Loaded {row_data[4]} ({row_data[5]}) Size {row_data[3]} Offset {row_data[2]}"
yield (description, timeliner.TimeLinerType.CREATED, row_data[6])
def run(self):
volatility3/framework/plugins/windows/dumpfiles.py
+1 -7
View File
@@ -192,13 +192,7 @@ 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,
)
desired_file_name = f"file.{file_obj.vol.offset:#x}.{memory_object.vol.offset:#x}.{cache_name}.{ntpath.basename(obj_name)}.{extension}"
file_handle = cls.dump_file_producer(
file_obj, memory_object, open_method, layer, desired_file_name
volatility3/framework/plugins/windows/envars.py
+3 -3
View File
@@ -92,7 +92,7 @@ class Envars(interfaces.plugins.PluginInterface):
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
) as excp:
):
vollog.log(
constants.LOGLEVEL_VVV,
"Error while parsing global environment variables keys (some keys might be excluded)",
@@ -113,7 +113,7 @@ class Envars(interfaces.plugins.PluginInterface):
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
) as excp:
):
vollog.log(
constants.LOGLEVEL_VVV,
"Error while parsing user environment variables keys (some keys might be excluded)",
@@ -134,7 +134,7 @@ class Envars(interfaces.plugins.PluginInterface):
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
) as excp:
):
vollog.log(
constants.LOGLEVEL_VVV,
"Error while parsing volatile environment variables keys (some keys might be excluded)",
volatility3/framework/plugins/windows/getservicesids.py
+1 -1
View File
@@ -55,7 +55,7 @@ class GetServiceSIDs(interfaces.plugins.PluginInterface):
)
# Get service sids dictionary (we need only the service sids).
with open(sids_json_file_name, "r") as file_handle:
with open(sids_json_file_name) as file_handle:
self.servicesids = json.load(file_handle)["service sids"]
@classmethod
volatility3/framework/plugins/windows/getsids.py
+3 -3
View File
@@ -58,7 +58,7 @@ class GetSIDs(interfaces.plugins.PluginInterface):
)
# Get all the sids from the json file.
with open(sids_json_file_name, "r") as file_handle:
with open(sids_json_file_name) as file_handle:
sids_json_data = json.load(file_handle)
self.servicesids = sids_json_data["service sids"]
self.well_known_sids = sids_json_data["well known"]
@@ -122,7 +122,7 @@ class GetSIDs(interfaces.plugins.PluginInterface):
except (
exceptions.InvalidAddressException,
layers.registry.RegistryFormatException,
) as excp:
):
continue
try:
value_data = node.decode_data()
@@ -156,7 +156,7 @@ class GetSIDs(interfaces.plugins.PluginInterface):
ValueError,
exceptions.InvalidAddressException,
layers.registry.RegistryFormatException,
) as excp:
):
continue
except (KeyError, exceptions.InvalidAddressException):
continue
volatility3/framework/plugins/windows/hollowprocesses.py
+12 -25
View File
@@ -12,20 +12,15 @@ from volatility3.plugins.windows import pslist, vadinfo
vollog = logging.getLogger(__name__)
VadData = NamedTuple(
"VadData",
[
("protection", str),
("path", str),
],
)
DLLData = NamedTuple(
"DLLData",
[
("path", str),
],
)
class VadData(NamedTuple):
protection: str
path: str
class DLLData(NamedTuple):
path: str
### Useful references on process hollowing
# https://cysinfo.com/detecting-deceptive-hollowing-techniques/
@@ -146,9 +141,7 @@ class HollowProcesses(interfaces.plugins.PluginInterface):
"""
image_base = self._get_image_base(proc)
if image_base is not None and image_base != proc.SectionBaseAddress:
yield "The ImageBaseAddress reported from the PEB ({:#x}) does not match the process SectionBaseAddress ({:#x})".format(
image_base, proc.SectionBaseAddress
)
yield f"The ImageBaseAddress reported from the PEB ({image_base:#x}) does not match the process SectionBaseAddress ({proc.SectionBaseAddress:#x})"
def _check_exe_protection(
self, proc, vads: Dict[int, VadData], __
@@ -166,13 +159,9 @@ class HollowProcesses(interfaces.plugins.PluginInterface):
base = proc.SectionBaseAddress
if base not in vads:
yield "There is no VAD starting at the base address of the process executable ({:#x})".format(
base
)
yield f"There is no VAD starting at the base address of the process executable ({base:#x})"
elif vads[base].protection != "PAGE_EXECUTE_WRITECOPY":
yield "Unexpected protection ({}) for VAD hosting the process executable ({:#x}) with path {}".format(
vads[base].protection, base, vads[base].path
)
yield f"Unexpected protection ({vads[base].protection}) for VAD hosting the process executable ({base:#x}) with path {vads[base].path}"
def _check_dlls_protection(
self, _, vads: Dict[int, VadData], dlls: Dict[int, DLLData]
@@ -184,9 +173,7 @@ class HollowProcesses(interfaces.plugins.PluginInterface):
# PAGE_EXECUTE_WRITECOPY is the only valid permission for mapped DLLs and .exe files
if vads[dll_base].protection != "PAGE_EXECUTE_WRITECOPY":
yield "Unexpected protection ({}) for DLL in the PEB's load order list ({:#x}) with path {}".format(
vads[dll_base].protection, dll_base, dlls[dll_base].path
)
yield f"Unexpected protection ({vads[dll_base].protection}) for DLL in the PEB's load order list ({dll_base:#x}) with path {dlls[dll_base].path}"
def _generator(self, procs):
checks = [
volatility3/framework/plugins/windows/iat.py
+4 -4
View File
@@ -1,7 +1,9 @@
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
import logging, io, pefile
import logging
import io
import pefile
from volatility3.framework.symbols import intermed
from volatility3.framework import renderers, interfaces, exceptions, constants
from volatility3.framework.configuration import requirements
@@ -119,9 +121,7 @@ class IAT(interfaces.plugins.PluginInterface):
)
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
proc_id, excp.invalid_address, excp.layer_name
)
f"Process {proc_id}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
continue
volatility3/framework/plugins/windows/malfind.py
+2 -6
View File
@@ -106,9 +106,7 @@ class Malfind(interfaces.plugins.PluginInterface):
proc_layer_name = proc.add_process_layer()
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
proc_id, excp.invalid_address, excp.layer_name
)
f"Process {proc_id}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
return None
@@ -211,9 +209,7 @@ class Malfind(interfaces.plugins.PluginInterface):
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
)
f"Unable to dump PE with pid {proc.UniqueProcessId}.{vad.get_start():#x}: {excp}"
)
yield (
volatility3/framework/plugins/windows/memmap.py
+2 -8
View File
@@ -53,9 +53,7 @@ class Memmap(interfaces.plugins.PluginInterface):
proc_layer = self.context.layers[proc_layer_name]
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
pid, excp.invalid_address, excp.layer_name
)
f"Process {pid}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
continue
@@ -80,11 +78,7 @@ class Memmap(interfaces.plugins.PluginInterface):
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,
)
f"Unable to write {proc_layer_name}'s address {offset} to {file_handle.preferred_filename}"
)
yield (
volatility3/framework/plugins/windows/modules.py
+1 -3
View File
@@ -177,9 +177,7 @@ class Modules(interfaces.plugins.PluginInterface):
except exceptions.InvalidAddressException:
vollog.log(
constants.LOGLEVEL_VVV,
"Process {} does not have a valid Session or a layer could not be constructed for it".format(
proc_id
),
f"Process {proc_id} does not have a valid Session or a layer could not be constructed for it",
)
continue
volatility3/framework/plugins/windows/netscan.py
+5 -26
View File
@@ -169,12 +169,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
)
vollog.debug(
"Determined OS Version: {}.{} {}.{}".format(
kuser.NtMajorVersion,
kuser.NtMinorVersion,
vers.MajorVersion,
vers.MinorVersion,
)
f"Determined OS Version: {kuser.NtMajorVersion}.{kuser.NtMinorVersion} {vers.MajorVersion}.{vers.MinorVersion}"
)
if nt_major_version == 10 and arch == "x64":
@@ -272,9 +267,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
if ver:
tcpip_mod_version = ver[3]
vollog.debug(
"Determined tcpip.sys's FileVersion: {}".format(
tcpip_mod_version
)
f"Determined tcpip.sys's FileVersion: {tcpip_mod_version}"
)
else:
vollog.debug("Could not determine tcpip.sys's FileVersion.")
@@ -316,12 +309,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
else:
raise NotImplementedError(
"This version of Windows is not supported: {}.{} {}.{}!".format(
nt_major_version,
nt_minor_version,
vers.MajorVersion,
vers_minor_version,
)
f"This version of Windows is not supported: {nt_major_version}.{nt_minor_version} {vers.MajorVersion}.{vers_minor_version}!"
)
vollog.debug(f"Determined symbol filename: {filename}")
@@ -510,17 +498,8 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
for i in row_data
]
description = (
"Network connection: Process {} {} Local Address {}:{} "
"Remote Address {}:{} State {} Protocol {} ".format(
row_data[7],
row_data[8],
row_data[2],
row_data[3],
row_data[4],
row_data[5],
row_data[6],
row_data[1],
)
f"Network connection: Process {row_data[7]} {row_data[8]} Local Address {row_data[2]}:{row_data[3]} "
f"Remote Address {row_data[4]}:{row_data[5]} State {row_data[6]} Protocol {row_data[1]} "
)
yield (description, timeliner.TimeLinerType.CREATED, row_data[9])
volatility3/framework/plugins/windows/netstat.py
+2 -6
View File
@@ -311,9 +311,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
part_table.Partitions.count = part_count
vollog.debug(
"Found TCP connection PartitionTable @ 0x{:x} (partition count: {})".format(
part_table_addr, part_count
)
f"Found TCP connection PartitionTable @ 0x{part_table_addr:x} (partition count: {part_count})"
)
entry_offset = context.symbol_space.get_type(obj_name).relative_child_offset(
"ListEntry"
@@ -624,9 +622,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
proto = "TCPv6"
else:
vollog.debug(
"TCP Endpoint @ 0x{:2x} has unknown address family 0x{:x}".format(
netw_obj.vol.offset, netw_obj.get_address_family()
)
f"TCP Endpoint @ 0x{netw_obj.vol.offset:2x} has unknown address family 0x{netw_obj.get_address_family():x}"
)
proto = "TCPv?"
volatility3/framework/plugins/windows/pe_symbols.py
+1 -1
View File
@@ -645,7 +645,7 @@ class PESymbols(interfaces.plugins.PluginInterface):
and wanted_addresses_identifier not in wanted_symbols
):
vollog.warning(
f"Invalid `wanted_symbols` sent to `find_symbols`. addresses and names keys both misssing."
"Invalid `wanted_symbols` sent to `find_symbols`. addresses and names keys both misssing."
)
return
volatility3/framework/plugins/windows/pedump.py
+24 -30
View File
@@ -64,27 +64,30 @@ class PEDump(interfaces.plugins.PluginInterface):
"""
Returns the filename of the dump file or None
"""
with open_method(file_name) as file_handle:
try:
dos_header = context.object(
pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
offset=base,
layer_name=layer_name,
)
try:
file_handle = open_method(file_name)
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(f"Unable to dump PE file at offset {base}: {excp}")
return None
dos_header = context.object(
pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
offset=base,
layer_name=layer_name,
)
return file_handle.preferred_filename
for offset, data in dos_header.reconstruct():
file_handle.seek(offset)
file_handle.write(data)
except (
OSError,
exceptions.VolatilityException,
OverflowError,
ValueError,
) as excp:
vollog.debug(f"Unable to dump PE file at offset {base}: {excp}")
return None
finally:
file_handle.close()
return file_handle.preferred_filename
@classmethod
def dump_ldr_entry(
@@ -116,12 +119,7 @@ class PEDump(interfaces.plugins.PluginInterface):
if layer_name is None:
layer_name = ldr_entry.vol.layer_name
file_name = "{}{}.{:#x}.{:#x}.dmp".format(
prefix,
ntpath.basename(name),
ldr_entry.vol.offset,
ldr_entry.DllBase,
)
file_name = f"{prefix}{ntpath.basename(name)}.{ldr_entry.vol.offset:#x}.{ldr_entry.DllBase:#x}.dmp"
return cls.dump_pe(
context,
@@ -143,11 +141,7 @@ class PEDump(interfaces.plugins.PluginInterface):
pid: int,
base: int,
) -> Optional[str]:
file_name = "PE.{:#x}.{:d}.{:#x}.dmp".format(
proc_offset,
pid,
base,
)
file_name = f"PE.{proc_offset:#x}.{pid:d}.{base:#x}.dmp"
return PEDump.dump_pe(
context, pe_table_name, layer_name, open_method, file_name, base
volatility3/framework/plugins/windows/privileges.py
+1 -1
View File
@@ -39,7 +39,7 @@ class Privs(interfaces.plugins.PluginInterface):
)
# Get service sids dictionary (we need only the service sids).
with open(sids_json_file_name, "r") as file_handle:
with open(sids_json_file_name) as file_handle:
temp_json = json.load(file_handle)["privileges"]
self.privilege_info = {
int(priv_num): temp_json[priv_num] for priv_num in temp_json
volatility3/framework/plugins/windows/psxview.py
-1
View File
@@ -14,7 +14,6 @@ from volatility3.plugins.windows import (
info,
pslist,
psscan,
sessions,
thrdscan,
)
volatility3/framework/plugins/windows/registry/hivelist.py
+8 -18
View File
@@ -232,10 +232,8 @@ class HiveList(interfaces.plugins.PluginInterface):
for hive in hg:
if hive.vol.offset in seen:
vollog.debug(
"Hivelist found an already seen offset {} while "
"traversing forwards, this should not occur".format(
hex(hive.vol.offset)
)
f"Hivelist found an already seen offset {hex(hive.vol.offset)} while "
"traversing forwards, this should not occur"
)
break
seen.add(hive.vol.offset)
@@ -249,18 +247,14 @@ class HiveList(interfaces.plugins.PluginInterface):
forward_invalid = hg.invalid
if forward_invalid:
vollog.debug(
"Hivelist failed traversing the list forwards at {}, traversing backwards".format(
hex(forward_invalid)
)
f"Hivelist failed traversing the list forwards at {hex(forward_invalid)}, traversing backwards"
)
hg = HiveGenerator(cmhive, forward=False)
for hive in hg:
if hive.vol.offset in seen:
vollog.debug(
"Hivelist found an already seen offset {} while "
"traversing backwards, list walking met in the middle".format(
hex(hive.vol.offset)
)
f"Hivelist found an already seen offset {hex(hive.vol.offset)} while "
"traversing backwards, list walking met in the middle"
)
break
seen.add(hive.vol.offset)
@@ -281,10 +275,8 @@ class HiveList(interfaces.plugins.PluginInterface):
# by walking the list, so revert to scanning, and walk the list forwards and backwards from each
# found hive
vollog.debug(
"Hivelist failed traversing backwards at {}, a different "
"location from forwards, revert to scanning".format(
hex(backward_invalid)
)
f"Hivelist failed traversing backwards at {hex(backward_invalid)}, a different "
"location from forwards, revert to scanning"
)
for hive in hivescan.HiveScan.scan_hives(
context, layer_name, symbol_table
@@ -320,9 +312,7 @@ class HiveList(interfaces.plugins.PluginInterface):
yield linked_hive
except exceptions.InvalidAddressException:
vollog.debug(
"InvalidAddressException when traversing hive {} found from scan, skipping".format(
hex(hive.vol.offset)
)
f"InvalidAddressException when traversing hive {hex(hive.vol.offset)} found from scan, skipping"
)
def run(self) -> renderers.TreeGrid:
volatility3/framework/plugins/windows/registry/userassist.py
+2 -4
View File
@@ -39,7 +39,7 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
os.path.join(os.path.dirname(__file__), "userassist.json"), "rb"
) as fp:
self._folder_guids = json.load(fp)
except IOError:
except OSError:
vollog.error("Usersassist data file not found")
@classmethod
@@ -308,9 +308,7 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
)
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Invalid address identified in lower layer {}: {}".format(
excp.layer_name, excp.invalid_address
)
f"Invalid address identified in lower layer {excp.layer_name}: {excp.invalid_address}"
)
except KeyError:
vollog.debug(
volatility3/framework/plugins/windows/skeleton_key_check.py
+2 -6
View File
@@ -172,9 +172,7 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
except exceptions.InvalidAddressException:
vollog.debug(
"Unable to construct cSystems array at given offset: {:x}".format(
array_start
)
f"Unable to construct cSystems array at given offset: {array_start:x}"
)
array = None
@@ -291,9 +289,7 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
proc_id, excp.invalid_address, excp.layer_name
)
f"Process {proc_id}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
return None, None
volatility3/framework/plugins/windows/strings.py
+1 -3
View File
@@ -170,9 +170,7 @@ class Strings(interfaces.plugins.PluginInterface):
proc_layer_name = process.add_process_layer()
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
proc_id, excp.invalid_address, excp.layer_name
)
f"Process {proc_id}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
continue
volatility3/framework/plugins/windows/svclist.py
+1 -3
View File
@@ -85,9 +85,7 @@ class SvcList(svcscan.SvcScan):
layer_name = proc.add_process_layer()
except exceptions.InvalidAddressException:
vollog.warning(
"Unable to access memory of services.exe running with PID: {}".format(
proc.UniqueProcessId
)
f"Unable to access memory of services.exe running with PID: {proc.UniqueProcessId}"
)
continue
volatility3/framework/plugins/windows/svcscan.py
+4 -10
View File
@@ -26,13 +26,9 @@ from volatility3.plugins.windows.registry import hivelist
vollog = logging.getLogger(__name__)
ServiceBinaryInfo = NamedTuple(
"ServiceBinaryInfo",
[
("dll", Union[str, interfaces.renderers.BaseAbsentValue]),
("binary", Union[str, interfaces.renderers.BaseAbsentValue]),
],
)
class ServiceBinaryInfo(NamedTuple):
dll: Union[str, interfaces.renderers.BaseAbsentValue]
binary: Union[str, interfaces.renderers.BaseAbsentValue]
class SvcScan(interfaces.plugins.PluginInterface):
@@ -306,9 +302,7 @@ class SvcScan(interfaces.plugins.PluginInterface):
proc_layer_name = task.add_process_layer()
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
proc_id, excp.invalid_address, excp.layer_name
)
f"Process {proc_id}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
continue
volatility3/framework/plugins/windows/thrdscan.py
+1 -1
View File
@@ -82,7 +82,7 @@ class ThrdScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface)
ethread.get_exit_time()
) # datetime.datetime object / volatility3.framework.renderers.UnparsableValue object
except exceptions.InvalidAddressException:
vollog.debug("Thread invalid address {:#x}".format(ethread.vol.offset))
vollog.debug(f"Thread invalid address {ethread.vol.offset:#x}")
return None
return (
volatility3/framework/plugins/windows/threads.py
+1 -1
View File
@@ -3,7 +3,7 @@
#
import logging
from typing import Callable, Iterable, List, Generator
from typing import Iterable, List, Generator
from volatility3.framework import interfaces, constants
from volatility3.framework.configuration import requirements
volatility3/framework/plugins/windows/timers.py
+1 -1
View File
@@ -141,7 +141,7 @@ class Timers(interfaces.plugins.PluginInterface):
if dpc.DeferredRoutine == 0:
continue
deferred_routine = dpc.DeferredRoutine
except Exception as e:
except Exception:
continue
module_symbols = list(
volatility3/framework/plugins/windows/unhooked_system_calls.py
+1 -1
View File
@@ -191,7 +191,7 @@ class unhooked_system_calls(interfaces.plugins.PluginInterface):
# gather processes on small_idx since these are the malware infected ones
for pid, pname in cb[small_idx]:
ps.append("{:d}:{}".format(pid, pname))
ps.append(f"{pid:d}:{pname}")
proc_names = ", ".join(ps)
volatility3/framework/plugins/windows/vadinfo.py
+1 -3
View File
@@ -169,9 +169,7 @@ class VadInfo(interfaces.plugins.PluginInterface):
proc_layer_name = proc.add_process_layer()
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
proc_id, excp.invalid_address, excp.layer_name
)
f"Process {proc_id}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
return None
volatility3/framework/plugins/windows/verinfo.py
+1 -3
View File
@@ -212,9 +212,7 @@ class VerInfo(interfaces.plugins.PluginInterface):
proc_layer_name = proc.add_process_layer()
except exceptions.InvalidAddressException as excp:
vollog.debug(
"Process {}: invalid address {} in layer {}".format(
proc_id, excp.invalid_address, excp.layer_name
)
f"Process {proc_id}: invalid address {excp.invalid_address} in layer {excp.layer_name}"
)
continue
volatility3/framework/renderers/__init__.py
+2 -6
View File
@@ -88,9 +88,7 @@ class TreeNode(interfaces.renderers.TreeNode):
val = values[index]
if not isinstance(val, (column.type, interfaces.renderers.BaseAbsentValue)):
raise TypeError(
"Values item with index {} is the wrong type for column {} (got {} but expected {})".format(
index, column.name, type(val), column.type
)
f"Values item with index {index} is the wrong type for column {column.name} (got {type(val)} but expected {column.type})"
)
# TODO: Consider how to deal with timezone naive/aware datetimes (and alert plugin uses to be precise)
# if isinstance(val, datetime.datetime):
@@ -189,9 +187,7 @@ class TreeGrid(interfaces.renderers.TreeGrid):
is_simple_type = issubclass(column_type, self.base_types)
if not is_simple_type:
raise TypeError(
"Column {}'s type is not a simple type: {}".format(
name, column_type.__class__.__name__
)
f"Column {name}'s type is not a simple type: {column_type.__class__.__name__}"
)
converted_columns.append(interfaces.renderers.Column(name, column_type))
self.RowStructure = RowStructureConstructor(
volatility3/framework/symbols/intermed.py
+1 -1
View File
@@ -171,7 +171,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
(indicating that only additive changes have been made) than
the consumer (in this case, the file reader).
"""
major, minor, patch = [int(x) for x in version.split(".")]
major, minor, patch = (int(x) for x in version.split("."))
supported_versions = [x for x in versions if x[0] == major and x[1] >= minor]
if not supported_versions:
raise ValueError(
volatility3/framework/symbols/linux/__init__.py
+1 -1
View File
@@ -798,7 +798,7 @@ class RadixTree(IDStorage):
return True
class PageCache(object):
class PageCache:
"""Linux Page Cache abstraction"""
def __init__(
volatility3/framework/symbols/linux/extensions/__init__.py
+2 -2
View File
@@ -1469,7 +1469,7 @@ class mount(objects.StructType):
def next_peer(self):
table_name = self.vol.type_name.split(constants.BANG)[0]
mount_struct = "{0}{1}mount".format(table_name, constants.BANG)
mount_struct = f"{table_name}{constants.BANG}mount"
offset = self._context.symbol_space.get_type(
mount_struct
).relative_child_offset("mnt_share")
@@ -2487,7 +2487,7 @@ class address_space(objects.StructType):
class page(objects.StructType):
@property
@functools.lru_cache()
@functools.lru_cache
def pageflags_enum(self) -> Dict:
"""Returns 'pageflags' enumeration key/values
volatility3/framework/symbols/windows/extensions/__init__.py
+4 -4
View File
@@ -1081,7 +1081,7 @@ class KTIMER(objects.StructType):
return self.Header.Type in self.VALID_TYPES
def get_due_time(self):
return "{0:#010x}:{1:#010x}".format(self.DueTime.HighPart, self.DueTime.LowPart)
return f"{self.DueTime.HighPart:#010x}:{self.DueTime.LowPart:#010x}"
def get_dpc(self):
"""Return Dpc, and if Windows 7 or later, decode it"""
@@ -1388,7 +1388,7 @@ class SHARED_CACHE_MAP(objects.StructType):
)
# Iterate through the entries
for counter in range(0, self.VACB_ARRAY):
for counter in range(self.VACB_ARRAY):
# Check if the VACB entry is in use
if not vacb_array[counter]:
continue
@@ -1472,7 +1472,7 @@ class SHARED_CACHE_MAP(objects.StructType):
if not section_size > self.VACB_SIZE_OF_FIRST_LEVEL:
array_head = vacb_obj
for counter in range(0, full_blocks):
for counter in range(full_blocks):
vacb_entry = self._context.object(
symbol_table_name + constants.BANG + "pointer",
layer_name=self.vol.layer_name,
@@ -1531,7 +1531,7 @@ class SHARED_CACHE_MAP(objects.StructType):
# 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.
for counter in range(0, self.VACB_ARRAY):
for counter in range(self.VACB_ARRAY):
if not vacb_array[counter]:
continue
volatility3/framework/symbols/windows/extensions/consoles.py
+1 -1
View File
@@ -73,7 +73,7 @@ class ROW(objects.StructType):
)
for i in range(0, len(char_row), 3)
)
except Exception as e:
except Exception:
line = ""
if truncate:
volatility3/framework/symbols/windows/extensions/mbr.py
+1 -6
View File
@@ -8,12 +8,7 @@ from volatility3.framework import objects
class PARTITION_TABLE(objects.StructType):
def get_disk_signature(self) -> str:
"""Get Disk Signature (GUID)."""
return "{0:02x}-{1:02x}-{2:02x}-{3:02x}".format(
self.DiskSignature[0],
self.DiskSignature[1],
self.DiskSignature[2],
self.DiskSignature[3],
)
return f"{self.DiskSignature[0]:02x}-{self.DiskSignature[1]:02x}-{self.DiskSignature[2]:02x}-{self.DiskSignature[3]:02x}"
class PARTITION_ENTRY(objects.StructType):
volatility3/framework/symbols/windows/extensions/network.py
+3 -5
View File
@@ -22,7 +22,7 @@ def inet_ntop(address_family: int, packed_ip: Union[List[int], Array]) -> str:
raise RuntimeError(
"This version of python does not have socket.inet_ntop, please upgrade"
)
raise socket.error("[Errno 97] Address family not supported by protocol")
raise OSError("[Errno 97] Address family not supported by protocol")
# Python's socket.AF_INET6 is 0x1e but Microsoft defines it
@@ -167,11 +167,9 @@ class _TCP_LISTENER(objects.StructType):
def is_valid(self):
try:
if not self.get_address_family() in (AF_INET, AF_INET6):
if self.get_address_family() not 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()
)
f"netw obj 0x{self.vol.offset:x} invalid due to invalid address_family {self.get_address_family()}"
)
return False
volatility3/framework/symbols/windows/extensions/pe.py
+2 -2
View File
@@ -101,9 +101,9 @@ class IMAGE_DOS_HEADER(objects.StructType):
)
except OverflowError:
vollog.warning(
"Volatility was unable to fix the image base for the PE file at base address {:#x}. "
f"Volatility was unable to fix the image base for the PE file at base address {self.vol.offset:#x}. "
"This will cause issues with many static analysis tools if you do not inform the "
"tool of the in-memory load address.".format(self.vol.offset)
"tool of the in-memory load address."
)
new_pe = raw_data
volatility3/framework/symbols/windows/extensions/pool.py
+2 -4
View File
@@ -217,7 +217,7 @@ class POOL_HEADER(objects.StructType):
yield mem_object
@classmethod
@functools.lru_cache()
@functools.lru_cache
def _calculate_optional_header_lengths(
cls, context: interfaces.context.ContextInterface, symbol_table_name: str
) -> Tuple[List[str], List[int]]:
@@ -430,9 +430,7 @@ class OBJECT_HEADER(objects.StructType):
if header_offset == 0:
raise ValueError(
"Could not find _OBJECT_HEADER_NAME_INFO for object at {} of layer {}".format(
self.vol.offset, self.vol.layer_name
)
f"Could not find _OBJECT_HEADER_NAME_INFO for object at {self.vol.offset} of layer {self.vol.layer_name}"
)
header = ntkrnlmp.object(
volatility3/framework/symbols/windows/extensions/registry.py
+1 -3
View File
@@ -196,9 +196,7 @@ class CM_KEY_NODE(objects.StructType):
yield cast("CM_KEY_NODE", node)
else:
vollog.debug(
"Unexpected node type encountered when traversing subkeys: {}, signature: {}".format(
node.vol.type_name, signature
)
f"Unexpected node type encountered when traversing subkeys: {node.vol.type_name}, signature: {signature}"
)
if listjump:
volatility3/framework/symbols/windows/pdbconv.py
+2 -4
View File
@@ -263,9 +263,7 @@ class PdbReader:
)
if header.index_max < header.index_min:
raise ValueError(
"Maximum {} index is smaller than minimum TPI index, found: {} < {} ".format(
stream_name, header.index_max, header.index_min
)
f"Maximum {stream_name} index is smaller than minimum TPI index, found: {header.index_max} < {header.index_min} "
)
# Reset the state
info_references: Dict[str, int] = {}
@@ -976,7 +974,7 @@ class PdbRetreiver:
if __name__ == "__main__":
import argparse
class PrintedProgress(object):
class PrintedProgress:
"""A progress handler that prints the progress value and the
description onto the command line."""
volatility3/framework/symbols/windows/pdbutil.py
+2 -4
View File
@@ -94,7 +94,7 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
if not requirements.VersionRequirement.matches_required(
(1, 0, 0), symbol_cache.SqliteCache.version
):
vollog.debug(f"Required version of SQLiteCache not found")
vollog.debug("Required version of SQLiteCache not found")
return None
identifiers_path = os.path.join(
@@ -291,9 +291,7 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
break
except PermissionError:
vollog.warning(
"Cannot write necessary symbol file, please check permissions on {}".format(
potential_output_filename
)
f"Cannot write necessary symbol file, please check permissions on {potential_output_filename}"
)
continue
finally:
volatility3/plugins/windows/registry/certificates.py
+1 -1
View File
@@ -60,7 +60,7 @@ class Certificates(interfaces.plugins.PluginInterface):
open_method: Type[interfaces.plugins.FileHandlerInterface],
) -> Optional[interfaces.plugins.FileHandlerInterface]:
try:
dump_name = "{}-{}-{}.crt".format(hive_offset, reg_section, key_hash)
dump_name = f"{hive_offset}-{reg_section}-{key_hash}.crt"
file_handle = open_method(dump_name)
file_handle.write(certificate_data)
return file_handle
volatility3/plugins/windows/statistics.py
+1 -3
View File
@@ -64,9 +64,7 @@ class Statistics(plugins.PluginInterface):
other_invalid += 1
page_size = expected_page_size
vollog.debug(
"A non-page lookup invalid address exception occurred at: {} in layer {}".format(
hex(excp.invalid_address), excp.layer_name
)
f"A non-page lookup invalid address exception occurred at: {hex(excp.invalid_address)} in layer {excp.layer_name}"
)
page_addr += page_size
volatility3/schemas/__init__.py
+3 -3
View File
@@ -20,7 +20,7 @@ def load_cached_validations() -> Set[str]:
to revalidate them."""
validhashes: Set = set()
if os.path.exists(cached_validation_filepath):
with open(cached_validation_filepath, "r") as f:
with open(cached_validation_filepath) as f:
validhashes.update(json.load(f))
return validhashes
@@ -46,7 +46,7 @@ def validate(input: Dict[str, Any], use_cache: bool = True) -> bool:
if not os.path.exists(schema_path):
vollog.debug(f"Schema for format not found: {schema_path}")
return False
with open(schema_path, "r") as s:
with open(schema_path) as s:
schema = json.load(s)
return valid(input, schema, use_cache)
@@ -66,7 +66,7 @@ def create_json_hash(
if not os.path.exists(schema_path):
vollog.debug(f"Schema for format not found: {schema_path}")
return None
with open(schema_path, "r") as s:
with open(schema_path) as s:
schema = json.load(s)
return hashlib.sha1(
bytes(json.dumps((input, schema), sort_keys=True), "utf-8")