Add in type-annotations for the remaining interfaces.

volatility/framework/interfaces/layers.py

@@ -1,15 +1,15 @@
 """Defines layers for containing data.  One layer may combine other layers, map data based on the data itself,
  or map a procedure (such as decryption) across another layer of data."""
-import collections
 import collections.abc
 import functools
 import logging
 import math
 import multiprocessing
 import traceback
+import typing
 from abc import ABCMeta, abstractmethod
 
-from volatility.framework import constants, exceptions, validity
+from volatility.framework import constants, exceptions, validity, interfaces
 from volatility.framework.interfaces import configuration, context
 
 vollog = logging.getLogger(__name__)
@@ -23,6 +23,8 @@ try:
 except ImportError:
     pass
 
+ProgressValue = typing.Union['DummyProgress', multiprocessing.Value]
+
 
 class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta):
     """Class for layer scanners that return locations of particular values from within the data
@@ -47,32 +49,32 @@ class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta):
     """
     thread_safe = False
 
-    def __init__(self):
+    def __init__(self) -> None:
         self.chunk_size = 0x1000000  # Default to 16Mb chunks
         self.overlap = 0x1000  # A page of overlap by default
         self._context = None
-        self._layer_name = None
+        self._layer_name: typing.Optional[str] = None
 
     @property
-    def context(self):
+    def context(self) -> 'interfaces.context.ContextInterface':
         return self._context
 
     @context.setter
-    def context(self, ctx):
+    def context(self, ctx: 'interfaces.context.ContextInterface') -> None:
         """Stores the context locally in case the scanner needs to access the layer"""
         self._context = self._check_type(ctx, context.ContextInterface)
 
     @property
-    def layer_name(self):
+    def layer_name(self) -> str:
         return self._layer_name
 
     @layer_name.setter
-    def layer_name(self, layer_name):
+    def layer_name(self, layer_name: str) -> None:
         """Stores the layer_name being scanned locally in case the scanner needs to access the layer"""
         self._layer_name = self._check_type(layer_name, str)
 
     @abstractmethod
-    def __call__(self, data, data_offset):
+    def __call__(self, data: bytes, data_offset: int) -> typing.Iterable[typing.Any]:
         """Searches through a chunk of data for a particular value/pattern/etc
            Always returns an iterator of the same type of object (need not be a volatility object)
 
@@ -87,7 +89,11 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR
 
     _architecture = "Unknown"
 
-    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)
         self._name = self._check_type(name, str)
         self._os = self._check_type(os, str)
@@ -95,7 +101,7 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR
     # Memory specific attributes
 
     @property
-    def architecture(self):
+    def architecture(self) -> str:
         """The architecutre of the TranslationLayer
 
         This cannot be modified after construction outside of the class
@@ -103,43 +109,43 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR
         return self._architecture
 
     @property
-    def os(self):
+    def os(self) -> str:
         """The operating system related to the TranslationLayer"""
         return self._os
 
     @os.setter
-    def os(self, value):
+    def os(self, value: str) -> None:
         """Sets the operating system of the TranslationLayer"""
         self._os = self._check_type(value, str)
 
     # Standard attributes
 
     @property
-    def name(self):
+    def name(self) -> str:
         """Returns the layer name"""
         return self._name
 
     @property
     @abstractmethod
-    def maximum_address(self):
+    def maximum_address(self) -> int:
         """Returns the maximum valid address of the space"""
 
     @property
     @abstractmethod
-    def minimum_address(self):
+    def minimum_address(self) -> int:
         """Returns the minimum valid address of the space"""
 
     @property
-    def address_mask(self):
+    def address_mask(self) -> int:
         """Returns a mask which encapsulates all the actives bit of an address for this layer"""
         return (1 << int(math.ceil(math.log2(self.maximum_address)))) - 1
 
     @abstractmethod
-    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"""
 
     @abstractmethod
-    def read(self, offset, length, pad = False):
+    def read(self, offset: int, length: int, pad: bool = False) -> bytes:
         """Reads an offset for length bytes and returns 'bytes' (not 'str') of length size
 
            If there is a fault of any kind (such as a page fault), an exception will be thrown
@@ -147,14 +153,14 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR
         """
 
     @abstractmethod
-    def write(self, offset, data):
+    def write(self, offset: int, data: bytes) -> None:
         """Writes a chunk of data at offset.
 
            Any unavailable sections in the underlying bases will cause an exception to be thrown.
            Note: Writes are not atomic, therefore some data can be written, even if an exception is thrown.
         """
 
-    def destroy(self):
+    def destroy(self) -> None:
         """Allows DataLayers to close any open handles, etc.
 
            Systems that make use of Data Layers should called destroy when they are done with them.
@@ -163,19 +169,29 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR
         pass
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]:
         """Returns a list of Requirement objects for this type of layer"""
         return []
 
+    @property
+    def dependencies(self) -> typing.List[str]:
+        """DataLayers must never define on other layers"""
+        return []
+
     # ## General scanning methods
 
-    def scan(self, context, scanner, progress_callback = None, min_address = None, max_address = None):
+    def scan(self,
+             context: interfaces.context.ContextInterface,
+             scanner: ScannerInterface,
+             progress_callback: validity.ProgressCallback = None,
+             min_address: typing.Optional[int] = None,
+             max_address: typing.Optional[int] = None) -> typing.Iterable[typing.Any]:
         """Scans a Translation layer by chunk
 
            Note: this will skip missing/unmappable chunks of memory
         """
-        if progress_callback is not None:
-            self._check_type(progress_callback, collections.Callable)
+        if progress_callback is not None and not callable(progress_callback):
+            raise TypeError("Progress_callback is not callable")
 
         scanner = self._check_type(scanner, ScannerInterface)
         scanner.context = context
@@ -194,6 +210,7 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR
         max_address = min(self.maximum_address, max_address)
 
         try:
+            progress: ProgressValue
             scan_iterator = functools.partial(self._scan_iterator, scanner, min_address, max_address)
             scan_metric = functools.partial(self._scan_metric, scanner, min_address, max_address)
             if scanner.thread_safe and not constants.DISABLE_MULTITHREADED_SCANNING:
@@ -225,20 +242,32 @@ class DataLayerInterface(configuration.ConfigurableInterface, validity.ValidityR
             vollog.log(constants.LOGLEVEL_VVV,
                        "\n".join(traceback.TracebackException.from_exception(e).format(chain = True)))
 
-    def _scan_iterator(self, scanner, min_address, max_address):
+    def _scan_iterator(self,
+                       scanner: 'ScannerInterface',
+                       min_address: int,
+                       max_address: int) -> range:
         return range(min_address, max_address, scanner.chunk_size)
 
-    def _scan_chunk(self, scanner, min_address, max_address, progress, iterator_value):
+    def _scan_chunk(self,
+                    scanner: 'ScannerInterface',
+                    min_address: int,
+                    max_address: int,
+                    progress: multiprocessing.Value,
+                    iterator_value: int) -> 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%
         progress.value += min(scanner.chunk_size, max_address - iterator_value)
         return list(scanner(chunk, iterator_value))
 
-    def _scan_metric(self, _scanner, min_address, max_address, value):
+    def _scan_metric(self,
+                     _scanner: 'ScannerInterface',
+                     min_address: int,
+                     max_address: int,
+                     value: int) -> float:
         return max(0, (value * 100) / (max_address - min_address))
 
-    def build_configuration(self):
+    def build_configuration(self) -> interfaces.configuration.HierarchicalDict:
         config = super().build_configuration()
 
         # Translation Layers are constructable, and therefore require a class configuration variable
@@ -254,7 +283,10 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
     provides = {"type": "interface"}
 
     @abstractmethod
-    def mapping(self, offset, length, ignore_errors = False):
+    def mapping(self,
+                offset: int,
+                length: int,
+                ignore_errors: bool = False) -> typing.List[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
@@ -264,13 +296,13 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
 
     @property
     @abstractmethod
-    def dependencies(self):
+    def dependencies(self) -> typing.List[str]:
         """Returns a list of layer names that this layer translates onto"""
         return []
 
     ### Translation layer convenience function
 
-    def translate(self, offset, ignore_errors = False):
+    def translate(self, offset: int, ignore_errors: bool = False) -> typing.Tuple[int, str]:
         mapping = self.mapping(offset, 0, ignore_errors)
         if mapping:
             _, mapped_offset, _, layer = list(mapping)[0]
@@ -278,15 +310,16 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
             if ignore_errors:
                 # We should only hit this if we ignored errors, but check anyway
                 return None, None
-            raise exceptions.InvalidAddressException("Cannot translate {} in layer {}".format(offset, self.name))
+            raise exceptions.InvalidAddressException(self.name, offset,
+                                                     "Cannot translate {} in layer {}".format(offset, self.name))
         return mapped_offset, layer
 
     # ## Read/Write functions for mapped pages
 
-    def read(self, offset, length, pad = False):
+    def read(self, offset: int, length: int, pad: bool = False) -> bytes:
         """Reads an offset for length bytes and returns 'bytes' (not 'str') of length size"""
         current_offset = offset
-        output = []
+        output: typing.List[bytes] = []
         for (offset, mapped_offset, length, layer) in self.mapping(offset, length, ignore_errors = pad):
             if not pad and offset > current_offset:
                 raise exceptions.InvalidAddressException(self.name, current_offset,
@@ -302,7 +335,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
         recovered_data = b"".join(output)
         return recovered_data + b"\x00" * (length - len(recovered_data))
 
-    def write(self, offset, value):
+    def write(self, offset: int, value: bytes) -> None:
         """Writes a value at offset, distributing the writing across any underlying mapping"""
         current_offset = offset
         length = len(value)
@@ -318,9 +351,14 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
 
     # ## Scan implementation with knowledge of pages
 
-    def _scan_chunk(self, scanner, min_address, max_address, progress, iterator_value):
+    def _scan_chunk(self,
+                    scanner: 'interfaces.layers.ScannerInterface',
+                    min_address: int,
+                    max_address: int,
+                    progress: ProgressValue,
+                    iterator_value: int) -> typing.List[typing.Any]:
         size_to_scan = min(max_address - min_address, scanner.chunk_size + scanner.overlap)
-        result = []
+        result: typing.List[typing.Any] = []
         for map in self.mapping(iterator_value, size_to_scan, ignore_errors = True):
             offset, mapped_offset, length, layer = map
             progress.value += length
@@ -332,21 +370,28 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
 class Memory(validity.ValidityRoutines, collections.abc.Mapping):
     """Container for multiple layers of data"""
 
-    def __init__(self):
-        self._layers = {}
+    def __init__(self) -> None:
+        self._layers: typing.Dict[str, DataLayerInterface] = {}
 
-    def read(self, layer, offset, length, pad = False):
+    def read(self,
+             layer: str,
+             offset: int,
+             length: int,
+             pad: bool = False):
         """Reads from a particular layer at offset for length bytes
 
            Returns 'bytes' not 'str'
         """
         return self[layer].read(offset, length, pad)
 
-    def write(self, layer, offset, data):
+    def write(self,
+              layer: str,
+              offset: int,
+              data: bytes) -> None:
         """Writes to a particular layer at offset for length bytes"""
         self[layer].write(offset, data)
 
-    def add_layer(self, layer):
+    def add_layer(self, layer: DataLayerInterface) -> None:
         """Adds a layer to memory model
 
            This will throw an exception if the required dependencies are not met
@@ -361,20 +406,20 @@ class Memory(validity.ValidityRoutines, collections.abc.Mapping):
                     "Layer {} has unmet dependencies: {}".format(layer.name, ", ".join(missing_list)))
         self._layers[layer.name] = layer
 
-    def del_layer(self, name):
+    def del_layer(self, name: str) -> None:
         """Removes the layer called name
 
            This will throw an exception if other layers depend upon this layer
         """
         for layer in self._layers:
-            depend_list = [superlayer for superlayer in self._layers if name in superlayer.dependencies]
+            depend_list = [superlayer for superlayer in self._layers if name in self._layers[layer].dependencies]
             if depend_list:
                 raise exceptions.LayerException(
-                    "Layer {} is depended upon: {}".format(layer.name, ", ".join(depend_list)))
+                    "Layer {} is depended upon: {}".format(self._layers[layer].name, ", ".join(depend_list)))
         self._layers[name].destroy()
         del self._layers[name]
 
-    def free_layer_name(self, prefix = "layer"):
+    def free_layer_name(self, prefix: str = "layer") -> str:
         """Returns an unused layer name to ensure no collision occurs when inserting a layer"""
         self._check_type(prefix, str)
 
@@ -383,17 +428,17 @@ class Memory(validity.ValidityRoutines, collections.abc.Mapping):
             count += 1
         return prefix + str(count)
 
-    def __getitem__(self, name):
+    def __getitem__(self, name: str) -> DataLayerInterface:
         """Returns the layer of specified name"""
         return self._layers[name]
 
-    def __len__(self):
+    def __len__(self) -> int:
         return len(self._layers)
 
     def __iter__(self):
         return iter(self._layers)
 
-    def check_cycles(self):
+    def check_cycles(self) -> None:
         """Runs through the available layers and identifies if there are cycles in the DAG"""
         # TODO: Is having a cycle check necessary?
 
@@ -401,5 +446,3 @@ class Memory(validity.ValidityRoutines, collections.abc.Mapping):
 class DummyProgress(object):
     def __init__(self):
         self.value = 0
-
-

volatility/framework/interfaces/objects.py

@@ -4,9 +4,10 @@
 import collections
 import collections.abc
 import logging
+import typing
 from abc import ABCMeta, abstractmethod
 
-from volatility.framework import constants, validity
+from volatility.framework import constants, validity, interfaces
 from volatility.framework.interfaces import context as interfaces_context
 
 vollog = logging.getLogger(__name__)
@@ -18,16 +19,16 @@ 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):
+    def __init__(self, dictionary: typing.ChainMap[str, typing.Any]) -> None:
         self._dict = dictionary
 
-    def __getattr__(self, attr):
+    def __getattr__(self, attr: str) -> typing.Any:
         """Returns the item as an attribute"""
         if attr in self._dict:
             return self._dict[attr]
         raise AttributeError("Object has no attribute: {}.{}".format(self.__class__.__name__, attr))
 
-    def __getitem__(self, name):
+    def __getitem__(self, name: str) -> typing.Any:
         """Returns the item requested"""
         return self._dict[name]
 
@@ -35,7 +36,7 @@ class ReadOnlyMapping(validity.ValidityRoutines, collections.abc.Mapping):
         """Returns an iterator of the dictionary items"""
         return self._dict.__iter__()
 
-    def __len__(self):
+    def __len__(self) -> int:
         """Returns the length of the internal dictionary"""
         return len(self._dict)
 
@@ -63,7 +64,11 @@ class ObjectInformation(ReadOnlyMapping):
 class ObjectInterface(validity.ValidityRoutines, metaclass = ABCMeta):
     """A base object required to be the ancestor of every object used in volatility"""
 
-    def __init__(self, context, type_name, object_info, **kwargs):
+    def __init__(self,
+                 context: 'interfaces_context.ContextInterface',
+                 type_name: str,
+                 object_info: 'ObjectInformation',
+                 **kwargs) -> None:
         # Since objects are likely to be instantiated often,
         # we're only checking that context, offset and parent
         # Everything else may be wrong, but that will get caught later on
@@ -86,22 +91,22 @@ class ObjectInterface(validity.ValidityRoutines, metaclass = ABCMeta):
         self._context = context
 
     @property
-    def vol(self):
+    def vol(self) -> ReadOnlyMapping:
         """Returns the volatility specific object information"""
         # Wrap the outgoing vol in a read-only proxy
         return ReadOnlyMapping(self._vol)
 
     @abstractmethod
-    def write(self, value):
+    def write(self, value: typing.Any):
         """Writes the new value into the format at the offset the object currently resides at"""
 
-    def validate(self):
+    def validate(self) -> bool:
         """A method that can be overridden to validate this object.  It does not return and its return value should not be used.
 
         Raises InvalidDataException on failure to validate the data correctly.
         """
 
-    def get_symbol_table(self):
+    def get_symbol_table(self) -> 'interfaces.symbols.SymbolTableInterface':
         """Returns the symbol table for this particular object
 
         Returns none if the symbol table cannot be identified.
@@ -114,7 +119,9 @@ class ObjectInterface(validity.ValidityRoutines, metaclass = ABCMeta):
             vollog.debug("Symbol table not found in context's symbol_space for symbol: {}".format(self.vol.type_name))
         return self._context.symbol_space[table_name]
 
-    def cast(self, new_type_name, **additional):
+    def cast(self,
+             new_type_name: str,
+             **additional) -> 'ObjectInterface':
         """Returns a new object at the offset and from the layer that the current object inhabits
 
         .. note:: If new type name does not include a symbol table, the symbol table for the current object is used
@@ -143,21 +150,26 @@ class ObjectInterface(validity.ValidityRoutines, metaclass = ABCMeta):
         new templates for each and every potental object type."""
 
         @classmethod
-        def size(cls, template):
+        def size(cls, template: 'Template') -> int:
             """Returns the size of the template object"""
 
         @classmethod
-        def children(cls, template):
+        def children(cls, template: 'Template') -> 'typing.List[Template]':
             """Returns the children of the template"""
             return []
 
         @classmethod
-        def replace_child(cls, template, old_child, new_child):
+        def replace_child(cls,
+                          template: 'Template',
+                          old_child: 'Template',
+                          new_child: 'Template') -> None:
             """Substitutes the old_child for the new_child"""
             raise KeyError("Template does not contain any children to replace: {}".format(template.vol.type_name))
 
         @classmethod
-        def relative_child_offset(cls, template, child):
+        def relative_child_offset(cls,
+                                  template: 'Template',
+                                  child: 'Template') -> int:
             """Returns the relative offset from the head of the parent data to the child member"""
             raise KeyError("Template does not contain any children: {}".format(template.vol.type_name))
 
@@ -185,20 +197,21 @@ class Template(validity.ValidityRoutines):
     constructed at resolution time and then cached.
     """
 
-    def __init__(self, type_name, **arguments):
+    def __init__(self, type_name: str, **arguments) -> None:
         """Stores the keyword arguments for later use"""
         # Allow the updating of template arguments whilst still in template form
         super().__init__()
         self._arguments = arguments
-        self._vol = collections.ChainMap({}, self._arguments, {'type_name': type_name})
+        empty_dict: typing.Dict[str, typing.Any] = {}
+        self._vol = collections.ChainMap(empty_dict, self._arguments, {'type_name': type_name})
 
     @property
-    def vol(self):
+    def vol(self) -> ReadOnlyMapping:
         """Returns a volatility information object, much like the :class:`~volatility.framework.interfaces.objects.ObjectInformation` provides"""
         return ReadOnlyMapping(self._vol)
 
     @property
-    def children(self):
+    def children(self) -> typing.List['Template']:
         """The children of this template (such as member types, sub-types and base-types where they are relevant).
         Used to traverse the template tree.
         """
@@ -206,32 +219,34 @@ class Template(validity.ValidityRoutines):
 
     @property
     @abstractmethod
-    def size(self):
+    def size(self) -> int:
         """Returns the size of the template"""
 
     @abstractmethod
-    def relative_child_offset(self, child):
+    def relative_child_offset(self, child: 'Template') -> int:
         """Returns the relative offset of the `child` member from its parent offset"""
 
     @abstractmethod
-    def replace_child(self, old_child, new_child):
+    def replace_child(self, old_child: 'Template', new_child: 'Template') -> None:
         """Replaces `old_child` with `new_child` in the list of children"""
 
-    def clone(self):
+    def clone(self) -> 'Template':
         """Returns a copy of the original Template as constructed (without `update_vol` additions having been made)"""
         clone = self.__class__(**self._vol.parents.new_child())
         return clone
 
-    def update_vol(self, **new_arguments):
+    def update_vol(self, **new_arguments) -> None:
         """Updates the keyword arguments with values that will **not** be carried across to clones"""
         self._vol.update(new_arguments)
 
-    def __getattr__(self, attr):
+    def __getattr__(self, attr: str) -> typing.Any:
         """Exposes any other values stored in ._vol as attributes (for example, enumeration choices)"""
         if attr != '_vol':
             if attr in self._vol:
                 return self._vol[attr]
         raise AttributeError("{} object has no attribute {}".format(self.__class__.__name__, attr))
 
-    def __call__(self, context, object_info):
+    def __call__(self,
+                 context: 'interfaces_context.ContextInterface',
+                 object_info: ObjectInformation) -> ObjectInterface:
         """Constructs the object"""

volatility/framework/interfaces/plugins.py

@@ -5,6 +5,7 @@ They are called and carry out some algorithms on data stored in layers using obj
 
 # Configuration interfaces must be imported separately, since we're part of interfaces and can't import ourselves
 import logging
+import typing
 from abc import ABCMeta, abstractmethod
 
 from volatility.framework import exceptions
@@ -13,6 +14,9 @@ from volatility.framework.interfaces import configuration as interfaces_configur
 
 vollog = logging.getLogger(__name__)
 
+if typing.TYPE_CHECKING:
+    from volatility.framework import interfaces, renderers
+
 
 #
 # Plugins
@@ -34,7 +38,9 @@ class PluginInterface(interfaces_configuration.ConfigurableInterface, validity.V
     context it is passed.
     """
 
-    def __init__(self, context, config_path):
+    def __init__(self,
+                 context: 'interfaces.context.ContextInterface',
+                 config_path: str) -> None:
         super().__init__(context, config_path)
         # Plugins self validate on construction, it makes it more difficult to work with them, but then
         # the validation doesn't need to be repeated over and over again by externals
@@ -43,12 +49,12 @@ class PluginInterface(interfaces_configuration.ConfigurableInterface, validity.V
             raise exceptions.PluginRequirementException("The plugin configuration failed to validate")
 
     @classmethod
-    def get_requirements(cls):
+    def get_requirements(cls) -> typing.List['interfaces.configuration.RequirementInterface']:
         """Returns a list of Requirement objects for this plugin"""
         return []
 
     @abstractmethod
-    def run(self):
+    def run(self) -> 'renderers.TreeGrid':
         """Executes the functionality of the code
 
         .. note:: This method expects `self.validate` to have been called to ensure all necessary options have been provided

volatility/framework/interfaces/renderers.py

@@ -3,34 +3,37 @@ or in some other form.  This module defines both the output format (:class:`Tree
 which can interact with a TreeGrid to produce suitable output."""
 
 import collections
+import typing
 from abc import abstractmethod, ABCMeta
 
 from volatility.framework import validity
 
 Column = collections.namedtuple('Column', ['index', 'name', 'type'])
 
+RenderOption = typing.Any
+
 
 class Renderer(validity.ValidityRoutines, metaclass = ABCMeta):
     """Class that defines the interface that all output renderers must support"""
 
-    def __init__(self, options):
+    def __init__(self, options: typing.List[RenderOption]) -> None:
         """Accepts an options object to configure the renderers"""
         # FIXME: Once the config option objects are in place, put the _type_check in place
 
     @abstractmethod
-    def get_render_options(self):
+    def get_render_options(self) -> typing.List[RenderOption]:
         """Returns a list of rendering options"""
 
     @abstractmethod
-    def render(self, grid):
+    def render(self, grid: 'TreeGrid') -> None:
         """Takes a grid object and renders it based on the object's preferences"""
 
 
 class ColumnSortKey(metaclass = ABCMeta):
-    ascending = True
+    ascending: bool = True
 
     @abstractmethod
-    def __call__(self, values):
+    def __call__(self, values: typing.List[typing.Any]) -> typing.Any:
         """The key function passed as a sort key to the TreeGrid's visit function"""
 
 
@@ -40,12 +43,12 @@ class TreeNode(collections.Sequence, metaclass = ABCMeta):
 
     @property
     @abstractmethod
-    def values(self):
+    def values(self) -> typing.Any:
         """Returns the list of values from the particular node, based on column.index"""
 
     @property
     @abstractmethod
-    def path(self):
+    def path(self) -> str:
         """Returns a path identifying string
 
         This should be seen as opaque by external classes,
@@ -54,22 +57,27 @@ class TreeNode(collections.Sequence, metaclass = ABCMeta):
 
     @property
     @abstractmethod
-    def parent(self):
+    def parent(self) -> 'TreeNode':
         """Returns the parent node of this node or None"""
 
     @property
     @abstractmethod
-    def path_depth(self):
+    def path_depth(self) -> int:
         """Return the path depth of the current node"""
 
     @abstractmethod
-    def path_changed(self, path, added = False):
+    def path_changed(self, path: str, added: bool = False) -> None:
         """Updates the path based on the addition or removal of a node higher up in the tree
 
            This should only be called by the containing TreeGrid and expects to only be called for affected nodes.
         """
 
 
+ColumnsType = typing.List[typing.Tuple[str, typing.Type]]
+SimpleTypes = typing.Union[int, str, float, bytes]
+_T = typing.TypeVar("_T")
+
+
 class TreeGrid(object, metaclass = ABCMeta):
     """Class providing the interface for a TreeGrid (which contains TreeNodes)
 
@@ -83,9 +91,9 @@ class TreeGrid(object, metaclass = ABCMeta):
     and to create cycles.
     """
 
-    simple_types = {int, str, float, bytes}
+    simple_types: typing.ClassVar[typing.Set[typing.Type]] = {int, str, float, bytes}
 
-    def __init__(self, columns, generator):
+    def __init__(self, columns: ColumnsType, generator: typing.Generator) -> None:
         """Constructs a TreeGrid object using a specific set of columns
 
         The TreeGrid itself is a root element, that can have children but no values.
@@ -97,7 +105,10 @@ class TreeGrid(object, metaclass = ABCMeta):
         """
 
     @abstractmethod
-    def populate(self, func = None, initial_accumulator = None):
+    def populate(self,
+                 func: typing.Callable[[typing.Tuple[SimpleTypes]], TreeNode] = None,
+                 initial_accumulator: typing.Any = None) \
+            -> typing.Generator[typing.Tuple[SimpleTypes, ...], None, None]:
         """Generator that returns the next available Node
 
            This is equivalent to a one-time visit.
@@ -105,44 +116,48 @@ class TreeGrid(object, metaclass = ABCMeta):
 
     @property
     @abstractmethod
-    def populated(self):
+    def populated(self) -> bool:
         """Indicates that population has completed and the tree may now be manipulated separately"""
 
     @property
     @abstractmethod
-    def columns(self):
+    def columns(self) -> ColumnsType:
         """Returns the available columns and their ordering and types"""
 
     @abstractmethod
-    def children(self, node):
+    def children(self, node: TreeNode) -> typing.List[TreeNode]:
         """Returns the subnodes of a particular node in order"""
 
     @abstractmethod
-    def values(self, node):
+    def values(self, node: TreeNode) -> typing.Tuple[SimpleTypes, ...]:
         """Returns the values for a particular node
 
            The values returned are mutable,
         """
 
     @abstractmethod
-    def is_ancestor(self, node, descendant):
+    def is_ancestor(self, node: TreeNode, descendant: TreeNode) -> bool:
         """Returns true if descendent is a child, grandchild, etc of node"""
 
     @abstractmethod
-    def max_depth(self):
+    def max_depth(self) -> int:
         """Returns the maximum depth of the tree"""
 
     @staticmethod
-    def path_depth(node):
+    def path_depth(node: TreeNode) -> int:
         """Returns the path depth of a particular node"""
         return node.path_depth
 
-    def path_is_valid(self, node):
+    def path_is_valid(self, node: TreeNode) -> bool:
         """Returns True is a given path is valid for this treegrid"""
         return node in self.children(node.parent)
 
     @abstractmethod
-    def visit(self, node, function, initial_accumulator = None, sort_key = None):
+    def visit(self,
+              node: TreeNode,
+              function: typing.Callable[[TreeNode, _T], _T],
+              initial_accumulator: _T = None,
+              sort_key: ColumnSortKey = None) -> None:
         """Visits all the nodes in a tree, calling function on each one.
 
            function should have the signature function(node, accumulator) and return new_accumulator

volatility/framework/interfaces/symbols.py

@@ -2,16 +2,24 @@
 """
 import bisect
 import collections.abc
+import typing
 from abc import abstractmethod
 
 from volatility.framework import constants, exceptions, validity
 from volatility.framework.interfaces import configuration, objects
 
+if typing.TYPE_CHECKING:
+    from volatility.framework import interfaces
+
 
 class Symbol(validity.ValidityRoutines):
     """Contains information about a named location in a program's memory"""
 
-    def __init__(self, name, address, type = None, constant_data = None):
+    def __init__(self,
+                 name: str,
+                 address: int,
+                 type: typing.Optional[objects.Template] = None,
+                 constant_data: typing.Optional[bytes] = None) -> None:
         self._name = self._check_type(name, str)
         if constants.BANG in self._name:
             raise ValueError("Symbol names cannot contain the symbol differentiator ({})".format(constants.BANG))
@@ -29,61 +37,66 @@ class Symbol(validity.ValidityRoutines):
             self._constant_data = self._check_type(constant_data, bytes)
 
     @property
-    def name(self):
+    def name(self) -> str:
         """Returns the name of the symbol"""
         return self._name
 
     @property
-    def type(self):
+    def type_name(self) -> typing.Optional[str]:
         """Returns the name of the type that the symbol represents"""
+        return self.type.name
+
+    @property
+    def type(self) -> typing.Optional[objects.Template]:
+        """Returns the type that the symbol represents"""
         return self._type
 
     @property
-    def address(self):
+    def address(self) -> int:
         """Returns the relative address of the symbol within the compilation unit"""
         return self._address
 
     @property
-    def constant_data(self):
+    def constant_data(self) -> typing.Optional[bytes]:
         return self._constant_data
 
 
 class SymbolSpaceInterface(collections.abc.Mapping):
     """An interface for the container that holds all the symbol-containing tables for use within a context"""
 
-    def free_table_name(self, prefix = "layer"):
+    def free_table_name(self, prefix: str = "layer") -> str:
         """Returns an unused table name to ensure no collision occurs when inserting a symbol table"""
 
     @abstractmethod
-    def get_symbols_by_type(self, type_name):
+    def get_symbols_by_type(self, type_name: str) -> typing.List[Symbol]:
         """Returns all symbols based on the type of the symbol"""
 
     @abstractmethod
-    def get_symbols_by_location(self, address, table_name = None):
+    def get_symbols_by_location(self, address: int, table_name: typing.Optional[str] = None) -> typing.List[Symbol]:
         """Returns all symbols that exist at a specific relative address"""
 
     @abstractmethod
-    def get_type(self, type_name):
+    def get_type(self, type_name: str) -> objects.Template:
         """Look-up a type name across all the contained symbol tables"""
 
     @abstractmethod
-    def get_symbol(self, symbol_name):
+    def get_symbol(self, symbol_name: str) -> Symbol:
         """Look-up a symbol name across all the contained symbol tables"""
 
     @abstractmethod
-    def get_enumeration(self, enum_name):
+    def get_enumeration(self, enum_name: str) -> typing.Dict[str, typing.Any]:
         """Look-up an enumeration across all the contained symbol tables"""
 
     @abstractmethod
-    def has_type(self, name):
+    def has_type(self, name: str) -> bool:
         """Determines whether a type exists in the contained symbol tables"""
 
     @abstractmethod
-    def has_symbol(self, name):
+    def has_symbol(self, name: str) -> bool:
         """Determines whether a symbol exists in the contained symbol tables"""
 
     @abstractmethod
-    def has_enumeration(self, name):
+    def has_enumeration(self, name: str) -> bool:
         """Determines whether an enumeration choice exists in the contained symbol tables"""
 
 
@@ -96,7 +109,10 @@ class BaseSymbolTableInterface(validity.ValidityRoutines):
     Note: table_mapping is a rarely used feature (since symbol tables are typically self-contained)
     """
 
-    def __init__(self, name, native_types = None, table_mapping = None):
+    def __init__(self,
+                 name: str,
+                 native_types: typing.Optional['NativeTableInterface'] = None,
+                 table_mapping: typing.Optional[typing.Dict[str, str]] = None) -> None:
         if name:
             self._check_type(name, str)
         self.name = name or None
@@ -107,7 +123,7 @@ class BaseSymbolTableInterface(validity.ValidityRoutines):
 
     # ## Required Symbol functions
 
-    def get_symbol(self, name):
+    def get_symbol(self, name: str) -> Symbol:
         """Resolves a symbol name into a symbol object
 
            If the symbol isn't found, it raises a SymbolError exception
@@ -115,18 +131,18 @@ class BaseSymbolTableInterface(validity.ValidityRoutines):
         raise NotImplementedError("Abstract property get_symbol not implemented by subclass.")
 
     @property
-    def symbols(self):
+    def symbols(self) -> typing.Iterable[str]:
         """Returns an iterator of the Symbol names"""
         raise NotImplementedError("Abstract property symbols not implemented by subclass.")
 
     # ## Required Type functions
 
     @property
-    def types(self):
+    def types(self) -> typing.Iterable[str]:
         """Returns an iterator of the Symbol type names"""
         raise NotImplementedError("Abstract property types not implemented by subclass.")
 
-    def get_type(self, name):
+    def get_type(self, name: str) -> objects.Template:
         """Resolves a symbol name into an object template
 
            If the symbol isn't found it raises a SymbolError exception
@@ -136,19 +152,19 @@ class BaseSymbolTableInterface(validity.ValidityRoutines):
     # ## Required Symbol enumeration functions
 
     @property
-    def enumerations(self):
+    def enumerations(self) -> typing.Iterable[typing.Any]:
         """Returns an iterator of the Enumeration names"""
         raise NotImplementedError("Abstract property enumerations not implemented by subclass.")
 
     # ## Native Type Handler
 
     @property
-    def natives(self):
+    def natives(self) -> 'NativeTableInterface':
         """Returns None or a NativeTable for handling space specific native types"""
         return self._native_types
 
     @natives.setter
-    def natives(self, value):
+    def natives(self, value: 'NativeTableInterface') -> None:
         """Checks the natives value and then applies it internally
 
            WARNING: This allows changing the underlying size of all the other types referenced in the SymbolTable
@@ -158,28 +174,28 @@ class BaseSymbolTableInterface(validity.ValidityRoutines):
 
     # ## Functions for overriding classes
 
-    def set_type_class(self, name, clazz):
+    def set_type_class(self, name: str, clazz: 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):
+    def get_type_class(self, name: str) -> objects.ObjectInterface:
         """Returns the class associated with a Symbol type"""
         raise NotImplementedError("Abstract method get_type_class not implemented yet.")
 
-    def del_type_class(self, name):
+    def del_type_class(self, name: str) -> None:
         """Removes the associated class override for a specific Symbol type"""
         raise NotImplementedError("Abstract method del_type_class not implemented yet.")
 
     # ## Convenience functions for location symbols
 
-    def get_symbol_type(self, name):
+    def get_symbol_type(self, name: str) -> objects.Template:
         """Resolves a symbol name into a symbol and then resolves the symbol's type"""
         return self.get_type(self.get_symbol(name).type_name)
 
-    def get_symbols_by_type(self, type_name):
+    def get_symbols_by_type(self, type_name: str) -> typing.Generator[str, None, None]:
         """Returns the name of all symbols in this table that have type matching type_name"""
         for symbol_name in self.symbols:
             # This allows for searching with and without the table name (in case multiple tables contain
@@ -188,7 +204,7 @@ class BaseSymbolTableInterface(validity.ValidityRoutines):
             if symbol.type_name == type_name or (symbol.type_name.endswith(constants.BANG + type_name)):
                 yield symbol.name
 
-    def get_symbols_by_location(self, offset):
+    def get_symbols_by_location(self, offset: int) -> typing.Generator[str, None, None]:
         """Returns the name of all symbols in this table that live at a particular offset"""
         sort_symbols = sorted([(self.get_symbol(sn).address, sn) for sn in self.symbols])
         result = bisect.bisect_left(sort_symbols, (offset, ""))
@@ -200,11 +216,15 @@ class BaseSymbolTableInterface(validity.ValidityRoutines):
 class SymbolTableInterface(BaseSymbolTableInterface, configuration.ConfigurableInterface):
     """Handles a table of symbols"""
 
-    def __init__(self, context, config_path, name, native_types = None):
+    def __init__(self,
+                 context: 'interfaces.context.ContextInterface',
+                 config_path: str,
+                 name: str,
+                 native_types: 'NativeTableInterface' = None) -> None:
         configuration.ConfigurableInterface.__init__(self, context, config_path)
         BaseSymbolTableInterface.__init__(self, name, native_types)
 
-    def build_configuration(self):
+    def build_configuration(self) -> 'interfaces.configuration.HierarchicalDict':
         config = super().build_configuration()
 
         # Translation Layers are constructable, and therefore require a class configuration variable
@@ -215,16 +235,16 @@ class SymbolTableInterface(BaseSymbolTableInterface, configuration.ConfigurableI
 class NativeTableInterface(BaseSymbolTableInterface):
     """Class to distinguish NativeSymbolLists from other symbol lists"""
 
-    def get_symbol(self, name):
+    def get_symbol(self, name: str):
         raise exceptions.SymbolError("NativeTables never hold symbols")
 
     @property
-    def symbols(self):
+    def symbols(self) -> typing.List[str]:
         return []
 
-    def get_enumeration(self, name):
+    def get_enumeration(self, name: str):
         raise exceptions.SymbolError("NativeTables never hold enumerations")
 
     @property
-    def enumerations(self):
+    def enumerations(self) -> typing.List[str]:
         return []

volatility/framework/objects/templates.py

@@ -1,4 +1,5 @@
 import logging
+import typing
 
 from volatility.framework import interfaces, validity, exceptions
 
@@ -66,15 +67,16 @@ class ReferenceTemplate(interfaces.objects.Template):
     def children(self):
         return []
 
-    def _unresolved(self, *args, **kwargs):
+    def _unresolved(self, *args, **kwargs) -> typing.Any:
         """Referenced symbols must be appropriately resolved before they can provide information such as size
            This is because the size request has no context within which to determine the actual symbol structure.
         """
         raise exceptions.SymbolError(
             "Template contains no information about its structure: {}".format(self.vol.type_name))
 
-    size = property(_unresolved)
-    replace_child = relative_child_offset = _unresolved
+    size: typing.ClassVar[typing.Any] = property(_unresolved)
+    replace_child: typing.ClassVar[typing.Any] = _unresolved
+    relative_child_offset: typing.ClassVar[typing.Any] = _unresolved
 
     def __call__(self, context, object_info):
         template = context.symbol_space.get_type(self.vol.type_name)

volatility/framework/renderers/__init__.py

@@ -3,6 +3,7 @@
 Renderers display the unified output format in some manner (be it text or file or graphical output"""
 
 import collections
+import typing
 
 from volatility.framework import interfaces
 
@@ -226,7 +227,13 @@ class TreeGrid(interfaces.renderers.TreeGrid):
         """Returns the maximum depth of the tree"""
         return self.visit(None, lambda n, a: max(a, self.path_depth(n)), )
 
-    def visit(self, node, function, initial_accumulator = None, sort_key = None):
+    _T = typing.TypeVar("_T")
+
+    def visit(self,
+              node: interfaces.renderers.TreeNode,
+              function: typing.Callable[[interfaces.renderers.TreeNode, _T], _T],
+              initial_accumulator: _T = None,
+              sort_key: interfaces.renderers.ColumnSortKey = None):
         """Visits all the nodes in a tree, calling function on each one.
 
            function should have the signature function(node, accumulator) and return new_accumulator
@@ -257,7 +264,11 @@ class TreeGrid(interfaces.renderers.TreeGrid):
             accumulator = self._visit(children, function, accumulator, sort_key)
         return accumulator
 
-    def _visit(self, list_of_children, function, accumulator, sort_key = None):
+    def _visit(self,
+               list_of_children: typing.List['TreeNode'],
+               function: typing.Callable,
+               accumulator: _T,
+               sort_key: interfaces.renderers.ColumnSortKey = None) -> _T:
         """Visits all the nodes in a tree, calling function on each one"""
         if list_of_children is not None:
             for n, children in list_of_children:
@@ -271,7 +282,7 @@ class TreeGrid(interfaces.renderers.TreeGrid):
 
 
 class ColumnSortKey(interfaces.renderers.ColumnSortKey):
-    def __init__(self, treegrid, column_name, ascending = True):
+    def __init__(self, treegrid: TreeGrid, column_name: str, ascending: bool = True) -> None:
         self._index = None
         self.ascending = ascending
         for i in treegrid.columns:
@@ -280,6 +291,6 @@ class ColumnSortKey(interfaces.renderers.ColumnSortKey):
         if self._index is None:
             raise ValueError("Column not found in TreeGrid columns: {}".format(column_name))
 
-    def __call__(self, values):
+    def __call__(self, values: typing.List[typing.Any]) -> typing.Any:
         """The key function passed as the sort key"""
         return values[self._index]

volatility/framework/validity.py

@@ -2,7 +2,7 @@
 """
 import typing
 
-ProgressCallback = typing.Optional[typing.Callable[[int, str], None]]
+ProgressCallback = typing.Optional[typing.Callable[[float, str], None]]
 
 
 class ValidityRoutines(object):