diff --git a/volatility3/cli/volargparse.py b/volatility3/cli/volargparse.py index 3935d912f..8f239867f 100644 --- a/volatility3/cli/volargparse.py +++ b/volatility3/cli/volargparse.py @@ -5,7 +5,8 @@ import argparse import gettext import re -from typing import List +from typing import List, Optional, Sequence, Any, Union + # This effectively overrides/monkeypatches the core argparse module to provide more helpful output around choices # We shouldn't really steal a private member from argparse, but otherwise we're just duplicating code @@ -21,15 +22,22 @@ 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 + self.choices = None # type: ignore def __call__(self, - parser: 'HelpfulArgParser', + parser: argparse.ArgumentParser, namespace: argparse.Namespace, - values: List[str], - option_string: None = None) -> None: - parser_name = values[0] - arg_strings = values[1:] + values: Union[str, Sequence[Any], None], + option_string: Optional[str] = None) -> None: + + parser_name = '' + arg_strings = [] # type: List[str] + if values is not None: + for value in values: + if not parser_name: + parser_name = value + else: + arg_strings += [value] # set the parser name if requested if self.dest != argparse.SUPPRESS: diff --git a/volatility3/cli/volshell/generic.py b/volatility3/cli/volshell/generic.py index 238b12e04..78ab9ba2c 100644 --- a/volatility3/cli/volshell/generic.py +++ b/volatility3/cli/volshell/generic.py @@ -8,7 +8,7 @@ import random import string import struct import sys -from typing import Any, Dict, List, Optional, Tuple, Union, Type +from typing import Any, Dict, List, Optional, Tuple, Union, Type, Iterable from urllib import request, parse from volatility3.cli import text_renderer @@ -57,7 +57,7 @@ class Volshell(interfaces.plugins.PluginInterface): Return a TreeGrid but this is always empty since the point of this plugin is to run interactively """ - self._current_layer = self.config['primary'] + self.__current_layer = self.config['primary'] # Try to enable tab completion try: @@ -174,13 +174,13 @@ class Volshell(interfaces.plugins.PluginInterface): @property def current_layer(self): - return self._current_layer + return self.__current_layer def change_layer(self, layer_name = None): """Changes the current default layer""" if not layer_name: layer_name = self.config['primary'] - self._current_layer = layer_name + self.__current_layer = layer_name sys.ps1 = "({}) >>> ".format(self.current_layer) def display_bytes(self, offset, count = 128, layer_name = None): @@ -336,7 +336,7 @@ class Volshell(interfaces.plugins.PluginInterface): len_offset = len(hex(symbol.address)) print(" " * (longest_offset - len_offset), hex(symbol.address), " ", symbol.name) - def run_script(self, location: str = None): + def run_script(self, location: str): """Runs a python script within the context of volshell""" if not parse.urlparse(location).scheme: location = "file:" + request.pathname2url(location) @@ -346,7 +346,7 @@ class Volshell(interfaces.plugins.PluginInterface): self.__console.runsource(fp.read(), symbol = 'exec') print("\nCode complete") - def load_file(self, location: str = None): + def load_file(self, location: str): """Loads a file into a Filelayer and returns the name of the layer""" layer_name = self.context.layers.free_layer_name() if not parse.urlparse(location).scheme: @@ -399,10 +399,10 @@ class NullFileHandler(io.BytesIO, interfaces.plugins.FileHandlerInterface): interfaces.plugins.FileHandlerInterface.__init__(self, preferred_name) super().__init__() - def writelines(self, lines): + def writelines(self, lines: Iterable[bytes]): """Dummy method""" pass - def write(self, data): + def write(self, b: bytes): """Dummy method""" - return len(data) + return len(b) diff --git a/volatility3/framework/__init__.py b/volatility3/framework/__init__.py index e8c67588c..b0a2f2b5c 100644 --- a/volatility3/framework/__init__.py +++ b/volatility3/framework/__init__.py @@ -53,22 +53,22 @@ def require_interface_version(*args) -> None: ".".join([str(x) for x in interface_version()[0:1]]), ".".join([str(x) for x in args[0:2]]))) -class noninheritable(object): +class NonInheritable(object): def __init__(self, value: Any, cls: Type) -> None: self.default_value = value self.cls = cls - def __get__(self, obj: Any, type: Type = None) -> Any: + def __get__(self, obj: Any, get_type: Type = None) -> Any: if type == self.cls: if hasattr(self.default_value, '__get__'): - return self.default_value.__get__(obj, type) + return self.default_value.__get__(obj, get_type) return self.default_value raise AttributeError def hide_from_subclasses(cls: Type) -> Type: - cls.hidden = noninheritable(True, cls) + cls.hidden = NonInheritable(True, cls) return cls diff --git a/volatility3/framework/automagic/__init__.py b/volatility3/framework/automagic/__init__.py index be1badb6b..844d9a273 100644 --- a/volatility3/framework/automagic/__init__.py +++ b/volatility3/framework/automagic/__init__.py @@ -48,7 +48,7 @@ def available(context: interfaces.context.ContextInterface) -> List[interfaces.a def choose_automagic( - automagics: List[interfaces.automagic.AutomagicInterface], + automagics: List[Type[interfaces.automagic.AutomagicInterface]], plugin: Type[interfaces.plugins.PluginInterface]) -> List[Type[interfaces.automagic.AutomagicInterface]]: """Chooses which automagics to run, maintaining the order they were handed in.""" diff --git a/volatility3/framework/configuration/requirements.py b/volatility3/framework/configuration/requirements.py index 2a8881ace..20a7af0f6 100644 --- a/volatility3/framework/configuration/requirements.py +++ b/volatility3/framework/configuration/requirements.py @@ -208,6 +208,9 @@ class LayerListRequirement(ComplexListRequirement): num_layers_path = interfaces.configuration.path_join(new_config_path, "number_of_elements") number_of_layers = context.config[num_layers_path] + if not isinstance(number_of_layers, int): + raise TypeError("Number of layers must be an integer") + # Build all the layers that can be built for i in range(number_of_layers): layer_req = self.requirements.get(self.name + str(i), None) @@ -363,6 +366,8 @@ class SymbolTableRequirement(interfaces.configuration.ConstructableRequirementIn raise TypeError("Class requirement is not of type ClassRequirement: {}".format( repr(self.requirements["class"]))) cls = self.requirements["class"].cls + if cls is None: + return None node_config = context.config.branch(config_path) for req in cls.get_requirements(): if req.name in node_config.data and req.name != "class": @@ -392,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 + self._component = component # type: Type[interfaces.configuration.VersionableInterface] if version is None: raise TypeError("Version cannot be None") self._version = version diff --git a/volatility3/framework/exceptions.py b/volatility3/framework/exceptions.py index 117933bbd..7f381b166 100644 --- a/volatility3/framework/exceptions.py +++ b/volatility3/framework/exceptions.py @@ -96,6 +96,6 @@ class UnsatisfiedException(VolatilityException): class MissingModuleException(VolatilityException): - def __init__(self, module: str, *args, **kwargs) -> None: - super().__init__(*args, **kwargs) + def __init__(self, module: str, *args) -> None: + super().__init__(*args) self.module = module diff --git a/volatility3/framework/interfaces/configuration.py b/volatility3/framework/interfaces/configuration.py index 396926b68..8b02d0d93 100644 --- a/volatility3/framework/interfaces/configuration.py +++ b/volatility3/framework/interfaces/configuration.py @@ -23,7 +23,7 @@ import random import string import sys from abc import ABCMeta, abstractmethod -from typing import Any, ClassVar, Dict, Generator, Iterator, List, Optional, Type, Union, Tuple +from typing import Any, ClassVar, Dict, Generator, Iterator, List, Optional, Type, Union, Tuple, Set from volatility3 import classproperty from volatility3.framework import constants, interfaces @@ -186,7 +186,8 @@ class HierarchicalDict(collections.abc.Mapping): element_value = self._sanitize_value(element) if isinstance(element_value, list): raise TypeError("Configuration list types cannot contain list types") - new_list.append(element_value) + if element_value is not None: + new_list.append(element_value) return new_list elif value is None: return None @@ -483,7 +484,7 @@ class ClassRequirement(RequirementInterface): return super().__eq__(other) @property - def cls(self) -> Type: + def cls(self) -> Optional[Type]: """Contains the actual chosen class based on the configuration value's class name.""" return self._cls @@ -528,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() + self._current_class_requirements = set() # type: Set[Any] def __eq__(self, other): # We can just use super because it checks all member of `__dict__` @@ -581,6 +582,9 @@ class ConstructableRequirementInterface(RequirementInterface): return None cls = self.requirements["class"].cls + if cls is None: + return None + # These classes all have a name property # We could subclass this out as a NameableInterface, but it seems a little excessive # FIXME: We can't test this, because importing the other interfaces causes all kinds of import loops diff --git a/volatility3/framework/interfaces/layers.py b/volatility3/framework/interfaces/layers.py index 29b2922a7..cce80f0a1 100644 --- a/volatility3/framework/interfaces/layers.py +++ b/volatility3/framework/interfaces/layers.py @@ -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'} + _direct_metadata = {'architecture': 'Unknown', 'os': 'Unknown'} # type: Mapping def __init__(self, context: 'interfaces.context.ContextInterface', @@ -473,7 +473,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass = ABCMeta): assumed to have no holes """ for (section_start, section_length) in sections: - output = [] + output = [] # type: List[Tuple[str, int, int]] # Hold the offsets of each chunk (including how much has been filled) chunk_start = chunk_position = 0 diff --git a/volatility3/framework/plugins/mac/lsmod.py b/volatility3/framework/plugins/mac/lsmod.py index fe5d25adf..89a3c08af 100644 --- a/volatility3/framework/plugins/mac/lsmod.py +++ b/volatility3/framework/plugins/mac/lsmod.py @@ -3,7 +3,9 @@ # """A module containing a collection of plugins that produce data typically found in Mac's lsmod command.""" -from volatility3.framework import renderers, interfaces, contexts +from typing import Set + +from volatility3.framework import renderers, interfaces, contexts, exceptions from volatility3.framework.configuration import requirements from volatility3.framework.interfaces import plugins from volatility3.framework.objects import utility @@ -55,11 +57,11 @@ class Lsmod(plugins.PluginInterface): except exceptions.InvalidAddressException: return [] - seen = set() + seen = set() # type: Set while kmod != 0 and \ - kmod not in seen and \ - len(seen) < 1024: + kmod not in seen and \ + len(seen) < 1024: kmod_obj = kmod.dereference() diff --git a/volatility3/framework/plugins/windows/cachedump.py b/volatility3/framework/plugins/windows/cachedump.py index b436db753..dc8246ef8 100644 --- a/volatility3/framework/plugins/windows/cachedump.py +++ b/volatility3/framework/plugins/windows/cachedump.py @@ -3,12 +3,14 @@ # from struct import unpack +from typing import Tuple from Crypto.Cipher import ARC4, AES from Crypto.Hash import HMAC -from volatility3.framework import interfaces, renderers +from volatility3.framework import interfaces, renderers, exceptions from volatility3.framework.configuration import requirements +from volatility3.framework.layers import registry from volatility3.framework.symbols.windows import versions from volatility3.plugins.windows import hashdump, lsadump from volatility3.plugins.windows.registry import hivelist @@ -31,10 +33,12 @@ class Cachedump(interfaces.plugins.PluginInterface): requirements.PluginRequirement(name = 'lsadump', plugin = lsadump.Lsadump, version = (1, 0, 0)) ] - def get_nlkm(self, sechive, lsakey, is_vista_or_later): + @staticmethod + def get_nlkm(sechive: registry.RegistryHive, lsakey: bytes, is_vista_or_later: bool): return lsadump.Lsadump.get_secret_by_name(sechive, 'NL$KM', lsakey, is_vista_or_later) - def decrypt_hash(self, edata, nlkm, ch, xp): + @staticmethod + def decrypt_hash(edata: bytes, nlkm: bytes, ch, xp: bool): if xp: hmac_md5 = HMAC.new(nlkm, ch) rc4key = hmac_md5.digest() @@ -51,16 +55,19 @@ class Cachedump(interfaces.plugins.PluginInterface): data += aes.decrypt(buf) return data - def parse_cache_entry(self, cache_data): + @staticmethod + def parse_cache_entry(cache_data: bytes) -> Tuple[int, int, int, bytes, bytes]: (uname_len, domain_len) = unpack(" Tuple[str, str, str, bytes]: """Get the data from the cache and separate it into the username, domain name, and hash data""" uname_offset = 72 pad = 2 * ((uname_len / 2) % 2) @@ -68,12 +75,9 @@ class Cachedump(interfaces.plugins.PluginInterface): pad = 2 * ((domain_len / 2) % 2) domain_name_offset = int(domain_offset + domain_len + pad) hashh = dec_data[:0x10] - username = dec_data[uname_offset:uname_offset + uname_len] - username = username.decode('utf-16-le', 'replace') - domain = dec_data[domain_offset:domain_offset + domain_len] - domain = domain.decode('utf-16-le', 'replace') - domain_name = dec_data[domain_name_offset:domain_name_offset + domain_name_len] - domain_name = domain_name.decode('utf-16-le', 'replace') + username = dec_data[uname_offset:uname_offset + uname_len].decode('utf-16-le', 'replace') + domain = dec_data[domain_offset:domain_offset + domain_len].decode('utf-16-le', 'replace') + domain_name = dec_data[domain_name_offset:domain_name_offset + domain_name_len].decode('utf-16-le', 'replace') return (username, domain, domain_name, hashh) @@ -116,6 +120,8 @@ class Cachedump(interfaces.plugins.PluginInterface): def run(self): offset = self.config.get('offset', None) + syshive = sechive = None + for hive in hivelist.HiveList.list_hives(self.context, self.config_path, self.config['primary'], @@ -127,5 +133,10 @@ class Cachedump(interfaces.plugins.PluginInterface): if hive.get_name().split('\\')[-1].upper() == 'SECURITY': sechive = hive + if syshive is None: + raise exceptions.VolatilityException('Unable to locate SYSTEM hive') + if sechive is None: + raise exceptions.VolatilityException('Unable to locate SECURITY hive') + return renderers.TreeGrid([("Username", str), ("Domain", str), ("Domain name", str), ('Hashh', bytes)], self._generator(syshive, sechive)) diff --git a/volatility3/framework/plugins/windows/dumpfiles.py b/volatility3/framework/plugins/windows/dumpfiles.py index 9fc9daa54..f1e54d0f9 100755 --- a/volatility3/framework/plugins/windows/dumpfiles.py +++ b/volatility3/framework/plugins/windows/dumpfiles.py @@ -9,7 +9,8 @@ from volatility3.plugins.windows import handles from volatility3.plugins.windows import pslist from volatility3.framework.configuration import requirements from volatility3.framework.renderers import format_hints -from typing import List, Tuple, Type, Optional +from typing import List, Tuple, Type, Optional, Generator + vollog = logging.getLogger(__name__) FILE_DEVICE_DISK = 0x7 @@ -91,13 +92,13 @@ class DumpFiles(interfaces.plugins.PluginInterface): @classmethod def process_file_object(cls, context: interfaces.context.ContextInterface, primary_layer_name: str, open_method: Type[interfaces.plugins.FileHandlerInterface], - file_obj: interfaces.objects.ObjectInterface) -> Tuple: + file_obj: interfaces.objects.ObjectInterface) -> Generator[Tuple, None, None]: """Given a FILE_OBJECT, dump data to separate files for each of the three file caches. :param context: the context to operate upon :param primary_layer_name: primary/virtual layer to operate on :param open_method: class for constructing output files - :param file_object: the FILE_OBJECT + :param file_obj: the FILE_OBJECT """ # Filtering by these types of devices prevents us from processing other types of devices that diff --git a/volatility3/framework/plugins/windows/hashdump.py b/volatility3/framework/plugins/windows/hashdump.py index ed2f1988a..97fba7608 100644 --- a/volatility3/framework/plugins/windows/hashdump.py +++ b/volatility3/framework/plugins/windows/hashdump.py @@ -133,7 +133,8 @@ class Hashdump(interfaces.plugins.PluginInterface): return None @classmethod - def decrypt_single_salted_hash(cls, rid, hbootkey: bytes, enc_hash: bytes, lmntstr, salt: bytes) -> Optional[bytes]: + def decrypt_single_salted_hash(cls, rid, hbootkey: bytes, enc_hash: bytes, _lmntstr, + salt: bytes) -> Optional[bytes]: (des_k1, des_k2) = cls.sid_to_key(rid) des1 = DES.new(des_k1, DES.MODE_ECB) des2 = DES.new(des_k2, DES.MODE_ECB) @@ -143,7 +144,7 @@ class Hashdump(interfaces.plugins.PluginInterface): @classmethod def get_user_hashes(cls, user: registry.CM_KEY_NODE, samhive: registry.RegistryHive, - hbootkey: bytes) -> Tuple[bytes, bytes]: + hbootkey: bytes) -> Optional[Tuple[bytes, bytes]]: ## Will sometimes find extra user with rid = NAMES, returns empty strings right now try: rid = int(str(user.get_name()), 16) @@ -199,22 +200,16 @@ class Hashdump(interfaces.plugins.PluginInterface): @classmethod def sidbytes_to_key(cls, s: bytes) -> bytes: """Builds final DES key from the strings generated in sid_to_key""" - key = [] - key.append(s[0] >> 1) - key.append(((s[0] & 0x01) << 6) | (s[1] >> 2)) - key.append(((s[1] & 0x03) << 5) | (s[2] >> 3)) - key.append(((s[2] & 0x07) << 4) | (s[3] >> 4)) - key.append(((s[3] & 0x0F) << 3) | (s[4] >> 5)) - key.append(((s[4] & 0x1F) << 2) | (s[5] >> 6)) - key.append(((s[5] & 0x3F) << 1) | (s[6] >> 7)) - key.append(s[6] & 0x7F) + key = [s[0] >> 1, ((s[0] & 0x01) << 6) | (s[1] >> 2), ((s[1] & 0x03) << 5) | (s[2] >> 3), + ((s[2] & 0x07) << 4) | (s[3] >> 4), ((s[3] & 0x0F) << 3) | (s[4] >> 5), + ((s[4] & 0x1F) << 2) | (s[5] >> 6), ((s[5] & 0x3F) << 1) | (s[6] >> 7), s[6] & 0x7F] for i in range(8): key[i] = (key[i] << 1) key[i] = cls.odd_parity[key[i]] return bytes(key) @classmethod - def decrypt_single_hash(cls, rid, hbootkey, enc_hash: bytes, lmntstr): + def decrypt_single_hash(cls, rid: int, hbootkey: bytes, enc_hash: bytes, lmntstr: bytes): (des_k1, des_k2) = cls.sid_to_key(rid) des1 = DES.new(des_k1, DES.MODE_ECB) des2 = DES.new(des_k2, DES.MODE_ECB) @@ -225,24 +220,23 @@ class Hashdump(interfaces.plugins.PluginInterface): rc4 = ARC4.new(rc4_key) obfkey = rc4.encrypt(enc_hash) - hash = des1.decrypt(obfkey[:8]) + des2.decrypt(obfkey[8:]) - return hash + return des1.decrypt(obfkey[:8]) + des2.decrypt(obfkey[8:]) @classmethod - def get_user_name(cls, user: interfaces.objects.ObjectInterface, samhive: registry.RegistryHive) -> Optional[bytes]: - V = None + def get_user_name(cls, user: registry.CM_KEY_NODE, samhive: registry.RegistryHive) -> Optional[bytes]: + value = None for v in user.get_values(): if v.get_name() == 'V': - V = samhive.read(v.Data + 4, v.DataLength) - if not V: + value = samhive.read(v.Data + 4, v.DataLength) + if not value: return None - name_offset = unpack(" len(V): + name_offset = unpack(" len(value): return None - username = V[name_offset:name_offset + name_length] + username = value[name_offset:name_offset + name_length] return username # replaces the dump_hashes method in vol2 diff --git a/volatility3/framework/plugins/windows/lsadump.py b/volatility3/framework/plugins/windows/lsadump.py index b9f8df9fe..a9ee1737b 100644 --- a/volatility3/framework/plugins/windows/lsadump.py +++ b/volatility3/framework/plugins/windows/lsadump.py @@ -3,12 +3,14 @@ # import logging from struct import unpack +from typing import Optional from Crypto.Cipher import ARC4, DES, AES from Crypto.Hash import MD5, SHA256 from volatility3.framework import interfaces, renderers from volatility3.framework.configuration import requirements +from volatility3.framework.layers import registry from volatility3.framework.symbols.windows import versions from volatility3.plugins.windows import hashdump from volatility3.plugins.windows.registry import hivelist @@ -33,7 +35,7 @@ class Lsadump(interfaces.plugins.PluginInterface): ] @classmethod - def decrypt_aes(cls, secret, key): + def decrypt_aes(cls, secret: bytes, key: bytes) -> bytes: """ Based on code from http://lab.mediaservice.net/code/cachedump.rb """ @@ -54,7 +56,7 @@ class Lsadump(interfaces.plugins.PluginInterface): return data @classmethod - def get_lsa_key(cls, sechive, bootkey, vista_or_later): + def get_lsa_key(cls, sechive: registry.RegistryHive, bootkey: bytes, vista_or_later: bool) -> Optional[bytes]: if not bootkey: return None @@ -91,7 +93,7 @@ class Lsadump(interfaces.plugins.PluginInterface): return lsa_key @classmethod - def get_secret_by_name(cls, sechive, name, lsakey, is_vista_or_later): + def get_secret_by_name(cls, sechive: registry.RegistryHive, name: str, lsakey: bytes, is_vista_or_later: bool): try: enc_secret_key = sechive.get_key("Policy\\Secrets\\" + name + "\\CurrVal") except KeyError: @@ -112,7 +114,7 @@ class Lsadump(interfaces.plugins.PluginInterface): return secret @classmethod - def decrypt_secret(cls, secret, key): + def decrypt_secret(cls, secret: bytes, key: bytes): """Python implementation of SystemFunction005. Decrypts a block of data with DES using given key. @@ -135,7 +137,7 @@ class Lsadump(interfaces.plugins.PluginInterface): return decrypted_data[8:8 + dec_data_len] - def _generator(self, syshive, sechive): + def _generator(self, syshive: registry.RegistryHive, sechive: registry.RegistryHive): vista_or_later = versions.is_vista_or_later(context = self.context, symbol_table = self.config['nt_symbols']) @@ -174,6 +176,7 @@ class Lsadump(interfaces.plugins.PluginInterface): def run(self): offset = self.config.get('offset', None) + syshive = sechive = None for hive in hivelist.HiveList.list_hives(self.context, self.config_path, diff --git a/volatility3/framework/plugins/windows/netscan.py b/volatility3/framework/plugins/windows/netscan.py index 77bf67841..7c0bdef0b 100644 --- a/volatility3/framework/plugins/windows/netscan.py +++ b/volatility3/framework/plugins/windows/netscan.py @@ -4,7 +4,7 @@ import datetime import logging -from typing import Iterable, List, Optional +from typing import Iterable, List, Optional, Tuple, Type from volatility3.framework import constants, exceptions, interfaces, renderers, symbols from volatility3.framework.configuration import requirements @@ -82,7 +82,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): @classmethod def determine_tcpip_version(cls, context: interfaces.context.ContextInterface, layer_name: str, - nt_symbol_table: str) -> str: + nt_symbol_table: str) -> Tuple[str, Type]: """Tries to determine which symbol filename to use for the image's tcpip driver. The logic is partially taken from the info plugin. Args: diff --git a/volatility3/framework/plugins/windows/netstat.py b/volatility3/framework/plugins/windows/netstat.py index b6332e6e3..1d37258ab 100644 --- a/volatility3/framework/plugins/windows/netstat.py +++ b/volatility3/framework/plugins/windows/netstat.py @@ -4,7 +4,7 @@ import logging import datetime -from typing import Iterable, Optional +from typing import Iterable, Optional, Generator, Tuple from volatility3.framework import constants, exceptions, interfaces, renderers, symbols from volatility3.framework.configuration import requirements @@ -125,7 +125,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): ptr_offset = context.symbol_space.get_type(obj_name).relative_child_offset("Next") else: # invalid argument. - yield + return vollog.debug("Current Port: {}".format(port)) # the given port serves as a shifted index into the port pool lists @@ -144,7 +144,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): assignment = inpa.InPaBigPoolBase.Assignments[truncated_port] if not assignment: - yield + return # the value within assignment.Entry is a) masked and b) points inside of the network object # first decode the pointer @@ -165,7 +165,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): @classmethod def get_tcpip_module(cls, context: interfaces.context.ContextInterface, layer_name: str, - nt_symbols: str) -> interfaces.objects.ObjectInterface: + nt_symbols: str) -> Optional[interfaces.objects.ObjectInterface]: """Uses `windows.modules` to find tcpip.sys in memory. Args: @@ -180,10 +180,11 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): if mod.BaseDllName.get_string() == "tcpip.sys": vollog.debug("Found tcpip.sys image base @ 0x{:x}".format(mod.DllBase)) return mod + return None @classmethod def parse_hashtable(cls, context: interfaces.context.ContextInterface, layer_name: str, ht_offset: int, - ht_length: int, alignment: int, net_symbol_table: str) -> list: + ht_length: int, alignment: int, net_symbol_table: str) -> Generator[interfaces.objects.ObjectInterface, None, None]: """Parses a hashtable quick and dirty. Args: @@ -217,10 +218,9 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): Args: context: The context to retrieve required elements (layers, symbol tables) from layer_name: The name of the layer on which to operate - nt_symbols: The name of the table containing the kernel symbols net_symbol_table: The name of the table containing the tcpip types - tcpip_module: The created vol Windows module object of the given memory image tcpip_symbol_table: The name of the table containing the tcpip driver symbols + tcpip_module_offset: The offset of the tcpip module Returns: The list of TCP endpoint objects from the `layer_name` layer's `PartitionTable` @@ -289,7 +289,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): if not guids: raise exceptions.VolatilityException("Did not find GUID of tcpip.pdb in tcpip.sys module @ 0x{:x}!".format( - tcpip_module.DllBase)) + tcpip_module_offset)) guid = guids[0] @@ -305,7 +305,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface): @classmethod def find_port_pools(cls, context: interfaces.context.ContextInterface, layer_name: str, net_symbol_table: str, - tcpip_symbol_table: str, tcpip_module_offset: int) -> (int, int): + tcpip_symbol_table: str, tcpip_module_offset: int) -> Tuple[int, int]: """Finds the given image's port pools. Older Windows versions (presumably < Win10 build 14251) use driver symbols called `UdpPortPool` and `TcpPortPool` which point towards the pools. Newer Windows versions use `UdpCompartmentSet` and `TcpCompartmentSet`, which we first have to translate into diff --git a/volatility3/framework/plugins/windows/vadinfo.py b/volatility3/framework/plugins/windows/vadinfo.py index d50f596aa..aa6567ace 100644 --- a/volatility3/framework/plugins/windows/vadinfo.py +++ b/volatility3/framework/plugins/windows/vadinfo.py @@ -132,11 +132,11 @@ class VadInfo(interfaces.plugins.PluginInterface): vad_end = vad.get_end() except AttributeError: vollog.debug("Unable to find the starting/ending VPN member") - return + return None if maxsize > 0 and (vad_end - vad_start) > maxsize: vollog.debug("Skip VAD dump {0:#x}-{1:#x} due to maxsize limit".format(vad_start, vad_end)) - return + return None proc_id = "Unknown" try: @@ -163,7 +163,7 @@ class VadInfo(interfaces.plugins.PluginInterface): except Exception as excp: vollog.debug("Unable to dump VAD {}: {}".format(file_name, excp)) - return + return None return file_handle diff --git a/volatility3/framework/renderers/__init__.py b/volatility3/framework/renderers/__init__.py index c7dc126b2..a3fa7cef3 100644 --- a/volatility3/framework/renderers/__init__.py +++ b/volatility3/framework/renderers/__init__.py @@ -9,7 +9,7 @@ or file or graphical output import collections import datetime import logging -from typing import Any, Callable, Iterable, List, Optional, Sequence, Tuple, TypeVar, Union +from typing import Any, Callable, Iterable, List, Optional, Tuple, TypeVar, Union from volatility3.framework import interfaces from volatility3.framework.interfaces import renderers @@ -48,7 +48,7 @@ class NotAvailableValue(interfaces.renderers.BaseAbsentValue): class TreeNode(interfaces.renderers.TreeNode): """Class representing a particular node in a tree grid.""" - def __init__(self, path: str, treegrid: 'TreeGrid', parent: Optional['TreeNode'], + def __init__(self, path: str, treegrid: 'TreeGrid', parent: Optional[interfaces.renderers.TreeNode], values: List[interfaces.renderers.BaseTypes]) -> None: if not isinstance(treegrid, TreeGrid): raise TypeError("Treegrid must be an instance of TreeGrid") @@ -70,7 +70,7 @@ class TreeNode(interfaces.renderers.TreeNode): def _validate_values(self, values: List[interfaces.renderers.BaseTypes]) -> None: """A function for raising exceptions if a given set of values is invalid according to the column properties.""" - if not (isinstance(values, collections.abc.Sequence) and len(values) == len(self._treegrid.columns)): + if not (isinstance(values, collections.Sequence) and len(values) == len(self._treegrid.columns)): raise TypeError( "Values must be a list of objects made up of simple types and number the same as the columns") for index in range(len(self._treegrid.columns)): @@ -85,10 +85,10 @@ class TreeNode(interfaces.renderers.TreeNode): # tznaive = val.tzinfo is None or val.tzinfo.utcoffset(val) is None @property - def values(self) -> Sequence[interfaces.renderers.BaseTypes]: + def values(self) -> List[interfaces.renderers.BaseTypes]: """Returns the list of values from the particular node, based on column index.""" - return self._values + return list(self._values) @property def path(self) -> str: @@ -101,7 +101,7 @@ class TreeNode(interfaces.renderers.TreeNode): return self._path @property - def parent(self) -> Optional['TreeNode']: + def parent(self) -> Optional[interfaces.renderers.TreeNode]: """Returns the parent node of this node or None.""" return self._parent diff --git a/volatility3/framework/renderers/format_hints.py b/volatility3/framework/renderers/format_hints.py index 029049df3..b169f44c9 100644 --- a/volatility3/framework/renderers/format_hints.py +++ b/volatility3/framework/renderers/format_hints.py @@ -8,7 +8,7 @@ These hints allow a plugin to indicate how they would like data from a particula Text renderers should attempt to honour all hints provided in this module where possible """ -from typing import Type +from typing import Type, Union class Bin(int): @@ -30,13 +30,16 @@ class MultiTypeData(bytes): """The contents are supposed to be a string, but may contain binary data.""" def __new__(cls: Type['MultiTypeData'], - original: int, + original: Union[int, bytes], encoding: str = 'utf-16-le', split_nulls: bool = False, show_hex: bool = False) -> 'MultiTypeData': + if isinstance(original, int): - original = str(original).encode(encoding) - return super().__new__(cls, original) + data = str(original).encode(encoding) + else: + data = original + return super().__new__(cls, data) def __init__(self, original: bytes, diff --git a/volatility3/framework/symbols/windows/pdbutil.py b/volatility3/framework/symbols/windows/pdbutil.py index 0b4f760c9..7e39cbc1a 100644 --- a/volatility3/framework/symbols/windows/pdbutil.py +++ b/volatility3/framework/symbols/windows/pdbutil.py @@ -68,7 +68,7 @@ class PDBUtility: filter_string = os.path.join(pdb_name.strip('\x00'), guid.upper() + "-" + str(age)) - isf_path = False + isf_path = None # Take the first result of search for the intermediate file for value in intermed.IntermediateSymbolTable.file_symbol_url("windows", filter_string): isf_path = value