Many more typing fixes.

volatility/cli/__init__.py

@@ -14,7 +14,7 @@ import json
 import logging
 import os
 import sys
-from typing import Union, Type, Dict
+from typing import Any, Dict, Type, Union
 from urllib import parse, request
 
 import volatility.plugins

volatility/framework/automagic/mac.py

@@ -121,7 +121,7 @@ class MacUtilities(object):
             if not isinstance(sym_layer, layers.intel.Intel):
                 raise TypeError("Layer name {} is not an intel space")
             aslr_layer = sym_layer.config['memory_layer']
-            _, aslr_shift = cls.find_aslr(context, symbol_table, aslr_layer)
+            aslr_shift = cls.find_aslr(context, symbol_table, aslr_layer)
 
         symbols.mask_symbol_table(sym_table, sym_layer.address_mask, aslr_shift)
 
@@ -149,8 +149,7 @@ class MacUtilities(object):
                   layer_name: str,
                   compare_banner: str = "",
                   compare_banner_offset: int = 0,
-                  progress_callback: validity.ProgressCallback = None) \
-            -> Tuple[int, int]:
+                  progress_callback: validity.ProgressCallback = None) -> int:
         """Determines the offset of the actual DTB in physical space and its symbol offset"""
         version_symbol = symbol_table + constants.BANG + 'version'
         version_json_address = context.symbol_space.get_symbol(version_symbol).address

volatility/framework/automagic/pdbscan.py

@@ -205,7 +205,8 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
                         context.config[join(sub_config_path, "isf_url")] = isf_path
                         # Construct the appropriate symbol table
                         config_path = interfaces.configuration.parent_path(sub_config_path)
-                        requirement.construct(context, config_path)
+                        if isinstance(requirement, interfaces.configuration.ConstructableRequirementInterface):
+                            requirement.construct(context, config_path)
                         break
                     else:
                         vollog.debug("Required symbol library path not found: {}".format(filter_string))

volatility/framework/automagic/symbol_cache.py

@@ -4,7 +4,7 @@ import pickle
 import urllib
 import urllib.parse
 import urllib.request
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 from volatility.framework import constants, exceptions, interfaces
 from volatility.framework.symbols import intermed
@@ -21,12 +21,14 @@ 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
-    symbol_name = "banner_name"
-    banner_path = None
+    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
         if os.path.exists(cls.banner_path):
             with open(cls.banner_path, "rb") as f:

volatility/framework/automagic/symbol_finder.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Iterable, List, Tuple
+from typing import Any, Iterable, List, Tuple, Dict, Type, Optional
 
 from volatility.framework import interfaces, validity
 from volatility.framework.automagic import symbol_cache
@@ -13,13 +13,13 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
     """Symbol loader based on signature strings"""
     priority = 40
 
-    banner_config_key = "banner"
-    banner_cache = None
-    symbol_class = None
+    banner_config_key: str = "banner"
+    banner_cache: Optional[Type[symbol_cache.SymbolBannerCache]] = None
+    symbol_class: Optional[str] = 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.ConstructableRequirementInterface]]
+        self._requirements = []  # type: List[Tuple[str, interfaces.configuration.RequirementInterface]]
         self._banners = {}  # type: symbol_cache.BannersType
 
     @property

volatility/framework/contexts/__init__.py

@@ -103,14 +103,12 @@ class Context(interfaces.context.ContextInterface):
             object_info = interfaces.objects.ObjectInformation(
                 layer_name = layer_name, offset = offset, native_layer_name = native_layer_name))
 
-    @functools.lru_cache()
-    def module(
-            self,  # type: ignore # FIXME: mypy #5107
-            module_name: str,
-            layer_name: str,
-            offset: int,
-            native_layer_name: Optional[str] = None,
-            size: Optional[int] = None) -> interfaces.context.ModuleInterface:
+    def module(self,
+               module_name: str,
+               layer_name: str,
+               offset: int,
+               native_layer_name: Optional[str] = None,
+               size: Optional[int] = None) -> interfaces.context.ModuleInterface:
         """Creates a module object"""
         if size:
             return SizedModule(

volatility/framework/interfaces/automagic.py

@@ -3,13 +3,11 @@
 Automagic objects attempt to automatically fill configuration values that a user has not filled.
 """
 from abc import ABCMeta
-from typing import TypeVar, Any, List, Optional, Tuple, Union, Type
+from typing import Any, List, Optional, Tuple, Union, Type
 
 from volatility.framework import interfaces, validity
 from volatility.framework.configuration import requirements
 
-R = TypeVar('R', bound = interfaces.configuration.RequirementInterface)
-
 
 class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metaclass = ABCMeta):
     """Class that defines an automagic component that can help fulfill a Requirement
@@ -49,15 +47,16 @@ class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metacla
         """Runs the automagic over the configurable"""
         return []
 
-    # TODO: requirement_type can be made Union[Type[T], Tuple[Type[T], ...]]
+    # TODO: requirement_type can be made UnionType[Type[T], Tuple[Type[T], ...]]
     #       once mypy properly supports Tuples in instance
 
     def find_requirements(self,
                           context: interfaces.context.ContextInterface,
                           config_path: str,
                           requirement_root: interfaces.configuration.RequirementInterface,
-                          requirement_type: Union[Tuple[Type[R], ...], Type[R]],
-                          shortcut: bool = True) -> List[Tuple[str, R]]:
+                          requirement_type: Union[Tuple[Type[interfaces.configuration.RequirementInterface], ...],
+                                                  Type[interfaces.configuration.RequirementInterface]],
+                          shortcut: bool = True) -> List[Tuple[str, interfaces.configuration.RequirementInterface]]:
         """Determines if there is actually an unfulfilled requirement waiting
 
         This ensures we do not carry out an expensive search when there is no requirement for a particular requirement
@@ -73,7 +72,7 @@ class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metacla
             A list of tuples containing the config_path, sub_config_path and requirement identifying the SymbolRequirements
         """
         sub_config_path = interfaces.configuration.path_join(config_path, requirement_root.name)
-        results = []  # type: List[Tuple[str, R]]
+        results = []  # type: List[Tuple[str, interfaces.configuration.RequirementInterface]]
         recurse = not shortcut
         if isinstance(requirement_root, requirement_type):
             if recurse or requirement_root.unsatisfied(context, config_path):

volatility/framework/interfaces/configuration.py

@@ -16,7 +16,7 @@ import random
 import string
 import sys
 from abc import ABCMeta, abstractmethod
-from typing import Any, Dict, Generator, List, Optional, Type, Union
+from typing import Any, ClassVar, Dict, Generator, List, Optional, Type, Union
 
 from volatility.framework import constants, interfaces, validity
 from volatility.framework.interfaces.context import ContextInterface

volatility/framework/interfaces/context.py

@@ -75,7 +75,12 @@ class ContextInterface(object, metaclass = ABCMeta):
            Memory constraints may become an issue for this function depending on how much is actually stored in the context"""
         return copy.deepcopy(self)
 
-    def module(self, module_name: str, layer_name: str, offset: int, size: Optional[int] = None) -> 'ModuleInterface':
+    def module(self,
+               module_name: str,
+               layer_name: str,
+               offset: int,
+               native_layer_name: Optional[str] = None,
+               size: Optional[int] = None) -> 'ModuleInterface':
         """Create a module object """
 
 

volatility/framework/interfaces/objects.py

@@ -5,7 +5,7 @@ import collections
 import collections.abc
 import logging
 from abc import ABCMeta, abstractmethod
-from typing import Any, List, Mapping
+from typing import Any, Dict, List, Mapping, Optional
 
 from volatility.framework import constants, validity, interfaces
 from volatility.framework.interfaces import context as interfaces_context
@@ -53,7 +53,12 @@ class ObjectInformation(ReadOnlyMapping):
     in a single place.  These values are based on the :class:`ReadOnlyMapping` class, to prevent their modification.
     """
 
-    def __init__(self, layer_name, offset, member_name = None, parent = None, native_layer_name = None):
+    def __init__(self,
+                 layer_name: str,
+                 offset: int,
+                 member_name: Optional[str] = None,
+                 parent: Optional['ObjectInterface'] = None,
+                 native_layer_name: Optional[str] = None):
         self._check_type(offset, int)
         if parent:
             self._check_type(parent, ObjectInterface)

volatility/framework/interfaces/plugins.py

@@ -7,7 +7,7 @@ They are called and carry out some algorithms on data stored in layers using obj
 import io
 import logging
 from abc import ABCMeta, abstractmethod
-from typing import TYPE_CHECKING, List
+from typing import List, Optional, TYPE_CHECKING
 
 from volatility.framework import exceptions
 from volatility.framework import validity

volatility/framework/objects/__init__.py

@@ -2,7 +2,7 @@ import collections
 import logging
 import struct
 from collections import abc
-from typing import Any, ClassVar, Dict, List, Iterable, Optional, Tuple, Type, Union, overload
+from typing import Any, ClassVar, Dict, List, Iterable, Optional, Tuple, Type, Union as TUnion, overload
 
 from volatility.framework import interfaces
 from volatility.framework.interfaces.objects import ObjectInformation
@@ -13,8 +13,8 @@ vollog = logging.getLogger(__name__)
 DataFormatInfo = collections.namedtuple('DataFormatInfo', ['length', 'byteorder', 'signed'])
 
 
-def convert_data_to_value(data: bytes, struct_type: Type[Union[int, float, bytes, str, bool]],
-                          data_format: DataFormatInfo) -> Union[int, float, bytes, str, bool]:
+def convert_data_to_value(data: bytes, struct_type: Type[TUnion[int, float, bytes, str, bool]],
+                          data_format: DataFormatInfo) -> TUnion[int, float, bytes, str, bool]:
     """Converts a series of bytes to a particular type of value"""
     if struct_type == int:
         return int.from_bytes(data, byteorder = data_format.byteorder, signed = data_format.signed)
@@ -33,8 +33,8 @@ def convert_data_to_value(data: bytes, struct_type: Type[Union[int, float, bytes
     return struct.unpack(struct_format, data)[0]
 
 
-def convert_value_to_data(value: Union[int, float, bytes, str, bool],
-                          struct_type: Type[Union[int, float, bytes, str, bool]],
+def convert_value_to_data(value: TUnion[int, float, bytes, str, bool],
+                          struct_type: Type[TUnion[int, float, bytes, str, bool]],
                           data_format: DataFormatInfo) -> bytes:
     """Converts a particular value to a series of bytes"""
     if not isinstance(value, struct_type):
@@ -87,12 +87,12 @@ class PrimitiveObject(interfaces.objects.ObjectInterface):
             context = context, type_name = type_name, object_info = object_info, data_format = data_format)
         self._data_format = data_format
 
-    def __new__(cls: 'PrimitiveObject',
+    def __new__(cls: Type,
                 context: interfaces.context.ContextInterface,
                 type_name: str,
                 object_info: interfaces.objects.ObjectInformation,
                 data_format: DataFormatInfo,
-                new_value: Union[int, float, bool, bytes, str] = None,
+                new_value: TUnion[int, float, bool, bytes, str] = None,
                 **kwargs) -> 'PrimitiveObject':
         """Creates the appropriate class and returns it so that the native type is inherited
 
@@ -122,7 +122,7 @@ class PrimitiveObject(interfaces.objects.ObjectInterface):
 
     @classmethod
     def _unmarshall(cls, context: interfaces.context.ContextInterface, data_format: DataFormatInfo,
-                    object_info: ObjectInformation) -> Union[int, float, bool, bytes, str]:
+                    object_info: ObjectInformation) -> TUnion[int, float, bool, bytes, str]:
         data = context.memory.read(object_info.layer_name, object_info.offset, data_format.length)
         return convert_data_to_value(data, cls._struct_type, data_format)
 
@@ -133,7 +133,7 @@ class PrimitiveObject(interfaces.objects.ObjectInterface):
             """Returns the size of the templated object"""
             return template.vol.data_format.length
 
-    def write(self, value: Union[int, float, bool, bytes, str]) -> None:
+    def write(self, value: TUnion[int, float, bool, bytes, str]) -> None:
         """Writes the object into the layer of the context at the current offset"""
         data = convert_value_to_data(value, self._struct_type, self._data_format)
         return self._context.memory.write(self.vol.layer_name, self.vol.offset, data)
@@ -174,7 +174,7 @@ class Bytes(PrimitiveObject, bytes):
             data_format = DataFormatInfo(length, "big", False))
         self._vol['length'] = length
 
-    def __new__(cls: 'Bytes',
+    def __new__(cls: Type,
                 context: interfaces.context.ContextInterface,
                 type_name: str,
                 object_info: interfaces.objects.ObjectInformation,
@@ -215,7 +215,7 @@ class String(PrimitiveObject, str):
         self._vol['encoding'] = encoding
         self._vol['errors'] = errors
 
-    def __new__(cls,
+    def __new__(cls: Type,
                 context: interfaces.context.ContextInterface,
                 type_name: str,
                 object_info: interfaces.objects.ObjectInformation,
@@ -234,10 +234,9 @@ class String(PrimitiveObject, str):
             params['errors'] = errors
         # Pass the encoding and error parameters to the string constructor to appropriately encode the string
         value = cls._struct_type.__new__(
-            cls,  # type: ignore
+            cls,
             cls._unmarshall(
-                context, data_format = DataFormatInfo(max_length, "big", False), object_info = object_info),
-            **params)
+                context, data_format = DataFormatInfo(max_length, "big", False), object_info = object_info), **params)
         if value.find('\x00') >= 0:
             value = value[:value.find('\x00')]
         return value

volatility/framework/plugins/timeliner.py

@@ -47,16 +47,18 @@ class Timeliner(interfaces.plugins.PluginInterface):
         plugin_list = list(framework.class_subclasses(TimeLinerInterface))
 
         # Get the filter from the configuration
-        def passthrough(_n, _s):
+        def passthrough(name: str, selected: List[str]) -> bool:
             return True
 
         filter_func = passthrough
         if selected_list:
 
-            def filter_plugins(name, selected):
+            def filter_plugins(name: str, selected: List[str]) -> bool:
                 return any([s in name for s in selected])
 
             filter_func = filter_plugins
+        else:
+            selected_list = []
 
         return [plugin_class for plugin_class in plugin_list if filter_func(plugin_class.__name__, selected_list)]
 

volatility/framework/plugins/windows/handles.py

@@ -144,7 +144,7 @@ class Handles(interfaces_plugins.PluginInterface):
         ptrs = ntkrnlmp.object(
             type_name = "array", offset = kvo + table_addr, subtype = ntkrnlmp.get_type("pointer"), count = 100)
 
-        for i, ptr in enumerate(ptrs):
+        for i, ptr in enumerate(ptrs):  #type: ignore
             # the first entry in the table is always null. break the
             # loop when we encounter the first null entry after that
             if i > 0 and ptr == 0:

volatility/framework/plugins/windows/poolscanner.py

@@ -1,6 +1,6 @@
 import enum
 import logging
-from typing import Optional, Tuple, List, Generator
+from typing import Dict, Generator, List, Optional, Tuple
 
 import volatility.plugins.windows.handles as handles
 

volatility/framework/plugins/windows/pstree.py

@@ -1,4 +1,6 @@
-from volatility.framework import objects
+from typing import Dict, Set
+
+from volatility.framework import objects, interfaces
 from volatility.framework.renderers import format_hints
 from volatility.plugins.windows import pslist
 
@@ -8,9 +10,9 @@ class PsTree(pslist.PsList):
 
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
-        self._processes = {}
-        self._levels = {}
-        self._children = {}
+        self._processes = {}  # type: Dict[int, interfaces.objects.ObjectInterface]
+        self._levels = {}  # type: Dict[int, int]
+        self._children = {}  # type: Dict[int, Set[int]]
 
     def find_level(self, pid: objects.Pointer) -> None:
         """Finds how deep the pid is in the processes list"""

volatility/framework/plugins/windows/registry/userassist.py

@@ -3,7 +3,7 @@ import datetime
 import json
 import logging
 import os
-from typing import List
+from typing import Any, List, Tuple
 
 from volatility.framework import exceptions, renderers, constants, interfaces
 from volatility.framework.configuration import requirements

volatility/framework/plugins/windows/ssdt.py

@@ -23,7 +23,7 @@ class SSDT(plugins.PluginInterface):
             requirements.SymbolRequirement(name = "nt_symbols", description = "Windows OS")
         ]
 
-    def _generator(self, mods: Iterator[Any]) -> Iterator[Tuple[int, Tuple[int, int, str, str]]]:
+    def _generator(self, mods: Iterator[Any]) -> Iterator[Tuple[int, Tuple[int, int, Any, Any]]]:
 
         layer_name = self.config['primary']
         context_modules = []
@@ -67,14 +67,14 @@ class SSDT(plugins.PluginInterface):
         if is_kernel_64:
             array_subtype = "long"
 
-            def kvo_calulator(func):
+            def kvo_calulator(func: int) -> int:
                 return kvo + service_table_address + (func >> 4)
 
             find_address = kvo_calulator
         else:
             array_subtype = "unsigned long"
 
-            def passthrough(func):
+            def passthrough(func: int) -> int:
                 return func
 
             find_address = passthrough

volatility/framework/plugins/windows/vadinfo.py

@@ -73,13 +73,13 @@ class VadInfo(interfaces.plugins.PluginInterface):
 
     def _generator(self, procs):
 
-        def passthrough(_):
+        def passthrough(_: 'interfaces.objects.ObjectInterface') -> bool:
             return False
 
         filter_func = passthrough
         if self.config.get('address', None) is not None:
 
-            def filter_function(x):
+            def filter_function(x: 'interfaces.objects.ObjectInterface') -> bool:
                 return x.get_start() not in [self.config['address']]
 
             filter_func = filter_function

volatility/framework/plugins/windows/verinfo.py

@@ -1,6 +1,6 @@
 import io
 import logging
-from typing import Generator, List, Tuple
+from typing import Generator, List, Tuple, Union
 
 import volatility.framework.interfaces.plugins as interfaces_plugins
 import volatility.plugins.windows.moddump as moddump

volatility/framework/symbols/windows/extensions/__init__.py

@@ -2,7 +2,7 @@ import collections.abc
 import datetime
 import functools
 import logging
-from typing import Iterable, Iterator, Optional, Union
+from typing import Iterable, Iterator, Optional, Union, Dict
 
 from volatility.framework import constants, exceptions, interfaces, objects, renderers, symbols
 from volatility.framework.layers import intel
@@ -103,6 +103,7 @@ class _POOL_HEADER(objects.Struct):
                         return None
                 except (TypeError, exceptions.InvalidAddressException):
                     return None
+        return None
 
 
 class _KSYSTEM_TIME(objects.Struct):
@@ -456,7 +457,7 @@ class _OBJECT_HEADER(objects.Struct):
 
         return True
 
-    def get_object_type(self, type_map: dict, cookie: int = None) -> str:
+    def get_object_type(self, type_map: Dict[int, str], cookie: int = None) -> Optional[str]:
         """Across all Windows versions, the _OBJECT_HEADER embeds details on the type of
         object (i.e. process, file) but the way its embedded differs between versions.
         This API abstracts away those details."""

volatility/schemas/__init__.py

@@ -13,7 +13,7 @@ cached_validation_filepath = os.path.join(constants.CACHE_PATH, "valid_isf.cache
 
 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()
+    validhashes = set()  # type: Set
     if os.path.exists(cached_validation_filepath):
         with open(cached_validation_filepath, "r") as f:
             validhashes.update(json.load(f))