WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-08-29 03:09:48 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #468 from volatilityfoundation/feature/improved-typing

Browse Source 
Core: Improve typing across codebase
This commit is contained in:
ikelos
2021-03-04 23:36:42 +00:00
committed by GitHub
parent a75ccd72b1 e18e9700f2
commit 9ef972192d
19 changed files with 132 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/cli/volargparse.py
+15 -7
View File
@@ -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:
volatility3/cli/volshell/generic.py
+9 -9
View File
@@ -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)
volatility3/framework/__init__.py
+4 -4
View File
@@ -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
volatility3/framework/automagic/__init__.py
+1 -1
View File
@@ -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."""
volatility3/framework/configuration/requirements.py
+6 -1
View File
@@ -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
volatility3/framework/exceptions.py
+2 -2
View File
@@ -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
volatility3/framework/interfaces/configuration.py
+8 -4
View File
@@ -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
volatility3/framework/interfaces/layers.py
+2 -2
View File
@@ -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
volatility3/framework/plugins/mac/lsmod.py
+6 -4
View File
@@ -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()
volatility3/framework/plugins/windows/cachedump.py
+23 -12
View File
@@ -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("<HH", cache_data[:4])
if len(cache_data[60:62]) == 0:
return (uname_len, domain_len, 0, '', '')
return (uname_len, domain_len, 0, b'', b'')
(domain_name_len, ) = unpack("<H", cache_data[60:62])
ch = cache_data[64:80]
enc_data = cache_data[96:]
return (uname_len, domain_len, domain_name_len, enc_data, ch)
def parse_decrypted_cache(self, dec_data, uname_len, domain_len, domain_name_len):
@staticmethod
def parse_decrypted_cache(dec_data: bytes, uname_len: int, domain_len: int,
domain_name_len: int) -> 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))
volatility3/framework/plugins/windows/dumpfiles.py
+4 -3
View File
@@ -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
volatility3/framework/plugins/windows/hashdump.py
+16 -22
View File
@@ -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("<L", V[0x0c:0x10])[0] + 0xCC
name_length = unpack("<L", V[0x10:0x14])[0]
if name_length > len(V):
name_offset = unpack("<L", value[0x0c:0x10])[0] + 0xCC
name_length = unpack("<L", value[0x10:0x14])[0]
if name_length > 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
volatility3/framework/plugins/windows/lsadump.py
+8 -5
View File
@@ -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,
volatility3/framework/plugins/windows/netscan.py
+2 -2
View File
@@ -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:
volatility3/framework/plugins/windows/netstat.py
+9 -9
View File
@@ -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
volatility3/framework/plugins/windows/vadinfo.py
+3 -3
View File
@@ -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
volatility3/framework/renderers/__init__.py
+6 -6
View File
@@ -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
volatility3/framework/renderers/format_hints.py
+7 -4
View File
@@ -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,
volatility3/framework/symbols/windows/pdbutil.py
+1 -1
View File
@@ -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