diff --git a/volatility/cli/text_renderer.py b/volatility/cli/text_renderer.py index 2dc7d3f8a..625588dbe 100644 --- a/volatility/cli/text_renderer.py +++ b/volatility/cli/text_renderer.py @@ -8,8 +8,9 @@ import random import string import sys from functools import wraps -from typing import Any, List, Tuple +from typing import Any, List, Tuple, Dict +from volatility.framework.interfaces.renderers import Column from volatility.framework.renderers import format_hints vollog = logging.getLogger(__name__) @@ -235,7 +236,7 @@ class PrettyTextRenderer(CLIRenderer): tree_indent_column = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(20)) max_column_widths = dict([(column.name, len(column.name)) for column in grid.columns]) - def visitor(node, accumulator): + def visitor(node, accumulator: List[Tuple[int, Dict[Column, bytes]]]) -> List[Tuple[int, Dict[Column, bytes]]]: # Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case max_column_widths[tree_indent_column] = max(max_column_widths.get(tree_indent_column, 0), node.path_depth) line = {} @@ -249,7 +250,7 @@ class PrettyTextRenderer(CLIRenderer): accumulator.append((node.path_depth, line)) return accumulator - final_output = [] # type: List[Tuple[int, List]] + final_output = [] # type: List[Tuple[int, Dict[Column, bytes]]] grid.populate(visitor, final_output) # Always align the tree to the left diff --git a/volatility/framework/plugins/mac/psaux.py b/volatility/framework/plugins/mac/psaux.py index 08485d588..6f7b02f4e 100644 --- a/volatility/framework/plugins/mac/psaux.py +++ b/volatility/framework/plugins/mac/psaux.py @@ -45,7 +45,7 @@ class Psaux(plugins.PluginInterface): task_name = utility.array_to_string(task.p_comm) - args = [] + args = [] # type: List[bytes] while argc > 0: try: diff --git a/volatility/framework/plugins/mac/pslist.py b/volatility/framework/plugins/mac/pslist.py index 424bf2cb6..3a866b8d8 100644 --- a/volatility/framework/plugins/mac/pslist.py +++ b/volatility/framework/plugins/mac/pslist.py @@ -3,7 +3,7 @@ # import logging -from typing import Callable, Iterable, List +from typing import Callable, Iterable, List, Dict from volatility.framework import renderers, interfaces, contexts from volatility.framework.automagic import mac @@ -77,7 +77,7 @@ class PsList(interfaces.plugins.PluginInterface): proc = kernel.object_from_symbol(symbol_name = "allproc").lh_first - seen = {} + seen = {} # type: Dict[int, int] while proc is not None and proc.vol.offset != 0: if proc.vol.offset in seen: vollog.log(logging.INFO, "Recursive process list detected (a result of non-atomic acquisition).") diff --git a/volatility/framework/plugins/mac/tasks.py b/volatility/framework/plugins/mac/tasks.py index 076e19f5e..bfe965c5c 100644 --- a/volatility/framework/plugins/mac/tasks.py +++ b/volatility/framework/plugins/mac/tasks.py @@ -3,7 +3,7 @@ # import logging -from typing import Callable, Iterable +from typing import Callable, Dict, Iterable from volatility.framework import interfaces, contexts from volatility.framework.automagic import mac @@ -40,7 +40,7 @@ class Tasks(pslist.PsList): queue_entry = kernel.object_from_symbol(symbol_name = "tasks") - seen = {} + seen = {} # type: Dict[int, int] for task in queue_entry.walk_list(queue_entry, "tasks", "task"): if task.vol.offset in seen: vollog.log(logging.INFO, "Recursive process list detected (a result of non-atomic acquisition).") diff --git a/volatility/framework/plugins/windows/callbacks.py b/volatility/framework/plugins/windows/callbacks.py index 9a1aef25e..41ec0b450 100644 --- a/volatility/framework/plugins/windows/callbacks.py +++ b/volatility/framework/plugins/windows/callbacks.py @@ -3,7 +3,7 @@ # import logging -from typing import List, Iterable, Tuple +from typing import List, Iterable, Tuple, Optional, Union import volatility.framework.interfaces.plugins as interfaces_plugins from volatility.framework import constants, exceptions, renderers, interfaces, symbols @@ -62,7 +62,7 @@ class Callbacks(interfaces_plugins.PluginInterface): @classmethod def list_notify_routines(cls, context: interfaces.context.ContextInterface, layer_name: str, symbol_table: str, - callback_table_name: str) -> Iterable[Tuple[str, int, str]]: + callback_table_name: str) -> Iterable[Tuple[str, int, Optional[str]]]: """Lists all kernel notification routines. Args: @@ -114,7 +114,7 @@ class Callbacks(interfaces_plugins.PluginInterface): @classmethod def list_registry_callbacks(cls, context: interfaces.context.ContextInterface, layer_name: str, symbol_table: str, - callback_table_name: str) -> Iterable[Tuple[str, int, str]]: + callback_table_name: str) -> Iterable[Tuple[str, int, None]]: """Lists all registry callbacks. Args: @@ -193,7 +193,8 @@ class Callbacks(interfaces_plugins.PluginInterface): try: component = ntkrnlmp.object( - "string", absolute = True, offset = callback.Component, max_length = 64, errors = "replace") + "string", absolute = True, offset = callback.Component, max_length = 64, errors = "replace" + ) # type: Union[interfaces.renderers.BaseAbsentValue, interfaces.objects.ObjectInterface] except exceptions.InvalidAddressException: component = renderers.UnreadableValue() diff --git a/volatility/framework/plugins/windows/poolscanner.py b/volatility/framework/plugins/windows/poolscanner.py index af0db3670..b70dc0865 100644 --- a/volatility/framework/plugins/windows/poolscanner.py +++ b/volatility/framework/plugins/windows/poolscanner.py @@ -359,8 +359,8 @@ class PoolScanner(plugins.PluginInterface): cookie = handles.Handles.find_cookie(context = context, layer_name = layer_name, symbol_table = symbol_table) - is_windows_10 = cls.is_windows_10(context = context, symbol_table = symbol_table) - is_windows_8_or_later = cls.is_windows_8_or_later(context = context, symbol_table = symbol_table) + is_windows_10 = cls.is_windows_10(context, symbol_table) + is_windows_8_or_later = cls.is_windows_8_or_later(context, symbol_table) # start off with the primary virtual layer scan_layer = layer_name @@ -434,7 +434,7 @@ class PoolScanner(plugins.PluginInterface): # We have to manually load a symbol table if symbols.symbol_table_is_64bit(context, symbol_table): - is_win_7 = cls.is_windows_7(context = context, symbol_table = symbol_table) + is_win_7 = cls.is_windows_7(context, symbol_table) if is_win_7: pool_header_json_filename = "poolheader-x64-win7" else: diff --git a/volatility/framework/plugins/windows/registry/printkey.py b/volatility/framework/plugins/windows/registry/printkey.py index 7150825ac..593ac380b 100644 --- a/volatility/framework/plugins/windows/registry/printkey.py +++ b/volatility/framework/plugins/windows/registry/printkey.py @@ -4,7 +4,7 @@ import datetime import logging -from typing import List, Sequence, Iterable, Tuple +from typing import List, Sequence, Iterable, Tuple, Union from volatility.framework import objects, renderers, exceptions, interfaces, constants from volatility.framework.configuration import requirements @@ -36,8 +36,8 @@ class PrintKey(interfaces.plugins.PluginInterface): ] @classmethod - def key_iterator(cls, hive: RegistryHive, node_path: Sequence[objects.StructType] = None, - recurse: bool = False) -> Iterable[Tuple[int, bool, datetime.datetime, str, bool, bytes]]: + def key_iterator(cls, hive: RegistryHive, node_path: Sequence[objects.StructType] = None, recurse: bool = False + ) -> Iterable[Tuple[int, bool, datetime.datetime, str, bool, interfaces.objects.ObjectInterface]]: """Walks through a set of nodes from a given node (last one in node_path). Avoids loops by not traversing into nodes already present in the node_path. @@ -112,7 +112,7 @@ class PrintKey(interfaces.plugins.PluginInterface): value_node_name = renderers.UnreadableValue() try: - value_data = str(node.decode_data()) + value_data = str(node.decode_data()) # type: Union[interfaces.renderers.BaseAbsentValue, str] except (ValueError, exceptions.InvalidAddressException, RegistryFormatException) as excp: vollog.debug(excp) value_data = renderers.UnreadableValue() diff --git a/volatility/framework/symbols/intermed.py b/volatility/framework/symbols/intermed.py index a9e5e9e78..ecc17f79d 100644 --- a/volatility/framework/symbols/intermed.py +++ b/volatility/framework/symbols/intermed.py @@ -11,7 +11,7 @@ import os import pathlib import zipfile from abc import ABCMeta -from typing import Any, Dict, Generator, Iterable, List, Optional, Type, Tuple +from typing import Any, Dict, Generator, Iterable, List, Optional, Type, Tuple, Mapping import volatility import volatility.framework.layers.resources @@ -208,7 +208,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface): filename: str, native_types: Optional[interfaces.symbols.NativeTableInterface] = None, table_mapping: Optional[Dict[str, str]] = None, - class_types: Optional[Dict[str, Type[interfaces.objects.ObjectInterface]]] = None) -> str: + class_types: Optional[Mapping[str, Type[interfaces.objects.ObjectInterface]]] = None) -> str: """Takes a context and loads an intermediate symbol table based on a filename. diff --git a/volatility/framework/symbols/windows/extensions/registry.py b/volatility/framework/symbols/windows/extensions/registry.py index 448aa199c..b0ec164f9 100644 --- a/volatility/framework/symbols/windows/extensions/registry.py +++ b/volatility/framework/symbols/windows/extensions/registry.py @@ -70,6 +70,7 @@ class _HMAP_ENTRY(objects.StructType): except AttributeError: return self.BlockAddress + class _CMHIVE(objects.StructType): def get_name(self) -> Optional[interfaces.objects.ObjectInterface]: @@ -205,7 +206,7 @@ class _CM_KEY_NODE(objects.StructType): """Since this is just a casting convenience, it can be a property.""" return self.Name.cast("string", max_length = self.NameLength, encoding = "latin-1") - def get_key_path(self) -> interfaces.objects.ObjectInterface: + def get_key_path(self) -> str: reg = self._context.layers[self.vol.layer_name] if not isinstance(reg, RegistryHive): raise TypeError("Key was not instantiated on a RegistryHive layer") @@ -277,10 +278,10 @@ class _CM_KEY_VALUE(objects.StructType): return output if self_type == RegValueTypes.REG_MULTI_SZ: return str(data, encoding = "utf-16-le").split("\x00")[0] - if self_type in [RegValueTypes.REG_BINARY, - RegValueTypes.REG_FULL_RESOURCE_DESCRIPTOR, - RegValueTypes.REG_RESOURCE_LIST, - RegValueTypes.REG_RESOURCE_REQUIREMENTS_LIST]: + if self_type in [ + RegValueTypes.REG_BINARY, RegValueTypes.REG_FULL_RESOURCE_DESCRIPTOR, RegValueTypes.REG_RESOURCE_LIST, + RegValueTypes.REG_RESOURCE_REQUIREMENTS_LIST + ]: return data if self_type == RegValueTypes.REG_NONE: return ''