From 807614aa4508ea81303c75d1154394aa0fc8d2ad Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Sun, 23 Aug 2020 21:57:57 +0100 Subject: [PATCH] Windows: Run yapf over recent lsadump/cachedump plugins --- .../framework/plugins/windows/cachedump.py | 86 +++++++++---------- .../framework/plugins/windows/lsadump.py | 86 ++++++++----------- 2 files changed, 79 insertions(+), 93 deletions(-) diff --git a/volatility/framework/plugins/windows/cachedump.py b/volatility/framework/plugins/windows/cachedump.py index d03d0b57a..2b71adf45 100644 --- a/volatility/framework/plugins/windows/cachedump.py +++ b/volatility/framework/plugins/windows/cachedump.py @@ -2,36 +2,35 @@ # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 # +from struct import unpack + +from Crypto.Cipher import ARC4, AES +from Crypto.Hash import HMAC + from volatility.framework import interfaces, renderers from volatility.framework.configuration import requirements -from volatility.framework.renderers import format_hints -from volatility.framework.layers import intel -from volatility.plugins.windows.registry import hivelist from volatility.plugins.windows import hashdump, lsadump, poolscanner -from Crypto.Hash import HMAC -from Crypto.Cipher import ARC4, AES -from struct import unpack +from volatility.plugins.windows.registry import hivelist class Cachedump(interfaces.plugins.PluginInterface): """Dumps lsa secrets from memory""" _version = (1, 0, 0) - + @classmethod def get_requirements(cls): return [requirements.TranslationLayerRequirement(name = 'primary', - description = 'Memory layer for the kernel', - architectures = ["Intel32", "Intel64"]), - requirements.SymbolTableRequirement(name = "nt_symbols", - description = "Windows kernel symbols"), - requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0)), - requirements.PluginRequirement(name = 'lsadump', plugin = lsadump.Lsadump, version = (1, 0, 0)) - ] + description = 'Memory layer for the kernel', + architectures = ["Intel32", "Intel64"]), + requirements.SymbolTableRequirement(name = "nt_symbols", + description = "Windows kernel symbols"), + requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0)), + requirements.PluginRequirement(name = 'lsadump', plugin = lsadump.Lsadump, version = (1, 0, 0)) + ] def get_nlkm(self, sechive, lsakey, is_vista_or_later): return lsadump.Lsadump.get_secret_by_name(sechive, 'NL$KM', lsakey, is_vista_or_later) - def decrypt_hash(self, edata, nlkm, ch, xp): if xp: @@ -44,12 +43,12 @@ class Cachedump(interfaces.plugins.PluginInterface): aes = AES.new(nlkm[16:32], AES.MODE_CBC, ch) data = "" for i in range(0, len(edata), 16): - buf = edata[i : i + 16] + buf = edata[i: i + 16] if len(buf) < 16: buf += (16 - len(buf)) * "\00" data += aes.decrypt(buf) return data - + def parse_cache_entry(self, cache_data): (uname_len, domain_len) = unpack("= (6, 0), - fallback_checks = [("KdCopyDataBlock", None, True)]) + fallback_checks = [("KdCopyDataBlock", None, True)]) vista_or_later = is_vista_or_later(context = self.context, symbol_table = self.config['nt_symbols']) lsakey = lsadump.Lsadump.get_lsa_key(sechive, bootkey, vista_or_later) if not lsakey: - raise ValueError('Unable to find lsa key') + raise ValueError('Unable to find lsa key') nlkm = self.get_nlkm(sechive, lsakey, vista_or_later) if not nlkm: - raise ValueError('Unable to find nlkma key') + raise ValueError('Unable to find nlkma key') cache = sechive.get_key("Cache") if not cache: - raise ValueError('Unable to find cache key') - + raise ValueError('Unable to find cache key') for cache_item in cache.get_values(): if cache_item.Name == "NL$Control": continue - data = sechive.read(cache_item.Data+4, cache_item.DataLength) + data = sechive.read(cache_item.Data + 4, cache_item.DataLength) if data == None: continue (uname_len, domain_len, domain_name_len, - enc_data, ch) = self.parse_cache_entry(data) + enc_data, ch) = self.parse_cache_entry(data) # Skip if nothing in this cache entry if uname_len == 0 or len(ch) == 0: continue dec_data = self.decrypt_hash(enc_data, nlkm, ch, not vista_or_later) (username, domain, domain_name, - hashh) = self.parse_decrypted_cache(dec_data, uname_len, - domain_len, domain_name_len) - yield (0,(username, domain, domain_name, hashh)) + hashh) = self.parse_decrypted_cache(dec_data, uname_len, + domain_len, domain_name_len) + yield (0, (username, domain, domain_name, hashh)) def run(self): offset = self.config.get('offset', None) - for hive in hivelist.HiveList.list_hives(self.context, - self.config_path, - self.config['primary'], - self.config['nt_symbols'], - hive_offsets = None if offset is None else [offset]): + self.config_path, + self.config['primary'], + self.config['nt_symbols'], + hive_offsets = None if offset is None else [offset]): if hive.get_name().split('\\')[-1].upper() == 'SYSTEM': - syshive=hive + syshive = hive if hive.get_name().split('\\')[-1].upper() == 'SECURITY': - sechive=hive + sechive = hive - return renderers.TreeGrid([("Username", str), ("Domain", str), ("Domain name", str), ('Hashh', bytes)], - self._generator(syshive, sechive)) \ No newline at end of file + return renderers.TreeGrid([("Username", str), ("Domain", str), ("Domain name", str), ('Hashh', bytes)], + self._generator(syshive, sechive)) diff --git a/volatility/framework/plugins/windows/lsadump.py b/volatility/framework/plugins/windows/lsadump.py index cf659c7ed..6a2ee8d4c 100644 --- a/volatility/framework/plugins/windows/lsadump.py +++ b/volatility/framework/plugins/windows/lsadump.py @@ -2,34 +2,34 @@ # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 # import logging +from struct import unpack + +from Crypto.Cipher import ARC4, DES, AES +from Crypto.Hash import MD5, SHA256 + from volatility.framework import interfaces, renderers from volatility.framework.configuration import requirements -from volatility.framework.renderers import format_hints -from volatility.framework.layers import intel -from volatility.plugins.windows.registry import hivelist from volatility.plugins.windows import hashdump, poolscanner -from Crypto.Hash import MD5, SHA256 -from Crypto.Cipher import ARC4, DES, AES -from struct import unpack, pack -import collections +from volatility.plugins.windows.registry import hivelist vollog = logging.getLogger(__name__) + class Lsadump(interfaces.plugins.PluginInterface): """Dumps lsa secrets from memory""" _version = (1, 0, 0) - + @classmethod def get_requirements(cls): return [requirements.TranslationLayerRequirement(name = 'primary', - description = 'Memory layer for the kernel', - architectures = ["Intel32", "Intel64"]), - requirements.SymbolTableRequirement(name = "nt_symbols", - description = "Windows kernel symbols"), - requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0)) - ] - + description = 'Memory layer for the kernel', + architectures = ["Intel32", "Intel64"]), + requirements.SymbolTableRequirement(name = "nt_symbols", + description = "Windows kernel symbols"), + requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0)) + ] + @classmethod def decrypt_aes(cls, secret, key): """ @@ -44,33 +44,32 @@ class Lsadump(interfaces.plugins.PluginInterface): data = b"" for i in range(60, len(secret), 16): aes = AES.new(aeskey, AES.MODE_CBC, b'\x00' * 16) - buf = secret[i : i + 16] + buf = secret[i: i + 16] if len(buf) < 16: buf += (16 - len(buf)) * "\00" data += aes.decrypt(buf) return data - + @classmethod def get_lsa_key(cls, sechive, bootkey, vista_or_later): if not bootkey: return None - + if vista_or_later: policy_key = 'PolEKList' else: policy_key = 'PolSecretEncryptionKey' - enc_reg_key = sechive.get_key("Policy\\"+policy_key) + enc_reg_key = sechive.get_key("Policy\\" + policy_key) if not enc_reg_key: return None enc_reg_value = next(enc_reg_key.get_values()) - if not enc_reg_value: return None - - obf_lsa_key = sechive.read(enc_reg_value.Data+4, enc_reg_value.DataLength) + + obf_lsa_key = sechive.read(enc_reg_value.Data + 4, enc_reg_value.DataLength) if not obf_lsa_key: return None @@ -88,21 +87,20 @@ class Lsadump(interfaces.plugins.PluginInterface): lsa_key = cls.decrypt_aes(obf_lsa_key, bootkey) lsa_key = lsa_key[68:100] return lsa_key - + @classmethod def get_secret_by_name(cls, sechive, name, lsakey, is_vista_or_later): try: enc_secret_key = sechive.get_key("Policy\\Secrets\\" + name + "\\CurrVal") except KeyError: raise ValueError("Unable to read cache from memory") - enc_secret_value = next(enc_secret_key.get_values()) if not enc_secret_value: return None - enc_secret = sechive.read(enc_secret_value.Data+4, - enc_secret_value.DataLength) + enc_secret = sechive.read(enc_secret_value.Data + 4, + enc_secret_value.DataLength) if not enc_secret: return None @@ -119,7 +117,7 @@ class Lsadump(interfaces.plugins.PluginInterface): Decrypts a block of data with DES using given key. Note that key can be longer than 7 bytes.""" decrypted_data = b'' - j = 0 # key index + j = 0 # key index for i in range(0, len(secret), 8): enc_block = secret[i:i + 8] @@ -139,14 +137,14 @@ class Lsadump(interfaces.plugins.PluginInterface): def _generator(self, syshive, sechive): is_vista_or_later = poolscanner.os_distinguisher(version_check = lambda x: x >= (6, 0), - fallback_checks = [("KdCopyDataBlock", None, True)]) + fallback_checks = [("KdCopyDataBlock", None, True)]) vista_or_later = is_vista_or_later(context = self.context, symbol_table = self.config['nt_symbols']) bootkey = hashdump.Hashdump.get_bootkey(syshive) lsakey = self.get_lsa_key(sechive, bootkey, vista_or_later) if not bootkey: raise ValueError('Unable to find bootkey') - + if not lsakey: raise ValueError('Unable to find lsa key') @@ -156,7 +154,7 @@ class Lsadump(interfaces.plugins.PluginInterface): for key in secrets_key.get_subkeys(): - sec_val_key=sechive.get_key('Policy\\Secrets\\' + key.get_key_path().split('\\')[3] + '\\CurrVal') + sec_val_key = sechive.get_key('Policy\\Secrets\\' + key.get_key_path().split('\\')[3] + '\\CurrVal') if not sec_val_key: continue @@ -165,7 +163,7 @@ class Lsadump(interfaces.plugins.PluginInterface): continue enc_secret = sechive.read(enc_secret_value.Data + 4, - enc_secret_value.DataLength) + enc_secret_value.DataLength) if not enc_secret: continue if not vista_or_later: @@ -173,30 +171,22 @@ class Lsadump(interfaces.plugins.PluginInterface): else: secret = self.decrypt_aes(enc_secret, lsakey) - - yield (0,(key.get_name(), secret.decode('latin1'), secret)) - - + yield (0, (key.get_name(), secret.decode('latin1'), secret)) def run(self): offset = self.config.get('offset', None) - for hive in hivelist.HiveList.list_hives(self.context, - self.config_path, - self.config['primary'], - self.config['nt_symbols'], - hive_offsets = None if offset is None else [offset]): + self.config_path, + self.config['primary'], + self.config['nt_symbols'], + hive_offsets = None if offset is None else [offset]): if hive.get_name().split('\\')[-1].upper() == 'SYSTEM': - syshive=hive + syshive = hive if hive.get_name().split('\\')[-1].upper() == 'SECURITY': - sechive=hive + sechive = hive - return renderers.TreeGrid([("Key", str), ("Secret", str), ('Hex', bytes)], - self._generator(syshive, sechive)) - - - - + return renderers.TreeGrid([("Key", str), ("Secret", str), ('Hex', bytes)], + self._generator(syshive, sechive))