From 942de5f166aa1cd232e2f91a79abb8d2c1ba3845 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Sun, 10 Dec 2017 19:31:41 +0000 Subject: [PATCH] Finish adding type-annotations thoughout the code. --- volatility/framework/__init__.py | 7 +- volatility/framework/exceptions.py | 10 +- volatility/framework/interfaces/objects.py | 2 +- volatility/framework/interfaces/symbols.py | 4 +- .../framework/symbols/generic/__init__.py | 7 +- volatility/framework/symbols/intermed.py | 101 ++++++++++-------- .../framework/symbols/linux/__init__.py | 11 +- .../symbols/linux/extensions/__init__.py | 39 ++++--- volatility/framework/symbols/native.py | 20 ++-- .../framework/symbols/windows/__init__.py | 11 +- .../symbols/windows/extensions/__init__.py | 23 ++-- .../symbols/windows/extensions/registry.py | 23 ++-- volatility/framework/symbols/wrappers.py | 7 +- 13 files changed, 164 insertions(+), 101 deletions(-) diff --git a/volatility/framework/__init__.py b/volatility/framework/__init__.py index 5dbfc89de..395195aca 100644 --- a/volatility/framework/__init__.py +++ b/volatility/framework/__init__.py @@ -33,7 +33,7 @@ def interface_version(): vollog = logging.getLogger(__name__) -def require_interface_version(*args): +def require_interface_version(*args) -> None: """Checks the required version of a plugin""" if len(args): if args[0] != interface_version()[0]: @@ -66,12 +66,13 @@ def hide_from_subclasses(cls: typing.Type) -> typing.Type: return cls -def class_subclasses(cls): +def class_subclasses(cls: typing.Type) -> typing.Iterable[typing.Type]: """Returns all the (recursive) subclasses of a given class""" if not inspect.isclass(cls): raise TypeError("class_subclasses parameter not a valid class: {}".format(cls)) for clazz in cls.__subclasses__(): - if not hasattr(clazz, 'hidden') or not clazz.hidden: + # The typing system is not clever enough to realize that clazz has a hidden attr after the hasattr check + if not hasattr(clazz, 'hidden') or not clazz.hidden: # type: ignore yield clazz for return_value in class_subclasses(clazz): yield return_value diff --git a/volatility/framework/exceptions.py b/volatility/framework/exceptions.py index 09078eb6f..5e3bfa1dd 100644 --- a/volatility/framework/exceptions.py +++ b/volatility/framework/exceptions.py @@ -33,7 +33,11 @@ class PagedInvalidAddressException(InvalidAddressException): Includes the invalid address and the number of bits of the address that are invalid """ - def __init__(self, layer_name, invalid_address, invalid_bits, *args): + def __init__(self, + layer_name: str, + invalid_address: int, + invalid_bits: int, + *args) -> None: super().__init__(layer_name, invalid_address, *args) self.invalid_bits = invalid_bits @@ -41,8 +45,8 @@ class PagedInvalidAddressException(InvalidAddressException): class InvalidDataException(VolatilityException): """Thrown when an object contains some data known to be invalid for that structure""" - def __init__(self, invalid_object, *args): - super.__init__(invalid_object, *args) + def __init__(self, invalid_object: object, *args) -> None: + super().__init__(invalid_object, *args) self._invalid_object = invalid_object diff --git a/volatility/framework/interfaces/objects.py b/volatility/framework/interfaces/objects.py index 3588312fa..548799951 100644 --- a/volatility/framework/interfaces/objects.py +++ b/volatility/framework/interfaces/objects.py @@ -19,7 +19,7 @@ class ReadOnlyMapping(validity.ValidityRoutines, collections.abc.Mapping): This ensures that the data stored in the mapping should not be modified, making an immutable mapping. """ - def __init__(self, dictionary: typing.ChainMap[str, typing.Any]) -> None: + def __init__(self, dictionary: typing.Mapping[str, typing.Any]) -> None: self._dict = dictionary def __getattr__(self, attr: str) -> typing.Any: diff --git a/volatility/framework/interfaces/symbols.py b/volatility/framework/interfaces/symbols.py index 263607732..cb8f8e05f 100644 --- a/volatility/framework/interfaces/symbols.py +++ b/volatility/framework/interfaces/symbols.py @@ -174,14 +174,14 @@ class BaseSymbolTableInterface(validity.ValidityRoutines): # ## Functions for overriding classes - def set_type_class(self, name: str, clazz: objects.ObjectInterface) -> None: + def set_type_class(self, name: str, clazz: typing.Type[objects.ObjectInterface]) -> None: """Overrides the object class for a specific Symbol type Name *must* be present in self.types """ raise NotImplementedError("Abstract method set_type_class not implemented yet.") - def get_type_class(self, name: str) -> objects.ObjectInterface: + def get_type_class(self, name: str) -> typing.Type[objects.ObjectInterface]: """Returns the class associated with a Symbol type""" raise NotImplementedError("Abstract method get_type_class not implemented yet.") diff --git a/volatility/framework/symbols/generic/__init__.py b/volatility/framework/symbols/generic/__init__.py index e414c1002..899659d23 100644 --- a/volatility/framework/symbols/generic/__init__.py +++ b/volatility/framework/symbols/generic/__init__.py @@ -1,11 +1,16 @@ import random import string +import typing from volatility.framework import objects, interfaces class GenericIntelProcess(objects.Struct): - def _add_process_layer(self, context, dtb, config_prefix = None, preferred_name = None): + def _add_process_layer(self, + context: interfaces.context.ContextInterface, + dtb: typing.Union[int, interfaces.objects.ObjectInterface], + config_prefix: str = None, + preferred_name: str = None) -> str: """Constructs a new layer based on the process's DirectoryTableBase""" if config_prefix is None: diff --git a/volatility/framework/symbols/intermed.py b/volatility/framework/symbols/intermed.py index 1e1aece41..7f34309ad 100644 --- a/volatility/framework/symbols/intermed.py +++ b/volatility/framework/symbols/intermed.py @@ -5,7 +5,9 @@ import json import logging import os import pathlib +import typing import zipfile +from abc import ABCMeta from volatility import schemas from volatility.framework import class_subclasses, constants, exceptions, interfaces, objects, layers @@ -38,7 +40,7 @@ vollog = logging.getLogger(__name__) # for container types # -def _construct_delegate_function(name, is_property = False): +def _construct_delegate_function(name: str, is_property: bool = False) -> typing.Any: def _delegate_function(self, *args, **kwargs): if is_property: return getattr(self._delegate, name) @@ -50,30 +52,30 @@ def _construct_delegate_function(name, is_property = False): class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface): - def __init__(self, context, config_path, name, isf_url, native_types = None, validate = True): + def __init__(self, + context: interfaces.context.ContextInterface, + config_path: str, + name: str, + isf_url: str, + native_types: interfaces.symbols.NativeTableInterface = None, + validate: bool = True) -> None: """Instantiates an SymbolTable based on an IntermediateSymbolFormat JSON file. This is validated against the appropriate schema. The validation can be disabled by passing validate = False, but this should almost never be done. :param context: - :type context: :param config_path: - :type config_path: :param name: - :type name: :param isf_url: - :type isf_url: :param native_types: - :type native_types: :param validate: Determines whether the ISF file will be validated against the appropriate schema - :type validate: bool """ # Check there are no obvious errors # Open the file and test the version self._versions = dict([(x.version, x) for x in class_subclasses(ISFormatTable)]) fp = layers.ResourceAccessor().open(isf_url) reader = codecs.getreader("utf-8") - json_object = json.load(reader(fp)) + json_object = json.load(reader(fp)) # type: ignore fp.close() # Validation is expensive, but we cache to store the hashes of successfully validated json objects @@ -92,7 +94,10 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface): # Inherit super().__init__(context, config_path, name, native_types or self._delegate.natives) - def _closest_version(self, version, versions): + def _closest_version(self, + version: str, + versions: typing.Dict[typing.Tuple[int, int, int], typing.Type['ISFormatTable']]) \ + -> typing.Type['ISFormatTable']: """Determines the highest suitable handler for specified version format An interface version such as (Current-Age).Age.Revision means that (Current - Age) of the provider must be equal to that of the @@ -117,7 +122,9 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface): del_type_class = _construct_delegate_function('del_type_class') @classmethod - def file_symbol_url(cls, sub_path, filename = None): + def file_symbol_url(cls, + sub_path: str, + filename: typing.Optional[str] = None) -> typing.Generator[str, None, None]: """Returns an iterator of appropriate file-scheme symbol URLs that can be opened by a ResourceAccessor class Filter reduces the number of results returned to only those URLs containing that string @@ -150,28 +157,25 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface): yield "jar:file:" + str(pathlib.Path(zip_path)) + "!" + name -class ISFormatTable(interfaces.symbols.SymbolTableInterface): +class ISFormatTable(interfaces.symbols.SymbolTableInterface, metaclass = ABCMeta): """Provide a base class to identify all subclasses""" - pass + version = (0, 0, 0) - -class Version1Format(ISFormatTable): - """Class for storing intermediate debugging data as objects and classes""" - current = 1 - revision = 0 - age = 1 - version = (current - age, age, revision) - - def __init__(self, context, config_path, name, json_object, native_types = None): + def __init__(self, + context: interfaces.context.ContextInterface, + config_path: str, + name: str, + json_object: typing.Any, + native_types: interfaces.symbols.NativeTableInterface = None) -> None: self._json_object = json_object self._validate_json() nt = native_types or self._get_natives() nt.name = name + "_natives" super().__init__(context, config_path, name, nt) - self._overrides = {} - self._symbol_cache = {} + self._overrides = {} # type: typing.Dict[str, typing.Type[interfaces.objects.ObjectInterface]] + self._symbol_cache = {} # type: typing.Dict[str, interfaces.symbols.Symbol] - def _get_natives(self): + def _get_natives(self) -> typing.Optional[interfaces.symbols.NativeTableInterface]: """Determines the appropriate native_types to use from the JSON data""" # TODO: Consider how to generate the natives entirely from the ISF classes = {"x64": native.x64NativeTable, "x86": native.x86NativeTable} @@ -187,10 +191,11 @@ class Version1Format(ISFormatTable): else: vollog.debug("Choosing appropriate natives for symbol library: {}".format(nc)) return native_class.natives + return None # TODO: Check the format and make use of the other metadata - def _validate_json(self): + def _validate_json(self) -> None: if (not 'user_types' in self._json_object or not 'base_types' in self._json_object or not 'metadata' in self._json_object or @@ -198,7 +203,15 @@ class Version1Format(ISFormatTable): not 'enums' in self._json_object): raise exceptions.SymbolSpaceError("Malformed JSON file provided") - def get_symbol(self, name): + +class Version1Format(ISFormatTable): + """Class for storing intermediate debugging data as objects and classes""" + current = 1 + revision = 0 + age = 1 + version = (current - age, age, revision) + + def get_symbol(self, name: str) -> interfaces.symbols.Symbol: """Returns the location offset given by the symbol name""" # TODO: Add the ability to add/remove/change symbols after creation # note that this should invalidate/update the cache @@ -211,33 +224,33 @@ class Version1Format(ISFormatTable): return self._symbol_cache[name] @property - def symbols(self): + def symbols(self) -> typing.Iterable[str]: """Returns an iterator of the symbol names""" return self._json_object.get('symbols', {}).keys() @property - def enumerations(self): + def enumerations(self) -> typing.Iterable[str]: """Returns an iterator of the available enumerations""" return self._json_object.get('enums', {}).keys() @property - def types(self): + def types(self) -> typing.Iterable[str]: """Returns an iterator of the symbol type names""" return list(self._json_object.get('user_types', {}).keys()) + list(self.natives.types) - def get_type_class(self, name): + def get_type_class(self, name: str) -> typing.Type[interfaces.objects.ObjectInterface]: return self._overrides.get(name, objects.Struct) - def set_type_class(self, name, clazz): + def set_type_class(self, name: str, clazz: typing.Type[interfaces.objects.ObjectInterface]) -> None: if name not in self.types: raise ValueError("Symbol type not in {} SymbolTable: {}".format(self.name, name)) self._overrides[name] = clazz - def del_type_class(self, name): + def del_type_class(self, name: str) -> None: if name in self._overrides: del self._overrides[name] - def _interdict_to_template(self, dictionary): + def _interdict_to_template(self, dictionary: typing.Dict[str, typing.Any]) -> interfaces.objects.Template: """Converts an intermediate format dict into an object template""" if not dictionary: raise exceptions.SymbolSpaceError("Invalid intermediate dictionary: {}".format(dictionary)) @@ -280,7 +293,7 @@ class Version1Format(ISFormatTable): return objects.templates.ReferenceTemplate(type_name = reference_name) - def _lookup_enum(self, name): + def _lookup_enum(self, name: str) -> typing.Dict[str, typing.Any]: """Looks up an enumeration and returns a dictionary of __init__ parameters for an Enum""" lookup = self._json_object['enums'].get(name, None) if not lookup: @@ -289,7 +302,7 @@ class Version1Format(ISFormatTable): "base_type": self.natives.get_type(lookup['base'])} return result - def get_enumeration(self, enum_name): + def get_enumeration(self, enum_name: str) -> interfaces.objects.Template: """Resolves an individual enumeration""" if constants.BANG in enum_name: raise exceptions.SymbolError("Enumeration for a different table requested: {}".format(enum_name)) @@ -304,7 +317,7 @@ class Version1Format(ISFormatTable): size = curdict['size'], choices = curdict['constants']) - def get_type(self, type_name): + def get_type(self, type_name: str) -> interfaces.objects.Template: """Resolves an individual symbol""" if constants.BANG in type_name: raise exceptions.SymbolError("Symbol for a different table requested: {}".format(type_name)) @@ -331,7 +344,7 @@ class Version2Format(Version1Format): age = 0 version = (current - age, age, revision) - def _get_natives(self): + def _get_natives(self) -> typing.Optional[interfaces.symbols.NativeTableInterface]: """Determines the appropriate native_types to use from the JSON data""" classes = {"x64": native.x64NativeTable, "x86": native.x86NativeTable} for nc in sorted(classes): @@ -346,8 +359,9 @@ class Version2Format(Version1Format): else: vollog.debug("Choosing appropriate natives for symbol library: {}".format(nc)) return native_class.natives + return None - def get_type(self, type_name): + def get_type(self, type_name: str) -> interfaces.objects.Template: """Resolves an individual symbol""" if constants.BANG in type_name: raise exceptions.SymbolError("Symbol for a different table requested: {}".format(type_name)) @@ -374,7 +388,7 @@ class Version3Format(Version2Format): age = 1 version = (current - age, age, revision) - def get_symbol(self, name): + def get_symbol(self, name: str) -> interfaces.symbols.Symbol: """Returns the symbol given by the symbol name""" if self._symbol_cache.get(name, None): return self._symbol_cache[name] @@ -407,7 +421,7 @@ class Version4Format(Version3Format): 'bool': ({1: '?'}, objects.Integer), 'char': ({1: 'c'}, objects.Char)} - def _get_natives(self): + def _get_natives(self) -> typing.Optional[interfaces.symbols.NativeTableInterface]: """Determines the appropriate native_types to use from the JSON data""" native_dict = {} base_types = self._json_object['base_types'] @@ -415,7 +429,8 @@ class Version4Format(Version3Format): # Void are ignored because voids are not a volatility primitive, they are a specific Volatility object if base_type != 'void': current = base_types[base_type] - size_map, object_type = self.format_str_mapping.get(current['kind'], ({}, None)) + # TODO: Fix up the typing of this, it bugs out because of the tuple assignment + size_map, object_type = self.format_str_mapping.get(current['kind'], ({}, None)) # type: ignore format_str = size_map.get(current['size'], None) if format_str is None or object_type is None: raise ValueError("Unsupported kind/size combination in base_type {}".format(base_type)) @@ -434,7 +449,7 @@ class Version5Format(Version4Format): age = 1 version = (current - age, age, revision) - def get_symbol(self, name): + def get_symbol(self, name: str) -> interfaces.symbols.Symbol: """Returns the symbol given by the symbol name""" if self._symbol_cache.get(name, None): return self._symbol_cache[name] diff --git a/volatility/framework/symbols/linux/__init__.py b/volatility/framework/symbols/linux/__init__.py index 8219e35aa..f8d365294 100644 --- a/volatility/framework/symbols/linux/__init__.py +++ b/volatility/framework/symbols/linux/__init__.py @@ -1,3 +1,6 @@ +import typing + +from volatility.framework import interfaces from volatility.framework.configuration import requirements from volatility.framework.symbols import intermed from volatility.framework.symbols.linux import extensions @@ -6,7 +9,11 @@ from volatility.framework.symbols.linux import extensions class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable): provides = {"type": "interface"} - def __init__(self, context, config_path, name, isf_url): + def __init__(self, + context: interfaces.context.ContextInterface, + config_path: str, + name: str, + isf_url: str) -> None: super().__init__(context = context, config_path = config_path, name = name, isf_url = isf_url) # Set-up Linux specific types @@ -18,6 +25,6 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable): self.set_type_class('vm_area_struct', extensions.vm_area_struct) @classmethod - def get_requirements(cls): + def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]: return [requirements.StringRequirement("isf_url", description = "JSON file containing the symbols encoded in the Intermediate Symbol Format")] diff --git a/volatility/framework/symbols/linux/extensions/__init__.py b/volatility/framework/symbols/linux/extensions/__init__.py index 642d4e2d5..880668a4c 100644 --- a/volatility/framework/symbols/linux/extensions/__init__.py +++ b/volatility/framework/symbols/linux/extensions/__init__.py @@ -1,16 +1,19 @@ import collections.abc +import typing -from volatility.framework import objects from volatility.framework import constants -from volatility.framework.symbols import generic +from volatility.framework import objects, interfaces from volatility.framework.objects import utility +from volatility.framework.symbols import generic # Keep these in a basic module, to prevent import cycles when symbol providers require them class task_struct(generic.GenericIntelProcess): - def add_process_layer(self, config_prefix = None, preferred_name = None): + def add_process_layer(self, + config_prefix: str = None, + preferred_name: str = None) -> typing.Optional[str]: """Constructs a new layer based on the process's DTB. Returns the name of the Layer or None. """ @@ -30,7 +33,7 @@ class task_struct(generic.GenericIntelProcess): class mm_struct(objects.Struct): @property - def mmap_iter(self): + def mmap_iter(self) -> typing.Iterable[interfaces.objects.ObjectInterface]: """Returns an iterator for the mmap list member of an mm_struct.""" if not self.mmap: @@ -52,22 +55,22 @@ class super_block(objects.Struct): MINORBITS = 20 @property - def major(self): + def major(self) -> int: return self.s_dev >> self.MINORBITS @property - def minor(self): + def minor(self) -> int: return self.s_dev & ((1 << self.MINORBITS) - 1) class vm_area_struct(objects.Struct): # include/linux/mm.h - VM_READ = 0x00000001 + VM_READ = 0x00000001 VM_WRITE = 0x00000002 - VM_EXEC = 0x00000004 + VM_EXEC = 0x00000004 @property - def flags(self): + def flags(self) -> str: """Returns an rwx string representation of the flags in a vm_area_struct.""" retval = "" @@ -80,19 +83,20 @@ class vm_area_struct(objects.Struct): return retval - def page_offset(self): + def page_offset(self) -> int: if self.vm_file == 0: return 0 return self.vm_pgoff << constants.linux.PAGE_SHIFT + class struct_file(objects.Struct): @property - def full_path(self): - parts = [] + def full_path(self) -> str: + parts = [] # type: typing.List[str] path = self.f_path path_dentry = path.dentry - seen = set() + seen = set() # type: typing.Set[int] while path_dentry != 0 and path_dentry.vol.offset not in seen: name = utility.pointer_to_string(path_dentry.d_name.name, path_dentry.d_name.len) if name == "/": @@ -105,7 +109,12 @@ class struct_file(objects.Struct): class list_head(objects.Struct, collections.abc.Iterable): - def to_list(self, symbol_type, member, forward = True, sentinel = True, layer = None): + def to_list(self, + symbol_type: str, + member: str, + forward: bool = True, + sentinel: bool = True, + layer: typing.Optional[str] = None) -> typing.Iterator[interfaces.objects.ObjectInterface]: """Returns an iterator of the entries in the list.""" if layer is None: @@ -130,5 +139,5 @@ class list_head(objects.Struct, collections.abc.Iterable): seen.add(link.vol.offset) link = getattr(link, direction).dereference() - def __iter__(self): + def __iter__(self) -> typing.Iterator[interfaces.objects.ObjectInterface]: return self.to_list(self.vol.parent.vol.type_name, self.vol.member_name) diff --git a/volatility/framework/symbols/native.py b/volatility/framework/symbols/native.py index 3cbaa02cf..c057d6920 100644 --- a/volatility/framework/symbols/native.py +++ b/volatility/framework/symbols/native.py @@ -1,4 +1,5 @@ import copy +import typing from volatility.framework import constants, interfaces, objects @@ -6,10 +7,13 @@ from volatility.framework import constants, interfaces, objects class NativeTable(interfaces.symbols.NativeTableInterface): """Symbol List that handles Native types""" - def __init__(self, name, native_dictionary): + # FIXME: typing the native_dictionary as typing.Tuple[interfaces.objects.ObjectInterface, str] throws many errors + def __init__(self, + name: str, + native_dictionary: typing.Dict[str, typing.Any]) -> None: super().__init__(name, self) self._native_dictionary = copy.deepcopy(native_dictionary) - self._overrides = {} + self._overrides = {} # type: typing.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 @@ -17,16 +21,16 @@ class NativeTable(interfaces.symbols.NativeTableInterface): self._types = set(self._native_dictionary).union( {'enum', 'array', 'bitfield', 'void', 'pointer', 'string', 'bytes', 'function'}) - def get_type_class(self, name): - ntype, fmt = self._native_dictionary.get(name, (objects.Integer, '')) + def get_type_class(self, name: str) -> typing.Type[interfaces.objects.ObjectInterface]: + ntype, _ = self._native_dictionary.get(name, (objects.Integer, None)) return ntype @property - def types(self): + def types(self) -> typing.Iterable[str]: """Returns an iterator of the symbol type names""" return self._types - def get_type(self, type_name): + def get_type(self, type_name: str) -> interfaces.objects.Template: """Resolves a symbol name into an object template symbol_space is used to resolve any subtype symbols if they don't exist in this list @@ -40,8 +44,8 @@ class NativeTable(interfaces.symbols.NativeTableInterface): table_name, type_name = name_split prefix = table_name + constants.BANG - additional = {} - obj = None + additional = {} # type: typing.Dict[str, typing.Any] + obj = None # type: typing.Type[interfaces.objects.ObjectInterface] if type_name == 'void' or type_name == 'function': obj = objects.Void elif type_name == 'array': diff --git a/volatility/framework/symbols/windows/__init__.py b/volatility/framework/symbols/windows/__init__.py index dbc3fe028..0409c4920 100644 --- a/volatility/framework/symbols/windows/__init__.py +++ b/volatility/framework/symbols/windows/__init__.py @@ -1,3 +1,6 @@ +import typing + +from volatility.framework import interfaces from volatility.framework.configuration import requirements from volatility.framework.symbols import intermed from volatility.framework.symbols.windows import extensions @@ -7,7 +10,11 @@ from volatility.framework.symbols.windows.extensions import registry class WindowsKernelIntermedSymbols(intermed.IntermediateSymbolTable): provides = {"type": "interface"} - def __init__(self, context, config_path, name, isf_url): + def __init__(self, + context: interfaces.context.ContextInterface, + config_path: str, + name: str, + isf_url: str) -> None: super().__init__(context = context, config_path = config_path, name = name, isf_url = isf_url) # Set-up windows specific types @@ -26,6 +33,6 @@ class WindowsKernelIntermedSymbols(intermed.IntermediateSymbolTable): self.set_type_class('_HMAP_ENTRY', registry._HMAP_ENTRY) @classmethod - def get_requirements(cls): + def get_requirements(cls) -> typing.List[interfaces.configuration.RequirementInterface]: return [requirements.StringRequirement("isf_url", description = "JSON file containing the symbols encoded in the Intermediate Symbol Format")] diff --git a/volatility/framework/symbols/windows/extensions/__init__.py b/volatility/framework/symbols/windows/extensions/__init__.py index 07ccbb70d..301747511 100644 --- a/volatility/framework/symbols/windows/extensions/__init__.py +++ b/volatility/framework/symbols/windows/extensions/__init__.py @@ -1,6 +1,7 @@ import collections.abc +import typing -from volatility.framework import constants, objects +from volatility.framework import constants, objects, interfaces from volatility.framework.symbols import generic from volatility.framework import exceptions @@ -108,14 +109,14 @@ class _OBJECT_HEADER(objects.Struct): class _ETHREAD(objects.Struct): - def owning_process(self, kernel_layer = None): + def owning_process(self, kernel_layer: str = None) -> interfaces.objects.ObjectInterface: """Return the EPROCESS that owns this thread""" return self.ThreadsProcess.dereference(kernel_layer) class _UNICODE_STRING(objects.Struct): @property - def helper_string(self): + def helper_string(self) -> interfaces.objects.ObjectInterface: # We explicitly do *not* catch errors here, we allow an exception to be thrown # (otherwise there's no way to determine anything went wrong) # It's up to the user of this method to catch exceptions @@ -126,7 +127,10 @@ class _UNICODE_STRING(objects.Struct): class _EPROCESS(generic.GenericIntelProcess): - def add_process_layer(self, context, config_prefix = None, preferred_name = None): + def add_process_layer(self, + context: interfaces.context.ContextInterface, + config_prefix: str = None, + preferred_name: str = None): """Constructs a new layer based on the process's DirectoryTableBase""" parent_layer = context.memory[self.vol.layer_name] @@ -140,7 +144,7 @@ class _EPROCESS(generic.GenericIntelProcess): # Add the constructed layer and return the name return self._add_process_layer(context, dtb, config_prefix, preferred_name) - def load_order_modules(self): + def load_order_modules(self) -> typing.Iterable[int]: """Generator for DLLs in the order that they were loaded""" if constants.BANG not in self.vol.type_name: @@ -162,7 +166,12 @@ class _EPROCESS(generic.GenericIntelProcess): class _LIST_ENTRY(objects.Struct, collections.abc.Iterable): - def to_list(self, symbol_type, member, forward = True, sentinel = True, layer = None): + def to_list(self, + symbol_type: str, + member: str, + forward: bool = True, + sentinel: bool = True, + layer: typing.Optional[str] = None) -> typing.Iterator[interfaces.objects.ObjectInterface]: """Returns an iterator of the entries in the list""" if layer is None: @@ -187,5 +196,5 @@ class _LIST_ENTRY(objects.Struct, collections.abc.Iterable): seen.add(link.vol.offset) link = getattr(link, direction).dereference() - def __iter__(self): + def __iter__(self) -> typing.Iterator[interfaces.objects.ObjectInterface]: return self.to_list(self.vol.parent.vol.type_name, self.vol.member_name) diff --git a/volatility/framework/symbols/windows/extensions/registry.py b/volatility/framework/symbols/windows/extensions/registry.py index c50a1605d..de72476cf 100644 --- a/volatility/framework/symbols/windows/extensions/registry.py +++ b/volatility/framework/symbols/windows/extensions/registry.py @@ -1,8 +1,9 @@ import enum import logging import struct +import typing -from volatility.framework import constants, exceptions, objects +from volatility.framework import constants, exceptions, objects, interfaces from volatility.framework.layers.registry import RegistryHive vollog = logging.getLogger(__name__) @@ -27,7 +28,7 @@ class RegValueTypes(enum.Enum): class _HMAP_ENTRY(objects.Struct): @property - def helper_block_offset(self): + def helper_block_offset(self) -> int: try: return self.PermanentBinAddress ^ (self.PermanentBinAddress & 0x3) except AttributeError: @@ -36,7 +37,7 @@ class _HMAP_ENTRY(objects.Struct): class _CMHIVE(objects.Struct): @property - def helper_name(self): + def helper_name(self) -> typing.Optional[interfaces.objects.ObjectInterface]: """Determine a name for the hive. Note that some attributes are unpredictably blank across different OS versions while others are populated, so we check all possibilities and take the first one that's not empty""" @@ -56,12 +57,12 @@ class _CM_KEY_NODE(objects.Struct): """Extension to allow traversal of registry keys""" @property - def helper_volatile(self): + def helper_volatile(self) -> bool: if not isinstance(self._context.memory[self.vol.layer_name], RegistryHive): raise ValueError("Cannot determine volatility of registry key without an offset in a RegistryHive layer") return bool(self.vol.offset & 0x80000000) - def get_subkeys(self): + def get_subkeys(self) -> typing.Iterable[interfaces.objects.ObjectInterface]: """Returns a list of the key nodes""" hive = self._context.memory[self.vol.layer_name] if not isinstance(hive, RegistryHive): @@ -100,7 +101,7 @@ class _CM_KEY_NODE(objects.Struct): vollog.log(constants.LOGLEVEL_VVV, "Node found with address outside the valid Hive size: {}".format(key_offset)) - def get_values(self): + def get_values(self) -> typing.Iterable[interfaces.objects.ObjectInterface]: """Returns a list of the Value nodes for a key""" hive = self._context.memory[self.vol.layer_name] if not isinstance(hive, RegistryHive): @@ -114,11 +115,11 @@ class _CM_KEY_NODE(objects.Struct): yield node @property - def helper_name(self): + def helper_name(self) -> interfaces.objects.ObjectInterface: """Since this is just a casting convenience, it can be a property""" return self.Name.cast("string", max_length = self.NameLength, encoding = "latin-1") - def get_key_path(self): + def get_key_path(self) -> interfaces.objects.ObjectInterface: reg = self._context.memory[self.vol.layer_name] # Using the offset adds a significant delay (since it cannot be cached easily) # if self.vol.offset == reg.get_node(reg.root_cell_offset).vol.offset: @@ -131,12 +132,12 @@ class _CM_KEY_VALUE(objects.Struct): """Extensions to extract data from CM_KEY_VALUE nodes""" @property - def helper_name(self): + def helper_name(self) -> interfaces.objects.ObjectInterface: """Since this is just a casting convenience, it can be a property""" self.Name.count = self.NameLength return self.Name.cast("string", max_length = self.NameLength, encoding = "latin-1") - def decode_data(self): + def decode_data(self) -> typing.Union[str, bytes]: """Since this is just a casting convenience, it can be a property""" # Determine if the data is stored inline datalen = self.DataLength & 0x7fffffff @@ -184,7 +185,7 @@ class _CM_KEY_VALUE(objects.Struct): output = output[:output.find("\x00")] return output if self_type == RegValueTypes.REG_MULTI_SZ: - return str(data, encoding = "utf-16-le").split("\x00") + return str(data, encoding = "utf-16-le").split("\x00")[0] if self_type == RegValueTypes.REG_BINARY: return data if self_type == RegValueTypes.REG_NONE: diff --git a/volatility/framework/symbols/wrappers.py b/volatility/framework/symbols/wrappers.py index 94c6f2640..bb9ff4c16 100644 --- a/volatility/framework/symbols/wrappers.py +++ b/volatility/framework/symbols/wrappers.py @@ -1,4 +1,5 @@ import collections +import typing from volatility.framework import interfaces, validity @@ -6,7 +7,7 @@ from volatility.framework import interfaces, validity class Flags(validity.ValidityRoutines): """Object that converts an integer into a set of flags based on their masks""" - def __init__(self, choices = None): + def __init__(self, choices: typing.Mapping[str, int] = None) -> None: self._check_type(choices, collections.Mapping) for k, v in choices.items(): self._check_type(k, str) @@ -14,10 +15,10 @@ class Flags(validity.ValidityRoutines): self._choices = interfaces.objects.ReadOnlyMapping(choices) @property - def choices(self): + def choices(self) -> interfaces.objects.ReadOnlyMapping: return self._choices - def __call__(self, value): + def __call__(self, value: int) -> typing.List[str]: """Return the appropriate Flags """ result = [] for k, v in self.choices.items():