diff --git a/volatility/framework/automagic/__init__.py b/volatility/framework/automagic/__init__.py index 76ddef24b..5cf12304f 100644 --- a/volatility/framework/automagic/__init__.py +++ b/volatility/framework/automagic/__init__.py @@ -31,7 +31,7 @@ import sys import traceback from typing import List, Type, Union -from volatility.framework import class_subclasses, import_files, interfaces, validity, constants +from volatility.framework import class_subclasses, import_files, interfaces, constants from volatility.framework.automagic import construct_layers, stacker, windows, pdbscan from volatility.framework.configuration import requirements @@ -88,7 +88,7 @@ def run(automagics: List[interfaces.automagic.AutomagicInterface], configurable: Union[interfaces.configuration.ConfigurableInterface, Type[interfaces.configuration. ConfigurableInterface]], config_path: str, - progress_callback: validity.ProgressCallback = None) -> List[traceback.TracebackException]: + progress_callback: constants.ProgressCallback = None) -> List[traceback.TracebackException]: """Runs through the list of `automagics` in order, allowing them to make changes to the context Args: diff --git a/volatility/framework/automagic/linux.py b/volatility/framework/automagic/linux.py index 56fb7ccfb..a8e35df62 100644 --- a/volatility/framework/automagic/linux.py +++ b/volatility/framework/automagic/linux.py @@ -21,8 +21,7 @@ import logging from typing import List, Optional, Tuple, Type -import volatility.framework.objects.utility -from volatility.framework import interfaces, constants, validity, exceptions, layers +from volatility.framework import interfaces, constants, exceptions, layers from volatility.framework import symbols, objects from volatility.framework.automagic import symbol_cache, symbol_finder from volatility.framework.layers import intel, scanners @@ -54,7 +53,7 @@ class LintelStacker(interfaces.automagic.StackerLayerInterface): def stack(cls, context: interfaces.context.ContextInterface, layer_name: str, - progress_callback: validity.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: + progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: """Attempts to identify linux within this layer""" # Bail out by default unless we can stack properly layer = context.memory[layer_name] @@ -290,7 +289,7 @@ class LinuxUtilities(object): context: interfaces.context.ContextInterface, symbol_table: str, layer_name: str, - progress_callback: validity.ProgressCallback = None) \ + progress_callback: constants.ProgressCallback = None) \ -> Tuple[int, int]: """Determines the offset of the actual DTB in physical space and its symbol offset""" init_task_symbol = symbol_table + constants.BANG + 'init_task' diff --git a/volatility/framework/automagic/mac.py b/volatility/framework/automagic/mac.py index ff00eae64..7e3f4617b 100644 --- a/volatility/framework/automagic/mac.py +++ b/volatility/framework/automagic/mac.py @@ -22,7 +22,7 @@ import logging import struct from typing import Optional -from volatility.framework import interfaces, constants, validity, layers +from volatility.framework import interfaces, constants, layers from volatility.framework import symbols from volatility.framework.automagic import symbol_cache, symbol_finder from volatility.framework.layers import intel, scanners @@ -53,7 +53,7 @@ class MacintelStacker(interfaces.automagic.StackerLayerInterface): def stack(cls, context: interfaces.context.ContextInterface, layer_name: str, - progress_callback: validity.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: + progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: """Attempts to identify mac within this layer""" # Bail out by default unless we can stack properly layer = context.memory[layer_name] @@ -171,7 +171,7 @@ class MacUtilities(object): layer_name: str, compare_banner: str = "", compare_banner_offset: int = 0, - progress_callback: validity.ProgressCallback = None) -> int: + progress_callback: constants.ProgressCallback = None) -> int: """Determines the offset of the actual DTB in physical space and its symbol offset""" version_symbol = symbol_table + constants.BANG + 'version' version_json_address = context.symbol_space.get_symbol(version_symbol).address diff --git a/volatility/framework/automagic/pdbscan.py b/volatility/framework/automagic/pdbscan.py index 45d2d821a..e70e745d0 100644 --- a/volatility/framework/automagic/pdbscan.py +++ b/volatility/framework/automagic/pdbscan.py @@ -29,7 +29,7 @@ import os import struct from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Tuple, Union -from volatility.framework import constants, exceptions, interfaces, layers, validity +from volatility.framework import constants, exceptions, interfaces, layers from volatility.framework.configuration import requirements from volatility.framework.layers import intel, scanners from volatility.framework.symbols import intermed, native @@ -87,7 +87,7 @@ class PdbSignatureScanner(interfaces.layers.ScannerInterface): def scan(ctx: interfaces.context.ContextInterface, layer_name: str, page_size: int, - progress_callback: validity.ProgressCallback = None, + progress_callback: constants.ProgressCallback = None, start: Optional[int] = None, end: Optional[int] = None) -> Generator[Dict[str, Optional[Union[bytes, str, int]]], None, None]: """Scans through `layer_name` at `ctx` looking for RSDS headers that indicate one of four common pdb kernel names @@ -245,7 +245,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): def method_fixed_mapping(self, context: interfaces.context.ContextInterface, vlayer: layers.intel.Intel, - progress_callback: validity.ProgressCallback = None) -> ValidKernelsType: + progress_callback: constants.ProgressCallback = None) -> ValidKernelsType: # TODO: Verify this is a windows image vollog.debug("Kernel base determination - testing fixed base address") valid_kernels = {} @@ -286,7 +286,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): def method_module_offset(self, context: interfaces.context.ContextInterface, vlayer: layers.intel.Intel, - progress_callback: validity.ProgressCallback = None) -> ValidKernelsType: + progress_callback: constants.ProgressCallback = None) -> ValidKernelsType: """Method for finding a suitable kernel offset based on a module table""" vollog.debug("Kernel base determination - searching layer module list structure") valid_kernels = {} # type: ValidKernelsType @@ -319,7 +319,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): def method_kdbg_offset(self, context: interfaces.context.ContextInterface, vlayer: layers.intel.Intel, - progress_callback: validity.ProgressCallback = None) -> ValidKernelsType: + progress_callback: constants.ProgressCallback = None) -> ValidKernelsType: vollog.debug("Kernel base determination - using KDBG structure for kernel offset") valid_kernels = {} # type: ValidKernelsType physical_layer_name = self.get_physical_layer_name(context, vlayer) @@ -347,7 +347,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): context: interfaces.context.ContextInterface, vlayer: layers.intel.Intel, address: int, - progress_callback: validity.ProgressCallback = None) -> ValidKernelsType: + progress_callback: constants.ProgressCallback = None) -> ValidKernelsType: """Scans a virtual address """ # Scan a few megs of the virtual space at the location to see if they're potential kernels @@ -376,7 +376,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): def determine_valid_kernels(self, context: interfaces.context.ContextInterface, potential_layers: List[str], - progress_callback: validity.ProgressCallback = None) -> ValidKernelsType: + progress_callback: constants.ProgressCallback = None) -> ValidKernelsType: """Runs through the identified potential kernels and verifies their suitability This carries out a scan using the pdb_signature scanner on a physical layer. It uses the @@ -408,7 +408,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): context: interfaces.context.ContextInterface, config_path: str, requirement: interfaces.configuration.RequirementInterface, - progress_callback: validity.ProgressCallback = None) -> None: + progress_callback: constants.ProgressCallback = None) -> None: if requirement.unsatisfied(context, config_path): if "pdbscan" not in context.symbol_space: context.symbol_space.append(native.NativeTable("pdbscan", native.std_ctypes)) diff --git a/volatility/framework/automagic/stacker.py b/volatility/framework/automagic/stacker.py index be2d4641f..b2b3526e6 100644 --- a/volatility/framework/automagic/stacker.py +++ b/volatility/framework/automagic/stacker.py @@ -31,7 +31,7 @@ import traceback from typing import List, Optional, Tuple from volatility import framework -from volatility.framework import interfaces, constants, validity +from volatility.framework import interfaces, constants from volatility.framework.automagic import construct_layers from volatility.framework.configuration import requirements from volatility.framework.layers import physical @@ -61,7 +61,7 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): context: interfaces.context.ContextInterface, config_path: str, requirement: interfaces.configuration.RequirementInterface, - progress_callback: validity.ProgressCallback = None) -> Optional[List[str]]: + progress_callback: constants.ProgressCallback = None) -> Optional[List[str]]: """Runs the automagic over the configurable""" # Quick exit if we're not needed @@ -75,7 +75,6 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): return list(unsatisfied) if not self.config or not self.config.get('single_location', None): raise ValueError("Unable to run LayerStacker, single_location parameter not provided") - self._check_type(requirement, interfaces.configuration.RequirementInterface) # Search for suitable requirements self.stack(context, config_path, requirement, progress_callback) @@ -84,7 +83,7 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): def stack(self, context: interfaces.context.ContextInterface, config_path: str, requirement: interfaces.configuration.RequirementInterface, - progress_callback: validity.ProgressCallback) -> None: + progress_callback: constants.ProgressCallback) -> None: """Stacks the various layers and attaches these to a specific requirement Args: @@ -106,7 +105,6 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): new_context = context.clone() location = self.config.get('single_location', None) - self._check_type(location, str) # Setup the local copy of the resource current_layer_name = context.memory.free_layer_name("FileLayer") diff --git a/volatility/framework/automagic/symbol_finder.py b/volatility/framework/automagic/symbol_finder.py index 58cbd1348..680ba9cb4 100644 --- a/volatility/framework/automagic/symbol_finder.py +++ b/volatility/framework/automagic/symbol_finder.py @@ -21,7 +21,7 @@ import logging from typing import Any, Iterable, List, Tuple, Type, Optional -from volatility.framework import interfaces, validity +from volatility.framework import interfaces, constants from volatility.framework.automagic import symbol_cache from volatility.framework.configuration import requirements from volatility.framework.layers import scanners @@ -55,7 +55,7 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface): context: interfaces.context.ContextInterface, config_path: str, requirement: interfaces.configuration.RequirementInterface, - progress_callback: validity.ProgressCallback = None) -> None: + progress_callback: constants.ProgressCallback = None) -> None: """Searches for SymbolRequirements and attempt to populate them""" # Bomb out early if our details haven't been configured @@ -88,7 +88,7 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface): config_path: str, requirement: interfaces.configuration.ConstructableRequirementInterface, layer_name: str, - progress_callback: validity.ProgressCallback = None) -> None: + progress_callback: constants.ProgressCallback = None) -> None: """Accepts a context, config_path and SymbolRequirement, with a constructed layer_name and scans the layer for banners""" diff --git a/volatility/framework/automagic/windows.py b/volatility/framework/automagic/windows.py index 7414a3bee..b7f7e771b 100644 --- a/volatility/framework/automagic/windows.py +++ b/volatility/framework/automagic/windows.py @@ -45,14 +45,14 @@ import logging import struct from typing import Any, Generator, List, Optional, Tuple, Type -from volatility.framework import interfaces, layers, validity +from volatility.framework import interfaces, layers, constants from volatility.framework.configuration import requirements from volatility.framework.layers import intel vollog = logging.getLogger(__name__) -class DtbTest(validity.ValidityRoutines): +class DtbTest: """This class generically contains the tests for a page based on a set of class parameters When constructed it contains all the information necessary to extract a specific index from a page @@ -60,11 +60,11 @@ class DtbTest(validity.ValidityRoutines): """ def __init__(self, layer_type: Type[layers.intel.Intel], ptr_struct: str, ptr_reference: int, mask: int) -> None: - self.layer_type = self._check_class(layer_type, layers.intel.Intel) - self.ptr_struct = self._check_type(ptr_struct, str) + self.layer_type = layer_type + self.ptr_struct = ptr_struct self.ptr_size = struct.calcsize(ptr_struct) - self.ptr_reference = self._check_type(ptr_reference, int) - self.mask = self._check_type(mask, int) + self.ptr_reference = ptr_reference + self.mask = mask self.page_size = layer_type.page_size # type: int def _unpack(self, value: bytes) -> int: @@ -214,8 +214,6 @@ class PageMapScanner(interfaces.layers.ScannerInterface): def __init__(self, tests: List[DtbTest]) -> None: super().__init__() - for value in tests: - self._check_type(value, DtbTest) self.tests = tests def __call__(self, data: bytes, data_offset: int) -> Generator[Tuple[DtbTest, int], None, None]: @@ -242,7 +240,7 @@ class WintelHelper(interfaces.automagic.AutomagicInterface): context: interfaces.context.ContextInterface, config_path: str, requirement: interfaces.configuration.RequirementInterface, - progress_callback: validity.ProgressCallback = None) -> None: + progress_callback: constants.ProgressCallback = None) -> None: useful = [] sub_config_path = interfaces.configuration.path_join(config_path, requirement.name) if (isinstance(requirement, requirements.TranslationLayerRequirement) @@ -289,7 +287,7 @@ class WintelStacker(interfaces.automagic.StackerLayerInterface): def stack(cls, context: interfaces.context.ContextInterface, layer_name: str, - progress_callback: validity.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: + progress_callback: constants.ProgressCallback = None) -> 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. @@ -373,7 +371,7 @@ class WinSwapLayers(interfaces.automagic.AutomagicInterface): context: interfaces.context.ContextInterface, config_path: str, requirement: interfaces.configuration.RequirementInterface, - progress_callback: validity.ProgressCallback = None) -> None: + progress_callback: constants.ProgressCallback = None) -> None: """Finds translation layers that can have swap layers added""" path_join = interfaces.configuration.path_join self._translation_requirement = self.find_requirements( diff --git a/volatility/framework/configuration/requirements.py b/volatility/framework/configuration/requirements.py index 0794c7547..8062b8577 100644 --- a/volatility/framework/configuration/requirements.py +++ b/volatility/framework/configuration/requirements.py @@ -122,7 +122,7 @@ class ListRequirement(configuration.RequirementInterface): if self.max_elements and not (len(value) < self.max_elements): vollog.log(constants.LOGLEVEL_V, "TypeError - Too many values provided to list option.") return {config_path: self} - if not all([self._check_type(element, self.element_type) for element in value]): + if not all([isinstance(element, self.element_type) for element in value]): vollog.log(constants.LOGLEVEL_V, "TypeError - At least one element in the list is not of the correct type.") return {config_path: self} return {} diff --git a/volatility/framework/constants/__init__.py b/volatility/framework/constants/__init__.py index 3750a40cb..5b2f27c2d 100644 --- a/volatility/framework/constants/__init__.py +++ b/volatility/framework/constants/__init__.py @@ -23,6 +23,7 @@ Stores all the constant values that are generally fixed throughout volatility This includes default scanning block sizes, etc.""" import os.path import sys +from typing import Optional, Callable import volatility.framework.constants.linux import volatility.framework.constants.windows @@ -53,3 +54,5 @@ os.makedirs(CACHE_PATH, exist_ok = True) LINUX_BANNERS_PATH = os.path.join(CACHE_PATH, "linux_banners.cache") MAC_BANNERS_PATH = os.path.join(CACHE_PATH, "mac_banners.cache") + +ProgressCallback = Optional[Callable[[float, str], None]] diff --git a/volatility/framework/contexts/__init__.py b/volatility/framework/contexts/__init__.py index 7ea40fbe0..90ee560ad 100644 --- a/volatility/framework/contexts/__init__.py +++ b/volatility/framework/contexts/__init__.py @@ -26,7 +26,7 @@ import functools import hashlib from typing import Callable, Dict, Iterable, List, Optional, Set, Tuple, Union -from volatility.framework import constants, interfaces, symbols, validity +from volatility.framework import constants, interfaces, symbols class Context(interfaces.context.ContextInterface): @@ -149,7 +149,6 @@ def get_module_wrapper(method: str) -> Callable: """Returns a symbol using the symbol_table_name of the Module""" def wrapper(self, name: str) -> Callable: - self._check_type(name, str) if constants.BANG in name: raise ValueError("Name cannot reference another module") return getattr(self._context.symbol_space, method)(self._module_name + constants.BANG + name) @@ -175,7 +174,6 @@ class Module(interfaces.context.ModuleInterface): """ type_arg = None # type: Optional[Union[str, interfaces.objects.Template]] if symbol_name is not None: - self._check_type(symbol_name, str) if constants.BANG in symbol_name: raise ValueError("Symbol_name cannot reference another module") symbol = self._context.symbol_space.get_symbol(self.symbol_table_name + constants.BANG + symbol_name) @@ -186,8 +184,6 @@ class Module(interfaces.context.ModuleInterface): if not self._absolute_symbol_addresses: offset += self._offset elif type_name is not None and offset is not None: - self._check_type(type_name, str) - self._check_type(offset, int) if constants.BANG in type_name: raise ValueError("Type_name cannot reference another module") type_arg = self.symbol_table_name + constants.BANG + type_name @@ -225,7 +221,7 @@ class SizedModule(Module): native_layer_name = native_layer_name, symbol_table_name = symbol_table_name, absolute_symbol_addresses = absolute_symbol_addresses) - self._size = self._check_type(size, int) + self._size = size @property def size(self) -> int: @@ -256,12 +252,10 @@ class SizedModule(Module): offset = offset - self._offset, size = size, table_name = self.symbol_table_name)) -class ModuleCollection(validity.ValidityRoutines): +class ModuleCollection: """Class to contain a collection of SizedModules and reason about their contents""" def __init__(self, modules: List[SizedModule]) -> None: - for module in modules: - self._check_type(module, SizedModule) self._modules = modules def deduplicate(self) -> 'ModuleCollection': diff --git a/volatility/framework/interfaces/automagic.py b/volatility/framework/interfaces/automagic.py index 30a99375e..462ece817 100644 --- a/volatility/framework/interfaces/automagic.py +++ b/volatility/framework/interfaces/automagic.py @@ -24,7 +24,7 @@ Automagic objects attempt to automatically fill configuration values that a user from abc import ABCMeta from typing import Any, List, Optional, Tuple, Union, Type -from volatility.framework import interfaces, validity +from volatility.framework import interfaces, constants from volatility.framework.configuration import requirements @@ -62,7 +62,7 @@ class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metacla context: interfaces.context.ContextInterface, config_path: str, requirement: interfaces.configuration.RequirementInterface, - progress_callback: validity.ProgressCallback = None) -> Optional[List[Any]]: + progress_callback: constants.ProgressCallback = None) -> Optional[List[Any]]: """Runs the automagic over the configurable""" return [] @@ -104,7 +104,7 @@ class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metacla return results -class StackerLayerInterface(validity.ValidityRoutines, metaclass = ABCMeta): +class StackerLayerInterface(metaclass = ABCMeta): """Class that takes a lower layer and attempts to build on it stack_order determines the order (from low to high) that stacking layers @@ -117,7 +117,7 @@ class StackerLayerInterface(validity.ValidityRoutines, metaclass = ABCMeta): def stack(self, context: interfaces.context.ContextInterface, layer_name: str, - progress_callback: validity.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: + progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: """ Method to determine whether this builder can operate on the named layer. If so, modify the context appropriately. diff --git a/volatility/framework/interfaces/configuration.py b/volatility/framework/interfaces/configuration.py index 39a677bf7..54c2f5b0a 100644 --- a/volatility/framework/interfaces/configuration.py +++ b/volatility/framework/interfaces/configuration.py @@ -37,7 +37,7 @@ import sys from abc import ABCMeta, abstractmethod from typing import Any, ClassVar, Dict, Generator, List, Optional, Type, Union -from volatility.framework import constants, interfaces, validity +from volatility.framework import constants, interfaces from volatility.framework.interfaces.context import ContextInterface CONFIG_SEPARATOR = "." @@ -266,7 +266,7 @@ class HierarchicalDict(collections.abc.Mapping): return json.dumps(dict([(key, self[key]) for key in sorted(self.generator())]), indent = 2) -class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): +class RequirementInterface(metaclass = ABCMeta): """Class that defines a requirement A requirement is a means for plugins and other framework components to request specific configuration data. @@ -284,7 +284,6 @@ class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): default: Optional[ConfigSimpleType] = None, optional: bool = False) -> None: super().__init__() - self._check_type(name, str) if CONFIG_SEPARATOR in name: raise ValueError("Name cannot contain the config-hierarchy divider ({})".format(CONFIG_SEPARATOR)) self._name = name @@ -321,9 +320,7 @@ class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): """Sets the optional value for a requirement""" self._optional = bool(value) - def config_value(self, - context: interfaces.context.ContextInterface, - config_path: str, + def config_value(self, context: ContextInterface, config_path: str, default: ConfigSimpleType = None) -> ConfigSimpleType: """Returns the value for this Requirement from its config path""" return context.config.get(config_path, default) @@ -336,16 +333,13 @@ class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): def add_requirement(self, requirement: 'RequirementInterface') -> None: """Adds a child to the list of requirements""" - self._check_type(requirement, RequirementInterface) self._requirements[requirement.name] = requirement def remove_requirement(self, requirement: 'RequirementInterface') -> None: """Removes a child from the list of requirements""" - self._check_type(requirement, RequirementInterface) del self._requirements[requirement.name] - def unsatisfied_children(self, context: interfaces.context.ContextInterface, - config_path: str) -> Dict[str, 'RequirementInterface']: + def unsatisfied_children(self, context: ContextInterface, config_path: str) -> Dict[str, 'RequirementInterface']: """Method that will validate all child requirements""" result = {} for requirement in self.requirements.values(): @@ -356,8 +350,7 @@ class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): # Validation routines @abstractmethod - def unsatisfied(self, context: interfaces.context.ContextInterface, - config_path: str) -> Dict[str, 'RequirementInterface']: + def unsatisfied(self, context: ContextInterface, config_path: str) -> Dict[str, 'RequirementInterface']: """Method to validate the value stored at config_path for the configuration object against a context Returns a list containing its own name (or multiple unsatisfied requirement names) when invalid @@ -376,8 +369,7 @@ class SimpleTypeRequirement(RequirementInterface): """Always raises a TypeError as instance requirements cannot have children""" raise TypeError("Instance Requirements cannot have subrequirements") - def unsatisfied(self, context: interfaces.context.ContextInterface, - config_path: str) -> Dict[str, RequirementInterface]: + def unsatisfied(self, context: ContextInterface, config_path: str) -> Dict[str, RequirementInterface]: """Validates the instance requirement based upon its `instance_type`.""" config_path = path_join(config_path, self.name) @@ -403,8 +395,7 @@ class ClassRequirement(RequirementInterface): def cls(self) -> Type: return self._cls - def unsatisfied(self, context: interfaces.context.ContextInterface, - config_path: str) -> Dict[str, RequirementInterface]: + def unsatisfied(self, context: ContextInterface, config_path: str) -> Dict[str, RequirementInterface]: """Checks to see if a class can be recovered""" config_path = path_join(config_path, self.name) @@ -441,10 +432,10 @@ class ConstructableRequirementInterface(RequirementInterface): self._current_class_requirements = set() @abstractmethod - def construct(self, context: interfaces.context.ContextInterface, config_path: str) -> None: + def construct(self, context: ContextInterface, config_path: str) -> None: """Method for constructing within the context any required elements from subrequirements""" - def _validate_class(self, context: interfaces.context.ContextInterface, config_path: str) -> None: + def _validate_class(self, context: ContextInterface, config_path: str) -> None: """Method to check if the class Requirement is valid and if so populate the other requirements (but no need to validate, since we're invalid already) """ @@ -462,9 +453,7 @@ class ConstructableRequirementInterface(RequirementInterface): self._current_class_requirements.add(requirement.name) self.add_requirement(requirement) - def _construct_class(self, - context: interfaces.context.ContextInterface, - config_path: str, + def _construct_class(self, context: ContextInterface, config_path: str, requirement_dict: Dict[str, object] = None) -> Optional['interfaces.objects.ObjectInterface']: """Constructs the class, handing args and the subrequirements as parameters to __init__""" if self.requirements["class"].unsatisfied(context, config_path): @@ -494,23 +483,22 @@ class ConstructableRequirementInterface(RequirementInterface): class ConfigurableRequirementInterface(RequirementInterface): """Simple Abstract class to provide build_required_config""" - def build_configuration(self, context: interfaces.context.ContextInterface, config_path: str, - value: Any) -> HierarchicalDict: + def build_configuration(self, context: ContextInterface, config_path: str, value: Any) -> HierarchicalDict: """Proxies to a ConfigurableInterface if necessary""" -class ConfigurableInterface(validity.ValidityRoutines, metaclass = ABCMeta): +class ConfigurableInterface(metaclass = ABCMeta): """Class to allow objects to have requirements and read configuration data from the context config tree""" - def __init__(self, context: interfaces.context.ContextInterface, config_path: str) -> None: + def __init__(self, context: ContextInterface, config_path: str) -> None: """Basic initializer that allows configurables to access their own config settings""" super().__init__() - self._context = self._check_type(context, ContextInterface) - self._config_path = self._check_type(config_path, str) + self._context = context + self._config_path = config_path self._config_cache = None # type: Optional[HierarchicalDict] @property - def context(self) -> 'interfaces.context.ContextInterface': + def context(self) -> ContextInterface: return self._context @property @@ -519,7 +507,7 @@ class ConfigurableInterface(validity.ValidityRoutines, metaclass = ABCMeta): @config_path.setter def config_path(self, value: str) -> None: - self._config_path = self._check_type(value, str) + self._config_path = value self._config_cache = None @property @@ -552,8 +540,7 @@ class ConfigurableInterface(validity.ValidityRoutines, metaclass = ABCMeta): return [] @classmethod - def unsatisfied(cls, context: interfaces.context.ContextInterface, - config_path: str) -> Dict[str, RequirementInterface]: + def unsatisfied(cls, context: ContextInterface, config_path: str) -> Dict[str, RequirementInterface]: """Returns a list of the names of all unsatisfied requirements Since a satisfied set of requirements will return [], it can be used in tests as follows: diff --git a/volatility/framework/interfaces/context.py b/volatility/framework/interfaces/context.py index 1a5a42786..9d49c2e9a 100644 --- a/volatility/framework/interfaces/context.py +++ b/volatility/framework/interfaces/context.py @@ -27,7 +27,7 @@ import copy from abc import ABCMeta, abstractmethod from typing import Optional, Union -from volatility.framework import interfaces, validity +from volatility.framework import interfaces class ContextInterface(object, metaclass = ABCMeta): @@ -103,7 +103,7 @@ class ContextInterface(object, metaclass = ABCMeta): """Create a module object """ -class ModuleInterface(validity.ValidityRoutines, metaclass = ABCMeta): +class ModuleInterface(metaclass = ABCMeta): """Maintains state concerning a particular loaded module in memory This object is OS-independent. @@ -117,13 +117,13 @@ class ModuleInterface(validity.ValidityRoutines, metaclass = ABCMeta): symbol_table_name: Optional[str] = None, native_layer_name: Optional[str] = None, absolute_symbol_addresses: bool = False) -> None: - self._context = self._check_type(context, ContextInterface) - self._module_name = self._check_type(module_name, str) - self._layer_name = self._check_type(layer_name, str) - self._offset = self._check_type(offset, int) + self._context = context + self._module_name = module_name + self._layer_name = layer_name + self._offset = offset self._native_layer_name = None if native_layer_name: - self._native_layer_name = self._check_type(native_layer_name, str) + self._native_layer_name = native_layer_name self.symbol_table_name = symbol_table_name or self._module_name self._absolute_symbol_addresses = absolute_symbol_addresses super().__init__() diff --git a/volatility/framework/interfaces/layers.py b/volatility/framework/interfaces/layers.py index 2faafdfb5..5eb7e4f6f 100644 --- a/volatility/framework/interfaces/layers.py +++ b/volatility/framework/interfaces/layers.py @@ -28,7 +28,7 @@ import traceback from abc import ABCMeta, abstractmethod from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Tuple, Union -from volatility.framework import constants, exceptions, interfaces, validity +from volatility.framework import constants, exceptions, interfaces vollog = logging.getLogger(__name__) @@ -45,7 +45,7 @@ ProgressValue = Union['DummyProgress', multiprocessing.Value] IteratorValue = Tuple[List[Tuple[str, int, int]], int] -class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta): +class ScannerInterface(metaclass = ABCMeta): """Class for layer scanners that return locations of particular values from within the data These are designed to be given a chunk of data and return a generator which yields @@ -85,7 +85,7 @@ class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta): @context.setter 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, interfaces.context.ContextInterface) + self._context = ctx @property def layer_name(self) -> Optional[str]: @@ -94,7 +94,7 @@ class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta): @layer_name.setter 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) + self._layer_name = layer_name @abstractmethod def __call__(self, data: bytes, data_offset: int) -> Iterable[Any]: @@ -106,8 +106,7 @@ class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta): """ -class DataLayerInterface( - interfaces.configuration.ConfigurableInterface, validity.ValidityRoutines, metaclass = ABCMeta): +class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metaclass = ABCMeta): """A Layer that directly holds data (and does not translate it). This is effectively a leaf node in a layer tree. It directly accesses a data source and exposes it within volatility.""" @@ -122,7 +121,7 @@ class DataLayerInterface( name: str, metadata: Optional[Dict[str, Any]] = None) -> None: super().__init__(context, config_path) - self._name = self._check_type(name, str) + self._name = name if metadata: self._direct_metadata.update(metadata) @@ -191,7 +190,7 @@ class DataLayerInterface( def scan(self, context: interfaces.context.ContextInterface, scanner: ScannerInterface, - progress_callback: validity.ProgressCallback = None, + progress_callback: constants.ProgressCallback = None, sections: Iterable[Tuple[int, int]] = None) -> Iterable[Any]: """Scans a Translation layer by chunk @@ -200,7 +199,7 @@ class DataLayerInterface( 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 = scanner scanner.context = context scanner.layer_name = self.name @@ -419,7 +418,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta): offset += chunk_size -class Memory(validity.ValidityRoutines, collections.abc.Mapping): +class Memory(collections.abc.Mapping): """Container for multiple layers of data""" def __init__(self) -> None: @@ -441,7 +440,6 @@ class Memory(validity.ValidityRoutines, collections.abc.Mapping): This will throw an exception if the required dependencies are not met """ - self._check_type(layer, DataLayerInterface) if layer.name in self._layers: raise exceptions.LayerException("Layer already exists: {}".format(layer.name)) if isinstance(layer, TranslationLayerInterface): @@ -466,8 +464,6 @@ class Memory(validity.ValidityRoutines, collections.abc.Mapping): 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) - count = 1 while prefix + str(count) in self: count += 1 diff --git a/volatility/framework/interfaces/objects.py b/volatility/framework/interfaces/objects.py index 3f3274fb6..efe2606eb 100644 --- a/volatility/framework/interfaces/objects.py +++ b/volatility/framework/interfaces/objects.py @@ -26,13 +26,13 @@ import logging from abc import ABCMeta, abstractmethod from typing import Any, Dict, List, Mapping, Optional -from volatility.framework import constants, validity, interfaces +from volatility.framework import constants, interfaces from volatility.framework.interfaces import context as interfaces_context vollog = logging.getLogger(__name__) -class ReadOnlyMapping(validity.ValidityRoutines, collections.abc.Mapping): +class ReadOnlyMapping(collections.abc.Mapping): """A read-only mapping of various values that offer attribute access as well This ensures that the data stored in the mapping should not be modified, making an immutable mapping. @@ -78,9 +78,6 @@ class ObjectInformation(ReadOnlyMapping): member_name: Optional[str] = None, parent: Optional['ObjectInterface'] = None, native_layer_name: Optional[str] = None): - self._check_type(offset, int) - if parent: - self._check_type(parent, ObjectInterface) super().__init__({ 'layer_name': layer_name, 'offset': offset, @@ -90,16 +87,14 @@ class ObjectInformation(ReadOnlyMapping): }) -class ObjectInterface(validity.ValidityRoutines, metaclass = ABCMeta): +class ObjectInterface(metaclass = ABCMeta): """A base object required to be the ancestor of every object used in volatility""" 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 + # we're reliant on type_checking to ensure correctness of context, offset and parent # Everything else may be wrong, but that will get caught later on - self._check_type(context, interfaces_context.ContextInterface) - self._check_type(object_info, ObjectInformation) # Add an empty dictionary at the start to allow objects to add their own data to the vol object # @@ -201,7 +196,7 @@ class ObjectInterface(validity.ValidityRoutines, metaclass = ABCMeta): return False -class Template(validity.ValidityRoutines): +class Template: """Class for all Factories that take offsets, and data layers and produce objects This is effectively a class for currying object calls. It creates a callable that can be called with the following diff --git a/volatility/framework/interfaces/plugins.py b/volatility/framework/interfaces/plugins.py index 1706e4b49..f814102f7 100644 --- a/volatility/framework/interfaces/plugins.py +++ b/volatility/framework/interfaces/plugins.py @@ -28,8 +28,7 @@ import logging from abc import ABCMeta, abstractmethod from typing import List, Optional, TYPE_CHECKING -from volatility.framework import exceptions -from volatility.framework import validity +from volatility.framework import exceptions, constants from volatility.framework.interfaces import configuration as interfaces_configuration vollog = logging.getLogger(__name__) @@ -38,7 +37,7 @@ if TYPE_CHECKING: from volatility.framework import interfaces, renderers -class FileInterface(validity.ValidityRoutines, metaclass = ABCMeta): +class FileInterface(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) -> None: @@ -73,7 +72,7 @@ class FileConsumerInterface(object): # The plugin runs and produces a TreeGrid output -class PluginInterface(interfaces_configuration.ConfigurableInterface, validity.ValidityRoutines, metaclass = ABCMeta): +class PluginInterface(interfaces_configuration.ConfigurableInterface, metaclass = ABCMeta): """Class that defines the basic interface that all Plugins must maintain. The constructor must only take a `context` and `config_path`, so that plugins can be launched automatically. As such all configuration information must be provided through the requirements and configuration information in the @@ -83,7 +82,7 @@ class PluginInterface(interfaces_configuration.ConfigurableInterface, validity.V def __init__(self, context: 'interfaces.context.ContextInterface', config_path: str, - progress_callback: validity.ProgressCallback = None) -> None: + progress_callback: constants.ProgressCallback = None) -> None: super().__init__(context, config_path) self._progress_callback = progress_callback or (lambda f, s: None) # Plugins self validate on construction, it makes it more difficult to work with them, but then @@ -94,7 +93,7 @@ class PluginInterface(interfaces_configuration.ConfigurableInterface, validity.V self._file_consumer = None # type: Optional[FileConsumerInterface] def set_file_consumer(self, consumer: FileConsumerInterface) -> None: - self._file_consumer = self._check_type(consumer, FileConsumerInterface) + self._file_consumer = consumer def produce_file(self, filedata: FileInterface) -> None: """Adds a file to the plugin's file store and returns the chosen filename for the file""" diff --git a/volatility/framework/interfaces/renderers.py b/volatility/framework/interfaces/renderers.py index 364db07ac..47d29c4ed 100644 --- a/volatility/framework/interfaces/renderers.py +++ b/volatility/framework/interfaces/renderers.py @@ -26,14 +26,12 @@ import datetime from abc import abstractmethod, ABCMeta from typing import Any, Callable, ClassVar, Generator, Iterable, List, NamedTuple, Optional, TypeVar, Type, Tuple, Union -from volatility.framework import validity - Column = NamedTuple('Column', [('index', int), ('name', str), ('type', Any)]) RenderOption = Any -class Renderer(validity.ValidityRoutines, metaclass = ABCMeta): +class Renderer(metaclass = ABCMeta): """Class that defines the interface that all output renderers must support""" def __init__(self, options: List[RenderOption]) -> None: diff --git a/volatility/framework/interfaces/symbols.py b/volatility/framework/interfaces/symbols.py index cfeaa9e4c..4ccc11e4b 100644 --- a/volatility/framework/interfaces/symbols.py +++ b/volatility/framework/interfaces/symbols.py @@ -24,11 +24,11 @@ import collections.abc from abc import abstractmethod, ABC from typing import Any, Dict, Iterable, List, Optional, Tuple, Type -from volatility.framework import constants, exceptions, validity +from volatility.framework import constants, exceptions from volatility.framework.interfaces import configuration, objects, context as interfaces_context -class SymbolInterface(validity.ValidityRoutines): +class SymbolInterface: """Contains information about a named location in a program's memory""" def __init__(self, @@ -36,21 +36,15 @@ class SymbolInterface(validity.ValidityRoutines): address: int, type: Optional[objects.Template] = None, constant_data: Optional[bytes] = None) -> None: - self._name = self._check_type(name, str) + self._name = name if constants.BANG in self._name: raise ValueError("Symbol names cannot contain the symbol differentiator ({})".format(constants.BANG)) # Scope can be added at a later date self._location = None - self._address = self._check_type(address, int) - - self._type = None - if type is not None: - self._type = self._check_type(type, objects.Template) - - self._constant_data = None - if constant_data is not None: - self._constant_data = self._check_type(constant_data, bytes) + self._address = address + self._type = type + self._constant_data = constant_data @property def name(self) -> str: @@ -80,7 +74,7 @@ class SymbolInterface(validity.ValidityRoutines): return self._constant_data -class BaseSymbolTableInterface(validity.ValidityRoutines): +class BaseSymbolTableInterface: """The base interface, inherited by both NativeTables and SymbolTables native_types is a NativeTableInterface used for native types for the particular loaded symbol table @@ -91,11 +85,11 @@ class BaseSymbolTableInterface(validity.ValidityRoutines): def __init__(self, name: str, native_types: 'NativeTableInterface', table_mapping: Optional[Dict[str, str]] = None) -> None: - self.name = self._check_type(name, str) + self.name = name if table_mapping is None: table_mapping = {} - self.table_mapping = self._check_type(table_mapping, dict) - self._native_types = self._check_type(native_types, NativeTableInterface) + self.table_mapping = table_mapping + self._native_types = native_types self._sort_symbols = [] # type: List[Tuple[int, str]] # ## Required Symbol functions @@ -146,7 +140,6 @@ class BaseSymbolTableInterface(validity.ValidityRoutines): WARNING: This allows changing the underlying size of all the other types referenced in the SymbolTable """ - self._check_type(value, NativeTableInterface) self._native_types = value # ## Functions for overriding classes diff --git a/volatility/framework/layers/crash.py b/volatility/framework/layers/crash.py index afba3de27..59d44a469 100644 --- a/volatility/framework/layers/crash.py +++ b/volatility/framework/layers/crash.py @@ -21,7 +21,7 @@ import struct from typing import Tuple, Optional -from volatility.framework import constants, exceptions, interfaces, validity +from volatility.framework import constants, exceptions, interfaces from volatility.framework.layers import segmented from volatility.framework.symbols import intermed @@ -116,8 +116,7 @@ class WindowsCrashDump32Stacker(interfaces.automagic.StackerLayerInterface): def stack(cls, context: interfaces.context.ContextInterface, layer_name: str, - progress_callback: validity.ProgressCallback = None) \ - -> Optional[interfaces.layers.DataLayerInterface]: + progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: try: WindowsCrashDump32Layer._check_header(context.memory[layer_name]) except WindowsCrashDump32FormatException: diff --git a/volatility/framework/layers/intel.py b/volatility/framework/layers/intel.py index 45e45e273..32b35a1b8 100644 --- a/volatility/framework/layers/intel.py +++ b/volatility/framework/layers/intel.py @@ -66,10 +66,9 @@ class Intel(interfaces.layers.TranslationLayerInterface): name: str, metadata: Optional[Dict[str, Any]] = None) -> None: super().__init__(context = context, config_path = config_path, name = name, metadata = metadata) - self._base_layer = self._check_type(self.config["memory_layer"], str) + self._base_layer = self.config["memory_layer"] self._swap_layers = [] # type: List[str] - self._check_type(self.config.get("swap_layers", False), bool) - self._page_map_offset = self._check_type(self.config["page_map_offset"], int) + self._page_map_offset = self.config["page_map_offset"] # These can vary depending on the type of space self._index_shift = int(math.ceil(math.log2(struct.calcsize(self._entry_format)))) diff --git a/volatility/framework/layers/lime.py b/volatility/framework/layers/lime.py index 422d9a9c6..837b1ce74 100644 --- a/volatility/framework/layers/lime.py +++ b/volatility/framework/layers/lime.py @@ -21,7 +21,7 @@ import struct from typing import Optional, Tuple -from volatility.framework import exceptions, interfaces, validity +from volatility.framework import exceptions, interfaces, constants from volatility.framework.layers import segmented @@ -95,8 +95,7 @@ class LimeStacker(interfaces.automagic.StackerLayerInterface): def stack(cls, context: interfaces.context.ContextInterface, layer_name: str, - progress_callback: validity.ProgressCallback = None) \ - -> Optional[interfaces.layers.DataLayerInterface]: + progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: try: LimeLayer._check_header(context.memory[layer_name]) except LimeFormatException: diff --git a/volatility/framework/layers/physical.py b/volatility/framework/layers/physical.py index be2bba721..c0e42afcb 100644 --- a/volatility/framework/layers/physical.py +++ b/volatility/framework/layers/physical.py @@ -37,7 +37,7 @@ class BufferDataLayer(interfaces.layers.DataLayerInterface): buffer: bytes, metadata: Optional[Dict[str, Any]] = None) -> None: super().__init__(context = context, config_path = config_path, name = name, metadata = metadata) - self._buffer = self._check_type(buffer, bytes) + self._buffer = buffer @property def maximum_address(self) -> int: @@ -66,7 +66,6 @@ class BufferDataLayer(interfaces.layers.DataLayerInterface): 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 diff --git a/volatility/framework/layers/resources.py b/volatility/framework/layers/resources.py index 6f6393086..e5e14b07b 100644 --- a/volatility/framework/layers/resources.py +++ b/volatility/framework/layers/resources.py @@ -32,7 +32,7 @@ import zipfile from typing import List, Optional from volatility import framework -from volatility.framework import validity, constants +from volatility.framework import constants try: import magic @@ -56,7 +56,7 @@ class ResourceAccessor(object): """Object for openning URLs as files (downloading locally first if necessary)""" def __init__(self, - progress_callback: Optional[validity.ProgressCallback] = None, + progress_callback: Optional[constants.ProgressCallback] = None, context: Optional[ssl.SSLContext] = None) -> None: """Creates a resource accessor diff --git a/volatility/framework/layers/scanners/__init__.py b/volatility/framework/layers/scanners/__init__.py index aa085f7dc..0528f8960 100644 --- a/volatility/framework/layers/scanners/__init__.py +++ b/volatility/framework/layers/scanners/__init__.py @@ -30,7 +30,7 @@ class BytesScanner(layers.ScannerInterface): def __init__(self, needle: bytes) -> None: super().__init__() - self.needle = self._check_type(needle, bytes) + self.needle = needle def __call__(self, data: bytes, data_offset: int) -> Generator[int, None, None]: """Runs through the data looking for the needle, and yields all offsets where the needle is found @@ -48,7 +48,7 @@ class RegExScanner(layers.ScannerInterface): def __init__(self, pattern: bytes, flags: int = 0) -> None: super().__init__() - self.regex = re.compile(self._check_type(pattern, bytes), self._check_type(flags, int)) + self.regex = re.compile(pattern, flags) def __call__(self, data: bytes, data_offset: int) -> Generator[int, None, None]: """Runs through the data looking for the needle, and yields all offsets where the needle is found @@ -65,10 +65,8 @@ class MultiStringScanner(layers.ScannerInterface): def __init__(self, patterns: List[bytes]) -> None: super().__init__() - self._check_type(patterns, list) self._patterns = multiregexp.MultiRegexp() for pattern in patterns: - self._check_type(pattern, bytes) self._patterns.add_pattern(pattern) self._patterns.preprocess() diff --git a/volatility/framework/layers/vmware.py b/volatility/framework/layers/vmware.py index 8afc222ee..dbf9e0d51 100644 --- a/volatility/framework/layers/vmware.py +++ b/volatility/framework/layers/vmware.py @@ -21,7 +21,7 @@ import struct from typing import Any, Dict, List, Optional -from volatility.framework import interfaces, validity +from volatility.framework import interfaces, constants from volatility.framework.configuration import requirements from volatility.framework.layers import physical, segmented, resources from volatility.framework.symbols import native @@ -129,8 +129,7 @@ class VmwareStacker(interfaces.automagic.StackerLayerInterface): def stack(cls, context: interfaces.context.ContextInterface, layer_name: str, - progress_callback: validity.ProgressCallback = None) \ - -> Optional[interfaces.layers.DataLayerInterface]: + progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: """Attempt to stack this based on the starting information""" memlayer = context.memory[layer_name] if not isinstance(memlayer, physical.FileLayer): diff --git a/volatility/framework/objects/__init__.py b/volatility/framework/objects/__init__.py index e5c9ab465..82f5d7b4e 100644 --- a/volatility/framework/objects/__init__.py +++ b/volatility/framework/objects/__init__.py @@ -271,7 +271,6 @@ class Pointer(Integer): object_info: interfaces.objects.ObjectInformation, data_format: DataFormatInfo, subtype: Optional[templates.ObjectTemplate] = None) -> None: - self._check_type(subtype, templates.ObjectTemplate) super().__init__(context = context, object_info = object_info, type_name = type_name, data_format = data_format) self._vol['subtype'] = subtype @@ -400,7 +399,6 @@ class Enumeration(interfaces.objects.ObjectInterface, int): def __new__(cls, context: interfaces.context.ContextInterface, type_name: str, object_info: interfaces.objects.ObjectInformation, base_type: interfaces.objects.Template, choices: Dict[str, int], **kwargs) -> 'Enumeration': - cls._check_class(base_type.vol.object_class, Integer) value = base_type(context = context, object_info = object_info) return int.__new__(cls, value) # type: ignore @@ -410,9 +408,7 @@ class Enumeration(interfaces.objects.ObjectInterface, int): super().__init__(context, type_name, object_info) self._inverse_choices = {} # type: Dict[int, str] - for k, v in self._check_type(choices, dict).items(): - self._check_type(k, str) - self._check_type(v, int) + for k, v in choices.items(): if v in self._inverse_choices: # Technically this shouldn't be a problem, but since we inverse cache # and can't map one value to two possibilities we throw an exception during build @@ -478,9 +474,8 @@ class Array(interfaces.objects.ObjectInterface, abc.Sequence): object_info: interfaces.objects.ObjectInformation, count: int = 0, subtype: templates.ObjectTemplate = None) -> None: - self._check_type(subtype, templates.ObjectTemplate) super().__init__(context = context, type_name = type_name, object_info = object_info) - self._vol['count'] = self._check_type(count, int) + self._vol['count'] = count self._vol['subtype'] = subtype # This overrides the little known Sequence.count(val) that returns the number of items in the list that match val @@ -493,7 +488,7 @@ class Array(interfaces.objects.ObjectInterface, abc.Sequence): @count.setter def count(self, value: int) -> None: """Sets the count to a specific value""" - self._vol['count'] = self._check_type(value, int) + self._vol['count'] = value class VolTemplateProxy(interfaces.objects.ObjectInterface.VolTemplateProxy): @@ -574,7 +569,7 @@ class Struct(interfaces.objects.ObjectInterface): members: Dict[str, Tuple[int, interfaces.objects.Template]]) -> None: super().__init__( context = context, type_name = type_name, object_info = object_info, size = size, members = members) - self._check_members(members) + # self._check_members(members) self._concrete_members = {} # type: Dict[str, Dict] def has_member(self, member_name: str) -> bool: @@ -628,10 +623,10 @@ class Struct(interfaces.objects.ObjectInterface): def _check_members(cls, members: Dict[str, Tuple[int, interfaces.objects.Template]]) -> None: # Members should be an iterable mapping of symbol names to tuples of (relative_offset, ObjectTemplate) # An object template is a callable that when called with a context, offset, layer_name and type_name - if not isinstance(members, abc.Mapping): - raise TypeError("Struct members parameter must be a mapping: {}".format(type(members))) - if not all([(isinstance(member, tuple) and len(member) == 2) for member in members.values()]): - raise TypeError("Struct members must be a tuple of relative_offsets and templates") + assert isinstance(members, abc.Mapping) + "Struct members parameter must be a mapping: {}".format(type(members)) + assert all([(isinstance(member, tuple) and len(member) == 2) for member in members.values()]) + "Struct members must be a tuple of relative_offsets and templates" def member(self, attr: str = 'member') -> object: """Specifically named method for retrieving members.""" diff --git a/volatility/framework/objects/templates.py b/volatility/framework/objects/templates.py index 848dd11f9..32e6dd707 100644 --- a/volatility/framework/objects/templates.py +++ b/volatility/framework/objects/templates.py @@ -21,12 +21,12 @@ import logging from typing import Any, ClassVar, Dict, List, Type -from volatility.framework import interfaces, validity, exceptions +from volatility.framework import interfaces, exceptions vollog = logging.getLogger(__name__) -class ObjectTemplate(interfaces.objects.Template, validity.ValidityRoutines): +class ObjectTemplate(interfaces.objects.Template): """Factory class that produces objects that adhere to the Object interface on demand This is effectively a method of currying, but adds more structure to avoid abuse. @@ -39,7 +39,6 @@ class ObjectTemplate(interfaces.objects.Template, validity.ValidityRoutines): def __init__(self, object_class: Type[interfaces.objects.ObjectInterface], type_name: str, **arguments) -> None: super().__init__(type_name = type_name, **arguments) - self._check_class(object_class, interfaces.objects.ObjectInterface) self._arguments['object_class'] = object_class @property diff --git a/volatility/framework/plugins/__init__.py b/volatility/framework/plugins/__init__.py index d00d4c3b8..5b53c6fc8 100644 --- a/volatility/framework/plugins/__init__.py +++ b/volatility/framework/plugins/__init__.py @@ -25,14 +25,14 @@ These modules should only be imported from volatility.plugins NOT volatility.fra import logging from typing import List, Type -from volatility.framework import interfaces, automagic, exceptions, constants, validity +from volatility.framework import interfaces, automagic, exceptions, constants vollog = logging.getLogger(__name__) def run_plugin(context: interfaces.context.ContextInterface, automagics: List[interfaces.automagic.AutomagicInterface], plugin: Type[interfaces.plugins.PluginInterface], base_config_path: str, - progress_callback: validity.ProgressCallback, + progress_callback: constants.ProgressCallback, file_consumer: interfaces.plugins.FileConsumerInterface) -> interfaces.plugins.PluginInterface: """Constructs a plugin object based on the parameters diff --git a/volatility/framework/plugins/windows/cmdline.py b/volatility/framework/plugins/windows/cmdline.py index 3516d1394..0f8545555 100644 --- a/volatility/framework/plugins/windows/cmdline.py +++ b/volatility/framework/plugins/windows/cmdline.py @@ -20,9 +20,8 @@ from typing import List -import volatility.framework.constants as constants import volatility.framework.interfaces.plugins as interfaces_plugins -from volatility.framework import exceptions, renderers, interfaces +from volatility.framework import constants, exceptions, renderers, interfaces from volatility.framework.configuration import requirements from volatility.framework.objects import utility from volatility.plugins.windows import pslist diff --git a/volatility/framework/plugins/windows/poolscanner.py b/volatility/framework/plugins/windows/poolscanner.py index 3adf315b4..cc35b0b35 100644 --- a/volatility/framework/plugins/windows/poolscanner.py +++ b/volatility/framework/plugins/windows/poolscanner.py @@ -24,7 +24,7 @@ from typing import Dict, Generator, List, Optional, Tuple import volatility.plugins.windows.handles as handles -from volatility.framework import constants, interfaces, renderers, validity, exceptions, symbols +from volatility.framework import constants, interfaces, renderers, exceptions, symbols from volatility.framework.configuration import requirements from volatility.framework.interfaces import plugins, configuration from volatility.framework.layers import scanners @@ -51,7 +51,7 @@ class PoolHeaderSymbolTable(intermed.IntermediateSymbolTable): self.set_type_class('_POOL_HEADER', extensions._POOL_HEADER) -class PoolConstraint(validity.ValidityRoutines): +class PoolConstraint: """Class to maintain tag/size/index/type information about Pool header tags""" def __init__(self, @@ -62,7 +62,7 @@ class PoolConstraint(validity.ValidityRoutines): size: Optional[Tuple[Optional[int], Optional[int]]] = None, index: Optional[Tuple[Optional[int], Optional[int]]] = None, alignment: Optional[int] = 1) -> None: - self.tag = self._check_type(tag, bytes) + self.tag = tag self.type_name = type_name self.object_type = object_type self.page_type = page_type @@ -204,7 +204,7 @@ class PoolScanner(plugins.PluginInterface): symbol_table: str, pool_constraints: List[PoolConstraint], alignment: int = 8, - progress_callback: Optional[validity.ProgressCallback] = None) \ + progress_callback: Optional[constants.ProgressCallback] = None) \ -> Generator[Tuple[PoolConstraint, interfaces.objects.ObjectInterface], None, None]: """Returns the _POOL_HEADER object (based on the symbol_table template) after scanning through layer_name returning all headers that match any of the constraints provided. Only one constraint can be provided per tag""" diff --git a/volatility/framework/plugins/windows/vadyarascan.py b/volatility/framework/plugins/windows/vadyarascan.py index abbe168b9..a7be33000 100644 --- a/volatility/framework/plugins/windows/vadyarascan.py +++ b/volatility/framework/plugins/windows/vadyarascan.py @@ -25,7 +25,6 @@ from volatility.framework import interfaces, renderers from volatility.framework.configuration import requirements from volatility.framework.layers import resources from volatility.framework.renderers import format_hints -from volatility.framework.symbols.windows import extensions from volatility.plugins import yarascan from volatility.plugins.windows import pslist @@ -90,8 +89,6 @@ class VadYaraScan(interfaces.plugins.PluginInterface): def get_vad_maps(self, task: Any) -> Iterable[Tuple[int, int]]: - task = self._check_type(task, extensions._EPROCESS) - vad_root = task.get_vad_root() for vad in vad_root.traverse(): end = vad.get_end() diff --git a/volatility/framework/symbols/__init__.py b/volatility/framework/symbols/__init__.py index ff3a9bf19..07263ac97 100644 --- a/volatility/framework/symbols/__init__.py +++ b/volatility/framework/symbols/__init__.py @@ -24,7 +24,7 @@ import enum import logging from typing import Any, Dict, Iterable, Iterator, Set, TypeVar -from volatility.framework import constants, exceptions, interfaces, objects, validity +from volatility.framework import constants, exceptions, interfaces, objects vollog = logging.getLogger(__name__) @@ -39,7 +39,7 @@ SymbolSpaceReturnType = TypeVar("SymbolSpaceReturnType", interfaces.objects.Temp interfaces.symbols.SymbolInterface, Dict[str, Any]) -class SymbolSpace(interfaces.symbols.SymbolSpaceInterface, validity.ValidityRoutines): +class SymbolSpace(interfaces.symbols.SymbolSpaceInterface): """Handles an ordered collection of SymbolTables This collection is ordered so that resolution of symbols can @@ -55,8 +55,6 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface, validity.ValidityRout def free_table_name(self, prefix: str = "layer") -> str: """Returns an unused table name to ensure no collision occurs when inserting a symbol table""" - self._check_type(prefix, str) - count = 1 while prefix + str(count) in self: count += 1 diff --git a/volatility/framework/symbols/intermed.py b/volatility/framework/symbols/intermed.py index e58394850..d0cf3ca48 100644 --- a/volatility/framework/symbols/intermed.py +++ b/volatility/framework/symbols/intermed.py @@ -245,7 +245,7 @@ class ISFormatTable(interfaces.symbols.SymbolTableInterface, metaclass = ABCMeta table_mapping: Optional[Dict[str, str]] = None) -> None: self._json_object = json_object self._validate_json() - self.name = self._check_type(name, str) + self.name = name nt = native_types or self._get_natives() if nt is None: raise ValueError("Native table not provided") diff --git a/volatility/framework/symbols/windows/extensions/__init__.py b/volatility/framework/symbols/windows/extensions/__init__.py index 1846a643e..f73160103 100644 --- a/volatility/framework/symbols/windows/extensions/__init__.py +++ b/volatility/framework/symbols/windows/extensions/__init__.py @@ -269,16 +269,17 @@ class _MMVAD_SHORT(objects.Struct): if self.has_member("StartingVpn"): if self.has_member("StartingVpnHigh"): - return (self.StartingVpn << 12) | (self.StartingVpnHigh << 44) + return (self.StartingVpn.cast("unsigned int") << 12) | (self.StartingVpnHigh.cast("unsigned int") << 44) else: - return self.StartingVpn << 12 + return self.StartingVpn.cast("unsigned int") << 12 elif self.has_member("Core"): if self.Core.has_member("StartingVpnHigh"): - return (self.Core.StartingVpn << 12) | (self.Core.StartingVpnHigh << 44) + return (self.Core.StartingVpn.cast("unsigned int") << 12) | ( + self.Core.StartingVpnHigh.cast("unsigned int") << 44) else: - return self.Core.StartingVpn << 12 + return self.Core.StartingVpn.cast("unsigned int") << 12 raise AttributeError("Unable to find the starting VPN member") @@ -288,16 +289,17 @@ class _MMVAD_SHORT(objects.Struct): if self.has_member("EndingVpn"): if self.has_member("EndingVpnHigh"): - return (self.EndingVpn << 12) | (self.EndingVpnHigh << 44) + return (self.EndingVpn.cast("unsigned int") << 12) | (self.EndingVpnHigh.cast("unsigned int") << 44) else: - return ((self.EndingVpn + 1) << 12) - 1 + return ((self.EndingVpn.cast("unsigned int") + 1) << 12) - 1 elif self.has_member("Core"): if self.Core.has_member("EndingVpnHigh"): - return (self.Core.EndingVpn << 12) | (self.Core.EndingVpnHigh << 44) + return (self.Core.EndingVpn.cast("unsigned int") << 12) | ( + self.Core.EndingVpnHigh.cast("unsigned int") << 44) else: - return ((self.Core.EndingVpn + 1) << 12) - 1 + return ((self.Core.EndingVpn.cast("unsigned int") + 1) << 12) - 1 raise AttributeError("Unable to find the ending VPN member") diff --git a/volatility/framework/symbols/wrappers.py b/volatility/framework/symbols/wrappers.py index e229231c2..33f204ed2 100644 --- a/volatility/framework/symbols/wrappers.py +++ b/volatility/framework/symbols/wrappers.py @@ -18,20 +18,15 @@ # specific language governing rights and limitations under the License. # -import collections from typing import List, Mapping -from volatility.framework import interfaces, validity +from volatility.framework import interfaces -class Flags(validity.ValidityRoutines): +class Flags: """Object that converts an integer into a set of flags based on their masks""" def __init__(self, choices: Mapping[str, int]) -> None: - self._check_type(choices, collections.Mapping) - for k, v in choices.items(): - self._check_type(k, str) - self._check_type(v, int) self._choices = interfaces.objects.ReadOnlyMapping(choices) @property diff --git a/volatility/framework/validity.py b/volatility/framework/validity.py deleted file mode 100644 index 2bc5a4cdf..000000000 --- a/volatility/framework/validity.py +++ /dev/null @@ -1,61 +0,0 @@ -# This file was contributed to the Volatility Framework Version 3. -# Copyright (C) 2018 Volatility Foundation. -# -# THE LICENSED WORK IS PROVIDED UNDER THE TERMS OF THE Volatility Contributors -# Public License V1.0("LICENSE") AS FIRST COMPLETED BY: Volatility Foundation, -# Inc. ANY USE, PUBLIC DISPLAY, PUBLIC PERFORMANCE, REPRODUCTION OR DISTRIBUTION -# OF, OR PREPARATION OF SUBSEQUENT WORKS, DERIVATIVE WORKS OR DERIVED WORKS BASED -# ON, THE LICENSED WORK CONSTITUTES RECIPIENT'S ACCEPTANCE OF THIS LICENSE AND ITS -# TERMS, WHETHER OR NOT SUCH RECIPIENT READS THE TERMS OF THE LICENSE. "LICENSED -# WORK,” “RECIPIENT" AND “DISTRIBUTOR" ARE DEFINED IN THE LICENSE. A COPY OF THE -# LICENSE IS LOCATED IN THE TEXT FILE ENTITLED "LICENSE.txt" ACCOMPANYING THE -# CONTENTS OF THIS FILE. IF A COPY OF THE LICENSE DOES NOT ACCOMPANY THIS FILE, A -# COPY OF THE LICENSE MAY ALSO BE OBTAINED AT THE FOLLOWING WEB SITE: -# https://www.volatilityfoundation.org/license/vcpl_v1.0 -# -# Software distributed under the License is distributed on an "AS IS" basis, -# WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for the -# specific language governing rights and limitations under the License. -# -"""A set of classes providing consistent type checking and error handling for type/class validity -""" -from typing import Callable, Optional, TypeVar, Type - -ProgressCallback = Optional[Callable[[float, str], None]] - - -class ValidityRoutines(object): - """Class to hold all validation routines, such as type checking - - Contains only private class methods, including `_check_type(cls, value, valid_type)` and - `_check_class(cls, klass, valid_class)`. These may eventually be made obsolete by PEP 484 - and appropriate static type verification by software such as mypy. - - These are currently implemented by assertions that will be optimized out of production code. - """ - - V = TypeVar('V') - - @classmethod - def _check_type(cls, value: V, valid_type: Type) -> V: - """Checks that value is an instance of valid_type, and returns value if it is, or throws a TypeError otherwise - - Args: - value: The value of which to validate the type - valid_type: The type against which to validate - """ - assert isinstance( - value, valid_type), cls.__name__ + " expected " + valid_type.__name__ + ", not " + type(value).__name__ - return value - - @classmethod - def _check_class(cls, klass: Type, valid_class: Type) -> Type: - """Checks that class is an instance of valid_class, and returns klass if it is, or throws a TypeError otherwise - - Args: - klass: Class to validate - valid_class: Valid class against which to check class validity - """ - assert issubclass(klass, - valid_class), cls.__name__ + " expected " + valid_class.__name__ + ", not " + klass.__name__ - return klass