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

Core: Convert to format strings across the whole base

Browse Source
This commit is contained in:
Mike Auty
2021-07-18 15:53:58 +01:00
parent 99e959a15a
commit 9f52a15734
92 changed files with 386 additions and 386 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/cli/__init__.py
+30 -30
View File
@@ -56,7 +56,7 @@ class PrintedProgress(object):
Args:
progress: Percentage of progress of the current procedure
"""
message = "\rProgress: {0: 7.2f}\t\t{1:}".format(round(progress, 2), description or '')
message = f"\rProgress: {round(progress, 2): 7.2f}\t\t{description or ''}"
message_len = len(message)
self._max_message_len = max([self._max_message_len, message_len])
sys.stderr.write(message + (' ' * (self._max_message_len - message_len)) + '\r')
@@ -144,7 +144,7 @@ class CommandLine:
parser.add_argument("-r",
"--renderer",
metavar = 'RENDERER',
help = "Determines how to render the output ({})".format(", ".join(list(renderers))),
help = f"Determines how to render the output ({', '.join(list(renderers))})",
default = "quick",
choices = list(renderers))
parser.add_argument("-f",
@@ -162,7 +162,7 @@ class CommandLine:
default = False,
action = 'store_true')
parser.add_argument("--cache-path",
help = "Change the default path ({}) used to store the cache".format(constants.CACHE_PATH),
help = f"Change the default path ({constants.CACHE_PATH}) used to store the cache",
default = constants.CACHE_PATH,
type = str)
@@ -174,7 +174,7 @@ class CommandLine:
banner_output = sys.stdout
if renderers[partial_args.renderer].structured_output:
banner_output = sys.stderr
banner_output.write("Volatility 3 Framework {}\n".format(constants.PACKAGE_VERSION))
banner_output.write(f"Volatility 3 Framework {constants.PACKAGE_VERSION}\n")
if partial_args.plugin_dirs:
volatility3.plugins.__path__ = [os.path.abspath(p)
@@ -202,8 +202,8 @@ class CommandLine:
else:
console.setLevel(10 - (partial_args.verbosity - 2))
vollog.info("Volatility plugins path: {}".format(volatility3.plugins.__path__))
vollog.info("Volatility symbols path: {}".format(volatility3.symbols.__path__))
vollog.info(f"Volatility plugins path: {volatility3.plugins.__path__}")
vollog.info(f"Volatility symbols path: {volatility3.symbols.__path__}")
# Set the PARALLELISM
if partial_args.parallelism == 'processes':
@@ -256,7 +256,7 @@ class CommandLine:
if args.plugin is None:
parser.error("Please select a plugin to run")
vollog.log(constants.LOGLEVEL_VVV, "Cache directory used: {}".format(constants.CACHE_PATH))
vollog.log(constants.LOGLEVEL_VVV, f"Cache directory used: {constants.CACHE_PATH}")
plugin = plugin_list[args.plugin]
chosen_configurables_list[args.plugin] = plugin
@@ -289,7 +289,7 @@ class CommandLine:
ctx.config['automagic.LayerStacker.stackers'] = stacker.choose_os_stackers(plugin)
self.output_dir = args.output_dir
if not os.path.exists(self.output_dir):
parser.error("The output directory specified does not exist: {}".format(self.output_dir))
parser.error(f"The output directory specified does not exist: {self.output_dir}")
self.populate_config(ctx, chosen_configurables_list, args, plugin_config_path)
@@ -318,7 +318,7 @@ class CommandLine:
json.dump(dict(constructed.build_configuration()), f, sort_keys = True, indent = 2)
except exceptions.UnsatisfiedException as excp:
self.process_unsatisfied_exceptions(excp)
parser.exit(1, "Unable to validate the plugin requirements: {}\n".format([x for x in excp.unsatisfied]))
parser.exit(1, f"Unable to validate the plugin requirements: {[x for x in excp.unsatisfied]}\n")
try:
# Construct and run the plugin
@@ -346,7 +346,7 @@ class CommandLine:
filename = request.url2pathname(single_location.path)
if not filename:
raise ValueError("File URL looks incorrect (potentially missing /)")
raise ValueError("File does not exist: {}".format(filename))
raise ValueError(f"File does not exist: {filename}")
return parse.urlunparse(single_location)
def process_exceptions(self, excp):
@@ -363,20 +363,20 @@ class CommandLine:
if isinstance(excp, exceptions.InvalidAddressException):
general = "Volatility was unable to read a requested page:"
if isinstance(excp, exceptions.SwappedInvalidAddressException):
detail = "Swap error {} in layer {} ({})".format(hex(excp.invalid_address), excp.layer_name, excp)
detail = f"Swap error {hex(excp.invalid_address)} in layer {excp.layer_name} ({excp})"
caused_by = [
"No suitable swap file having been provided (locate and provide the correct swap file)",
"An intentionally invalid page (operating system protection)"
]
elif isinstance(excp, exceptions.PagedInvalidAddressException):
detail = "Page error {} in layer {} ({})".format(hex(excp.invalid_address), excp.layer_name, excp)
detail = f"Page error {hex(excp.invalid_address)} in layer {excp.layer_name} ({excp})"
caused_by = [
"Memory smear during acquisition (try re-acquiring if possible)",
"An intentionally invalid page lookup (operating system protection)",
"A bug in the plugin/volatility3 (re-run with -vvv and file a bug)"
]
else:
detail = "{} in layer {} ({})".format(hex(excp.invalid_address), excp.layer_name, excp)
detail = f"{hex(excp.invalid_address)} in layer {excp.layer_name} ({excp})"
caused_by = [
"The base memory file being incomplete (try re-acquiring if possible)",
"Memory smear during acquisition (try re-acquiring if possible)",
@@ -385,7 +385,7 @@ class CommandLine:
]
elif isinstance(excp, exceptions.SymbolError):
general = "Volatility experienced a symbol-related issue:"
detail = "{}{}{}: {}".format(excp.table_name, constants.BANG, excp.symbol_name, excp)
detail = f"{excp.table_name}{constants.BANG}{excp.symbol_name}: {excp}"
caused_by = [
"An invalid symbol table",
"A plugin requesting a bad symbol",
@@ -393,32 +393,32 @@ class CommandLine:
]
elif isinstance(excp, exceptions.SymbolSpaceError):
general = "Volatility experienced an issue related to a symbol table:"
detail = "{}".format(excp)
detail = f"{excp}"
caused_by = [
"An invalid symbol table", "A plugin requesting a bad symbol",
"A plugin requesting a symbol from the wrong table"
]
elif isinstance(excp, exceptions.LayerException):
general = "Volatility experienced a layer-related issue: {}".format(excp.layer_name)
detail = "{}".format(excp)
general = f"Volatility experienced a layer-related issue: {excp.layer_name}"
detail = f"{excp}"
caused_by = ["A faulty layer implementation (re-run with -vvv and file a bug)"]
elif isinstance(excp, exceptions.MissingModuleException):
general = "Volatility could not import a necessary module: {}".format(excp.module)
detail = "{}".format(excp)
general = f"Volatility could not import a necessary module: {excp.module}"
detail = f"{excp}"
caused_by = ["A required python module is not installed (install the module and re-run)"]
else:
general = "Volatilty encountered an unexpected situation."
detail = ""
caused_by = [
"Please re-run using with -vvv and file a bug with the output", "at {}".format(constants.BUG_URL)
"Please re-run using with -vvv and file a bug with the output", f"at {constants.BUG_URL}"
]
# Code that actually renders the exception
output = sys.stderr
output.write(general + "\n")
output.write(detail + "\n\n")
output.write(f"{general}\n")
output.write(f"{detail}\n\n")
for cause in caused_by:
output.write("\t* " + cause + "\n")
output.write(f" * {cause}\n")
output.write("\nNo further results will be produced\n")
sys.exit(1)
@@ -434,7 +434,7 @@ class CommandLine:
symbols_failed = symbols_failed or isinstance(excp.unsatisfied[config_path],
configuration.requirements.SymbolTableRequirement)
print("Unsatisfied requirement {}: {}".format(config_path, excp.unsatisfied[config_path].description))
print(f"Unsatisfied requirement {config_path}: {excp.unsatisfied[config_path].description}")
if symbols_failed:
print("\nA symbol table requirement was not fulfilled. Please verify that:\n"
@@ -471,8 +471,8 @@ class CommandLine:
if not scheme or len(scheme) <= 1:
if not os.path.exists(value):
raise FileNotFoundError(
"Non-existant file {} passed to URIRequirement".format(value))
value = "file://" + request.pathname2url(os.path.abspath(value))
f"Non-existant file {value} passed to URIRequirement")
value = f"file://{request.pathname2url(os.path.abspath(value))}"
if isinstance(requirement, requirements.ListRequirement):
if not isinstance(value, list):
raise TypeError("Configuration for ListRequirement was not a list: {}".format(
@@ -499,11 +499,11 @@ class CommandLine:
pref_name_array = self.preferred_filename.split('.')
filename, extension = os.path.join(output_dir, '.'.join(pref_name_array[:-1])), pref_name_array[-1]
output_filename = "{}.{}".format(filename, extension)
output_filename = f"{filename}.{extension}"
counter = 1
while os.path.exists(output_filename):
output_filename = "{}-{}.{}".format(filename, counter, extension)
output_filename = f"{filename}-{counter}.{extension}"
counter += 1
return output_filename
@@ -525,7 +525,7 @@ class CommandLine:
with open(output_filename, "wb") as current_file:
current_file.write(self.read())
self._committed = True
vollog.log(logging.INFO, "Saved stored plugin file: {}".format(output_filename))
vollog.log(logging.INFO, f"Saved stored plugin file: {output_filename}")
super().close()
@@ -578,7 +578,7 @@ class CommandLine:
configurable: The plugin object to pull the requirements from
"""
if not issubclass(configurable, interfaces.configuration.ConfigurableInterface):
raise TypeError("Expected ConfigurableInterface type, not: {}".format(type(configurable)))
raise TypeError(f"Expected ConfigurableInterface type, not: {type(configurable)}")
# Construct an argparse group
volatility3/cli/text_renderer.py
+19 -19
View File
@@ -33,13 +33,13 @@ def hex_bytes_as_text(value: bytes) -> str:
A text representation of the hexadecimal bytes plus their ascii equivalents, separated by newline characters
"""
if not isinstance(value, bytes):
raise TypeError("hex_bytes_as_text takes bytes not: {}".format(type(value)))
raise TypeError(f"hex_bytes_as_text takes bytes not: {type(value)}")
ascii = []
hex = []
count = 0
output = ""
for byte in value:
hex.append("{:02x}".format(byte))
hex.append(f"{byte:02x}")
ascii.append(chr(byte) if 0x20 < byte <= 0x7E else ".")
if (count % 8) == 7:
output += "\n"
@@ -87,10 +87,10 @@ def quoted_optional(func: Callable) -> Callable:
if result == "-" or result == "N/A":
return ""
if isinstance(x, format_hints.MultiTypeData) and x.converted_int:
return "{}".format(result)
return f"{result}"
if isinstance(x, int) and not isinstance(x, (format_hints.Hex, format_hints.Bin)):
return "{}".format(result)
return "\"{}\"".format(result)
return f"{result}"
return f"\"{result}\""
return wrapped
@@ -115,7 +115,7 @@ def display_disassembly(disasm: interfaces.renderers.Disassembly) -> str:
output = ""
if disasm.architecture is not None:
for i in disasm_types[disasm.architecture].disasm(disasm.data, disasm.offset):
output += "\n0x%x:\t%s\t%s" % (i.address, i.mnemonic, i.op_str)
output += f"\n0x{i.address:x}:\t{i.mnemonic}\t{i.op_str}"
return output
return QuickTextRenderer._type_renderers[bytes](disasm.data)
@@ -128,14 +128,14 @@ class CLIRenderer(interfaces.renderers.Renderer):
class QuickTextRenderer(CLIRenderer):
_type_renderers = {
format_hints.Bin: optional(lambda x: "0b{:b}".format(x)),
format_hints.Hex: optional(lambda x: "0x{:x}".format(x)),
format_hints.Bin: optional(lambda x: f"0b{x:b}"),
format_hints.Hex: optional(lambda x: f"0x{x:x}"),
format_hints.HexBytes: optional(hex_bytes_as_text),
format_hints.MultiTypeData: quoted_optional(multitypedata_as_text),
interfaces.renderers.Disassembly: optional(display_disassembly),
bytes: optional(lambda x: " ".join(["{0:02x}".format(b) for b in x])),
bytes: optional(lambda x: " ".join([f"{b:02x}" for b in x])),
datetime.datetime: optional(lambda x: x.strftime("%Y-%m-%d %H:%M:%S.%f %Z")),
'default': optional(lambda x: "{}".format(x))
'default': optional(lambda x: f"{x}")
}
name = "quick"
@@ -158,7 +158,7 @@ class QuickTextRenderer(CLIRenderer):
line = []
for column in grid.columns:
# Ignore the type because namedtuples don't realize they have accessible attributes
line.append("{}".format(column.name))
line.append(f"{column.name}")
outfd.write("\n{}\n".format("\t".join(line)))
def visitor(node: interfaces.renderers.TreeNode, accumulator):
@@ -184,14 +184,14 @@ class QuickTextRenderer(CLIRenderer):
class CSVRenderer(CLIRenderer):
_type_renderers = {
format_hints.Bin: quoted_optional(lambda x: "0b{:b}".format(x)),
format_hints.Hex: quoted_optional(lambda x: "0x{:x}".format(x)),
format_hints.Bin: quoted_optional(lambda x: f"0b{x:b}"),
format_hints.Hex: quoted_optional(lambda x: f"0x{x:x}"),
format_hints.HexBytes: quoted_optional(hex_bytes_as_text),
format_hints.MultiTypeData: quoted_optional(multitypedata_as_text),
interfaces.renderers.Disassembly: quoted_optional(display_disassembly),
bytes: quoted_optional(lambda x: " ".join(["{0:02x}".format(b) for b in x])),
bytes: quoted_optional(lambda x: " ".join([f"{b:02x}" for b in x])),
datetime.datetime: quoted_optional(lambda x: x.strftime("%Y-%m-%d %H:%M:%S.%f %Z")),
'default': quoted_optional(lambda x: "{}".format(x))
'default': quoted_optional(lambda x: f"{x}")
}
name = "csv"
@@ -212,7 +212,7 @@ class CSVRenderer(CLIRenderer):
for column in grid.columns:
# Ignore the type because namedtuples don't realize they have accessible attributes
line.append("{}".format('"' + column.name + '"'))
outfd.write("{}".format(",".join(line)))
outfd.write(f"{','.join(line)}")
def visitor(node: interfaces.renderers.TreeNode, accumulator):
accumulator.write("\n")
@@ -223,7 +223,7 @@ class CSVRenderer(CLIRenderer):
column = grid.columns[column_index]
renderer = self._type_renderers.get(column.type, self._type_renderers['default'])
line.append(renderer(node.values[column_index]))
accumulator.write("{}".format(",".join(line)))
accumulator.write(f"{','.join(line)}")
return accumulator
if not grid.populated:
@@ -273,7 +273,7 @@ class PrettyTextRenderer(CLIRenderer):
renderer = self._type_renderers.get(column.type, self._type_renderers['default'])
data = renderer(node.values[column_index])
max_column_widths[column.name] = max(max_column_widths.get(column.name, len(column.name)),
len("{}".format(data)))
len(f"{data}"))
line[column] = data
accumulator.append((node.path_depth, line))
return accumulator
@@ -304,7 +304,7 @@ class JsonRenderer(CLIRenderer):
format_hints.HexBytes: quoted_optional(hex_bytes_as_text),
interfaces.renderers.Disassembly: quoted_optional(display_disassembly),
format_hints.MultiTypeData: quoted_optional(multitypedata_as_text),
bytes: optional(lambda x: " ".join(["{0:02x}".format(b) for b in x])),
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,
'default': lambda x: x
}
volatility3/cli/volargparse.py
+3 -3
View File
@@ -46,10 +46,10 @@ class HelpfulSubparserAction(argparse._SubParsersAction):
matched_parsers = [name for name in self._name_parser_map if parser_name in name]
if len(matched_parsers) < 1:
msg = 'invalid choice {} (choose from {})'.format(parser_name, ', '.join(self._name_parser_map))
msg = f"invalid choice {parser_name} (choose from {', '.join(self._name_parser_map)})"
raise argparse.ArgumentError(self, msg)
if len(matched_parsers) > 1:
msg = 'plugin {} matches multiple plugins ({})'.format(parser_name, ', '.join(matched_parsers))
msg = f"plugin {parser_name} matches multiple plugins ({', '.join(matched_parsers)})"
raise argparse.ArgumentError(self, msg)
parser = self._name_parser_map[matched_parsers[0]]
setattr(namespace, 'plugin', matched_parsers[0])
@@ -88,7 +88,7 @@ class HelpfulArgParser(argparse.ArgumentParser):
if msg is None:
msg = gettext.ngettext('expected %s argument', 'expected %s arguments', action.nargs) % action.nargs
if action.choices:
msg = "{} (from: {})".format(msg, ", ".join(action.choices))
msg = f"{msg} (from: {', '.join(action.choices)})"
raise argparse.ArgumentError(action, msg)
# return the number of arguments matched
volatility3/cli/volshell/__init__.py
+6 -6
View File
@@ -41,7 +41,7 @@ class VolShell(cli.CommandLine):
def run(self):
"""Executes the command line module, taking the system arguments,
determining the plugin to run and then running it."""
sys.stdout.write("Volshell (Volatility 3 Framework) {}\n".format(constants.PACKAGE_VERSION))
sys.stdout.write(f"Volshell (Volatility 3 Framework) {constants.PACKAGE_VERSION}\n")
framework.require_interface_version(1, 0, 0)
@@ -90,7 +90,7 @@ class VolShell(cli.CommandLine):
default = False,
action = 'store_true')
parser.add_argument("--cache-path",
help = "Change the default path ({}) used to store the cache".format(constants.CACHE_PATH),
help = f"Change the default path ({constants.CACHE_PATH}) used to store the cache",
default = constants.CACHE_PATH,
type = str)
@@ -119,8 +119,8 @@ class VolShell(cli.CommandLine):
if partial_args.cache_path:
constants.CACHE_PATH = partial_args.cache_path
vollog.info("Volatility plugins path: {}".format(volatility3.plugins.__path__))
vollog.info("Volatility symbols path: {}".format(volatility3.symbols.__path__))
vollog.info(f"Volatility plugins path: {volatility3.plugins.__path__}")
vollog.info(f"Volatility symbols path: {volatility3.symbols.__path__}")
if partial_args.log:
file_logger = logging.FileHandler(partial_args.log)
@@ -180,7 +180,7 @@ class VolShell(cli.CommandLine):
# Run the argparser
args = parser.parse_args()
vollog.log(constants.LOGLEVEL_VVV, "Cache directory used: {}".format(constants.CACHE_PATH))
vollog.log(constants.LOGLEVEL_VVV, f"Cache directory used: {constants.CACHE_PATH}")
plugin = generic.Volshell
if args.windows:
@@ -243,7 +243,7 @@ class VolShell(cli.CommandLine):
constructed.run()
except exceptions.VolatilityException as excp:
self.process_exceptions(excp)
parser.exit(1, "Unable to validate the plugin requirements: {}\n".format([x for x in excp.unsatisfied]))
parser.exit(1, f"Unable to validate the plugin requirements: {[x for x in excp.unsatisfied]}\n")
def main():
volatility3/cli/volshell/generic.py
+13 -13
View File
@@ -76,14 +76,14 @@ class Volshell(interfaces.plugins.PluginInterface):
mode = self.__module__.split('.')[-1]
mode = mode[0].upper() + mode[1:]
banner = """
banner = f"""
Call help() to see available functions
Volshell mode: {}
Current Layer: {}
""".format(mode, self.current_layer)
Volshell mode: {mode}
Current Layer: {self.current_layer}
"""
sys.ps1 = "({}) >>> ".format(self.current_layer)
sys.ps1 = f"({self.current_layer}) >>> "
self.__console = code.InteractiveConsole(locals = self._construct_locals_dict())
# Since we have to do work to add the option only once for all different modes of volshell, we can't
# rely on the default having been set
@@ -105,14 +105,14 @@ class Volshell(interfaces.plugins.PluginInterface):
for aliases, item in self.construct_locals():
name = ", ".join(aliases)
if item.__doc__ and callable(item):
print("* {}".format(name))
print(" {}".format(item.__doc__))
print(f"* {name}")
print(f" {item.__doc__}")
else:
variables.append(name)
print("\nVariables:")
for var in variables:
print(" {}".format(var))
print(f" {var}")
def construct_locals(self) -> List[Tuple[List[str], Any]]:
"""Returns a dictionary listing the functions to be added to the
@@ -181,7 +181,7 @@ class Volshell(interfaces.plugins.PluginInterface):
if not layer_name:
layer_name = self.config['primary']
self.__current_layer = layer_name
sys.ps1 = "({}) >>> ".format(self.current_layer)
sys.ps1 = f"({self.current_layer}) >>> "
def display_bytes(self, offset, count = 128, layer_name = None):
"""Displays byte values and ASCII characters"""
@@ -221,7 +221,7 @@ class Volshell(interfaces.plugins.PluginInterface):
}
if architecture is not None:
for i in disasm_types[architecture].disasm(remaining_data, offset):
print("0x%x:\t%s\t%s" % (i.address, i.mnemonic, i.op_str))
print(f"0x{i.address:x}:\t{i.mnemonic}\t{i.op_str}")
def display_type(self,
object: Union[str, interfaces.objects.ObjectInterface, interfaces.objects.Template],
@@ -245,7 +245,7 @@ class Volshell(interfaces.plugins.PluginInterface):
volobject = self.context.object(volobject.vol.type_name, layer_name = self.current_layer, offset = offset)
if hasattr(volobject.vol, 'size'):
print("{} ({} bytes)".format(volobject.vol.type_name, volobject.vol.size))
print(f"{volobject.vol.type_name} ({volobject.vol.size} bytes)")
elif hasattr(volobject.vol, 'data_format'):
data_format = volobject.vol.data_format
print("{} ({} bytes, {} endian, {})".format(volobject.vol.type_name, data_format.length,
@@ -301,7 +301,7 @@ class Volshell(interfaces.plugins.PluginInterface):
constructed = plugins.construct_plugin(self.context, [], plugin, plugin_path, None, NullFileHandler)
return constructed.run()
except exceptions.UnsatisfiedException as excp:
print("Unable to validate the plugin requirements: {}\n".format([x for x in excp.unsatisfied]))
print(f"Unable to validate the plugin requirements: {[x for x in excp.unsatisfied]}\n")
return None
def render_treegrid(self,
@@ -340,7 +340,7 @@ class Volshell(interfaces.plugins.PluginInterface):
"""Runs a python script within the context of volshell"""
if not parse.urlparse(location).scheme:
location = "file:" + request.pathname2url(location)
print("Running code from {}\n".format(location))
print(f"Running code from {location}\n")
accessor = resources.ResourceAccessor()
with io.TextIOWrapper(accessor.open(url = location), encoding = 'utf-8') as fp:
self.__console.runsource(fp.read(), symbol = 'exec')
volatility3/cli/volshell/linux.py
+2 -2
View File
@@ -30,9 +30,9 @@ class Volshell(generic.Volshell):
if process_layer is not None:
self.change_layer(process_layer)
return
print("Layer for task ID {} could not be constructed".format(pid))
print(f"Layer for task ID {pid} could not be constructed")
return
print("No task with task ID {} found".format(pid))
print(f"No task with task ID {pid} found")
def list_tasks(self):
"""Returns a list of task objects from the primary layer"""
volatility3/cli/volshell/mac.py
+2 -2
View File
@@ -30,9 +30,9 @@ class Volshell(generic.Volshell):
if process_layer is not None:
self.change_layer(process_layer)
return
print("Layer for task ID {} could not be constructed".format(pid))
print(f"Layer for task ID {pid} could not be constructed")
return
print("No task with task ID {} found".format(pid))
print(f"No task with task ID {pid} found")
def list_tasks(self):
"""Returns a list of task objects from the primary layer"""
volatility3/cli/volshell/windows.py
+1 -1
View File
@@ -29,7 +29,7 @@ class Volshell(generic.Volshell):
process_layer = process.add_process_layer()
self.change_layer(process_layer)
return
print("No process with process ID {} found".format(pid))
print(f"No process with process ID {pid} found")
def list_processes(self):
"""Returns a list of EPROCESS objects from the primary layer"""
volatility3/framework/__init__.py
+2 -2
View File
@@ -77,7 +77,7 @@ T = TypeVar('T')
def class_subclasses(cls: Type[T]) -> Generator[Type[T], None, None]:
"""Returns all the (recursive) subclasses of a given class."""
if not inspect.isclass(cls):
raise TypeError("class_subclasses parameter not a valid class: {}".format(cls))
raise TypeError(f"class_subclasses parameter not a valid class: {cls}")
for clazz in cls.__subclasses__():
# The typing system is not clever enough to realize that clazz has a hidden attr after the hasattr check
if not hasattr(clazz, 'hidden') or not clazz.hidden: # type: ignore
@@ -92,7 +92,7 @@ def import_files(base_module, ignore_errors = False) -> List[str]:
if not isinstance(base_module.__path__, list):
raise TypeError("[base_module].__path__ must be a list of paths")
vollog.log(constants.LOGLEVEL_VVVV,
"Importing from the following paths: {}".format(", ".join(base_module.__path__)))
f"Importing from the following paths: {', '.join(base_module.__path__)}")
for path in base_module.__path__:
for root, _, files in os.walk(path, followlinks = True):
# TODO: Figure out how to import pycache files
volatility3/framework/automagic/__init__.py
+2 -2
View File
@@ -72,7 +72,7 @@ def choose_automagic(
vollog.info("No plugin category detected")
return automagics
vollog.info("Detected a {} category plugin".format(plugin_category))
vollog.info(f"Detected a {plugin_category} category plugin")
output = []
for amagic in automagics:
if amagic.__class__.__name__ in automagic_categories[plugin_category]:
@@ -127,7 +127,7 @@ def run(automagics: List[interfaces.automagic.AutomagicInterface],
for automagic in automagics:
try:
vollog.info("Running automagic: {}".format(automagic.__class__.__name__))
vollog.info(f"Running automagic: {automagic.__class__.__name__}")
automagic(context, config_path, requirement, progress_callback)
except Exception as excp:
exceptions.append(traceback.TracebackException.from_exception(excp))
volatility3/framework/automagic/construct_layers.py
+2 -2
View File
@@ -48,12 +48,12 @@ class ConstructionMagic(interfaces.automagic.AutomagicInterface):
self(context, subreq_config_path, subreq, optional = optional or subreq.optional)
except Exception as e:
# We don't really care if this fails, it tends to mean the configuration isn't complete for that item
vollog.log(constants.LOGLEVEL_VVVV, "Construction Exception occurred: {}".format(e))
vollog.log(constants.LOGLEVEL_VVVV, f"Construction Exception occurred: {e}")
invalid = subreq.unsatisfied(context, subreq_config_path)
# We want to traverse optional paths, so don't check until we've tried to validate
# We also don't want to emit a debug message when a parent is optional, hence the optional parameter
if invalid and not (optional or subreq.optional):
vollog.log(constants.LOGLEVEL_V, "Failed on requirement: {}".format(subreq_config_path))
vollog.log(constants.LOGLEVEL_V, f"Failed on requirement: {subreq_config_path}")
result.append(interfaces.configuration.path_join(subreq_config_path, subreq.name))
if result:
return result
volatility3/framework/automagic/linux.py
+2 -2
View File
@@ -41,7 +41,7 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
mss = scanners.MultiStringScanner([x for x in linux_banners if x is not None])
for _, banner in layer.scan(context = context, scanner = mss, progress_callback = progress_callback):
dtb = None
vollog.debug("Identified banner: {}".format(repr(banner)))
vollog.debug(f"Identified banner: {repr(banner)}")
symbol_files = linux_banners.get(banner, None)
if symbol_files:
@@ -82,7 +82,7 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
metadata = {'kaslr_value': aslr_shift, 'os': 'Linux'})
if layer and dtb:
vollog.debug("DTB was found at: 0x{:0x}".format(dtb))
vollog.debug(f"DTB was found at: 0x{dtb:0x}")
return layer
vollog.debug("No suitable linux banner could be matched")
return None
volatility3/framework/automagic/mac.py
+5 -5
View File
@@ -44,7 +44,7 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
for banner_offset, banner in layer.scan(context = context, scanner = mss,
progress_callback = progress_callback):
dtb = None
vollog.debug("Identified banner: {}".format(repr(banner)))
vollog.debug(f"Identified banner: {repr(banner)}")
symbol_files = mac_banners.get(banner, None)
if symbol_files:
@@ -63,7 +63,7 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
progress_callback = progress_callback)
if kaslr_shift == 0:
vollog.log(constants.LOGLEVEL_VVV, "Invalid kalsr_shift found at offset: {}".format(banner_offset))
vollog.log(constants.LOGLEVEL_VVV, f"Invalid kalsr_shift found at offset: {banner_offset}")
continue
bootpml4_addr = cls.virtual_to_physical_address(table.get_symbol("BootPML4").address + kaslr_shift)
@@ -90,7 +90,7 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
tmp_dtb = idlepml4_addr
if tmp_dtb % 4096:
vollog.log(constants.LOGLEVEL_VVV, "Skipping non-page aligned DTB: 0x{:0x}".format(tmp_dtb))
vollog.log(constants.LOGLEVEL_VVV, f"Skipping non-page aligned DTB: 0x{tmp_dtb:0x}")
continue
dtb = tmp_dtb
@@ -108,7 +108,7 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
metadata = {'kaslr_value': kaslr_shift})
if new_layer and dtb:
vollog.debug("DTB was found at: 0x{:0x}".format(dtb))
vollog.debug(f"DTB was found at: 0x{dtb:0x}")
return new_layer
vollog.debug("No suitable mac banner could be matched")
return None
@@ -164,7 +164,7 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
aslr_shift = tmp_aslr_shift & 0xffffffff
break
vollog.log(constants.LOGLEVEL_VVVV, "Mac find_aslr returned: {:0x}".format(aslr_shift))
vollog.log(constants.LOGLEVEL_VVVV, f"Mac find_aslr returned: {aslr_shift:0x}")
return aslr_shift
volatility3/framework/automagic/pdbscan.py
+2 -2
View File
@@ -127,7 +127,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
kvo_path = interfaces.configuration.path_join(context.layers[virtual_layer].config_path,
'kernel_virtual_offset')
context.config[kvo_path] = kvo
vollog.debug("Setting kernel_virtual_offset to {}".format(hex(kvo)))
vollog.debug(f"Setting kernel_virtual_offset to {hex(kvo)}")
def get_physical_layer_name(self, context, vlayer):
return context.config.get(interfaces.configuration.path_join(vlayer.config_path, 'memory_layer'), None)
@@ -166,7 +166,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
vollog.debug("Potential kernel_virtual_offset did not map to expected location: {}".format(
hex(kvo)))
except exceptions.InvalidAddressException:
vollog.debug("Potential kernel_virtual_offset caused a page fault: {}".format(hex(kvo)))
vollog.debug(f"Potential kernel_virtual_offset caused a page fault: {hex(kvo)}")
vollog.debug("Kernel base determination - testing fixed base address")
return self._method_layer_pdb_scan(context, vlayer, test_physical_kernel, True, progress_callback)
volatility3/framework/automagic/stacker.py
+6 -6
View File
@@ -58,7 +58,7 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
# Bow out quickly if the UI hasn't provided a single_location
unsatisfied = self.unsatisfied(self.context, self.config_path)
if unsatisfied:
vollog.info("Unable to run LayerStacker, unsatisfied requirement: {}".format(unsatisfied))
vollog.info(f"Unable to run LayerStacker, unsatisfied requirement: {unsatisfied}")
return list(unsatisfied)
if not self.config or not self.config.get('single_location', None):
raise ValueError("Unable to run LayerStacker, single_location parameter not provided")
@@ -123,7 +123,7 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
# Stash the changed config items
self._cached = context.config.get(path, None), context.config.branch(path)
vollog.debug("Stacked layers: {}".format(stacked_layers))
vollog.debug(f"Stacked layers: {stacked_layers}")
@classmethod
def stack_layer(cls,
@@ -158,7 +158,7 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
for stacker_item in stack_set:
if not issubclass(stacker_item, interfaces.automagic.StackerLayerInterface):
raise TypeError("Stacker {} is not a descendent of StackerLayerInterface".format(stacker_item.__name__))
raise TypeError(f"Stacker {stacker_item.__name__} is not a descendent of StackerLayerInterface")
while stacked:
stacked = False
@@ -167,17 +167,17 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
for stacker_cls in stack_set:
stacker = stacker_cls()
try:
vollog.log(constants.LOGLEVEL_VV, "Attempting to stack using {}".format(stacker_cls.__name__))
vollog.log(constants.LOGLEVEL_VV, f"Attempting to stack using {stacker_cls.__name__}")
new_layer = stacker.stack(context, initial_layer, progress_callback)
if new_layer:
context.layers.add_layer(new_layer)
vollog.log(constants.LOGLEVEL_VV,
"Stacked {} using {}".format(new_layer.name, stacker_cls.__name__))
f"Stacked {new_layer.name} using {stacker_cls.__name__}")
break
except Exception as excp:
# Stacking exceptions are likely only of interest to developers, so the lowest level of logging
fulltrace = traceback.TracebackException.from_exception(excp).format(chain = True)
vollog.log(constants.LOGLEVEL_VVV, "Exception during stacking: {}".format(str(excp)))
vollog.log(constants.LOGLEVEL_VVV, f"Exception during stacking: {str(excp)}")
vollog.log(constants.LOGLEVEL_VVVV, "\n".join(fulltrace))
else:
stacked = False
volatility3/framework/automagic/symbol_cache.py
+5 -5
View File
@@ -90,10 +90,10 @@ class SymbolBannerCache(interfaces.automagic.AutomagicInterface):
total = len(cacheables)
if total > 0:
vollog.info("Building {} caches...".format(self.os))
vollog.info(f"Building {self.os} caches...")
for current in range(total):
if progress_callback is not None:
progress_callback(current * 100 / total, "Building {} caches".format(self.os))
progress_callback(current * 100 / total, f"Building {self.os} caches")
isf_url = cacheables[current]
isf = None
@@ -105,7 +105,7 @@ class SymbolBannerCache(interfaces.automagic.AutomagicInterface):
# We don't bother with the hash (it'll likely take too long to validate)
# but we should check at least that the banner matches on load.
banner = isf.get_symbol(self.symbol_name).constant_data
vollog.log(constants.LOGLEVEL_VV, "Caching banner {} for file {}".format(banner, isf_url))
vollog.log(constants.LOGLEVEL_VV, f"Caching banner {banner} for file {isf_url}")
bannerlist = banners.get(banner, [])
bannerlist.append(isf_url)
@@ -113,7 +113,7 @@ class SymbolBannerCache(interfaces.automagic.AutomagicInterface):
except exceptions.SymbolError:
pass
except json.JSONDecodeError:
vollog.log(constants.LOGLEVEL_VV, "Caching file {} failed due to JSON error".format(isf_url))
vollog.log(constants.LOGLEVEL_VV, f"Caching file {isf_url} failed due to JSON error")
finally:
# Get rid of the loaded file, in case it sits in memory
if isf:
@@ -124,4 +124,4 @@ class SymbolBannerCache(interfaces.automagic.AutomagicInterface):
self.save_banners(banners)
if progress_callback is not None:
progress_callback(100, "Built {} caches".format(self.os))
progress_callback(100, f"Built {self.os} caches")
volatility3/framework/automagic/symbol_finder.py
+4 -4
View File
@@ -33,7 +33,7 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
requested."""
if not self._banners:
if not self.banner_cache:
raise RuntimeError("Cache has not been properly defined for {}".format(self.__class__.__name__))
raise RuntimeError(f"Cache has not been properly defined for {self.__class__.__name__}")
self._banners = self.banner_cache.load_banners()
return self._banners
@@ -98,11 +98,11 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
banner_list = layer.scan(context = context, scanner = mss, progress_callback = progress_callback)
for _, banner in banner_list:
vollog.debug("Identified banner: {}".format(repr(banner)))
vollog.debug(f"Identified banner: {repr(banner)}")
symbol_files = self.banners.get(banner, None)
if symbol_files:
isf_path = symbol_files[0]
vollog.debug("Using symbol library: {}".format(symbol_files[0]))
vollog.debug(f"Using symbol library: {symbol_files[0]}")
clazz = self.symbol_class
# Set the discovered options
path_join = interfaces.configuration.path_join
@@ -134,7 +134,7 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
break
else:
if symbol_files:
vollog.debug("Symbol library path not found: {}".format(symbol_files[0]))
vollog.debug(f"Symbol library path not found: {symbol_files[0]}")
# print("Kernel", banner, hex(banner_offset))
else:
vollog.debug("No existing banners found")
volatility3/framework/automagic/windows.py
+1 -1
View File
@@ -272,7 +272,7 @@ class WintelHelper(interfaces.automagic.AutomagicInterface):
physical_layer_name = requirement.requirements["memory_layer"].config_value(
context, sub_config_path)
if not isinstance(physical_layer_name, str):
raise TypeError("Physical layer name is not a string: {}".format(sub_config_path))
raise TypeError(f"Physical layer name is not a string: {sub_config_path}")
physical_layer = context.layers[physical_layer_name]
# Check lower layer metadata first
if physical_layer.metadata.get('page_map_offset', None):
volatility3/framework/configuration/requirements.py
+8 -8
View File
@@ -105,7 +105,7 @@ class ListRequirement(interfaces.configuration.RequirementInterface):
context.config[config_path] = []
if not isinstance(value, list):
# TODO: Check this is the correct response for an error
raise TypeError("Unexpected config value found: {}".format(repr(value)))
raise TypeError(f"Unexpected config value found: {repr(value)}")
if not (self.min_elements <= len(value)):
vollog.log(constants.LOGLEVEL_V, "TypeError - Too few values provided to list option.")
return {config_path: self}
@@ -264,20 +264,20 @@ class TranslationLayerRequirement(interfaces.configuration.ConstructableRequirem
value = self.config_value(context, config_path, None)
if isinstance(value, str):
if value not in context.layers:
vollog.log(constants.LOGLEVEL_V, "IndexError - Layer not found in memory space: {}".format(value))
vollog.log(constants.LOGLEVEL_V, f"IndexError - Layer not found in memory space: {value}")
return {config_path: self}
if self.oses and context.layers[value].metadata.get('os', None) not in self.oses:
vollog.log(constants.LOGLEVEL_V, "TypeError - Layer is not the required OS: {}".format(value))
vollog.log(constants.LOGLEVEL_V, f"TypeError - Layer is not the required OS: {value}")
return {config_path: self}
if (self.architectures
and context.layers[value].metadata.get('architecture', None) not in self.architectures):
vollog.log(constants.LOGLEVEL_V, "TypeError - Layer is not the required Architecture: {}".format(value))
vollog.log(constants.LOGLEVEL_V, f"TypeError - Layer is not the required Architecture: {value}")
return {config_path: self}
return {}
if value is not None:
vollog.log(constants.LOGLEVEL_V,
"TypeError - Translation Layer Requirement only accepts string labels: {}".format(repr(value)))
f"TypeError - Translation Layer Requirement only accepts string labels: {repr(value)}")
return {config_path: self}
# TODO: check that the space in the context lives up to the requirements for arch/os etc
@@ -285,7 +285,7 @@ class TranslationLayerRequirement(interfaces.configuration.ConstructableRequirem
### NOTE: This validate method has side effects (the dependencies can change)!!!
self._validate_class(context, interfaces.configuration.parent_path(config_path))
vollog.log(constants.LOGLEVEL_V, "IndexError - No configuration provided: {}".format(config_path))
vollog.log(constants.LOGLEVEL_V, f"IndexError - No configuration provided: {config_path}")
return {config_path: self}
def construct(self, context: interfaces.context.ContextInterface, config_path: str) -> None:
@@ -333,7 +333,7 @@ class SymbolTableRequirement(interfaces.configuration.ConstructableRequirementIn
value = self.config_value(context, config_path, None)
if not isinstance(value, str) and value is not None:
vollog.log(constants.LOGLEVEL_V,
"TypeError - SymbolTableRequirement only accepts string labels: {}".format(repr(value)))
f"TypeError - SymbolTableRequirement only accepts string labels: {repr(value)}")
return {config_path: self}
if value and value in context.symbol_space:
# This is an expected situation, so return rather than raise
@@ -345,7 +345,7 @@ class SymbolTableRequirement(interfaces.configuration.ConstructableRequirementIn
### NOTE: This validate method has side effects (the dependencies can change)!!!
self._validate_class(context, interfaces.configuration.parent_path(config_path))
vollog.log(constants.LOGLEVEL_V, "Symbol table requirement not yet fulfilled: {}".format(config_path))
vollog.log(constants.LOGLEVEL_V, f"Symbol table requirement not yet fulfilled: {config_path}")
return {config_path: self}
def construct(self, context: interfaces.context.ContextInterface, config_path: str) -> None:
volatility3/framework/contexts/__init__.py
+2 -2
View File
@@ -155,7 +155,7 @@ def get_module_wrapper(method: str) -> Callable:
if constants.BANG not in name:
name = self._module_name + constants.BANG + name
else:
raise ValueError("Cannot reference another module when calling {}".format(method))
raise ValueError(f"Cannot reference another module when calling {method}")
return getattr(self._context.symbol_space, method)(name)
for entry in ['__annotations__', '__doc__', '__module__', '__name__', '__qualname__']:
@@ -232,7 +232,7 @@ class Module(interfaces.context.ModuleInterface):
offset += self._offset
if symbol_val.type is None:
raise TypeError("Symbol {} has no associated type".format(symbol_val.name))
raise TypeError(f"Symbol {symbol_val.name} has no associated type")
# Ensure we don't use a layer_name other than the module's, why would anyone do that?
if 'layer_name' in kwargs:
volatility3/framework/interfaces/configuration.py
+3 -3
View File
@@ -77,7 +77,7 @@ class HierarchicalDict(collections.abc.Mapping):
separator: A custom hierarchy separator (defaults to CONFIG_SEPARATOR)
"""
if not (isinstance(separator, str) and len(separator) == 1):
raise TypeError("Separator must be a one character string: {}".format(separator))
raise TypeError(f"Separator must be a one character string: {separator}")
self._separator = separator
self._data = {} # type: Dict[str, ConfigSimpleType]
self._subdict = {} # type: Dict[str, 'HierarchicalDict']
@@ -88,7 +88,7 @@ class HierarchicalDict(collections.abc.Mapping):
self[k] = v
elif initial_dict is not None:
raise TypeError(
"Initial_dict must be a dictionary or JSON string containing a dictionary: {}".format(initial_dict))
f"Initial_dict must be a dictionary or JSON string containing a dictionary: {initial_dict}")
def __eq__(self, other):
"""Define equality between HierarchicalDicts"""
@@ -315,7 +315,7 @@ class RequirementInterface(metaclass = ABCMeta):
"""
super().__init__()
if CONFIG_SEPARATOR in name:
raise ValueError("Name cannot contain the config-hierarchy divider ({})".format(CONFIG_SEPARATOR))
raise ValueError(f"Name cannot contain the config-hierarchy divider ({CONFIG_SEPARATOR})")
self._name = name
self._description = description or ""
self._default = default
volatility3/framework/interfaces/layers.py
+11 -11
View File
@@ -236,7 +236,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla
for value in scan_iterator():
if progress_callback:
progress_callback(scan_metric(progress.value),
"Scanning {} using {}".format(self.name, scanner.__class__.__name__))
f"Scanning {self.name} using {scanner.__class__.__name__}")
yield from scan_chunk(value)
else:
progress = multiprocessing.Manager().Value("Q", 0)
@@ -251,7 +251,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla
if progress_callback:
# Run the progress_callback
progress_callback(scan_metric(progress.value),
"Scanning {} using {}".format(self.name, scanner.__class__.__name__))
f"Scanning {self.name} using {scanner.__class__.__name__}")
# Ensures we don't burn CPU cycles going round in a ready waiting loop
# without delaying the user too long between progress updates/results
result.wait(0.1)
@@ -259,7 +259,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla
yield from result_value
except Exception as e:
# We don't care the kind of exception, so catch and report on everything, yielding nothing further
vollog.debug("Scan Failure: {}".format(str(e)))
vollog.debug(f"Scan Failure: {str(e)}")
vollog.log(constants.LOGLEVEL_VVV,
"\n".join(traceback.TracebackException.from_exception(e).format(chain = True)))
@@ -325,7 +325,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla
layer_name, self.name, address))
if len(data) > scanner.chunk_size + scanner.overlap:
vollog.debug("Scan chunk too large: {}".format(hex(len(data))))
vollog.debug(f"Scan chunk too large: {hex(len(data))}")
progress.value = chunk_end
return list(scanner(data, chunk_end - len(data)))
@@ -429,7 +429,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
ignore_errors = pad):
if not pad and layer_offset > current_offset:
raise exceptions.InvalidAddressException(
self.name, current_offset, "Layer {} cannot map offset: {}".format(self.name, current_offset))
self.name, current_offset, f"Layer {self.name} cannot map offset: {current_offset}")
elif layer_offset > current_offset:
output += b"\x00" * (layer_offset - current_offset)
current_offset = layer_offset
@@ -452,7 +452,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
for (layer_offset, sublength, mapped_offset, mapped_length, layer) in self.mapping(offset, length):
if layer_offset > current_offset:
raise exceptions.InvalidAddressException(
self.name, current_offset, "Layer {} cannot map offset: {}".format(self.name, current_offset))
self.name, current_offset, f"Layer {self.name} cannot map offset: {current_offset}")
value_chunk = value[layer_offset - offset:layer_offset - offset + sublength]
new_data = self._encode_data(layer, mapped_offset, layer_offset, value_chunk)
@@ -566,12 +566,12 @@ class LayerContainer(collections.abc.Mapping):
layer: the layer to add to the list of layers (based on layer.name)
"""
if layer.name in self._layers:
raise exceptions.LayerException(layer.name, "Layer already exists: {}".format(layer.name))
raise exceptions.LayerException(layer.name, f"Layer already exists: {layer.name}")
if isinstance(layer, TranslationLayerInterface):
missing_list = [sublayer for sublayer in layer.dependencies if sublayer not in self._layers]
if missing_list:
raise exceptions.LayerException(
layer.name, "Layer {} has unmet dependencies: {}".format(layer.name, ", ".join(missing_list)))
layer.name, f"Layer {layer.name} has unmet dependencies: {', '.join(missing_list)}")
self._layers[layer.name] = layer
def del_layer(self, name: str) -> None:
@@ -587,7 +587,7 @@ class LayerContainer(collections.abc.Mapping):
if depend_list:
raise exceptions.LayerException(
self._layers[layer].name,
"Layer {} is depended upon: {}".format(self._layers[layer].name, ", ".join(depend_list)))
f"Layer {self._layers[layer].name} is depended upon: {', '.join(depend_list)}")
self._layers[name].destroy()
del self._layers[name]
@@ -604,9 +604,9 @@ class LayerContainer(collections.abc.Mapping):
if prefix not in self:
return prefix
count = 1
while "{}_{}".format(prefix, count) in self:
while f"{prefix}_{count}" in self:
count += 1
return "{}_{}".format(prefix, count)
return f"{prefix}_{count}"
def __getitem__(self, name: str) -> DataLayerInterface:
"""Returns the layer of specified name."""
volatility3/framework/interfaces/objects.py
+6 -6
View File
@@ -31,7 +31,7 @@ class ReadOnlyMapping(collections.abc.Mapping):
return super().__getattribute__(attr)
if attr in self._dict:
return self._dict[attr]
raise AttributeError("Object has no attribute: {}.{}".format(self.__class__.__name__, attr))
raise AttributeError(f"Object has no attribute: {self.__class__.__name__}.{attr}")
def __getitem__(self, name: str) -> Any:
"""Returns the item requested."""
@@ -141,10 +141,10 @@ class ObjectInterface(metaclass = abc.ABCMeta):
KeyError: If the table_name is not valid within the object's context
"""
if constants.BANG not in self.vol.type_name:
raise ValueError("Unable to determine table for symbol: {}".format(self.vol.type_name))
raise ValueError(f"Unable to determine table for symbol: {self.vol.type_name}")
table_name = self.vol.type_name[:self.vol.type_name.index(constants.BANG)]
if table_name not in self._context.symbol_space:
raise KeyError("Symbol table not found in context's symbol_space for symbol: {}".format(self.vol.type_name))
raise KeyError(f"Symbol table not found in context's symbol_space for symbol: {self.vol.type_name}")
return table_name
def cast(self, new_type_name: str, **additional) -> 'ObjectInterface':
@@ -231,14 +231,14 @@ class ObjectInterface(metaclass = abc.ABCMeta):
@abc.abstractmethod
def replace_child(cls, template: 'Template', old_child: 'Template', new_child: 'Template') -> None:
"""Substitutes the old_child for the new_child."""
raise KeyError("Template does not contain any children to replace: {}".format(template.vol.type_name))
raise KeyError(f"Template does not contain any children to replace: {template.vol.type_name}")
@classmethod
@abc.abstractmethod
def relative_child_offset(cls, template: 'Template', child: str) -> int:
"""Returns the relative offset from the head of the parent data to
the child member."""
raise KeyError("Template does not contain any children: {}".format(template.vol.type_name))
raise KeyError(f"Template does not contain any children: {template.vol.type_name}")
@classmethod
@abc.abstractmethod
@@ -330,7 +330,7 @@ class Template:
if attr != '_vol':
if attr in self._vol:
return self._vol[attr]
raise AttributeError("{} object has no attribute {}".format(self.__class__.__name__, attr))
raise AttributeError(f"{self.__class__.__name__} object has no attribute {attr}")
def __call__(self, context: 'interfaces.context.ContextInterface',
object_info: ObjectInformation) -> ObjectInterface:
volatility3/framework/interfaces/plugins.py
+1 -1
View File
@@ -65,7 +65,7 @@ class FileHandlerInterface(io.RawIOBase):
if exc_type is None and exc_value is None and traceback is None:
self.close()
else:
vollog.warning("File {} could not be written: {}".format(self._preferred_filename, str(exc_value)))
vollog.warning(f"File {self._preferred_filename} could not be written: {str(exc_value)}")
self.close()
volatility3/framework/interfaces/symbols.py
+1 -1
View File
@@ -31,7 +31,7 @@ class SymbolInterface:
"""
self._name = name
if constants.BANG in self._name:
raise ValueError("Symbol names cannot contain the symbol differentiator ({})".format(constants.BANG))
raise ValueError(f"Symbol names cannot contain the symbol differentiator ({constants.BANG})")
# Scope can be added at a later date
self._location = None
volatility3/framework/layers/crash.py
+8 -8
View File
@@ -70,8 +70,8 @@ class WindowsCrashDump32Layer(segmented.SegmentedLayer):
# Verify that it is a supported format
if header.DumpType not in self.supported_dumptypes:
vollog.log(constants.LOGLEVEL_VVVV, "unsupported dump format 0x{:x}".format(header.DumpType))
raise WindowsCrashDumpFormatException(name, "unsupported dump format 0x{:x}".format(header.DumpType))
vollog.log(constants.LOGLEVEL_VVVV, f"unsupported dump format 0x{header.DumpType:x}")
raise WindowsCrashDumpFormatException(name, f"unsupported dump format 0x{header.DumpType:x}")
# Then call the super, which will call load_segments (which needs the base_layer before it'll work)
super().__init__(context, config_path, name)
@@ -143,11 +143,11 @@ class WindowsCrashDump32Layer(segmented.SegmentedLayer):
segment_length = (last_bit_seen - first_bit + 1) * 0x1000
segments.append((first_bit * 0x1000, first_offset, segment_length, segment_length))
else:
vollog.log(constants.LOGLEVEL_VVVV, "unsupported dump format 0x{:x}".format(self.dump_type))
raise WindowsCrashDumpFormatException(self.name, "unsupported dump format 0x{:x}".format(self.dump_type))
vollog.log(constants.LOGLEVEL_VVVV, f"unsupported dump format 0x{self.dump_type:x}")
raise WindowsCrashDumpFormatException(self.name, f"unsupported dump format 0x{self.dump_type:x}")
if len(segments) == 0:
raise WindowsCrashDumpFormatException(self.name, "No Crash segments defined in {}".format(self._base_layer))
raise WindowsCrashDumpFormatException(self.name, f"No Crash segments defined in {self._base_layer}")
else:
# report the segments for debugging. this is valuable for dev/troubleshooting but
# not important enough for a dedicated plugin.
@@ -167,15 +167,15 @@ class WindowsCrashDump32Layer(segmented.SegmentedLayer):
header_data = base_layer.read(offset, cls._magic_struct.size)
except exceptions.InvalidAddressException:
raise WindowsCrashDumpFormatException(base_layer.name,
"Crashdump header not found at offset {}".format(offset))
f"Crashdump header not found at offset {offset}")
(signature, validdump) = cls._magic_struct.unpack(header_data)
if signature != cls.SIGNATURE:
raise WindowsCrashDumpFormatException(
base_layer.name, "Bad signature 0x{:x} at file offset 0x{:x}".format(signature, offset))
base_layer.name, f"Bad signature 0x{signature:x} at file offset 0x{offset:x}")
if validdump != cls.VALIDDUMP:
raise WindowsCrashDumpFormatException(base_layer.name,
"Invalid dump 0x{:x} at file offset 0x{:x}".format(validdump, offset))
f"Invalid dump 0x{validdump:x} at file offset 0x{offset:x}")
return signature, validdump
volatility3/framework/layers/elf.py
+5 -5
View File
@@ -46,7 +46,7 @@ class Elf64Layer(segmented.SegmentedLayer):
segments.append((int(phdr.p_paddr), int(phdr.p_offset), int(phdr.p_memsz), int(phdr.p_memsz)))
if len(segments) == 0:
raise ElfFormatException(self.name, "No ELF segments defined in {}".format(self._base_layer))
raise ElfFormatException(self.name, f"No ELF segments defined in {self._base_layer}")
self._segments = segments
@@ -56,12 +56,12 @@ class Elf64Layer(segmented.SegmentedLayer):
header_data = base_layer.read(offset, cls._header_struct.size)
except exceptions.InvalidAddressException:
raise ElfFormatException(base_layer.name,
"Offset 0x{:0x} does not exist within the base layer".format(offset))
f"Offset 0x{offset:0x} does not exist within the base layer")
(magic, elf_class, elf_data_encoding, elf_version) = cls._header_struct.unpack(header_data)
if magic != cls.MAGIC:
raise ElfFormatException(base_layer.name, "Bad magic 0x{:x} at file offset 0x{:x}".format(magic, offset))
raise ElfFormatException(base_layer.name, f"Bad magic 0x{magic:x} at file offset 0x{offset:x}")
if elf_class != cls.ELF_CLASS:
raise ElfFormatException(base_layer.name, "ELF class is not 64-bit (2): {:d}".format(elf_class))
raise ElfFormatException(base_layer.name, f"ELF class is not 64-bit (2): {elf_class:d}")
# Virtualbox uses an ELF version of 0, which isn't to specification, but is ok to deal with
return True
@@ -78,7 +78,7 @@ class Elf64Stacker(interfaces.automagic.StackerLayerInterface):
if not Elf64Layer._check_header(context.layers[layer_name]):
return None
except ElfFormatException as excp:
vollog.log(constants.LOGLEVEL_VVVV, "Exception: {}".format(excp))
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
volatility3/framework/layers/intel.py
+1 -1
View File
@@ -107,7 +107,7 @@ class Intel(linear.LinearlyMappedLayer):
# Now we're done
if not self._page_is_valid(entry):
raise exceptions.PagedInvalidAddressException(self.name, offset, position + 1, entry,
"Page Fault at entry {} in page entry".format(hex(entry)))
f"Page Fault at entry {hex(entry)} in page entry")
page = self._mask(entry, self._maxphyaddr - 1, position + 1) | self._mask(offset, position, 0)
return page, 1 << (position + 1), self._base_layer
volatility3/framework/layers/leechcore.py
+1 -1
View File
@@ -43,7 +43,7 @@ if HAS_LEECHCORE:
try:
self._handle = leechcorepyc.LeechCore(self._device)
except TypeError:
raise IOError("Unable to open LeechCore device {}".format(self._device))
raise IOError(f"Unable to open LeechCore device {self._device}")
return self._handle
def fileno(self):
volatility3/framework/layers/lime.py
+5 -5
View File
@@ -45,7 +45,7 @@ class LimeLayer(segmented.SegmentedLayer):
if start < maxaddr or end < start:
raise LimeFormatException(
self.name, "Bad start/end 0x{:x}/0x{:x} at file offset 0x{:x}".format(start, end, offset))
self.name, f"Bad start/end 0x{start:x}/0x{end:x} at file offset 0x{offset:x}")
segment_length = end - start + 1
segments.append((start, offset + header_size, segment_length, segment_length))
@@ -53,7 +53,7 @@ class LimeLayer(segmented.SegmentedLayer):
offset = offset + header_size + segment_length
if len(segments) == 0:
raise LimeFormatException(self.name, "No LiME segments defined in {}".format(self._base_layer))
raise LimeFormatException(self.name, f"No LiME segments defined in {self._base_layer}")
self._segments = segments
@@ -63,13 +63,13 @@ class LimeLayer(segmented.SegmentedLayer):
header_data = base_layer.read(offset, cls._header_struct.size)
except exceptions.InvalidAddressException:
raise LimeFormatException(base_layer.name,
"Offset 0x{:0x} does not exist within the base layer".format(offset))
f"Offset 0x{offset:0x} does not exist within the base layer")
(magic, version, start, end, reserved) = cls._header_struct.unpack(header_data)
if magic != cls.MAGIC:
raise LimeFormatException(base_layer.name, "Bad magic 0x{:x} at file offset 0x{:x}".format(magic, offset))
raise LimeFormatException(base_layer.name, f"Bad magic 0x{magic:x} at file offset 0x{offset:x}")
if version != cls.VERSION:
raise LimeFormatException(base_layer.name,
"Unexpected version {:d} at file offset 0x{:x}".format(version, offset))
f"Unexpected version {version:d} at file offset 0x{offset:x}")
return start, end
volatility3/framework/layers/linear.py
+4 -4
View File
@@ -16,13 +16,13 @@ class LinearlyMappedLayer(interfaces.layers.TranslationLayerInterface):
original_offset, _, mapped_offset, _, layer = mapping[0]
if original_offset != offset:
raise exceptions.LayerException(self.name,
"Layer {} claims to map linearly but does not".format(self.name))
f"Layer {self.name} claims to map linearly but does not")
else:
if ignore_errors:
# We should only hit this if we ignored errors, but check anyway
return None, None
raise exceptions.InvalidAddressException(self.name, offset,
"Cannot translate {} in layer {}".format(offset, self.name))
f"Cannot translate {offset} in layer {self.name}")
return mapped_offset, layer
# ## Read/Write functions for mapped pages
@@ -37,7 +37,7 @@ class LinearlyMappedLayer(interfaces.layers.TranslationLayerInterface):
for (offset, _, mapped_offset, mapped_length, layer) in self.mapping(offset, length, ignore_errors = pad):
if not pad and offset > current_offset:
raise exceptions.InvalidAddressException(
self.name, current_offset, "Layer {} cannot map offset: {}".format(self.name, current_offset))
self.name, current_offset, f"Layer {self.name} cannot map offset: {current_offset}")
elif offset > current_offset:
output += [b"\x00" * (offset - current_offset)]
current_offset = offset
@@ -57,7 +57,7 @@ class LinearlyMappedLayer(interfaces.layers.TranslationLayerInterface):
for (offset, _, mapped_offset, length, layer) in self.mapping(offset, length):
if offset > current_offset:
raise exceptions.InvalidAddressException(
self.name, current_offset, "Layer {} cannot map offset: {}".format(self.name, current_offset))
self.name, current_offset, f"Layer {self.name} cannot map offset: {current_offset}")
elif offset < current_offset:
raise exceptions.LayerException(self.name, "Mapping returned an overlapping element")
self._context.layers.write(layer, mapped_offset, value[:length])
volatility3/framework/layers/msf.py
+1 -1
View File
@@ -224,7 +224,7 @@ class PdbMSFStream(linear.LinearlyMappedLayer):
def _pdb_layer(self) -> PdbMultiStreamFormat:
if self._base_layer not in self._context.layers:
raise PDBFormatException(self._base_layer,
"No PdbMultiStreamFormat layer found: {}".format(self._base_layer))
f"No PdbMultiStreamFormat layer found: {self._base_layer}")
result = self._context.layers[self._base_layer]
if isinstance(result, PdbMultiStreamFormat):
return result
volatility3/framework/layers/physical.py
+1 -1
View File
@@ -164,7 +164,7 @@ class FileLayer(interfaces.layers.DataLayerInterface):
if not self._file.writable():
if not self._write_warning:
self._write_warning = True
vollog.warning("Try to write to unwritable layer: {}".format(self.name))
vollog.warning(f"Try to write to unwritable layer: {self.name}")
return None
if not self.is_valid(offset, len(data)):
invalid_address = offset
volatility3/framework/layers/qemu.py
+4 -4
View File
@@ -179,21 +179,21 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
index += 4
if section_id != current_section_id:
raise exceptions.LayerException(
self._name, 'QEMU section footer mismatch: {} and {}'.format(current_section_id, section_id))
self._name, f'QEMU section footer mismatch: {current_section_id} and {section_id}')
elif section_byte == self.QEVM_EOF:
pass
else:
raise exceptions.LayerException(self._name, 'QEMU unknown section encountered: {}'.format(section_byte))
raise exceptions.LayerException(self._name, f'QEMU unknown section encountered: {section_byte}')
def extract_data(self, index, name, version_id):
if name == 'ram':
if version_id != 4:
raise exceptions.LayerException("QEMU unknown RAM version_id {}".format(version_id))
raise exceptions.LayerException(f"QEMU unknown RAM version_id {version_id}")
new_segments, index = self._get_ram_segments(index, self._configuration.get('page_size', None) or 4096)
self._segments += new_segments
elif name == 'spapr/htab':
if version_id != 1:
raise exceptions.LayerException("QEMU unknown HTAB version_id {}".format(version_id))
raise exceptions.LayerException(f"QEMU unknown HTAB version_id {version_id}")
header = self.context.object(self._qemu_table_name + constants.BANG + 'unsigned long',
offset = index,
layer_name = self._base_layer)
volatility3/framework/layers/registry.py
+1 -1
View File
@@ -48,7 +48,7 @@ class RegistryHive(linear.LinearlyMappedLayer):
# TODO: Check the checksum
if self.hive.Signature != 0xbee0bee0:
raise RegistryFormatException(
self.name, "Registry hive at {} does not have a valid signature".format(self._hive_offset))
self.name, f"Registry hive at {self._hive_offset} does not have a valid signature")
# Win10 17063 introduced the Registry process to map most hives. Check
# if it exists and update RegistryHive._base_layer
volatility3/framework/layers/resources.py
+5 -5
View File
@@ -74,7 +74,7 @@ class ResourceAccessor(object):
self._enable_cache = enable_cache
if self.list_handlers:
vollog.log(constants.LOGLEVEL_VVV,
"Available URL handlers: {}".format(", ".join([x.__name__ for x in self._handlers])))
f"Available URL handlers: {', '.join([x.__name__ for x in self._handlers])}")
self.__class__.list_handlers = False
def uses_cache(self, url: str) -> bool:
@@ -132,7 +132,7 @@ class ResourceAccessor(object):
"data_" + hashlib.sha512(bytes(url, 'raw_unicode_escape')).hexdigest() + ".cache")
if not os.path.exists(temp_filename):
vollog.debug("Caching file at: {}".format(temp_filename))
vollog.debug(f"Caching file at: {temp_filename}")
try:
content_length = fp.info().get('Content-Length', -1)
@@ -147,7 +147,7 @@ class ResourceAccessor(object):
count += len(block)
if self._progress_callback:
self._progress_callback(count * 100 / max(count, int(content_length)),
"Reading file {}".format(url))
f"Reading file {url}")
cache_file.write(block)
block = fp.read(block_size)
cache_file.close()
@@ -237,13 +237,13 @@ class JarHandler(VolatilityHandler):
if req.type == 'jar':
subscheme, remainder = req.full_url.split(":")[1], ":".join(req.full_url.split(":")[2:])
if subscheme != 'file':
vollog.log(constants.LOGLEVEL_VVV, "Unsupported jar subscheme {}".format(subscheme))
vollog.log(constants.LOGLEVEL_VVV, f"Unsupported jar subscheme {subscheme}")
return None
zipsplit = remainder.split("!")
if len(zipsplit) != 2:
vollog.log(constants.LOGLEVEL_VVV,
"Path did not contain exactly one fragment indicator: {}".format(remainder))
f"Path did not contain exactly one fragment indicator: {remainder}")
return None
zippath, filepath = zipsplit
volatility3/framework/layers/segmented.py
+1 -1
View File
@@ -67,7 +67,7 @@ class NonLinearlySegmentedLayer(interfaces.layers.TranslationLayerInterface, met
if next:
if i < len(self._segments):
return self._segments[i]
raise exceptions.InvalidAddressException(self.name, offset, "Invalid address at {:0x}".format(offset))
raise exceptions.InvalidAddressException(self.name, offset, f"Invalid address at {offset:0x}")
def mapping(self,
offset: int,
volatility3/framework/layers/vmware.py
+2 -2
View File
@@ -53,7 +53,7 @@ class VmwareLayer(segmented.SegmentedLayer):
data = meta_layer.read(0, header_size)
magic, unknown, groupCount = struct.unpack(self.header_structure, data)
if magic not in [b"\xD0\xBE\xD2\xBE", b"\xD1\xBA\xD1\xBA", b"\xD2\xBE\xD2\xBE", b"\xD3\xBE\xD3\xBE"]:
raise VmwareFormatException(self.name, "Wrong magic bytes for Vmware layer: {}".format(repr(magic)))
raise VmwareFormatException(self.name, f"Wrong magic bytes for Vmware layer: {repr(magic)}")
version = magic[0] & 0xf
group_size = struct.calcsize(self.group_structure)
@@ -171,7 +171,7 @@ class VmwareStacker(interfaces.automagic.StackerLayerInterface):
except IOError:
pass
vollog.log(constants.LOGLEVEL_VVVV, "Metadata found: VMSS ({}) or VMSN ({})".format(vmss_success, vmsn_success))
vollog.log(constants.LOGLEVEL_VVVV, f"Metadata found: VMSS ({vmss_success}) or VMSN ({vmsn_success})")
if not vmss_success and not vmsn_success:
return None
volatility3/framework/objects/__init__.py
+14 -14
View File
@@ -31,7 +31,7 @@ def convert_data_to_value(data: bytes, struct_type: Type[TUnion[int, float, byte
elif struct_type in [bytes, str]:
struct_format = str(data_format.length) + "s"
else:
raise TypeError("Cannot construct struct format for type {}".format(type(struct_type)))
raise TypeError(f"Cannot construct struct format for type {type(struct_type)}")
return struct.unpack(struct_format, data)[0]
@@ -41,7 +41,7 @@ def convert_value_to_data(value: TUnion[int, float, bytes, str, bool], struct_ty
data_format: DataFormatInfo) -> bytes:
"""Converts a particular value to a series of bytes."""
if not isinstance(value, struct_type):
raise TypeError("Written value is not of the correct type for {}".format(struct_type.__name__))
raise TypeError(f"Written value is not of the correct type for {struct_type.__name__}")
if struct_type == int and isinstance(value, int):
# Doubling up on the isinstance is for mypy
@@ -61,7 +61,7 @@ def convert_value_to_data(value: TUnion[int, float, bytes, str, bool], struct_ty
value = bytes(value, 'latin-1')
struct_format = str(data_format.length) + "s"
else:
raise TypeError("Cannot construct struct format for type {}".format(type(struct_type)))
raise TypeError(f"Cannot construct struct format for type {type(struct_type)}")
return struct.pack(struct_format, value)
@@ -459,7 +459,7 @@ class Enumeration(interfaces.objects.ObjectInterface, int):
# Technically this shouldn't be a problem, but since we inverse cache
# and can't map one value to two possibilities we throw an exception during build
# We can remove/work around this if it proves a common issue
raise ValueError("Enumeration value {} duplicated as {} and {}".format(v, k, inverse_choices[v]))
raise ValueError(f"Enumeration value {v} duplicated as {k} and {inverse_choices[v]}")
inverse_choices[v] = k
return inverse_choices
@@ -489,7 +489,7 @@ class Enumeration(interfaces.objects.ObjectInterface, int):
"""Returns the value for a specific name."""
if attr in self._vol['choices']:
return self._vol['choices'][attr]
raise AttributeError("Unknown attribute {} for Enumeration {}".format(attr, self._vol['type_name']))
raise AttributeError(f"Unknown attribute {attr} for Enumeration {self._vol['type_name']}")
def write(self, value: bytes):
raise NotImplementedError("Writing to Enumerations is not yet implemented")
@@ -589,7 +589,7 @@ class Array(interfaces.objects.ObjectInterface, collections.abc.Sequence):
the child member."""
if 'subtype' in template.vol and child == 'subtype':
return 0
raise IndexError("Member not present in array template: {}".format(child))
raise IndexError(f"Member not present in array template: {child}")
@overload
def __getitem__(self, i: int) -> interfaces.objects.Template:
@@ -660,10 +660,10 @@ class AggregateType(interfaces.objects.ObjectInterface):
"""Describes the object appropriately"""
extras = member_name = ''
if self.vol.native_layer_name != self.vol.layer_name:
extras += " (Native: {})".format(self.vol.native_layer_name)
extras += f" (Native: {self.vol.native_layer_name})"
if self.vol.member_name:
member_name = " (.{})".format(self.vol.member_name)
return "<{} {}{}: {} @ 0x{:x} #{}{}>".format(self.__class__.__name__, self.vol.type_name, member_name, self.vol.layer_name, self.vol.offset, self.vol.size, extras)
member_name = f" (.{self.vol.member_name})"
return f"<{self.__class__.__name__} {self.vol.type_name}{member_name}: {self.vol.layer_name} @ 0x{self.vol.offset:x} #{self.vol.size}{extras}>"
class VolTemplateProxy(interfaces.objects.ObjectInterface.VolTemplateProxy):
@@ -701,7 +701,7 @@ class AggregateType(interfaces.objects.ObjectInterface):
"""Returns the relative offset of a child to its parent."""
retlist = template.vol.members.get(child, None)
if retlist is None:
raise IndexError("Member not present in template: {}".format(child))
raise IndexError(f"Member not present in template: {child}")
return retlist[0]
@classmethod
@@ -722,9 +722,9 @@ class AggregateType(interfaces.objects.ObjectInterface):
agg_name = agg_type.__name__
assert isinstance(members, collections.abc.Mapping)
"{} members parameter must be a mapping: {}".format(agg_name, type(members))
f"{agg_name} members parameter must be a mapping: {type(members)}"
assert all([(isinstance(member, tuple) and len(member) == 2) for member in members.values()])
"{} members must be a tuple of relative_offsets and templates".format(agg_name)
f"{agg_name} members must be a tuple of relative_offsets and templates"
def member(self, attr: str = 'member') -> object:
"""Specifically named method for retrieving members."""
@@ -758,7 +758,7 @@ class AggregateType(interfaces.objects.ObjectInterface):
for agg_type in AggregateTypes:
if isinstance(self, agg_type):
agg_name = agg_type.__name__
raise AttributeError("{} has no attribute: {}.{}".format(agg_name, self.vol.type_name, attr))
raise AttributeError(f"{agg_name} has no attribute: {self.vol.type_name}.{attr}")
# Disable messing around with setattr until the consequences have been considered properly
# For example pdbutil constructs objects and then sets values for them
@@ -782,7 +782,7 @@ class AggregateType(interfaces.objects.ObjectInterface):
if isinstance(self, agg_type):
agg_name = agg_type.__name__
raise TypeError(
"{}s cannot be written to directly, individual members must be written instead".format(agg_name))
f"{agg_name}s cannot be written to directly, individual members must be written instead")
class StructType(AggregateType):
volatility3/framework/objects/templates.py
+1 -1
View File
@@ -94,7 +94,7 @@ class ReferenceTemplate(interfaces.objects.Template):
symbol_name = type_name[-1]
raise exceptions.SymbolError(
symbol_name, table_name,
"Template contains no information about its structure: {}".format(self.vol.type_name))
f"Template contains no information about its structure: {self.vol.type_name}")
size = property(_unresolved) # type: ClassVar[Any]
replace_child = _unresolved # type: ClassVar[Any]
volatility3/framework/plugins/__init__.py
+1 -1
View File
@@ -44,7 +44,7 @@ def construct_plugin(context: interfaces.context.ContextInterface,
if unsatisfied:
for error in errors:
error_string = [x for x in error.format_exception_only()][-1]
vollog.warning("Automagic exception occurred: {}".format(error_string[:-1]))
vollog.warning(f"Automagic exception occurred: {error_string[:-1]}")
vollog.log(constants.LOGLEVEL_V, "".join(error.format(chain = True)))
raise exceptions.UnsatisfiedException(unsatisfied)
volatility3/framework/plugins/configwriter.py
+1 -1
View File
@@ -42,7 +42,7 @@ class ConfigWriter(plugins.PluginInterface):
with self.open(filename) as file_data:
file_data.write(bytes(json.dumps(config, sort_keys = True, indent = 2), 'raw_unicode_escape'))
except Exception as excp:
vollog.warning("Unable to JSON encode configuration: {}".format(excp))
vollog.warning(f"Unable to JSON encode configuration: {excp}")
for k, v in config.items():
yield (0, (k, json.dumps(v)))
volatility3/framework/plugins/isfinfo.py
+1 -1
View File
@@ -117,7 +117,7 @@ class IsfInfo(plugins.PluginInterface):
windows_info = os.path.splitext(os.path.basename(entry))[0]
valid = check_valid(data)
except (UnicodeDecodeError, json.decoder.JSONDecodeError):
vollog.warning("Invalid ISF: {}".format(entry))
vollog.warning(f"Invalid ISF: {entry}")
yield (0, (entry, valid, num_bases, num_types, num_symbols, num_enums, windows_info, linux_banner,
mac_banner))
volatility3/framework/plugins/layerwriter.py
+4 -4
View File
@@ -73,7 +73,7 @@ class LayerWriter(plugins.PluginInterface):
data = layer.read(i, current_chunk_size, pad = True)
file_handle.write(data)
if progress_callback:
progress_callback((i / layer.maximum_address) * 100, 'Writing layer {}'.format(layer_name))
progress_callback((i / layer.maximum_address) * 100, f'Writing layer {layer_name}')
return file_handle
def _generator(self):
@@ -91,7 +91,7 @@ class LayerWriter(plugins.PluginInterface):
for name in self.config['layers']:
# Check the layer exists and validate the output file
if name not in self.context.layers:
yield 0, ('Layer Name {} does not exist'.format(name), )
yield 0, (f'Layer Name {name} does not exist', )
else:
output_name = self.config.get('output', ".".join([name, "raw"]))
try:
@@ -103,9 +103,9 @@ class LayerWriter(plugins.PluginInterface):
progress_callback = self._progress_callback)
file_handle.close()
except IOError as excp:
yield 0, ('Layer cannot be written to {}: {}'.format(self.config['output_name'], excp), )
yield 0, (f"Layer cannot be written to {self.config['output_name']}: {excp}", )
yield 0, ('Layer has been written to {}'.format(output_name), )
yield 0, (f'Layer has been written to {output_name}', )
def _generate_layers(self):
"""List layer names from this run"""
volatility3/framework/plugins/linux/bash.py
+1 -1
View File
@@ -106,5 +106,5 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
self.config['vmlinux'],
filter_func = filter_func)):
_depth, row_data = row
description = "{} ({}): \"{}\"".format(row_data[0], row_data[1], row_data[3])
description = f"{row_data[0]} ({row_data[1]}): \"{row_data[3]}\""
yield (description, timeliner.TimeLinerType.CREATED, row_data[2])
volatility3/framework/plugins/linux/check_creds.py
+1 -1
View File
@@ -55,7 +55,7 @@ class Check_creds(interfaces.plugins.PluginInterface):
if len(pids) > 1:
pid_str = ""
for pid in pids:
pid_str = pid_str + "{0:d}, ".format(pid)
pid_str = pid_str + f"{pid:d}, "
pid_str = pid_str[:-2]
yield (0, [str(pid_str)])
volatility3/framework/plugins/mac/bash.py
+1 -1
View File
@@ -107,5 +107,5 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
for row in self._generator(
list_tasks(self.context, self.config['primary'], self.config['darwin'], filter_func = filter_func)):
_depth, row_data = row
description = "{} ({}): \"{}\"".format(row_data[0], row_data[1], row_data[3])
description = f"{row_data[0]} ({row_data[1]}): \"{row_data[3]}\""
yield (description, timeliner.TimeLinerType.CREATED, row_data[2])
volatility3/framework/plugins/mac/ifconfig.py
+1 -1
View File
@@ -45,7 +45,7 @@ class Ifconfig(plugins.PluginInterface):
for ifaddr in mac.MacUtilities.walk_tailq(ifnet.if_addrhead, "ifa_link"):
ip = ifaddr.ifa_addr.get_address()
yield (0, ("{0}{1}".format(name, unit), ip, mac_addr, prom))
yield (0, (f"{name}{unit}", ip, mac_addr, prom))
def run(self):
return renderers.TreeGrid([("Interface", str), ("IP Address", str), ("Mac Address", str),
volatility3/framework/plugins/mac/netstat.py
+2 -2
View File
@@ -95,7 +95,7 @@ class Netstat(plugins.PluginInterface):
continue
yield (0, (format_hints.Hex(socket.vol.offset), "UNIX", path, 0, "", 0, "",
"{}/{:d}".format(task_name, pid)))
f"{task_name}/{pid:d}"))
elif family in [2, 30]:
state = socket.get_state()
@@ -107,7 +107,7 @@ class Netstat(plugins.PluginInterface):
(lip, lport, rip, rport) = vals
yield (0, (format_hints.Hex(socket.vol.offset), proto, lip, lport, rip, rport, state,
"{}/{:d}".format(task_name, pid)))
f"{task_name}/{pid:d}"))
def run(self):
return renderers.TreeGrid([("Offset", format_hints.Hex), ("Proto", str), ("Local IP", str), ("Local Port", int),
volatility3/framework/plugins/mac/pslist.py
+1 -1
View File
@@ -68,7 +68,7 @@ class PsList(interfaces.plugins.PluginInterface):
list_tasks = cls.list_tasks_pid_hash_table
else:
raise ValueError("Impossible method choice chosen")
vollog.debug("Using method {}".format(method))
vollog.debug(f"Using method {method}")
return list_tasks
volatility3/framework/plugins/timeliner.py
+4 -4
View File
@@ -113,9 +113,9 @@ class Timeliner(interfaces.plugins.PluginInterface):
for plugin in runable_plugins:
plugin_name = plugin.__class__.__name__
self._progress_callback((runable_plugins.index(plugin) * 100) // len(runable_plugins),
"Running plugin {}...".format(plugin_name))
f"Running plugin {plugin_name}...")
try:
vollog.log(logging.INFO, "Running {}".format(plugin_name))
vollog.log(logging.INFO, f"Running {plugin_name}")
for (item, timestamp_type, timestamp) in plugin.generate_timeline():
times = self.timeline.get((plugin_name, item), {})
if times.get(timestamp_type, None) is not None:
@@ -131,7 +131,7 @@ class Timeliner(interfaces.plugins.PluginInterface):
times.get(TimeLinerType.CHANGED, renderers.NotApplicableValue())
]))
except Exception:
vollog.log(logging.INFO, "Exception occurred running plugin: {}".format(plugin_name))
vollog.log(logging.INFO, f"Exception occurred running plugin: {plugin_name}")
vollog.log(logging.DEBUG, traceback.format_exc())
for data_item in sorted(data, key = self._sort_function):
yield data_item
@@ -206,7 +206,7 @@ class Timeliner(interfaces.plugins.PluginInterface):
plugins_to_run.append(plugin)
except exceptions.UnsatisfiedException as excp:
# Remove the failed plugin from the list and continue
vollog.debug("Unable to satisfy {}: {}".format(plugin_class.__name__, excp.unsatisfied))
vollog.debug(f"Unable to satisfy {plugin_class.__name__}: {excp.unsatisfied}")
continue
if self.config.get('record-config', False):
volatility3/framework/plugins/windows/callbacks.py
+1 -1
View File
@@ -90,7 +90,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
try:
symbol_offset = ntkrnlmp.get_symbol(symbol_name).address
except exceptions.SymbolError:
vollog.debug("Cannot find {}".format(symbol_name))
vollog.debug(f"Cannot find {symbol_name}")
continue
if is_vista_or_later and extended_list:
volatility3/framework/plugins/windows/cmdline.py
+2 -2
View File
@@ -66,10 +66,10 @@ class CmdLine(interfaces.plugins.PluginInterface):
result_text = self.get_cmdline(self.context, self.config["nt_symbols"], proc)
except exceptions.SwappedInvalidAddressException as exp:
result_text = "Required memory at {0:#x} is inaccessible (swapped)".format(exp.invalid_address)
result_text = f"Required memory at {exp.invalid_address:#x} is inaccessible (swapped)"
except exceptions.PagedInvalidAddressException as exp:
result_text = "Required memory at {0:#x} is not valid (process exited?)".format(exp.invalid_address)
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(
volatility3/framework/plugins/windows/crashinfo.py
+1 -1
View File
@@ -34,7 +34,7 @@ class Crashinfo(interfaces.plugins.PluginInterface):
dump_type = "Bitmap Dump (0x5)"
else:
# this should never happen since the crash layer only accepts 0x1 and 0x5
dump_type = "Unknown/Unsupported ({:#x})".format(header.DumpType)
dump_type = f"Unknown/Unsupported ({header.DumpType:#x})"
if header.DumpType == 0x5:
summary_header = layer.get_summary_header()
volatility3/framework/plugins/windows/dlllist.py
+2 -2
View File
@@ -83,7 +83,7 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
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))
vollog.debug(f"Unable to dump dll at offset {dll_entry.DllBase}: {excp}")
return None
return file_handle
@@ -131,7 +131,7 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
entry,
self.open,
proc_layer_name,
prefix = "pid.{}.".format(proc_id))
prefix = f"pid.{proc_id}.")
file_output = "Error outputting file"
if file_handle:
file_handle.close()
volatility3/framework/plugins/windows/dumpfiles.py
+10 -10
View File
@@ -80,13 +80,13 @@ class DumpFiles(interfaces.plugins.PluginInterface):
filedata.write(data)
if not bytes_written:
vollog.debug("No data is cached for the file at {0:#x}".format(file_object.vol.offset))
vollog.debug(f"No data is cached for the file at {file_object.vol.offset:#x}")
return None
else:
vollog.debug("Stored {}".format(filedata.preferred_filename))
vollog.debug(f"Stored {filedata.preferred_filename}")
return filedata
except exceptions.InvalidAddressException:
vollog.debug("Unable to dump file at {0:#x}".format(file_object.vol.offset))
vollog.debug(f"Unable to dump file at {file_object.vol.offset:#x}")
return None
@classmethod
@@ -105,7 +105,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
# use the "File" object type, such as \Device\Tcp and \Device\NamedPipe.
if file_obj.DeviceObject.DeviceType not in [FILE_DEVICE_DISK, FILE_DEVICE_NETWORK_FILE_SYSTEM]:
vollog.log(constants.LOGLEVEL_VVV,
"The file object at {0:#x} is not a file on disk".format(file_obj.vol.offset))
f"The file object at {file_obj.vol.offset:#x} is not a file on disk")
return
# Depending on the type of object (DataSection, ImageSection, SharedCacheMap) we may need to
@@ -134,7 +134,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
dump_parameters.append((control_area, memory_layer, extension))
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV,
"{0} is unavailable for file {1:#x}".format(member_name, file_obj.vol.offset))
f"{member_name} is unavailable for file {file_obj.vol.offset:#x}")
# The SharedCacheMap is handled differently than the caches above.
# We carve these "pages" from the primary_layer.
@@ -145,7 +145,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
dump_parameters.append((shared_cache_map, primary_layer, "vacb"))
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV,
"SharedCacheMap is unavailable for file {0:#x}".format(file_obj.vol.offset))
f"SharedCacheMap is unavailable for file {file_obj.vol.offset:#x}")
for memory_object, layer, extension in dump_parameters:
cache_name = EXTENSION_CACHE_MAP[extension]
@@ -187,7 +187,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
object_table = proc.ObjectTable
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV,
"Cannot access _EPROCESS.ObjectTable at {0:#x}".format(proc.vol.offset))
f"Cannot access _EPROCESS.ObjectTable at {proc.vol.offset:#x}")
continue
for entry in handles_plugin.handles(object_table):
@@ -200,7 +200,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
yield (0, result)
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV,
"Cannot extract file from _OBJECT_HEADER at {0:#x}".format(entry.vol.offset))
f"Cannot extract file from _OBJECT_HEADER at {entry.vol.offset:#x}")
# Pull file objects from the VADs. This will produce DLLs and EXEs that are
# mapped into the process as images, but that the process doesn't have an
@@ -224,7 +224,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
yield (0, result)
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV,
"Cannot extract file from VAD at {0:#x}".format(vad.vol.offset))
f"Cannot extract file from VAD at {vad.vol.offset:#x}")
elif offsets:
# Now process any offsets explicitly requested by the user.
@@ -242,7 +242,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
for result in self.process_file_object(self.context, self.config["primary"], self.open, file_obj):
yield (0, result)
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV, "Cannot extract file at {0:#x}".format(offset))
vollog.log(constants.LOGLEVEL_VVV, f"Cannot extract file at {offset:#x}")
def run(self):
# a list of tuples (<int>, <bool>) where <int> is the address and <bool> is True for virtual.
volatility3/framework/plugins/windows/handles.py
+5 -5
View File
@@ -203,7 +203,7 @@ class Handles(interfaces.plugins.PluginInterface):
type_name = objt.Name.String
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV,
"Cannot access _OBJECT_HEADER Name at {0:#x}".format(objt.vol.offset))
f"Cannot access _OBJECT_HEADER Name at {objt.vol.offset:#x}")
continue
type_map[i] = type_name
@@ -305,7 +305,7 @@ class Handles(interfaces.plugins.PluginInterface):
object_table = proc.ObjectTable
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV,
"Cannot access _EPROCESS.ObjectType at {0:#x}".format(proc.vol.offset))
f"Cannot access _EPROCESS.ObjectType at {proc.vol.offset:#x}")
continue
process_name = utility.array_to_string(proc.ImageFileName)
@@ -320,10 +320,10 @@ class Handles(interfaces.plugins.PluginInterface):
obj_name = item.file_name_with_device()
elif obj_type == "Process":
item = entry.Body.cast("_EPROCESS")
obj_name = "{} Pid {}".format(utility.array_to_string(proc.ImageFileName), item.UniqueProcessId)
obj_name = f"{utility.array_to_string(proc.ImageFileName)} Pid {item.UniqueProcessId}"
elif obj_type == "Thread":
item = entry.Body.cast("_ETHREAD")
obj_name = "Tid {} Pid {}".format(item.Cid.UniqueThread, item.Cid.UniqueProcess)
obj_name = f"Tid {item.Cid.UniqueThread} Pid {item.Cid.UniqueProcess}"
elif obj_type == "Key":
item = entry.Body.cast("_CM_KEY_BODY")
obj_name = item.get_full_key_name()
@@ -335,7 +335,7 @@ class Handles(interfaces.plugins.PluginInterface):
except (exceptions.InvalidAddressException):
vollog.log(constants.LOGLEVEL_VVV,
"Cannot access _OBJECT_HEADER at {0:#x}".format(entry.vol.offset))
f"Cannot access _OBJECT_HEADER at {entry.vol.offset:#x}")
continue
yield (0, (proc.UniqueProcessId, process_name, format_hints.Hex(entry.Body.vol.offset),
volatility3/framework/plugins/windows/hashdump.py
+2 -2
View File
@@ -68,7 +68,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
result = hive.get_key(key)
except KeyError:
vollog.info(
"Unable to load the required registry key {}\\{} from this memory image".format(hive.get_name(), key))
f"Unable to load the required registry key {hive.get_name()}\\{key} from this memory image")
return result
@classmethod
@@ -84,7 +84,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
@classmethod
def get_bootkey(cls, syshive: registry.RegistryHive) -> Optional[bytes]:
cs = 1
lsa_base = "ControlSet{0:03}".format(cs) + "\\Control\\Lsa"
lsa_base = f"ControlSet{cs:03}" + "\\Control\\Lsa"
lsa_keys = ["JD", "Skew1", "GBG", "Data"]
lsa = cls.get_hive_key(syshive, lsa_base)
volatility3/framework/plugins/windows/info.py
+2 -2
View File
@@ -162,7 +162,7 @@ class Info(plugins.PluginInterface):
yield (0, ("IsPAE", str(self.context.layers[layer_name].metadata.get("pae", False))))
for i, layer in self.get_depends(self.context, "primary"):
yield (0, (layer.name, "{} {}".format(i, layer.__class__.__name__)))
yield (0, (layer.name, f"{i} {layer.__class__.__name__}"))
if kdbg.Header.OwnerTag == 0x4742444B:
@@ -173,7 +173,7 @@ class Info(plugins.PluginInterface):
vers = self.get_version_structure(self.context, layer_name, symbol_table)
yield (0, ("KdVersionBlock", hex(vers.vol.offset)))
yield (0, ("Major/Minor", "{0}.{1}".format(vers.MajorVersion, vers.MinorVersion)))
yield (0, ("Major/Minor", f"{vers.MajorVersion}.{vers.MinorVersion}"))
yield (0, ("MachineType", str(vers.MachineType)))
ntkrnlmp = self.get_kernel_module(self.context, layer_name, symbol_table)
volatility3/framework/plugins/windows/memmap.py
+1 -1
View File
@@ -48,7 +48,7 @@ class Memmap(interfaces.plugins.PluginInterface):
excp.layer_name))
continue
file_handle = self.open("pid.{}.dmp".format(pid))
file_handle = self.open(f"pid.{pid}.dmp")
with file_handle as file_data:
file_offset = 0
for mapval in proc_layer.mapping(0x0, proc_layer.maximum_address, ignore_errors = True):
volatility3/framework/plugins/windows/modscan.py
+1 -1
View File
@@ -160,7 +160,7 @@ class ModScan(interfaces.plugins.PluginInterface):
if self.config['dump']:
session_layer_name = self.find_session_layer(self.context, session_layers, mod.DllBase)
file_output = "Cannot find a viable session layer for {0:#x}".format(mod.DllBase)
file_output = f"Cannot find a viable session layer for {mod.DllBase:#x}"
if session_layer_name:
file_handle = dlllist.DllList.dump_pe(self.context,
pe_table_name,
volatility3/framework/plugins/windows/netscan.py
+7 -7
View File
@@ -212,12 +212,12 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
latest_version = current_versions[-1]
filename = version_dict.get(latest_version)
vollog.debug("Unable to find exact matching symbol file, going with latest: {}".format(filename))
vollog.debug(f"Unable to find exact matching symbol file, going with latest: {filename}")
else:
raise NotImplementedError("This version of Windows is not supported: {}.{} {}.{}!".format(
nt_major_version, nt_minor_version, vers.MajorVersion, vers_minor_version))
vollog.debug("Determined symbol filename: {}".format(filename))
vollog.debug(f"Determined symbol filename: {filename}")
return filename, class_types
@@ -285,13 +285,13 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
for netw_obj in self.scan(self.context, self.config['primary'], self.config['nt_symbols'],
netscan_symbol_table):
vollog.debug("Found netw obj @ 0x{:2x} of assumed type {}".format(netw_obj.vol.offset, type(netw_obj)))
vollog.debug(f"Found netw obj @ 0x{netw_obj.vol.offset:2x} of assumed type {type(netw_obj)}")
# objects passed pool header constraints. check for additional constraints if strict flag is set.
if not show_corrupt_results and not netw_obj.is_valid():
continue
if isinstance(netw_obj, network._UDP_ENDPOINT):
vollog.debug("Found UDP_ENDPOINT @ 0x{:2x}".format(netw_obj.vol.offset))
vollog.debug(f"Found UDP_ENDPOINT @ 0x{netw_obj.vol.offset:2x}")
# For UdpA, the state is always blank and the remote end is asterisks
for ver, laddr, _ in netw_obj.dual_stack_sockets():
@@ -301,7 +301,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
or renderers.UnreadableValue()))
elif isinstance(netw_obj, network._TCP_ENDPOINT):
vollog.debug("Found _TCP_ENDPOINT @ 0x{:2x}".format(netw_obj.vol.offset))
vollog.debug(f"Found _TCP_ENDPOINT @ 0x{netw_obj.vol.offset:2x}")
if netw_obj.get_address_family() == network.AF_INET:
proto = "TCPv4"
elif netw_obj.get_address_family() == network.AF_INET6:
@@ -322,7 +322,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
# check for isinstance of tcp listener last, because all other objects are inherited from here
elif isinstance(netw_obj, network._TCP_LISTENER):
vollog.debug("Found _TCP_LISTENER @ 0x{:2x}".format(netw_obj.vol.offset))
vollog.debug(f"Found _TCP_LISTENER @ 0x{netw_obj.vol.offset:2x}")
# For TcpL, the state is always listening and the remote port is zero
for ver, laddr, raddr in netw_obj.dual_stack_sockets():
@@ -332,7 +332,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
or renderers.UnreadableValue()))
else:
# this should not happen therefore we log it.
vollog.debug("Found network object unsure of its type: {} of type {}".format(netw_obj, type(netw_obj)))
vollog.debug(f"Found network object unsure of its type: {netw_obj} of type {type(netw_obj)}")
def generate_timeline(self):
for row in self._generator():
volatility3/framework/plugins/windows/netstat.py
+11 -11
View File
@@ -128,7 +128,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
# invalid argument.
return
vollog.debug("Current Port: {}".format(port))
vollog.debug(f"Current Port: {port}")
# the given port serves as a shifted index into the port pool lists
list_index = port >> 8
truncated_port = port & 0xff
@@ -179,7 +179,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""
for mod in modules.Modules.list_modules(context, layer_name, nt_symbols):
if mod.BaseDllName.get_string() == "tcpip.sys":
vollog.debug("Found tcpip.sys image base @ 0x{:x}".format(mod.DllBase))
vollog.debug(f"Found tcpip.sys image base @ 0x{mod.DllBase:x}")
return mod
return None
@@ -255,7 +255,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
part_table_addr, part_count))
entry_offset = context.symbol_space.get_type(obj_name).relative_child_offset("ListEntry")
for ctr, partition in enumerate(part_table.Partitions):
vollog.debug("Parsing partition {}".format(ctr))
vollog.debug(f"Parsing partition {ctr}")
if partition.Endpoints.NumEntries > 0:
for endpoint_entry in cls.parse_hashtable(context, layer_name, partition.Endpoints.Directory,
partition.Endpoints.TableSize, alignment, net_symbol_table):
@@ -347,9 +347,9 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
# this branch should not be reached.
raise exceptions.SymbolError(
"UdpPortPool", tcpip_symbol_table,
"Neither UdpPortPool nor UdpCompartmentSet found in {} table".format(tcpip_symbol_table))
f"Neither UdpPortPool nor UdpCompartmentSet found in {tcpip_symbol_table} table")
vollog.debug("Found PortPools @ 0x{:x} (UDP) && 0x{:x} (TCP)".format(upp_addr, tpp_addr))
vollog.debug(f"Found PortPools @ 0x{upp_addr:x} (UDP) && 0x{tpp_addr:x} (TCP)")
return upp_addr, tpp_addr
@classmethod
@@ -399,8 +399,8 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
tcpl_ports = cls.parse_bitmap(context, layer_name, tpp_obj.PortBitMap.Buffer,
tpp_obj.PortBitMap.SizeOfBitMap // 8)
vollog.debug("Found TCP Ports: {}".format(tcpl_ports))
vollog.debug("Found UDP Ports: {}".format(udpa_ports))
vollog.debug(f"Found TCP Ports: {tcpl_ports}")
vollog.debug(f"Found UDP Ports: {udpa_ports}")
# given the list of TCP / UDP ports, calculate the address of their respective objects and yield them.
for port in tcpl_ports:
# port value can be 0, which we can skip
@@ -439,7 +439,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
continue
if isinstance(netw_obj, network._UDP_ENDPOINT):
vollog.debug("Found UDP_ENDPOINT @ 0x{:2x}".format(netw_obj.vol.offset))
vollog.debug(f"Found UDP_ENDPOINT @ 0x{netw_obj.vol.offset:2x}")
# For UdpA, the state is always blank and the remote end is asterisks
for ver, laddr, _ in netw_obj.dual_stack_sockets():
@@ -449,7 +449,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
or renderers.UnreadableValue()))
elif isinstance(netw_obj, network._TCP_ENDPOINT):
vollog.debug("Found _TCP_ENDPOINT @ 0x{:2x}".format(netw_obj.vol.offset))
vollog.debug(f"Found _TCP_ENDPOINT @ 0x{netw_obj.vol.offset:2x}")
if netw_obj.get_address_family() == network.AF_INET:
proto = "TCPv4"
elif netw_obj.get_address_family() == network.AF_INET6:
@@ -472,7 +472,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
# check for isinstance of tcp listener last, because all other objects are inherited from here
elif isinstance(netw_obj, network._TCP_LISTENER):
vollog.debug("Found _TCP_LISTENER @ 0x{:2x}".format(netw_obj.vol.offset))
vollog.debug(f"Found _TCP_LISTENER @ 0x{netw_obj.vol.offset:2x}")
# For TcpL, the state is always listening and the remote port is zero
for ver, laddr, raddr in netw_obj.dual_stack_sockets():
@@ -482,7 +482,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
or renderers.UnreadableValue()))
else:
# this should not happen therefore we log it.
vollog.debug("Found network object unsure of its type: {} of type {}".format(netw_obj, type(netw_obj)))
vollog.debug(f"Found network object unsure of its type: {netw_obj} of type {type(netw_obj)}")
def generate_timeline(self):
for row in self._generator():
volatility3/framework/plugins/windows/poolscanner.py
+4 -4
View File
@@ -141,7 +141,7 @@ class PoolScanner(plugins.PluginInterface):
try:
name = mem_object.FileName.String
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV, "Skipping file at {0:#x}".format(mem_object.vol.offset))
vollog.log(constants.LOGLEVEL_VVV, f"Skipping file at {mem_object.vol.offset:#x}")
continue
else:
name = renderers.NotApplicableValue()
@@ -298,7 +298,7 @@ class PoolScanner(plugins.PluginInterface):
kernel_symbol_table = symbol_table)
if mem_object is None:
vollog.log(constants.LOGLEVEL_VVV, "Cannot create an instance of {}".format(constraint.type_name))
vollog.log(constants.LOGLEVEL_VVV, f"Cannot create an instance of {constraint.type_name}")
continue
if constraint.object_type is not None and not constraint.skip_type_test:
@@ -307,7 +307,7 @@ class PoolScanner(plugins.PluginInterface):
continue
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV,
"Cannot test instance type check for {}".format(constraint.type_name))
f"Cannot test instance type check for {constraint.type_name}")
continue
yield constraint, mem_object, header
@@ -341,7 +341,7 @@ class PoolScanner(plugins.PluginInterface):
constraint_lookup = {} # type: Dict[bytes, PoolConstraint]
for constraint in pool_constraints:
if constraint.tag in constraint_lookup:
raise ValueError("Constraint tag is used for more than one constraint: {}".format(repr(constraint.tag)))
raise ValueError(f"Constraint tag is used for more than one constraint: {repr(constraint.tag)}")
constraint_lookup[constraint.tag] = constraint
pool_header_table_name = cls.get_pool_header_table(context, symbol_table)
volatility3/framework/plugins/windows/privileges.py
+1 -1
View File
@@ -64,7 +64,7 @@ class Privs(interfaces.plugins.PluginInterface):
# Skip privileges whose bit positions cannot be
# translated to a privilege name
if not self.privilege_info.get(int(value)):
vollog.log(constants.LOGLEVEL_VVV, 'Skeep invalid privilege ({}).'.format(value))
vollog.log(constants.LOGLEVEL_VVV, f'Skeep invalid privilege ({value}).')
continue
name, desc = self.privilege_info.get(int(value))
volatility3/framework/plugins/windows/pslist.py
+5 -5
View File
@@ -73,12 +73,12 @@ 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 = open_method("pid.{0}.{1:#x}.dmp".format(proc.UniqueProcessId, peb.ImageBaseAddress))
file_handle = open_method(f"pid.{proc.UniqueProcessId}.{peb.ImageBaseAddress:#x}.dmp")
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))
vollog.debug(f"Unable to dump PE with pid {proc.UniqueProcessId}: {excp}")
return file_handle
@@ -209,12 +209,12 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
proc.get_is_wow64(), proc.get_create_time(), proc.get_exit_time(), file_output))
except exceptions.InvalidAddressException:
vollog.info("Invalid process found at address: {:x}. Skipping".format(proc.vol.offset))
vollog.info(f"Invalid process found at address: {proc.vol.offset:x}. Skipping")
def generate_timeline(self):
for row in self._generator():
_depth, row_data = row
description = "Process: {} {} ({})".format(row_data[0], row_data[2], row_data[3])
description = f"Process: {row_data[0]} {row_data[2]} ({row_data[3]})"
yield (description, timeliner.TimeLinerType.CREATED, row_data[8])
yield (description, timeliner.TimeLinerType.MODIFIED, row_data[9])
@@ -222,7 +222,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
offsettype = "(V)" if not self.config.get('physical', self.PHYSICAL_DEFAULT) else "(P)"
return renderers.TreeGrid([("PID", int), ("PPID", int), ("ImageFileName", str),
("Offset{0}".format(offsettype), format_hints.Hex), ("Threads", int),
(f"Offset{offsettype}", format_hints.Hex), ("Threads", int),
("Handles", int), ("SessionId", int), ("Wow64", bool),
("CreateTime", datetime.datetime), ("ExitTime", datetime.datetime),
("File output", str)], self._generator())
volatility3/framework/plugins/windows/psscan.py
+1 -1
View File
@@ -178,7 +178,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
def generate_timeline(self):
for row in self._generator():
_depth, row_data = row
description = "Process: {} {} ({})".format(row_data[0], row_data[2], row_data[3])
description = f"Process: {row_data[0]} {row_data[2]} ({row_data[3]})"
yield (description, timeliner.TimeLinerType.CREATED, row_data[8])
yield (description, timeliner.TimeLinerType.MODIFIED, row_data[9])
volatility3/framework/plugins/windows/pstree.py
+1 -1
View File
@@ -90,7 +90,7 @@ class PsTree(interfaces.plugins.PluginInterface):
offsettype = "(V)" if not self.config.get('physical', pslist.PsList.PHYSICAL_DEFAULT) else "(P)"
return renderers.TreeGrid([("PID", int), ("PPID", int), ("ImageFileName", str),
("Offset{0}".format(offsettype), format_hints.Hex), ("Threads", int),
(f"Offset{offsettype}", format_hints.Hex), ("Threads", int),
("Handles", int), ("SessionId", int), ("Wow64", bool),
("CreateTime", datetime.datetime), ("ExitTime", datetime.datetime)],
self._generator())
volatility3/framework/plugins/windows/registry/hivelist.py
+2 -2
View File
@@ -83,7 +83,7 @@ class HiveList(interfaces.plugins.PluginInterface):
maxaddr = hive.hive.Storage[0].Length
hive_name = self._sanitize_hive_name(hive.get_name())
file_handle = self.open('registry.{}.{}.hive'.format(hive_name, hex(hive.hive_offset)))
file_handle = self.open(f'registry.{hive_name}.{hex(hive.hive_offset)}.hive')
with file_handle as file_data:
if hive._base_block:
hive_data = self.context.layers[hive.dependencies[0]].read(hive.hive.BaseBlock, 1 << 12)
@@ -143,7 +143,7 @@ class HiveList(interfaces.plugins.PluginInterface):
try:
hive = registry.RegistryHive(context, reg_config_path, name = 'hive' + hex(hive_offset))
except exceptions.InvalidAddressException:
vollog.warning("Couldn't create RegistryHive layer at offset {}, skipping".format(hex(hive_offset)))
vollog.warning(f"Couldn't create RegistryHive layer at offset {hex(hive_offset)}, skipping")
continue
context.layers.add_layer(hive)
yield hive
volatility3/framework/plugins/windows/registry/printkey.py
+2 -2
View File
@@ -176,11 +176,11 @@ class PrintKey(interfaces.plugins.PluginInterface):
yield (x - len(node_path), y)
except (exceptions.InvalidAddressException, KeyError, RegistryFormatException) as excp:
if isinstance(excp, KeyError):
vollog.debug("Key '{}' not found in Hive at offset {}.".format(key, hex(hive.hive_offset)))
vollog.debug(f"Key '{key}' not found in Hive at offset {hex(hive.hive_offset)}.")
elif isinstance(excp, RegistryFormatException):
vollog.debug(excp)
elif isinstance(excp, exceptions.InvalidAddressException):
vollog.debug("Invalid address identified in Hive: {}".format(hex(excp.invalid_address)))
vollog.debug(f"Invalid address identified in Hive: {hex(excp.invalid_address)}")
result = (0, (renderers.UnreadableValue(), format_hints.Hex(hive.hive_offset), "Key",
'?\\' + (key or ''), renderers.UnreadableValue(), renderers.UnreadableValue(),
renderers.UnreadableValue()))
volatility3/framework/plugins/windows/registry/userassist.py
+1 -1
View File
@@ -227,7 +227,7 @@ class UserAssist(interfaces.plugins.PluginInterface):
yield from self.list_userassist(hive)
continue
except exceptions.PagedInvalidAddressException as excp:
vollog.debug("Invalid address identified in Hive: {}".format(hex(excp.invalid_address)))
vollog.debug(f"Invalid address identified in Hive: {hex(excp.invalid_address)}")
except exceptions.InvalidAddressException as excp:
vollog.debug("Invalid address identified in lower layer {}: {}".format(
excp.layer_name, excp.invalid_address))
volatility3/framework/plugins/windows/strings.py
+3 -3
View File
@@ -57,7 +57,7 @@ class Strings(interfaces.plugins.PluginInterface):
offset, string = self._parse_line(line)
string_list.append((offset, string))
except ValueError:
vollog.error("Line in unrecognized format: line {}".format(count))
vollog.error(f"Line in unrecognized format: line {count}")
line = strings_fp.readline()
revmap = self.generate_mapping(self.context,
@@ -150,11 +150,11 @@ class Strings(interfaces.plugins.PluginInterface):
mapped_offset, _, offset, mapped_size, maplayer = mapval
for val in range(mapped_offset, mapped_offset + mapped_size, 0x1000):
cur_set = reverse_map.get(mapped_offset >> 12, set())
cur_set.add(("Process {}".format(process.UniqueProcessId), offset))
cur_set.add((f"Process {process.UniqueProcessId}", offset))
reverse_map[mapped_offset >> 12] = cur_set
# FIXME: make the progress for all processes, rather than per-process
if progress_callback:
progress_callback((offset * 100) / layer.maximum_address,
"Creating mapping for task {}".format(process.UniqueProcessId))
f"Creating mapping for task {process.UniqueProcessId}")
return reverse_map
volatility3/framework/plugins/windows/symlinkscan.py
+1 -1
View File
@@ -68,7 +68,7 @@ class SymlinkScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfa
def generate_timeline(self):
for row in self._generator():
_depth, row_data = row
description = "Symlink: {} -> {}".format(row_data[2], row_data[3])
description = f"Symlink: {row_data[2]} -> {row_data[3]}"
yield (description, timeliner.TimeLinerType.CREATED, row_data[1])
def run(self):
volatility3/framework/plugins/windows/vadinfo.py
+3 -3
View File
@@ -135,7 +135,7 @@ class VadInfo(interfaces.plugins.PluginInterface):
return None
if maxsize > 0 and (vad_end - vad_start) > maxsize:
vollog.debug("Skip VAD dump {0:#x}-{1:#x} due to maxsize limit".format(vad_start, vad_end))
vollog.debug(f"Skip VAD dump {vad_start:#x}-{vad_end:#x} due to maxsize limit")
return None
proc_id = "Unknown"
@@ -148,7 +148,7 @@ class VadInfo(interfaces.plugins.PluginInterface):
return None
proc_layer = context.layers[proc_layer_name]
file_name = "pid.{0}.vad.{1:#x}-{2:#x}.dmp".format(proc_id, vad_start, vad_end)
file_name = f"pid.{proc_id}.vad.{vad_start:#x}-{vad_end:#x}.dmp"
try:
file_handle = open_method(file_name)
chunk_size = 1024 * 1024 * 10
@@ -162,7 +162,7 @@ class VadInfo(interfaces.plugins.PluginInterface):
offset += to_read
except Exception as excp:
vollog.debug("Unable to dump VAD {}: {}".format(file_name, excp))
vollog.debug(f"Unable to dump VAD {file_name}: {excp}")
return None
return file_handle
volatility3/framework/plugins/yarascan.py
+2 -2
View File
@@ -75,12 +75,12 @@ class YaraScan(plugins.PluginInterface):
if config.get('yara_rules', None) is not None:
rule = config['yara_rules']
if rule[0] not in ["{", "/"]:
rule = '"{}"'.format(rule)
rule = f'"{rule}"'
if config.get('case', False):
rule += " nocase"
if config.get('wide', False):
rule += " wide ascii"
rules = yara.compile(sources = {'n': 'rule r1 {{strings: $a = {} condition: $a}}'.format(rule)})
rules = yara.compile(sources = {'n': f'rule r1 {{strings: $a = {rule} condition: $a}}'})
elif config.get('yara_file', None) is not None:
rules = yara.compile(file = resources.ResourceAccessor().open(config['yara_file'], "rb"))
elif config.get('yara_compiled_file', None) is not None:
volatility3/framework/renderers/__init__.py
+3 -3
View File
@@ -59,7 +59,7 @@ class TreeNode(interfaces.renderers.TreeNode):
self._values = treegrid.RowStructure(*values) # type: ignore
def __repr__(self) -> str:
return "<TreeNode [{}] - {}>".format(self.path, self._values)
return f"<TreeNode [{self.path}] - {self._values}>"
def __getitem__(self, item: Union[int, slice]) -> Any:
return self._treegrid.children(self).__getitem__(item)
@@ -219,7 +219,7 @@ class TreeGrid(interfaces.renderers.TreeGrid):
except Exception as excp:
if fail_on_errors:
raise
vollog.debug("Exception during population: {}".format(excp))
vollog.debug(f"Exception during population: {excp}")
self._populated = True
return excp
self._populated = True
@@ -363,7 +363,7 @@ class ColumnSortKey(interfaces.renderers.ColumnSortKey):
_index = i
self._type = column.type
if _index is None:
raise ValueError("Column not found in TreeGrid columns: {}".format(column_name))
raise ValueError(f"Column not found in TreeGrid columns: {column_name}")
self._index = _index
def __call__(self, values: List[Any]) -> Any:
volatility3/framework/symbols/__init__.py
+6 -6
View File
@@ -117,7 +117,7 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
"""
def __init__(self, type_name: str, **kwargs) -> None:
vollog.debug("Unresolved reference: {}".format(type_name))
vollog.debug(f"Unresolved reference: {type_name}")
super().__init__(type_name = type_name, **kwargs)
def _weak_resolve(self, resolve_type: SymbolType, name: str) -> SymbolSpaceReturnType:
@@ -139,8 +139,8 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
return getattr(self._dict[table_name], get_function)(component_name)
except KeyError as e:
raise exceptions.SymbolError(component_name, table_name,
'Type {} references missing Type/Symbol/Enum: {}'.format(name, e))
raise exceptions.SymbolError(name, None, "Malformed name: {}".format(name))
f'Type {name} references missing Type/Symbol/Enum: {e}')
raise exceptions.SymbolError(name, None, f"Malformed name: {name}")
def _iterative_resolve(self, traverse_list):
"""Iteratively resolves a type, populating linked child
@@ -185,7 +185,7 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
index = type_name.find(constants.BANG)
if index > 0:
table_name, type_name = type_name[:index], type_name[index + 1:]
raise exceptions.SymbolError(type_name, table_name, "Unresolvable symbol requested: {}".format(type_name))
raise exceptions.SymbolError(type_name, table_name, f"Unresolvable symbol requested: {type_name}")
return self._resolved[type_name]
def get_symbol(self, symbol_name: str) -> interfaces.symbols.SymbolInterface:
@@ -198,7 +198,7 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
index = symbol_name.find(constants.BANG)
if index > 0:
table_name, symbol_name = symbol_name[:index], symbol_name[index + 1:]
raise exceptions.SymbolError(symbol_name, table_name, "Unresolvable Symbol: {}".format(symbol_name))
raise exceptions.SymbolError(symbol_name, table_name, f"Unresolvable Symbol: {symbol_name}")
return retval
def _subresolve(self, object_template: interfaces.objects.Template) -> interfaces.objects.Template:
@@ -220,7 +220,7 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
index = enum_name.find(constants.BANG)
if index > 0:
table_name, enum_name = enum_name[:index], enum_name[index + 1:]
raise exceptions.SymbolError(enum_name, table_name, "Unresolvable Enumeration: {}".format(enum_name))
raise exceptions.SymbolError(enum_name, table_name, f"Unresolvable Enumeration: {enum_name}")
return retval
def _membership(self, member_type: SymbolType, name: str) -> bool:
volatility3/framework/symbols/intermed.py
+19 -19
View File
@@ -111,7 +111,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
# Validation is expensive, but we cache to store the hashes of successfully validated json objects
if validate and not schemas.validate(json_object):
raise exceptions.SymbolSpaceError("File does not pass version validation: {}".format(isf_url))
raise exceptions.SymbolSpaceError(f"File does not pass version validation: {isf_url}")
metadata = json_object.get('metadata', None)
@@ -123,7 +123,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
raise RuntimeError("ISF version {} is no longer supported: {}".format(metadata.get('format', "0.0.0"),
isf_url))
elif self._delegate.version < constants.ISF_MINIMUM_DEPRECATED:
vollog.warning("ISF version {} has been deprecated: {}".format(metadata.get('format', "0.0.0"), isf_url))
vollog.warning(f"ISF version {metadata.get('format', '0.0.0')} has been deprecated: {isf_url}")
# Inherit
super().__init__(context,
@@ -154,7 +154,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
supported_versions = [x for x in versions if x[0] == major and x[1] >= minor]
if not supported_versions:
raise ValueError(
"No Intermediate Format interface versions support file interface version: {}".format(version))
f"No Intermediate Format interface versions support file interface version: {version}")
return versions[max(supported_versions)]
symbols = _construct_delegate_function('symbols', True)
@@ -188,7 +188,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
zip_match = "/".join(os.path.split(filename))
# Check user symbol directory first, then fallback to the framework's library to allow for overloading
vollog.log(constants.LOGLEVEL_VVVV, "Searching for symbols in {}".format(", ".join(symbols.__path__)))
vollog.log(constants.LOGLEVEL_VVVV, f"Searching for symbols in {', '.join(symbols.__path__)}")
for path in symbols.__path__:
if not os.path.isabs(path):
path = os.path.abspath(os.path.join(__file__, path))
@@ -300,7 +300,7 @@ class ISFormatTable(interfaces.symbols.SymbolTableInterface, metaclass = ABCMeta
# TODO: determine whether we should give voids a size - We don't give voids a length, whereas microsoft seemingly do
pass
else:
vollog.debug("Choosing appropriate natives for symbol library: {}".format(nc))
vollog.debug(f"Choosing appropriate natives for symbol library: {nc}")
return native_class.natives
return None
@@ -335,7 +335,7 @@ class Version1Format(ISFormatTable):
return self._symbol_cache[name]
symbol = self._json_object['symbols'].get(name, None)
if not symbol:
raise exceptions.SymbolError(name, self.name, "Unknown symbol: {}".format(name))
raise exceptions.SymbolError(name, self.name, f"Unknown symbol: {name}")
address = symbol['address'] + self.config.get('symbol_shift', 0)
if self.config.get('symbol_mask', 0):
address = address & self.config['symbol_mask']
@@ -362,7 +362,7 @@ class Version1Format(ISFormatTable):
def set_type_class(self, name: str, clazz: Type[interfaces.objects.ObjectInterface]) -> None:
if name not in self.types:
raise ValueError("Symbol type not in {} SymbolTable: {}".format(self.name, name))
raise ValueError(f"Symbol type not in {self.name} SymbolTable: {name}")
self._overrides[name] = clazz
def del_type_class(self, name: str) -> None:
@@ -372,7 +372,7 @@ class Version1Format(ISFormatTable):
def _interdict_to_template(self, dictionary: Dict[str, Any]) -> interfaces.objects.Template:
"""Converts an intermediate format dict into an object template."""
if not dictionary:
raise exceptions.SymbolSpaceError("Invalid intermediate dictionary: {}".format(dictionary))
raise exceptions.SymbolSpaceError(f"Invalid intermediate dictionary: {dictionary}")
type_name = dictionary['kind']
if type_name == 'base':
@@ -407,7 +407,7 @@ class Version1Format(ISFormatTable):
# Otherwise
if dictionary['kind'] not in objects.AggregateTypes.values():
raise exceptions.SymbolSpaceError("Unknown Intermediate format: {}".format(dictionary))
raise exceptions.SymbolSpaceError(f"Unknown Intermediate format: {dictionary}")
reference_name = dictionary['name']
if constants.BANG not in reference_name:
@@ -424,7 +424,7 @@ class Version1Format(ISFormatTable):
parameters for an Enum."""
lookup = self._json_object['enums'].get(name, None)
if not lookup:
raise exceptions.SymbolSpaceError("Unknown enumeration: {}".format(name))
raise exceptions.SymbolSpaceError(f"Unknown enumeration: {name}")
result = {"choices": copy.deepcopy(lookup['constants']), "base_type": self.natives.get_type(lookup['base'])}
return result
@@ -432,11 +432,11 @@ class Version1Format(ISFormatTable):
"""Resolves an individual enumeration."""
if constants.BANG in enum_name:
raise exceptions.SymbolError(enum_name, self.name,
"Enumeration for a different table requested: {}".format(enum_name))
f"Enumeration for a different table requested: {enum_name}")
if enum_name not in self._json_object['enums']:
# Fall back to the natives table
raise exceptions.SymbolError(enum_name, self.name,
"Enumeration not found in {} table: {}".format(self.name, enum_name))
f"Enumeration not found in {self.name} table: {enum_name}")
curdict = self._json_object['enums'][enum_name]
base_type = self.natives.get_type(curdict['base'])
# The size isn't actually used, the base-type defines it.
@@ -452,7 +452,7 @@ class Version1Format(ISFormatTable):
table_name, type_name = type_name[:index], type_name[index + 1:]
raise exceptions.SymbolError(
type_name, table_name,
"Symbol for a different table requested: {}".format(table_name + constants.BANG + type_name))
f"Symbol for a different table requested: {table_name + constants.BANG + type_name}")
if type_name not in self._json_object['user_types']:
# Fall back to the natives table
return self.natives.get_type(self.name + constants.BANG + type_name)
@@ -491,7 +491,7 @@ class Version2Format(Version1Format):
# TODO: determine whether we should give voids a size - We don't give voids a length, whereas microsoft seemingly do
pass
else:
vollog.debug("Choosing appropriate natives for symbol library: {}".format(nc))
vollog.debug(f"Choosing appropriate natives for symbol library: {nc}")
return native_class.natives
return None
@@ -502,13 +502,13 @@ class Version2Format(Version1Format):
table_name, type_name = type_name[:index], type_name[index + 1:]
raise exceptions.SymbolError(
type_name, table_name,
"Symbol for a different table requested: {}".format(table_name + constants.BANG + type_name))
f"Symbol for a different table requested: {table_name + constants.BANG + type_name}")
if type_name not in self._json_object['user_types']:
# Fall back to the natives table
if type_name in self.natives.types:
return self.natives.get_type(self.name + constants.BANG + type_name)
else:
raise exceptions.SymbolError(type_name, self.name, "Unknown symbol: {}".format(type_name))
raise exceptions.SymbolError(type_name, self.name, f"Unknown symbol: {type_name}")
curdict = self._json_object['user_types'][type_name]
members = {}
for member_name in curdict['fields']:
@@ -536,7 +536,7 @@ class Version3Format(Version2Format):
return self._symbol_cache[name]
symbol = self._json_object['symbols'].get(name, None)
if not symbol:
raise exceptions.SymbolError(name, self.name, "Unknown symbol: {}".format(name))
raise exceptions.SymbolError(name, self.name, f"Unknown symbol: {name}")
symbol_type = None
if 'type' in symbol:
symbol_type = self._interdict_to_template(symbol['type'])
@@ -593,7 +593,7 @@ class Version5Format(Version4Format):
return self._symbol_cache[name]
symbol = self._json_object['symbols'].get(name, None)
if not symbol:
raise exceptions.SymbolError(name, self.name, "Unknown symbol: {}".format(name))
raise exceptions.SymbolError(name, self.name, f"Unknown symbol: {name}")
symbol_type = None
if 'type' in symbol:
symbol_type = self._interdict_to_template(symbol['type'])
@@ -666,7 +666,7 @@ class Version8Format(Version7Format):
table_name, type_name = type_name[:index], type_name[index + 1:]
raise exceptions.SymbolError(
type_name, table_name,
"Symbol for a different table requested: {}".format(table_name + constants.BANG + type_name))
f"Symbol for a different table requested: {table_name + constants.BANG + type_name}")
type_definition = self._json_object['user_types'].get(type_name)
if type_definition is None:
volatility3/framework/symbols/linux/__init__.py
+4 -4
View File
@@ -83,7 +83,7 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
except exceptions.InvalidAddressException:
ino = 0
ret_val = ret_val[:-1] + ":[{0}]".format(ino)
ret_val = ret_val[:-1] + f":[{ino}]"
else:
ret_val = ret_val.replace("/", "")
@@ -132,12 +132,12 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
pre_name = cls._get_path_file(task, filp)
else:
pre_name = "<unsupported d_op symbol: {0}>".format(sym)
pre_name = f"<unsupported d_op symbol: {sym}>"
ret = "{0}:[{1:d}]".format(pre_name, dentry.d_inode.i_ino)
ret = f"{pre_name}:[{dentry.d_inode.i_ino:d}]"
else:
ret = "<invalid d_dname pointer> {0:x}".format(sym_addr)
ret = f"<invalid d_dname pointer> {sym_addr:x}"
return ret
volatility3/framework/symbols/linux/extensions/__init__.py
+3 -3
View File
@@ -181,7 +181,7 @@ class task_struct(generic.GenericIntelProcess):
return None
if preferred_name is None:
preferred_name = self.vol.layer_name + "_Process{}".format(self.pid)
preferred_name = self.vol.layer_name + f"_Process{self.pid}"
# Add the constructed layer and return the name
return self._add_process_layer(self._context, dtb, config_prefix, preferred_name)
@@ -197,7 +197,7 @@ class task_struct(generic.GenericIntelProcess):
continue
else:
# FIXME: Check if this actually needs to be printed out or not
vollog.info("adding vma: {:x} {:x} | {:x} {:x}".format(start, self.mm.brk, end, self.mm.start_brk))
vollog.info(f"adding vma: {start:x} {self.mm.brk:x} | {end:x} {self.mm.start_brk:x}")
yield (start, end - start)
@@ -496,7 +496,7 @@ class vfsmount(objects.StructType):
def _get_real_mnt(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")
return self._context.object(mount_struct, self.vol.layer_name, offset = self.vol.offset - offset)
volatility3/framework/symbols/linux/extensions/elf.py
+1 -1
View File
@@ -45,7 +45,7 @@ class elf(objects.StructType):
elif ei_class == 2:
self._type_prefix = "Elf64_"
else:
raise ValueError("Unsupported ei_class value {}".format(ei_class))
raise ValueError(f"Unsupported ei_class value {ei_class}")
# Construct the full header
self._hdr = self._context.object(symbol_table_name + constants.BANG + self._type_prefix + "Ehdr",
volatility3/framework/symbols/mac/__init__.py
+1 -1
View File
@@ -149,7 +149,7 @@ class MacUtilities(interfaces.configuration.VersionableInterface):
vnode = f.f_fglob.fg_data.dereference().cast("vnode")
path = vnode.full_path()
elif ftype:
path = "<{}>".format(ftype.lower())
path = f"<{ftype.lower()}>"
yield f, path, fd_num
volatility3/framework/symbols/mac/extensions/__init__.py
+2 -2
View File
@@ -32,7 +32,7 @@ class proc(generic.GenericIntelProcess):
return None
if preferred_name is None:
preferred_name = self.vol.layer_name + "_Process{}".format(self.p_pid)
preferred_name = self.vol.layer_name + f"_Process{self.p_pid}"
# Add the constructed layer and return the name
return self._add_process_layer(self._context, dtb, config_prefix, preferred_name)
@@ -478,7 +478,7 @@ class sockaddr_dl(objects.StructType):
e = e.cast("unsigned char")
ret = ret + "{:02X}:".format(e)
ret = ret + f"{e:02X}:"
if ret and ret[-1] == ":":
ret = ret[:-1]
volatility3/framework/symbols/native.py
+1 -1
View File
@@ -45,7 +45,7 @@ class NativeTable(interfaces.symbols.NativeTableInterface):
if constants.BANG in type_name:
name_split = type_name.split(constants.BANG)
if len(name_split) > 2:
raise ValueError("SymbolName cannot contain multiple {} separators".format(constants.BANG))
raise ValueError(f"SymbolName cannot contain multiple {constants.BANG} separators")
table_name, type_name = name_split
prefix = table_name + constants.BANG
volatility3/framework/symbols/windows/extensions/__init__.py
+15 -15
View File
@@ -94,7 +94,7 @@ class MMVAD_SHORT(objects.StructType):
# any node other than the root that doesn't have a recognized tag
# is just garbage and we skip the node entirely
vollog.log(constants.LOGLEVEL_VVV,
"Skipping VAD at {} depth {} with tag {}".format(self.vol.offset, depth, tag))
f"Skipping VAD at {self.vol.offset} depth {depth} with tag {tag}")
return
if target:
@@ -105,13 +105,13 @@ class MMVAD_SHORT(objects.StructType):
for vad_node in self.get_left_child().dereference().traverse(visited, depth + 1):
yield vad_node
except exceptions.InvalidAddressException as excp:
vollog.log(constants.LOGLEVEL_VVV, "Invalid address on LeftChild: {0:#x}".format(excp.invalid_address))
vollog.log(constants.LOGLEVEL_VVV, f"Invalid address on LeftChild: {excp.invalid_address:#x}")
try:
for vad_node in self.get_right_child().dereference().traverse(visited, depth + 1):
yield vad_node
except exceptions.InvalidAddressException as excp:
vollog.log(constants.LOGLEVEL_VVV, "Invalid address on RightChild: {0:#x}".format(excp.invalid_address))
vollog.log(constants.LOGLEVEL_VVV, f"Invalid address on RightChild: {excp.invalid_address:#x}")
def get_right_child(self):
"""Get the right child member."""
@@ -329,7 +329,7 @@ class EX_FAST_REF(objects.StructType):
def dereference(self) -> interfaces.objects.ObjectInterface:
if constants.BANG not in self.vol.type_name:
raise ValueError("Invalid symbol table name syntax (no {} found)".format(constants.BANG))
raise ValueError(f"Invalid symbol table name syntax (no {constants.BANG} found)")
# the mask value is different on 32 and 64 bits
symbol_table_name = self.vol.type_name.split(constants.BANG)[0]
@@ -394,7 +394,7 @@ class FILE_OBJECT(objects.StructType, pool.ExecutiveObject):
# be instantiated from a primary (virtual) layer or a memory (physical) layer.
if self._context.layers[self.vol.native_layer_name].is_valid(self.DeviceObject):
try:
name = "\\Device\\{}".format(self.DeviceObject.get_device_name())
name = f"\\Device\\{self.DeviceObject.get_device_name()}"
except ValueError:
pass
@@ -449,7 +449,7 @@ class ETHREAD(objects.StructType):
stringCrossThreadFlags = ''
for flag in dictCrossThreadFlags:
if flags & 2 ** dictCrossThreadFlags[flag]:
stringCrossThreadFlags += '{} '.format(flag)
stringCrossThreadFlags += f'{flag} '
return stringCrossThreadFlags[:-1] if stringCrossThreadFlags else stringCrossThreadFlags
@@ -534,7 +534,7 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
dtb = dtb & ((1 << parent_layer.bits_per_register) - 1)
if preferred_name is None:
preferred_name = self.vol.layer_name + "_Process{}".format(self.UniqueProcessId)
preferred_name = self.vol.layer_name + f"_Process{self.UniqueProcessId}"
# Add the constructed layer and return the name
return self._add_process_layer(self._context, dtb, config_prefix, preferred_name)
@@ -542,7 +542,7 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
def get_peb(self) -> interfaces.objects.ObjectInterface:
"""Constructs a PEB object"""
if constants.BANG not in self.vol.type_name:
raise ValueError("Invalid symbol table name syntax (no {} found)".format(constants.BANG))
raise ValueError(f"Invalid symbol table name syntax (no {constants.BANG} found)")
# add_process_layer can raise InvalidAddressException.
# if that happens, we let the exception propagate upwards
@@ -551,10 +551,10 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
proc_layer = self._context.layers[proc_layer_name]
if not proc_layer.is_valid(self.Peb):
raise exceptions.InvalidAddressException(proc_layer_name, self.Peb,
"Invalid address at {:0x}".format(self.Peb))
f"Invalid address at {self.Peb:0x}")
sym_table = self.vol.type_name.split(constants.BANG)[0]
peb = self._context.object("{}{}_PEB".format(sym_table, constants.BANG),
peb = self._context.object(f"{sym_table}{constants.BANG}_PEB",
layer_name = proc_layer_name,
offset = self.Peb)
return peb
@@ -565,7 +565,7 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
try:
peb = self.get_peb()
for entry in peb.Ldr.InLoadOrderModuleList.to_list(
"{}{}_LDR_DATA_TABLE_ENTRY".format(self.get_symbol_table_name(), constants.BANG),
f"{self.get_symbol_table_name()}{constants.BANG}_LDR_DATA_TABLE_ENTRY",
"InLoadOrderLinks"):
yield entry
except exceptions.InvalidAddressException:
@@ -577,7 +577,7 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
try:
peb = self.get_peb()
for entry in peb.Ldr.InInitializationOrderModuleList.to_list(
"{}{}_LDR_DATA_TABLE_ENTRY".format(self.get_symbol_table_name(), constants.BANG),
f"{self.get_symbol_table_name()}{constants.BANG}_LDR_DATA_TABLE_ENTRY",
"InInitializationOrderLinks"):
yield entry
except exceptions.InvalidAddressException:
@@ -589,7 +589,7 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
try:
peb = self.get_peb()
for entry in peb.Ldr.InMemoryOrderModuleList.to_list(
"{}{}_LDR_DATA_TABLE_ENTRY".format(self.get_symbol_table_name(), constants.BANG),
f"{self.get_symbol_table_name()}{constants.BANG}_LDR_DATA_TABLE_ENTRY",
"InMemoryOrderLinks"):
yield entry
except exceptions.InvalidAddressException:
@@ -603,7 +603,7 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV,
"Cannot access _EPROCESS.ObjectTable.HandleCount at {0:#x}".format(self.vol.offset))
f"Cannot access _EPROCESS.ObjectTable.HandleCount at {self.vol.offset:#x}")
return renderers.UnreadableValue()
@@ -626,7 +626,7 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV,
"Cannot access _EPROCESS.Session.SessionId at {0:#x}".format(self.vol.offset))
f"Cannot access _EPROCESS.Session.SessionId at {self.vol.offset:#x}")
return renderers.UnreadableValue()
volatility3/framework/symbols/windows/extensions/network.py
+5 -5
View File
@@ -164,7 +164,7 @@ class _TCP_LISTENER(objects.StructType):
return False
except exceptions.InvalidAddressException:
vollog.debug("netw obj 0x{:x} invalid due to invalid address access".format(self.vol.offset))
vollog.debug(f"netw obj 0x{self.vol.offset:x} invalid due to invalid address access")
return False
return True
@@ -200,21 +200,21 @@ class _TCP_ENDPOINT(_TCP_LISTENER):
def is_valid(self):
if self.State not in self.State.choices.values():
vollog.debug("{} 0x{:x} invalid due to invalid tcp state {}".format(type(self), self.vol.offset, self.State))
vollog.debug(f"{type(self)} 0x{self.vol.offset:x} invalid due to invalid tcp state {self.State}")
return False
try:
if self.get_address_family() not in (AF_INET, AF_INET6):
vollog.debug("{} 0x{:x} invalid due to invalid address_family {}".format(type(self), self.vol.offset, self.get_address_family()))
vollog.debug(f"{type(self)} 0x{self.vol.offset:x} invalid due to invalid address_family {self.get_address_family()}")
return False
if not self.get_local_address() and (not self.get_owner() or self.get_owner().UniqueProcessId == 0
or self.get_owner().UniqueProcessId > 65535):
vollog.debug("{} 0x{:x} invalid due to invalid owner data".format(type(self), self.vol.offset))
vollog.debug(f"{type(self)} 0x{self.vol.offset:x} invalid due to invalid owner data")
return False
except exceptions.InvalidAddressException:
vollog.debug("{} 0x{:x} invalid due to invalid address access".format(type(self), self.vol.offset))
vollog.debug(f"{type(self)} 0x{self.vol.offset:x} invalid due to invalid address access")
return False
return True
volatility3/framework/symbols/windows/extensions/pe.py
+6 -6
View File
@@ -22,7 +22,7 @@ class IMAGE_DOS_HEADER(objects.StructType):
"""
if self.e_magic != 0x5a4d:
raise ValueError("e_magic {0:04X} is not a valid DOS signature.".format(self.e_magic))
raise ValueError(f"e_magic {self.e_magic:04X} is not a valid DOS signature.")
layer_name = self.vol.layer_name
symbol_table_name = self.get_symbol_table_name()
@@ -32,7 +32,7 @@ class IMAGE_DOS_HEADER(objects.StructType):
offset = self.vol.offset + self.e_lfanew)
if nt_header.Signature != 0x4550:
raise ValueError("NT header signature {0:04X} is not a valid".format(nt_header.Signature))
raise ValueError(f"NT header signature {nt_header.Signature:04X} is not a valid")
# this checks if we need a PE32+ header
if nt_header.FileHeader.Machine == 34404:
@@ -110,7 +110,7 @@ class IMAGE_DOS_HEADER(objects.StructType):
# no legitimate PE is going to be larger than this
if size_of_image > (1024 * 1024 * 100):
raise ValueError("The claimed SizeOfImage is too large: {}".format(size_of_image))
raise ValueError(f"The claimed SizeOfImage is too large: {size_of_image}")
read_layer = self._context.layers[layer_name]
@@ -127,13 +127,13 @@ class IMAGE_DOS_HEADER(objects.StructType):
for sect in nt_header.get_sections():
if sect.VirtualAddress > size_of_image:
raise ValueError("Section VirtualAddress is too large: {}".format(sect.VirtualAddress))
raise ValueError(f"Section VirtualAddress is too large: {sect.VirtualAddress}")
if sect.Misc.VirtualSize > size_of_image:
raise ValueError("Section VirtualSize is too large: {}".format(sect.Misc.VirtualSize))
raise ValueError(f"Section VirtualSize is too large: {sect.Misc.VirtualSize}")
if sect.SizeOfRawData > size_of_image:
raise ValueError("Section SizeOfRawData is too large: {}".format(sect.SizeOfRawData))
raise ValueError(f"Section SizeOfRawData is too large: {sect.SizeOfRawData}")
if sect is not None:
# It doesn't matter if this is too big, because it'll get overwritten by the later layers
volatility3/framework/symbols/windows/extensions/pool.py
+5 -5
View File
@@ -175,7 +175,7 @@ class POOL_HEADER(objects.StructType):
'HANDLE_REVOCATION_INFO', 'PADDING_INFO'
]:
try:
type_name = "{}{}_OBJECT_HEADER_{}".format(symbol_table_name, constants.BANG, header)
type_name = f"{symbol_table_name}{constants.BANG}_OBJECT_HEADER_{header}"
header_type = context.symbol_space.get_type(type_name)
headers.append(header)
sizes.append(header_type.size)
@@ -240,7 +240,7 @@ class POOL_TRACKER_BIG_PAGES(objects.StructType):
if hasattr(self, 'PoolType'):
if not self.pool_type_lookup:
self._generate_pool_type_lookup()
return self.pool_type_lookup.get(self.PoolType, "Unknown choice {}".format(self.PoolType))
return self.pool_type_lookup.get(self.PoolType, f"Unknown choice {self.PoolType}")
else:
return renderers.NotApplicableValue()
@@ -259,7 +259,7 @@ class ExecutiveObject(interfaces.objects.ObjectInterface):
def get_object_header(self) -> 'OBJECT_HEADER':
if constants.BANG not in self.vol.type_name:
raise ValueError("Invalid symbol table name syntax (no {} found)".format(constants.BANG))
raise ValueError(f"Invalid symbol table name syntax (no {constants.BANG} found)")
symbol_table_name = self.vol.type_name.split(constants.BANG)[0]
body_offset = self._context.symbol_space.get_type(symbol_table_name + constants.BANG +
"_OBJECT_HEADER").relative_child_offset("Body")
@@ -315,7 +315,7 @@ class OBJECT_HEADER(objects.StructType):
@property
def NameInfo(self) -> interfaces.objects.ObjectInterface:
if constants.BANG not in self.vol.type_name:
raise ValueError("Invalid symbol table name syntax (no {} found)".format(constants.BANG))
raise ValueError(f"Invalid symbol table name syntax (no {constants.BANG} found)")
symbol_table_name = self.vol.type_name.split(constants.BANG)[0]
@@ -329,7 +329,7 @@ class OBJECT_HEADER(objects.StructType):
kvo = layer.config.get("kernel_virtual_offset", None)
if kvo is None:
raise AttributeError("Could not find kernel_virtual_offset for layer: {}".format(self.vol.layer_name))
raise AttributeError(f"Could not find kernel_virtual_offset for layer: {self.vol.layer_name}")
ntkrnlmp = self._context.module(symbol_table_name, layer_name = self.vol.layer_name, offset = kvo)
address = ntkrnlmp.get_symbol("ObpInfoMaskToOffset").address
volatility3/framework/symbols/windows/extensions/registry.py
+9 -9
View File
@@ -166,13 +166,13 @@ class CM_KEY_NODE(objects.StructType):
for subnode_offset in node.List[::listjump]:
if (subnode_offset & 0x7fffffff) > hive.maximum_address:
vollog.log(constants.LOGLEVEL_VVV,
"Node found with address outside the valid Hive size: {}".format(hex(subnode_offset)))
f"Node found with address outside the valid Hive size: {hex(subnode_offset)}")
else:
try:
subnode = hive.get_node(subnode_offset)
except (exceptions.InvalidAddressException, RegistryFormatException):
vollog.log(constants.LOGLEVEL_VVV,
"Failed to get node at {}, skipping".format(hex(subnode_offset)))
f"Failed to get node at {hex(subnode_offset)}, skipping")
continue
yield from self._get_subkeys_recursive(hive, subnode)
@@ -190,12 +190,12 @@ class CM_KEY_NODE(objects.StructType):
try:
node = hive.get_node(v)
except (RegistryInvalidIndex, RegistryFormatException) as excp:
vollog.debug("Invalid address {}".format(excp))
vollog.debug(f"Invalid address {excp}")
continue
if node.vol.type_name.endswith(constants.BANG + '_CM_KEY_VALUE'):
yield node
except (exceptions.InvalidAddressException, RegistryFormatException) as excp:
vollog.debug("Invalid address in get_values iteration: {}".format(excp))
vollog.debug(f"Invalid address in get_values iteration: {excp}")
return
def get_name(self) -> interfaces.objects.ObjectInterface:
@@ -240,7 +240,7 @@ class CM_KEY_VALUE(objects.StructType):
# Remove the high bit
datalen = datalen & 0x7fffffff
if (0 > datalen or datalen > 4):
raise ValueError("Unable to read inline registry value with excessive length: {}".format(datalen))
raise ValueError(f"Unable to read inline registry value with excessive length: {datalen}")
else:
data = layer.read(self.Data.vol.offset, datalen)
elif layer.hive.Version == 5 and datalen > 0x4000:
@@ -263,15 +263,15 @@ class CM_KEY_VALUE(objects.StructType):
self_type = RegValueTypes(self.Type)
if self_type == RegValueTypes.REG_DWORD:
if len(data) != struct.calcsize("<L"):
raise ValueError("Size of data does not match the type of registry value {}".format(self.get_name()))
raise ValueError(f"Size of data does not match the type of registry value {self.get_name()}")
return struct.unpack("<L", data)[0]
if self_type == RegValueTypes.REG_DWORD_BIG_ENDIAN:
if len(data) != struct.calcsize(">L"):
raise ValueError("Size of data does not match the type of registry value {}".format(self.get_name()))
raise ValueError(f"Size of data does not match the type of registry value {self.get_name()}")
return struct.unpack(">L", data)[0]
if self_type == RegValueTypes.REG_QWORD:
if len(data) != struct.calcsize("<Q"):
raise ValueError("Size of data does not match the type of registry value {}".format(self.get_name()))
raise ValueError(f"Size of data does not match the type of registry value {self.get_name()}")
return struct.unpack("<Q", data)[0]
if self_type in [
RegValueTypes.REG_SZ, RegValueTypes.REG_EXPAND_SZ, RegValueTypes.REG_LINK, RegValueTypes.REG_MULTI_SZ,
@@ -283,5 +283,5 @@ class CM_KEY_VALUE(objects.StructType):
return b''
# Fall back if it's something weird
vollog.debug("Unknown registry value type encountered: {}".format(self.Type))
vollog.debug(f"Unknown registry value type encountered: {self.Type}")
return data
volatility3/framework/symbols/windows/pdbconv.py
+18 -18
View File
@@ -364,19 +364,19 @@ class PdbReader:
def _read_info_stream(self, stream_number, stream_name, info_list):
vollog.debug("Reading {}".format(stream_name))
vollog.debug(f"Reading {stream_name}")
info_layer = self._context.layers.get(self._layer_name + "_stream" + str(stream_number), None)
if not info_layer:
raise ValueError("No {} stream available".format(stream_name))
raise ValueError(f"No {stream_name} stream available")
module = self._context.module(module_name = info_layer.pdb_symbol_table,
layer_name = info_layer.name,
offset = 0)
header = module.object(object_type = "TPI_HEADER", offset = 0)
# Check the header
if not (56 <= header.header_size < 1024):
raise ValueError("{} Stream Header size outside normal bounds".format(stream_name))
raise ValueError(f"{stream_name} Stream Header size outside normal bounds")
if header.index_min < 4096:
raise ValueError("Minimum {} index is 4096, found: {}".format(stream_name, header.index_min))
raise ValueError(f"Minimum {stream_name} index is 4096, found: {header.index_min}")
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))
@@ -396,8 +396,8 @@ class PdbReader:
output, consumed = self.consume_type(module, offset, length)
leaf_type, name, value = output
for tag_type in ['unnamed', 'anonymous']:
if name == '<{}-tag>'.format(tag_type) or name == '__{}'.format(tag_type):
name = '__{}_'.format(tag_type) + hex(len(info_list) + 0x1000)[2:]
if name == f'<{tag_type}-tag>' or name == f'__{tag_type}':
name = f'__{tag_type}_' + hex(len(info_list) + 0x1000)[2:]
if name:
info_references[name] = len(info_list)
info_list.append((leaf_type, name, value))
@@ -493,7 +493,7 @@ class PdbReader:
name = self.parse_string(sym.name, False, sym.length - sym.vol.size + 2)
address = self._sections[sym.segment - 1].VirtualAddress + sym.offset
else:
vollog.debug("Only v2 and v3 symbols are supported: {:x}".format(leaf_type))
vollog.debug(f"Only v2 and v3 symbols are supported: {leaf_type:x}")
if name:
if self._omap_mapping:
address = self.omap_lookup(address)
@@ -674,9 +674,9 @@ class PdbReader:
elif leaf_type in [leaf_type.LF_PROCEDURE]:
raise ValueError("LF_PROCEDURE size could not be identified")
else:
raise ValueError("Unable to determine size of leaf_type {}".format(leaf_type.lookup()))
raise ValueError(f"Unable to determine size of leaf_type {leaf_type.lookup()}")
if result <= 0:
raise ValueError("Invalid size identified: {} ({})".format(index, name))
raise ValueError(f"Invalid size identified: {index} ({name})")
return result
### TYPE HANDLING CODE
@@ -824,7 +824,7 @@ class PdbReader:
consumed += buildinfo.arguments.vol.size
result = leaf_type, None, buildinfo
else:
raise TypeError("Unhandled leaf_type: {}".format(leaf_type))
raise TypeError(f"Unhandled leaf_type: {leaf_type}")
return result, consumed
@@ -934,20 +934,20 @@ class PdbRetreiver:
vollog.info("Download PDB file...")
file_name = ".".join(file_name.split(".")[:-1] + ['pdb'])
for sym_url in ['http://msdl.microsoft.com/download/symbols']:
url = sym_url + "/{}/{}/".format(file_name, guid)
url = sym_url + f"/{file_name}/{guid}/"
result = None
for suffix in [file_name, file_name[:-1] + '_']:
try:
vollog.debug("Attempting to retrieve {}".format(url + suffix))
vollog.debug(f"Attempting to retrieve {url + suffix}")
# We have to cache this because the file is opened by a layer and we can't control whether that caches
result = resources.ResourceAccessor(progress_callback).open(url + suffix)
except (error.HTTPError, error.URLError) as excp:
vollog.debug("Failed with {}".format(excp))
vollog.debug(f"Failed with {excp}")
if result:
break
if progress_callback is not None:
progress_callback(100, "Downloading {}".format(url + suffix))
progress_callback(100, f"Downloading {url + suffix}")
if result is None:
return None
return url + suffix
@@ -972,7 +972,7 @@ if __name__ == '__main__':
Args:
progress: Percentage of progress of the current procedure
"""
message = "\rProgress: {0: 7.2f}\t\t{1:}".format(round(progress, 2), description or '')
message = f"\rProgress: {round(progress, 2): 7.2f}\t\t{description or ''}"
message_len = len(message)
self._max_message_len = max([self._max_message_len, message_len])
print(message, end = (' ' * (self._max_message_len - message_len)) + '\r')
@@ -1017,7 +1017,7 @@ if __name__ == '__main__':
url = parse.urlparse(filename, scheme = 'file')
if url.scheme == 'file':
if not os.path.exists(filename):
parser.error("File {} does not exists".format(filename))
parser.error(f"File {filename} does not exists")
location = "file:" + request.pathname2url(os.path.abspath(filename))
else:
location = filename
@@ -1032,7 +1032,7 @@ if __name__ == '__main__':
else:
guid = converted_json['metadata']['windows']['pdb']['GUID']
age = converted_json['metadata']['windows']['pdb']['age']
args.output = "{}-{}.json.xz".format(guid, age)
args.output = f"{guid}-{age}.json.xz"
output_url = os.path.abspath(args.output)
@@ -1049,6 +1049,6 @@ if __name__ == '__main__':
f.write(bytes(json_string, 'latin-1'))
if args.keep:
print("Temporary PDB file: {}".format(filename))
print(f"Temporary PDB file: {filename}")
elif delfile:
os.remove(filename)
volatility3/framework/symbols/windows/pdbutil.py
+5 -5
View File
@@ -85,13 +85,13 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
break
if not isf_path:
vollog.debug("Required symbol library path not found: {}".format(filter_string))
vollog.debug(f"Required symbol library path not found: {filter_string}")
vollog.info("The symbols can be downloaded later using pdbconv.py -p {} -g {}".format(
pdb_name.strip('\x00'),
guid.upper() + str(age)))
return None
vollog.debug("Using symbol library: {}".format(filter_string))
vollog.debug(f"Using symbol library: {filter_string}")
# Set the discovered options
join = interfaces.configuration.path_join
@@ -225,7 +225,7 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
try:
os.remove(filename)
except PermissionError:
vollog.warning("Temporary file could not be removed: {}".format(filename))
vollog.warning(f"Temporary file could not be removed: {filename}")
else:
vollog.warning("Cannot write downloaded symbols, please add the appropriate symbols"
" or add/modify a symbols directory that is writable")
@@ -310,11 +310,11 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
if not guids:
raise exceptions.VolatilityException(
"Did not find GUID of {} in module @ 0x{:x}!".format(pdb_name, module_offset))
f"Did not find GUID of {pdb_name} in module @ 0x{module_offset:x}!")
guid = guids[0]
vollog.debug("Found {}: {}-{}".format(guid["pdb_name"], guid["GUID"], guid["age"]))
vollog.debug(f"Found {guid['pdb_name']}: {guid['GUID']}-{guid['age']}")
return cls.load_windows_symbol_table(context,
guid["GUID"],
volatility3/schemas/__init__.py
+1 -1
View File
@@ -44,7 +44,7 @@ def validate(input: Dict[str, Any], use_cache: bool = True) -> bool:
basepath = os.path.abspath(os.path.dirname(__file__))
schema_path = os.path.join(basepath, 'schema-' + format + '.json')
if not os.path.exists(schema_path):
vollog.debug("Schema for format not found: {}".format(schema_path))
vollog.debug(f"Schema for format not found: {schema_path}")
return False
with open(schema_path, 'r') as s:
schema = json.load(s)