Lots and lots of typing additions found by an initial monkeytype run.

volatility/framework/exceptions.py

@@ -22,7 +22,7 @@ class SymbolError(VolatilityException):
 class InvalidAddressException(VolatilityException):
     """Thrown when an address is not valid in the space it was requested"""
 
-    def __init__(self, layer_name, invalid_address, *args):
+    def __init__(self, layer_name: str, invalid_address: int, *args) -> None:
         super().__init__(layer_name, invalid_address, *args)
         self.invalid_address = invalid_address
         self.layer_name = layer_name

volatility/plugins/configwriter.py

@@ -1,7 +1,8 @@
 import json
 import logging
+import typing
 
-from volatility.framework import renderers
+from volatility.framework import renderers, interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.interfaces import plugins
 
@@ -12,7 +13,7 @@ class ConfigWriter(plugins.PluginInterface):
     """Runs the automagics and both prints and outputs configuration in the output directory"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),

volatility/plugins/linux/bash.py

@@ -4,8 +4,9 @@ typically found in Linux's /proc file system.
 
 import datetime
 import struct
+import typing
 
-from volatility.framework import constants, renderers, symbols
+from volatility.framework import constants, renderers, symbols, interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.interfaces import plugins
 from volatility.framework.layers import scanners
@@ -19,7 +20,7 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
     """Recovers bash command history from memory"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),

volatility/plugins/linux/check_afinfo.py

@@ -2,8 +2,9 @@
 typically found in Linux's /proc file system.
 """
 import logging
+import typing
 
-from volatility.framework import exceptions
+from volatility.framework import exceptions, interfaces
 from volatility.framework import renderers
 from volatility.framework.automagic import linux
 from volatility.framework.configuration import requirements
@@ -17,7 +18,7 @@ class Check_afinfo(plugins.PluginInterface):
     """Verifies the operation function pointers of network protocols"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),

volatility/plugins/linux/check_syscall.py

@@ -2,8 +2,9 @@
 typically found in Linux's /proc file system.
 """
 import logging
+import typing
 
-from volatility.framework import exceptions
+from volatility.framework import exceptions, interfaces
 from volatility.framework import renderers, constants
 from volatility.framework.automagic import linux
 from volatility.framework.configuration import requirements
@@ -24,7 +25,7 @@ class Check_syscall(plugins.PluginInterface):
     """Check system call table for hooks"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),

volatility/plugins/linux/elfs.py

@@ -1,8 +1,9 @@
 """A module containing a collection of plugins that produce data
 typically found in Linux's /proc file system.
 """
+import typing
 
-from volatility.framework import renderers
+from volatility.framework import renderers, interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.interfaces import plugins
 from volatility.framework.objects import utility
@@ -14,7 +15,7 @@ class Elfs(plugins.PluginInterface):
     """Lists all memory mapped ELF files for all processes"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),

volatility/plugins/linux/malfind.py

@@ -1,7 +1,9 @@
+import typing
+
 import volatility.framework.interfaces.plugins as interfaces_plugins
 import volatility.framework.interfaces.renderers as interfaces_renderers
 import volatility.plugins.linux.pslist as pslist
-from volatility.framework import constants
+from volatility.framework import constants, interfaces
 from volatility.framework import renderers
 from volatility.framework.configuration import requirements
 from volatility.framework.objects import utility
@@ -12,7 +14,7 @@ class Malfind(interfaces_plugins.PluginInterface):
     """Lists process memory ranges that potentially contain injected code"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),

volatility/plugins/linux/pslist.py

@@ -11,7 +11,7 @@ class PsList(interfaces_plugins.PluginInterface):
     """Lists the processes present in a particular linux memory image"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),

volatility/plugins/mac/psaux.py

@@ -1,6 +1,7 @@
 """In-memory artifacts from OSX systems"""
+import typing
 
-from volatility.framework import exceptions, renderers
+from volatility.framework import exceptions, renderers, interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.interfaces import plugins
 from volatility.framework.objects import utility
@@ -11,14 +12,15 @@ class Psaux(plugins.PluginInterface):
     """Recovers program command line arguments"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),
                 requirements.SymbolRequirement(name = "darwin",
                                                description = "Mac Kernel")]
 
-    def _generator(self, tasks):
+    def _generator(self, tasks: typing.Iterator[typing.Any]) -> \
+            typing.Iterator[typing.Tuple[int, typing.Tuple[int, str, int, str]]]:
         for task in tasks:
             proc_layer_name = task.add_process_layer()
             if proc_layer_name is None:
@@ -80,7 +82,7 @@ class Psaux(plugins.PluginInterface):
 
             yield (0, (task.p_pid, task_name, task.p_argc, args_str))
 
-    def run(self):
+    def run(self) -> renderers.TreeGrid:
         filter = pslist.PsList.create_filter([self.config.get('pid', None)])
 
         plugin = pslist.PsList.list_tasks

volatility/plugins/windows/cmdline.py

@@ -1,6 +1,8 @@
+import typing
+
 import volatility.framework.constants as constants
 import volatility.framework.interfaces.plugins as interfaces_plugins
-from volatility.framework import exceptions, renderers
+from volatility.framework import exceptions, renderers, interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.objects import utility
 from volatility.plugins.windows import pslist
@@ -10,7 +12,7 @@ class CmdLine(interfaces_plugins.PluginInterface):
     """Lists process command line arguments"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         # Since we're calling the plugin, make sure we have the plugin's requirements
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',

volatility/plugins/windows/dlldump.py

@@ -1,5 +1,6 @@
 import logging
 import ntpath
+import typing
 
 import volatility.framework.constants as constants
 import volatility.framework.interfaces.plugins as interfaces_plugins
@@ -18,7 +19,7 @@ class DllDump(interfaces_plugins.PluginInterface):
     """Dumps process memory ranges as DLLs"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         # Since we're calling the plugin, make sure we have the plugin's requirements
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',

volatility/plugins/windows/dlllist.py

@@ -1,5 +1,7 @@
+import typing
+
 import volatility.framework.interfaces.plugins as interfaces_plugins
-from volatility.framework import exceptions, renderers
+from volatility.framework import exceptions, renderers, interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.renderers import format_hints
 from volatility.plugins.windows import pslist
@@ -9,7 +11,7 @@ class DllList(interfaces_plugins.PluginInterface):
     """Lists the loaded modules in a particular windows memory image"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         # Since we're calling the plugin, make sure we have the plugin's requirements
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',

volatility/plugins/windows/info.py

@@ -1,7 +1,8 @@
 import time
+import typing
 
 import volatility.framework.interfaces.plugins as plugins
-from volatility.framework import constants
+from volatility.framework import constants, interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.renderers import TreeGrid
 from volatility.framework.symbols.windows.kdbg import KdbgIntermedSymbols
@@ -12,7 +13,7 @@ class Info(plugins.PluginInterface):
     """Show OS & kernel details of the memory sample being analyzed"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),

volatility/plugins/windows/moddump.py

@@ -19,7 +19,7 @@ class ModDump(interfaces_plugins.PluginInterface):
     """Dumps kernel modules"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         # Reuse the requirements from the plugins we use
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',

volatility/plugins/windows/procdump.py

@@ -1,10 +1,12 @@
 import logging
+import typing
 
 import volatility.framework.constants as constants
 import volatility.framework.exceptions as exceptions
 import volatility.framework.interfaces.plugins as interfaces_plugins
 import volatility.framework.renderers as renderers
 import volatility.plugins.windows.pslist as pslist
+from volatility.framework import interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.objects import utility
 from volatility.framework.symbols.windows.pe import PEIntermedSymbols
@@ -16,7 +18,7 @@ class ProcDump(interfaces_plugins.PluginInterface):
     """Dumps process executable images"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         # Since we're calling the plugin, make sure we have the plugin's requirements
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',

volatility/plugins/windows/pstree.py

@@ -1,3 +1,4 @@
+from volatility.framework.interfaces import objects
 from volatility.framework.renderers import format_hints
 from volatility.plugins.windows import pslist
 
@@ -5,13 +6,13 @@ from volatility.plugins.windows import pslist
 class PsTree(pslist.PsList):
     """Plugin for listing processes in a tree based on their parent process ID """
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
         self._processes = {}
         self._levels = {}
         self._children = {}
 
-    def find_level(self, pid):
+    def find_level(self, pid: objects.Pointer) -> None:
         """Finds how deep the pid is in the processes list"""
         seen = set([])
         seen.add(pid)
@@ -26,7 +27,7 @@ class PsTree(pslist.PsList):
         self._levels[pid] = level
 
     def _generator(self):
-        """Generates the """
+        """Generates the Tree of processes"""
         for proc in self.list_processes(self.context, self.config['primary'], self.config['nt_symbols']):
 
             if not self.config.get('physical', self.PHYSICAL_DEFAULT):

volatility/plugins/windows/registry/hivelist.py

@@ -1,5 +1,7 @@
+import typing
+
 import volatility.framework.interfaces.plugins as plugins
-from volatility.framework import renderers
+from volatility.framework import renderers, interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.renderers import format_hints
 
@@ -8,7 +10,7 @@ class HiveList(plugins.PluginInterface):
     """Lists the registry hives present in a particular memory image"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),
@@ -18,7 +20,7 @@ class HiveList(plugins.PluginInterface):
                                                optional = True,
                                                default = None)]
 
-    def _generator(self):
+    def _generator(self) -> typing.Iterator[typing.Tuple[int, typing.Tuple[int, str]]]:
         for hive in self.list_hives(context = self.context,
                                     layer_name = self.config["primary"],
                                     symbol_table = self.config["nt_symbols"],
@@ -28,7 +30,11 @@ class HiveList(plugins.PluginInterface):
                        hive.get_name() or ""))
 
     @classmethod
-    def list_hives(cls, context, layer_name, symbol_table, filter_string = None):
+    def list_hives(cls,
+                   context: interfaces.context.ContextInterface,
+                   layer_name: str,
+                   symbol_table: str,
+                   filter_string: None = None) -> typing.Iterator[interfaces.objects.ObjectInterface]:
         """Lists all the hives in the primary layer"""
 
         # We only use the object factory to demonstrate how to use one
@@ -44,7 +50,7 @@ class HiveList(plugins.PluginInterface):
             if filter_string is None or filter_string.lower() in str(hive.get_name() or "").lower():
                 yield hive
 
-    def run(self):
+    def run(self) -> renderers.TreeGrid:
         return renderers.TreeGrid([("Offset", format_hints.Hex),
                                    ("FileFullPath", str)],
                                   self._generator())

volatility/plugins/windows/registry/userassist.py

@@ -28,7 +28,7 @@ class UserAssist(interfaces.plugins.PluginInterface):
         self._folder_guids = json.load(open(os.path.join(os.path.dirname(__file__), "userassist.json"), "rb"))
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),

volatility/plugins/windows/ssdt.py

@@ -1,6 +1,7 @@
 import os
+import typing
 
-from volatility.framework import constants
+from volatility.framework import constants, interfaces
 from volatility.framework import contexts
 from volatility.framework import exceptions, symbols
 from volatility.framework import renderers
@@ -15,13 +16,14 @@ class SSDT(plugins.PluginInterface):
     """Lists the system call table"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),
                 requirements.SymbolRequirement(name = "nt_symbols", description = "Windows OS")]
 
-    def _generator(self, modules):
+    def _generator(self, modules: typing.Iterator[typing.Any]) -> \
+            typing.Iterator[typing.Tuple[int, typing.Tuple[int, int, str, str]]]:
 
         layer_name = self.config['primary']
         context_modules = []
@@ -98,7 +100,7 @@ class SSDT(plugins.PluginInterface):
                                module_name,
                                renderers.NotAvailableValue()))
 
-    def run(self):
+    def run(self) -> renderers.TreeGrid:
         return renderers.TreeGrid([("Index", int),
                                    ("Address", format_hints.Hex),
                                    ("Module", str),

volatility/plugins/windows/statistics.py

@@ -1,4 +1,6 @@
-from volatility.framework import renderers, exceptions
+import typing
+
+from volatility.framework import renderers, exceptions, interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.interfaces import plugins
 from volatility.framework.layers import intel
@@ -7,7 +9,7 @@ from volatility.framework.layers import intel
 class Statistics(plugins.PluginInterface):
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"])]

volatility/plugins/windows/strings.py

@@ -14,7 +14,7 @@ vollog = logging.getLogger(__name__)
 class Strings(interfaces.plugins.PluginInterface):
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',
                                                          architectures = ["Intel32", "Intel64"]),

volatility/plugins/windows/vaddump.py

@@ -1,9 +1,10 @@
 import logging
+import typing
 
 import volatility.framework.interfaces.plugins as interfaces_plugins
 import volatility.plugins.windows.pslist as pslist
 import volatility.plugins.windows.vadinfo as vadinfo
-from volatility.framework import renderers
+from volatility.framework import renderers, interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.objects import utility
 
@@ -14,7 +15,7 @@ class VadDump(interfaces_plugins.PluginInterface):
     """Dumps process memory ranges"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         # Since we're calling the plugin, make sure we have the plugin's requirements
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = 'Kernel Address Space',

volatility/plugins/windows/vadyarascan.py

@@ -19,7 +19,7 @@ except ImportError:
 class VadYaraScan(interfaces.plugins.PluginInterface):
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'primary',
                                                          description = "Primary kernel address space",
                                                          architectures = ["Intel32", "Intel64"]),

volatility/plugins/windows/verinfo.py

@@ -24,7 +24,7 @@ class VerInfo(interfaces_plugins.PluginInterface):
     """Lists version information from PE files"""
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         ## TODO: we might add a regex option on the name later, but otherwise we're good
         ## TODO: and we don't want any CLI options from pslist, modules, or moddump
         return [requirements.TranslationLayerRequirement(name = 'primary',

volatility/schemas/__init__.py

@@ -2,6 +2,7 @@ import hashlib
 import json
 import logging
 import os
+import typing
 
 from volatility.framework import constants
 
@@ -10,7 +11,7 @@ vollog = logging.getLogger(__name__)
 cached_validation_filepath = os.path.join(constants.CACHE_PATH, "valid_isf.cache")
 
 
-def load_cached_validations():
+def load_cached_validations() -> typing.Set[str]:
     """Loads up the list of successfully cached json objects, so we don't need to revalidate them"""
     validhashes = set()
     if os.path.exists(cached_validation_filepath):
@@ -28,7 +29,7 @@ def record_cached_validations(validations):
 cached_validations = load_cached_validations()
 
 
-def validate(input, use_cache = True):
+def validate(input: typing.Dict[str, typing.Any], use_cache: bool = True) -> bool:
     """Validates an input JSON file based upon """
     format = input.get('metadata', {}).get('format', None)
     if not format:
@@ -44,12 +45,12 @@ def validate(input, use_cache = True):
     return valid(input, schema, use_cache)
 
 
-def create_json_hash(input, schema):
+def create_json_hash(input: typing.Dict[str, typing.Any], schema: typing.Dict[str, typing.Any]) -> str:
     """Constructs the hash of the input and schema to create a unique indentifier for a particular JSON file"""
     return hashlib.sha1(bytes(json.dumps((input, schema), sort_keys = True), 'utf-8')).hexdigest()
 
 
-def valid(input, schema, use_cache = True):
+def valid(input: Dict[str, Any], schema: Dict[str, Any], use_cache: bool = True) -> bool:
     """Validates a json schema"""
     input_hash = create_json_hash(input, schema)
     if input_hash in cached_validations and use_cache: