mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-10 11:47:38 +02:00
Core: Change all 3.5 type hints to variable annotations
This commit is contained in:
Mike Auty
@@ -43,7 +43,7 @@ class WarningFindSpec(abc.MetaPathFinder):
|
||||
raise Warning(warning)
|
||||
|
||||
|
||||
warning_find_spec = [WarningFindSpec()] # type: List[abc.MetaPathFinder]
|
||||
warning_find_spec: List[abc.MetaPathFinder] = [WarningFindSpec()]
|
||||
sys.meta_path = warning_find_spec + sys.meta_path
|
||||
|
||||
# We point the volatility3.plugins __path__ variable at BOTH
|
||||
|
||||
@@ -583,7 +583,7 @@ class CommandLine:
|
||||
# Construct an argparse group
|
||||
|
||||
for requirement in configurable.get_requirements():
|
||||
additional = {} # type: Dict[str, Any]
|
||||
additional: Dict[str, Any] = {}
|
||||
if not isinstance(requirement, interfaces.configuration.RequirementInterface):
|
||||
raise TypeError("Plugin contains requirements that are not RequirementInterfaces: {}".format(
|
||||
configurable.__name__))
|
||||
|
||||
@@ -278,7 +278,7 @@ class PrettyTextRenderer(CLIRenderer):
|
||||
accumulator.append((node.path_depth, line))
|
||||
return accumulator
|
||||
|
||||
final_output = [] # type: List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]]
|
||||
final_output: List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]] = []
|
||||
if not grid.populated:
|
||||
grid.populate(visitor, final_output)
|
||||
else:
|
||||
@@ -323,15 +323,15 @@ class JsonRenderer(CLIRenderer):
|
||||
outfd = sys.stdout
|
||||
|
||||
outfd.write("\n")
|
||||
final_output = (
|
||||
{}, []) # type: Tuple[Dict[str, List[interfaces.renderers.TreeNode]], List[interfaces.renderers.TreeNode]]
|
||||
final_output: Tuple[Dict[str, List[interfaces.renderers.TreeNode]], List[interfaces.renderers.TreeNode]] = (
|
||||
{}, [])
|
||||
|
||||
def visitor(
|
||||
node: interfaces.renderers.TreeNode, accumulator: Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]]
|
||||
) -> Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]]:
|
||||
# Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case
|
||||
acc_map, final_tree = accumulator
|
||||
node_dict = {'__children': []} # type: Dict[str, Any]
|
||||
node_dict: Dict[str, Any] = {'__children': []}
|
||||
for column_index in range(len(grid.columns)):
|
||||
column = grid.columns[column_index]
|
||||
renderer = self._type_renderers.get(column.type, self._type_renderers['default'])
|
||||
|
||||
@@ -12,7 +12,7 @@ from typing import List, Optional, Sequence, Any, Union
|
||||
# We shouldn't really steal a private member from argparse, but otherwise we're just duplicating code
|
||||
|
||||
# HelpfulSubparserAction gives more information about the possible choices from a subparsed choice
|
||||
# HelpfulArgParser gives the list of choices when no arguments are provided to a choice option whilst still using a METAVAR
|
||||
# HelpfulArgParser gives the list of choices when no arguments are provided to a choice option whilst still using a
|
||||
|
||||
|
||||
class HelpfulSubparserAction(argparse._SubParsersAction):
|
||||
@@ -22,7 +22,7 @@ class HelpfulSubparserAction(argparse._SubParsersAction):
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
# We don't want the action self-check to kick in, so we remove the choices list, the check happens in __call__
|
||||
self.choices = None # type: ignore
|
||||
self.choices = None
|
||||
|
||||
def __call__(self,
|
||||
parser: argparse.ArgumentParser,
|
||||
@@ -31,7 +31,7 @@ class HelpfulSubparserAction(argparse._SubParsersAction):
|
||||
option_string: Optional[str] = None) -> None:
|
||||
|
||||
parser_name = ''
|
||||
arg_strings = [] # type: List[str]
|
||||
arg_strings: List[str] = []
|
||||
if values is not None:
|
||||
for value in values:
|
||||
if not parser_name:
|
||||
|
||||
@@ -30,7 +30,7 @@ class Volshell(interfaces.plugins.PluginInterface):
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.__current_layer = None # type: Optional[str]
|
||||
self.__current_layer: Optional[str] = None
|
||||
self.__console = None
|
||||
|
||||
def random_string(self, length: int = 32) -> str:
|
||||
@@ -38,7 +38,7 @@ class Volshell(interfaces.plugins.PluginInterface):
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
||||
reqs = [] # type: List[interfaces.configuration.RequirementInterface]
|
||||
reqs: List[interfaces.configuration.RequirementInterface] = []
|
||||
if cls == Volshell:
|
||||
reqs = [
|
||||
requirements.URIRequirement(name = 'script',
|
||||
|
||||
@@ -37,7 +37,7 @@ class ConstructionMagic(interfaces.automagic.AutomagicInterface):
|
||||
# Make sure we import the layers, so they can reconstructed
|
||||
framework.import_files(sys.modules['volatility3.framework.layers'])
|
||||
|
||||
result = [] # type: List[str]
|
||||
result: List[str] = []
|
||||
if requirement.unsatisfied(context, config_path):
|
||||
# Having called validate at the top level tells us both that we need to dig deeper
|
||||
# but also ensures that TranslationLayerRequirements have got the correct subrequirements if their class is populated
|
||||
|
||||
@@ -57,7 +57,7 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
|
||||
layer_name,
|
||||
progress_callback = progress_callback)
|
||||
|
||||
layer_class = intel.Intel # type: Type
|
||||
layer_class: Type = intel.Intel
|
||||
if 'init_top_pgt' in table.symbols:
|
||||
layer_class = intel.Intel32e
|
||||
dtb_symbol_name = 'init_top_pgt'
|
||||
|
||||
@@ -62,7 +62,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
|
||||
A list of (layer_name, scan_results)
|
||||
"""
|
||||
sub_config_path = interfaces.configuration.path_join(config_path, requirement.name)
|
||||
results = [] # type: List[str]
|
||||
results: List[str] = []
|
||||
if isinstance(requirement, requirements.TranslationLayerRequirement):
|
||||
# Check for symbols in this layer
|
||||
# FIXME: optionally allow a full (slow) scan
|
||||
@@ -207,13 +207,13 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
|
||||
"""Method for finding a suitable kernel offset based on a module
|
||||
table."""
|
||||
vollog.debug("Kernel base determination - searching layer module list structure")
|
||||
valid_kernel = None # type: Optional[ValidKernelType]
|
||||
valid_kernel: Optional[ValidKernelType] = None
|
||||
# If we're here, chances are high we're in a Win10 x64 image with kernel base randomization
|
||||
physical_layer_name = self.get_physical_layer_name(context, vlayer)
|
||||
physical_layer = context.layers[physical_layer_name]
|
||||
# TODO: On older windows, this might be \WINDOWS\system32\nt rather than \SystemRoot\system32\nt
|
||||
results = physical_layer.scan(context, scanners.BytesScanner(pattern), progress_callback = progress_callback)
|
||||
seen = set() # type: Set[int]
|
||||
seen: 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
|
||||
@@ -252,7 +252,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
|
||||
"""Scans a virtual address."""
|
||||
# Scan a few megs of the virtual space at the location to see if they're potential kernels
|
||||
|
||||
valid_kernel = None # type: Optional[ValidKernelType]
|
||||
valid_kernel: Optional[ValidKernelType] = None
|
||||
kernel_pdb_names = [bytes(name + ".pdb", "utf-8") for name in constants.windows.KERNEL_MODULE_NAMES]
|
||||
|
||||
virtual_layer_name = vlayer.name
|
||||
@@ -295,7 +295,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
|
||||
Returns:
|
||||
A dictionary of valid kernels
|
||||
"""
|
||||
valid_kernel = None # type: Optional[ValidKernelType]
|
||||
valid_kernel: Optional[ValidKernelType] = None
|
||||
for virtual_layer_name in potential_layers:
|
||||
vlayer = context.layers.get(virtual_layer_name, None)
|
||||
if isinstance(vlayer, layers.intel.Intel):
|
||||
|
||||
@@ -26,15 +26,15 @@ 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 # type: Optional[str]
|
||||
symbol_name = "banner_name" # type: str
|
||||
banner_path = None # type: Optional[str]
|
||||
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
|
||||
banners: BannersType = {}
|
||||
if os.path.exists(cls.banner_path):
|
||||
with open(cls.banner_path, "rb") as f:
|
||||
# We use pickle over JSON because we're dealing with bytes objects
|
||||
|
||||
@@ -17,15 +17,15 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
|
||||
"""Symbol loader based on signature strings."""
|
||||
priority = 40
|
||||
|
||||
banner_config_key = "banner" # type: str
|
||||
banner_cache = None # type: Optional[Type[symbol_cache.SymbolBannerCache]]
|
||||
symbol_class = None # type: Optional[str]
|
||||
find_aslr = None # type: Optional[Callable]
|
||||
banner_config_key: str = "banner"
|
||||
banner_cache: Optional[Type[symbol_cache.SymbolBannerCache]] = None
|
||||
symbol_class: Optional[str] = None
|
||||
find_aslr: Optional[Callable] = 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.RequirementInterface]]
|
||||
self._banners = {} # type: symbol_cache.BannersType
|
||||
self._requirements: List[Tuple[str, interfaces.configuration.RequirementInterface]] = []
|
||||
self._banners: symbol_cache.BannersType = {}
|
||||
|
||||
@property
|
||||
def banners(self) -> symbol_cache.BannersType:
|
||||
|
||||
@@ -325,7 +325,7 @@ class WindowsIntelStacker(interfaces.automagic.StackerLayerInterface):
|
||||
if arch not in ['Intel32', 'Intel64']:
|
||||
return None
|
||||
# Set the layer type
|
||||
layer_type = intel.WindowsIntel # type: Type
|
||||
layer_type: Type = intel.WindowsIntel
|
||||
if arch == 'Intel64':
|
||||
layer_type = intel.WindowsIntel32e
|
||||
elif base_layer.metadata.get('pae', False):
|
||||
|
||||
@@ -36,13 +36,13 @@ class BooleanRequirement(interfaces.configuration.SimpleTypeRequirement):
|
||||
|
||||
class IntRequirement(interfaces.configuration.SimpleTypeRequirement):
|
||||
"""A requirement type that contains a single integer."""
|
||||
instance_type = int # type: ClassVar[Type]
|
||||
instance_type: ClassVar[Type] = int
|
||||
|
||||
|
||||
class StringRequirement(interfaces.configuration.SimpleTypeRequirement):
|
||||
"""A requirement type that contains a single unicode string."""
|
||||
# TODO: Maybe add string length limits?
|
||||
instance_type = str # type: ClassVar[Type]
|
||||
instance_type: ClassVar[Type] = str
|
||||
|
||||
|
||||
class URIRequirement(StringRequirement):
|
||||
@@ -53,7 +53,7 @@ class URIRequirement(StringRequirement):
|
||||
|
||||
class BytesRequirement(interfaces.configuration.SimpleTypeRequirement):
|
||||
"""A requirement type that contains a byte string."""
|
||||
instance_type = bytes # type: ClassVar[Type]
|
||||
instance_type: ClassVar[Type] = bytes
|
||||
|
||||
|
||||
class ListRequirement(interfaces.configuration.RequirementInterface):
|
||||
@@ -83,9 +83,9 @@ class ListRequirement(interfaces.configuration.RequirementInterface):
|
||||
super().__init__(*args, **kwargs)
|
||||
if not issubclass(element_type, interfaces.configuration.BasicTypes):
|
||||
raise TypeError("ListRequirements can only be populated with simple InstanceRequirements")
|
||||
self.element_type = element_type # type: Type
|
||||
self.min_elements = min_elements or 0 # type: int
|
||||
self.max_elements = max_elements # type: Optional[int]
|
||||
self.element_type: Type = element_type
|
||||
self.min_elements: int = min_elements or 0
|
||||
self.max_elements: Optional[int] = max_elements
|
||||
|
||||
def unsatisfied(self, context: interfaces.context.ContextInterface,
|
||||
config_path: str) -> Dict[str, interfaces.configuration.RequirementInterface]:
|
||||
@@ -397,7 +397,7 @@ class VersionRequirement(interfaces.configuration.RequirementInterface):
|
||||
super().__init__(name = name, description = description, default = default, optional = optional)
|
||||
if component is None:
|
||||
raise TypeError("Component cannot be None")
|
||||
self._component = component # type: Type[interfaces.configuration.VersionableInterface]
|
||||
self._component: Type[interfaces.configuration.VersionableInterface] = component
|
||||
if version is None:
|
||||
raise TypeError("Version cannot be None")
|
||||
self._version = version
|
||||
|
||||
@@ -319,7 +319,7 @@ class ModuleCollection:
|
||||
included in the deduplicated version
|
||||
"""
|
||||
new_modules = []
|
||||
seen = set() # type: Set[str]
|
||||
seen: Set[str] = set()
|
||||
for mod in self._modules:
|
||||
if mod.hash not in seen or mod.size == 0:
|
||||
new_modules.append(mod)
|
||||
@@ -334,7 +334,7 @@ class ModuleCollection:
|
||||
|
||||
@classmethod
|
||||
def _generate_module_dict(cls, modules: List[SizedModule]) -> Dict[str, List[SizedModule]]:
|
||||
result = {} # type: Dict[str, List[SizedModule]]
|
||||
result: Dict[str, List[SizedModule]] = {}
|
||||
for module in modules:
|
||||
modlist = result.get(module.name, [])
|
||||
modlist.append(module)
|
||||
|
||||
@@ -82,7 +82,7 @@ class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metacla
|
||||
A list of tuples containing the config_path, sub_config_path and requirement identifying the unsatisfied `Requirements`
|
||||
"""
|
||||
sub_config_path = interfaces.configuration.path_join(config_path, requirement_root.name)
|
||||
results = [] # type: List[Tuple[str, interfaces.configuration.RequirementInterface]]
|
||||
results: List[Tuple[str, interfaces.configuration.RequirementInterface]] = []
|
||||
recurse = not shortcut
|
||||
if isinstance(requirement_root, requirement_type):
|
||||
if recurse or requirement_root.unsatisfied(context, config_path):
|
||||
@@ -105,7 +105,7 @@ class StackerLayerInterface(metaclass = ABCMeta):
|
||||
|
||||
stack_order = 0
|
||||
"""The order in which to attempt stacking, the lower the earlier"""
|
||||
exclusion_list = [] # type: List[str]
|
||||
exclusion_list: List[str] = []
|
||||
"""The list operating systems/first-level plugin hierarchy that should exclude this stacker"""
|
||||
|
||||
@classmethod
|
||||
|
||||
@@ -79,8 +79,8 @@ class HierarchicalDict(collections.abc.Mapping):
|
||||
if not (isinstance(separator, str) and len(separator) == 1):
|
||||
raise TypeError(f"Separator must be a one character string: {separator}")
|
||||
self._separator = separator
|
||||
self._data = {} # type: Dict[str, ConfigSimpleType]
|
||||
self._subdict = {} # type: Dict[str, 'HierarchicalDict']
|
||||
self._data: Dict[str, ConfigSimpleType] = {}
|
||||
self._subdict: Dict[str, 'HierarchicalDict'] = {}
|
||||
if isinstance(initial_dict, str):
|
||||
initial_dict = json.loads(initial_dict)
|
||||
if isinstance(initial_dict, dict):
|
||||
@@ -320,7 +320,7 @@ class RequirementInterface(metaclass = ABCMeta):
|
||||
self._description = description or ""
|
||||
self._default = default
|
||||
self._optional = optional
|
||||
self._requirements = {} # type: Dict[str, RequirementInterface]
|
||||
self._requirements: Dict[str, RequirementInterface] = {}
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return "<" + self.__class__.__name__ + ": " + self.name + ">"
|
||||
@@ -438,7 +438,7 @@ class RequirementInterface(metaclass = ABCMeta):
|
||||
class SimpleTypeRequirement(RequirementInterface):
|
||||
"""Class to represent a single simple type (such as a boolean, a string, an
|
||||
integer or a series of bytes)"""
|
||||
instance_type = bool # type: ClassVar[Type]
|
||||
instance_type: ClassVar[Type] = bool
|
||||
|
||||
def add_requirement(self, requirement: RequirementInterface):
|
||||
"""Always raises a TypeError as instance requirements cannot have
|
||||
@@ -529,7 +529,7 @@ class ConstructableRequirementInterface(RequirementInterface):
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
self.add_requirement(ClassRequirement("class", "Class of the constructable requirement"))
|
||||
self._current_class_requirements = set() # type: Set[Any]
|
||||
self._current_class_requirements: Set[Any] = set()
|
||||
|
||||
def __eq__(self, other):
|
||||
# We can just use super because it checks all member of `__dict__`
|
||||
@@ -620,7 +620,7 @@ class ConfigurableInterface(metaclass = ABCMeta):
|
||||
super().__init__()
|
||||
self._context = context
|
||||
self._config_path = config_path
|
||||
self._config_cache = None # type: Optional[HierarchicalDict]
|
||||
self._config_cache: Optional[HierarchicalDict] = None
|
||||
|
||||
@property
|
||||
def context(self) -> 'interfaces.context.ContextInterface':
|
||||
@@ -729,7 +729,7 @@ class VersionableInterface:
|
||||
|
||||
All version number should use semantic versioning
|
||||
"""
|
||||
_version = (0, 0, 0) # type: Tuple[int, int, int]
|
||||
_version: Tuple[int, int, int] = (0, 0, 0)
|
||||
|
||||
@classproperty
|
||||
def version(cls) -> Tuple[int, int, int]:
|
||||
|
||||
@@ -58,8 +58,8 @@ class ScannerInterface(interfaces.configuration.VersionableInterface, metaclass
|
||||
super().__init__()
|
||||
self.chunk_size = 0x1000000 # Default to 16Mb chunks
|
||||
self.overlap = 0x1000 # A page of overlap by default
|
||||
self._context = None # type: Optional[interfaces.context.ContextInterface]
|
||||
self._layer_name = None # type: Optional[str]
|
||||
self._context: Optional[interfaces.context.ContextInterface] = None
|
||||
self._layer_name: Optional[str] = None
|
||||
|
||||
@property
|
||||
def context(self) -> Optional['interfaces.context.ContextInterface']:
|
||||
@@ -99,7 +99,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla
|
||||
accesses a data source and exposes it within volatility.
|
||||
"""
|
||||
|
||||
_direct_metadata = {'architecture': 'Unknown', 'os': 'Unknown'} # type: Mapping
|
||||
_direct_metadata: Mapping = {'architecture': 'Unknown', 'os': 'Unknown'}
|
||||
|
||||
def __init__(self,
|
||||
context: 'interfaces.context.ContextInterface',
|
||||
@@ -227,7 +227,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla
|
||||
sections = list(self._coalesce_sections(sections))
|
||||
|
||||
try:
|
||||
progress = DummyProgress() # type: ProgressValue
|
||||
progress: ProgressValue = DummyProgress()
|
||||
scan_iterator = functools.partial(self._scan_iterator, scanner, sections)
|
||||
scan_metric = self._scan_metric(scanner, sections)
|
||||
if not scanner.thread_safe or constants.PARALLELISM == constants.Parallelism.Off:
|
||||
@@ -240,7 +240,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla
|
||||
yield from scan_chunk(value)
|
||||
else:
|
||||
progress = multiprocessing.Manager().Value("Q", 0)
|
||||
parallel_module = multiprocessing # type: types.ModuleType
|
||||
parallel_module: types.ModuleType = multiprocessing
|
||||
if constants.PARALLELISM == constants.Parallelism.Threading:
|
||||
progress = DummyProgress()
|
||||
parallel_module = threading
|
||||
@@ -266,7 +266,7 @@ class DataLayerInterface(interfaces.configuration.ConfigurableInterface, metacla
|
||||
def _coalesce_sections(self, sections: Iterable[Tuple[int, int]]) -> Iterable[Tuple[int, int]]:
|
||||
"""Take a list of (start, length) sections and coalesce any adjacent
|
||||
sections."""
|
||||
result = [] # type: List[Tuple[int, int]]
|
||||
result: List[Tuple[int, int]] = []
|
||||
position = 0
|
||||
for (start, length) in sorted(sections):
|
||||
if result and start <= position:
|
||||
@@ -423,7 +423,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
|
||||
"""Reads an offset for length bytes and returns 'bytes' (not 'str') of
|
||||
length size."""
|
||||
current_offset = offset
|
||||
output = b'' # type: bytes
|
||||
output: bytes = b''
|
||||
for (layer_offset, sublength, mapped_offset, mapped_length, layer) in self.mapping(offset,
|
||||
length,
|
||||
ignore_errors = pad):
|
||||
@@ -473,7 +473,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta):
|
||||
assumed to have no holes
|
||||
"""
|
||||
for (section_start, section_length) in sections:
|
||||
output = [] # type: List[Tuple[str, int, int]]
|
||||
output: List[Tuple[str, int, int]] = []
|
||||
|
||||
# Hold the offsets of each chunk (including how much has been filled)
|
||||
chunk_start = chunk_position = 0
|
||||
@@ -532,7 +532,7 @@ class LayerContainer(collections.abc.Mapping):
|
||||
"""Container for multiple layers of data."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._layers = {} # type: Dict[str, DataLayerInterface]
|
||||
self._layers: Dict[str, DataLayerInterface] = {}
|
||||
|
||||
def read(self, layer: str, offset: int, length: int, pad: bool = False) -> bytes:
|
||||
"""Reads from a particular layer at offset for length bytes.
|
||||
|
||||
@@ -214,7 +214,7 @@ class ObjectInterface(metaclass = abc.ABCMeta):
|
||||
to control how their templates respond without needing to write
|
||||
new templates for each and every potental object type.
|
||||
"""
|
||||
_methods = [] # type: List[str]
|
||||
_methods: List[str] = []
|
||||
|
||||
@classmethod
|
||||
@abc.abstractmethod
|
||||
@@ -275,7 +275,7 @@ class Template:
|
||||
"""Stores the keyword arguments for later object creation."""
|
||||
# Allow the updating of template arguments whilst still in template form
|
||||
super().__init__()
|
||||
empty_dict = {} # type: Dict[str, Any]
|
||||
empty_dict: Dict[str, Any] = {}
|
||||
self._vol = collections.ChainMap(empty_dict, arguments, {'type_name': type_name})
|
||||
|
||||
@property
|
||||
|
||||
@@ -95,7 +95,7 @@ class PluginInterface(interfaces.configuration.ConfigurableInterface,
|
||||
"""
|
||||
|
||||
# Be careful with inheritance around this (We default to requiring a version which doesn't exist, so it must be set)
|
||||
_required_framework_version = (0, 0, 0) # type: Tuple[int, int, int]
|
||||
_required_framework_version: Tuple[int, int, int] = (0, 0, 0)
|
||||
"""The _version variable is a quick way for plugins to define their current interface, it should follow SemVer rules"""
|
||||
|
||||
def __init__(self,
|
||||
@@ -121,7 +121,7 @@ class PluginInterface(interfaces.configuration.ConfigurableInterface,
|
||||
if requirement.name not in self.config:
|
||||
self.config[requirement.name] = requirement.default
|
||||
|
||||
self._file_handler = FileHandlerInterface # type: Type[FileHandlerInterface]
|
||||
self._file_handler: Type[FileHandlerInterface] = FileHandlerInterface
|
||||
|
||||
framework.require_interface_version(*self._required_framework_version)
|
||||
|
||||
|
||||
@@ -38,7 +38,7 @@ class Renderer(metaclass = ABCMeta):
|
||||
|
||||
|
||||
class ColumnSortKey(metaclass = ABCMeta):
|
||||
ascending = True # type: bool
|
||||
ascending: bool = True
|
||||
|
||||
@abstractmethod
|
||||
def __call__(self, values: List[Any]) -> Any:
|
||||
@@ -129,7 +129,7 @@ class TreeGrid(object, metaclass = ABCMeta):
|
||||
and to create cycles.
|
||||
"""
|
||||
|
||||
base_types = (int, str, float, bytes, datetime.datetime, Disassembly) # type: ClassVar[Tuple]
|
||||
base_types: ClassVar[Tuple] = (int, str, float, bytes, datetime.datetime, Disassembly)
|
||||
|
||||
def __init__(self, columns: ColumnsType, generator: Generator) -> None:
|
||||
"""Constructs a TreeGrid object using a specific set of columns.
|
||||
|
||||
@@ -96,7 +96,7 @@ class BaseSymbolTableInterface:
|
||||
table_mapping = {}
|
||||
self.table_mapping = table_mapping
|
||||
self._native_types = native_types
|
||||
self._sort_symbols = [] # type: List[Tuple[int, str]]
|
||||
self._sort_symbols: List[Tuple[int, str]] = []
|
||||
|
||||
# Set any provisioned class_types
|
||||
if class_types:
|
||||
|
||||
@@ -39,7 +39,7 @@ class Intel(linear.LinearlyMappedLayer):
|
||||
metadata: Optional[Dict[str, Any]] = None) -> None:
|
||||
super().__init__(context = context, config_path = config_path, name = name, metadata = metadata)
|
||||
self._base_layer = self.config["memory_layer"]
|
||||
self._swap_layers = [] # type: List[str]
|
||||
self._swap_layers: List[str] = []
|
||||
self._page_map_offset = self.config["page_map_offset"]
|
||||
|
||||
# Assign constants
|
||||
|
||||
@@ -33,7 +33,8 @@ class LinearlyMappedLayer(interfaces.layers.TranslationLayerInterface):
|
||||
"""Reads an offset for length bytes and returns 'bytes' (not 'str') of
|
||||
length size."""
|
||||
current_offset = offset
|
||||
output = [] # type: List[bytes]
|
||||
output: List[bytes] = []
|
||||
output: List[bytes] = []
|
||||
for (offset, _, mapped_offset, mapped_length, layer) in self.mapping(offset, length, ignore_errors = pad):
|
||||
if not pad and offset > current_offset:
|
||||
raise exceptions.InvalidAddressException(
|
||||
|
||||
@@ -34,7 +34,7 @@ class PdbMultiStreamFormat(linear.LinearlyMappedLayer):
|
||||
if response is None:
|
||||
raise PDBFormatException(name, "Could not find a suitable header")
|
||||
self._version, self._header = response
|
||||
self._streams = {} # type: Dict[int, str]
|
||||
self._streams: Dict[int, str] = {}
|
||||
|
||||
@property
|
||||
def pdb_symbol_table(self) -> str:
|
||||
|
||||
@@ -86,10 +86,10 @@ class FileLayer(interfaces.layers.DataLayerInterface):
|
||||
self._write_warning = False
|
||||
self._location = self.config["location"]
|
||||
self._accessor = resources.ResourceAccessor()
|
||||
self._file_ = None # type: Optional[IO[Any]]
|
||||
self._size = None # type: Optional[int]
|
||||
self._file_: Optional[IO[Any]] = None
|
||||
self._size: Optional[int] = None
|
||||
# Construct the lock now (shared if made before threading) in case we ever need it
|
||||
self._lock = DummyLock() # type: Union[DummyLock, threading.Lock]
|
||||
self._lock: Union[DummyLock, threading.Lock] = DummyLock()
|
||||
if constants.PARALLELISM == constants.Parallelism.Threading:
|
||||
self._lock = threading.Lock()
|
||||
# Instantiate the file to throw exceptions if the file doesn't open
|
||||
|
||||
@@ -40,7 +40,7 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
|
||||
metadata: Optional[Dict[str, Any]] = None) -> None:
|
||||
self._qemu_table_name = intermed.IntermediateSymbolTable.create(context, config_path, 'generic', 'qemu')
|
||||
self._configuration = None
|
||||
self._compressed = set() # type: Set[int]
|
||||
self._compressed: Set[int] = set()
|
||||
self._current_segment_name = b''
|
||||
super().__init__(context = context, config_path = config_path, name = name, metadata = metadata)
|
||||
|
||||
|
||||
@@ -141,7 +141,7 @@ class RegistryHive(linear.LinearlyMappedLayer):
|
||||
if key.endswith("\\"):
|
||||
key = key[:-1]
|
||||
key_array = key.split('\\')
|
||||
found_key = [] # type: List[str]
|
||||
found_key: List[str] = []
|
||||
while key_array and node_key:
|
||||
subkeys = node_key[-1].get_subkeys()
|
||||
for subkey in subkeys:
|
||||
|
||||
@@ -48,7 +48,7 @@ class MultiStringScanner(layers.ScannerInterface):
|
||||
|
||||
def __init__(self, patterns: List[bytes]) -> None:
|
||||
super().__init__()
|
||||
self._pattern_trie = {} # type: Optional[Dict[int, Optional[Dict]]]
|
||||
self._pattern_trie: Optional[Dict[int, Optional[Dict]]] = {}
|
||||
for pattern in patterns:
|
||||
self._process_pattern(pattern)
|
||||
self._regex = self._process_trie(self._pattern_trie)
|
||||
|
||||
@@ -10,7 +10,7 @@ class MultiRegexp(object):
|
||||
"""Algorithm for multi-string matching."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._pattern_strings = [] # type: List[bytes]
|
||||
self._pattern_strings: List[bytes] = []
|
||||
self._regex = re.compile(b'')
|
||||
|
||||
def add_pattern(self, pattern: bytes) -> None:
|
||||
|
||||
@@ -25,9 +25,9 @@ class NonLinearlySegmentedLayer(interfaces.layers.TranslationLayerInterface, met
|
||||
super().__init__(context = context, config_path = config_path, name = name, metadata = metadata)
|
||||
|
||||
self._base_layer = self.config["base_layer"]
|
||||
self._segments = [] # type: List[Tuple[int, int, int, int]]
|
||||
self._minaddr = None # type: Optional[int]
|
||||
self._maxaddr = None # type: Optional[int]
|
||||
self._segments: List[Tuple[int, int, int, int]] = []
|
||||
self._minaddr: Optional[int] = None
|
||||
self._maxaddr: Optional[int] = None
|
||||
|
||||
self._load_segments()
|
||||
|
||||
|
||||
@@ -95,7 +95,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 # type: ClassVar[Type]
|
||||
_struct_type: ClassVar[Type] = int
|
||||
|
||||
def __init__(self, context: interfaces.context.ContextInterface, type_name: str,
|
||||
object_info: interfaces.objects.ObjectInformation, data_format: DataFormatInfo) -> None:
|
||||
@@ -164,7 +164,7 @@ class PrimitiveObject(interfaces.objects.ObjectInterface):
|
||||
# https://mail.python.org/pipermail/python-dev/2004-February/042537.html
|
||||
class Boolean(PrimitiveObject, int):
|
||||
"""Primitive Object that handles boolean types."""
|
||||
_struct_type = int # type: ClassVar[Type]
|
||||
_struct_type: ClassVar[Type] = int
|
||||
|
||||
|
||||
class Integer(PrimitiveObject, int):
|
||||
@@ -173,17 +173,17 @@ class Integer(PrimitiveObject, int):
|
||||
|
||||
class Float(PrimitiveObject, float):
|
||||
"""Primitive Object that handles double or floating point numbers."""
|
||||
_struct_type = float # type: ClassVar[Type]
|
||||
_struct_type: ClassVar[Type] = float
|
||||
|
||||
|
||||
class Char(PrimitiveObject, int):
|
||||
"""Primitive Object that handles characters."""
|
||||
_struct_type = int # type: ClassVar[Type]
|
||||
_struct_type: ClassVar[Type] = int
|
||||
|
||||
|
||||
class Bytes(PrimitiveObject, bytes):
|
||||
"""Primitive Object that handles specific series of bytes."""
|
||||
_struct_type = bytes # type: ClassVar[Type]
|
||||
_struct_type: ClassVar[Type] = bytes
|
||||
|
||||
def __init__(self,
|
||||
context: interfaces.context.ContextInterface,
|
||||
@@ -227,7 +227,7 @@ class String(PrimitiveObject, str):
|
||||
max_length: specifies the maximum possible length that the string could hold within memory
|
||||
(for multibyte characters, this will not be the maximum length of the string)
|
||||
"""
|
||||
_struct_type = str # type: ClassVar[Type]
|
||||
_struct_type: ClassVar[Type] = str
|
||||
|
||||
def __init__(self,
|
||||
context: interfaces.context.ContextInterface,
|
||||
@@ -453,7 +453,7 @@ class Enumeration(interfaces.objects.ObjectInterface, int):
|
||||
@classmethod
|
||||
def _generate_inverse_choices(cls, choices: Dict[str, int]) -> Dict[int, str]:
|
||||
"""Generates the inverse choices for the object."""
|
||||
inverse_choices = {} # type: Dict[int, str]
|
||||
inverse_choices: Dict[int, str] = {}
|
||||
for k, v in choices.items():
|
||||
if v in inverse_choices:
|
||||
# Technically this shouldn't be a problem, but since we inverse cache
|
||||
@@ -601,7 +601,7 @@ class Array(interfaces.objects.ObjectInterface, collections.abc.Sequence):
|
||||
|
||||
def __getitem__(self, i):
|
||||
"""Returns the i-th item from the array."""
|
||||
result = [] # type: List[interfaces.objects.Template]
|
||||
result: List[interfaces.objects.Template] = []
|
||||
mask = self._context.layers[self.vol.layer_name].address_mask
|
||||
# We use the range function to deal with slices for us
|
||||
series = range(self.vol.count)[i]
|
||||
@@ -649,7 +649,7 @@ class AggregateType(interfaces.objects.ObjectInterface):
|
||||
size = size,
|
||||
members = members)
|
||||
# self._check_members(members)
|
||||
self._concrete_members = {} # type: Dict[str, Dict]
|
||||
self._concrete_members: Dict[str, Dict] = {}
|
||||
|
||||
def has_member(self, member_name: str) -> bool:
|
||||
"""Returns whether the object would contain a member called
|
||||
|
||||
@@ -65,7 +65,7 @@ class ObjectTemplate(interfaces.objects.Template):
|
||||
|
||||
Returns: an object adhereing to the :class:`~volatility3.framework.interfaces.objects.ObjectInterface`
|
||||
"""
|
||||
arguments = {} # type: Dict[str, Any]
|
||||
arguments: Dict[str, Any] = {}
|
||||
for arg in self.vol:
|
||||
if arg != 'object_class':
|
||||
arguments[arg] = self.vol[arg]
|
||||
@@ -96,10 +96,10 @@ class ReferenceTemplate(interfaces.objects.Template):
|
||||
symbol_name, table_name,
|
||||
f"Template contains no information about its structure: {self.vol.type_name}")
|
||||
|
||||
size = property(_unresolved) # type: ClassVar[Any]
|
||||
replace_child = _unresolved # type: ClassVar[Any]
|
||||
relative_child_offset = _unresolved # type: ClassVar[Any]
|
||||
has_member = _unresolved # type: ClassVar[Any]
|
||||
size: ClassVar[Any] = property(_unresolved)
|
||||
replace_child: ClassVar[Any] = _unresolved
|
||||
relative_child_offset: ClassVar[Any] = _unresolved
|
||||
has_member: ClassVar[Any] = _unresolved
|
||||
|
||||
def __call__(self, context: interfaces.context.ContextInterface, object_info: interfaces.objects.ObjectInformation):
|
||||
template = context.symbol_space.get_type(self.vol.type_name)
|
||||
|
||||
@@ -57,7 +57,7 @@ class Lsmod(plugins.PluginInterface):
|
||||
except exceptions.InvalidAddressException:
|
||||
return []
|
||||
|
||||
seen = set() # type: Set
|
||||
seen: Set = set()
|
||||
|
||||
while kmod != 0 and \
|
||||
kmod not in seen and \
|
||||
|
||||
@@ -52,7 +52,7 @@ class Psaux(plugins.PluginInterface):
|
||||
|
||||
task_name = utility.array_to_string(task.p_comm)
|
||||
|
||||
args = [] # type: List[bytes]
|
||||
args: List[bytes] = []
|
||||
|
||||
while argc > 0:
|
||||
try:
|
||||
|
||||
@@ -124,7 +124,7 @@ class PsList(interfaces.plugins.PluginInterface):
|
||||
|
||||
proc = kernel.object_from_symbol(symbol_name = "allproc").lh_first
|
||||
|
||||
seen = {} # type: Dict[int, int]
|
||||
seen: Dict[int, int] = {}
|
||||
while proc is not None and proc.vol.offset != 0:
|
||||
if proc.vol.offset in seen:
|
||||
vollog.log(logging.INFO, "Recursive process list detected (a result of non-atomic acquisition).")
|
||||
@@ -165,7 +165,7 @@ class PsList(interfaces.plugins.PluginInterface):
|
||||
|
||||
queue_entry = kernel.object_from_symbol(symbol_name = "tasks")
|
||||
|
||||
seen = {} # type: Dict[int, int]
|
||||
seen: Dict[int, int] = {}
|
||||
for task in queue_entry.walk_list(queue_entry, "tasks", "task"):
|
||||
if task.vol.offset in seen:
|
||||
vollog.log(logging.INFO, "Recursive process list detected (a result of non-atomic acquisition).")
|
||||
|
||||
@@ -48,7 +48,7 @@ class Timeliner(interfaces.plugins.PluginInterface):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.timeline = {}
|
||||
self.usable_plugins = None
|
||||
self.automagics = None # type: Optional[List[interfaces.automagic.AutomagicInterface]]
|
||||
self.automagics: Optional[List[interfaces.automagic.AutomagicInterface]] = None
|
||||
|
||||
@classmethod
|
||||
def get_usable_plugins(cls, selected_list: List[str] = None) -> List[Type]:
|
||||
|
||||
@@ -191,9 +191,9 @@ class Callbacks(interfaces.plugins.PluginInterface):
|
||||
continue
|
||||
|
||||
try:
|
||||
component = ntkrnlmp.object(
|
||||
component: Union[interfaces.renderers.BaseAbsentValue, interfaces.objects.ObjectInterface] = ntkrnlmp.object(
|
||||
"string", absolute = True, offset = callback.Component, max_length = 64, errors = "replace"
|
||||
) # type: Union[interfaces.renderers.BaseAbsentValue, interfaces.objects.ObjectInterface]
|
||||
)
|
||||
except exceptions.InvalidAddressException:
|
||||
component = renderers.UnreadableValue()
|
||||
|
||||
|
||||
@@ -172,7 +172,7 @@ class Handles(interfaces.plugins.PluginInterface):
|
||||
A mapping of type indicies to type names
|
||||
"""
|
||||
|
||||
type_map = {} # type: Dict[int, str]
|
||||
type_map: Dict[int, str] = {}
|
||||
|
||||
kvo = context.layers[layer_name].config['kernel_virtual_offset']
|
||||
ntkrnlmp = context.module(symbol_table, layer_name = layer_name, offset = kvo)
|
||||
|
||||
@@ -81,7 +81,7 @@ class ModScan(interfaces.plugins.PluginInterface):
|
||||
Returns:
|
||||
A list of session layer names
|
||||
"""
|
||||
seen_ids = [] # type: List[interfaces.objects.ObjectInterface]
|
||||
seen_ids: List[interfaces.objects.ObjectInterface] = []
|
||||
filter_func = pslist.PsList.create_pid_filter(pids or [])
|
||||
|
||||
for proc in pslist.PsList.list_processes(context = context,
|
||||
|
||||
@@ -86,7 +86,7 @@ class Modules(interfaces.plugins.PluginInterface):
|
||||
Returns:
|
||||
A list of session layer names
|
||||
"""
|
||||
seen_ids = [] # type: List[interfaces.objects.ObjectInterface]
|
||||
seen_ids: List[interfaces.objects.ObjectInterface] = []
|
||||
filter_func = pslist.PsList.create_pid_filter(pids or [])
|
||||
|
||||
for proc in pslist.PsList.list_processes(context = context,
|
||||
|
||||
@@ -338,7 +338,7 @@ class PoolScanner(plugins.PluginInterface):
|
||||
An Iterable of pool constraints and the pool headers associated with them
|
||||
"""
|
||||
# Setup the pattern
|
||||
constraint_lookup = {} # type: Dict[bytes, PoolConstraint]
|
||||
constraint_lookup: Dict[bytes, PoolConstraint] = {}
|
||||
for constraint in pool_constraints:
|
||||
if constraint.tag in constraint_lookup:
|
||||
raise ValueError(f"Constraint tag is used for more than one constraint: {repr(constraint.tag)}")
|
||||
|
||||
@@ -18,9 +18,9 @@ class PsTree(interfaces.plugins.PluginInterface):
|
||||
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
self._processes = {} # type: Dict[int, interfaces.objects.ObjectInterface]
|
||||
self._levels = {} # type: Dict[int, int]
|
||||
self._children = {} # type: Dict[int, Set[int]]
|
||||
self._processes: Dict[int, interfaces.objects.ObjectInterface] = {}
|
||||
self._levels: Dict[int, int] = {}
|
||||
self._children: Dict[int, Set[int]] = {}
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls):
|
||||
|
||||
@@ -130,7 +130,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
|
||||
|
||||
if isinstance(value_type, renderers.UnreadableValue):
|
||||
vollog.debug("Couldn't read registry value type, so data is unreadable")
|
||||
value_data = renderers.UnreadableValue() # type: Union[interfaces.renderers.BaseAbsentValue, bytes]
|
||||
value_data: Union[interfaces.renderers.BaseAbsentValue, bytes] = renderers.UnreadableValue()
|
||||
else:
|
||||
try:
|
||||
value_data = node.decode_data()
|
||||
|
||||
@@ -151,12 +151,12 @@ class UserAssist(interfaces.plugins.PluginInterface):
|
||||
countkey_last_write_time = conversion.wintime_to_datetime(countkey.LastWriteTime.QuadPart)
|
||||
|
||||
# output the parent Count key
|
||||
result = (
|
||||
result: Tuple[int, Tuple[format_hints.Hex, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]] = (
|
||||
0, (renderers.format_hints.Hex(hive.hive_offset), hive_name, countkey_path,
|
||||
countkey_last_write_time, "Key", renderers.NotApplicableValue(), renderers.NotApplicableValue(),
|
||||
renderers.NotApplicableValue(), renderers.NotApplicableValue(), renderers.NotApplicableValue(),
|
||||
renderers.NotApplicableValue(), renderers.NotApplicableValue())
|
||||
) # type: Tuple[int, Tuple[format_hints.Hex, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]]
|
||||
)
|
||||
yield result
|
||||
|
||||
# output any subkeys under Count
|
||||
|
||||
@@ -40,17 +40,18 @@ class Strings(interfaces.plugins.PluginInterface):
|
||||
|
||||
def run(self):
|
||||
return renderers.TreeGrid([("String", str), ("Physical Address", format_hints.Hex), ("Result", str)],
|
||||
self._generator())
|
||||
self._generator)
|
||||
|
||||
@property
|
||||
def _generator(self) -> Generator[Tuple, None, None]:
|
||||
"""Generates results from a strings file."""
|
||||
string_list = [] # type: List[Tuple[int,bytes]]
|
||||
string_list: List[Tuple[int,bytes]] = []
|
||||
|
||||
# Test strings file format is accurate
|
||||
accessor = resources.ResourceAccessor()
|
||||
strings_fp = accessor.open(self.config['strings_file'], "rb")
|
||||
line = strings_fp.readline()
|
||||
count = 0 # type: float
|
||||
count: float = 0
|
||||
while line:
|
||||
count += 1
|
||||
try:
|
||||
@@ -66,7 +67,8 @@ class Strings(interfaces.plugins.PluginInterface):
|
||||
progress_callback = self._progress_callback,
|
||||
pid_list = self.config['pid'])
|
||||
|
||||
last_prog = line_count = 0 # type: float
|
||||
last_prog: float = 0
|
||||
line_count: float = 0
|
||||
num_strings = len(string_list)
|
||||
for offset, string in string_list:
|
||||
line_count += 1
|
||||
@@ -119,7 +121,7 @@ class Strings(interfaces.plugins.PluginInterface):
|
||||
filter = pslist.PsList.create_pid_filter(pid_list)
|
||||
|
||||
layer = context.layers[layer_name]
|
||||
reverse_map = dict() # type: Dict[int, Set[Tuple[str, int]]]
|
||||
reverse_map: Dict[int, Set[Tuple[str, int]]] = dict()
|
||||
if isinstance(layer, intel.Intel):
|
||||
# We don't care about errors, we just wanted chunks that map correctly
|
||||
for mapval in layer.mapping(0x0, layer.maximum_address, ignore_errors = True):
|
||||
|
||||
@@ -41,7 +41,7 @@ class VirtMap(interfaces.plugins.PluginInterface):
|
||||
if not isinstance(layer, intel.Intel):
|
||||
raise
|
||||
|
||||
result = {} # type: Dict[str, List[Tuple[int, int]]]
|
||||
result: Dict[str, List[Tuple[int, int]]] = {}
|
||||
system_va_type = module.get_enumeration('_MI_SYSTEM_VA_TYPE')
|
||||
large_page_size = (layer.page_size ** 2) // module.get_type("_MMPTE").size
|
||||
|
||||
@@ -84,7 +84,7 @@ class VirtMap(interfaces.plugins.PluginInterface):
|
||||
def _enumerate_system_va_type(cls, large_page_size: int, system_range_start: int,
|
||||
module: interfaces.context.ModuleInterface,
|
||||
type_array: interfaces.objects.ObjectInterface) -> Dict[str, List[Tuple[int, int]]]:
|
||||
result = {} # type: Dict[str, List[Tuple[int, int]]]
|
||||
result: Dict[str, List[Tuple[int, int]]] = {}
|
||||
system_va_type = module.get_enumeration('_MI_SYSTEM_VA_TYPE')
|
||||
start = system_range_start
|
||||
prev_entry = -1
|
||||
|
||||
@@ -158,8 +158,8 @@ class TreeGrid(interfaces.renderers.TreeGrid):
|
||||
"""
|
||||
self._populated = False
|
||||
self._row_count = 0
|
||||
self._children = [] # type: List[interfaces.renderers.TreeNode]
|
||||
converted_columns = [] # type: List[interfaces.renderers.Column]
|
||||
self._children: List[interfaces.renderers.TreeNode] = []
|
||||
converted_columns: List[interfaces.renderers.Column] = []
|
||||
if len(columns) < 1:
|
||||
raise ValueError("Columns must be a list containing at least one column")
|
||||
for (name, column_type) in columns:
|
||||
@@ -207,7 +207,7 @@ class TreeGrid(interfaces.renderers.TreeGrid):
|
||||
|
||||
if not self.populated:
|
||||
try:
|
||||
prev_nodes = [] # type: List[interfaces.renderers.TreeNode]
|
||||
prev_nodes: List[interfaces.renderers.TreeNode] = []
|
||||
for (level, item) in self._generator:
|
||||
parent_index = min(len(prev_nodes), level)
|
||||
parent = prev_nodes[parent_index - 1] if parent_index > 0 else None
|
||||
|
||||
@@ -24,7 +24,7 @@ def wintime_to_datetime(wintime: int) -> Union[interfaces.renderers.BaseAbsentVa
|
||||
|
||||
|
||||
def unixtime_to_datetime(unixtime: int) -> Union[interfaces.renderers.BaseAbsentValue, datetime.datetime]:
|
||||
ret = renderers.UnparsableValue() # type: Union[interfaces.renderers.BaseAbsentValue, datetime.datetime]
|
||||
ret: Union[interfaces.renderers.BaseAbsentValue, datetime.datetime] = renderers.UnparsableValue()
|
||||
|
||||
if unixtime > 0:
|
||||
try:
|
||||
|
||||
@@ -46,7 +46,7 @@ class MultiTypeData(bytes):
|
||||
encoding: str = 'utf-16-le',
|
||||
split_nulls: bool = False,
|
||||
show_hex: bool = False) -> None:
|
||||
self.converted_int = False # type: bool
|
||||
self.converted_int: bool = False
|
||||
if isinstance(original, int):
|
||||
self.converted_int = True
|
||||
self.encoding = encoding
|
||||
|
||||
@@ -31,15 +31,15 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self._dict = collections.OrderedDict() # type: Dict[str, interfaces.symbols.BaseSymbolTableInterface]
|
||||
self._dict: Dict[str, interfaces.symbols.BaseSymbolTableInterface] = collections.OrderedDict()
|
||||
# Permanently cache all resolved symbols
|
||||
self._resolved = {} # type: Dict[str, interfaces.objects.Template]
|
||||
self._resolved_symbols = {} # type: Dict[str, interfaces.objects.Template]
|
||||
self._resolved: Dict[str, interfaces.objects.Template] = {}
|
||||
self._resolved_symbols: Dict[str, interfaces.objects.Template] = {}
|
||||
|
||||
def clear_symbol_cache(self, table_name: str = None) -> None:
|
||||
"""Clears the symbol cache for the specified table name. If no table
|
||||
name is specified, the caches of all symbol tables are cleared."""
|
||||
table_list = list() # type: List[interfaces.symbols.BaseSymbolTableInterface]
|
||||
table_list: List[interfaces.symbols.BaseSymbolTableInterface] = list()
|
||||
if table_name is None:
|
||||
table_list = list(self._dict.values())
|
||||
else:
|
||||
|
||||
@@ -282,8 +282,8 @@ class ISFormatTable(interfaces.symbols.SymbolTableInterface, metaclass = ABCMeta
|
||||
raise TypeError("Native table not provided")
|
||||
nt.name = name + "_natives"
|
||||
super().__init__(context, config_path, name, nt, table_mapping = table_mapping)
|
||||
self._overrides = {} # type: Dict[str, Type[interfaces.objects.ObjectInterface]]
|
||||
self._symbol_cache = {} # type: Dict[str, interfaces.symbols.SymbolInterface]
|
||||
self._overrides: Dict[str, Type[interfaces.objects.ObjectInterface]] = {}
|
||||
self._symbol_cache: Dict[str, interfaces.symbols.SymbolInterface] = {}
|
||||
|
||||
def _get_natives(self) -> Optional[interfaces.symbols.NativeTableInterface]:
|
||||
"""Determines the appropriate native_types to use from the JSON
|
||||
|
||||
@@ -46,7 +46,7 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
|
||||
@classmethod
|
||||
def _do_get_path(cls, rdentry, rmnt, dentry, vfsmnt) -> str:
|
||||
|
||||
ret_path = [] # type: List[str]
|
||||
ret_path: List[str] = []
|
||||
|
||||
while dentry != rdentry or vfsmnt != rmnt:
|
||||
dname = dentry.path()
|
||||
|
||||
@@ -160,7 +160,7 @@ class MacUtilities(interfaces.configuration.VersionableInterface):
|
||||
list_next_member: str,
|
||||
next_member: str,
|
||||
max_elements: int = 4096) -> Iterable[interfaces.objects.ObjectInterface]:
|
||||
seen = set() # type: Set[int]
|
||||
seen: Set[int] = set()
|
||||
|
||||
try:
|
||||
current = queue.member(attr = list_head_member)
|
||||
|
||||
@@ -48,7 +48,7 @@ class proc(generic.GenericIntelProcess):
|
||||
except exceptions.InvalidAddressException:
|
||||
return
|
||||
|
||||
seen = set() # type: Set[int]
|
||||
seen: Set[int] = set()
|
||||
|
||||
for i in range(task.map.hdr.nentries):
|
||||
if not current_map or current_map.vol.offset in seen:
|
||||
|
||||
@@ -15,7 +15,7 @@ class NativeTable(interfaces.symbols.NativeTableInterface):
|
||||
def __init__(self, name: str, native_dictionary: Dict[str, Any]) -> None:
|
||||
super().__init__(name, self)
|
||||
self._native_dictionary = copy.deepcopy(native_dictionary)
|
||||
self._overrides = {} # type: Dict[str, interfaces.objects.ObjectInterface]
|
||||
self._overrides: Dict[str, interfaces.objects.ObjectInterface] = {}
|
||||
for native_type in self._native_dictionary:
|
||||
native_class, _native_struct = self._native_dictionary[native_type]
|
||||
self._overrides[native_type] = native_class
|
||||
@@ -49,8 +49,8 @@ class NativeTable(interfaces.symbols.NativeTableInterface):
|
||||
table_name, type_name = name_split
|
||||
prefix = table_name + constants.BANG
|
||||
|
||||
additional = {} # type: Dict[str, Any]
|
||||
obj = None # type: Optional[Type[interfaces.objects.ObjectInterface]]
|
||||
additional: Dict[str, Any] = {}
|
||||
obj: Optional[Type[interfaces.objects.ObjectInterface]] = None
|
||||
if type_name == 'void' or type_name == 'function':
|
||||
obj = objects.Void
|
||||
elif type_name == 'array':
|
||||
|
||||
@@ -388,7 +388,7 @@ class FILE_OBJECT(objects.StructType, pool.ExecutiveObject):
|
||||
self.FileName.Buffer)
|
||||
|
||||
def file_name_with_device(self) -> Union[str, interfaces.renderers.BaseAbsentValue]:
|
||||
name = renderers.UnreadableValue() # type: Union[str, interfaces.renderers.BaseAbsentValue]
|
||||
name: Union[str, interfaces.renderers.BaseAbsentValue] = renderers.UnreadableValue()
|
||||
|
||||
# this pointer needs to be checked against native_layer_name because the object may
|
||||
# be instantiated from a primary (virtual) layer or a memory (physical) layer.
|
||||
|
||||
@@ -214,7 +214,7 @@ class POOL_HEADER_VISTA(POOL_HEADER):
|
||||
class POOL_TRACKER_BIG_PAGES(objects.StructType):
|
||||
"""A kernel big page pool tracker."""
|
||||
|
||||
pool_type_lookup = {} # type: Dict[str, str]
|
||||
pool_type_lookup: Dict[str, str] = {}
|
||||
|
||||
def _generate_pool_type_lookup(self):
|
||||
# Enumeration._generate_inverse_choices() raises ValueError because multiple enum names map to the same
|
||||
|
||||
@@ -265,18 +265,18 @@ class PdbReader:
|
||||
database_name: Optional[str] = None,
|
||||
progress_callback: constants.ProgressCallback = None) -> None:
|
||||
self._layer_name, self._context = self.load_pdb_layer(context, location)
|
||||
self._dbiheader = None # type: Optional[interfaces.objects.ObjectInterface]
|
||||
self._dbiheader: Optional[interfaces.objects.ObjectInterface] = None
|
||||
if not progress_callback:
|
||||
progress_callback = lambda x, y: None
|
||||
self._progress_callback = progress_callback
|
||||
self.types = [
|
||||
] # type: List[Tuple[interfaces.objects.ObjectInterface, Optional[str], interfaces.objects.ObjectInterface]]
|
||||
self.bases = {} # type: Dict[str, Any]
|
||||
self.user_types = {} # type: Dict[str, Any]
|
||||
self.enumerations = {} # type: Dict[str, Any]
|
||||
self.symbols = {} # type: Dict[str, Any]
|
||||
self._omap_mapping = [] # type: List[Tuple[int, int]]
|
||||
self._sections = [] # type: List[interfaces.objects.ObjectInterface]
|
||||
self.types: List[Tuple[interfaces.objects.ObjectInterface, Optional[str], interfaces.objects.ObjectInterface]] = [
|
||||
]
|
||||
self.bases: Dict[str, Any] = {}
|
||||
self.user_types: Dict[str, Any] = {}
|
||||
self.enumerations: Dict[str, Any] = {}
|
||||
self.symbols: Dict[str, Any] = {}
|
||||
self._omap_mapping: List[Tuple[int, int]] = []
|
||||
self._sections: List[interfaces.objects.ObjectInterface] = []
|
||||
self.metadata = {"format": "6.1.0", "windows": {}}
|
||||
self._database_name = database_name
|
||||
|
||||
@@ -381,7 +381,7 @@ class PdbReader:
|
||||
raise ValueError("Maximum {} index is smaller than minimum TPI index, found: {} < {} ".format(
|
||||
stream_name, header.index_max, header.index_min))
|
||||
# Reset the state
|
||||
info_references = {} # type: Dict[str, int]
|
||||
info_references: Dict[str, int] = {}
|
||||
offset = header.header_size
|
||||
# Ensure we use the same type everywhere
|
||||
length_type = "unsigned short"
|
||||
@@ -586,7 +586,7 @@ class PdbReader:
|
||||
if index < 0x1000:
|
||||
base_name, base = primatives[index & 0xff]
|
||||
self.bases[base_name] = base
|
||||
result = {"kind": "base", "name": base_name} # type: Union[List[Dict[str, Any]], Dict[str, Any]]
|
||||
result: Union[List[Dict[str, Any]], Dict[str, Any]] = {"kind": "base", "name": base_name}
|
||||
indirection = (index & 0xf00)
|
||||
if indirection:
|
||||
pointer_name, pointer_base = indirections[indirection]
|
||||
@@ -639,7 +639,7 @@ class PdbReader:
|
||||
"""Returns the size of the structure based on the type index
|
||||
provided."""
|
||||
result = -1
|
||||
name = '' # type: Optional[str]
|
||||
name: Optional[str] = ''
|
||||
if index < 0x1000:
|
||||
if (index & 0xf00):
|
||||
_, base = indirections[index & 0xf00]
|
||||
@@ -837,7 +837,7 @@ class PdbReader:
|
||||
|
||||
def convert_fields(self, fields: int) -> Dict[Optional[str], Dict[str, Any]]:
|
||||
"""Converts a field list into a list of fields."""
|
||||
result = {} # type: Dict[Optional[str], Dict[str, Any]]
|
||||
result: Dict[Optional[str], Dict[str, Any]] = {}
|
||||
_, _, fields_struct = self.types[fields]
|
||||
if not isinstance(fields_struct, list):
|
||||
vollog.warning("Fields structure did not contain a list of fields")
|
||||
|
||||
@@ -18,7 +18,7 @@ cached_validation_filepath = os.path.join(constants.CACHE_PATH, "valid_isf.hashc
|
||||
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() # type: Set
|
||||
validhashes: Set = set()
|
||||
if os.path.exists(cached_validation_filepath):
|
||||
with open(cached_validation_filepath, "r") as f:
|
||||
validhashes.update(json.load(f))
|
||||
|
||||
Reference in New Issue
Block a user