Add type-annotations for reamining layer files.

volatility/framework/__init__.py

@@ -18,6 +18,7 @@ import sys
 # 4. If changes or removals of the interface have been made, set age to 0
 
 # We use the libtool library versioning
+import typing
 
 CURRENT = 0  # Number of releases of the library with any change
 REVISION = 0  # Number of changes that don't affect the interface
@@ -48,19 +49,19 @@ def require_interface_version(*args):
 
 
 class noninheritable(object):
-    def __init__(self, f, cls):
-        self.f = f
+    def __init__(self, value: typing.Any, cls: typing.Type) -> None:
+        self.default_value = value
         self.cls = cls
 
-    def __get__(self, obj, type = None):
+    def __get__(self, obj: typing.Any, type: typing.Type = None) -> typing.Any:
         if type == self.cls:
-            if hasattr(self.f, '__get__'):
-                return self.f.__get__(obj, type)
-            return self.f
+            if hasattr(self.default_value, '__get__'):
+                return self.default_value.__get__(obj, type)
+            return self.default_value
         raise AttributeError
 
 
-def hide_from_subclasses(cls):
+def hide_from_subclasses(cls: typing.Type) -> typing.Type:
     cls.hidden = noninheritable(True, cls)
     return cls
 
@@ -76,7 +77,7 @@ def class_subclasses(cls):
             yield return_value
 
 
-def import_files(base_module):
+def import_files(base_module) -> None:
     """Imports all plugins present under plugins path"""
     if not isinstance(base_module.__path__, list):
         raise TypeError("[base_module].__path__ must be a list of paths")
@@ -99,6 +100,7 @@ def import_files(base_module):
                             raise
                     else:
                         vollog.info("Skipping existing module: {}".format(module))
+    return None
 
 
 # Check the python version to ensure it's suitable

volatility/framework/contexts/__init__.py

@@ -73,7 +73,7 @@ class Context(interfaces.context.ContextInterface):
     # ## Object Factory Functions
 
     def object(self,
-               symbol: str,
+               symbol: typing.Union[str, interfaces.objects.Template],
                layer_name: str,
                offset: int,
                **arguments) -> interfaces.objects.ObjectInterface:
@@ -91,9 +91,10 @@ class Context(interfaces.context.ContextInterface):
         :return: A fully constructed object
         :rtype: :py:class:`volatility.framework.interfaces.objects.ObjectInterface`
         """
-        object_template = self.symbol_space.get_type(symbol)
         if not isinstance(symbol, interfaces.objects.Template):
             object_template = self._symbol_space.get_type(symbol)
+        else:
+            object_template = symbol
         object_template = object_template.clone()
         object_template.update_vol(**arguments)
         return object_template(context = self,

volatility/framework/interfaces/context.py

@@ -5,6 +5,7 @@ of symbols that can be used to interpret data in a layer.  The context also prov
 notably the object constructor function, `object`, which will construct a symbol on a layer at a particular offset.
 """
 import copy
+import typing
 from abc import ABCMeta, abstractmethod
 
 from volatility.framework import validity, interfaces
@@ -54,7 +55,7 @@ class ContextInterface(object, metaclass = ABCMeta):
 
     @abstractmethod
     def object(self,
-               symbol: str,
+               symbol: typing.Union[str, 'interfaces.objects.Template'],
                layer_name: str,
                offset: int,
                **arguments):

volatility/framework/interfaces/layers.py

@@ -245,15 +245,15 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR
     def _scan_iterator(self,
                        scanner: 'ScannerInterface',
                        min_address: int,
-                       max_address: int) -> range:
+                       max_address: int) -> typing.Iterable[typing.Any]:
         return range(min_address, max_address, scanner.chunk_size)
 
     def _scan_chunk(self,
                     scanner: 'ScannerInterface',
                     min_address: int,
                     max_address: int,
-                    progress: multiprocessing.Value,
-                    iterator_value: int) -> typing.List[typing.Any]:
+                    progress: ProgressValue,
+                    iterator_value: typing.Any) -> typing.List[typing.Any]:
         length = min(scanner.chunk_size + scanner.overlap, max_address - iterator_value)
         chunk = self.read(iterator_value, length)
         # Don't include the overlaps, or we'll go over 100%
@@ -286,7 +286,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
     def mapping(self,
                 offset: int,
                 length: int,
-                ignore_errors: bool = False) -> typing.List[typing.Tuple[int, int, int, str]]:
+                ignore_errors: bool = False) -> typing.Iterable[typing.Tuple[int, int, int, str]]:
         """Returns a sorted iterable of (offset, mapped_offset, length, layer) mappings
 
            ignore_errors will provide all available maps with gaps, but their total length may not add up to the requested length
@@ -302,7 +302,8 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
 
     ### Translation layer convenience function
 
-    def translate(self, offset: int, ignore_errors: bool = False) -> typing.Tuple[int, str]:
+    def translate(self, offset: int, ignore_errors: bool = False) \
+            -> typing.Tuple[typing.Optional[int], typing.Optional[str]]:
         mapping = self.mapping(offset, 0, ignore_errors)
         if mapping:
             _, mapped_offset, _, layer = list(mapping)[0]
@@ -356,7 +357,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
                     min_address: int,
                     max_address: int,
                     progress: ProgressValue,
-                    iterator_value: int) -> typing.List[typing.Any]:
+                    iterator_value: typing.Any) -> typing.List[typing.Any]:
         size_to_scan = min(max_address - min_address, scanner.chunk_size + scanner.overlap)
         result = []  # type: typing.List[typing.Any]
         for map in self.mapping(iterator_value, size_to_scan, ignore_errors = True):

volatility/framework/layers/intel.py

@@ -1,20 +1,25 @@
 import logging
 import math
 import struct
+import typing
 
 from volatility.framework import exceptions, interfaces
 from volatility.framework.configuration import requirements
 
 vollog = logging.getLogger(__name__)
 
+IteratorValue = typing.Tuple[typing.List[typing.Tuple[str, int, int]], int]
+
 
 class classproperty(object):
-    """Class property decorator"""
+    """Class property decorator
 
-    def __init__(self, func):
+    Note this will change the return type """
+
+    def __init__(self, func: typing.Callable[[typing.Any], typing.Any]) -> None:
         self._func = func
 
-    def __get__(self, owner_self, owner_cls):
+    def __get__(self, _owner_self, owner_cls: typing.Type) -> typing.Any:
         return self._func(owner_cls)
 
 
@@ -32,7 +37,10 @@ class Intel(interfaces.layers.TranslationLayerInterface):
     _structure = [('page directory', 10, False),
                   ('page table', 10, True)]
 
-    def __init__(self, context, config_path, name):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 name: str) -> None:
         super().__init__(context, config_path, name)
         self._base_layer = self._check_type(self.config["memory_layer"], str)
         self._page_map_offset = self._check_type(self.config["page_map_offset"], int)
@@ -42,31 +50,31 @@ class Intel(interfaces.layers.TranslationLayerInterface):
         self._index_shift = int(math.ceil(math.log2(struct.calcsize(self._entry_format))))
 
     @classproperty
-    def page_size(cls):
+    def page_size(cls) -> int:
         """Page size for the intel address space.
 
         All Intel address spaces work on 4096 byte pages"""
         return 1 << cls._page_size_in_bits
 
     @classproperty
-    def bits_per_register(cls):
+    def bits_per_register(cls) -> int:
         """Returns the bits_per_register to determine the range of an IntelTranslationLayer"""
         return cls._bits_per_register
 
     @classproperty
-    def minimum_address(cls):
+    def minimum_address(cls) -> int:  # type: ignore
         return 0
 
     @classproperty
-    def maximum_address(cls):
+    def maximum_address(cls) -> int:  # type: ignore
         return (1 << cls._maxvirtaddr) - 1
 
     @classproperty
-    def structure(cls):
+    def structure(cls) -> typing.List[typing.Tuple[str, int, bool]]:
         return cls._structure
 
     @staticmethod
-    def _mask(value, high_bit, low_bit):
+    def _mask(value: int, high_bit: int, low_bit: int) -> int:
         """Returns the bits of a value between highbit and lowbit inclusive"""
         high_mask = (2 ** (high_bit + 1)) - 1
         low_mask = (2 ** low_bit) - 1
@@ -75,11 +83,11 @@ class Intel(interfaces.layers.TranslationLayerInterface):
         return value & mask
 
     @staticmethod
-    def _page_is_valid(entry):
+    def _page_is_valid(entry: int) -> bool:
         """Returns whether a particular page is valid based on its entry"""
-        return entry & 1
+        return bool(entry & 1)
 
-    def _translate(self, offset):
+    def _translate(self, offset: int) -> typing.Tuple[int, int, str]:
         """Translates a specific offset based on paging tables
 
            Returns the translated offset, the contiguous pagesize that the translated address lives in and the layer_name that the address lives in
@@ -121,7 +129,7 @@ class Intel(interfaces.layers.TranslationLayerInterface):
         page = self._mask(entry, self._maxphyaddr - 1, position + 1) | self._mask(offset, position, 0)
         return page, 1 << (position + 1), self._base_layer
 
-    def is_valid(self, offset, length = 1):
+    def is_valid(self, offset: int, length: int = 1) -> bool:
         """Returns whether the address offset can be translated to a valid address"""
         try:
             # TODO: Consider reimplementing this, since calls to mapping can call is_valid
@@ -130,12 +138,14 @@ class Intel(interfaces.layers.TranslationLayerInterface):
         except exceptions.InvalidAddressException:
             return False
 
-    def mapping(self, offset, length, ignore_errors = False):
+    def mapping(self,
+                offset: int,
+                length: int,
+                ignore_errors: bool = False) -> typing.Iterable[typing.Tuple[int, int, int, str]]:
         """Returns a sorted iterable of (offset, mapped_offset, length, layer) mappings
 
            This allows translation layers to provide maps of contiguous regions in one layer
         """
-        result = []
         if length == 0:
             if ignore_errors and not self.is_valid(offset):
                 raise StopIteration
@@ -156,13 +166,13 @@ class Intel(interfaces.layers.TranslationLayerInterface):
             offset += chunk_size
 
     @property
-    def dependencies(self):
+    def dependencies(self) -> typing.List[str]:
         """Returns a list of the lower layer names that this layer is dependent upon"""
         # TODO: Add in the whole buffalo
         return [self._base_layer]
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'memory_layer',
                                                          optional = False),
                 requirements.TranslationLayerRequirement(name = 'swap_layer',
@@ -174,10 +184,14 @@ class Intel(interfaces.layers.TranslationLayerInterface):
                 requirements.StringRequirement(name = 'linux_banner',
                                                optional = True)]
 
-    def _scan_iterator(self, scanner, min_address, max_address):
+    def _scan_iterator(self,
+                       scanner: interfaces.layers.ScannerInterface,
+                       min_address: int,
+                       max_address: int) \
+            -> typing.Iterable[IteratorValue]:
         previous = None
-        data_to_scan = []
-        scanned_pairs = set()
+        data_to_scan = []  # type: typing.List[typing.Tuple[str, int, int]]
+        scanned_pairs = set()  # type: typing.Set[typing.Tuple[int, int]]
         chunk_end = min_address
         while chunk_end <= max_address:
             try:
@@ -206,7 +220,13 @@ class Intel(interfaces.layers.TranslationLayerInterface):
             previous = address
             chunk_end += chunk_size
 
-    def _scan_chunk(self, scanner, min_address, max_address, progress, iterator_value):
+    # We ignore the type due to the iterator_value, actually it only needs to match the output from _scan_iterator
+    def _scan_chunk(self,
+                    scanner: interfaces.layers.ScannerInterface,
+                    min_address: int,
+                    max_address: int,
+                    progress: interfaces.layers.ProgressValue,
+                    iterator_value: IteratorValue) -> typing.List[typing.Any]:
         data_to_scan, chunk_end = iterator_value
         data = b''
         for layer_name, address, chunk_size in data_to_scan:
@@ -219,7 +239,11 @@ class Intel(interfaces.layers.TranslationLayerInterface):
         progress.value = chunk_end
         return list(scanner(data, chunk_end - len(data_to_scan)))
 
-    def _scan_metric(self, _scanner, min_address, max_address, value):
+    def _scan_metric(self,
+                     _scanner: interfaces.layers.ScannerInterface,
+                     min_address: int,
+                     max_address: int,
+                     value: int) -> float:
         return max(0, ((value - min_address) * 100) / (max_address - min_address))
 
 
@@ -254,7 +278,7 @@ class Intel32e(Intel):
 
 class WindowsMixin(object):
     @staticmethod
-    def _page_is_valid(entry):
+    def _page_is_valid(entry: int) -> bool:
         """Returns whether a particular page is valid based on its entry
 
            Windows uses additional "available" bits to store flags
@@ -264,7 +288,7 @@ class WindowsMixin(object):
 
            For more information, see Windows Internals (6th Ed, Part 2, pages 268-269)
         """
-        return (entry & 1) or ((entry & 1 << 11) and not entry & 1 << 10)
+        return bool((entry & 1) or ((entry & 1 << 11) and not entry & 1 << 10))
 
 
 ### These must be full separate classes so that JSON configs re-create them properly

volatility/framework/layers/lime.py

@@ -5,8 +5,9 @@ Created on 6 Apr 2016
 """
 
 import struct
+import typing
 
-from volatility.framework import exceptions, interfaces
+from volatility.framework import exceptions, interfaces, validity
 from volatility.framework.layers import segmented
 
 
@@ -29,13 +30,16 @@ class LimeLayer(segmented.SegmentedLayer):
     # XXX move this to a custom SymbolSpace?
     _header_struct = struct.Struct('<IIQQQ')
 
-    def __init__(self, context, config_path, name):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 name: str) -> None:
         super().__init__(context, config_path, name)
 
         # We must run this on creation in order to get the right min/maxaddr in case scanning is our first action
         self._load_segments()
 
-    def _load_segments(self):
+    def _load_segments(self) -> None:
         base_layer = self._context.memory[self._base_layer]
         base_maxaddr = base_layer.maximum_address
         maxaddr = 0
@@ -61,7 +65,9 @@ class LimeLayer(segmented.SegmentedLayer):
         self._segments = segments
 
     @classmethod
-    def _check_header(cls, base_layer, offset = 0):
+    def _check_header(cls,
+                      base_layer: interfaces.layers.DataLayerInterface,
+                      offset: int = 0) -> typing.Tuple[int, int]:
         header_data = base_layer.read(offset, cls._header_struct.size)
         (magic, version, start, end, reserved) = cls._header_struct.unpack(header_data)
         if magic != cls.MAGIC:
@@ -75,11 +81,15 @@ class LimeStacker(interfaces.automagic.StackerLayerInterface):
     stack_order = 10
 
     @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.Optional[interfaces.layers.DataLayerInterface]:
         try:
             LimeLayer._check_header(context.memory[layer_name])
         except LimeFormatException:
-            return
+            return None
         new_name = context.memory.free_layer_name("LimeLayer")
         context.config[interfaces.configuration.path_join(new_name, "base_layer")] = layer_name
         return LimeLayer(context, new_name, new_name)

volatility/framework/layers/physical.py

@@ -1,3 +1,5 @@
+import typing
+
 from volatility.framework import exceptions, interfaces, layers
 from volatility.framework.configuration import requirements
 
@@ -8,26 +10,30 @@ class BufferDataLayer(interfaces.layers.DataLayerInterface):
     provides = {"type": "physical"}
     priority = 10
 
-    def __init__(self, context, config_path, name, buffer):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 name: str,
+                 buffer: bytes) -> None:
         super().__init__(context, config_path, name)
         self._buffer = self._check_type(buffer, bytes)
 
     @property
-    def maximum_address(self):
+    def maximum_address(self) -> int:
         """Returns the largest available address in the space"""
         return len(self._buffer) - 1
 
     @property
-    def minimum_address(self):
+    def minimum_address(self) -> int:
         """Returns the smallest available address in the space"""
         return 0
 
-    def is_valid(self, offset, length = 1):
+    def is_valid(self, offset: int, length: int = 1) -> bool:
         """Returns whether the offset is valid or not"""
-        return (self.minimum_address <= offset <= self.maximum_address and
-                self.minimum_address <= offset + length - 1 <= self.maximum_address)
+        return bool(self.minimum_address <= offset <= self.maximum_address and
+                    self.minimum_address <= offset + length - 1 <= self.maximum_address)
 
-    def read(self, address, length, pad = False):
+    def read(self, address: int, length: int, pad: bool = False) -> bytes:
         """Reads the data from the buffer"""
         if not self.is_valid(address, length):
             invalid_address = address
@@ -37,13 +43,13 @@ class BufferDataLayer(interfaces.layers.DataLayerInterface):
                                                      "Offset outside of the buffer boundaries")
         return self._buffer[address:address + length]
 
-    def write(self, address, data):
+    def write(self, address: int, data: bytes):
         """Writes the data from to the buffer"""
         self._check_type(data, bytes)
         self._buffer = self._buffer[:address] + data + self._buffer[address + len(data):]
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         # No real requirements (only the buffer).  Need to figure out if there's a better way of representing this
         return [requirements.BytesRequirement(name = 'buffer', description = "The direct bytes to interact with",
                                               optional = False)]
@@ -55,23 +61,26 @@ class FileLayer(interfaces.layers.DataLayerInterface):
     provides = {"type": "physical"}
     priority = 20
 
-    def __init__(self, context, config_path, name):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 name: str) -> None:
         super().__init__(context, config_path, name)
 
         self._location = self.config["location"]
         self._accessor = layers.ResourceAccessor()
         self._file_ = None
-        self._size = None
+        self._size = None  # type: typing.Optional[int]
         # Instantiate the file to throw exceptions if the file doesn't open
         _ = self._file
 
     @property
-    def location(self):
+    def location(self) -> str:
         """Returns the location on which this Layer abstracts"""
         return self._location
 
     @property
-    def _file(self):
+    def _file(self) -> typing.IO[typing.Any]:
         """Property to prevent the initializer storing an unserializable open file (for context cloning)"""
         # FIXME: Add "+" to the mode once we've determined whether write mode is enabled
         mode = "rb"
@@ -80,7 +89,7 @@ class FileLayer(interfaces.layers.DataLayerInterface):
         return self._file_
 
     @property
-    def maximum_address(self):
+    def maximum_address(self) -> int:
         """Returns the largest available address in the space"""
         # Zero based, so we return the size of the file minus 1
         if self._size:
@@ -92,18 +101,18 @@ class FileLayer(interfaces.layers.DataLayerInterface):
         return self._size
 
     @property
-    def minimum_address(self):
+    def minimum_address(self) -> int:
         """Returns the smallest available address in the space"""
         return 0
 
-    def is_valid(self, offset, length = 1):
+    def is_valid(self, offset: int, length: int = 1) -> bool:
         """Returns whether the offset is valid or not"""
         if length <= 0:
             raise TypeError("Length must be positive")
-        return (self.minimum_address <= offset <= self.maximum_address and
-                self.minimum_address <= offset + length - 1 <= self.maximum_address)
+        return bool(self.minimum_address <= offset <= self.maximum_address and
+                    self.minimum_address <= offset + length - 1 <= self.maximum_address)
 
-    def read(self, offset, length, pad = False):
+    def read(self, offset: int, length: int, pad: bool = False) -> bytes:
         """Reads from the file at offset for length"""
         if not self.is_valid(offset, length):
             invalid_address = offset
@@ -122,7 +131,7 @@ class FileLayer(interfaces.layers.DataLayerInterface):
                                                          self.name + " file")
         return data
 
-    def write(self, offset, data):
+    def write(self, offset: int, data: bytes) -> None:
         """Writes to the file
 
            This will technically allow writes beyond the extent of the file
@@ -136,7 +145,7 @@ class FileLayer(interfaces.layers.DataLayerInterface):
         self._file.seek(offset)
         self._file.write(data)
 
-    def __getstate__(self):
+    def __getstate__(self) -> typing.Dict[str, typing.Any]:
         """Do not store the open _file_ attribute, our property will ensure the file is open when needed
 
            This is necessary for multi-processing
@@ -144,10 +153,10 @@ class FileLayer(interfaces.layers.DataLayerInterface):
         self._file_ = None
         return self.__dict__
 
-    def destroy(self):
+    def destroy(self) -> None:
         """Closes the file handle"""
         self._file.close()
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.StringRequirement(name = 'location', optional = False)]

volatility/framework/layers/registry.py

@@ -1,7 +1,8 @@
 import logging
 import os.path as os_path
+import typing
 
-from volatility.framework import constants, exceptions, interfaces
+from volatility.framework import constants, exceptions, interfaces, objects
 from volatility.framework.configuration import requirements
 from volatility.framework.configuration.requirements import IntRequirement
 from volatility.framework.interfaces.configuration import TranslationLayerRequirement
@@ -19,7 +20,11 @@ class RegistryInvalidIndex(exceptions.LayerException):
 
 
 class RegistryHive(interfaces.layers.TranslationLayerInterface):
-    def __init__(self, context, config_path, name, os = "Unknown"):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 name: str,
+                 os: str = "Unknown") -> None:
         super().__init__(context, config_path, name, os)
 
         self._base_layer = self.config["base_layer"]
@@ -51,23 +56,23 @@ class RegistryHive(interfaces.layers.TranslationLayerInterface):
                 "Invalid registry base_block length: {}".format(self._base_block.Length))
 
     @property
-    def address_mask(self):
+    def address_mask(self) -> int:
         """Return a mask that allows for the volatile bit to be set"""
         return super().address_mask | 0x80000000
 
     @property
-    def root_cell_offset(self):
+    def root_cell_offset(self) -> int:
         """Returns the offset for the root cell in this hive"""
         return self._base_block.RootCell
 
-    def get_cell(self, cell_offset):
+    def get_cell(self, cell_offset: int) -> 'objects.Struct':
         """Returns the appropriate Cell value for a cell offset"""
         # This would be an _HCELL containing CELL_DATA, but to save time we skip the size of the HCELL
         cell = self._context.object(symbol = self._table_name + constants.BANG + "_CELL_DATA", offset = cell_offset + 4,
                                     layer_name = self.name)
         return cell
 
-    def get_node(self, cell_offset):
+    def get_node(self, cell_offset: int) -> 'objects.Struct':
         """Returns the appropriate Node, interpreted from the Cell based on its Signature"""
         cell = self.get_cell(cell_offset)
         signature = cell.cast('string', max_length = 2, encoding = 'latin-1')
@@ -89,13 +94,13 @@ class RegistryHive(interfaces.layers.TranslationLayerInterface):
                 "Unknown Signature {} (0x{:x}) at offset {}".format(signature, cell.u.KeyNode.Signature, cell_offset))
             return cell
 
-    def get_key(self, key):
+    def get_key(self, key: str) -> interfaces.objects.ObjectInterface:
         """Gets a specific registry key by key path"""
         node_key = self.get_node(self.root_cell_offset)
         if key.endswith("\\"):
             key = key[:-1]
         key_array = key.split('\\')
-        found_key = []
+        found_key = []  # type: typing.List[str]
         while key_array and node_key:
             for subkey in node_key.get_subkeys():
                 if subkey.helper_name == key_array[0]:
@@ -105,10 +110,12 @@ class RegistryHive(interfaces.layers.TranslationLayerInterface):
             else:
                 node_key = None
         if not node_key:
-            raise KeyError("Key {} not found under {}", key_array[0], found_key.join('\\'))
+            raise KeyError("Key {} not found under {}", key_array[0], '\\'.join(found_key))
         return node_key
 
-    def visit_nodes(self, visitor, node = None):
+    def visit_nodes(self,
+                    visitor: typing.Callable[[objects.Struct], None],
+                    node: typing.Optional[objects.Struct] = None) -> None:
         """Applies a callable (visitor) to all nodes within the registry tree from a given node"""
         if not node:
             node = self.get_node(self.root_cell_offset)
@@ -117,7 +124,7 @@ class RegistryHive(interfaces.layers.TranslationLayerInterface):
             self.visit_nodes(visitor, node)
 
     @staticmethod
-    def _mask(value, high_bit, low_bit):
+    def _mask(value: int, high_bit: int, low_bit: int) -> int:
         """Returns the bits of a value between highbit and lowbit inclusive"""
         high_mask = (2 ** (high_bit + 1)) - 1
         low_mask = (2 ** low_bit) - 1
@@ -125,12 +132,13 @@ class RegistryHive(interfaces.layers.TranslationLayerInterface):
         # print(high_bit, low_bit, bin(mask), bin(value))
         return value & mask
 
-    def get_requirements(cls):
+    @classmethod
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [IntRequirement(name = 'hive_offset', description = '', default = 0, optional = False),
                 requirements.SymbolRequirement(name = "nt_symbols", description = "Windows OS"),
                 TranslationLayerRequirement(name = 'base_layer', optional = False)]
 
-    def _translate(self, offset):
+    def _translate(self, offset: int) -> int:
         """Translates a single cell index to a cell memory offset and the suboffset within it"""
 
         # Ignore the volatile bit when determining maxaddr validity
@@ -147,7 +155,10 @@ class RegistryHive(interfaces.layers.TranslationLayerInterface):
         entry = table.Table[table_index]
         return entry.helper_block_offset + suboffset
 
-    def mapping(self, offset, length, ignore_errors = False):
+    def mapping(self,
+                offset: int,
+                length: int,
+                ignore_errors: bool = False) -> typing.Iterable[typing.Tuple[int, int, int, str]]:
 
         # TODO: Check the offset and offset + length are not outside the norms
         if (length < 0):
@@ -177,19 +188,19 @@ class RegistryHive(interfaces.layers.TranslationLayerInterface):
         return response
 
     @property
-    def dependencies(self):
+    def dependencies(self) -> typing.List[str]:
         """Returns a list of layer names that this layer translates onto"""
         return [self.config['base_layer']]
 
-    def is_valid(self, offset, length = 1):
+    def is_valid(self, offset: int, length: int = 1) -> bool:
         """Returns a boolean based on whether the offset is valid or not"""
         # TODO: Fix me
         return True
 
     @property
-    def minimum_address(self):
+    def minimum_address(self) -> int:
         return self._minaddr
 
     @property
-    def maximum_address(self):
+    def maximum_address(self) -> int:
         return self._maxaddr

volatility/framework/layers/segmented.py

@@ -1,3 +1,4 @@
+import typing
 from abc import ABCMeta, abstractmethod
 from bisect import bisect_right
 
@@ -11,32 +12,35 @@ class SegmentedLayer(interfaces.layers.TranslationLayerInterface, metaclass = AB
        In the documentation "mapped address" or "mapped offset" refers to an offset once it has been mapped to the underlying layer
     """
 
-    def __init__(self, context, config_path, name):
+    def __init__(self,
+                 context: interfaces.configuration.ContextInterface,
+                 config_path: str,
+                 name: str) -> None:
         super().__init__(context, config_path = config_path, name = name)
 
         self._base_layer = self.config["base_layer"]
-        self._segments = []
-        self._minaddr = None
-        self._maxaddr = None
+        self._segments = []  # type: typing.List[typing.Tuple[int, int, int]]
+        self._minaddr = None  # type: typing.Optional[int]
+        self._maxaddr = None  # type: typing.Optional[int]
 
         self._load_segments()
 
     @abstractmethod
-    def _load_segments(self):
+    def _load_segments(self) -> None:
         """Populates the _segments variable
 
            Segments must be (address, mapped address, length) and must be sorted by address when this method exits
         """
 
-    def is_valid(self, offset, length = 1):
+    def is_valid(self, offset: int, length: int = 1) -> bool:
         """Returns whether the address offset can be translated to a valid address"""
         try:
-            return all([self._context.memory[self._base_layer].is_valid(mapped_offset) for _, mapped_offset, _, _ in
+            return all([self._context.memory[self._base_layer].is_valid(mapped_offset) for _i, mapped_offset, _i, _s in
                         self.mapping(offset, length)])
         except exceptions.InvalidAddressException:
             return False
 
-    def _find_segment(self, offset, next = False):
+    def _find_segment(self, offset: int, next: bool = False) -> typing.Tuple[int, int, int]:
         """Finds the segment containing a given offset
 
            Returns the segment tuple (offset, mapped_offset, length)
@@ -57,7 +61,8 @@ class SegmentedLayer(interfaces.layers.TranslationLayerInterface, metaclass = AB
                     return self._segments[i]
         raise exceptions.InvalidAddressException(self.name, offset, "Invalid address at {:0x}".format(offset))
 
-    def mapping(self, offset, length, ignore_errors = False):
+    def mapping(self, offset: int, length: int, ignore_errors: bool = False) \
+            -> typing.Iterable[typing.Tuple[int, int, int, str]]:
         """Returns a sorted iterable of (offset, mapped_offset, length, layer) mappings"""
         done = False
         current_offset = offset
@@ -94,7 +99,7 @@ class SegmentedLayer(interfaces.layers.TranslationLayerInterface, metaclass = AB
                 done = True
 
     @property
-    def minimum_address(self):
+    def minimum_address(self) -> int:
         if not self._segments:
             raise ValueError("SegmentedLayer must contain some segments")
         if self._minaddr is None:
@@ -103,7 +108,7 @@ class SegmentedLayer(interfaces.layers.TranslationLayerInterface, metaclass = AB
         return self._minaddr
 
     @property
-    def maximum_address(self):
+    def maximum_address(self) -> int:
         if not self._segments:
             raise ValueError("SegmentedLayer must contain some segments")
         if self._maxaddr is None:
@@ -112,11 +117,11 @@ class SegmentedLayer(interfaces.layers.TranslationLayerInterface, metaclass = AB
         return self._maxaddr
 
     @property
-    def dependencies(self):
+    def dependencies(self) -> typing.List[str]:
         """Returns a list of the lower layers that this layer is dependent upon"""
         return [self._base_layer]
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         return [requirements.TranslationLayerRequirement(name = 'base_layer',
                                                          optional = False)]

volatility/framework/layers/vmware.py

@@ -1,7 +1,8 @@
 import os
 import struct
+import typing
 
-from volatility.framework import interfaces
+from volatility.framework import interfaces, validity
 from volatility.framework.configuration import requirements
 from volatility.framework.layers import physical, segmented
 from volatility.framework.symbols import native
@@ -14,7 +15,10 @@ class VmwareLayer(segmented.SegmentedLayer):
     header_structure = "<4sII"
     group_structure = "64sQQ"
 
-    def __init__(self, context, config_path, name):
+    def __init__(self,
+                 context: interfaces.context.ContextInterface,
+                 config_path: str,
+                 name: str) -> None:
         # Construct these so we can use self.config
         self._context = context
         self._config_path = config_path
@@ -23,11 +27,11 @@ class VmwareLayer(segmented.SegmentedLayer):
         # Then call the super, which will call load_segments (which needs the base_layer before it'll work)
         super().__init__(context, config_path = config_path, name = name)
 
-    def _load_segments(self):
+    def _load_segments(self) -> None:
         """Loads up the segments from the meta_layer"""
         self._read_header()
 
-    def _read_header(self):
+    def _read_header(self) -> None:
         """Checks the vmware header to make sure it's valid"""
         if "vmware" not in self._context.symbol_space:
             self._context.symbol_space.append(native.NativeTable("vmware", native.std_ctypes))
@@ -86,11 +90,11 @@ class VmwareLayer(segmented.SegmentedLayer):
             self._segments.append((offset, mapped_offset, length))
 
     @property
-    def dependencies(self):
+    def dependencies(self) -> typing.List[str]:
         return [self._base_layer, self._meta_layer]
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         """This vmware translation layer always requires a separate metadata layer"""
         return [requirements.TranslationLayerRequirement(name = 'base_layer',
                                                          optional = False),
@@ -101,10 +105,14 @@ class VmwareLayer(segmented.SegmentedLayer):
 
 class VmwareStacker(interfaces.automagic.StackerLayerInterface):
     @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.Optional[interfaces.layers.DataLayerInterface]:
         """Attempt to stack this based on the starting information"""
         if not isinstance(context.memory[layer_name], physical.FileLayer):
-            return
+            return None
         location = context.memory[layer_name].location
         if location.endswith(".vmem"):
             vmss = location[:-5] + ".vmss"
@@ -119,10 +127,11 @@ class VmwareStacker(interfaces.automagic.StackerLayerInterface):
                 context.config[interfaces.configuration.path_join(current_config_path, "location")] = vmss
                 context.memory.add_layer(physical.FileLayer(context, current_config_path, current_layer_name))
             else:
-                return
+                return None
             new_layer_name = context.memory.free_layer_name("VmwareLayer")
             context.config[interfaces.configuration.path_join(current_config_path, "base_layer")] = layer_name
             context.config[
                 interfaces.configuration.path_join(current_config_path, "meta_layer")] = current_layer_name
             new_layer = VmwareLayer(context, current_config_path, new_layer_name)
             return new_layer
+        return None