From 623180ddbda404addc5d46b5e29f93ae233ae446 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Sun, 22 Apr 2018 20:45:59 +0100 Subject: [PATCH] Make many typing fixes, based on mypy-0.590. --- volatility/framework/__init__.py | 2 +- volatility/framework/automagic/__init__.py | 2 +- .../framework/automagic/construct_layers.py | 2 +- volatility/framework/automagic/pdbscan.py | 21 +++++++++---------- volatility/framework/automagic/stacker.py | 2 +- volatility/framework/automagic/windows.py | 5 ++++- volatility/framework/contexts/__init__.py | 1 + volatility/framework/interfaces/automagic.py | 19 +++++++++-------- .../framework/interfaces/configuration.py | 7 +++---- volatility/framework/interfaces/layers.py | 7 +++---- volatility/framework/interfaces/plugins.py | 2 +- volatility/framework/interfaces/renderers.py | 7 +++---- volatility/framework/layers/vmware.py | 5 +++-- volatility/framework/objects/utility.py | 7 +++---- volatility/framework/renderers/__init__.py | 3 +-- .../symbols/linux/extensions/__init__.py | 3 +++ .../symbols/windows/extensions/__init__.py | 13 +++++++++--- .../symbols/windows/extensions/registry.py | 7 ++++++- 18 files changed, 65 insertions(+), 50 deletions(-) diff --git a/volatility/framework/__init__.py b/volatility/framework/__init__.py index bebd2ca29..541fce32c 100644 --- a/volatility/framework/__init__.py +++ b/volatility/framework/__init__.py @@ -67,7 +67,7 @@ def hide_from_subclasses(cls: typing.Type) -> typing.Type: return cls -T = typing.TypeVar('T', typing.Type, typing.Type) +T = typing.TypeVar('T', bound = typing.Type) def class_subclasses(cls: T) -> typing.Generator[T, None, None]: diff --git a/volatility/framework/automagic/__init__.py b/volatility/framework/automagic/__init__.py index 8da855edb..528cfd969 100644 --- a/volatility/framework/automagic/__init__.py +++ b/volatility/framework/automagic/__init__.py @@ -32,7 +32,7 @@ linux_automagic = ['ConstructionMagic', def available(context: interfaces.context.ContextInterface) \ - -> typing.List[typing.Type[interfaces.automagic.AutomagicInterface]]: + -> typing.List[interfaces.automagic.AutomagicInterface]: """Returns an ordered list of all subclasses of :class:`~volatility.framework.interfaces.automagic.AutomagicInterface`. The order is based on the priority attributes of the subclasses, in order to ensure the automagics are listed in diff --git a/volatility/framework/automagic/construct_layers.py b/volatility/framework/automagic/construct_layers.py index 267481bd4..7eee0605d 100644 --- a/volatility/framework/automagic/construct_layers.py +++ b/volatility/framework/automagic/construct_layers.py @@ -26,7 +26,7 @@ class ConstructionMagic(interfaces.automagic.AutomagicInterface): config_path: str, requirement: interfaces.configuration.RequirementInterface, progress_callback = None, optional = False) -> typing.List[str]: - result = [] + result = [] # type: typing.List[str] if requirement.unsatisfied(context, config_path): # Having called validate at the top level tells us both that we need to dig deeper # but also ensures that TranslationLayerRequirements have got the correct subrequirements if their class is populated diff --git a/volatility/framework/automagic/pdbscan.py b/volatility/framework/automagic/pdbscan.py index 8d040d0e4..8fac8ec82 100644 --- a/volatility/framework/automagic/pdbscan.py +++ b/volatility/framework/automagic/pdbscan.py @@ -11,7 +11,7 @@ import struct import typing from volatility.framework import exceptions, layers, validity -from volatility.framework.layers import scanners +from volatility.framework.layers import scanners, intel from volatility.framework.symbols import intermed, native if __name__ == "__main__": @@ -165,25 +165,24 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): virtual_layer_name = context.config.get(sub_config_path, None) layer_name = context.config.get(interfaces.configuration.path_join(sub_config_path, "memory_layer"), None) if layer_name and virtual_layer_name: - page_size = context.memory[virtual_layer_name].page_size - results = {virtual_layer_name: scan(context, - layer_name, - page_size, - progress_callback = progress_callback)} + memlayer = context.memory[virtual_layer_name] + if isinstance(memlayer, intel.Intel): + page_size = memlayer.page_size + results = {virtual_layer_name: scan(context, + layer_name, + page_size, + progress_callback = progress_callback)} else: for subreq in requirement.requirements.values(): results.update(self.recurse_pdb_finder(context, sub_config_path, subreq)) return results - def recurse_symbol_fulfiller(self, - context: interfaces.context.ContextInterface) \ - -> None: + def recurse_symbol_fulfiller(self, context: interfaces.context.ContextInterface) -> None: """Fulfills the SymbolRequirements in `self._symbol_requirements` found by the `recurse_symbol_requirements`. This pass will construct any requirements that may need it in the context it was passed :param context: Context on which to operate - :type context: ~volatility.framework.interfaces.context.ContextInterface """ join = interfaces.configuration.path_join for config_path, sub_config_path, requirement in self._symbol_requirements: @@ -323,7 +322,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): def __call__(self, context: interfaces.context.ContextInterface, config_path: str, - requirement: interfaces.configuration.ConstructableRequirementInterface, + requirement: interfaces.configuration.RequirementInterface, progress_callback: validity.ProgressCallback = None) -> None: if requirement.unsatisfied(context, config_path): if "pdbscan" not in context.symbol_space: diff --git a/volatility/framework/automagic/stacker.py b/volatility/framework/automagic/stacker.py index f0b1f0279..0780e244d 100644 --- a/volatility/framework/automagic/stacker.py +++ b/volatility/framework/automagic/stacker.py @@ -42,7 +42,7 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): def __call__(self, context: interfaces.context.ContextInterface, config_path: str, - requirement: interfaces.configuration.ConstructableRequirementInterface, + requirement: interfaces.configuration.RequirementInterface, progress_callback: validity.ProgressCallback = None) \ -> typing.Optional[typing.List[str]]: """Runs the automagic over the configurable""" diff --git a/volatility/framework/automagic/windows.py b/volatility/framework/automagic/windows.py index e2b4c1a03..5f0da6ae8 100644 --- a/volatility/framework/automagic/windows.py +++ b/volatility/framework/automagic/windows.py @@ -264,6 +264,8 @@ class WintelHelper(interfaces.automagic.AutomagicInterface): # Only bother getting the DTB if we don't already have one if not context.config.get(interfaces.configuration.path_join(sub_config_path, "page_map_offset"), None): physical_layer = requirement.requirements["memory_layer"].config_value(context, sub_config_path) + if not isinstance(physical_layer, str): + raise TypeError("Physical layer name is not a string: {}".format(sub_config_path)) hits = context.memory[physical_layer].scan(context, PageMapScanner(useful), progress_callback) for test, dtb in hits: context.config[interfaces.configuration.path_join(sub_config_path, "page_map_offset")] = dtb @@ -282,7 +284,8 @@ class WintelStacker(interfaces.automagic.StackerLayerInterface): def stack(cls, context: interfaces.context.ContextInterface, layer_name: str, - progress_callback: validity.ProgressCallback = None) -> typing.Optional[str]: + progress_callback: validity.ProgressCallback = None) \ + -> typing.Optional[interfaces.layers.DataLayerInterface]: """Attempts to determine and stack an intel layer on a physical layer where possible Where the DTB scan fails, it attempts a heuristic of checking for the DTB within a specific range. diff --git a/volatility/framework/contexts/__init__.py b/volatility/framework/contexts/__init__.py index 48c983e47..29221b05e 100644 --- a/volatility/framework/contexts/__init__.py +++ b/volatility/framework/contexts/__init__.py @@ -138,6 +138,7 @@ class Module(interfaces.context.Module): """ symbol_type = symbol_name and not (type_name or offset) type_type = (type_name and offset) and not symbol_name + type_arg = None # type: typing.Optional[typing.Union[str, interfaces.objects.Template]] if symbol_type and type_type or not (symbol_type or type_type): raise ValueError("One of symbol_name, or type_name & offset, must be specified to construct a module") if symbol_type is not None: diff --git a/volatility/framework/interfaces/automagic.py b/volatility/framework/interfaces/automagic.py index 311db1ccf..7580084ae 100644 --- a/volatility/framework/interfaces/automagic.py +++ b/volatility/framework/interfaces/automagic.py @@ -3,13 +3,13 @@ Automagic objects attempt to automatically fill configuration values that a user has not filled. """ import typing -from abc import ABCMeta, abstractmethod +from abc import ABCMeta import volatility.framework.configuration.requirements from volatility.framework import validity, interfaces from volatility.framework.interfaces import configuration as interfaces_configuration -RequirementInterfaceType = typing.Type[interfaces.configuration.RequirementInterface] +R = typing.TypeVar('R', bound = interfaces.configuration.RequirementInterface) class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metaclass = ABCMeta): @@ -47,22 +47,24 @@ class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metacla raise ValueError( "Automagic requirements must be an InstanceRequirement, ChoiceRequirement or ListRequirement") - @abstractmethod def __call__(self, context: interfaces.context.ContextInterface, config_path: str, - configurable: interfaces.configuration.ConfigurableInterface, + requirement: interfaces.configuration.RequirementInterface, progress_callback: validity.ProgressCallback = None) -> typing.List[str]: """Runs the automagic over the configurable""" + return [] + + # TODO: requirement_type can be made typing.Union[typing.Type[T], typing.Tuple[typing.Type[T], ...]] + # once mypy properly supports Tuples in instance def find_requirements(self, context: interfaces.context.ContextInterface, config_path: str, requirement_root: interfaces.configuration.RequirementInterface, - requirement_type: typing.Union[RequirementInterfaceType, - typing.Tuple[RequirementInterfaceType, ...]], + requirement_type: typing.Type[R], shortcut: bool = True) \ - -> typing.List[typing.Tuple[str, str, interfaces_configuration.ConstructableRequirementInterface]]: + -> typing.List[typing.Tuple[str, str, R]]: """Determines if there is actually an unfulfilled requirement waiting This ensures we do not carry out an expensive search when there is no requirement for a particular requirement @@ -75,7 +77,7 @@ class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metacla :return: A list of tuples containing the config_path, sub_config_path and requirement identifying the SymbolRequirements """ sub_config_path = interfaces_configuration.path_join(config_path, requirement_root.name) - results = [] + results = [] # type: typing.List[typing.Tuple[str, str, R]] recurse = not shortcut if isinstance(requirement_root, requirement_type): if recurse or requirement_root.unsatisfied(context, config_path): @@ -98,7 +100,6 @@ class StackerLayerInterface(validity.ValidityRoutines, metaclass = ABCMeta): stack_order = 0 @classmethod - @abstractmethod def stack(self, context: interfaces.context.ContextInterface, layer_name: str, diff --git a/volatility/framework/interfaces/configuration.py b/volatility/framework/interfaces/configuration.py index 09d3d134f..e593f7ac3 100644 --- a/volatility/framework/interfaces/configuration.py +++ b/volatility/framework/interfaces/configuration.py @@ -8,6 +8,7 @@ provide configurations values that fulfill those requirements. Where the user d values, automagic modules may extend the configuration tree themselves. """ +import collections.abc import copy import json import logging @@ -17,8 +18,6 @@ import sys import typing from abc import ABCMeta, abstractmethod -import collections.abc - from volatility.framework import constants, interfaces from volatility.framework import validity from volatility.framework.interfaces.context import ContextInterface @@ -631,7 +630,7 @@ class TranslationLayerRequirement(ConstructableRequirementInterface, Configurabl return None obj = self._construct_class(context, config_path, args) - if obj is not None: + if obj is not None and isinstance(obj, interfaces.layers.DataLayerInterface): context.add_layer(obj) # This should already be done by the _construct_class method # context.config[config_path] = obj.name @@ -687,7 +686,7 @@ class SymbolRequirement(ConstructableRequirementInterface, ConfigurableRequireme args[req.name] = node_config.data[req.name] obj = self._construct_class(context, config_path, args) - if obj is not None: + if obj is not None and isinstance(obj, interfaces.symbols.SymbolTableInterface): context.symbol_space.append(obj) return None diff --git a/volatility/framework/interfaces/layers.py b/volatility/framework/interfaces/layers.py index 5c419405c..edba0511b 100644 --- a/volatility/framework/interfaces/layers.py +++ b/volatility/framework/interfaces/layers.py @@ -1,5 +1,6 @@ """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.abc import functools import logging import math @@ -8,8 +9,6 @@ import traceback import typing from abc import ABCMeta, abstractmethod -import collections.abc - from volatility.framework import constants, exceptions, validity, interfaces from volatility.framework.interfaces import configuration, context @@ -53,7 +52,7 @@ class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta): 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._context = None # type: typing.Optional[interfaces.context.ContextInterface] self._layer_name = None # type: typing.Optional[str] @property @@ -348,7 +347,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta): current_offset)) elif offset < current_offset: raise exceptions.LayerException("Mapping returned an overlapping element") - self._context.memory.write(layer, mapped_offset, length) + self._context.memory.write(layer, mapped_offset, value) current_offset += length # ## Scan implementation with knowledge of pages diff --git a/volatility/framework/interfaces/plugins.py b/volatility/framework/interfaces/plugins.py index 559225bc7..fc0c7b9b6 100644 --- a/volatility/framework/interfaces/plugins.py +++ b/volatility/framework/interfaces/plugins.py @@ -22,7 +22,7 @@ if typing.TYPE_CHECKING: class FileInterface(validity.ValidityRoutines, metaclass = ABCMeta): """Class for storing Files in the plugin as a means to output a file or files when necessary""" - def __init__(self, filename: str, data: bytes = None): + def __init__(self, filename: str, data: bytes = None) -> None: self.preferred_filename = filename self.data = io.BytesIO(data) diff --git a/volatility/framework/interfaces/renderers.py b/volatility/framework/interfaces/renderers.py index 3d4f66b96..5594da6c4 100644 --- a/volatility/framework/interfaces/renderers.py +++ b/volatility/framework/interfaces/renderers.py @@ -2,12 +2,11 @@ or in some other form. This module defines both the output format (:class:`TreeGrid`) and the renderer interface which can interact with a TreeGrid to produce suitable output.""" +import collections import datetime import typing from abc import abstractmethod, ABCMeta -import collections - from volatility.framework import validity Column = collections.namedtuple('Column', ['index', 'name', 'type']) @@ -83,7 +82,7 @@ class Disassembly(object): """A class to indicate that the bytes provided should be disassembled (based on the architecture)""" possible_architectures = ['intel', 'intel64', 'arm', 'arm64'] - def __init__(self, data: bytes, offset: int = 0, architecture: str = 'intel64'): + def __init__(self, data: bytes, offset: int = 0, architecture: str = 'intel64') -> None: self.data = data self.architecture = None if architecture in self.possible_architectures: @@ -96,7 +95,7 @@ class Disassembly(object): # We don't class these off a shared base, because the BaseTypes must only # contain the types that the validator will accept (which would not include the base) -_Type = typing.TypeVar("_Type") +_Type = typing.TypeVar("_Type", bound = typing.Type) ColumnsType = typing.List[typing.Tuple[str, typing.Type]] BaseTypes = typing.Union[typing.Type[int], typing.Type[str], diff --git a/volatility/framework/layers/vmware.py b/volatility/framework/layers/vmware.py index 27eb48792..af37cf4d7 100644 --- a/volatility/framework/layers/vmware.py +++ b/volatility/framework/layers/vmware.py @@ -111,9 +111,10 @@ class VmwareStacker(interfaces.automagic.StackerLayerInterface): 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): + memlayer = context.memory[layer_name] + if not isinstance(memlayer, physical.FileLayer): return None - location = context.memory[layer_name].location + location = memlayer.location if location.endswith(".vmem"): vmss = location[:-5] + ".vmss" vmsn = location[:-5] + ".vmsn" diff --git a/volatility/framework/objects/utility.py b/volatility/framework/objects/utility.py index fc70c1ce4..a63b36b2a 100644 --- a/volatility/framework/objects/utility.py +++ b/volatility/framework/objects/utility.py @@ -2,7 +2,6 @@ import datetime import typing from volatility.framework import interfaces, objects, renderers -from volatility.framework.objects import templates def array_to_string(array: objects.Array, @@ -31,13 +30,13 @@ def pointer_to_string(pointer: objects.Pointer, def array_of_pointers(array: interfaces.objects.ObjectInterface, count: int, - subtype: typing.Optional[typing.Union[str, templates.ObjectTemplate]] = None, + subtype: typing.Optional[typing.Union[str, interfaces.objects.Template]] = None, context: interfaces.context.ContextInterface = None) -> interfaces.objects.ObjectInterface: """Takes an object, and recasts it as an array of pointers to subtype""" if isinstance(subtype, str) and context is not None: subtype = context.symbol_space.get_type(subtype) - if not isinstance(subtype, templates.ObjectTemplate) or subtype is None: - raise TypeError("Subtype must be a valid object template (or string name of an object template)") + if not isinstance(subtype, interfaces.objects.Template) or subtype is None: + raise TypeError("Subtype must be a valid template (or string name of an object template)") subtype_pointer = objects.templates.ObjectTemplate(objects.Pointer, type_name = 'pointer', subtype = subtype) return array.cast("array", count = count, subtype = subtype_pointer) diff --git a/volatility/framework/renderers/__init__.py b/volatility/framework/renderers/__init__.py index 5783d9f74..833f69cb6 100644 --- a/volatility/framework/renderers/__init__.py +++ b/volatility/framework/renderers/__init__.py @@ -1,11 +1,10 @@ """Renderers Renderers display the unified output format in some manner (be it text or file or graphical output""" +import collections import datetime import typing -import collections - from volatility.framework import interfaces diff --git a/volatility/framework/symbols/linux/extensions/__init__.py b/volatility/framework/symbols/linux/extensions/__init__.py index 880668a4c..25fc01caf 100644 --- a/volatility/framework/symbols/linux/extensions/__init__.py +++ b/volatility/framework/symbols/linux/extensions/__init__.py @@ -23,6 +23,9 @@ class task_struct(generic.GenericIntelProcess): if not pgd: return None + if not isinstance(parent_layer, interfaces.layers.TranslationLayerInterface): + raise TypeError("Parent layer is not a translation layer, unable to construct process layer") + dtb, layer_name = parent_layer.translate(pgd) if not dtb: return None diff --git a/volatility/framework/symbols/windows/extensions/__init__.py b/volatility/framework/symbols/windows/extensions/__init__.py index 459982fce..ebb4e674c 100644 --- a/volatility/framework/symbols/windows/extensions/__init__.py +++ b/volatility/framework/symbols/windows/extensions/__init__.py @@ -1,10 +1,10 @@ import collections.abc -import datetime import functools import logging import typing from volatility.framework import constants, exceptions, interfaces, objects, renderers +from volatility.framework.layers import intel from volatility.framework.objects import utility from volatility.framework.symbols import generic @@ -200,10 +200,12 @@ class _MMVAD_SHORT(objects.Struct): return self.u.VadFlags.PrivateMemory elif hasattr(self, "Core"): - if hasattr(self.Core, "u1") and hasattr(self.Core.u1, "VadFlags1") and hasattr(self.Core.u1.VadFlags1, "PrivateMemory"): + if hasattr(self.Core, "u1") and hasattr(self.Core.u1, "VadFlags1") and hasattr(self.Core.u1.VadFlags1, + "PrivateMemory"): return self.Core.u1.VadFlags1.PrivateMemory - elif hasattr(self.Core, "u") and hasattr(self.Core.u, "VadFlags") and hasattr(self.Core.u.VadFlags, "PrivateMemory"): + elif hasattr(self.Core, "u") and hasattr(self.Core.u, "VadFlags") and hasattr(self.Core.u.VadFlags, + "PrivateMemory"): return self.Core.u.VadFlags.PrivateMemory raise AttributeError("Unable to find the private memory member") @@ -393,6 +395,11 @@ class _EPROCESS(generic.GenericIntelProcess): """Constructs a new layer based on the process's DirectoryTableBase""" parent_layer = context.memory[self.vol.layer_name] + + if not isinstance(parent_layer, intel.Intel): + # We can't get bits_per_register unless we're an intel space (since that's not defined at the higher layer) + raise TypeError("Parent layer is not a translation layer, unable to construct process layer") + # Presumably for 64-bit systems, the DTB is defined as an array, rather than an unsigned long long if isinstance(self.Pcb.DirectoryTableBase, objects.Array): dtb = self.Pcb.DirectoryTableBase.cast("unsigned long long") diff --git a/volatility/framework/symbols/windows/extensions/registry.py b/volatility/framework/symbols/windows/extensions/registry.py index ea7528da7..072e0447d 100644 --- a/volatility/framework/symbols/windows/extensions/registry.py +++ b/volatility/framework/symbols/windows/extensions/registry.py @@ -117,6 +117,8 @@ class _CM_KEY_NODE(objects.Struct): def get_key_path(self) -> interfaces.objects.ObjectInterface: reg = self._context.memory[self.vol.layer_name] + if not isinstance(reg, RegistryHive): + raise TypeError("Key was not instantiated on a RegistryHive layer") # 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: if self.vol.offset == reg.root_cell_offset + 4: @@ -139,6 +141,9 @@ class _CM_KEY_VALUE(objects.Struct): data = b"" # Check if the data is stored inline layer = self._context.memory[self.vol.layer_name] + if not isinstance(layer, RegistryHive): + raise TypeError("Key value was not instantiated on a RegistryHive layer") + if self.DataLength & 0x80000000 and (0 > datalen or datalen > 4): raise ValueError("Unable to read inline registry value with excessive length: {}".format(datalen)) elif self.DataLength & 0x80000000: @@ -151,7 +156,7 @@ class _CM_KEY_VALUE(objects.Struct): # The value 4 should actually be unsigned-int.size, but since it's a file format that shouldn't change # the direct value 4 can be used instead block_offset = layer.get_cell(big_data.List + (i * 4)).cast("unsigned int") - if block_offset < layer.maximum_address: + if isinstance(block_offset, int) and block_offset < layer.maximum_address: amount = min(BIG_DATA_MAXLEN, datalen) data += layer.read(offset = layer.get_cell(block_offset).vol.offset, length = amount) datalen -= amount