Windows: Run yapf over recent lsadump/cachedump plugins

This commit is contained in:
Mike Auty
2020-08-23 21:57:57 +01:00
parent 22fdd1e065
commit 807614aa45
2 changed files with 79 additions and 93 deletions
@@ -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("<HH", cache_data[:4])
if len(cache_data[60:62]) == 0:
@@ -59,21 +58,20 @@ class Cachedump(interfaces.plugins.PluginInterface):
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):
domain_len, domain_name_len):
"""Get the data from the cache and separate it into the username, domain name, and hash data"""
uname_offset= 72
uname_offset = 72
pad = 2 * ((uname_len / 2) % 2)
domain_offset= int(uname_offset+ uname_len + pad)
domain_offset = int(uname_offset + uname_len + pad)
pad = 2 * ((domain_len / 2) % 2)
domain_name_offset= int(domain_offset+ domain_len + pad)
domain_name_offset = int(domain_offset + domain_len + pad)
hashh = dec_data[:0x10]
username = dec_data[uname_offset:uname_offset+ uname_len]
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 = 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 = dec_data[domain_name_offset:domain_name_offset + domain_name_len]
domain_name = domain_name.decode('utf-16-le', 'replace')
return (username, domain, domain_name, hashh)
@@ -82,57 +80,55 @@ class Cachedump(interfaces.plugins.PluginInterface):
bootkey = hashdump.Hashdump.get_bootkey(syshive)
if not bootkey:
raise ValueError('Unable to find bootkey')
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'])
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))
return renderers.TreeGrid([("Username", str), ("Domain", str), ("Domain name", str), ('Hashh', bytes)],
self._generator(syshive, sechive))