Add type annotations to most of the automagic files.

volatility/framework/automagic/__init__.py

@@ -10,8 +10,9 @@ loading of file format types) as well as a module to reconstruct layers based on
 import logging
 import sys
 import traceback
+import typing
 
-from volatility.framework import class_subclasses, import_files, interfaces
+from volatility.framework import class_subclasses, import_files, interfaces, validity
 from volatility.framework.automagic import construct_layers, stacker, windows, pdbscan
 from volatility.framework.configuration import requirements
 
@@ -29,7 +30,8 @@ linux_automagic = ['ConstructionMagic',
                    'LinuxSymbolFinder']
 
 
-def available(context):
+def available(context: interfaces.context.ContextInterface) \
+        -> typing.List[typing.Type[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
@@ -45,7 +47,12 @@ def available(context):
                   key = lambda x: x.priority)
 
 
-def run(automagics, context, configurable, config_path, progress_callback = None):
+def run(automagics: typing.List[interfaces.automagic.AutomagicInterface],
+        context: interfaces.context.ContextInterface,
+        configurable: typing.Union[interfaces.configuration.ConfigurableInterface,
+                                   typing.Type[interfaces.configuration.ConfigurableInterface]],
+        config_path: str,
+        progress_callback: validity.ProgressCallback = None) -> typing.List[traceback.TracebackException]:
     """Runs through the list of `automagics` in order, allowing them to make changes to the context
 
     :param automagics: A list of :class:`~volatility.framework.interfaces.automagic.AutomagicInterface` objects
@@ -66,14 +73,16 @@ def run(automagics, context, configurable, config_path, progress_callback = None
             raise TypeError("Automagics must only contain AutomagicInterface subclasses")
 
     if (not isinstance(configurable, interfaces.configuration.ConfigurableInterface)
-        and not issubclass(configurable, interfaces.configuration.ConfigurableInterface)):
+            and not issubclass(configurable, interfaces.configuration.ConfigurableInterface)):
         raise TypeError("Automagic operates on configurables only")
 
     # TODO: Fix need for top level config element just because we're using a MultiRequirement to group the
     # configurable's config requirements
-    configurable_class = configurable
+    configurable_class: typing.Type[interfaces.configuration.ConfigurableInterface]
     if isinstance(configurable, interfaces.configuration.ConfigurableInterface):
         configurable_class = configurable.__class__
+    else:
+        configurable_class = configurable
     requirement = requirements.MultiRequirement(name = configurable_class.__name__)
     for req in configurable.get_requirements():
         requirement.add_requirement(req)

volatility/framework/automagic/construct_layers.py

@@ -2,6 +2,7 @@
 of a :class:`~volatility.framework.interfaces.configuration.ConfigurableInterface`."""
 
 import logging
+import typing
 
 from volatility.framework import constants
 from volatility.framework import interfaces
@@ -20,7 +21,11 @@ class ConstructionMagic(interfaces.automagic.AutomagicInterface):
     """
     priority = 0
 
-    def __call__(self, context, config_path, requirement, progress_callback = None, optional = False):
+    def __call__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 requirement: interfaces.configuration.RequirementInterface,
+                 progress_callback = None, optional = False) -> typing.List[str]:
         result = []
         if requirement.unsatisfied(context, config_path):
             # Having called validate at the top level tells us both that we need to dig deeper

volatility/framework/automagic/linux.py

@@ -1,6 +1,7 @@
 import logging
+import typing
 
-from volatility.framework import interfaces, constants
+from volatility.framework import interfaces, constants, validity
 from volatility.framework.automagic import linux_symbol_cache
 from volatility.framework.layers import intel, scanners
 from volatility.framework.symbols import linux
@@ -12,19 +13,26 @@ class LinuxSymbolFinder(interfaces.automagic.AutomagicInterface):
     """Linux symbol loader based on uname signature strings"""
     priority = 40
 
-    def __init__(self, context, config_path):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str) -> None:
         super().__init__(context, config_path)
-        self._requirements = None
-        self._linux_banners_ = None
+        self._requirements: typing.List[
+            typing.Tuple[str, str, interfaces.configuration.ConstructableRequirementInterface]] = []
+        self._linux_banners_: linux_symbol_cache.LinuxBanners = {}
 
     @property
-    def _linux_banners(self):
+    def _linux_banners(self) -> linux_symbol_cache.LinuxBanners:
         """Creates a cached copy of the results, but only it's been requested"""
         if self._linux_banners_ is None:
             self._linux_banners_ = linux_symbol_cache.LinuxSymbolCache.load_linux_banners()
         return self._linux_banners_
 
-    def __call__(self, context, config_path, requirement, progress_callback = None):
+    def __call__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 requirement: interfaces.configuration.RequirementInterface,
+                 progress_callback: validity.ProgressCallback = None) -> None:
         """Searches for LinuxSymbolRequirements and attempt to populate them"""
         self._requirements = self.find_requirements(context, config_path, requirement,
                                                     (interfaces.configuration.TranslationLayerRequirement,
@@ -37,13 +45,18 @@ class LinuxSymbolFinder(interfaces.automagic.AutomagicInterface):
                 for (tl_path, tl_sub_path, tl_requirement) in self._requirements:
                     # Find the TranslationLayer sibling to the SymbolRequirement
                     if (isinstance(tl_requirement, interfaces.configuration.TranslationLayerRequirement) and
-                                tl_path == path):
+                            tl_path == path):
                         if context.config.get(tl_sub_path, None):
                             self._banner_scan(context, path, requirement, context.config[tl_sub_path],
                                               progress_callback)
                             break
 
-    def _banner_scan(self, context, config_path, requirement, layer_name, progress_callback = None):
+    def _banner_scan(self,
+                     context: interfaces.context.ContextInterface,
+                     config_path: str,
+                     requirement: interfaces.configuration.ConstructableRequirementInterface,
+                     layer_name: str,
+                     progress_callback: validity.ProgressCallback = None) -> None:
         """Accepts a context, config_path and SymbolRequirement, with a constructed layer_name
         and scans the layer for linux banners"""
 
@@ -90,7 +103,11 @@ class LintelStacker(interfaces.automagic.StackerLayerInterface):
     stack_order = 12
 
     @classmethod
-    def stack(cls, context, layer_name, progress_callback = None):
+    def stack(cls,
+              context: interfaces.context.ContextInterface,
+              layer_name: str,
+              progress_callback: validity.ProgressCallback = None) \
+            -> typing.Union[None, typing.Type[interfaces.layers.DataLayerInterface]]:
         """Attempts to identify linux within this layer"""
         layer = context.memory[layer_name]
         join = interfaces.configuration.path_join
@@ -116,11 +133,11 @@ class LintelStacker(interfaces.automagic.StackerLayerInterface):
                 kaslr_shift, _ = LinuxUtilities.find_aslr(context, table_name, layer_name,
                                                           progress_callback = progress_callback)
 
+                layer_class: typing.Type = intel.Intel
                 if ('init_level4_pgt' in table.symbols):
                     layer_class = intel.Intel32e
                     dtb_symbol_name = 'init_level4_pgt'
                 else:
-                    layer_class = intel.Intel
                     dtb_symbol_name = 'swapper_pg_dir'
 
                 dtb = LinuxUtilities.virtual_to_physical_address(table.get_symbol(dtb_symbol_name).address +
@@ -144,7 +161,12 @@ class LinuxUtilities(object):
     """Class with multiple useful linux functions"""
 
     @classmethod
-    def find_aslr(cls, context, symbol_table, layer_name, progress_callback = None):
+    def find_aslr(cls,
+                  context: interfaces.context.ContextInterface,
+                  symbol_table: str,
+                  layer_name: str,
+                  progress_callback: validity.ProgressCallback = None) \
+            -> typing.Tuple[typing.Union[None, int], typing.Union[None, int]]:
         """Determines the offset of the actual DTB in physical space and its symbol offset"""
         init_task_symbol = symbol_table + constants.BANG + 'init_task'
         table_dtb = context.symbol_space.get_symbol(init_task_symbol).address
@@ -174,7 +196,7 @@ class LinuxUtilities(object):
         return None, None
 
     @classmethod
-    def virtual_to_physical_address(cls, addr):
+    def virtual_to_physical_address(cls, addr: int) -> int:
         """Converts a virtual linux address to a physical one (does not account of ASLR)"""
         if addr > 0xffffffff80000000:
             return addr - 0xffffffff80000000

volatility/framework/automagic/linux_symbol_cache.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import pickle
+import typing
 from urllib import parse
 
 from volatility.framework import constants, exceptions, interfaces
@@ -8,6 +9,8 @@ from volatility.framework.symbols import intermed
 
 vollog = logging.getLogger(__name__)
 
+LinuxBanners = typing.Dict[bytes, typing.List[str]]
+
 
 class LinuxSymbolCache(interfaces.automagic.AutomagicInterface):
     """Runs through all Linux symbols tables and caches their banners"""
@@ -17,8 +20,8 @@ class LinuxSymbolCache(interfaces.automagic.AutomagicInterface):
     priority = 0
 
     @classmethod
-    def load_linux_banners(cls):
-        linuxbanners = {}
+    def load_linux_banners(cls) -> LinuxBanners:
+        linuxbanners: LinuxBanners = {}
         if os.path.exists(constants.LINUX_BANNERS_PATH):
             with open(constants.LINUX_BANNERS_PATH, "rb") as f:
                 # We use pickle over JSON because we're dealing with bytes objects

volatility/framework/automagic/nlpdtbfinder.py

@@ -1,19 +1,22 @@
 import logging
 import math
 import struct
+import typing
 
-from volatility.framework import exceptions, interfaces
+from volatility.framework import exceptions, interfaces, validity
 from volatility.framework.configuration import requirements
 from volatility.framework.layers import intel
 
 vollog = logging.getLogger(__name__)
 
-validity_tests = {intel.Intel: [],
-                  intel.IntelPAE: [],
-                  intel.Intel32e: [(0b10111011, 0b00100011),
-                                   (0b1, 0b1),
-                                   (0b1111011, 0b1100011),
-                                   (0b1111011, 0b1100011)]}
+validity_tests: typing.Dict[typing.Type[intel.Intel],
+                            typing.List[typing.Tuple[int, int]]] = {intel.Intel: [],
+                                                                    intel.IntelPAE: [],
+                                                                    intel.Intel32e: [
+                                                                        (0b10111011, 0b00100011),
+                                                                        (0b1, 0b1),
+                                                                        (0b1111011, 0b1100011),
+                                                                        (0b1111011, 0b1100011)]}
 
 
 class NlpDtbScanner(interfaces.layers.ScannerInterface):
@@ -28,12 +31,14 @@ class NlpDtbScanner(interfaces.layers.ScannerInterface):
     # Overlap must be a multiple of the page size
     overlap = 0x4000
 
-    def __init__(self, layer_class, physical_layer):
+    def __init__(self,
+                 layer_class: typing.Type[intel.Intel],
+                 physical_layer: interfaces.layers.DataLayerInterface) -> None:
         super().__init__()
         self._layer_class = layer_class
         self._physical_layer = physical_layer
 
-    def test_entries(self, valid_entries):
+    def test_entries(self, valid_entries: typing.List[typing.Tuple[int, int]]) -> bool:
         """Scans through valid_entries, descending to see whether one can be successfully mapped to completion
 
         Returns the first valid DTB or None is no valid DTBs could be found
@@ -43,12 +48,12 @@ class NlpDtbScanner(interfaces.layers.ScannerInterface):
                 return True
         return False
 
-    def _get_mask(self, value):
+    def _get_mask(self, value: int) -> int:
         """Returns a value that correctly masks all numbers less than that of value"""
         bits = int(math.ceil(math.log(value, 2)))
         return (1 << (bits + 1)) - 1
 
-    def test_entry(self, entry, level = 0):
+    def test_entry(self, entry: int, level: int = 0) -> bool:
         """Tests an individual entry at a particular level in the structure"""
         name, size, large_page = self._layer_class.structure[level]
 
@@ -89,7 +94,8 @@ class NlpDtbScanner(interfaces.layers.ScannerInterface):
                 return True
         return False
 
-    def __call__(self, data, data_offset):
+    def __call__(self, data: bytes, data_offset: int) \
+            -> typing.Generator[typing.Tuple[int, typing.List[typing.Tuple[int, int]]], None, None]:
         structure = self._layer_class.structure
         name, size, large_page = structure[0]
 
@@ -105,7 +111,7 @@ class NlpDtbScanner(interfaces.layers.ScannerInterface):
             if len(data[page_offset:page_offset + calcsize]) < calcsize:
                 continue
             entries = struct.unpack('<' + str(2 ** size) + format_str, data[page_offset:page_offset + calcsize])
-            valid_entries = []
+            valid_entries: typing.List[typing.Tuple[int, int]] = []
             invalid_count = 0
             user_count = 0
             supervisor_count = 0
@@ -137,7 +143,12 @@ class NlpDtbfinder(interfaces.automagic.AutomagicInterface):
     """
     priority = 11
 
-    def __call__(self, context, config_path, requirement, progress_callback = None):
+    def __call__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 requirement: interfaces.configuration.RequirementInterface,
+                 progress_callback: validity.ProgressCallback = None) \
+            -> None:
         results = {}
         sub_config_path = interfaces.configuration.path_join(config_path, requirement.name)
         if (not interfaces.configuration.path_join(sub_config_path, "page_map_offset") in context.config and
@@ -175,7 +186,10 @@ class NlpDtbStacker(interfaces.automagic.StackerLayerInterface):
     stack_order = 13
 
     @classmethod
-    def stack(cls, context, layer_name, progress_callback = None):
+    def stack(cls,
+              context: interfaces.context.ContextInterface,
+              layer_name: str,
+              progress_callback: validity.ProgressCallback = None):
         """Attempts to determine and stack an intel layer on a physical layer where possible"""
         if isinstance(context.memory[layer_name], intel.Intel):
             return None

volatility/framework/automagic/pdbscan.py

@@ -8,8 +8,9 @@ import logging
 import math
 import os
 import struct
+import typing
 
-from volatility.framework import exceptions, layers
+from volatility.framework import exceptions, layers, validity
 from volatility.framework.layers import scanners
 from volatility.framework.symbols import native, intermed
 
@@ -38,11 +39,12 @@ class PdbSignatureScanner(interfaces.layers.ScannerInterface):
 
     _RSDS_format = struct.Struct("<16BI")
 
-    def __init__(self, pdb_names):
+    def __init__(self, pdb_names: typing.List[bytes]) -> None:
         super().__init__()
         self._pdb_names = pdb_names
 
-    def __call__(self, data, data_offset):
+    def __call__(self, data: bytes, data_offset: int) \
+            -> typing.Generator[typing.Tuple[str, typing.Any, bytes, int], None, None]:
         sig = data.find(b"RSDS")
         while sig >= 0:
             null = data.find(b'\0', sig + 4 + self._RSDS_format.size)
@@ -52,7 +54,7 @@ class PdbSignatureScanner(interfaces.layers.ScannerInterface):
                     pdb_name = data[name_offset:null]
                     if pdb_name in self._pdb_names:
 
-                        ## thie ordering is intentional due to mixed endianness in the GUID
+                        ## this ordering is intentional due to mixed endianness in the GUID
                         (g3, g2, g1, g0, g5, g4, g7, g6, g8, g9, ga, gb, gc, gd, ge, gf, a) = \
                             self._RSDS_format.unpack(data[sig + 4:name_offset])
 
@@ -61,7 +63,13 @@ class PdbSignatureScanner(interfaces.layers.ScannerInterface):
             sig = data.find(b"RSDS", sig + 1)
 
 
-def scan(ctx, layer_name, page_size, progress_callback = None, start = None, end = None):
+def scan(ctx: interfaces.context.ContextInterface,
+         layer_name: str,
+         page_size: int,
+         progress_callback: validity.ProgressCallback = None,
+         start: typing.Union[int, None] = None,
+         end: typing.Union[int, None] = None) \
+        -> typing.Generator[typing.Dict[str, typing.Union[bytes, str, int]], None, None]:
     """Scans through `layer_name` at `ctx` looking for RSDS headers that indicate one of four common pdb kernel names
        (as listed in `self.pdb_names`) and returns the tuple (GUID, age, pdb_name, signature_offset, mz_offset)
 
@@ -124,11 +132,18 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
     suffixes = ['.json', '.json.xz']
     """Provides a list of supported suffixes for Intermediate Format data files"""
 
-    def __init__(self, context, config_path):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str) -> None:
         super().__init__(context, config_path)
-        self.valid_kernels = []
+        self.valid_kernels: typing.Dict[str, typing.Tuple[int, typing.Dict]] = {}
 
-    def recurse_pdb_finder(self, context, config_path, requirement, progress_callback = None):
+    def recurse_pdb_finder(self,
+                           context: interfaces.context.ContextInterface,
+                           config_path: str,
+                           requirement: interfaces.configuration.RequirementInterface,
+                           progress_callback: validity.ProgressCallback = None) \
+            -> typing.Dict[bytes, typing.Dict]:
         """Traverses the requirement tree, rooted at `requirement` looking for virtual layers that might contain a windows PDB.
 
         Returns a list of possible kernel locations in the physical memory
@@ -142,7 +157,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
         :return: A list of (layer_name, scan_results)
         """
         sub_config_path = interfaces.configuration.path_join(config_path, requirement.name)
-        results = {}
+        results: typing.Dict[bytes, typing.Dict] = {}
         if isinstance(requirement, interfaces.configuration.TranslationLayerRequirement):
             # Check for symbols in this layer
             # FIXME: optionally allow a full (slow) scan
@@ -160,7 +175,9 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
                 results.update(self.recurse_pdb_finder(context, sub_config_path, subreq))
         return results
 
-    def recurse_symbol_fulfiller(self, context):
+    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
@@ -195,7 +212,8 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
                 else:
                     vollog.debug("No suitable kernel pdb signature found")
 
-    def set_kernel_virtual_offset(self, context):
+    def set_kernel_virtual_offset(self,
+                                  context: interfaces.context.ContextInterface) -> None:
         """Traverses the requirement tree, looking for kernel_virtual_offset values that may need setting and sets
         it based on the previously identified `valid_kernels`.
 
@@ -210,7 +228,11 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
             context.config[kvo_path] = kvo
             vollog.debug("Setting kernel_virtual_offset to {}".format(hex(kvo)))
 
-    def determine_valid_kernels(self, context, potential_kernels, progress_callback = None):
+    def determine_valid_kernels(self,
+                                context: interfaces.context.ContextInterface,
+                                potential_kernels: typing.Dict[str, typing.Any],
+                                progress_callback: validity.ProgressCallback = None) \
+            -> typing.Dict[str, typing.Tuple[int, typing.Any]]:
         """Runs through the identified potential kernels and verifies their suitability
 
         This carries out a scan using the pdb_signature scanner on a physical layer.  It uses the
@@ -267,7 +289,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
                     # TODO:  On older windows, this might be \WINDOWS\system32\nt rather than \SystemRoot\system32\nt
                     results = physical_layer.scan(context, scanners.BytesScanner(b"\\SystemRoot\\system32\\nt"),
                                                   progress_callback = progress_callback)
-                    seen = set()
+                    seen: typing.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

volatility/framework/automagic/stacker.py

@@ -8,9 +8,10 @@ once a layer successfully stacks on top of the existing layers, it is removed fr
 """
 
 import logging
+import typing
 
 import volatility
-from volatility.framework import configuration, interfaces, constants
+from volatility.framework import configuration, interfaces, constants, validity
 from volatility.framework.automagic import construct_layers
 from volatility.framework.configuration import requirements
 from volatility.framework.layers import physical
@@ -33,7 +34,12 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
     priority = 10
     page_map_offset = None
 
-    def __call__(self, context, config_path, requirement, progress_callback = None):
+    def __call__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 requirement: interfaces.configuration.ConstructableRequirementInterface,
+                 progress_callback: validity.ProgressCallback = None) \
+            -> None:
         """Runs the automagic over the configurable"""
 
         # Quick exit if we're not needed
@@ -103,7 +109,11 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
                                                                                                 "ConstructionMagic"))
             constructor(context, config_path, requirement)
 
-    def find_suitable_requirements(self, stacked_layers, requirement, context, config_path):
+    def find_suitable_requirements(self,
+                                   stacked_layers: typing.List,
+                                   requirement: interfaces.configuration.RequirementInterface,
+                                   context: interfaces.context.ContextInterface,
+                                   config_path: str) -> typing.Union[None, typing.Tuple[str, str]]:
         """Looks for translation layer requirements and attempts to apply the stacked layers to it.  If it succeeds
         it returns the configuration path and layer name where the stacked nodes were spliced into the tree.
 
@@ -127,9 +137,10 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
             result = self.find_suitable_requirements(stacked_layers, req, context, child_config_path)
             if result:
                 return result
+        return None
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         # This is not optional for the stacker to run, so optional must be marked as False
         return [requirements.StringRequirement("single_location",
                                                description = "Specifies a base location on which to stack",

volatility/framework/configuration/requirements.py

@@ -6,6 +6,7 @@ etc) as well as indicating what they expect to be in the context (such as partic
 """
 
 import logging
+import typing
 
 from volatility.framework.interfaces import configuration as interfaces_configuration
 

volatility/framework/interfaces/automagic.py

@@ -2,7 +2,7 @@
 
 Automagic objects attempt to automatically fill configuration values that a user has not filled.
 """
-
+import typing
 from abc import ABCMeta, abstractmethod
 
 from volatility.framework import validity
@@ -46,7 +46,8 @@ class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metacla
     def __call__(self, context, config_path, configurable, progress_callback = None):
         """Runs the automagic over the configurable"""
 
-    def find_requirements(self, context, config_path, requirement_root, requirement_type, shortcut = True):
+    def find_requirements(self, context, config_path, requirement_root, requirement_type, shortcut = True) \
+            -> typing.List[typing.Tuple[str, str, interfaces_configuration.ConstructableRequirementInterface]]:
         """Determines if there is actually an unfulfilled symbol requirement waiting
 
         This ensures we do not carry out an expensive search when there is no requirement for a particular symbol table.

volatility/framework/interfaces/configuration.py

@@ -15,6 +15,7 @@ import logging
 import random
 import string
 import sys
+import typing
 from abc import ABCMeta, abstractmethod
 
 from volatility.framework import constants
@@ -303,7 +304,7 @@ class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta):
 
 class InstanceRequirement(RequirementInterface):
     """Class to represent a single simple type (such as a boolean, a string, an integer or a series of bytes)"""
-    instance_type = bool
+    instance_type: typing.Type = bool
 
     def add_requirement(self, requirement):
         """Always raises a TypeError as instance requirements cannot have children"""

volatility/framework/interfaces/context.py

@@ -69,6 +69,9 @@ 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, layer_name, offset):
+        """Create a module object """
+
 
 class Module(validity.ValidityRoutines, metaclass = ABCMeta):
     """Maintains state concerning a particular loaded module in memory

volatility/framework/objects/__init__.py

@@ -1,5 +1,6 @@
 import logging
 import struct
+import typing
 from collections import abc
 
 from volatility.framework import interfaces
@@ -29,7 +30,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
+    _struct_type: typing.Type = int
 
     def __init__(self, context, type_name, object_info, struct_format):
         super().__init__(context = context,
@@ -493,7 +494,7 @@ class Struct(interfaces.objects.ObjectInterface):
             member = member(context = self._context,
                             object_info = interfaces.objects.ObjectInformation(layer_name = self.vol.layer_name,
                                                                                offset = mask & (
-                                                                                   self.vol.offset + relative_offset),
+                                                                                       self.vol.offset + relative_offset),
                                                                                member_name = attr,
                                                                                parent = self))
             self._concrete_members[attr] = member

volatility/framework/validity.py

@@ -1,5 +1,8 @@
 """A set of classes providing consistent type checking and error handling for type/class validity
 """
+import typing
+
+ProgressCallback = typing.Union[typing.Callable[[int, str], None], None]
 
 
 class ValidityRoutines(object):