diff --git a/volatility/cli/__init__.py b/volatility/cli/__init__.py index 72d52b098..7b78813e6 100644 --- a/volatility/cli/__init__.py +++ b/volatility/cli/__init__.py @@ -14,7 +14,7 @@ import json import logging import os import sys -from typing import Union, Type, Dict +from typing import Any, Dict, Type, Union from urllib import parse, request import volatility.plugins diff --git a/volatility/framework/automagic/mac.py b/volatility/framework/automagic/mac.py index 387b559b5..3d1a9f942 100644 --- a/volatility/framework/automagic/mac.py +++ b/volatility/framework/automagic/mac.py @@ -121,7 +121,7 @@ class MacUtilities(object): if not isinstance(sym_layer, layers.intel.Intel): raise TypeError("Layer name {} is not an intel space") aslr_layer = sym_layer.config['memory_layer'] - _, aslr_shift = cls.find_aslr(context, symbol_table, aslr_layer) + aslr_shift = cls.find_aslr(context, symbol_table, aslr_layer) symbols.mask_symbol_table(sym_table, sym_layer.address_mask, aslr_shift) @@ -149,8 +149,7 @@ class MacUtilities(object): layer_name: str, compare_banner: str = "", compare_banner_offset: int = 0, - progress_callback: validity.ProgressCallback = None) \ - -> Tuple[int, int]: + progress_callback: validity.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 dec7baaa6..fd77032af 100644 --- a/volatility/framework/automagic/pdbscan.py +++ b/volatility/framework/automagic/pdbscan.py @@ -205,7 +205,8 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): context.config[join(sub_config_path, "isf_url")] = isf_path # Construct the appropriate symbol table config_path = interfaces.configuration.parent_path(sub_config_path) - requirement.construct(context, config_path) + if isinstance(requirement, interfaces.configuration.ConstructableRequirementInterface): + requirement.construct(context, config_path) break else: vollog.debug("Required symbol library path not found: {}".format(filter_string)) diff --git a/volatility/framework/automagic/symbol_cache.py b/volatility/framework/automagic/symbol_cache.py index 6bd744789..4641c5c59 100644 --- a/volatility/framework/automagic/symbol_cache.py +++ b/volatility/framework/automagic/symbol_cache.py @@ -4,7 +4,7 @@ import pickle import urllib import urllib.parse import urllib.request -from typing import Dict, List +from typing import Dict, List, Optional from volatility.framework import constants, exceptions, interfaces from volatility.framework.symbols import intermed @@ -21,12 +21,14 @@ class SymbolBannerCache(interfaces.automagic.AutomagicInterface): # The user would run it eventually either way, but running it first means it can be used that run priority = 0 - os = None - symbol_name = "banner_name" - banner_path = None + os: Optional[str] = None + symbol_name: str = "banner_name" + banner_path: Optional[str] = None @classmethod def load_banners(cls) -> BannersType: + if not cls.banner_path: + raise ValueError("Banner_path not appropriately set") banners = {} # type: BannersType if os.path.exists(cls.banner_path): with open(cls.banner_path, "rb") as f: diff --git a/volatility/framework/automagic/symbol_finder.py b/volatility/framework/automagic/symbol_finder.py index c16755c1c..002269b72 100644 --- a/volatility/framework/automagic/symbol_finder.py +++ b/volatility/framework/automagic/symbol_finder.py @@ -1,5 +1,5 @@ import logging -from typing import Any, Iterable, List, Tuple +from typing import Any, Iterable, List, Tuple, Dict, Type, Optional from volatility.framework import interfaces, validity from volatility.framework.automagic import symbol_cache @@ -13,13 +13,13 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface): """Symbol loader based on signature strings""" priority = 40 - banner_config_key = "banner" - banner_cache = None - symbol_class = None + banner_config_key: str = "banner" + banner_cache: Optional[Type[symbol_cache.SymbolBannerCache]] = None + symbol_class: Optional[str] = None def __init__(self, context: interfaces.context.ContextInterface, config_path: str) -> None: super().__init__(context, config_path) - self._requirements = [] # type: List[Tuple[str, interfaces.configuration.ConstructableRequirementInterface]] + self._requirements = [] # type: List[Tuple[str, interfaces.configuration.RequirementInterface]] self._banners = {} # type: symbol_cache.BannersType @property diff --git a/volatility/framework/contexts/__init__.py b/volatility/framework/contexts/__init__.py index 6329d2387..eca54fcb8 100644 --- a/volatility/framework/contexts/__init__.py +++ b/volatility/framework/contexts/__init__.py @@ -103,14 +103,12 @@ class Context(interfaces.context.ContextInterface): object_info = interfaces.objects.ObjectInformation( layer_name = layer_name, offset = offset, native_layer_name = native_layer_name)) - @functools.lru_cache() - def module( - self, # type: ignore # FIXME: mypy #5107 - module_name: str, - layer_name: str, - offset: int, - native_layer_name: Optional[str] = None, - size: Optional[int] = None) -> interfaces.context.ModuleInterface: + def module(self, + module_name: str, + layer_name: str, + offset: int, + native_layer_name: Optional[str] = None, + size: Optional[int] = None) -> interfaces.context.ModuleInterface: """Creates a module object""" if size: return SizedModule( diff --git a/volatility/framework/interfaces/automagic.py b/volatility/framework/interfaces/automagic.py index a70533371..6eb3707b1 100644 --- a/volatility/framework/interfaces/automagic.py +++ b/volatility/framework/interfaces/automagic.py @@ -3,13 +3,11 @@ Automagic objects attempt to automatically fill configuration values that a user has not filled. """ from abc import ABCMeta -from typing import TypeVar, Any, List, Optional, Tuple, Union, Type +from typing import Any, List, Optional, Tuple, Union, Type from volatility.framework import interfaces, validity from volatility.framework.configuration import requirements -R = TypeVar('R', bound = interfaces.configuration.RequirementInterface) - class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metaclass = ABCMeta): """Class that defines an automagic component that can help fulfill a Requirement @@ -49,15 +47,16 @@ class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metacla """Runs the automagic over the configurable""" return [] - # TODO: requirement_type can be made Union[Type[T], Tuple[Type[T], ...]] + # TODO: requirement_type can be made UnionType[Type[T], Tuple[Type[T], ...]] # once mypy properly supports Tuples in instance def find_requirements(self, context: interfaces.context.ContextInterface, config_path: str, requirement_root: interfaces.configuration.RequirementInterface, - requirement_type: Union[Tuple[Type[R], ...], Type[R]], - shortcut: bool = True) -> List[Tuple[str, R]]: + requirement_type: Union[Tuple[Type[interfaces.configuration.RequirementInterface], ...], + Type[interfaces.configuration.RequirementInterface]], + shortcut: bool = True) -> List[Tuple[str, interfaces.configuration.RequirementInterface]]: """Determines if there is actually an unfulfilled requirement waiting This ensures we do not carry out an expensive search when there is no requirement for a particular requirement @@ -73,7 +72,7 @@ class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metacla A list of tuples containing the config_path, sub_config_path and requirement identifying the SymbolRequirements """ sub_config_path = interfaces.configuration.path_join(config_path, requirement_root.name) - results = [] # type: List[Tuple[str, R]] + results = [] # type: List[Tuple[str, interfaces.configuration.RequirementInterface]] recurse = not shortcut if isinstance(requirement_root, requirement_type): if recurse or requirement_root.unsatisfied(context, config_path): diff --git a/volatility/framework/interfaces/configuration.py b/volatility/framework/interfaces/configuration.py index 49e310681..25664bdf5 100644 --- a/volatility/framework/interfaces/configuration.py +++ b/volatility/framework/interfaces/configuration.py @@ -16,7 +16,7 @@ import random import string import sys from abc import ABCMeta, abstractmethod -from typing import Any, Dict, Generator, List, Optional, Type, Union +from typing import Any, ClassVar, Dict, Generator, List, Optional, Type, Union from volatility.framework import constants, interfaces, validity from volatility.framework.interfaces.context import ContextInterface diff --git a/volatility/framework/interfaces/context.py b/volatility/framework/interfaces/context.py index 5df398732..8aa8d650d 100644 --- a/volatility/framework/interfaces/context.py +++ b/volatility/framework/interfaces/context.py @@ -75,7 +75,12 @@ 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: str, layer_name: str, offset: int, size: Optional[int] = None) -> 'ModuleInterface': + def module(self, + module_name: str, + layer_name: str, + offset: int, + native_layer_name: Optional[str] = None, + size: Optional[int] = None) -> 'ModuleInterface': """Create a module object """ diff --git a/volatility/framework/interfaces/objects.py b/volatility/framework/interfaces/objects.py index f196759c1..1deb09372 100644 --- a/volatility/framework/interfaces/objects.py +++ b/volatility/framework/interfaces/objects.py @@ -5,7 +5,7 @@ import collections import collections.abc import logging from abc import ABCMeta, abstractmethod -from typing import Any, List, Mapping +from typing import Any, Dict, List, Mapping, Optional from volatility.framework import constants, validity, interfaces from volatility.framework.interfaces import context as interfaces_context @@ -53,7 +53,12 @@ class ObjectInformation(ReadOnlyMapping): in a single place. These values are based on the :class:`ReadOnlyMapping` class, to prevent their modification. """ - def __init__(self, layer_name, offset, member_name = None, parent = None, native_layer_name = None): + def __init__(self, + layer_name: str, + offset: int, + 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) diff --git a/volatility/framework/interfaces/plugins.py b/volatility/framework/interfaces/plugins.py index c4fafe535..17caf112d 100644 --- a/volatility/framework/interfaces/plugins.py +++ b/volatility/framework/interfaces/plugins.py @@ -7,7 +7,7 @@ They are called and carry out some algorithms on data stored in layers using obj import io import logging from abc import ABCMeta, abstractmethod -from typing import TYPE_CHECKING, List +from typing import List, Optional, TYPE_CHECKING from volatility.framework import exceptions from volatility.framework import validity diff --git a/volatility/framework/objects/__init__.py b/volatility/framework/objects/__init__.py index 9237dc61f..3077b642f 100644 --- a/volatility/framework/objects/__init__.py +++ b/volatility/framework/objects/__init__.py @@ -2,7 +2,7 @@ import collections import logging import struct from collections import abc -from typing import Any, ClassVar, Dict, List, Iterable, Optional, Tuple, Type, Union, overload +from typing import Any, ClassVar, Dict, List, Iterable, Optional, Tuple, Type, Union as TUnion, overload from volatility.framework import interfaces from volatility.framework.interfaces.objects import ObjectInformation @@ -13,8 +13,8 @@ vollog = logging.getLogger(__name__) DataFormatInfo = collections.namedtuple('DataFormatInfo', ['length', 'byteorder', 'signed']) -def convert_data_to_value(data: bytes, struct_type: Type[Union[int, float, bytes, str, bool]], - data_format: DataFormatInfo) -> Union[int, float, bytes, str, bool]: +def convert_data_to_value(data: bytes, struct_type: Type[TUnion[int, float, bytes, str, bool]], + data_format: DataFormatInfo) -> TUnion[int, float, bytes, str, bool]: """Converts a series of bytes to a particular type of value""" if struct_type == int: return int.from_bytes(data, byteorder = data_format.byteorder, signed = data_format.signed) @@ -33,8 +33,8 @@ def convert_data_to_value(data: bytes, struct_type: Type[Union[int, float, bytes return struct.unpack(struct_format, data)[0] -def convert_value_to_data(value: Union[int, float, bytes, str, bool], - struct_type: Type[Union[int, float, bytes, str, bool]], +def convert_value_to_data(value: TUnion[int, float, bytes, str, bool], + struct_type: Type[TUnion[int, float, bytes, str, bool]], data_format: DataFormatInfo) -> bytes: """Converts a particular value to a series of bytes""" if not isinstance(value, struct_type): @@ -87,12 +87,12 @@ class PrimitiveObject(interfaces.objects.ObjectInterface): context = context, type_name = type_name, object_info = object_info, data_format = data_format) self._data_format = data_format - def __new__(cls: 'PrimitiveObject', + def __new__(cls: Type, context: interfaces.context.ContextInterface, type_name: str, object_info: interfaces.objects.ObjectInformation, data_format: DataFormatInfo, - new_value: Union[int, float, bool, bytes, str] = None, + new_value: TUnion[int, float, bool, bytes, str] = None, **kwargs) -> 'PrimitiveObject': """Creates the appropriate class and returns it so that the native type is inherited @@ -122,7 +122,7 @@ class PrimitiveObject(interfaces.objects.ObjectInterface): @classmethod def _unmarshall(cls, context: interfaces.context.ContextInterface, data_format: DataFormatInfo, - object_info: ObjectInformation) -> Union[int, float, bool, bytes, str]: + object_info: ObjectInformation) -> TUnion[int, float, bool, bytes, str]: data = context.memory.read(object_info.layer_name, object_info.offset, data_format.length) return convert_data_to_value(data, cls._struct_type, data_format) @@ -133,7 +133,7 @@ class PrimitiveObject(interfaces.objects.ObjectInterface): """Returns the size of the templated object""" return template.vol.data_format.length - def write(self, value: Union[int, float, bool, bytes, str]) -> None: + def write(self, value: TUnion[int, float, bool, bytes, str]) -> None: """Writes the object into the layer of the context at the current offset""" data = convert_value_to_data(value, self._struct_type, self._data_format) return self._context.memory.write(self.vol.layer_name, self.vol.offset, data) @@ -174,7 +174,7 @@ class Bytes(PrimitiveObject, bytes): data_format = DataFormatInfo(length, "big", False)) self._vol['length'] = length - def __new__(cls: 'Bytes', + def __new__(cls: Type, context: interfaces.context.ContextInterface, type_name: str, object_info: interfaces.objects.ObjectInformation, @@ -215,7 +215,7 @@ class String(PrimitiveObject, str): self._vol['encoding'] = encoding self._vol['errors'] = errors - def __new__(cls, + def __new__(cls: Type, context: interfaces.context.ContextInterface, type_name: str, object_info: interfaces.objects.ObjectInformation, @@ -234,10 +234,9 @@ class String(PrimitiveObject, str): params['errors'] = errors # Pass the encoding and error parameters to the string constructor to appropriately encode the string value = cls._struct_type.__new__( - cls, # type: ignore + cls, cls._unmarshall( - context, data_format = DataFormatInfo(max_length, "big", False), object_info = object_info), - **params) + context, data_format = DataFormatInfo(max_length, "big", False), object_info = object_info), **params) if value.find('\x00') >= 0: value = value[:value.find('\x00')] return value diff --git a/volatility/framework/plugins/timeliner.py b/volatility/framework/plugins/timeliner.py index ae29dbe3f..eff420d3d 100644 --- a/volatility/framework/plugins/timeliner.py +++ b/volatility/framework/plugins/timeliner.py @@ -47,16 +47,18 @@ class Timeliner(interfaces.plugins.PluginInterface): plugin_list = list(framework.class_subclasses(TimeLinerInterface)) # Get the filter from the configuration - def passthrough(_n, _s): + def passthrough(name: str, selected: List[str]) -> bool: return True filter_func = passthrough if selected_list: - def filter_plugins(name, selected): + def filter_plugins(name: str, selected: List[str]) -> bool: return any([s in name for s in selected]) filter_func = filter_plugins + else: + selected_list = [] return [plugin_class for plugin_class in plugin_list if filter_func(plugin_class.__name__, selected_list)] diff --git a/volatility/framework/plugins/windows/handles.py b/volatility/framework/plugins/windows/handles.py index d7f715077..cb90b1ffe 100644 --- a/volatility/framework/plugins/windows/handles.py +++ b/volatility/framework/plugins/windows/handles.py @@ -144,7 +144,7 @@ class Handles(interfaces_plugins.PluginInterface): ptrs = ntkrnlmp.object( type_name = "array", offset = kvo + table_addr, subtype = ntkrnlmp.get_type("pointer"), count = 100) - for i, ptr in enumerate(ptrs): + for i, ptr in enumerate(ptrs): #type: ignore # the first entry in the table is always null. break the # loop when we encounter the first null entry after that if i > 0 and ptr == 0: diff --git a/volatility/framework/plugins/windows/poolscanner.py b/volatility/framework/plugins/windows/poolscanner.py index a83936e07..eb53146ca 100644 --- a/volatility/framework/plugins/windows/poolscanner.py +++ b/volatility/framework/plugins/windows/poolscanner.py @@ -1,6 +1,6 @@ import enum import logging -from typing import Optional, Tuple, List, Generator +from typing import Dict, Generator, List, Optional, Tuple import volatility.plugins.windows.handles as handles diff --git a/volatility/framework/plugins/windows/pstree.py b/volatility/framework/plugins/windows/pstree.py index 2074587dd..0a85d2275 100644 --- a/volatility/framework/plugins/windows/pstree.py +++ b/volatility/framework/plugins/windows/pstree.py @@ -1,4 +1,6 @@ -from volatility.framework import objects +from typing import Dict, Set + +from volatility.framework import objects, interfaces from volatility.framework.renderers import format_hints from volatility.plugins.windows import pslist @@ -8,9 +10,9 @@ class PsTree(pslist.PsList): def __init__(self, *args, **kwargs) -> None: super().__init__(*args, **kwargs) - self._processes = {} - self._levels = {} - self._children = {} + self._processes = {} # type: Dict[int, interfaces.objects.ObjectInterface] + self._levels = {} # type: Dict[int, int] + self._children = {} # type: Dict[int, Set[int]] def find_level(self, pid: objects.Pointer) -> None: """Finds how deep the pid is in the processes list""" diff --git a/volatility/framework/plugins/windows/registry/userassist.py b/volatility/framework/plugins/windows/registry/userassist.py index 52ced92d9..d83cc1c70 100644 --- a/volatility/framework/plugins/windows/registry/userassist.py +++ b/volatility/framework/plugins/windows/registry/userassist.py @@ -3,7 +3,7 @@ import datetime import json import logging import os -from typing import List +from typing import Any, List, Tuple from volatility.framework import exceptions, renderers, constants, interfaces from volatility.framework.configuration import requirements diff --git a/volatility/framework/plugins/windows/ssdt.py b/volatility/framework/plugins/windows/ssdt.py index 53b1717b5..056beebfe 100644 --- a/volatility/framework/plugins/windows/ssdt.py +++ b/volatility/framework/plugins/windows/ssdt.py @@ -23,7 +23,7 @@ class SSDT(plugins.PluginInterface): requirements.SymbolRequirement(name = "nt_symbols", description = "Windows OS") ] - def _generator(self, mods: Iterator[Any]) -> Iterator[Tuple[int, Tuple[int, int, str, str]]]: + def _generator(self, mods: Iterator[Any]) -> Iterator[Tuple[int, Tuple[int, int, Any, Any]]]: layer_name = self.config['primary'] context_modules = [] @@ -67,14 +67,14 @@ class SSDT(plugins.PluginInterface): if is_kernel_64: array_subtype = "long" - def kvo_calulator(func): + def kvo_calulator(func: int) -> int: return kvo + service_table_address + (func >> 4) find_address = kvo_calulator else: array_subtype = "unsigned long" - def passthrough(func): + def passthrough(func: int) -> int: return func find_address = passthrough diff --git a/volatility/framework/plugins/windows/vadinfo.py b/volatility/framework/plugins/windows/vadinfo.py index a8a3d6fc3..379d01d01 100644 --- a/volatility/framework/plugins/windows/vadinfo.py +++ b/volatility/framework/plugins/windows/vadinfo.py @@ -73,13 +73,13 @@ class VadInfo(interfaces.plugins.PluginInterface): def _generator(self, procs): - def passthrough(_): + def passthrough(_: 'interfaces.objects.ObjectInterface') -> bool: return False filter_func = passthrough if self.config.get('address', None) is not None: - def filter_function(x): + def filter_function(x: 'interfaces.objects.ObjectInterface') -> bool: return x.get_start() not in [self.config['address']] filter_func = filter_function diff --git a/volatility/framework/plugins/windows/verinfo.py b/volatility/framework/plugins/windows/verinfo.py index 6aeb9d277..466392fb2 100644 --- a/volatility/framework/plugins/windows/verinfo.py +++ b/volatility/framework/plugins/windows/verinfo.py @@ -1,6 +1,6 @@ import io import logging -from typing import Generator, List, Tuple +from typing import Generator, List, Tuple, Union import volatility.framework.interfaces.plugins as interfaces_plugins import volatility.plugins.windows.moddump as moddump diff --git a/volatility/framework/symbols/windows/extensions/__init__.py b/volatility/framework/symbols/windows/extensions/__init__.py index beb5e2254..bcb257ca9 100644 --- a/volatility/framework/symbols/windows/extensions/__init__.py +++ b/volatility/framework/symbols/windows/extensions/__init__.py @@ -2,7 +2,7 @@ import collections.abc import datetime import functools import logging -from typing import Iterable, Iterator, Optional, Union +from typing import Iterable, Iterator, Optional, Union, Dict from volatility.framework import constants, exceptions, interfaces, objects, renderers, symbols from volatility.framework.layers import intel @@ -103,6 +103,7 @@ class _POOL_HEADER(objects.Struct): return None except (TypeError, exceptions.InvalidAddressException): return None + return None class _KSYSTEM_TIME(objects.Struct): @@ -456,7 +457,7 @@ class _OBJECT_HEADER(objects.Struct): return True - def get_object_type(self, type_map: dict, cookie: int = None) -> str: + def get_object_type(self, type_map: Dict[int, str], cookie: int = None) -> Optional[str]: """Across all Windows versions, the _OBJECT_HEADER embeds details on the type of object (i.e. process, file) but the way its embedded differs between versions. This API abstracts away those details.""" diff --git a/volatility/schemas/__init__.py b/volatility/schemas/__init__.py index 5cdb99fff..2cdb6b112 100644 --- a/volatility/schemas/__init__.py +++ b/volatility/schemas/__init__.py @@ -13,7 +13,7 @@ cached_validation_filepath = os.path.join(constants.CACHE_PATH, "valid_isf.cache def load_cached_validations() -> Set[str]: """Loads up the list of successfully cached json objects, so we don't need to revalidate them""" - validhashes = set() + validhashes = set() # type: Set if os.path.exists(cached_validation_filepath): with open(cached_validation_filepath, "r") as f: validhashes.update(json.load(f))