From 0d48261201e09735e791f2a301bb39d3bd5e90d9 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Sun, 18 Jul 2021 16:49:37 +0100 Subject: [PATCH] Core: Change all 3.5 type hints to variable annotations --- volatility3/__init__.py | 2 +- volatility3/cli/__init__.py | 2 +- volatility3/cli/text_renderer.py | 8 +++--- volatility3/cli/volargparse.py | 6 ++--- volatility3/cli/volshell/generic.py | 4 +-- .../framework/automagic/construct_layers.py | 2 +- volatility3/framework/automagic/linux.py | 2 +- volatility3/framework/automagic/pdbscan.py | 10 +++---- .../framework/automagic/symbol_cache.py | 8 +++--- .../framework/automagic/symbol_finder.py | 12 ++++----- volatility3/framework/automagic/windows.py | 2 +- .../framework/configuration/requirements.py | 14 +++++----- volatility3/framework/contexts/__init__.py | 4 +-- volatility3/framework/interfaces/automagic.py | 4 +-- .../framework/interfaces/configuration.py | 14 +++++----- volatility3/framework/interfaces/layers.py | 18 ++++++------- volatility3/framework/interfaces/objects.py | 4 +-- volatility3/framework/interfaces/plugins.py | 4 +-- volatility3/framework/interfaces/renderers.py | 4 +-- volatility3/framework/interfaces/symbols.py | 2 +- volatility3/framework/layers/intel.py | 2 +- volatility3/framework/layers/linear.py | 3 ++- volatility3/framework/layers/msf.py | 2 +- volatility3/framework/layers/physical.py | 6 ++--- volatility3/framework/layers/qemu.py | 2 +- volatility3/framework/layers/registry.py | 2 +- .../framework/layers/scanners/__init__.py | 2 +- .../framework/layers/scanners/multiregexp.py | 2 +- volatility3/framework/layers/segmented.py | 6 ++--- volatility3/framework/objects/__init__.py | 18 ++++++------- volatility3/framework/objects/templates.py | 10 +++---- volatility3/framework/plugins/mac/lsmod.py | 2 +- volatility3/framework/plugins/mac/psaux.py | 2 +- volatility3/framework/plugins/mac/pslist.py | 4 +-- volatility3/framework/plugins/timeliner.py | 2 +- .../framework/plugins/windows/callbacks.py | 4 +-- .../framework/plugins/windows/handles.py | 2 +- .../framework/plugins/windows/modscan.py | 2 +- .../framework/plugins/windows/modules.py | 2 +- .../framework/plugins/windows/poolscanner.py | 2 +- .../framework/plugins/windows/pstree.py | 6 ++--- .../plugins/windows/registry/printkey.py | 2 +- .../plugins/windows/registry/userassist.py | 4 +-- .../framework/plugins/windows/strings.py | 12 +++++---- .../framework/plugins/windows/virtmap.py | 4 +-- volatility3/framework/renderers/__init__.py | 6 ++--- volatility3/framework/renderers/conversion.py | 2 +- .../framework/renderers/format_hints.py | 2 +- volatility3/framework/symbols/__init__.py | 8 +++--- volatility3/framework/symbols/intermed.py | 4 +-- .../framework/symbols/linux/__init__.py | 2 +- volatility3/framework/symbols/mac/__init__.py | 2 +- .../symbols/mac/extensions/__init__.py | 2 +- volatility3/framework/symbols/native.py | 6 ++--- .../symbols/windows/extensions/__init__.py | 2 +- .../symbols/windows/extensions/pool.py | 2 +- .../framework/symbols/windows/pdbconv.py | 26 +++++++++---------- volatility3/schemas/__init__.py | 2 +- 58 files changed, 150 insertions(+), 147 deletions(-) diff --git a/volatility3/__init__.py b/volatility3/__init__.py index 5d3c34e43..db52aa9b0 100644 --- a/volatility3/__init__.py +++ b/volatility3/__init__.py @@ -43,7 +43,7 @@ class WarningFindSpec(abc.MetaPathFinder): raise Warning(warning) -warning_find_spec = [WarningFindSpec()] # type: List[abc.MetaPathFinder] +warning_find_spec: List[abc.MetaPathFinder] = [WarningFindSpec()] sys.meta_path = warning_find_spec + sys.meta_path # We point the volatility3.plugins __path__ variable at BOTH diff --git a/volatility3/cli/__init__.py b/volatility3/cli/__init__.py index 9ee4d570c..e6f00728a 100644 --- a/volatility3/cli/__init__.py +++ b/volatility3/cli/__init__.py @@ -583,7 +583,7 @@ class CommandLine: # Construct an argparse group for requirement in configurable.get_requirements(): - additional = {} # type: Dict[str, Any] + additional: Dict[str, Any] = {} if not isinstance(requirement, interfaces.configuration.RequirementInterface): raise TypeError("Plugin contains requirements that are not RequirementInterfaces: {}".format( configurable.__name__)) diff --git a/volatility3/cli/text_renderer.py b/volatility3/cli/text_renderer.py index 19507b11c..35b2468e8 100644 --- a/volatility3/cli/text_renderer.py +++ b/volatility3/cli/text_renderer.py @@ -278,7 +278,7 @@ class PrettyTextRenderer(CLIRenderer): accumulator.append((node.path_depth, line)) return accumulator - final_output = [] # type: List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]] + final_output: List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]] = [] if not grid.populated: grid.populate(visitor, final_output) else: @@ -323,15 +323,15 @@ class JsonRenderer(CLIRenderer): outfd = sys.stdout outfd.write("\n") - final_output = ( - {}, []) # type: Tuple[Dict[str, List[interfaces.renderers.TreeNode]], List[interfaces.renderers.TreeNode]] + final_output: Tuple[Dict[str, List[interfaces.renderers.TreeNode]], List[interfaces.renderers.TreeNode]] = ( + {}, []) def visitor( node: interfaces.renderers.TreeNode, accumulator: Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]] ) -> Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]]: # Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case acc_map, final_tree = accumulator - node_dict = {'__children': []} # type: Dict[str, Any] + node_dict: Dict[str, Any] = {'__children': []} for column_index in range(len(grid.columns)): column = grid.columns[column_index] renderer = self._type_renderers.get(column.type, self._type_renderers['default']) diff --git a/volatility3/cli/volargparse.py b/volatility3/cli/volargparse.py index 5ced541ae..996acc150 100644 --- a/volatility3/cli/volargparse.py +++ b/volatility3/cli/volargparse.py @@ -12,7 +12,7 @@ from typing import List, Optional, Sequence, Any, Union # We shouldn't really steal a private member from argparse, but otherwise we're just duplicating code # HelpfulSubparserAction gives more information about the possible choices from a subparsed choice -# HelpfulArgParser gives the list of choices when no arguments are provided to a choice option whilst still using a METAVAR +# HelpfulArgParser gives the list of choices when no arguments are provided to a choice option whilst still using a class HelpfulSubparserAction(argparse._SubParsersAction): @@ -22,7 +22,7 @@ class HelpfulSubparserAction(argparse._SubParsersAction): def __init__(self, *args, **kwargs) -> None: super().__init__(*args, **kwargs) # We don't want the action self-check to kick in, so we remove the choices list, the check happens in __call__ - self.choices = None # type: ignore + self.choices = None def __call__(self, parser: argparse.ArgumentParser, @@ -31,7 +31,7 @@ class HelpfulSubparserAction(argparse._SubParsersAction): option_string: Optional[str] = None) -> None: parser_name = '' - arg_strings = [] # type: List[str] + arg_strings: List[str] = [] if values is not None: for value in values: if not parser_name: diff --git a/volatility3/cli/volshell/generic.py b/volatility3/cli/volshell/generic.py index 46701b4ac..7252c3b69 100644 --- a/volatility3/cli/volshell/generic.py +++ b/volatility3/cli/volshell/generic.py @@ -30,7 +30,7 @@ class Volshell(interfaces.plugins.PluginInterface): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) - self.__current_layer = None # type: Optional[str] + self.__current_layer: Optional[str] = None self.__console = None def random_string(self, length: int = 32) -> str: @@ -38,7 +38,7 @@ class Volshell(interfaces.plugins.PluginInterface): @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: - reqs = [] # type: List[interfaces.configuration.RequirementInterface] + reqs: List[interfaces.configuration.RequirementInterface] = [] if cls == Volshell: reqs = [ requirements.URIRequirement(name = 'script', diff --git a/volatility3/framework/automagic/construct_layers.py b/volatility3/framework/automagic/construct_layers.py index 8afc6326e..40a17419f 100644 --- a/volatility3/framework/automagic/construct_layers.py +++ b/volatility3/framework/automagic/construct_layers.py @@ -37,7 +37,7 @@ class ConstructionMagic(interfaces.automagic.AutomagicInterface): # Make sure we import the layers, so they can reconstructed framework.import_files(sys.modules['volatility3.framework.layers']) - result = [] # type: List[str] + result: List[str] = [] if requirement.unsatisfied(context, config_path): # Having called validate at the top level tells us both that we need to dig deeper # but also ensures that TranslationLayerRequirements have got the correct subrequirements if their class is populated diff --git a/volatility3/framework/automagic/linux.py b/volatility3/framework/automagic/linux.py index 4d0b95ce3..767dd3407 100644 --- a/volatility3/framework/automagic/linux.py +++ b/volatility3/framework/automagic/linux.py @@ -57,7 +57,7 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface): layer_name, progress_callback = progress_callback) - layer_class = intel.Intel # type: Type + layer_class: Type = intel.Intel if 'init_top_pgt' in table.symbols: layer_class = intel.Intel32e dtb_symbol_name = 'init_top_pgt' diff --git a/volatility3/framework/automagic/pdbscan.py b/volatility3/framework/automagic/pdbscan.py index d1df22ebe..b32010506 100644 --- a/volatility3/framework/automagic/pdbscan.py +++ b/volatility3/framework/automagic/pdbscan.py @@ -62,7 +62,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): A list of (layer_name, scan_results) """ sub_config_path = interfaces.configuration.path_join(config_path, requirement.name) - results = [] # type: List[str] + results: List[str] = [] if isinstance(requirement, requirements.TranslationLayerRequirement): # Check for symbols in this layer # FIXME: optionally allow a full (slow) scan @@ -207,13 +207,13 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): """Method for finding a suitable kernel offset based on a module table.""" vollog.debug("Kernel base determination - searching layer module list structure") - valid_kernel = None # type: Optional[ValidKernelType] + valid_kernel: Optional[ValidKernelType] = None # If we're here, chances are high we're in a Win10 x64 image with kernel base randomization physical_layer_name = self.get_physical_layer_name(context, vlayer) physical_layer = context.layers[physical_layer_name] # TODO: On older windows, this might be \WINDOWS\system32\nt rather than \SystemRoot\system32\nt results = physical_layer.scan(context, scanners.BytesScanner(pattern), progress_callback = progress_callback) - seen = set() # type: Set[int] + seen: Set[int] = set() # Because this will launch a scan of the virtual layer, we want to be careful for result in results: # TODO: Identify the specific structure we're finding and document this a bit better @@ -252,7 +252,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): """Scans a virtual address.""" # Scan a few megs of the virtual space at the location to see if they're potential kernels - valid_kernel = None # type: Optional[ValidKernelType] + valid_kernel: Optional[ValidKernelType] = None kernel_pdb_names = [bytes(name + ".pdb", "utf-8") for name in constants.windows.KERNEL_MODULE_NAMES] virtual_layer_name = vlayer.name @@ -295,7 +295,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): Returns: A dictionary of valid kernels """ - valid_kernel = None # type: Optional[ValidKernelType] + valid_kernel: Optional[ValidKernelType] = None for virtual_layer_name in potential_layers: vlayer = context.layers.get(virtual_layer_name, None) if isinstance(vlayer, layers.intel.Intel): diff --git a/volatility3/framework/automagic/symbol_cache.py b/volatility3/framework/automagic/symbol_cache.py index 1b468a4cd..627ffac8b 100644 --- a/volatility3/framework/automagic/symbol_cache.py +++ b/volatility3/framework/automagic/symbol_cache.py @@ -26,15 +26,15 @@ class SymbolBannerCache(interfaces.automagic.AutomagicInterface): # The user would run it eventually either way, but running it first means it can be used that run priority = 0 - os = None # type: Optional[str] - symbol_name = "banner_name" # type: str - banner_path = None # type: Optional[str] + os: Optional[str] = None + symbol_name: str = "banner_name" + banner_path: Optional[str] = None @classmethod def load_banners(cls) -> BannersType: if not cls.banner_path: raise ValueError("Banner_path not appropriately set") - banners = {} # type: BannersType + banners: BannersType = {} if os.path.exists(cls.banner_path): with open(cls.banner_path, "rb") as f: # We use pickle over JSON because we're dealing with bytes objects diff --git a/volatility3/framework/automagic/symbol_finder.py b/volatility3/framework/automagic/symbol_finder.py index b57ab54c8..03d051c49 100644 --- a/volatility3/framework/automagic/symbol_finder.py +++ b/volatility3/framework/automagic/symbol_finder.py @@ -17,15 +17,15 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface): """Symbol loader based on signature strings.""" priority = 40 - banner_config_key = "banner" # type: str - banner_cache = None # type: Optional[Type[symbol_cache.SymbolBannerCache]] - symbol_class = None # type: Optional[str] - find_aslr = None # type: Optional[Callable] + banner_config_key: str = "banner" + banner_cache: Optional[Type[symbol_cache.SymbolBannerCache]] = None + symbol_class: Optional[str] = None + find_aslr: Optional[Callable] = None def __init__(self, context: interfaces.context.ContextInterface, config_path: str) -> None: super().__init__(context, config_path) - self._requirements = [] # type: List[Tuple[str, interfaces.configuration.RequirementInterface]] - self._banners = {} # type: symbol_cache.BannersType + self._requirements: List[Tuple[str, interfaces.configuration.RequirementInterface]] = [] + self._banners: symbol_cache.BannersType = {} @property def banners(self) -> symbol_cache.BannersType: diff --git a/volatility3/framework/automagic/windows.py b/volatility3/framework/automagic/windows.py index 1144978be..9f98bebfe 100644 --- a/volatility3/framework/automagic/windows.py +++ b/volatility3/framework/automagic/windows.py @@ -325,7 +325,7 @@ class WindowsIntelStacker(interfaces.automagic.StackerLayerInterface): if arch not in ['Intel32', 'Intel64']: return None # Set the layer type - layer_type = intel.WindowsIntel # type: Type + layer_type: Type = intel.WindowsIntel if arch == 'Intel64': layer_type = intel.WindowsIntel32e elif base_layer.metadata.get('pae', False): diff --git a/volatility3/framework/configuration/requirements.py b/volatility3/framework/configuration/requirements.py index ea0a5fcc2..e0f8bb2d8 100644 --- a/volatility3/framework/configuration/requirements.py +++ b/volatility3/framework/configuration/requirements.py @@ -36,13 +36,13 @@ class BooleanRequirement(interfaces.configuration.SimpleTypeRequirement): class IntRequirement(interfaces.configuration.SimpleTypeRequirement): """A requirement type that contains a single integer.""" - instance_type = int # type: ClassVar[Type] + instance_type: ClassVar[Type] = int class StringRequirement(interfaces.configuration.SimpleTypeRequirement): """A requirement type that contains a single unicode string.""" # TODO: Maybe add string length limits? - instance_type = str # type: ClassVar[Type] + instance_type: ClassVar[Type] = str class URIRequirement(StringRequirement): @@ -53,7 +53,7 @@ class URIRequirement(StringRequirement): class BytesRequirement(interfaces.configuration.SimpleTypeRequirement): """A requirement type that contains a byte string.""" - instance_type = bytes # type: ClassVar[Type] + instance_type: ClassVar[Type] = bytes class ListRequirement(interfaces.configuration.RequirementInterface): @@ -83,9 +83,9 @@ class ListRequirement(interfaces.configuration.RequirementInterface): super().__init__(*args, **kwargs) if not issubclass(element_type, interfaces.configuration.BasicTypes): raise TypeError("ListRequirements can only be populated with simple InstanceRequirements") - self.element_type = element_type # type: Type - self.min_elements = min_elements or 0 # type: int - self.max_elements = max_elements # type: Optional[int] + self.element_type: Type = element_type + self.min_elements: int = min_elements or 0 + self.max_elements: Optional[int] = max_elements def unsatisfied(self, context: interfaces.context.ContextInterface, config_path: str) -> Dict[str, interfaces.configuration.RequirementInterface]: @@ -397,7 +397,7 @@ class VersionRequirement(interfaces.configuration.RequirementInterface): super().__init__(name = name, description = description, default = default, optional = optional) if component is None: raise TypeError("Component cannot be None") - self._component = component # type: Type[interfaces.configuration.VersionableInterface] + self._component: Type[interfaces.configuration.VersionableInterface] = component if version is None: raise TypeError("Version cannot be None") self._version = version diff --git a/volatility3/framework/contexts/__init__.py b/volatility3/framework/contexts/__init__.py index 9fb950a03..412dfbc50 100644 --- a/volatility3/framework/contexts/__init__.py +++ b/volatility3/framework/contexts/__init__.py @@ -319,7 +319,7 @@ class ModuleCollection: included in the deduplicated version """ new_modules = [] - seen = set() # type: Set[str] + seen: Set[str] = set() for mod in self._modules: if mod.hash not in seen or mod.size == 0: new_modules.append(mod) @@ -334,7 +334,7 @@ class ModuleCollection: @classmethod def _generate_module_dict(cls, modules: List[SizedModule]) -> Dict[str, List[SizedModule]]: - result = {} # type: Dict[str, List[SizedModule]] + result: Dict[str, List[SizedModule]] = {} for module in modules: modlist = result.get(module.name, []) modlist.append(module) diff --git a/volatility3/framework/interfaces/automagic.py b/volatility3/framework/interfaces/automagic.py index c6eb2e5ce..04f495894 100644 --- a/volatility3/framework/interfaces/automagic.py +++ b/volatility3/framework/interfaces/automagic.py @@ -82,7 +82,7 @@ class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metacla A list of tuples containing the config_path, sub_config_path and requirement identifying the unsatisfied `Requirements` """ sub_config_path = interfaces.configuration.path_join(config_path, requirement_root.name) - results = [] # type: List[Tuple[str, interfaces.configuration.RequirementInterface]] + results: List[Tuple[str, interfaces.configuration.RequirementInterface]] = [] recurse = not shortcut if isinstance(requirement_root, requirement_type): if recurse or requirement_root.unsatisfied(context, config_path): @@ -105,7 +105,7 @@ class StackerLayerInterface(metaclass = ABCMeta): stack_order = 0 """The order in which to attempt stacking, the lower the earlier""" - exclusion_list = [] # type: List[str] + exclusion_list: List[str] = [] """The list operating systems/first-level plugin hierarchy that should exclude this stacker""" @classmethod diff --git a/volatility3/framework/interfaces/configuration.py b/volatility3/framework/interfaces/configuration.py index d3c05d9a9..d52d6a176 100644 --- a/volatility3/framework/interfaces/configuration.py +++ b/volatility3/framework/interfaces/configuration.py @@ -79,8 +79,8 @@ class HierarchicalDict(collections.abc.Mapping): if not (isinstance(separator, str) and len(separator) == 1): 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'] + self._data: Dict[str, ConfigSimpleType] = {} + self._subdict: Dict[str, 'HierarchicalDict'] = {} if isinstance(initial_dict, str): initial_dict = json.loads(initial_dict) if isinstance(initial_dict, dict): @@ -320,7 +320,7 @@ class RequirementInterface(metaclass = ABCMeta): self._description = description or "" self._default = default self._optional = optional - self._requirements = {} # type: Dict[str, RequirementInterface] + self._requirements: Dict[str, RequirementInterface] = {} def __repr__(self) -> str: return "<" + self.__class__.__name__ + ": " + self.name + ">" @@ -438,7 +438,7 @@ class RequirementInterface(metaclass = ABCMeta): class SimpleTypeRequirement(RequirementInterface): """Class to represent a single simple type (such as a boolean, a string, an integer or a series of bytes)""" - instance_type = bool # type: ClassVar[Type] + instance_type: ClassVar[Type] = bool def add_requirement(self, requirement: RequirementInterface): """Always raises a TypeError as instance requirements cannot have @@ -529,7 +529,7 @@ class ConstructableRequirementInterface(RequirementInterface): def __init__(self, *args, **kwargs) -> None: super().__init__(*args, **kwargs) self.add_requirement(ClassRequirement("class", "Class of the constructable requirement")) - self._current_class_requirements = set() # type: Set[Any] + self._current_class_requirements: Set[Any] = set() def __eq__(self, other): # We can just use super because it checks all member of `__dict__` @@ -620,7 +620,7 @@ class ConfigurableInterface(metaclass = ABCMeta): super().__init__() self._context = context self._config_path = config_path - self._config_cache = None # type: Optional[HierarchicalDict] + self._config_cache: Optional[HierarchicalDict] = None @property def context(self) -> 'interfaces.context.ContextInterface': @@ -729,7 +729,7 @@ class VersionableInterface: All version number should use semantic versioning """ - _version = (0, 0, 0) # type: Tuple[int, int, int] + _version: Tuple[int, int, int] = (0, 0, 0) @classproperty def version(cls) -> Tuple[int, int, int]: diff --git a/volatility3/framework/interfaces/layers.py b/volatility3/framework/interfaces/layers.py index 0b5a58d37..c7c4feb8c 100644 --- a/volatility3/framework/interfaces/layers.py +++ b/volatility3/framework/interfaces/layers.py @@ -58,8 +58,8 @@ class ScannerInterface(interfaces.configuration.VersionableInterface, metaclass super().__init__() self.chunk_size = 0x1000000 # Default to 16Mb chunks self.overlap = 0x1000 # A page of overlap by default - self._context = None # type: Optional[interfaces.context.ContextInterface] - self._layer_name = None # type: Optional[str] + self._context: Optional[interfaces.context.ContextInterface] = None + self._layer_name: Optional[str] = None @property def context(self) -> Optional['interfaces.context.ContextInterface']: @@ -99,7 +99,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla accesses a data source and exposes it within volatility. """ - _direct_metadata = {'architecture': 'Unknown', 'os': 'Unknown'} # type: Mapping + _direct_metadata: Mapping = {'architecture': 'Unknown', 'os': 'Unknown'} def __init__(self, context: 'interfaces.context.ContextInterface', @@ -227,7 +227,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla sections = list(self._coalesce_sections(sections)) try: - progress = DummyProgress() # type: ProgressValue + progress: ProgressValue = DummyProgress() scan_iterator = functools.partial(self._scan_iterator, scanner, sections) scan_metric = self._scan_metric(scanner, sections) if not scanner.thread_safe or constants.PARALLELISM == constants.Parallelism.Off: @@ -240,7 +240,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla yield from scan_chunk(value) else: progress = multiprocessing.Manager().Value("Q", 0) - parallel_module = multiprocessing # type: types.ModuleType + parallel_module: types.ModuleType = multiprocessing if constants.PARALLELISM == constants.Parallelism.Threading: progress = DummyProgress() parallel_module = threading @@ -266,7 +266,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla def _coalesce_sections(self, sections: Iterable[Tuple[int, int]]) -> Iterable[Tuple[int, int]]: """Take a list of (start, length) sections and coalesce any adjacent sections.""" - result = [] # type: List[Tuple[int, int]] + result: List[Tuple[int, int]] = [] position = 0 for (start, length) in sorted(sections): if result and start <= position: @@ -423,7 +423,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta): """Reads an offset for length bytes and returns 'bytes' (not 'str') of length size.""" current_offset = offset - output = b'' # type: bytes + output: bytes = b'' for (layer_offset, sublength, mapped_offset, mapped_length, layer) in self.mapping(offset, length, ignore_errors = pad): @@ -473,7 +473,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta): assumed to have no holes """ for (section_start, section_length) in sections: - output = [] # type: List[Tuple[str, int, int]] + output: List[Tuple[str, int, int]] = [] # Hold the offsets of each chunk (including how much has been filled) chunk_start = chunk_position = 0 @@ -532,7 +532,7 @@ class LayerContainer(collections.abc.Mapping): """Container for multiple layers of data.""" def __init__(self) -> None: - self._layers = {} # type: Dict[str, DataLayerInterface] + self._layers: Dict[str, DataLayerInterface] = {} def read(self, layer: str, offset: int, length: int, pad: bool = False) -> bytes: """Reads from a particular layer at offset for length bytes. diff --git a/volatility3/framework/interfaces/objects.py b/volatility3/framework/interfaces/objects.py index 4cb8bce42..11040c503 100644 --- a/volatility3/framework/interfaces/objects.py +++ b/volatility3/framework/interfaces/objects.py @@ -214,7 +214,7 @@ class ObjectInterface(metaclass = abc.ABCMeta): to control how their templates respond without needing to write new templates for each and every potental object type. """ - _methods = [] # type: List[str] + _methods: List[str] = [] @classmethod @abc.abstractmethod @@ -275,7 +275,7 @@ class Template: """Stores the keyword arguments for later object creation.""" # Allow the updating of template arguments whilst still in template form super().__init__() - empty_dict = {} # type: Dict[str, Any] + empty_dict: Dict[str, Any] = {} self._vol = collections.ChainMap(empty_dict, arguments, {'type_name': type_name}) @property diff --git a/volatility3/framework/interfaces/plugins.py b/volatility3/framework/interfaces/plugins.py index 06316c221..d091f27cb 100644 --- a/volatility3/framework/interfaces/plugins.py +++ b/volatility3/framework/interfaces/plugins.py @@ -95,7 +95,7 @@ class PluginInterface(interfaces.configuration.ConfigurableInterface, """ # Be careful with inheritance around this (We default to requiring a version which doesn't exist, so it must be set) - _required_framework_version = (0, 0, 0) # type: Tuple[int, int, int] + _required_framework_version: Tuple[int, int, int] = (0, 0, 0) """The _version variable is a quick way for plugins to define their current interface, it should follow SemVer rules""" def __init__(self, @@ -121,7 +121,7 @@ class PluginInterface(interfaces.configuration.ConfigurableInterface, if requirement.name not in self.config: self.config[requirement.name] = requirement.default - self._file_handler = FileHandlerInterface # type: Type[FileHandlerInterface] + self._file_handler: Type[FileHandlerInterface] = FileHandlerInterface framework.require_interface_version(*self._required_framework_version) diff --git a/volatility3/framework/interfaces/renderers.py b/volatility3/framework/interfaces/renderers.py index 54b9f922f..dd53a96b3 100644 --- a/volatility3/framework/interfaces/renderers.py +++ b/volatility3/framework/interfaces/renderers.py @@ -38,7 +38,7 @@ class Renderer(metaclass = ABCMeta): class ColumnSortKey(metaclass = ABCMeta): - ascending = True # type: bool + ascending: bool = True @abstractmethod def __call__(self, values: List[Any]) -> Any: @@ -129,7 +129,7 @@ class TreeGrid(object, metaclass = ABCMeta): and to create cycles. """ - base_types = (int, str, float, bytes, datetime.datetime, Disassembly) # type: ClassVar[Tuple] + base_types: ClassVar[Tuple] = (int, str, float, bytes, datetime.datetime, Disassembly) def __init__(self, columns: ColumnsType, generator: Generator) -> None: """Constructs a TreeGrid object using a specific set of columns. diff --git a/volatility3/framework/interfaces/symbols.py b/volatility3/framework/interfaces/symbols.py index 8b4419829..37c2824eb 100644 --- a/volatility3/framework/interfaces/symbols.py +++ b/volatility3/framework/interfaces/symbols.py @@ -96,7 +96,7 @@ class BaseSymbolTableInterface: table_mapping = {} self.table_mapping = table_mapping self._native_types = native_types - self._sort_symbols = [] # type: List[Tuple[int, str]] + self._sort_symbols: List[Tuple[int, str]] = [] # Set any provisioned class_types if class_types: diff --git a/volatility3/framework/layers/intel.py b/volatility3/framework/layers/intel.py index 555a8417b..48fa205bd 100644 --- a/volatility3/framework/layers/intel.py +++ b/volatility3/framework/layers/intel.py @@ -39,7 +39,7 @@ class Intel(linear.LinearlyMappedLayer): metadata: Optional[Dict[str, Any]] = None) -> None: super().__init__(context = context, config_path = config_path, name = name, metadata = metadata) self._base_layer = self.config["memory_layer"] - self._swap_layers = [] # type: List[str] + self._swap_layers: List[str] = [] self._page_map_offset = self.config["page_map_offset"] # Assign constants diff --git a/volatility3/framework/layers/linear.py b/volatility3/framework/layers/linear.py index 40341d86d..d94f7bcc0 100644 --- a/volatility3/framework/layers/linear.py +++ b/volatility3/framework/layers/linear.py @@ -33,7 +33,8 @@ class LinearlyMappedLayer(interfaces.layers.TranslationLayerInterface): """Reads an offset for length bytes and returns 'bytes' (not 'str') of length size.""" current_offset = offset - output = [] # type: List[bytes] + output: List[bytes] = [] + output: List[bytes] = [] 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( diff --git a/volatility3/framework/layers/msf.py b/volatility3/framework/layers/msf.py index 7713c5024..02fc570bc 100644 --- a/volatility3/framework/layers/msf.py +++ b/volatility3/framework/layers/msf.py @@ -34,7 +34,7 @@ class PdbMultiStreamFormat(linear.LinearlyMappedLayer): if response is None: raise PDBFormatException(name, "Could not find a suitable header") self._version, self._header = response - self._streams = {} # type: Dict[int, str] + self._streams: Dict[int, str] = {} @property def pdb_symbol_table(self) -> str: diff --git a/volatility3/framework/layers/physical.py b/volatility3/framework/layers/physical.py index 998d8cf12..228815fea 100644 --- a/volatility3/framework/layers/physical.py +++ b/volatility3/framework/layers/physical.py @@ -86,10 +86,10 @@ class FileLayer(interfaces.layers.DataLayerInterface): self._write_warning = False self._location = self.config["location"] self._accessor = resources.ResourceAccessor() - self._file_ = None # type: Optional[IO[Any]] - self._size = None # type: Optional[int] + self._file_: Optional[IO[Any]] = None + self._size: Optional[int] = None # Construct the lock now (shared if made before threading) in case we ever need it - self._lock = DummyLock() # type: Union[DummyLock, threading.Lock] + self._lock: Union[DummyLock, threading.Lock] = DummyLock() if constants.PARALLELISM == constants.Parallelism.Threading: self._lock = threading.Lock() # Instantiate the file to throw exceptions if the file doesn't open diff --git a/volatility3/framework/layers/qemu.py b/volatility3/framework/layers/qemu.py index 95d7d2a8f..ff0644dbc 100644 --- a/volatility3/framework/layers/qemu.py +++ b/volatility3/framework/layers/qemu.py @@ -40,7 +40,7 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer): metadata: Optional[Dict[str, Any]] = None) -> None: self._qemu_table_name = intermed.IntermediateSymbolTable.create(context, config_path, 'generic', 'qemu') self._configuration = None - self._compressed = set() # type: Set[int] + self._compressed: Set[int] = set() self._current_segment_name = b'' super().__init__(context = context, config_path = config_path, name = name, metadata = metadata) diff --git a/volatility3/framework/layers/registry.py b/volatility3/framework/layers/registry.py index ed6d045f6..55a6e5186 100644 --- a/volatility3/framework/layers/registry.py +++ b/volatility3/framework/layers/registry.py @@ -141,7 +141,7 @@ class RegistryHive(linear.LinearlyMappedLayer): if key.endswith("\\"): key = key[:-1] key_array = key.split('\\') - found_key = [] # type: List[str] + found_key: List[str] = [] while key_array and node_key: subkeys = node_key[-1].get_subkeys() for subkey in subkeys: diff --git a/volatility3/framework/layers/scanners/__init__.py b/volatility3/framework/layers/scanners/__init__.py index 2a167fc4e..3407b5784 100644 --- a/volatility3/framework/layers/scanners/__init__.py +++ b/volatility3/framework/layers/scanners/__init__.py @@ -48,7 +48,7 @@ class MultiStringScanner(layers.ScannerInterface): def __init__(self, patterns: List[bytes]) -> None: super().__init__() - self._pattern_trie = {} # type: Optional[Dict[int, Optional[Dict]]] + self._pattern_trie: Optional[Dict[int, Optional[Dict]]] = {} for pattern in patterns: self._process_pattern(pattern) self._regex = self._process_trie(self._pattern_trie) diff --git a/volatility3/framework/layers/scanners/multiregexp.py b/volatility3/framework/layers/scanners/multiregexp.py index 36c6410aa..45feb51d1 100644 --- a/volatility3/framework/layers/scanners/multiregexp.py +++ b/volatility3/framework/layers/scanners/multiregexp.py @@ -10,7 +10,7 @@ class MultiRegexp(object): """Algorithm for multi-string matching.""" def __init__(self) -> None: - self._pattern_strings = [] # type: List[bytes] + self._pattern_strings: List[bytes] = [] self._regex = re.compile(b'') def add_pattern(self, pattern: bytes) -> None: diff --git a/volatility3/framework/layers/segmented.py b/volatility3/framework/layers/segmented.py index 076838da6..80c89723a 100644 --- a/volatility3/framework/layers/segmented.py +++ b/volatility3/framework/layers/segmented.py @@ -25,9 +25,9 @@ class NonLinearlySegmentedLayer(interfaces.layers.TranslationLayerInterface, met super().__init__(context = context, config_path = config_path, name = name, metadata = metadata) self._base_layer = self.config["base_layer"] - self._segments = [] # type: List[Tuple[int, int, int, int]] - self._minaddr = None # type: Optional[int] - self._maxaddr = None # type: Optional[int] + self._segments: List[Tuple[int, int, int, int]] = [] + self._minaddr: Optional[int] = None + self._maxaddr: Optional[int] = None self._load_segments() diff --git a/volatility3/framework/objects/__init__.py b/volatility3/framework/objects/__init__.py index bf8dda515..15d2ef63e 100644 --- a/volatility3/framework/objects/__init__.py +++ b/volatility3/framework/objects/__init__.py @@ -95,7 +95,7 @@ class Function(interfaces.objects.ObjectInterface): class PrimitiveObject(interfaces.objects.ObjectInterface): """PrimitiveObject is an interface for any objects that should simulate a Python primitive.""" - _struct_type = int # type: ClassVar[Type] + _struct_type: ClassVar[Type] = int def __init__(self, context: interfaces.context.ContextInterface, type_name: str, object_info: interfaces.objects.ObjectInformation, data_format: DataFormatInfo) -> None: @@ -164,7 +164,7 @@ class PrimitiveObject(interfaces.objects.ObjectInterface): # https://mail.python.org/pipermail/python-dev/2004-February/042537.html class Boolean(PrimitiveObject, int): """Primitive Object that handles boolean types.""" - _struct_type = int # type: ClassVar[Type] + _struct_type: ClassVar[Type] = int class Integer(PrimitiveObject, int): @@ -173,17 +173,17 @@ class Integer(PrimitiveObject, int): class Float(PrimitiveObject, float): """Primitive Object that handles double or floating point numbers.""" - _struct_type = float # type: ClassVar[Type] + _struct_type: ClassVar[Type] = float class Char(PrimitiveObject, int): """Primitive Object that handles characters.""" - _struct_type = int # type: ClassVar[Type] + _struct_type: ClassVar[Type] = int class Bytes(PrimitiveObject, bytes): """Primitive Object that handles specific series of bytes.""" - _struct_type = bytes # type: ClassVar[Type] + _struct_type: ClassVar[Type] = bytes def __init__(self, context: interfaces.context.ContextInterface, @@ -227,7 +227,7 @@ class String(PrimitiveObject, str): max_length: specifies the maximum possible length that the string could hold within memory (for multibyte characters, this will not be the maximum length of the string) """ - _struct_type = str # type: ClassVar[Type] + _struct_type: ClassVar[Type] = str def __init__(self, context: interfaces.context.ContextInterface, @@ -453,7 +453,7 @@ class Enumeration(interfaces.objects.ObjectInterface, int): @classmethod def _generate_inverse_choices(cls, choices: Dict[str, int]) -> Dict[int, str]: """Generates the inverse choices for the object.""" - inverse_choices = {} # type: Dict[int, str] + inverse_choices: Dict[int, str] = {} for k, v in choices.items(): if v in inverse_choices: # Technically this shouldn't be a problem, but since we inverse cache @@ -601,7 +601,7 @@ class Array(interfaces.objects.ObjectInterface, collections.abc.Sequence): def __getitem__(self, i): """Returns the i-th item from the array.""" - result = [] # type: List[interfaces.objects.Template] + result: List[interfaces.objects.Template] = [] mask = self._context.layers[self.vol.layer_name].address_mask # We use the range function to deal with slices for us series = range(self.vol.count)[i] @@ -649,7 +649,7 @@ class AggregateType(interfaces.objects.ObjectInterface): size = size, members = members) # self._check_members(members) - self._concrete_members = {} # type: Dict[str, Dict] + self._concrete_members: Dict[str, Dict] = {} def has_member(self, member_name: str) -> bool: """Returns whether the object would contain a member called diff --git a/volatility3/framework/objects/templates.py b/volatility3/framework/objects/templates.py index 62094ff3d..b544d117f 100644 --- a/volatility3/framework/objects/templates.py +++ b/volatility3/framework/objects/templates.py @@ -65,7 +65,7 @@ class ObjectTemplate(interfaces.objects.Template): Returns: an object adhereing to the :class:`~volatility3.framework.interfaces.objects.ObjectInterface` """ - arguments = {} # type: Dict[str, Any] + arguments: Dict[str, Any] = {} for arg in self.vol: if arg != 'object_class': arguments[arg] = self.vol[arg] @@ -96,10 +96,10 @@ class ReferenceTemplate(interfaces.objects.Template): symbol_name, table_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] - relative_child_offset = _unresolved # type: ClassVar[Any] - has_member = _unresolved # type: ClassVar[Any] + size: ClassVar[Any] = property(_unresolved) + replace_child: ClassVar[Any] = _unresolved + relative_child_offset: ClassVar[Any] = _unresolved + has_member: ClassVar[Any] = _unresolved def __call__(self, context: interfaces.context.ContextInterface, object_info: interfaces.objects.ObjectInformation): template = context.symbol_space.get_type(self.vol.type_name) diff --git a/volatility3/framework/plugins/mac/lsmod.py b/volatility3/framework/plugins/mac/lsmod.py index 89a3c08af..7227c3289 100644 --- a/volatility3/framework/plugins/mac/lsmod.py +++ b/volatility3/framework/plugins/mac/lsmod.py @@ -57,7 +57,7 @@ class Lsmod(plugins.PluginInterface): except exceptions.InvalidAddressException: return [] - seen = set() # type: Set + seen: Set = set() while kmod != 0 and \ kmod not in seen and \ diff --git a/volatility3/framework/plugins/mac/psaux.py b/volatility3/framework/plugins/mac/psaux.py index 97f61c02e..14d765ef8 100644 --- a/volatility3/framework/plugins/mac/psaux.py +++ b/volatility3/framework/plugins/mac/psaux.py @@ -52,7 +52,7 @@ class Psaux(plugins.PluginInterface): task_name = utility.array_to_string(task.p_comm) - args = [] # type: List[bytes] + args: List[bytes] = [] while argc > 0: try: diff --git a/volatility3/framework/plugins/mac/pslist.py b/volatility3/framework/plugins/mac/pslist.py index 0829b42fb..1ebd89c97 100644 --- a/volatility3/framework/plugins/mac/pslist.py +++ b/volatility3/framework/plugins/mac/pslist.py @@ -124,7 +124,7 @@ class PsList(interfaces.plugins.PluginInterface): proc = kernel.object_from_symbol(symbol_name = "allproc").lh_first - seen = {} # type: Dict[int, int] + seen: 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).") @@ -165,7 +165,7 @@ class PsList(interfaces.plugins.PluginInterface): queue_entry = kernel.object_from_symbol(symbol_name = "tasks") - seen = {} # type: Dict[int, int] + seen: 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/volatility3/framework/plugins/timeliner.py b/volatility3/framework/plugins/timeliner.py index e95bba126..c3fe424b9 100644 --- a/volatility3/framework/plugins/timeliner.py +++ b/volatility3/framework/plugins/timeliner.py @@ -48,7 +48,7 @@ class Timeliner(interfaces.plugins.PluginInterface): super().__init__(*args, **kwargs) self.timeline = {} self.usable_plugins = None - self.automagics = None # type: Optional[List[interfaces.automagic.AutomagicInterface]] + self.automagics: Optional[List[interfaces.automagic.AutomagicInterface]] = None @classmethod def get_usable_plugins(cls, selected_list: List[str] = None) -> List[Type]: diff --git a/volatility3/framework/plugins/windows/callbacks.py b/volatility3/framework/plugins/windows/callbacks.py index 4c58c62c0..ef9ca98c0 100644 --- a/volatility3/framework/plugins/windows/callbacks.py +++ b/volatility3/framework/plugins/windows/callbacks.py @@ -191,9 +191,9 @@ class Callbacks(interfaces.plugins.PluginInterface): continue try: - component = ntkrnlmp.object( + component: Union[interfaces.renderers.BaseAbsentValue, interfaces.objects.ObjectInterface] = ntkrnlmp.object( "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/volatility3/framework/plugins/windows/handles.py b/volatility3/framework/plugins/windows/handles.py index 49098bb8b..9b249f83c 100644 --- a/volatility3/framework/plugins/windows/handles.py +++ b/volatility3/framework/plugins/windows/handles.py @@ -172,7 +172,7 @@ class Handles(interfaces.plugins.PluginInterface): A mapping of type indicies to type names """ - type_map = {} # type: Dict[int, str] + type_map: Dict[int, str] = {} kvo = context.layers[layer_name].config['kernel_virtual_offset'] ntkrnlmp = context.module(symbol_table, layer_name = layer_name, offset = kvo) diff --git a/volatility3/framework/plugins/windows/modscan.py b/volatility3/framework/plugins/windows/modscan.py index 968609ebc..4820a6fb8 100644 --- a/volatility3/framework/plugins/windows/modscan.py +++ b/volatility3/framework/plugins/windows/modscan.py @@ -81,7 +81,7 @@ class ModScan(interfaces.plugins.PluginInterface): Returns: A list of session layer names """ - seen_ids = [] # type: List[interfaces.objects.ObjectInterface] + seen_ids: List[interfaces.objects.ObjectInterface] = [] filter_func = pslist.PsList.create_pid_filter(pids or []) for proc in pslist.PsList.list_processes(context = context, diff --git a/volatility3/framework/plugins/windows/modules.py b/volatility3/framework/plugins/windows/modules.py index b030066db..ad2faefc9 100644 --- a/volatility3/framework/plugins/windows/modules.py +++ b/volatility3/framework/plugins/windows/modules.py @@ -86,7 +86,7 @@ class Modules(interfaces.plugins.PluginInterface): Returns: A list of session layer names """ - seen_ids = [] # type: List[interfaces.objects.ObjectInterface] + seen_ids: List[interfaces.objects.ObjectInterface] = [] filter_func = pslist.PsList.create_pid_filter(pids or []) for proc in pslist.PsList.list_processes(context = context, diff --git a/volatility3/framework/plugins/windows/poolscanner.py b/volatility3/framework/plugins/windows/poolscanner.py index 62a98615a..a6a6cae0c 100644 --- a/volatility3/framework/plugins/windows/poolscanner.py +++ b/volatility3/framework/plugins/windows/poolscanner.py @@ -338,7 +338,7 @@ class PoolScanner(plugins.PluginInterface): An Iterable of pool constraints and the pool headers associated with them """ # Setup the pattern - constraint_lookup = {} # type: Dict[bytes, PoolConstraint] + constraint_lookup: Dict[bytes, PoolConstraint] = {} for constraint in pool_constraints: if constraint.tag in constraint_lookup: raise ValueError(f"Constraint tag is used for more than one constraint: {repr(constraint.tag)}") diff --git a/volatility3/framework/plugins/windows/pstree.py b/volatility3/framework/plugins/windows/pstree.py index 151a35d52..b8c99688f 100644 --- a/volatility3/framework/plugins/windows/pstree.py +++ b/volatility3/framework/plugins/windows/pstree.py @@ -18,9 +18,9 @@ class PsTree(interfaces.plugins.PluginInterface): def __init__(self, *args, **kwargs) -> None: super().__init__(*args, **kwargs) - self._processes = {} # type: Dict[int, interfaces.objects.ObjectInterface] - self._levels = {} # type: Dict[int, int] - self._children = {} # type: Dict[int, Set[int]] + self._processes: Dict[int, interfaces.objects.ObjectInterface] = {} + self._levels: Dict[int, int] = {} + self._children: Dict[int, Set[int]] = {} @classmethod def get_requirements(cls): diff --git a/volatility3/framework/plugins/windows/registry/printkey.py b/volatility3/framework/plugins/windows/registry/printkey.py index 380d3bbca..bad438256 100644 --- a/volatility3/framework/plugins/windows/registry/printkey.py +++ b/volatility3/framework/plugins/windows/registry/printkey.py @@ -130,7 +130,7 @@ class PrintKey(interfaces.plugins.PluginInterface): if isinstance(value_type, renderers.UnreadableValue): vollog.debug("Couldn't read registry value type, so data is unreadable") - value_data = renderers.UnreadableValue() # type: Union[interfaces.renderers.BaseAbsentValue, bytes] + value_data: Union[interfaces.renderers.BaseAbsentValue, bytes] = renderers.UnreadableValue() else: try: value_data = node.decode_data() diff --git a/volatility3/framework/plugins/windows/registry/userassist.py b/volatility3/framework/plugins/windows/registry/userassist.py index 804e9f60e..b3a3b8f3a 100644 --- a/volatility3/framework/plugins/windows/registry/userassist.py +++ b/volatility3/framework/plugins/windows/registry/userassist.py @@ -151,12 +151,12 @@ class UserAssist(interfaces.plugins.PluginInterface): countkey_last_write_time = conversion.wintime_to_datetime(countkey.LastWriteTime.QuadPart) # output the parent Count key - result = ( + result: Tuple[int, Tuple[format_hints.Hex, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]] = ( 0, (renderers.format_hints.Hex(hive.hive_offset), hive_name, countkey_path, countkey_last_write_time, "Key", renderers.NotApplicableValue(), renderers.NotApplicableValue(), renderers.NotApplicableValue(), renderers.NotApplicableValue(), renderers.NotApplicableValue(), renderers.NotApplicableValue(), renderers.NotApplicableValue()) - ) # type: Tuple[int, Tuple[format_hints.Hex, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]] + ) yield result # output any subkeys under Count diff --git a/volatility3/framework/plugins/windows/strings.py b/volatility3/framework/plugins/windows/strings.py index 7a55079a9..775ffc732 100644 --- a/volatility3/framework/plugins/windows/strings.py +++ b/volatility3/framework/plugins/windows/strings.py @@ -40,17 +40,18 @@ class Strings(interfaces.plugins.PluginInterface): def run(self): return renderers.TreeGrid([("String", str), ("Physical Address", format_hints.Hex), ("Result", str)], - self._generator()) + self._generator) + @property def _generator(self) -> Generator[Tuple, None, None]: """Generates results from a strings file.""" - string_list = [] # type: List[Tuple[int,bytes]] + string_list: List[Tuple[int,bytes]] = [] # Test strings file format is accurate accessor = resources.ResourceAccessor() strings_fp = accessor.open(self.config['strings_file'], "rb") line = strings_fp.readline() - count = 0 # type: float + count: float = 0 while line: count += 1 try: @@ -66,7 +67,8 @@ class Strings(interfaces.plugins.PluginInterface): progress_callback = self._progress_callback, pid_list = self.config['pid']) - last_prog = line_count = 0 # type: float + last_prog: float = 0 + line_count: float = 0 num_strings = len(string_list) for offset, string in string_list: line_count += 1 @@ -119,7 +121,7 @@ class Strings(interfaces.plugins.PluginInterface): filter = pslist.PsList.create_pid_filter(pid_list) layer = context.layers[layer_name] - reverse_map = dict() # type: Dict[int, Set[Tuple[str, int]]] + reverse_map: Dict[int, Set[Tuple[str, int]]] = dict() if isinstance(layer, intel.Intel): # We don't care about errors, we just wanted chunks that map correctly for mapval in layer.mapping(0x0, layer.maximum_address, ignore_errors = True): diff --git a/volatility3/framework/plugins/windows/virtmap.py b/volatility3/framework/plugins/windows/virtmap.py index 552564cad..238b0df19 100644 --- a/volatility3/framework/plugins/windows/virtmap.py +++ b/volatility3/framework/plugins/windows/virtmap.py @@ -41,7 +41,7 @@ class VirtMap(interfaces.plugins.PluginInterface): if not isinstance(layer, intel.Intel): raise - result = {} # type: Dict[str, List[Tuple[int, int]]] + result: Dict[str, List[Tuple[int, int]]] = {} system_va_type = module.get_enumeration('_MI_SYSTEM_VA_TYPE') large_page_size = (layer.page_size ** 2) // module.get_type("_MMPTE").size @@ -84,7 +84,7 @@ class VirtMap(interfaces.plugins.PluginInterface): def _enumerate_system_va_type(cls, large_page_size: int, system_range_start: int, module: interfaces.context.ModuleInterface, type_array: interfaces.objects.ObjectInterface) -> Dict[str, List[Tuple[int, int]]]: - result = {} # type: Dict[str, List[Tuple[int, int]]] + result: Dict[str, List[Tuple[int, int]]] = {} system_va_type = module.get_enumeration('_MI_SYSTEM_VA_TYPE') start = system_range_start prev_entry = -1 diff --git a/volatility3/framework/renderers/__init__.py b/volatility3/framework/renderers/__init__.py index 5b1013574..23e686a07 100644 --- a/volatility3/framework/renderers/__init__.py +++ b/volatility3/framework/renderers/__init__.py @@ -158,8 +158,8 @@ class TreeGrid(interfaces.renderers.TreeGrid): """ self._populated = False self._row_count = 0 - self._children = [] # type: List[interfaces.renderers.TreeNode] - converted_columns = [] # type: List[interfaces.renderers.Column] + self._children: List[interfaces.renderers.TreeNode] = [] + converted_columns: List[interfaces.renderers.Column] = [] if len(columns) < 1: raise ValueError("Columns must be a list containing at least one column") for (name, column_type) in columns: @@ -207,7 +207,7 @@ class TreeGrid(interfaces.renderers.TreeGrid): if not self.populated: try: - prev_nodes = [] # type: List[interfaces.renderers.TreeNode] + prev_nodes: List[interfaces.renderers.TreeNode] = [] for (level, item) in self._generator: parent_index = min(len(prev_nodes), level) parent = prev_nodes[parent_index - 1] if parent_index > 0 else None diff --git a/volatility3/framework/renderers/conversion.py b/volatility3/framework/renderers/conversion.py index b60b47411..5737abef5 100644 --- a/volatility3/framework/renderers/conversion.py +++ b/volatility3/framework/renderers/conversion.py @@ -24,7 +24,7 @@ def wintime_to_datetime(wintime: int) -> Union[interfaces.renderers.BaseAbsentVa def unixtime_to_datetime(unixtime: int) -> Union[interfaces.renderers.BaseAbsentValue, datetime.datetime]: - ret = renderers.UnparsableValue() # type: Union[interfaces.renderers.BaseAbsentValue, datetime.datetime] + ret: Union[interfaces.renderers.BaseAbsentValue, datetime.datetime] = renderers.UnparsableValue() if unixtime > 0: try: diff --git a/volatility3/framework/renderers/format_hints.py b/volatility3/framework/renderers/format_hints.py index b169f44c9..486e164b3 100644 --- a/volatility3/framework/renderers/format_hints.py +++ b/volatility3/framework/renderers/format_hints.py @@ -46,7 +46,7 @@ class MultiTypeData(bytes): encoding: str = 'utf-16-le', split_nulls: bool = False, show_hex: bool = False) -> None: - self.converted_int = False # type: bool + self.converted_int: bool = False if isinstance(original, int): self.converted_int = True self.encoding = encoding diff --git a/volatility3/framework/symbols/__init__.py b/volatility3/framework/symbols/__init__.py index 6a8ab8246..3d0e01803 100644 --- a/volatility3/framework/symbols/__init__.py +++ b/volatility3/framework/symbols/__init__.py @@ -31,15 +31,15 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface): def __init__(self) -> None: super().__init__() - self._dict = collections.OrderedDict() # type: Dict[str, interfaces.symbols.BaseSymbolTableInterface] + self._dict: Dict[str, interfaces.symbols.BaseSymbolTableInterface] = collections.OrderedDict() # Permanently cache all resolved symbols - self._resolved = {} # type: Dict[str, interfaces.objects.Template] - self._resolved_symbols = {} # type: Dict[str, interfaces.objects.Template] + self._resolved: Dict[str, interfaces.objects.Template] = {} + self._resolved_symbols: Dict[str, interfaces.objects.Template] = {} def clear_symbol_cache(self, table_name: str = None) -> None: """Clears the symbol cache for the specified table name. If no table name is specified, the caches of all symbol tables are cleared.""" - table_list = list() # type: List[interfaces.symbols.BaseSymbolTableInterface] + table_list: List[interfaces.symbols.BaseSymbolTableInterface] = list() if table_name is None: table_list = list(self._dict.values()) else: diff --git a/volatility3/framework/symbols/intermed.py b/volatility3/framework/symbols/intermed.py index b20760d6b..a6e76ae7a 100644 --- a/volatility3/framework/symbols/intermed.py +++ b/volatility3/framework/symbols/intermed.py @@ -282,8 +282,8 @@ class ISFormatTable(interfaces.symbols.SymbolTableInterface, metaclass = ABCMeta raise TypeError("Native table not provided") nt.name = name + "_natives" super().__init__(context, config_path, name, nt, table_mapping = table_mapping) - self._overrides = {} # type: Dict[str, Type[interfaces.objects.ObjectInterface]] - self._symbol_cache = {} # type: Dict[str, interfaces.symbols.SymbolInterface] + self._overrides: Dict[str, Type[interfaces.objects.ObjectInterface]] = {} + self._symbol_cache: Dict[str, interfaces.symbols.SymbolInterface] = {} def _get_natives(self) -> Optional[interfaces.symbols.NativeTableInterface]: """Determines the appropriate native_types to use from the JSON diff --git a/volatility3/framework/symbols/linux/__init__.py b/volatility3/framework/symbols/linux/__init__.py index cd0495a2c..6d09bbe46 100644 --- a/volatility3/framework/symbols/linux/__init__.py +++ b/volatility3/framework/symbols/linux/__init__.py @@ -46,7 +46,7 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface): @classmethod def _do_get_path(cls, rdentry, rmnt, dentry, vfsmnt) -> str: - ret_path = [] # type: List[str] + ret_path: List[str] = [] while dentry != rdentry or vfsmnt != rmnt: dname = dentry.path() diff --git a/volatility3/framework/symbols/mac/__init__.py b/volatility3/framework/symbols/mac/__init__.py index 241b4ffba..152f0bbc2 100644 --- a/volatility3/framework/symbols/mac/__init__.py +++ b/volatility3/framework/symbols/mac/__init__.py @@ -160,7 +160,7 @@ class MacUtilities(interfaces.configuration.VersionableInterface): list_next_member: str, next_member: str, max_elements: int = 4096) -> Iterable[interfaces.objects.ObjectInterface]: - seen = set() # type: Set[int] + seen: Set[int] = set() try: current = queue.member(attr = list_head_member) diff --git a/volatility3/framework/symbols/mac/extensions/__init__.py b/volatility3/framework/symbols/mac/extensions/__init__.py index 538c4101e..b0d75b93b 100644 --- a/volatility3/framework/symbols/mac/extensions/__init__.py +++ b/volatility3/framework/symbols/mac/extensions/__init__.py @@ -48,7 +48,7 @@ class proc(generic.GenericIntelProcess): except exceptions.InvalidAddressException: return - seen = set() # type: Set[int] + seen: Set[int] = set() for i in range(task.map.hdr.nentries): if not current_map or current_map.vol.offset in seen: diff --git a/volatility3/framework/symbols/native.py b/volatility3/framework/symbols/native.py index c53ff6f16..d9833e26d 100644 --- a/volatility3/framework/symbols/native.py +++ b/volatility3/framework/symbols/native.py @@ -15,7 +15,7 @@ class NativeTable(interfaces.symbols.NativeTableInterface): def __init__(self, name: str, native_dictionary: Dict[str, Any]) -> None: super().__init__(name, self) self._native_dictionary = copy.deepcopy(native_dictionary) - self._overrides = {} # type: Dict[str, interfaces.objects.ObjectInterface] + self._overrides: Dict[str, interfaces.objects.ObjectInterface] = {} for native_type in self._native_dictionary: native_class, _native_struct = self._native_dictionary[native_type] self._overrides[native_type] = native_class @@ -49,8 +49,8 @@ class NativeTable(interfaces.symbols.NativeTableInterface): table_name, type_name = name_split prefix = table_name + constants.BANG - additional = {} # type: Dict[str, Any] - obj = None # type: Optional[Type[interfaces.objects.ObjectInterface]] + additional: Dict[str, Any] = {} + obj: Optional[Type[interfaces.objects.ObjectInterface]] = None if type_name == 'void' or type_name == 'function': obj = objects.Void elif type_name == 'array': diff --git a/volatility3/framework/symbols/windows/extensions/__init__.py b/volatility3/framework/symbols/windows/extensions/__init__.py index 5b057ff62..13e07cb3b 100755 --- a/volatility3/framework/symbols/windows/extensions/__init__.py +++ b/volatility3/framework/symbols/windows/extensions/__init__.py @@ -388,7 +388,7 @@ class FILE_OBJECT(objects.StructType, pool.ExecutiveObject): self.FileName.Buffer) def file_name_with_device(self) -> Union[str, interfaces.renderers.BaseAbsentValue]: - name = renderers.UnreadableValue() # type: Union[str, interfaces.renderers.BaseAbsentValue] + name: Union[str, interfaces.renderers.BaseAbsentValue] = renderers.UnreadableValue() # this pointer needs to be checked against native_layer_name because the object may # be instantiated from a primary (virtual) layer or a memory (physical) layer. diff --git a/volatility3/framework/symbols/windows/extensions/pool.py b/volatility3/framework/symbols/windows/extensions/pool.py index 59213aaff..d470797b5 100644 --- a/volatility3/framework/symbols/windows/extensions/pool.py +++ b/volatility3/framework/symbols/windows/extensions/pool.py @@ -214,7 +214,7 @@ class POOL_HEADER_VISTA(POOL_HEADER): class POOL_TRACKER_BIG_PAGES(objects.StructType): """A kernel big page pool tracker.""" - pool_type_lookup = {} # type: Dict[str, str] + pool_type_lookup: Dict[str, str] = {} def _generate_pool_type_lookup(self): # Enumeration._generate_inverse_choices() raises ValueError because multiple enum names map to the same diff --git a/volatility3/framework/symbols/windows/pdbconv.py b/volatility3/framework/symbols/windows/pdbconv.py index e530fcbd7..b582fc299 100644 --- a/volatility3/framework/symbols/windows/pdbconv.py +++ b/volatility3/framework/symbols/windows/pdbconv.py @@ -265,18 +265,18 @@ class PdbReader: database_name: Optional[str] = None, progress_callback: constants.ProgressCallback = None) -> None: self._layer_name, self._context = self.load_pdb_layer(context, location) - self._dbiheader = None # type: Optional[interfaces.objects.ObjectInterface] + self._dbiheader: Optional[interfaces.objects.ObjectInterface] = None if not progress_callback: progress_callback = lambda x, y: None self._progress_callback = progress_callback - self.types = [ - ] # type: List[Tuple[interfaces.objects.ObjectInterface, Optional[str], interfaces.objects.ObjectInterface]] - self.bases = {} # type: Dict[str, Any] - self.user_types = {} # type: Dict[str, Any] - self.enumerations = {} # type: Dict[str, Any] - self.symbols = {} # type: Dict[str, Any] - self._omap_mapping = [] # type: List[Tuple[int, int]] - self._sections = [] # type: List[interfaces.objects.ObjectInterface] + self.types: List[Tuple[interfaces.objects.ObjectInterface, Optional[str], interfaces.objects.ObjectInterface]] = [ + ] + self.bases: Dict[str, Any] = {} + self.user_types: Dict[str, Any] = {} + self.enumerations: Dict[str, Any] = {} + self.symbols: Dict[str, Any] = {} + self._omap_mapping: List[Tuple[int, int]] = [] + self._sections: List[interfaces.objects.ObjectInterface] = [] self.metadata = {"format": "6.1.0", "windows": {}} self._database_name = database_name @@ -381,7 +381,7 @@ class PdbReader: raise ValueError("Maximum {} index is smaller than minimum TPI index, found: {} < {} ".format( stream_name, header.index_max, header.index_min)) # Reset the state - info_references = {} # type: Dict[str, int] + info_references: Dict[str, int] = {} offset = header.header_size # Ensure we use the same type everywhere length_type = "unsigned short" @@ -586,7 +586,7 @@ class PdbReader: if index < 0x1000: base_name, base = primatives[index & 0xff] self.bases[base_name] = base - result = {"kind": "base", "name": base_name} # type: Union[List[Dict[str, Any]], Dict[str, Any]] + result: Union[List[Dict[str, Any]], Dict[str, Any]] = {"kind": "base", "name": base_name} indirection = (index & 0xf00) if indirection: pointer_name, pointer_base = indirections[indirection] @@ -639,7 +639,7 @@ class PdbReader: """Returns the size of the structure based on the type index provided.""" result = -1 - name = '' # type: Optional[str] + name: Optional[str] = '' if index < 0x1000: if (index & 0xf00): _, base = indirections[index & 0xf00] @@ -837,7 +837,7 @@ class PdbReader: def convert_fields(self, fields: int) -> Dict[Optional[str], Dict[str, Any]]: """Converts a field list into a list of fields.""" - result = {} # type: Dict[Optional[str], Dict[str, Any]] + result: Dict[Optional[str], Dict[str, Any]] = {} _, _, fields_struct = self.types[fields] if not isinstance(fields_struct, list): vollog.warning("Fields structure did not contain a list of fields") diff --git a/volatility3/schemas/__init__.py b/volatility3/schemas/__init__.py index 9340b29f4..5fbed7e09 100644 --- a/volatility3/schemas/__init__.py +++ b/volatility3/schemas/__init__.py @@ -18,7 +18,7 @@ cached_validation_filepath = os.path.join(constants.CACHE_PATH, "valid_isf.hashc def load_cached_validations() -> Set[str]: """Loads up the list of successfully cached json objects, so we don't need to revalidate them.""" - validhashes = set() # type: Set + validhashes: Set = set() if os.path.exists(cached_validation_filepath): with open(cached_validation_filepath, "r") as f: validhashes.update(json.load(f))