Make many typing fixes, based on mypy-0.590.

volatility/framework/__init__.py

@@ -67,7 +67,7 @@ def hide_from_subclasses(cls: typing.Type) -> typing.Type:
     return cls
 
 
-T = typing.TypeVar('T', typing.Type, typing.Type)
+T = typing.TypeVar('T', bound = typing.Type)
 
 
 def class_subclasses(cls: T) -> typing.Generator[T, None, None]:

volatility/framework/automagic/__init__.py

@@ -32,7 +32,7 @@ linux_automagic = ['ConstructionMagic',
 
 
 def available(context: interfaces.context.ContextInterface) \
-        -> typing.List[typing.Type[interfaces.automagic.AutomagicInterface]]:
+        -> typing.List[interfaces.automagic.AutomagicInterface]:
     """Returns an ordered list of all subclasses of :class:`~volatility.framework.interfaces.automagic.AutomagicInterface`.
 
     The order is based on the priority attributes of the subclasses, in order to ensure the automagics are listed in

volatility/framework/automagic/construct_layers.py

@@ -26,7 +26,7 @@ class ConstructionMagic(interfaces.automagic.AutomagicInterface):
                  config_path: str,
                  requirement: interfaces.configuration.RequirementInterface,
                  progress_callback = None, optional = False) -> typing.List[str]:
-        result = []
+        result = []  # type: typing.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

volatility/framework/automagic/pdbscan.py

@@ -11,7 +11,7 @@ import struct
 import typing
 
 from volatility.framework import exceptions, layers, validity
-from volatility.framework.layers import scanners
+from volatility.framework.layers import scanners, intel
 from volatility.framework.symbols import intermed, native
 
 if __name__ == "__main__":
@@ -165,25 +165,24 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
             virtual_layer_name = context.config.get(sub_config_path, None)
             layer_name = context.config.get(interfaces.configuration.path_join(sub_config_path, "memory_layer"), None)
             if layer_name and virtual_layer_name:
-                page_size = context.memory[virtual_layer_name].page_size
-                results = {virtual_layer_name: scan(context,
-                                                    layer_name,
-                                                    page_size,
-                                                    progress_callback = progress_callback)}
+                memlayer = context.memory[virtual_layer_name]
+                if isinstance(memlayer, intel.Intel):
+                    page_size = memlayer.page_size
+                    results = {virtual_layer_name: scan(context,
+                                                        layer_name,
+                                                        page_size,
+                                                        progress_callback = progress_callback)}
         else:
             for subreq in requirement.requirements.values():
                 results.update(self.recurse_pdb_finder(context, sub_config_path, subreq))
         return results
 
-    def recurse_symbol_fulfiller(self,
-                                 context: interfaces.context.ContextInterface) \
-            -> None:
+    def recurse_symbol_fulfiller(self, context: interfaces.context.ContextInterface) -> None:
         """Fulfills the SymbolRequirements in `self._symbol_requirements` found by the `recurse_symbol_requirements`.
 
         This pass will construct any requirements that may need it in the context it was passed
 
         :param context: Context on which to operate
-        :type context: ~volatility.framework.interfaces.context.ContextInterface
         """
         join = interfaces.configuration.path_join
         for config_path, sub_config_path, requirement in self._symbol_requirements:
@@ -323,7 +322,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
     def __call__(self,
                  context: interfaces.context.ContextInterface,
                  config_path: str,
-                 requirement: interfaces.configuration.ConstructableRequirementInterface,
+                 requirement: interfaces.configuration.RequirementInterface,
                  progress_callback: validity.ProgressCallback = None) -> None:
         if requirement.unsatisfied(context, config_path):
             if "pdbscan" not in context.symbol_space:

volatility/framework/automagic/stacker.py

@@ -42,7 +42,7 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
     def __call__(self,
                  context: interfaces.context.ContextInterface,
                  config_path: str,
-                 requirement: interfaces.configuration.ConstructableRequirementInterface,
+                 requirement: interfaces.configuration.RequirementInterface,
                  progress_callback: validity.ProgressCallback = None) \
             -> typing.Optional[typing.List[str]]:
         """Runs the automagic over the configurable"""

volatility/framework/automagic/windows.py

@@ -264,6 +264,8 @@ class WintelHelper(interfaces.automagic.AutomagicInterface):
                 # Only bother getting the DTB if we don't already have one
                 if not context.config.get(interfaces.configuration.path_join(sub_config_path, "page_map_offset"), None):
                     physical_layer = requirement.requirements["memory_layer"].config_value(context, sub_config_path)
+                    if not isinstance(physical_layer, str):
+                        raise TypeError("Physical layer name is not a string: {}".format(sub_config_path))
                     hits = context.memory[physical_layer].scan(context, PageMapScanner(useful), progress_callback)
                     for test, dtb in hits:
                         context.config[interfaces.configuration.path_join(sub_config_path, "page_map_offset")] = dtb
@@ -282,7 +284,8 @@ class WintelStacker(interfaces.automagic.StackerLayerInterface):
     def stack(cls,
               context: interfaces.context.ContextInterface,
               layer_name: str,
-              progress_callback: validity.ProgressCallback = None) -> typing.Optional[str]:
+              progress_callback: validity.ProgressCallback = None) \
+            -> typing.Optional[interfaces.layers.DataLayerInterface]:
         """Attempts to determine and stack an intel layer on a physical layer where possible
 
         Where the DTB scan fails, it attempts a heuristic of checking for the DTB within a specific range.

volatility/framework/contexts/__init__.py

@@ -138,6 +138,7 @@ class Module(interfaces.context.Module):
         """
         symbol_type = symbol_name and not (type_name or offset)
         type_type = (type_name and offset) and not symbol_name
+        type_arg = None  # type: typing.Optional[typing.Union[str, interfaces.objects.Template]]
         if symbol_type and type_type or not (symbol_type or type_type):
             raise ValueError("One of symbol_name, or type_name & offset, must be specified to construct a module")
         if symbol_type is not None:

volatility/framework/interfaces/automagic.py

@@ -3,13 +3,13 @@
 Automagic objects attempt to automatically fill configuration values that a user has not filled.
 """
 import typing
-from abc import ABCMeta, abstractmethod
+from abc import ABCMeta
 
 import volatility.framework.configuration.requirements
 from volatility.framework import validity, interfaces
 from volatility.framework.interfaces import configuration as interfaces_configuration
 
-RequirementInterfaceType = typing.Type[interfaces.configuration.RequirementInterface]
+R = typing.TypeVar('R', bound = interfaces.configuration.RequirementInterface)
 
 
 class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metaclass = ABCMeta):
@@ -47,22 +47,24 @@ class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metacla
                 raise ValueError(
                     "Automagic requirements must be an InstanceRequirement, ChoiceRequirement or ListRequirement")
 
-    @abstractmethod
     def __call__(self,
                  context: interfaces.context.ContextInterface,
                  config_path: str,
-                 configurable: interfaces.configuration.ConfigurableInterface,
+                 requirement: interfaces.configuration.RequirementInterface,
                  progress_callback: validity.ProgressCallback = None) -> typing.List[str]:
         """Runs the automagic over the configurable"""
+        return []
+
+    # TODO: requirement_type can be made typing.Union[typing.Type[T], typing.Tuple[typing.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: typing.Union[RequirementInterfaceType,
-                                                         typing.Tuple[RequirementInterfaceType, ...]],
+                          requirement_type: typing.Type[R],
                           shortcut: bool = True) \
-            -> typing.List[typing.Tuple[str, str, interfaces_configuration.ConstructableRequirementInterface]]:
+            -> typing.List[typing.Tuple[str, str, R]]:
         """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
@@ -75,7 +77,7 @@ class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metacla
         :return: 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 = []
+        results = []  # type: typing.List[typing.Tuple[str, str, R]]
         recurse = not shortcut
         if isinstance(requirement_root, requirement_type):
             if recurse or requirement_root.unsatisfied(context, config_path):
@@ -98,7 +100,6 @@ class StackerLayerInterface(validity.ValidityRoutines, metaclass = ABCMeta):
     stack_order = 0
 
     @classmethod
-    @abstractmethod
     def stack(self,
               context: interfaces.context.ContextInterface,
               layer_name: str,

volatility/framework/interfaces/configuration.py

@@ -8,6 +8,7 @@ provide configurations values that fulfill those requirements.  Where the user d
 values, automagic modules may extend the configuration tree themselves.
 """
 
+import collections.abc
 import copy
 import json
 import logging
@@ -17,8 +18,6 @@ import sys
 import typing
 from abc import ABCMeta, abstractmethod
 
-import collections.abc
-
 from volatility.framework import constants, interfaces
 from volatility.framework import validity
 from volatility.framework.interfaces.context import ContextInterface
@@ -631,7 +630,7 @@ class TranslationLayerRequirement(ConstructableRequirementInterface, Configurabl
             return None
 
         obj = self._construct_class(context, config_path, args)
-        if obj is not None:
+        if obj is not None and isinstance(obj, interfaces.layers.DataLayerInterface):
             context.add_layer(obj)
             # This should already be done by the _construct_class method
             # context.config[config_path] = obj.name
@@ -687,7 +686,7 @@ class SymbolRequirement(ConstructableRequirementInterface, ConfigurableRequireme
                 args[req.name] = node_config.data[req.name]
 
         obj = self._construct_class(context, config_path, args)
-        if obj is not None:
+        if obj is not None and isinstance(obj, interfaces.symbols.SymbolTableInterface):
             context.symbol_space.append(obj)
         return None
 

volatility/framework/interfaces/layers.py

@@ -1,5 +1,6 @@
 """Defines layers for containing data.  One layer may combine other layers, map data based on the data itself,
  or map a procedure (such as decryption) across another layer of data."""
+import collections.abc
 import functools
 import logging
 import math
@@ -8,8 +9,6 @@ import traceback
 import typing
 from abc import ABCMeta, abstractmethod
 
-import collections.abc
-
 from volatility.framework import constants, exceptions, validity, interfaces
 from volatility.framework.interfaces import configuration, context
 
@@ -53,7 +52,7 @@ class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta):
     def __init__(self) -> None:
         self.chunk_size = 0x1000000  # Default to 16Mb chunks
         self.overlap = 0x1000  # A page of overlap by default
-        self._context = None
+        self._context = None  # type: typing.Optional[interfaces.context.ContextInterface]
         self._layer_name = None  # type: typing.Optional[str]
 
     @property
@@ -348,7 +347,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
                                                                                                  current_offset))
             elif offset < current_offset:
                 raise exceptions.LayerException("Mapping returned an overlapping element")
-            self._context.memory.write(layer, mapped_offset, length)
+            self._context.memory.write(layer, mapped_offset, value)
             current_offset += length
 
     # ## Scan implementation with knowledge of pages

volatility/framework/interfaces/plugins.py

@@ -22,7 +22,7 @@ if typing.TYPE_CHECKING:
 class FileInterface(validity.ValidityRoutines, metaclass = ABCMeta):
     """Class for storing Files in the plugin as a means to output a file or files when necessary"""
 
-    def __init__(self, filename: str, data: bytes = None):
+    def __init__(self, filename: str, data: bytes = None) -> None:
         self.preferred_filename = filename
         self.data = io.BytesIO(data)
 

volatility/framework/interfaces/renderers.py

@@ -2,12 +2,11 @@
 or in some other form.  This module defines both the output format (:class:`TreeGrid`) and the renderer interface
 which can interact with a TreeGrid to produce suitable output."""
 
+import collections
 import datetime
 import typing
 from abc import abstractmethod, ABCMeta
 
-import collections
-
 from volatility.framework import validity
 
 Column = collections.namedtuple('Column', ['index', 'name', 'type'])
@@ -83,7 +82,7 @@ class Disassembly(object):
     """A class to indicate that the bytes provided should be disassembled (based on the architecture)"""
     possible_architectures = ['intel', 'intel64', 'arm', 'arm64']
 
-    def __init__(self, data: bytes, offset: int = 0, architecture: str = 'intel64'):
+    def __init__(self, data: bytes, offset: int = 0, architecture: str = 'intel64') -> None:
         self.data = data
         self.architecture = None
         if architecture in self.possible_architectures:
@@ -96,7 +95,7 @@ class Disassembly(object):
 # We don't class these off a shared base, because the BaseTypes must only
 # contain the types that the validator will accept (which would not include the base)
 
-_Type = typing.TypeVar("_Type")
+_Type = typing.TypeVar("_Type", bound = typing.Type)
 ColumnsType = typing.List[typing.Tuple[str, typing.Type]]
 BaseTypes = typing.Union[typing.Type[int],
                          typing.Type[str],

volatility/framework/layers/vmware.py

@@ -111,9 +111,10 @@ class VmwareStacker(interfaces.automagic.StackerLayerInterface):
               progress_callback: validity.ProgressCallback = None) \
             -> typing.Optional[interfaces.layers.DataLayerInterface]:
         """Attempt to stack this based on the starting information"""
-        if not isinstance(context.memory[layer_name], physical.FileLayer):
+        memlayer = context.memory[layer_name]
+        if not isinstance(memlayer, physical.FileLayer):
             return None
-        location = context.memory[layer_name].location
+        location = memlayer.location
         if location.endswith(".vmem"):
             vmss = location[:-5] + ".vmss"
             vmsn = location[:-5] + ".vmsn"

volatility/framework/objects/utility.py

@@ -2,7 +2,6 @@ import datetime
 import typing
 
 from volatility.framework import interfaces, objects, renderers
-from volatility.framework.objects import templates
 
 
 def array_to_string(array: objects.Array,
@@ -31,13 +30,13 @@ def pointer_to_string(pointer: objects.Pointer,
 
 def array_of_pointers(array: interfaces.objects.ObjectInterface,
                       count: int,
-                      subtype: typing.Optional[typing.Union[str, templates.ObjectTemplate]] = None,
+                      subtype: typing.Optional[typing.Union[str, interfaces.objects.Template]] = None,
                       context: interfaces.context.ContextInterface = None) -> interfaces.objects.ObjectInterface:
     """Takes an object, and recasts it as an array of pointers to subtype"""
     if isinstance(subtype, str) and context is not None:
         subtype = context.symbol_space.get_type(subtype)
-    if not isinstance(subtype, templates.ObjectTemplate) or subtype is None:
-        raise TypeError("Subtype must be a valid object template (or string name of an object template)")
+    if not isinstance(subtype, interfaces.objects.Template) or subtype is None:
+        raise TypeError("Subtype must be a valid template (or string name of an object template)")
     subtype_pointer = objects.templates.ObjectTemplate(objects.Pointer, type_name = 'pointer', subtype = subtype)
     return array.cast("array", count = count, subtype = subtype_pointer)
 

volatility/framework/renderers/__init__.py

@@ -1,11 +1,10 @@
 """Renderers
 
 Renderers display the unified output format in some manner (be it text or file or graphical output"""
+import collections
 import datetime
 import typing
 
-import collections
-
 from volatility.framework import interfaces
 
 

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

@@ -23,6 +23,9 @@ class task_struct(generic.GenericIntelProcess):
         if not pgd:
             return None
 
+        if not isinstance(parent_layer, interfaces.layers.TranslationLayerInterface):
+            raise TypeError("Parent layer is not a translation layer, unable to construct process layer")
+
         dtb, layer_name = parent_layer.translate(pgd)
         if not dtb:
             return None

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

@@ -1,10 +1,10 @@
 import collections.abc
-import datetime
 import functools
 import logging
 import typing
 
 from volatility.framework import constants, exceptions, interfaces, objects, renderers
+from volatility.framework.layers import intel
 from volatility.framework.objects import utility
 from volatility.framework.symbols import generic
 
@@ -200,10 +200,12 @@ class _MMVAD_SHORT(objects.Struct):
             return self.u.VadFlags.PrivateMemory
 
         elif hasattr(self, "Core"):
-            if hasattr(self.Core, "u1") and hasattr(self.Core.u1, "VadFlags1") and hasattr(self.Core.u1.VadFlags1, "PrivateMemory"):
+            if hasattr(self.Core, "u1") and hasattr(self.Core.u1, "VadFlags1") and hasattr(self.Core.u1.VadFlags1,
+                                                                                           "PrivateMemory"):
                 return self.Core.u1.VadFlags1.PrivateMemory
 
-            elif hasattr(self.Core, "u") and hasattr(self.Core.u, "VadFlags") and hasattr(self.Core.u.VadFlags, "PrivateMemory"):
+            elif hasattr(self.Core, "u") and hasattr(self.Core.u, "VadFlags") and hasattr(self.Core.u.VadFlags,
+                                                                                          "PrivateMemory"):
                 return self.Core.u.VadFlags.PrivateMemory
 
         raise AttributeError("Unable to find the private memory member")
@@ -393,6 +395,11 @@ class _EPROCESS(generic.GenericIntelProcess):
         """Constructs a new layer based on the process's DirectoryTableBase"""
 
         parent_layer = context.memory[self.vol.layer_name]
+
+        if not isinstance(parent_layer, intel.Intel):
+            # We can't get bits_per_register unless we're an intel space (since that's not defined at the higher layer)
+            raise TypeError("Parent layer is not a translation layer, unable to construct process layer")
+
         # Presumably for 64-bit systems, the DTB is defined as an array, rather than an unsigned long long
         if isinstance(self.Pcb.DirectoryTableBase, objects.Array):
             dtb = self.Pcb.DirectoryTableBase.cast("unsigned long long")

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

@@ -117,6 +117,8 @@ class _CM_KEY_NODE(objects.Struct):
 
     def get_key_path(self) -> interfaces.objects.ObjectInterface:
         reg = self._context.memory[self.vol.layer_name]
+        if not isinstance(reg, RegistryHive):
+            raise TypeError("Key was not instantiated on a RegistryHive layer")
         # Using the offset adds a significant delay (since it cannot be cached easily)
         # if self.vol.offset == reg.get_node(reg.root_cell_offset).vol.offset:
         if self.vol.offset == reg.root_cell_offset + 4:
@@ -139,6 +141,9 @@ class _CM_KEY_VALUE(objects.Struct):
         data = b""
         # Check if the data is stored inline
         layer = self._context.memory[self.vol.layer_name]
+        if not isinstance(layer, RegistryHive):
+            raise TypeError("Key value was not instantiated on a RegistryHive layer")
+
         if self.DataLength & 0x80000000 and (0 > datalen or datalen > 4):
             raise ValueError("Unable to read inline registry value with excessive length: {}".format(datalen))
         elif self.DataLength & 0x80000000:
@@ -151,7 +156,7 @@ class _CM_KEY_VALUE(objects.Struct):
                 # The value 4 should actually be unsigned-int.size, but since it's a file format that shouldn't change
                 # the direct value 4 can be used instead
                 block_offset = layer.get_cell(big_data.List + (i * 4)).cast("unsigned int")
-                if block_offset < layer.maximum_address:
+                if isinstance(block_offset, int) and block_offset < layer.maximum_address:
                     amount = min(BIG_DATA_MAXLEN, datalen)
                     data += layer.read(offset = layer.get_cell(block_offset).vol.offset, length = amount)
                     datalen -= amount