diff --git a/volatility/framework/automagic/__init__.py b/volatility/framework/automagic/__init__.py index 9fe98fc5f..005298297 100644 --- a/volatility/framework/automagic/__init__.py +++ b/volatility/framework/automagic/__init__.py @@ -10,8 +10,9 @@ loading of file format types) as well as a module to reconstruct layers based on import logging import sys import traceback +import typing -from volatility.framework import class_subclasses, import_files, interfaces +from volatility.framework import class_subclasses, import_files, interfaces, validity from volatility.framework.automagic import construct_layers, stacker, windows, pdbscan from volatility.framework.configuration import requirements @@ -29,7 +30,8 @@ linux_automagic = ['ConstructionMagic', 'LinuxSymbolFinder'] -def available(context): +def available(context: interfaces.context.ContextInterface) \ + -> typing.List[typing.Type[interfaces.automagic.AutomagicInterface]]: """Returns an ordered list of all subclasses of :class:`~volatility.framework.interfaces.automagic.AutomagicInterface`. The order is based on the priority attributes of the subclasses, in order to ensure the automagics are listed in @@ -45,7 +47,12 @@ def available(context): key = lambda x: x.priority) -def run(automagics, context, configurable, config_path, progress_callback = None): +def run(automagics: typing.List[interfaces.automagic.AutomagicInterface], + context: interfaces.context.ContextInterface, + configurable: typing.Union[interfaces.configuration.ConfigurableInterface, + typing.Type[interfaces.configuration.ConfigurableInterface]], + config_path: str, + progress_callback: validity.ProgressCallback = None) -> typing.List[traceback.TracebackException]: """Runs through the list of `automagics` in order, allowing them to make changes to the context :param automagics: A list of :class:`~volatility.framework.interfaces.automagic.AutomagicInterface` objects @@ -66,14 +73,16 @@ def run(automagics, context, configurable, config_path, progress_callback = None raise TypeError("Automagics must only contain AutomagicInterface subclasses") if (not isinstance(configurable, interfaces.configuration.ConfigurableInterface) - and not issubclass(configurable, interfaces.configuration.ConfigurableInterface)): + and not issubclass(configurable, interfaces.configuration.ConfigurableInterface)): raise TypeError("Automagic operates on configurables only") # TODO: Fix need for top level config element just because we're using a MultiRequirement to group the # configurable's config requirements - configurable_class = configurable + configurable_class: typing.Type[interfaces.configuration.ConfigurableInterface] if isinstance(configurable, interfaces.configuration.ConfigurableInterface): configurable_class = configurable.__class__ + else: + configurable_class = configurable requirement = requirements.MultiRequirement(name = configurable_class.__name__) for req in configurable.get_requirements(): requirement.add_requirement(req) diff --git a/volatility/framework/automagic/construct_layers.py b/volatility/framework/automagic/construct_layers.py index 17faa067b..f9acbf058 100644 --- a/volatility/framework/automagic/construct_layers.py +++ b/volatility/framework/automagic/construct_layers.py @@ -2,6 +2,7 @@ of a :class:`~volatility.framework.interfaces.configuration.ConfigurableInterface`.""" import logging +import typing from volatility.framework import constants from volatility.framework import interfaces @@ -20,7 +21,11 @@ class ConstructionMagic(interfaces.automagic.AutomagicInterface): """ priority = 0 - def __call__(self, context, config_path, requirement, progress_callback = None, optional = False): + def __call__(self, + context: interfaces.context.ContextInterface, + config_path: str, + requirement: interfaces.configuration.RequirementInterface, + progress_callback = None, optional = False) -> typing.List[str]: result = [] if requirement.unsatisfied(context, config_path): # Having called validate at the top level tells us both that we need to dig deeper diff --git a/volatility/framework/automagic/linux.py b/volatility/framework/automagic/linux.py index 25a2158a6..2df88daef 100644 --- a/volatility/framework/automagic/linux.py +++ b/volatility/framework/automagic/linux.py @@ -1,6 +1,7 @@ import logging +import typing -from volatility.framework import interfaces, constants +from volatility.framework import interfaces, constants, validity from volatility.framework.automagic import linux_symbol_cache from volatility.framework.layers import intel, scanners from volatility.framework.symbols import linux @@ -12,19 +13,26 @@ class LinuxSymbolFinder(interfaces.automagic.AutomagicInterface): """Linux symbol loader based on uname signature strings""" priority = 40 - def __init__(self, context, config_path): + def __init__(self, + context: interfaces.context.ContextInterface, + config_path: str) -> None: super().__init__(context, config_path) - self._requirements = None - self._linux_banners_ = None + self._requirements: typing.List[ + typing.Tuple[str, str, interfaces.configuration.ConstructableRequirementInterface]] = [] + self._linux_banners_: linux_symbol_cache.LinuxBanners = {} @property - def _linux_banners(self): + def _linux_banners(self) -> linux_symbol_cache.LinuxBanners: """Creates a cached copy of the results, but only it's been requested""" if self._linux_banners_ is None: self._linux_banners_ = linux_symbol_cache.LinuxSymbolCache.load_linux_banners() return self._linux_banners_ - def __call__(self, context, config_path, requirement, progress_callback = None): + def __call__(self, + context: interfaces.context.ContextInterface, + config_path: str, + requirement: interfaces.configuration.RequirementInterface, + progress_callback: validity.ProgressCallback = None) -> None: """Searches for LinuxSymbolRequirements and attempt to populate them""" self._requirements = self.find_requirements(context, config_path, requirement, (interfaces.configuration.TranslationLayerRequirement, @@ -37,13 +45,18 @@ class LinuxSymbolFinder(interfaces.automagic.AutomagicInterface): for (tl_path, tl_sub_path, tl_requirement) in self._requirements: # Find the TranslationLayer sibling to the SymbolRequirement if (isinstance(tl_requirement, interfaces.configuration.TranslationLayerRequirement) and - tl_path == path): + tl_path == path): if context.config.get(tl_sub_path, None): self._banner_scan(context, path, requirement, context.config[tl_sub_path], progress_callback) break - def _banner_scan(self, context, config_path, requirement, layer_name, progress_callback = None): + def _banner_scan(self, + context: interfaces.context.ContextInterface, + config_path: str, + requirement: interfaces.configuration.ConstructableRequirementInterface, + layer_name: str, + progress_callback: validity.ProgressCallback = None) -> None: """Accepts a context, config_path and SymbolRequirement, with a constructed layer_name and scans the layer for linux banners""" @@ -90,7 +103,11 @@ class LintelStacker(interfaces.automagic.StackerLayerInterface): stack_order = 12 @classmethod - def stack(cls, context, layer_name, progress_callback = None): + def stack(cls, + context: interfaces.context.ContextInterface, + layer_name: str, + progress_callback: validity.ProgressCallback = None) \ + -> typing.Union[None, typing.Type[interfaces.layers.DataLayerInterface]]: """Attempts to identify linux within this layer""" layer = context.memory[layer_name] join = interfaces.configuration.path_join @@ -116,11 +133,11 @@ class LintelStacker(interfaces.automagic.StackerLayerInterface): kaslr_shift, _ = LinuxUtilities.find_aslr(context, table_name, layer_name, progress_callback = progress_callback) + layer_class: typing.Type = intel.Intel if ('init_level4_pgt' in table.symbols): layer_class = intel.Intel32e dtb_symbol_name = 'init_level4_pgt' else: - layer_class = intel.Intel dtb_symbol_name = 'swapper_pg_dir' dtb = LinuxUtilities.virtual_to_physical_address(table.get_symbol(dtb_symbol_name).address + @@ -144,7 +161,12 @@ class LinuxUtilities(object): """Class with multiple useful linux functions""" @classmethod - def find_aslr(cls, context, symbol_table, layer_name, progress_callback = None): + def find_aslr(cls, + context: interfaces.context.ContextInterface, + symbol_table: str, + layer_name: str, + progress_callback: validity.ProgressCallback = None) \ + -> typing.Tuple[typing.Union[None, int], typing.Union[None, int]]: """Determines the offset of the actual DTB in physical space and its symbol offset""" init_task_symbol = symbol_table + constants.BANG + 'init_task' table_dtb = context.symbol_space.get_symbol(init_task_symbol).address @@ -174,7 +196,7 @@ class LinuxUtilities(object): return None, None @classmethod - def virtual_to_physical_address(cls, addr): + def virtual_to_physical_address(cls, addr: int) -> int: """Converts a virtual linux address to a physical one (does not account of ASLR)""" if addr > 0xffffffff80000000: return addr - 0xffffffff80000000 diff --git a/volatility/framework/automagic/linux_symbol_cache.py b/volatility/framework/automagic/linux_symbol_cache.py index c37556197..ab7dfc32d 100644 --- a/volatility/framework/automagic/linux_symbol_cache.py +++ b/volatility/framework/automagic/linux_symbol_cache.py @@ -1,6 +1,7 @@ import logging import os import pickle +import typing from urllib import parse from volatility.framework import constants, exceptions, interfaces @@ -8,6 +9,8 @@ from volatility.framework.symbols import intermed vollog = logging.getLogger(__name__) +LinuxBanners = typing.Dict[bytes, typing.List[str]] + class LinuxSymbolCache(interfaces.automagic.AutomagicInterface): """Runs through all Linux symbols tables and caches their banners""" @@ -17,8 +20,8 @@ class LinuxSymbolCache(interfaces.automagic.AutomagicInterface): priority = 0 @classmethod - def load_linux_banners(cls): - linuxbanners = {} + def load_linux_banners(cls) -> LinuxBanners: + linuxbanners: LinuxBanners = {} if os.path.exists(constants.LINUX_BANNERS_PATH): with open(constants.LINUX_BANNERS_PATH, "rb") as f: # We use pickle over JSON because we're dealing with bytes objects diff --git a/volatility/framework/automagic/nlpdtbfinder.py b/volatility/framework/automagic/nlpdtbfinder.py index 0d9702c55..054d32825 100644 --- a/volatility/framework/automagic/nlpdtbfinder.py +++ b/volatility/framework/automagic/nlpdtbfinder.py @@ -1,19 +1,22 @@ import logging import math import struct +import typing -from volatility.framework import exceptions, interfaces +from volatility.framework import exceptions, interfaces, validity from volatility.framework.configuration import requirements from volatility.framework.layers import intel vollog = logging.getLogger(__name__) -validity_tests = {intel.Intel: [], - intel.IntelPAE: [], - intel.Intel32e: [(0b10111011, 0b00100011), - (0b1, 0b1), - (0b1111011, 0b1100011), - (0b1111011, 0b1100011)]} +validity_tests: typing.Dict[typing.Type[intel.Intel], + typing.List[typing.Tuple[int, int]]] = {intel.Intel: [], + intel.IntelPAE: [], + intel.Intel32e: [ + (0b10111011, 0b00100011), + (0b1, 0b1), + (0b1111011, 0b1100011), + (0b1111011, 0b1100011)]} class NlpDtbScanner(interfaces.layers.ScannerInterface): @@ -28,12 +31,14 @@ class NlpDtbScanner(interfaces.layers.ScannerInterface): # Overlap must be a multiple of the page size overlap = 0x4000 - def __init__(self, layer_class, physical_layer): + def __init__(self, + layer_class: typing.Type[intel.Intel], + physical_layer: interfaces.layers.DataLayerInterface) -> None: super().__init__() self._layer_class = layer_class self._physical_layer = physical_layer - def test_entries(self, valid_entries): + def test_entries(self, valid_entries: typing.List[typing.Tuple[int, int]]) -> bool: """Scans through valid_entries, descending to see whether one can be successfully mapped to completion Returns the first valid DTB or None is no valid DTBs could be found @@ -43,12 +48,12 @@ class NlpDtbScanner(interfaces.layers.ScannerInterface): return True return False - def _get_mask(self, value): + def _get_mask(self, value: int) -> int: """Returns a value that correctly masks all numbers less than that of value""" bits = int(math.ceil(math.log(value, 2))) return (1 << (bits + 1)) - 1 - def test_entry(self, entry, level = 0): + def test_entry(self, entry: int, level: int = 0) -> bool: """Tests an individual entry at a particular level in the structure""" name, size, large_page = self._layer_class.structure[level] @@ -89,7 +94,8 @@ class NlpDtbScanner(interfaces.layers.ScannerInterface): return True return False - def __call__(self, data, data_offset): + def __call__(self, data: bytes, data_offset: int) \ + -> typing.Generator[typing.Tuple[int, typing.List[typing.Tuple[int, int]]], None, None]: structure = self._layer_class.structure name, size, large_page = structure[0] @@ -105,7 +111,7 @@ class NlpDtbScanner(interfaces.layers.ScannerInterface): if len(data[page_offset:page_offset + calcsize]) < calcsize: continue entries = struct.unpack('<' + str(2 ** size) + format_str, data[page_offset:page_offset + calcsize]) - valid_entries = [] + valid_entries: typing.List[typing.Tuple[int, int]] = [] invalid_count = 0 user_count = 0 supervisor_count = 0 @@ -137,7 +143,12 @@ class NlpDtbfinder(interfaces.automagic.AutomagicInterface): """ priority = 11 - def __call__(self, context, config_path, requirement, progress_callback = None): + def __call__(self, + context: interfaces.context.ContextInterface, + config_path: str, + requirement: interfaces.configuration.RequirementInterface, + progress_callback: validity.ProgressCallback = None) \ + -> None: results = {} sub_config_path = interfaces.configuration.path_join(config_path, requirement.name) if (not interfaces.configuration.path_join(sub_config_path, "page_map_offset") in context.config and @@ -175,7 +186,10 @@ class NlpDtbStacker(interfaces.automagic.StackerLayerInterface): stack_order = 13 @classmethod - def stack(cls, context, layer_name, progress_callback = None): + def stack(cls, + context: interfaces.context.ContextInterface, + layer_name: str, + progress_callback: validity.ProgressCallback = None): """Attempts to determine and stack an intel layer on a physical layer where possible""" if isinstance(context.memory[layer_name], intel.Intel): return None diff --git a/volatility/framework/automagic/pdbscan.py b/volatility/framework/automagic/pdbscan.py index ffafb03cb..e6947cb8a 100644 --- a/volatility/framework/automagic/pdbscan.py +++ b/volatility/framework/automagic/pdbscan.py @@ -8,8 +8,9 @@ import logging import math import os import struct +import typing -from volatility.framework import exceptions, layers +from volatility.framework import exceptions, layers, validity from volatility.framework.layers import scanners from volatility.framework.symbols import native, intermed @@ -38,11 +39,12 @@ class PdbSignatureScanner(interfaces.layers.ScannerInterface): _RSDS_format = struct.Struct("<16BI") - def __init__(self, pdb_names): + def __init__(self, pdb_names: typing.List[bytes]) -> None: super().__init__() self._pdb_names = pdb_names - def __call__(self, data, data_offset): + def __call__(self, data: bytes, data_offset: int) \ + -> typing.Generator[typing.Tuple[str, typing.Any, bytes, int], None, None]: sig = data.find(b"RSDS") while sig >= 0: null = data.find(b'\0', sig + 4 + self._RSDS_format.size) @@ -52,7 +54,7 @@ class PdbSignatureScanner(interfaces.layers.ScannerInterface): pdb_name = data[name_offset:null] if pdb_name in self._pdb_names: - ## thie ordering is intentional due to mixed endianness in the GUID + ## this ordering is intentional due to mixed endianness in the GUID (g3, g2, g1, g0, g5, g4, g7, g6, g8, g9, ga, gb, gc, gd, ge, gf, a) = \ self._RSDS_format.unpack(data[sig + 4:name_offset]) @@ -61,7 +63,13 @@ class PdbSignatureScanner(interfaces.layers.ScannerInterface): sig = data.find(b"RSDS", sig + 1) -def scan(ctx, layer_name, page_size, progress_callback = None, start = None, end = None): +def scan(ctx: interfaces.context.ContextInterface, + layer_name: str, + page_size: int, + progress_callback: validity.ProgressCallback = None, + start: typing.Union[int, None] = None, + end: typing.Union[int, None] = None) \ + -> typing.Generator[typing.Dict[str, typing.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 (as listed in `self.pdb_names`) and returns the tuple (GUID, age, pdb_name, signature_offset, mz_offset) @@ -124,11 +132,18 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): suffixes = ['.json', '.json.xz'] """Provides a list of supported suffixes for Intermediate Format data files""" - def __init__(self, context, config_path): + def __init__(self, + context: interfaces.context.ContextInterface, + config_path: str) -> None: super().__init__(context, config_path) - self.valid_kernels = [] + self.valid_kernels: typing.Dict[str, typing.Tuple[int, typing.Dict]] = {} - def recurse_pdb_finder(self, context, config_path, requirement, progress_callback = None): + def recurse_pdb_finder(self, + context: interfaces.context.ContextInterface, + config_path: str, + requirement: interfaces.configuration.RequirementInterface, + progress_callback: validity.ProgressCallback = None) \ + -> typing.Dict[bytes, typing.Dict]: """Traverses the requirement tree, rooted at `requirement` looking for virtual layers that might contain a windows PDB. Returns a list of possible kernel locations in the physical memory @@ -142,7 +157,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): :return: A list of (layer_name, scan_results) """ sub_config_path = interfaces.configuration.path_join(config_path, requirement.name) - results = {} + results: typing.Dict[bytes, typing.Dict] = {} if isinstance(requirement, interfaces.configuration.TranslationLayerRequirement): # Check for symbols in this layer # FIXME: optionally allow a full (slow) scan @@ -160,7 +175,9 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): results.update(self.recurse_pdb_finder(context, sub_config_path, subreq)) return results - def recurse_symbol_fulfiller(self, context): + def recurse_symbol_fulfiller(self, + context: interfaces.context.ContextInterface) \ + -> None: """Fulfills the SymbolRequirements in `self._symbol_requirements` found by the `recurse_symbol_requirements`. This pass will construct any requirements that may need it in the context it was passed @@ -195,7 +212,8 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): else: vollog.debug("No suitable kernel pdb signature found") - def set_kernel_virtual_offset(self, context): + def set_kernel_virtual_offset(self, + context: interfaces.context.ContextInterface) -> None: """Traverses the requirement tree, looking for kernel_virtual_offset values that may need setting and sets it based on the previously identified `valid_kernels`. @@ -210,7 +228,11 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): context.config[kvo_path] = kvo vollog.debug("Setting kernel_virtual_offset to {}".format(hex(kvo))) - def determine_valid_kernels(self, context, potential_kernels, progress_callback = None): + def determine_valid_kernels(self, + context: interfaces.context.ContextInterface, + potential_kernels: typing.Dict[str, typing.Any], + progress_callback: validity.ProgressCallback = None) \ + -> typing.Dict[str, typing.Tuple[int, typing.Any]]: """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 @@ -267,7 +289,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): # TODO: On older windows, this might be \WINDOWS\system32\nt rather than \SystemRoot\system32\nt results = physical_layer.scan(context, scanners.BytesScanner(b"\\SystemRoot\\system32\\nt"), progress_callback = progress_callback) - seen = set() + seen: typing.Set[int] = set() # Because this will launch a scan of the virtual layer, we want to be careful for result in results: # TODO: Identify the specific structure we're finding and document this a bit better diff --git a/volatility/framework/automagic/stacker.py b/volatility/framework/automagic/stacker.py index 102e0e59f..978b17362 100644 --- a/volatility/framework/automagic/stacker.py +++ b/volatility/framework/automagic/stacker.py @@ -8,9 +8,10 @@ once a layer successfully stacks on top of the existing layers, it is removed fr """ import logging +import typing import volatility -from volatility.framework import configuration, interfaces, constants +from volatility.framework import configuration, interfaces, constants, validity from volatility.framework.automagic import construct_layers from volatility.framework.configuration import requirements from volatility.framework.layers import physical @@ -33,7 +34,12 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): priority = 10 page_map_offset = None - def __call__(self, context, config_path, requirement, progress_callback = None): + def __call__(self, + context: interfaces.context.ContextInterface, + config_path: str, + requirement: interfaces.configuration.ConstructableRequirementInterface, + progress_callback: validity.ProgressCallback = None) \ + -> None: """Runs the automagic over the configurable""" # Quick exit if we're not needed @@ -103,7 +109,11 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): "ConstructionMagic")) constructor(context, config_path, requirement) - def find_suitable_requirements(self, stacked_layers, requirement, context, config_path): + def find_suitable_requirements(self, + stacked_layers: typing.List, + requirement: interfaces.configuration.RequirementInterface, + context: interfaces.context.ContextInterface, + config_path: str) -> typing.Union[None, typing.Tuple[str, str]]: """Looks for translation layer requirements and attempts to apply the stacked layers to it. If it succeeds it returns the configuration path and layer name where the stacked nodes were spliced into the tree. @@ -127,9 +137,10 @@ class LayerStacker(interfaces.automagic.AutomagicInterface): result = self.find_suitable_requirements(stacked_layers, req, context, child_config_path) if result: return result + return None @classmethod - def get_requirements(cls): + def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]: # This is not optional for the stacker to run, so optional must be marked as False return [requirements.StringRequirement("single_location", description = "Specifies a base location on which to stack", diff --git a/volatility/framework/configuration/requirements.py b/volatility/framework/configuration/requirements.py index 6597fafb7..39a4a2fa0 100644 --- a/volatility/framework/configuration/requirements.py +++ b/volatility/framework/configuration/requirements.py @@ -6,6 +6,7 @@ etc) as well as indicating what they expect to be in the context (such as partic """ import logging +import typing from volatility.framework.interfaces import configuration as interfaces_configuration diff --git a/volatility/framework/interfaces/automagic.py b/volatility/framework/interfaces/automagic.py index 0f7792f1a..c50bf79ad 100644 --- a/volatility/framework/interfaces/automagic.py +++ b/volatility/framework/interfaces/automagic.py @@ -2,7 +2,7 @@ Automagic objects attempt to automatically fill configuration values that a user has not filled. """ - +import typing from abc import ABCMeta, abstractmethod from volatility.framework import validity @@ -46,7 +46,8 @@ class AutomagicInterface(interfaces_configuration.ConfigurableInterface, metacla def __call__(self, context, config_path, configurable, progress_callback = None): """Runs the automagic over the configurable""" - def find_requirements(self, context, config_path, requirement_root, requirement_type, shortcut = True): + def find_requirements(self, context, config_path, requirement_root, requirement_type, shortcut = True) \ + -> typing.List[typing.Tuple[str, str, interfaces_configuration.ConstructableRequirementInterface]]: """Determines if there is actually an unfulfilled symbol requirement waiting This ensures we do not carry out an expensive search when there is no requirement for a particular symbol table. diff --git a/volatility/framework/interfaces/configuration.py b/volatility/framework/interfaces/configuration.py index a50b7479f..983484c77 100644 --- a/volatility/framework/interfaces/configuration.py +++ b/volatility/framework/interfaces/configuration.py @@ -15,6 +15,7 @@ import logging import random import string import sys +import typing from abc import ABCMeta, abstractmethod from volatility.framework import constants @@ -303,7 +304,7 @@ class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): class InstanceRequirement(RequirementInterface): """Class to represent a single simple type (such as a boolean, a string, an integer or a series of bytes)""" - instance_type = bool + instance_type: typing.Type = bool def add_requirement(self, requirement): """Always raises a TypeError as instance requirements cannot have children""" diff --git a/volatility/framework/interfaces/context.py b/volatility/framework/interfaces/context.py index a4e7544c2..2dc4c35b7 100644 --- a/volatility/framework/interfaces/context.py +++ b/volatility/framework/interfaces/context.py @@ -69,6 +69,9 @@ class ContextInterface(object, metaclass = ABCMeta): Memory constraints may become an issue for this function depending on how much is actually stored in the context""" return copy.deepcopy(self) + def module(self, module_name, layer_name, offset): + """Create a module object """ + class Module(validity.ValidityRoutines, metaclass = ABCMeta): """Maintains state concerning a particular loaded module in memory diff --git a/volatility/framework/objects/__init__.py b/volatility/framework/objects/__init__.py index 59372e239..70eeb57b1 100644 --- a/volatility/framework/objects/__init__.py +++ b/volatility/framework/objects/__init__.py @@ -1,5 +1,6 @@ import logging import struct +import typing from collections import abc from volatility.framework import interfaces @@ -29,7 +30,7 @@ class Function(interfaces.objects.ObjectInterface): class PrimitiveObject(interfaces.objects.ObjectInterface): """PrimitiveObject is an interface for any objects that should simulate a Python primitive""" - _struct_type = int + _struct_type: typing.Type = int def __init__(self, context, type_name, object_info, struct_format): super().__init__(context = context, @@ -493,7 +494,7 @@ class Struct(interfaces.objects.ObjectInterface): member = member(context = self._context, object_info = interfaces.objects.ObjectInformation(layer_name = self.vol.layer_name, offset = mask & ( - self.vol.offset + relative_offset), + self.vol.offset + relative_offset), member_name = attr, parent = self)) self._concrete_members[attr] = member diff --git a/volatility/framework/validity.py b/volatility/framework/validity.py index 6f1098759..150e5e581 100644 --- a/volatility/framework/validity.py +++ b/volatility/framework/validity.py @@ -1,5 +1,8 @@ """A set of classes providing consistent type checking and error handling for type/class validity """ +import typing + +ProgressCallback = typing.Union[typing.Callable[[int, str], None], None] class ValidityRoutines(object):