Finish adding type-annotations thoughout the code.

volatility/framework/__init__.py

@@ -33,7 +33,7 @@ def interface_version():
 vollog = logging.getLogger(__name__)
 
 
-def require_interface_version(*args):
+def require_interface_version(*args) -> None:
     """Checks the required version of a plugin"""
     if len(args):
         if args[0] != interface_version()[0]:
@@ -66,12 +66,13 @@ def hide_from_subclasses(cls: typing.Type) -> typing.Type:
     return cls
 
 
-def class_subclasses(cls):
+def class_subclasses(cls: typing.Type) -> typing.Iterable[typing.Type]:
     """Returns all the (recursive) subclasses of a given class"""
     if not inspect.isclass(cls):
         raise TypeError("class_subclasses parameter not a valid class: {}".format(cls))
     for clazz in cls.__subclasses__():
-        if not hasattr(clazz, 'hidden') or not clazz.hidden:
+        # The typing system is not clever enough to realize that clazz has a hidden attr after the hasattr check
+        if not hasattr(clazz, 'hidden') or not clazz.hidden:  # type: ignore
             yield clazz
         for return_value in class_subclasses(clazz):
             yield return_value

volatility/framework/exceptions.py

@@ -33,7 +33,11 @@ class PagedInvalidAddressException(InvalidAddressException):
     Includes the invalid address and the number of bits of the address that are invalid
     """
 
-    def __init__(self, layer_name, invalid_address, invalid_bits, *args):
+    def __init__(self,
+                 layer_name: str,
+                 invalid_address: int,
+                 invalid_bits: int,
+                 *args) -> None:
         super().__init__(layer_name, invalid_address, *args)
         self.invalid_bits = invalid_bits
 
@@ -41,8 +45,8 @@ class PagedInvalidAddressException(InvalidAddressException):
 class InvalidDataException(VolatilityException):
     """Thrown when an object contains some data known to be invalid for that structure"""
 
-    def __init__(self, invalid_object, *args):
-        super.__init__(invalid_object, *args)
+    def __init__(self, invalid_object: object, *args) -> None:
+        super().__init__(invalid_object, *args)
         self._invalid_object = invalid_object
 
 

volatility/framework/interfaces/objects.py

@@ -19,7 +19,7 @@ class ReadOnlyMapping(validity.ValidityRoutines, collections.abc.Mapping):
     This ensures that the data stored in the mapping should not be modified, making an immutable mapping.
     """
 
-    def __init__(self, dictionary: typing.ChainMap[str, typing.Any]) -> None:
+    def __init__(self, dictionary: typing.Mapping[str, typing.Any]) -> None:
         self._dict = dictionary
 
     def __getattr__(self, attr: str) -> typing.Any:

volatility/framework/interfaces/symbols.py

@@ -174,14 +174,14 @@ class BaseSymbolTableInterface(validity.ValidityRoutines):
 
     # ## Functions for overriding classes
 
-    def set_type_class(self, name: str, clazz: objects.ObjectInterface) -> None:
+    def set_type_class(self, name: str, clazz: typing.Type[objects.ObjectInterface]) -> None:
         """Overrides the object class for a specific Symbol type
 
            Name *must* be present in self.types
         """
         raise NotImplementedError("Abstract method set_type_class not implemented yet.")
 
-    def get_type_class(self, name: str) -> objects.ObjectInterface:
+    def get_type_class(self, name: str) -> typing.Type[objects.ObjectInterface]:
         """Returns the class associated with a Symbol type"""
         raise NotImplementedError("Abstract method get_type_class not implemented yet.")
 

volatility/framework/symbols/generic/__init__.py

@@ -1,11 +1,16 @@
 import random
 import string
+import typing
 
 from volatility.framework import objects, interfaces
 
 
 class GenericIntelProcess(objects.Struct):
-    def _add_process_layer(self, context, dtb, config_prefix = None, preferred_name = None):
+    def _add_process_layer(self,
+                           context: interfaces.context.ContextInterface,
+                           dtb: typing.Union[int, interfaces.objects.ObjectInterface],
+                           config_prefix: str = None,
+                           preferred_name: str = None) -> str:
         """Constructs a new layer based on the process's DirectoryTableBase"""
 
         if config_prefix is None:

volatility/framework/symbols/intermed.py

@@ -5,7 +5,9 @@ import json
 import logging
 import os
 import pathlib
+import typing
 import zipfile
+from abc import ABCMeta
 
 from volatility import schemas
 from volatility.framework import class_subclasses, constants, exceptions, interfaces, objects, layers
@@ -38,7 +40,7 @@ vollog = logging.getLogger(__name__)
 # for container types
 #
 
-def _construct_delegate_function(name, is_property = False):
+def _construct_delegate_function(name: str, is_property: bool = False) -> typing.Any:
     def _delegate_function(self, *args, **kwargs):
         if is_property:
             return getattr(self._delegate, name)
@@ -50,30 +52,30 @@ def _construct_delegate_function(name, is_property = False):
 
 
 class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
-    def __init__(self, context, config_path, name, isf_url, native_types = None, validate = True):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 name: str,
+                 isf_url: str,
+                 native_types: interfaces.symbols.NativeTableInterface = None,
+                 validate: bool = True) -> None:
         """Instantiates an SymbolTable based on an IntermediateSymbolFormat JSON file.  This is validated against the
         appropriate schema.  The validation can be disabled by passing validate = False, but this should almost never be
         done.
 
         :param context:
-        :type context:
         :param config_path:
-        :type config_path:
         :param name:
-        :type name:
         :param isf_url:
-        :type isf_url:
         :param native_types:
-        :type native_types:
         :param validate: Determines whether the ISF file will be validated against the appropriate schema
-        :type validate: bool
         """
         # Check there are no obvious errors
         # Open the file and test the version
         self._versions = dict([(x.version, x) for x in class_subclasses(ISFormatTable)])
         fp = layers.ResourceAccessor().open(isf_url)
         reader = codecs.getreader("utf-8")
-        json_object = json.load(reader(fp))
+        json_object = json.load(reader(fp))  # type: ignore
         fp.close()
 
         # Validation is expensive, but we cache to store the hashes of successfully validated json objects
@@ -92,7 +94,10 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
         # Inherit
         super().__init__(context, config_path, name, native_types or self._delegate.natives)
 
-    def _closest_version(self, version, versions):
+    def _closest_version(self,
+                         version: str,
+                         versions: typing.Dict[typing.Tuple[int, int, int], typing.Type['ISFormatTable']]) \
+            -> typing.Type['ISFormatTable']:
         """Determines the highest suitable handler for specified version format
 
         An interface version such as (Current-Age).Age.Revision means that (Current - Age) of the provider must be equal to that of the
@@ -117,7 +122,9 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
     del_type_class = _construct_delegate_function('del_type_class')
 
     @classmethod
-    def file_symbol_url(cls, sub_path, filename = None):
+    def file_symbol_url(cls,
+                        sub_path: str,
+                        filename: typing.Optional[str] = None) -> typing.Generator[str, None, None]:
         """Returns an iterator of appropriate file-scheme symbol URLs that can be opened by a ResourceAccessor class
 
         Filter reduces the number of results returned to only those URLs containing that string
@@ -150,28 +157,25 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
                                 yield "jar:file:" + str(pathlib.Path(zip_path)) + "!" + name
 
 
-class ISFormatTable(interfaces.symbols.SymbolTableInterface):
+class ISFormatTable(interfaces.symbols.SymbolTableInterface, metaclass = ABCMeta):
     """Provide a base class to identify all subclasses"""
-    pass
+    version = (0, 0, 0)
 
-
-class Version1Format(ISFormatTable):
-    """Class for storing intermediate debugging data as objects and classes"""
-    current = 1
-    revision = 0
-    age = 1
-    version = (current - age, age, revision)
-
-    def __init__(self, context, config_path, name, json_object, native_types = None):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 name: str,
+                 json_object: typing.Any,
+                 native_types: interfaces.symbols.NativeTableInterface = None) -> None:
         self._json_object = json_object
         self._validate_json()
         nt = native_types or self._get_natives()
         nt.name = name + "_natives"
         super().__init__(context, config_path, name, nt)
-        self._overrides = {}
-        self._symbol_cache = {}
+        self._overrides = {}  # type: typing.Dict[str, typing.Type[interfaces.objects.ObjectInterface]]
+        self._symbol_cache = {}  # type: typing.Dict[str, interfaces.symbols.Symbol]
 
-    def _get_natives(self):
+    def _get_natives(self) -> typing.Optional[interfaces.symbols.NativeTableInterface]:
         """Determines the appropriate native_types to use from the JSON data"""
         # TODO: Consider how to generate the natives entirely from the ISF
         classes = {"x64": native.x64NativeTable, "x86": native.x86NativeTable}
@@ -187,10 +191,11 @@ class Version1Format(ISFormatTable):
             else:
                 vollog.debug("Choosing appropriate natives for symbol library: {}".format(nc))
                 return native_class.natives
+        return None
 
     # TODO: Check the format and make use of the other metadata
 
-    def _validate_json(self):
+    def _validate_json(self) -> None:
         if (not 'user_types' in self._json_object or
                 not 'base_types' in self._json_object or
                 not 'metadata' in self._json_object or
@@ -198,7 +203,15 @@ class Version1Format(ISFormatTable):
                 not 'enums' in self._json_object):
             raise exceptions.SymbolSpaceError("Malformed JSON file provided")
 
-    def get_symbol(self, name):
+
+class Version1Format(ISFormatTable):
+    """Class for storing intermediate debugging data as objects and classes"""
+    current = 1
+    revision = 0
+    age = 1
+    version = (current - age, age, revision)
+
+    def get_symbol(self, name: str) -> interfaces.symbols.Symbol:
         """Returns the location offset given by the symbol name"""
         # TODO: Add the ability to add/remove/change symbols after creation
         # note that this should invalidate/update the cache
@@ -211,33 +224,33 @@ class Version1Format(ISFormatTable):
         return self._symbol_cache[name]
 
     @property
-    def symbols(self):
+    def symbols(self) -> typing.Iterable[str]:
         """Returns an iterator of the symbol names"""
         return self._json_object.get('symbols', {}).keys()
 
     @property
-    def enumerations(self):
+    def enumerations(self) -> typing.Iterable[str]:
         """Returns an iterator of the available enumerations"""
         return self._json_object.get('enums', {}).keys()
 
     @property
-    def types(self):
+    def types(self) -> typing.Iterable[str]:
         """Returns an iterator of the symbol type names"""
         return list(self._json_object.get('user_types', {}).keys()) + list(self.natives.types)
 
-    def get_type_class(self, name):
+    def get_type_class(self, name: str) -> typing.Type[interfaces.objects.ObjectInterface]:
         return self._overrides.get(name, objects.Struct)
 
-    def set_type_class(self, name, clazz):
+    def set_type_class(self, name: str, clazz: typing.Type[interfaces.objects.ObjectInterface]) -> None:
         if name not in self.types:
             raise ValueError("Symbol type not in {} SymbolTable: {}".format(self.name, name))
         self._overrides[name] = clazz
 
-    def del_type_class(self, name):
+    def del_type_class(self, name: str) -> None:
         if name in self._overrides:
             del self._overrides[name]
 
-    def _interdict_to_template(self, dictionary):
+    def _interdict_to_template(self, dictionary: typing.Dict[str, typing.Any]) -> interfaces.objects.Template:
         """Converts an intermediate format dict into an object template"""
         if not dictionary:
             raise exceptions.SymbolSpaceError("Invalid intermediate dictionary: {}".format(dictionary))
@@ -280,7 +293,7 @@ class Version1Format(ISFormatTable):
 
         return objects.templates.ReferenceTemplate(type_name = reference_name)
 
-    def _lookup_enum(self, name):
+    def _lookup_enum(self, name: str) -> typing.Dict[str, typing.Any]:
         """Looks up an enumeration and returns a dictionary of __init__ parameters for an Enum"""
         lookup = self._json_object['enums'].get(name, None)
         if not lookup:
@@ -289,7 +302,7 @@ class Version1Format(ISFormatTable):
                   "base_type": self.natives.get_type(lookup['base'])}
         return result
 
-    def get_enumeration(self, enum_name):
+    def get_enumeration(self, enum_name: str) -> interfaces.objects.Template:
         """Resolves an individual enumeration"""
         if constants.BANG in enum_name:
             raise exceptions.SymbolError("Enumeration for a different table requested: {}".format(enum_name))
@@ -304,7 +317,7 @@ class Version1Format(ISFormatTable):
                                                 size = curdict['size'],
                                                 choices = curdict['constants'])
 
-    def get_type(self, type_name):
+    def get_type(self, type_name: str) -> interfaces.objects.Template:
         """Resolves an individual symbol"""
         if constants.BANG in type_name:
             raise exceptions.SymbolError("Symbol for a different table requested: {}".format(type_name))
@@ -331,7 +344,7 @@ class Version2Format(Version1Format):
     age = 0
     version = (current - age, age, revision)
 
-    def _get_natives(self):
+    def _get_natives(self) -> typing.Optional[interfaces.symbols.NativeTableInterface]:
         """Determines the appropriate native_types to use from the JSON data"""
         classes = {"x64": native.x64NativeTable, "x86": native.x86NativeTable}
         for nc in sorted(classes):
@@ -346,8 +359,9 @@ class Version2Format(Version1Format):
             else:
                 vollog.debug("Choosing appropriate natives for symbol library: {}".format(nc))
                 return native_class.natives
+        return None
 
-    def get_type(self, type_name):
+    def get_type(self, type_name: str) -> interfaces.objects.Template:
         """Resolves an individual symbol"""
         if constants.BANG in type_name:
             raise exceptions.SymbolError("Symbol for a different table requested: {}".format(type_name))
@@ -374,7 +388,7 @@ class Version3Format(Version2Format):
     age = 1
     version = (current - age, age, revision)
 
-    def get_symbol(self, name):
+    def get_symbol(self, name: str) -> interfaces.symbols.Symbol:
         """Returns the symbol given by the symbol name"""
         if self._symbol_cache.get(name, None):
             return self._symbol_cache[name]
@@ -407,7 +421,7 @@ class Version4Format(Version3Format):
                           'bool': ({1: '?'}, objects.Integer),
                           'char': ({1: 'c'}, objects.Char)}
 
-    def _get_natives(self):
+    def _get_natives(self) -> typing.Optional[interfaces.symbols.NativeTableInterface]:
         """Determines the appropriate native_types to use from the JSON data"""
         native_dict = {}
         base_types = self._json_object['base_types']
@@ -415,7 +429,8 @@ class Version4Format(Version3Format):
             # Void are ignored because voids are not a volatility primitive, they are a specific Volatility object
             if base_type != 'void':
                 current = base_types[base_type]
-                size_map, object_type = self.format_str_mapping.get(current['kind'], ({}, None))
+                # TODO: Fix up the typing of this, it bugs out because of the tuple assignment
+                size_map, object_type = self.format_str_mapping.get(current['kind'], ({}, None))  # type: ignore
                 format_str = size_map.get(current['size'], None)
                 if format_str is None or object_type is None:
                     raise ValueError("Unsupported kind/size combination in base_type {}".format(base_type))
@@ -434,7 +449,7 @@ class Version5Format(Version4Format):
     age = 1
     version = (current - age, age, revision)
 
-    def get_symbol(self, name):
+    def get_symbol(self, name: str) -> interfaces.symbols.Symbol:
         """Returns the symbol given by the symbol name"""
         if self._symbol_cache.get(name, None):
             return self._symbol_cache[name]

volatility/framework/symbols/linux/__init__.py

@@ -1,3 +1,6 @@
+import typing
+
+from volatility.framework import interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.symbols import intermed
 from volatility.framework.symbols.linux import extensions
@@ -6,7 +9,11 @@ from volatility.framework.symbols.linux import extensions
 class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
     provides = {"type": "interface"}
 
-    def __init__(self, context, config_path, name, isf_url):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 name: str,
+                 isf_url: str) -> None:
         super().__init__(context = context, config_path = config_path, name = name, isf_url = isf_url)
 
         # Set-up Linux specific types
@@ -18,6 +25,6 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
         self.set_type_class('vm_area_struct', extensions.vm_area_struct)
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.StringRequirement("isf_url",
                                                description = "JSON file containing the symbols encoded in the Intermediate Symbol Format")]

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

@@ -1,16 +1,19 @@
 import collections.abc
+import typing
 
-from volatility.framework import objects
 from volatility.framework import constants
-from volatility.framework.symbols import generic
+from volatility.framework import objects, interfaces
 from volatility.framework.objects import utility
+from volatility.framework.symbols import generic
 
 
 # Keep these in a basic module, to prevent import cycles when symbol providers require them
 
 
 class task_struct(generic.GenericIntelProcess):
-    def add_process_layer(self, config_prefix = None, preferred_name = None):
+    def add_process_layer(self,
+                          config_prefix: str = None,
+                          preferred_name: str = None) -> typing.Optional[str]:
         """Constructs a new layer based on the process's DTB.
         Returns the name of the Layer or None.
         """
@@ -30,7 +33,7 @@ class task_struct(generic.GenericIntelProcess):
 
 class mm_struct(objects.Struct):
     @property
-    def mmap_iter(self):
+    def mmap_iter(self) -> typing.Iterable[interfaces.objects.ObjectInterface]:
         """Returns an iterator for the mmap list member of an mm_struct."""
 
         if not self.mmap:
@@ -52,22 +55,22 @@ class super_block(objects.Struct):
     MINORBITS = 20
 
     @property
-    def major(self):
+    def major(self) -> int:
         return self.s_dev >> self.MINORBITS
 
     @property
-    def minor(self):
+    def minor(self) -> int:
         return self.s_dev & ((1 << self.MINORBITS) - 1)
 
 
 class vm_area_struct(objects.Struct):
     # include/linux/mm.h
-    VM_READ  = 0x00000001
+    VM_READ = 0x00000001
     VM_WRITE = 0x00000002
-    VM_EXEC  = 0x00000004
+    VM_EXEC = 0x00000004
 
     @property
-    def flags(self):
+    def flags(self) -> str:
         """Returns an rwx string representation of the flags in a vm_area_struct."""
 
         retval = ""
@@ -80,19 +83,20 @@ class vm_area_struct(objects.Struct):
 
         return retval
 
-    def page_offset(self):
+    def page_offset(self) -> int:
         if self.vm_file == 0:
             return 0
 
         return self.vm_pgoff << constants.linux.PAGE_SHIFT
 
+
 class struct_file(objects.Struct):
     @property
-    def full_path(self):
-        parts = []
+    def full_path(self) -> str:
+        parts = []  # type: typing.List[str]
         path = self.f_path
         path_dentry = path.dentry
-        seen = set()
+        seen = set()  # type: typing.Set[int]
         while path_dentry != 0 and path_dentry.vol.offset not in seen:
             name = utility.pointer_to_string(path_dentry.d_name.name, path_dentry.d_name.len)
             if name == "/":
@@ -105,7 +109,12 @@ class struct_file(objects.Struct):
 
 
 class list_head(objects.Struct, collections.abc.Iterable):
-    def to_list(self, symbol_type, member, forward = True, sentinel = True, layer = None):
+    def to_list(self,
+                symbol_type: str,
+                member: str,
+                forward: bool = True,
+                sentinel: bool = True,
+                layer: typing.Optional[str] = None) -> typing.Iterator[interfaces.objects.ObjectInterface]:
         """Returns an iterator of the entries in the list."""
 
         if layer is None:
@@ -130,5 +139,5 @@ class list_head(objects.Struct, collections.abc.Iterable):
             seen.add(link.vol.offset)
             link = getattr(link, direction).dereference()
 
-    def __iter__(self):
+    def __iter__(self) -> typing.Iterator[interfaces.objects.ObjectInterface]:
         return self.to_list(self.vol.parent.vol.type_name, self.vol.member_name)

volatility/framework/symbols/native.py

@@ -1,4 +1,5 @@
 import copy
+import typing
 
 from volatility.framework import constants, interfaces, objects
 
@@ -6,10 +7,13 @@ from volatility.framework import constants, interfaces, objects
 class NativeTable(interfaces.symbols.NativeTableInterface):
     """Symbol List that handles Native types"""
 
-    def __init__(self, name, native_dictionary):
+    # FIXME: typing the native_dictionary as typing.Tuple[interfaces.objects.ObjectInterface, str] throws many errors
+    def __init__(self,
+                 name: str,
+                 native_dictionary: typing.Dict[str, typing.Any]) -> None:
         super().__init__(name, self)
         self._native_dictionary = copy.deepcopy(native_dictionary)
-        self._overrides = {}
+        self._overrides = {}  # type: typing.Dict[str, interfaces.objects.ObjectInterface]
         for native_type in self._native_dictionary:
             native_class, _native_struct = self._native_dictionary[native_type]
             self._overrides[native_type] = native_class
@@ -17,16 +21,16 @@ class NativeTable(interfaces.symbols.NativeTableInterface):
         self._types = set(self._native_dictionary).union(
             {'enum', 'array', 'bitfield', 'void', 'pointer', 'string', 'bytes', 'function'})
 
-    def get_type_class(self, name):
-        ntype, fmt = self._native_dictionary.get(name, (objects.Integer, ''))
+    def get_type_class(self, name: str) -> typing.Type[interfaces.objects.ObjectInterface]:
+        ntype, _ = self._native_dictionary.get(name, (objects.Integer, None))
         return ntype
 
     @property
-    def types(self):
+    def types(self) -> typing.Iterable[str]:
         """Returns an iterator of the symbol type names"""
         return self._types
 
-    def get_type(self, type_name):
+    def get_type(self, type_name: str) -> interfaces.objects.Template:
         """Resolves a symbol name into an object template
 
            symbol_space is used to resolve any subtype symbols if they don't exist in this list
@@ -40,8 +44,8 @@ class NativeTable(interfaces.symbols.NativeTableInterface):
             table_name, type_name = name_split
             prefix = table_name + constants.BANG
 
-        additional = {}
-        obj = None
+        additional = {}  # type: typing.Dict[str, typing.Any]
+        obj = None  # type: typing.Type[interfaces.objects.ObjectInterface]
         if type_name == 'void' or type_name == 'function':
             obj = objects.Void
         elif type_name == 'array':

volatility/framework/symbols/windows/__init__.py

@@ -1,3 +1,6 @@
+import typing
+
+from volatility.framework import interfaces
 from volatility.framework.configuration import requirements
 from volatility.framework.symbols import intermed
 from volatility.framework.symbols.windows import extensions
@@ -7,7 +10,11 @@ from volatility.framework.symbols.windows.extensions import registry
 class WindowsKernelIntermedSymbols(intermed.IntermediateSymbolTable):
     provides = {"type": "interface"}
 
-    def __init__(self, context, config_path, name, isf_url):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 name: str,
+                 isf_url: str) -> None:
         super().__init__(context = context, config_path = config_path, name = name, isf_url = isf_url)
 
         # Set-up windows specific types
@@ -26,6 +33,6 @@ class WindowsKernelIntermedSymbols(intermed.IntermediateSymbolTable):
         self.set_type_class('_HMAP_ENTRY', registry._HMAP_ENTRY)
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.StringRequirement("isf_url",
                                                description = "JSON file containing the symbols encoded in the Intermediate Symbol Format")]

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

@@ -1,6 +1,7 @@
 import collections.abc
+import typing
 
-from volatility.framework import constants, objects
+from volatility.framework import constants, objects, interfaces
 from volatility.framework.symbols import generic
 from volatility.framework import exceptions
 
@@ -108,14 +109,14 @@ class _OBJECT_HEADER(objects.Struct):
         
 
 class _ETHREAD(objects.Struct):
-    def owning_process(self, kernel_layer = None):
+    def owning_process(self, kernel_layer: str = None) -> interfaces.objects.ObjectInterface:
         """Return the EPROCESS that owns this thread"""
         return self.ThreadsProcess.dereference(kernel_layer)
 
 
 class _UNICODE_STRING(objects.Struct):
     @property
-    def helper_string(self):
+    def helper_string(self) -> interfaces.objects.ObjectInterface:
         # We explicitly do *not* catch errors here, we allow an exception to be thrown
         # (otherwise there's no way to determine anything went wrong)
         # It's up to the user of this method to catch exceptions
@@ -126,7 +127,10 @@ class _UNICODE_STRING(objects.Struct):
 
 
 class _EPROCESS(generic.GenericIntelProcess):
-    def add_process_layer(self, context, config_prefix = None, preferred_name = None):
+    def add_process_layer(self,
+                          context: interfaces.context.ContextInterface,
+                          config_prefix: str = None,
+                          preferred_name: str = None):
         """Constructs a new layer based on the process's DirectoryTableBase"""
 
         parent_layer = context.memory[self.vol.layer_name]
@@ -140,7 +144,7 @@ class _EPROCESS(generic.GenericIntelProcess):
         # Add the constructed layer and return the name
         return self._add_process_layer(context, dtb, config_prefix, preferred_name)
 
-    def load_order_modules(self):
+    def load_order_modules(self) -> typing.Iterable[int]:
         """Generator for DLLs in the order that they were loaded"""
 
         if constants.BANG not in self.vol.type_name:
@@ -162,7 +166,12 @@ class _EPROCESS(generic.GenericIntelProcess):
 
 
 class _LIST_ENTRY(objects.Struct, collections.abc.Iterable):
-    def to_list(self, symbol_type, member, forward = True, sentinel = True, layer = None):
+    def to_list(self,
+                symbol_type: str,
+                member: str,
+                forward: bool = True,
+                sentinel: bool = True,
+                layer: typing.Optional[str] = None) -> typing.Iterator[interfaces.objects.ObjectInterface]:
         """Returns an iterator of the entries in the list"""
 
         if layer is None:
@@ -187,5 +196,5 @@ class _LIST_ENTRY(objects.Struct, collections.abc.Iterable):
             seen.add(link.vol.offset)
             link = getattr(link, direction).dereference()
 
-    def __iter__(self):
+    def __iter__(self) -> typing.Iterator[interfaces.objects.ObjectInterface]:
         return self.to_list(self.vol.parent.vol.type_name, self.vol.member_name)

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

@@ -1,8 +1,9 @@
 import enum
 import logging
 import struct
+import typing
 
-from volatility.framework import constants, exceptions, objects
+from volatility.framework import constants, exceptions, objects, interfaces
 from volatility.framework.layers.registry import RegistryHive
 
 vollog = logging.getLogger(__name__)
@@ -27,7 +28,7 @@ class RegValueTypes(enum.Enum):
 
 class _HMAP_ENTRY(objects.Struct):
     @property
-    def helper_block_offset(self):
+    def helper_block_offset(self) -> int:
         try:
             return self.PermanentBinAddress ^ (self.PermanentBinAddress & 0x3)
         except AttributeError:
@@ -36,7 +37,7 @@ class _HMAP_ENTRY(objects.Struct):
 
 class _CMHIVE(objects.Struct):
     @property
-    def helper_name(self):
+    def helper_name(self) -> typing.Optional[interfaces.objects.ObjectInterface]:
         """Determine a name for the hive. Note that some attributes are
         unpredictably blank across different OS versions while others are populated,
         so we check all possibilities and take the first one that's not empty"""
@@ -56,12 +57,12 @@ class _CM_KEY_NODE(objects.Struct):
     """Extension to allow traversal of registry keys"""
 
     @property
-    def helper_volatile(self):
+    def helper_volatile(self) -> bool:
         if not isinstance(self._context.memory[self.vol.layer_name], RegistryHive):
             raise ValueError("Cannot determine volatility of registry key without an offset in a RegistryHive layer")
         return bool(self.vol.offset & 0x80000000)
 
-    def get_subkeys(self):
+    def get_subkeys(self) -> typing.Iterable[interfaces.objects.ObjectInterface]:
         """Returns a list of the key nodes"""
         hive = self._context.memory[self.vol.layer_name]
         if not isinstance(hive, RegistryHive):
@@ -100,7 +101,7 @@ class _CM_KEY_NODE(objects.Struct):
                     vollog.log(constants.LOGLEVEL_VVV,
                                "Node found with address outside the valid Hive size: {}".format(key_offset))
 
-    def get_values(self):
+    def get_values(self) -> typing.Iterable[interfaces.objects.ObjectInterface]:
         """Returns a list of the Value nodes for a key"""
         hive = self._context.memory[self.vol.layer_name]
         if not isinstance(hive, RegistryHive):
@@ -114,11 +115,11 @@ class _CM_KEY_NODE(objects.Struct):
                     yield node
 
     @property
-    def helper_name(self):
+    def helper_name(self) -> interfaces.objects.ObjectInterface:
         """Since this is just a casting convenience, it can be a property"""
         return self.Name.cast("string", max_length = self.NameLength, encoding = "latin-1")
 
-    def get_key_path(self):
+    def get_key_path(self) -> interfaces.objects.ObjectInterface:
         reg = self._context.memory[self.vol.layer_name]
         # 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:
@@ -131,12 +132,12 @@ class _CM_KEY_VALUE(objects.Struct):
     """Extensions to extract data from CM_KEY_VALUE nodes"""
 
     @property
-    def helper_name(self):
+    def helper_name(self) -> interfaces.objects.ObjectInterface:
         """Since this is just a casting convenience, it can be a property"""
         self.Name.count = self.NameLength
         return self.Name.cast("string", max_length = self.NameLength, encoding = "latin-1")
 
-    def decode_data(self):
+    def decode_data(self) -> typing.Union[str, bytes]:
         """Since this is just a casting convenience, it can be a property"""
         # Determine if the data is stored inline
         datalen = self.DataLength & 0x7fffffff
@@ -184,7 +185,7 @@ class _CM_KEY_VALUE(objects.Struct):
                 output = output[:output.find("\x00")]
             return output
         if self_type == RegValueTypes.REG_MULTI_SZ:
-            return str(data, encoding = "utf-16-le").split("\x00")
+            return str(data, encoding = "utf-16-le").split("\x00")[0]
         if self_type == RegValueTypes.REG_BINARY:
             return data
         if self_type == RegValueTypes.REG_NONE:

volatility/framework/symbols/wrappers.py

@@ -1,4 +1,5 @@
 import collections
+import typing
 
 from volatility.framework import interfaces, validity
 
@@ -6,7 +7,7 @@ from volatility.framework import interfaces, validity
 class Flags(validity.ValidityRoutines):
     """Object that converts an integer into a set of flags based on their masks"""
 
-    def __init__(self, choices = None):
+    def __init__(self, choices: typing.Mapping[str, int] = None) -> None:
         self._check_type(choices, collections.Mapping)
         for k, v in choices.items():
             self._check_type(k, str)
@@ -14,10 +15,10 @@ class Flags(validity.ValidityRoutines):
         self._choices = interfaces.objects.ReadOnlyMapping(choices)
 
     @property
-    def choices(self):
+    def choices(self) -> interfaces.objects.ReadOnlyMapping:
         return self._choices
 
-    def __call__(self, value):
+    def __call__(self, value: int) -> typing.List[str]:
         """Return the appropriate Flags """
         result = []
         for k, v in self.choices.items():