From 7a52ac9debc6e0463a52ecc1c5cacc5dbb40a0d1 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Mon, 4 Jun 2018 01:25:02 +0100 Subject: [PATCH] Fix a large number of typing issues. There are several instances where mypy didn't detect if blah is not None: blah = thing and so were rewritten as: blah = blah or thing --- volatility/cli/__init__.py | 7 ++++++- volatility/cli/text_renderer.py | 5 +++-- volatility/framework/contexts/__init__.py | 4 ++-- .../framework/interfaces/configuration.py | 10 +++++----- volatility/framework/interfaces/context.py | 4 ++-- volatility/framework/interfaces/layers.py | 4 ++-- volatility/framework/interfaces/objects.py | 7 +++---- volatility/framework/interfaces/plugins.py | 4 +++- volatility/framework/interfaces/renderers.py | 8 ++------ volatility/framework/interfaces/symbols.py | 20 +++++++++++-------- volatility/framework/layers/__init__.py | 8 ++++++-- volatility/framework/layers/physical.py | 5 ++--- volatility/framework/layers/registry.py | 2 +- .../framework/layers/scanners/wumanber.py | 3 ++- volatility/framework/objects/__init__.py | 14 ++++++------- volatility/framework/objects/templates.py | 4 ++-- volatility/framework/renderers/__init__.py | 9 +++++---- volatility/framework/symbols/__init__.py | 2 +- volatility/framework/symbols/intermed.py | 2 ++ .../symbols/linux/extensions/__init__.py | 4 +--- volatility/framework/symbols/native.py | 2 +- .../symbols/windows/extensions/__init__.py | 3 +-- 22 files changed, 70 insertions(+), 61 deletions(-) diff --git a/volatility/cli/__init__.py b/volatility/cli/__init__.py index b7bf144e7..d1eb1e98b 100644 --- a/volatility/cli/__init__.py +++ b/volatility/cli/__init__.py @@ -55,6 +55,11 @@ class PrintedProgress(object): print(message, end = ' ' * (self._max_message_len - message_len)) +class MuteProgress(PrintedProgress): + def __call__(self, progress, description = None): + pass + + class CommandLine(interfaces.plugins.FileConsumerInterface): """Constructs a command-line interface object for users to run plugins""" @@ -207,7 +212,7 @@ class CommandLine(interfaces.plugins.FileConsumerInterface): """ progress_callback = PrintedProgress() if quiet: - progress_callback = None + progress_callback = MuteProgress() errors = automagic.run(automagics, context, plugin, "plugins", progress_callback = progress_callback) # Check all the requirements and/or go back to the automagic step diff --git a/volatility/cli/text_renderer.py b/volatility/cli/text_renderer.py index da4279e69..458291b23 100644 --- a/volatility/cli/text_renderer.py +++ b/volatility/cli/text_renderer.py @@ -57,8 +57,9 @@ def display_disassembly(disasm: interfaces.renderers.Disassembly) -> str: 'arm': capstone.Cs(capstone.CS_ARCH_ARM, capstone.CS_MODE_ARM), 'arm64': capstone.Cs(capstone.CS_ARCH_ARM64, capstone.CS_MODE_ARM)} output = "" - for i in disasm_types[disasm.architecture].disasm(disasm.data, disasm.offset): - output += "\n0x%x:\t%s\t%s" % (i.address, i.mnemonic, i.op_str) + if disasm.architecture is not None: + for i in disasm_types[disasm.architecture].disasm(disasm.data, disasm.offset): + output += "\n0x%x:\t%s\t%s" % (i.address, i.mnemonic, i.op_str) return output return QuickTextRenderer.type_renderers[bytes](disasm.data) diff --git a/volatility/framework/contexts/__init__.py b/volatility/framework/contexts/__init__.py index 8e2dae030..18601b17e 100644 --- a/volatility/framework/contexts/__init__.py +++ b/volatility/framework/contexts/__init__.py @@ -117,8 +117,8 @@ def get_module_wrapper(method: str) -> typing.Callable: class Module(interfaces.context.Module): def object(self, - symbol_name: str = None, - type_name: str = None, + symbol_name: str, + type_name: str, offset: int = None, **kwargs) -> interfaces.objects.ObjectInterface: """Returns an object created using the symbol_table and layer_name of the Module diff --git a/volatility/framework/interfaces/configuration.py b/volatility/framework/interfaces/configuration.py index 7fd9c125c..6e7b90a06 100644 --- a/volatility/framework/interfaces/configuration.py +++ b/volatility/framework/interfaces/configuration.py @@ -95,7 +95,7 @@ class HierarchicalDict(collections.abc.Mapping): """ if self.separator in key: return key[key.index(self.separator) + 1:] - return None + return '' def __iter__(self): """Returns an iterator object that supports the iterator protocol""" @@ -134,7 +134,7 @@ class HierarchicalDict(collections.abc.Mapping): if is_data: self._data[key] = value else: - if not isinstance(value, HierarchicalDict) and value is not None: + if not isinstance(value, HierarchicalDict): raise TypeError( "HierarchicalDicts can only store HierarchicalDicts within their structure: {}".format( type(value))) @@ -166,7 +166,7 @@ class HierarchicalDict(collections.abc.Mapping): """Returns the length of all items""" return len(self._data) + sum([len(subdict) for subdict in self._subdict]) - def branch(self, key: str) -> typing.Optional['HierarchicalDict']: + def branch(self, key: str) -> 'HierarchicalDict': """Returns the HierarchicalDict housed under the key This differs from the data property, in that it is directed by the `key`, and all layers under that key are @@ -239,7 +239,7 @@ class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): def __init__(self, name: str, description: str = None, - default: ConfigSimpleType = None, + default: typing.Optional[ConfigSimpleType] = None, optional: bool = False) -> None: super().__init__() self._check_type(name, str) @@ -265,7 +265,7 @@ class RequirementInterface(validity.ValidityRoutines, metaclass = ABCMeta): return self._description @property - def default(self) -> ConfigSimpleType: + def default(self) -> typing.Optional[ConfigSimpleType]: """Returns the default value if one is set""" return self._default diff --git a/volatility/framework/interfaces/context.py b/volatility/framework/interfaces/context.py index 23d1732d7..174e5ced4 100644 --- a/volatility/framework/interfaces/context.py +++ b/volatility/framework/interfaces/context.py @@ -104,8 +104,8 @@ class Module(validity.ValidityRoutines, metaclass = ABCMeta): @abstractmethod def object(self, - symbol_name: str = None, - type_name: str = None, + symbol_name: str, + type_name: str, offset: int = None, **kwargs) -> 'interfaces.objects.ObjectInterface': """Returns an object created using the symbol_table and layer_name of the Module""" diff --git a/volatility/framework/interfaces/layers.py b/volatility/framework/interfaces/layers.py index 0293d7800..fc16136fd 100644 --- a/volatility/framework/interfaces/layers.py +++ b/volatility/framework/interfaces/layers.py @@ -61,7 +61,7 @@ class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta): self._layer_name = None # type: typing.Optional[str] @property - def context(self) -> 'interfaces.context.ContextInterface': + def context(self) -> typing.Optional['interfaces.context.ContextInterface']: return self._context @context.setter @@ -70,7 +70,7 @@ class ScannerInterface(validity.ValidityRoutines, metaclass = ABCMeta): self._context = self._check_type(ctx, context.ContextInterface) @property - def layer_name(self) -> str: + def layer_name(self) -> typing.Optional[str]: return self._layer_name @layer_name.setter diff --git a/volatility/framework/interfaces/objects.py b/volatility/framework/interfaces/objects.py index 7b391d50f..79c0d93c1 100644 --- a/volatility/framework/interfaces/objects.py +++ b/volatility/framework/interfaces/objects.py @@ -1,13 +1,12 @@ """Objects are the core of volatility, and provide pythonic access to interpreted values of data from a layer. """ +import collections +import collections.abc import logging import typing from abc import ABCMeta, abstractmethod -import collections -import collections.abc - from volatility.framework import constants, validity, interfaces from volatility.framework.interfaces import context as interfaces_context @@ -109,7 +108,7 @@ class ObjectInterface(validity.ValidityRoutines, metaclass = ABCMeta): Raises InvalidDataException on failure to validate the data correctly. """ - def get_symbol_table(self) -> 'interfaces.symbols.SymbolTableInterface': + def get_symbol_table(self) -> typing.Optional['interfaces.symbols.SymbolTableInterface']: """Returns the symbol table for this particular object Returns none if the symbol table cannot be identified. diff --git a/volatility/framework/interfaces/plugins.py b/volatility/framework/interfaces/plugins.py index fc0c7b9b6..191993fe3 100644 --- a/volatility/framework/interfaces/plugins.py +++ b/volatility/framework/interfaces/plugins.py @@ -24,6 +24,8 @@ class FileInterface(validity.ValidityRoutines, metaclass = ABCMeta): def __init__(self, filename: str, data: bytes = None) -> None: self.preferred_filename = filename + if data is None: + data = b'' self.data = io.BytesIO(data) @@ -69,7 +71,7 @@ class PluginInterface(interfaces_configuration.ConfigurableInterface, validity.V if self.unsatisfied(context, config_path): vollog.warning("Plugin failed validation") raise exceptions.PluginRequirementException("The plugin configuration failed to validate") - self._file_consumer = None # type: FileConsumerInterface + self._file_consumer = None # type: typing.Optional[FileConsumerInterface] def set_file_consumer(self, consumer: FileConsumerInterface) -> None: self._file_consumer = self._check_type(consumer, FileConsumerInterface) diff --git a/volatility/framework/interfaces/renderers.py b/volatility/framework/interfaces/renderers.py index 5594da6c4..643d0197b 100644 --- a/volatility/framework/interfaces/renderers.py +++ b/volatility/framework/interfaces/renderers.py @@ -58,7 +58,7 @@ class TreeNode(collections.Sequence, metaclass = ABCMeta): @property @abstractmethod - def parent(self) -> 'TreeNode': + def parent(self) -> typing.Optional['TreeNode']: """Returns the parent node of this node or None""" @property @@ -182,15 +182,11 @@ class TreeGrid(object, metaclass = ABCMeta): """Returns the path depth of a particular node""" return node.path_depth - def path_is_valid(self, node: TreeNode) -> bool: - """Returns True is a given path is valid for this treegrid""" - return node in self.children(node.parent) - @abstractmethod def visit(self, node: TreeNode, function: VisitorSignature, - initial_accumulator: _Type = None, + initial_accumulator: _Type, sort_key: ColumnSortKey = None) -> None: """Visits all the nodes in a tree, calling function on each one. diff --git a/volatility/framework/interfaces/symbols.py b/volatility/framework/interfaces/symbols.py index c4b61cddd..36a9b406b 100644 --- a/volatility/framework/interfaces/symbols.py +++ b/volatility/framework/interfaces/symbols.py @@ -42,6 +42,8 @@ class Symbol(validity.ValidityRoutines): def type_name(self) -> typing.Optional[str]: """Returns the name of the type that the symbol represents""" # Objects and ObjectTemplates should *always* get a type_name when they're constructed, so allow the IndexError + if self.type is None: + return None return self.type.vol['type_name'] @property @@ -70,11 +72,9 @@ class BaseSymbolTableInterface(validity.ValidityRoutines): def __init__(self, name: str, - native_types: typing.Optional['NativeTableInterface'] = None, + native_types: 'NativeTableInterface', table_mapping: typing.Optional[typing.Dict[str, str]] = None) -> None: - if name: - self._check_type(name, str) - self.name = name or None + self.name = self._check_type(name, str) if table_mapping is None: table_mapping = {} self.table_mapping = self._check_type(table_mapping, dict) @@ -150,9 +150,12 @@ class BaseSymbolTableInterface(validity.ValidityRoutines): # ## Convenience functions for location symbols - def get_symbol_type(self, name: str) -> objects.Template: + def get_symbol_type(self, name: str) -> typing.Optional[objects.Template]: """Resolves a symbol name into a symbol and then resolves the symbol's type""" - return self.get_type(self.get_symbol(name).type_name) + type_name = self.get_symbol(name).type_name + if type_name is None: + return None + return self.get_type(type_name) def get_symbols_by_type(self, type_name: str) -> typing.Iterable[str]: """Returns the name of all symbols in this table that have type matching type_name""" @@ -160,7 +163,8 @@ class BaseSymbolTableInterface(validity.ValidityRoutines): # This allows for searching with and without the table name (in case multiple tables contain # the same symbol name and we've not specifically been told which one) symbol = self.get_symbol(symbol_name) - if symbol.type_name == type_name or (symbol.type_name.endswith(constants.BANG + type_name)): + if symbol.type_name is not None and ( + symbol.type_name == type_name or (symbol.type_name.endswith(constants.BANG + type_name))): yield symbol.name def get_symbols_by_location(self, offset: int) -> typing.Iterable[str]: @@ -222,7 +226,7 @@ class SymbolTableInterface(BaseSymbolTableInterface, configuration.ConfigurableI context: 'interfaces_context.ContextInterface', config_path: str, name: str, - native_types: 'NativeTableInterface' = None, + native_types: 'NativeTableInterface', table_mapping: typing.Optional[typing.Dict[str, str]] = None) -> None: configuration.ConfigurableInterface.__init__(self, context, config_path) BaseSymbolTableInterface.__init__(self, name, native_types, table_mapping) diff --git a/volatility/framework/layers/__init__.py b/volatility/framework/layers/__init__.py index 54a5a5982..48844b2bd 100644 --- a/volatility/framework/layers/__init__.py +++ b/volatility/framework/layers/__init__.py @@ -13,8 +13,10 @@ import zipfile try: import magic + + HAS_MAGIC = True except ImportError: - magic = None + HAS_MAGIC = False try: import smb.SMBHandler @@ -93,13 +95,15 @@ class ResourceAccessor(object): # Determine whether the file is a particular type of file, and if so, open it as such IMPORTED_MAGIC = False - if not magic is None: + if HAS_MAGIC: while True: detected = None try: # Detect the content detected = magic.detect_from_fobj(curfile) IMPORTED_MAGIC = True + # This is because python-magic and file provide a magic module + # Only file's python has magic.detect_from_fobj except AttributeError: pass except: diff --git a/volatility/framework/layers/physical.py b/volatility/framework/layers/physical.py index 17ab7eac7..95c70bcdf 100644 --- a/volatility/framework/layers/physical.py +++ b/volatility/framework/layers/physical.py @@ -69,7 +69,7 @@ class FileLayer(interfaces.layers.DataLayerInterface): self._location = self.config["location"] self._accessor = layers.ResourceAccessor() - self._file_ = None + self._file_ = None # type: typing.Optional[typing.IO[typing.Any]] self._size = None # type: typing.Optional[int] # Instantiate the file to throw exceptions if the file doesn't open _ = self._file @@ -84,8 +84,7 @@ class FileLayer(interfaces.layers.DataLayerInterface): """Property to prevent the initializer storing an unserializable open file (for context cloning)""" # FIXME: Add "+" to the mode once we've determined whether write mode is enabled mode = "rb" - if not self._file_: - self._file_ = self._accessor.open(self._location, mode) + self._file_ = self._file_ or self._accessor.open(self._location, mode) return self._file_ @property diff --git a/volatility/framework/layers/registry.py b/volatility/framework/layers/registry.py index d0c96e6b0..0186c3f58 100644 --- a/volatility/framework/layers/registry.py +++ b/volatility/framework/layers/registry.py @@ -115,7 +115,7 @@ class RegistryHive(interfaces.layers.TranslationLayerInterface): found_key, key_array = found_key + [key_array[0]], key_array[1:] break else: - node_key = None + node_key = [] if not node_key: raise KeyError("Key {} not found under {}", key_array[0], '\\'.join(found_key)) if return_list: diff --git a/volatility/framework/layers/scanners/wumanber.py b/volatility/framework/layers/scanners/wumanber.py index fe5587a0f..b26fdd8ce 100644 --- a/volatility/framework/layers/scanners/wumanber.py +++ b/volatility/framework/layers/scanners/wumanber.py @@ -5,7 +5,8 @@ class WuManber(object): """Algorithm for multi-string matching""" def __init__(self, block_size: int = 3) -> None: - self.minimum_pattern_length = None # type: typing.Optional[int] + # Set a suitably large minimum + self.minimum_pattern_length = 1000000000000 self._block_size = block_size self._maximum_hash = self._hash_function(b"\xff\xff\xff") + 1 # This depends on the hash function used diff --git a/volatility/framework/objects/__init__.py b/volatility/framework/objects/__init__.py index e1ab90521..894e534e6 100644 --- a/volatility/framework/objects/__init__.py +++ b/volatility/framework/objects/__init__.py @@ -248,8 +248,7 @@ class Pointer(Integer): Layer_name is identifies the appropriate layer within the context that the pointer points to. If layer_name is None, it defaults to the same layer that the pointer is currently instantiated in. """ - if layer_name is None: - layer_name = self.vol.layer_name + layer_name = layer_name or self.vol.layer_name mask = self._context.memory[layer_name].address_mask offset = self & mask return self.vol.subtype(context = self._context, @@ -260,8 +259,7 @@ class Pointer(Integer): def is_readable(self, layer_name: typing.Optional[str] = None) -> bool: """Determines whether the address of this pointer can be read from memory""" - if layer_name is None: - layer_name = self.vol.layer_name + layer_name = layer_name or self.vol.layer_name return self._context.memory[layer_name].is_valid(self) def __getattr__(self, attr: str) -> typing.Any: @@ -352,8 +350,8 @@ class Enumeration(interfaces.objects.ObjectInterface, int): context: interfaces.context.ContextInterface, type_name: str, object_info: interfaces.objects.ObjectInformation, - base_type: interfaces.objects.Template = None, - choices: typing.Dict[str, int] = None, + base_type: interfaces.objects.Template, + choices: typing.Dict[str, int], **kwargs) -> typing.Type: cls._check_class(base_type.vol.object_class, Integer) value = base_type(context = context, @@ -364,8 +362,8 @@ class Enumeration(interfaces.objects.ObjectInterface, int): context: interfaces.context.ContextInterface, type_name: str, object_info: interfaces.objects.ObjectInformation, - base_type: typing.Optional[Integer] = None, - choices: typing.Optional[typing.Dict[str, int]] = None) -> None: + base_type: Integer, + choices: typing.Dict[str, int]) -> None: super().__init__(context, type_name, object_info) self._inverse_choices = {} # type: typing.Dict[int, str] diff --git a/volatility/framework/objects/templates.py b/volatility/framework/objects/templates.py index ba17a8180..816a9c776 100644 --- a/volatility/framework/objects/templates.py +++ b/volatility/framework/objects/templates.py @@ -18,8 +18,8 @@ class ObjectTemplate(interfaces.objects.Template, validity.ValidityRoutines): """ def __init__(self, - object_class: typing.Optional[typing.Type[interfaces.objects.ObjectInterface]] = None, - type_name: str = None, + object_class: typing.Type[interfaces.objects.ObjectInterface], + type_name: str, **arguments) -> None: super().__init__(type_name = type_name, **arguments) self._check_class(object_class, interfaces.objects.ObjectInterface) diff --git a/volatility/framework/renderers/__init__.py b/volatility/framework/renderers/__init__.py index 833f69cb6..96483eea4 100644 --- a/volatility/framework/renderers/__init__.py +++ b/volatility/framework/renderers/__init__.py @@ -177,7 +177,8 @@ class TreeGrid(interfaces.renderers.TreeGrid): parent = prev_nodes[parent_index - 1] if parent_index > 0 else None treenode = self._append(parent, item) prev_nodes = prev_nodes[0: parent_index] + [treenode] - accumulator = func(treenode, accumulator) + if func is not None: + accumulator = func(treenode, accumulator) self._row_count += 1 self._populated = True @@ -254,8 +255,8 @@ class TreeGrid(interfaces.renderers.TreeGrid): def visit(self, node: typing.Optional[interfaces.renderers.TreeNode], function: typing.Callable[[interfaces.renderers.TreeNode, _T], _T], - initial_accumulator: _T = None, - sort_key: interfaces.renderers.ColumnSortKey = None): + initial_accumulator: _T, + sort_key: typing.Optional[interfaces.renderers.ColumnSortKey] = None): """Visits all the nodes in a tree, calling function on each one. function should have the signature function(node, accumulator) and return new_accumulator @@ -290,7 +291,7 @@ class TreeGrid(interfaces.renderers.TreeGrid): list_of_children: typing.List['TreeNode'], function: typing.Callable, accumulator: _T, - sort_key: interfaces.renderers.ColumnSortKey = None) -> _T: + sort_key: typing.Optional[interfaces.renderers.ColumnSortKey] = None) -> _T: """Visits all the nodes in a tree, calling function on each one""" if list_of_children is not None: for n, children in list_of_children: diff --git a/volatility/framework/symbols/__init__.py b/volatility/framework/symbols/__init__.py index d6af6c3d0..1cd4c87b9 100644 --- a/volatility/framework/symbols/__init__.py +++ b/volatility/framework/symbols/__init__.py @@ -106,7 +106,7 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface, validity.ValidityRout indicate this failure in resolution. """ - def __init__(self, type_name: str = None, **kwargs) -> None: + def __init__(self, type_name: str, **kwargs) -> None: vollog.debug("Unresolved reference: {}".format(type_name)) super().__init__(type_name = type_name, **kwargs) diff --git a/volatility/framework/symbols/intermed.py b/volatility/framework/symbols/intermed.py index e6bf571c0..fa108cc50 100644 --- a/volatility/framework/symbols/intermed.py +++ b/volatility/framework/symbols/intermed.py @@ -196,6 +196,8 @@ class ISFormatTable(interfaces.symbols.SymbolTableInterface, metaclass = ABCMeta self._json_object = json_object self._validate_json() nt = native_types or self._get_natives() + if nt is None: + raise ValueError("Native table not provided") nt.name = name + "_natives" super().__init__(context, config_path, name, nt, table_mapping = table_mapping) self._overrides = {} # type: typing.Dict[str, typing.Type[interfaces.objects.ObjectInterface]] diff --git a/volatility/framework/symbols/linux/extensions/__init__.py b/volatility/framework/symbols/linux/extensions/__init__.py index 25fc01caf..03a0f47b1 100644 --- a/volatility/framework/symbols/linux/extensions/__init__.py +++ b/volatility/framework/symbols/linux/extensions/__init__.py @@ -119,9 +119,7 @@ class list_head(objects.Struct, collections.abc.Iterable): 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: - layer = self.vol.layer_name + layer = layer or self.vol.layer_name relative_offset = self._context.symbol_space.get_type(symbol_type).relative_child_offset(member) diff --git a/volatility/framework/symbols/native.py b/volatility/framework/symbols/native.py index c057d6920..8e92eb477 100644 --- a/volatility/framework/symbols/native.py +++ b/volatility/framework/symbols/native.py @@ -45,7 +45,7 @@ class NativeTable(interfaces.symbols.NativeTableInterface): prefix = table_name + constants.BANG additional = {} # type: typing.Dict[str, typing.Any] - obj = None # type: typing.Type[interfaces.objects.ObjectInterface] + obj = None # type: typing.Optional[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/extensions/__init__.py b/volatility/framework/symbols/windows/extensions/__init__.py index 3ab6a2704..6b85ad19f 100644 --- a/volatility/framework/symbols/windows/extensions/__init__.py +++ b/volatility/framework/symbols/windows/extensions/__init__.py @@ -509,8 +509,7 @@ class _LIST_ENTRY(objects.Struct, collections.abc.Iterable): layer: typing.Optional[str] = None) -> typing.Iterator[interfaces.objects.ObjectInterface]: """Returns an iterator of the entries in the list""" - if layer is None: - layer = self.vol.layer_name + layer = layer or self.vol.layer_name relative_offset = self._context.symbol_space.get_type(symbol_type).relative_child_offset(member)