WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-28 12:34:53 +02:00
Code Issues Packages Projects Releases Wiki Activity

Remove validity checking in favour of static-type checking.

Browse Source 
This removes the assertions that checked plugin devs didn't do anything
overly bad in favour of using mypy to ensure appropriate type-checking.

It also moves the ProgressCallback typing information to constants in
order to avoid circular imports (since constants doesn't import
anything).
This commit is contained in:
Mike Auty
2019-01-20 02:35:49 +00:00
parent 1071dca087
commit c64ed6a8dd
36 changed files with 142 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility/framework/automagic/__init__.py
+2 -2
View File
@@ -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:
volatility/framework/automagic/linux.py
+3 -4
View File
@@ -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'
volatility/framework/automagic/mac.py
+3 -3
View File
@@ -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
volatility/framework/automagic/pdbscan.py
+8 -8
View File
@@ -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))
volatility/framework/automagic/stacker.py
+3 -5
View File
@@ -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")
volatility/framework/automagic/symbol_finder.py
+3 -3
View File
@@ -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"""
volatility/framework/automagic/windows.py
+9 -11
View File
@@ -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(
volatility/framework/configuration/requirements.py
+1 -1
View File
@@ -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 {}
volatility/framework/constants/__init__.py
+3
View File
@@ -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]]
volatility/framework/contexts/__init__.py
+3 -9
View File
@@ -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':
volatility/framework/interfaces/automagic.py
+4 -4
View File
@@ -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.
volatility/framework/interfaces/configuration.py
+18 -31
View File
@@ -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:
volatility/framework/interfaces/context.py
+7 -7
View File
@@ -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__()
volatility/framework/interfaces/layers.py
+9 -13
View File
@@ -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
volatility/framework/interfaces/objects.py
+5 -10
View File
@@ -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
volatility/framework/interfaces/plugins.py
+5 -6
View File
@@ -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"""
volatility/framework/interfaces/renderers.py
+1 -3
View File
@@ -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:
volatility/framework/interfaces/symbols.py
+10 -17
View File
@@ -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
volatility/framework/layers/crash.py
+2 -3
View File
@@ -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:
volatility/framework/layers/intel.py
+2 -3
View File
@@ -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))))
volatility/framework/layers/lime.py
+2 -3
View File
@@ -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:
volatility/framework/layers/physical.py
+1 -2
View File
@@ -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
volatility/framework/layers/resources.py
+2 -2
View File
@@ -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
volatility/framework/layers/scanners/__init__.py
+2 -4
View File
@@ -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()
volatility/framework/layers/vmware.py
+2 -3
View File
@@ -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):
volatility/framework/objects/__init__.py
+8 -13
View File
@@ -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."""
volatility/framework/objects/templates.py
+2 -3
View File
@@ -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
volatility/framework/plugins/__init__.py
+2 -2
View File
@@ -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
volatility/framework/plugins/windows/cmdline.py
+1 -2
View File
@@ -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
volatility/framework/plugins/windows/poolscanner.py
+4 -4
View File
@@ -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"""
volatility/framework/plugins/windows/vadyarascan.py
-3
View File
@@ -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()
volatility/framework/symbols/__init__.py
+2 -4
View File
@@ -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
volatility/framework/symbols/intermed.py
+1 -1
View File
@@ -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")
volatility/framework/symbols/windows/extensions/__init__.py
+10 -8
View File
@@ -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")
volatility/framework/symbols/wrappers.py
+2 -7
View File
@@ -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
volatility/framework/validity.py
-61
View File
@@ -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