Merge pull request #1707 from volatilityfoundation/1471-registry-plugins-spread-across-directories-needs-standardization

1471 registry plugins spread across directories needs standardization
This commit is contained in:
ikelos
2025-03-27 16:36:58 +00:00
committed by GitHub
11 changed files with 3280 additions and 3130 deletions
+49
View File
@@ -88,3 +88,52 @@ def deprecated_method(
return wrapper
return decorator
def renamed_class(deprecated_class_name: str, message: str, removal_date: str):
"""A decorator for marking classes as being renamed and removed in the future.
Callers to this function should explicitly update to use the other plugins instead.
Args:
deprecated_class_name: The name of the class being deprecated
message: A message added to the standard deprecation warning. Should include the replacement API paths
removal_date: A YYYY-MM-DD formatted date of when the function will be removed from the framework
"""
def decorator(replacement_func):
@functools.wraps(replacement_func)
def wrapper(*args, **kwargs):
warnings.warn(
f"This plugin ({deprecated_class_name}) has been renamed and will be removed in the first release after {removal_date}. {message}",
FutureWarning,
)
return replacement_func(*args, **kwargs)
return wrapper
return decorator
class PluginRenameClass:
"""Class to move all classmethod invocations (for when a plugin has been moved)"""
def __init_subclass__(cls, replacement_class, removal_date, **kwargs):
deprecated_class_name = f"{cls.__module__}.{cls.__qualname__}"
super().__init_subclass__(**kwargs)
for attr, value in replacement_class.__dict__.items():
if isinstance(value, classmethod) and attr != "get_requirements":
setattr(
cls,
attr,
classmethod(
renamed_class(
deprecated_class_name=deprecated_class_name,
removal_date=removal_date,
message=f"Please ensure all method calls to this plugin are replaced with calls to {replacement_class.__module__}.{replacement_class.__qualname__}",
)(value.__func__)
),
)
else:
if not attr.startswith("__"):
setattr(cls, attr, value)
return super(PluginRenameClass).__init_subclass__(**kwargs)
+10 -646
View File
@@ -1,656 +1,20 @@
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
# This file is Copyright 2025 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import dataclasses
import datetime
import enum
import itertools
import logging
from typing import Dict, Iterable, Iterator, List, Optional, Tuple, Union
from volatility3.framework import interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import registry
from volatility3.framework.renderers import conversion
from volatility3.framework.symbols.windows.extensions import registry as reg_extensions
from volatility3.plugins import timeliner
from volatility3.plugins.windows.registry import hivelist
from volatility3.framework import interfaces, deprecation
from volatility3.plugins.windows.registry import amcache
vollog = logging.getLogger(__name__)
#######################################################################
# More information about the following enums can be found in the report
# 'Analysis of the AmCache` by Blanche Lagny, 2019
#######################################################################
class Win8FileValName(enum.Enum):
"""
An enumeration that creates a helpful mapping of opaque Windows 8 Amcache
'File' subkey value names to their human-readable equivalent.
"""
ProgramID = "100"
SHA1Hash = "101"
Product = "0"
Company = "1"
Size = "6"
SizeOfImage = "7"
PEHeaderChecksum = "9"
LastModTime = "11" # REG_QWORD FILETIME
CreateTime = "12" # REG_QWORD FILETIME
Path = "15"
LastModTime2 = "17" # REG_QWORD FILETIME
Version = "d"
CompileTime = "f" # REG_QWORD UNIX EPOCH
class Win8ProgramValName(enum.Enum):
"""
An enumeration that creates a helpful mapping of opaque Windows 8 Amcache
'Program' subkey value names to their human-readable equivalent.
"""
Product = "0"
Version = "1"
Publisher = "2"
InstallTime = "a"
MSIProductCode = "11"
MSIPackageCode = "12"
ProductCode = "f"
PackageCode = "10"
class Win10InvAppFileValName(enum.Enum):
"""
An enumeration containing the most useful Windows 10 Amcache
'InventoryApplicationFile' subkey value names.
"""
FileId = "FileId"
LinkDate = "LinkDate"
LowerCaseLongPath = "LowerCaseLongPath"
ProductName = "ProductName"
ProductVersion = "ProductVersion"
ProgramID = "ProgramId"
Publisher = "Publisher"
class Win10InvAppValName(enum.Enum):
"""
An enumeration containing the most useful Windows 10 Amcache
'InventoryApplication' subkey value names.
"""
InstallDate = "InstallDate"
Name = "Name"
Publisher = "Publisher"
RootDirPath = "RootDirPath"
Version = "Version"
class Win10DriverBinaryValName(enum.Enum):
"""
An enumeration containing the most useful Windows 10 Amcache
'InventoryDriverBinary' subkey value names.
"""
DriverId = "DriverId"
DriverName = "DriverName"
DriverCompany = "DriverCompany"
Product = "Product"
Service = "Service"
DriverTimeStamp = "DriverTimeStamp"
class AmcacheEntryType(enum.IntEnum):
Driver = 1
Program = 2
File = 3
NullableString = Union[str, None, interfaces.renderers.BaseAbsentValue]
NullableDatetime = Union[datetime.datetime, None, interfaces.renderers.BaseAbsentValue]
@dataclasses.dataclass
class _AmcacheEntry:
"""
A class containing all information about an entry from the Amcache registry hive.
Because all values could potentially be paged out of memory or malformed, they are all
a union between their expected value and `interfaces.renderers.BaseAbsentValue`.
"""
entry_type: str
path: NullableString = renderers.NotApplicableValue()
company: NullableString = renderers.NotApplicableValue()
last_modify_time: NullableDatetime = renderers.NotApplicableValue()
last_modify_time_2: NullableDatetime = renderers.NotApplicableValue()
install_time: NullableDatetime = renderers.NotApplicableValue()
compile_time: NullableDatetime = renderers.NotApplicableValue()
sha1_hash: NullableString = renderers.NotApplicableValue()
service: NullableString = renderers.NotApplicableValue()
product_name: NullableString = renderers.NotApplicableValue()
product_version: NullableString = renderers.NotApplicableValue()
def _entry_sort_key(entry_tuple: Tuple[NullableString, _AmcacheEntry]) -> str:
"""Sorts entries by program ID. This is broken out as a function here
to ensure consistency in sorting between the `group_by` and `sorted` function
invocations.
"""
program_id, _ = entry_tuple
key = program_id if isinstance(program_id, str) else ""
return key
def _get_string_value(
values: Dict[str, reg_extensions.CM_KEY_VALUE], name: str
) -> NullableString:
try:
value = values[name]
except KeyError:
return renderers.NotAvailableValue()
data = value.decode_data()
if not isinstance(data, bytes):
return renderers.UnparsableValue()
return data.decode("utf-16le", errors="replace").rstrip("\u0000")
def _get_datetime_filetime_value(
values: Dict[str, reg_extensions.CM_KEY_VALUE], name: str
) -> NullableDatetime:
try:
value = values[name]
except KeyError:
return renderers.NotAvailableValue()
data = value.decode_data()
if not isinstance(data, int):
return renderers.UnparsableValue()
return conversion.wintime_to_datetime(data)
def _get_datetime_utc_epoch_value(
values: Dict[str, reg_extensions.CM_KEY_VALUE], name: str
) -> NullableDatetime:
try:
value = values[name]
except KeyError:
return renderers.NotAvailableValue()
data = value.decode_data()
if not isinstance(data, (int, float)):
return renderers.UnparsableValue()
try:
return datetime.datetime.fromtimestamp(float(data), datetime.timezone.utc)
except (ValueError, OverflowError, OSError):
return renderers.UnparsableValue()
def _get_datetime_str_value(
values: Dict[str, reg_extensions.CM_KEY_VALUE], name: str
) -> NullableDatetime:
try:
value = values[name]
except KeyError:
return renderers.NotAvailableValue()
data = value.decode_data()
if not isinstance(data, int):
return renderers.UnparsableValue()
if isinstance(data, str):
try:
return datetime.datetime.strptime(data, "%m/%d/%Y %H:%M:%S")
except ValueError:
return renderers.UnparsableValue()
else:
return renderers.UnparsableValue()
class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Extract information on executed applications from the AmCache."""
class Amcache(
interfaces.plugins.PluginInterface,
deprecation.PluginRenameClass,
replacement_class=amcache.Amcache,
removal_date="2025-09-25",
):
"""Extract information on executed applications from the AmCache (deprecated)."""
_required_framework_version = (2, 0, 0)
# 2.0.0 - changed the signature of get_amcache_hive
_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.ModuleRequirement(
name="kernel",
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="hivelist", component=hivelist.HiveList, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="timeliner",
component=timeliner.TimeLinerInterface,
version=(1, 0, 0),
),
]
def generate_timeline(
self,
) -> Iterator[Tuple[str, timeliner.TimeLinerType, datetime.datetime]]:
for _, entry in self._generator():
if isinstance(entry.last_modify_time, datetime.datetime):
yield f"Amcache: {entry.entry_type} {entry.path} registry key modified", timeliner.TimeLinerType.MODIFIED, entry.last_modify_time
if isinstance(entry.last_modify_time_2, datetime.datetime):
yield f"Amcache: {entry.entry_type} {entry.path} STANDARD_INFORMATION create time", timeliner.TimeLinerType.CREATED, entry.last_modify_time_2
if isinstance(entry.install_time, datetime.datetime):
yield f"Amcache: {entry.entry_type} {entry.path} installed", timeliner.TimeLinerType.CREATED, entry.install_time
if isinstance(entry.compile_time, datetime.datetime):
yield f"Amcache: {entry.entry_type} {entry.path} compiled (PE metadata)", timeliner.TimeLinerType.MODIFIED, entry.compile_time
@classmethod
def get_amcache_hive(
cls,
context: interfaces.context.ContextInterface,
config_path: str,
kernel_module_name: str,
) -> Optional[registry.RegistryHive]:
"""Retrieves the `Amcache.hve` registry hive from the kernel module, if it can be located."""
return next(
hivelist.HiveList.list_hives(
context=context,
base_config_path=interfaces.configuration.path_join(
config_path, "hivelist"
),
kernel_module_name=kernel_module_name,
filter_string="amcache",
),
None,
)
@classmethod
def parse_file_key(
cls, file_key: reg_extensions.CM_KEY_NODE
) -> Iterator[Tuple[NullableString, _AmcacheEntry]]:
"""Parses File entries from the Windows 8 `Root\\File` key.
:param programs_key: The `Root\\File` registry key.
:return: An iterator of tuples, where the first member is the program ID string for
correlating `Root\\Program` entries, and the second member is the `AmcacheEntry`.
"""
val_enum = Win8FileValName
wanted_values = [key.value for key in val_enum]
for file_entry_key in itertools.chain(
*(key.get_subkeys() for key in file_key.get_subkeys())
):
vollog.debug(f"Checking Win8 File key {file_entry_key.get_name()}")
values = {
str(value.get_name()): value
for value in file_entry_key.get_values()
if value.get_name() in wanted_values
}
program_id = _get_string_value(values, val_enum.ProgramID.value)
path = _get_string_value(values, val_enum.Path.value)
company = _get_string_value(values, val_enum.Company.value)
last_mod_time = _get_datetime_filetime_value(
values, val_enum.LastModTime.value
)
last_mod_time_2 = _get_datetime_filetime_value(
values, val_enum.LastModTime2.value
)
install_time = _get_datetime_filetime_value(
values, val_enum.CreateTime.value
)
compile_time = _get_datetime_utc_epoch_value(
values, val_enum.CompileTime.value
)
sha1_hash = _get_string_value(values, val_enum.SHA1Hash.value)
vollog.debug(f"Found sha1hash {sha1_hash}")
product_name = _get_string_value(values, val_enum.Product.value)
yield program_id, _AmcacheEntry(
AmcacheEntryType.File.name,
path=path,
company=company,
last_modify_time=last_mod_time,
last_modify_time_2=last_mod_time_2,
install_time=install_time,
compile_time=compile_time,
sha1_hash=(
sha1_hash.lstrip("0000")
if isinstance(sha1_hash, str)
else sha1_hash
),
product_name=product_name,
)
@classmethod
def parse_programs_key(
cls, programs_key: reg_extensions.CM_KEY_NODE
) -> Iterator[Tuple[str, _AmcacheEntry]]:
"""Parses Program entries from the Windows 8 `Root\\Programs` key.
:param programs_key: The `Root\\Programs` registry key.
:return: An iterator of tuples, where the first member is the program ID string for
correlating `Root\\File` entries, and the second member is the `AmcacheEntry`.
"""
val_enum = Win8ProgramValName
wanted_values = [key.value for key in val_enum]
for program_key in programs_key.get_subkeys():
values = {
str(value.get_name()): value
for value in program_key.get_values()
if value.get_name() in wanted_values
}
vollog.debug(f"Parsing Win8 Program key {program_key.get_name()}")
program_id = program_key.get_name().strip().strip("\u0000")
product = _get_string_value(values, val_enum.Product.value)
company = _get_string_value(values, val_enum.Publisher.value)
install_time = _get_datetime_utc_epoch_value(
values, val_enum.InstallTime.value
)
version = _get_string_value(values, val_enum.Version.value)
yield program_id, _AmcacheEntry(
AmcacheEntryType.Program.name,
company=company,
last_modify_time=conversion.wintime_to_datetime(
program_key.LastWriteTime.QuadPart
),
install_time=install_time,
product_name=product,
product_version=version,
)
@classmethod
def parse_inventory_app_key(
cls, inv_app_key: reg_extensions.CM_KEY_NODE
) -> Iterator[Tuple[str, _AmcacheEntry]]:
"""Parses InventoryApplication entries from the Windows 10 `Root\\InventoryApplication` key.
:param programs_key: The `Root\\InventoryApplication` registry key.
:return: An iterator of tuples, where the first member is the program ID string for
correlating `Root\\InventoryApplicationFile` entries, and the second member is the `AmcacheEntry`.
"""
val_enum = Win10InvAppValName
wanted_values = [key.value for key in val_enum]
for program_key in inv_app_key.get_subkeys():
program_id = program_key.get_name()
values = {
str(value.get_name()): value
for value in program_key.get_values()
if value.get_name() in wanted_values
}
name = _get_string_value(values, val_enum.Name.value)
version = _get_string_value(values, val_enum.Version.value)
publisher = _get_string_value(values, val_enum.Publisher.value)
path = _get_string_value(values, val_enum.RootDirPath.value)
install_date = _get_datetime_str_value(values, val_enum.InstallDate.value)
last_mod = conversion.wintime_to_datetime(
program_key.LastWriteTime.QuadPart
)
product: str = name if isinstance(name, str) else "UNKNOWN" # type: ignore
yield program_id.strip().strip("\u0000"), _AmcacheEntry(
AmcacheEntryType.Program.name,
path=path,
last_modify_time=last_mod,
install_time=install_date,
product_name=product,
company=publisher,
product_version=version,
)
@classmethod
def parse_inventory_app_file_key(
cls, inv_app_file_key: reg_extensions.CM_KEY_NODE
) -> Iterator[Tuple[NullableString, _AmcacheEntry]]:
"""Parses executable file entries from the `Root\\InventoryApplicationFile` registry key.
:param inv_app_file_key: The `Root\\InventoryApplicationFile` registry key.
:return: An iterator of tuples, where the first member is the program ID string for correlating
with it's parent `InventoryApplication` program entry, and the second member is the `Amcache` entry.
"""
val_enum = Win10InvAppFileValName
wanted_values = [key.value for key in val_enum]
for file_key in inv_app_file_key.get_subkeys():
vollog.debug(
f"Parsing Win10 InventoryApplicationFile key {file_key.get_name()}"
)
values = {
str(value.get_name()): value
for value in file_key.get_values()
if value.get_name() in wanted_values
}
last_mod = conversion.wintime_to_datetime(file_key.LastWriteTime.QuadPart)
path = _get_string_value(values, val_enum.LowerCaseLongPath.value)
linkdate = _get_datetime_str_value(values, val_enum.LinkDate.value)
sha1_hash = _get_string_value(values, val_enum.FileId.value)
publisher = _get_string_value(values, val_enum.Publisher.value)
prod_name = _get_string_value(values, val_enum.ProductName.value)
prod_ver = _get_string_value(values, val_enum.ProductVersion.value)
program_id = _get_string_value(values, val_enum.ProgramID.value)
yield program_id, _AmcacheEntry(
AmcacheEntryType.File.name,
path=path,
company=publisher,
last_modify_time=last_mod,
compile_time=linkdate,
sha1_hash=(
sha1_hash.lstrip("0000")
if isinstance(sha1_hash, str)
else sha1_hash
),
product_name=prod_name,
product_version=prod_ver,
)
@classmethod
def parse_driver_binary_key(
cls, driver_binary_key: reg_extensions.CM_KEY_NODE
) -> Iterator[_AmcacheEntry]:
"""Parses information about installed drivers from the `Root\\InventoryDriverBinary` registry key.
:param driver_binary_key: The `Root\\InventoryDriverBinary` registry key
:return: An iterator of `AmcacheEntry`s
"""
val_enum = Win10DriverBinaryValName
wanted_values = [key.value for key in val_enum]
for binary_key in driver_binary_key.get_subkeys():
values = {
str(value.get_name()): value
for value in binary_key.get_values()
if value.get_name() in wanted_values
}
# Depending on the Windows version, the key name will be either the name
# of the driver, or its SHA1 hash.
if "/" in str(binary_key.get_name()):
driver_name = str(binary_key.get_name())
sha1_hash = _get_string_value(values, val_enum.DriverId.name)
else:
sha1_hash = str(binary_key.get_name())
driver_name = _get_string_value(values, val_enum.DriverName.name)
if isinstance(sha1_hash, str):
sha1_hash = sha1_hash[4:] if sha1_hash.startswith("0000") else sha1_hash
company, product, service, last_write_time, driver_timestamp = (
_get_string_value(values, val_enum.DriverCompany.name),
_get_string_value(values, val_enum.Product.name),
_get_string_value(values, val_enum.Service.name),
conversion.wintime_to_datetime(binary_key.LastWriteTime.QuadPart),
_get_datetime_utc_epoch_value(values, val_enum.DriverTimeStamp.name),
)
yield _AmcacheEntry(
entry_type=AmcacheEntryType.Driver.name,
path=driver_name,
company=company,
last_modify_time=last_write_time,
compile_time=driver_timestamp,
sha1_hash=(
sha1_hash.lstrip("0000")
if isinstance(sha1_hash, str)
else sha1_hash
),
service=service,
product_name=product,
)
def _generator(self) -> Iterator[Tuple[int, _AmcacheEntry]]:
def indented(
entry_gen: Iterable[_AmcacheEntry], indent: int = 0
) -> Iterator[Tuple[int, _AmcacheEntry]]:
for item in entry_gen:
yield indent, item
# Building the dictionary ahead of time is much better for performance
# vs looking up each service's DLL individually.
amcache = self.get_amcache_hive(
self.context, self.config_path, self.config["kernel"]
)
if amcache is None:
return
try:
yield from indented(
self.parse_driver_binary_key(
amcache.get_key("Root\\InventoryDriverBinary") # type: ignore
)
)
except (KeyError, registry.RegistryException):
# Registry key not found
pass
try:
programs: Dict[str, _AmcacheEntry] = {
program_id: entry
for program_id, entry in self.parse_programs_key(
amcache.get_key("Root\\Programs")
) # type: ignore
}
except (KeyError, registry.RegistryException):
programs = {}
try:
files = sorted(
list(
self.parse_file_key(amcache.get_key("Root\\File")), # type: ignore
),
key=_entry_sort_key,
)
except (KeyError, registry.RegistryException):
files = []
for program_id, file_entries in itertools.groupby(
files,
key=_entry_sort_key,
):
files_indent = 0
if isinstance(program_id, str):
try:
program_entry = programs.pop(program_id.strip().strip("\u0000"))
yield (0, program_entry)
files_indent = 1
except KeyError:
# No parent program for this file entry
pass
for _, entry in file_entries:
yield files_indent, entry
for empty_program in programs.values():
yield 0, empty_program
try:
programs: Dict[str, _AmcacheEntry] = dict(
self.parse_inventory_app_key(
amcache.get_key("Root\\InventoryApplication") # type: ignore
)
)
except (KeyError, registry.RegistryException):
programs = {}
try:
files = sorted(
list(
self.parse_inventory_app_file_key(amcache.get_key("Root\\InventoryApplicationFile")), # type: ignore
),
key=_entry_sort_key,
)
except (KeyError, registry.RegistryException):
files = []
for program_id, file_entries in itertools.groupby(
files,
key=_entry_sort_key,
):
files_indent = 0
if isinstance(program_id, str):
try:
program_entry = programs.pop(program_id.strip().strip("\u0000"))
yield (0, program_entry)
files_indent = 1
except KeyError:
# No parent program for this file entry
pass
for _, entry in file_entries:
yield files_indent, entry
for empty_program in programs.values():
yield 0, empty_program
def run(self):
return renderers.TreeGrid(
[
("EntryType", str),
("Path", str),
("Company", str),
("LastModifyTime", datetime.datetime),
("LastModifyTime2", datetime.datetime),
("InstallTime", datetime.datetime),
("CompileTime", datetime.datetime),
("SHA1", str),
("Service", str),
("ProductName", str),
("ProductVersion", str),
],
(
(indent, dataclasses.astuple(entry))
for indent, entry in self._generator()
),
)
@@ -1,187 +1,20 @@
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# This file is Copyright 2025 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
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, 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
from volatility3.framework import interfaces, deprecation
from volatility3.plugins.windows.registry import cachedump
vollog = logging.getLogger(__name__)
class Cachedump(interfaces.plugins.PluginInterface):
"""Dumps lsa secrets from memory"""
class Cachedump(
interfaces.plugins.PluginInterface,
deprecation.PluginRenameClass,
replacement_class=cachedump.Cachedump,
removal_date="2025-09-25",
):
"""Dumps lsa secrets from memory (deprecated)"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls):
return [
requirements.ModuleRequirement(
name="kernel",
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="hivelist", component=hivelist.HiveList, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="lsadump", component=lsadump.Lsadump, version=(1, 0, 0)
),
requirements.VersionRequirement(
name="hashdump", component=hashdump.Hashdump, version=(1, 1, 0)
),
]
@classmethod
def get_nlkm(
cls, 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
)
@classmethod
def decrypt_hash(cls, edata: bytes, nlkm: bytes, ch, xp: bool):
if xp:
hmac_md5 = HMAC.new(nlkm, ch)
rc4key = hmac_md5.digest()
rc4 = ARC4.new(rc4key)
data = rc4.encrypt(edata) # lgtm [py/weak-cryptographic-algorithm]
else:
# Based on code from http://lab.mediaservice.net/code/cachedump.rb
aes = AES.new(nlkm[16:32], AES.MODE_CBC, ch)
data = b""
for i in range(0, len(edata), 16):
buf = edata[i : i + 16]
if len(buf) < 16:
buf += (16 - len(buf)) * b"\00"
data += aes.decrypt(buf)
return data
@classmethod
def parse_cache_entry(cls, 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, 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)
@classmethod
def parse_decrypted_cache(
cls, 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)
domain_offset = int(uname_offset + uname_len + pad)
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].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)
def _generator(self, syshive, sechive):
if not syshive or not sechive:
if syshive is None:
vollog.warning("Unable to locate SYSTEM hive")
if sechive is None:
vollog.warning("Unable to locate SECURITY hive")
return None
bootkey = hashdump.Hashdump.get_bootkey(syshive)
if not bootkey:
vollog.warning("Unable to find bootkey")
return None
kernel = self.context.modules[self.config["kernel"]]
vista_or_later = versions.is_vista_or_later(
context=self.context, symbol_table=kernel.symbol_table_name
)
lsakey = lsadump.Lsadump.get_lsa_key(sechive, bootkey, vista_or_later)
if not lsakey:
vollog.warning("Unable to find lsa key")
return None
nlkm = self.get_nlkm(sechive, lsakey, vista_or_later)
if not nlkm:
vollog.warning("Unable to find nlkma key")
return None
cache = hashdump.Hashdump.get_hive_key(sechive, "Cache")
if not cache:
vollog.warning("Unable to find cache key")
return None
for cache_item in cache.get_values():
if cache_item.Name == "NL$Control":
continue
try:
data = sechive.read(cache_item.Data + 4, cache_item.DataLength)
except exceptions.InvalidAddressException:
continue
if not data:
continue
(
uname_len,
domain_len,
domain_name_len,
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))
def run(self):
offset = self.config.get("offset", None)
syshive = sechive = None
for hive in hivelist.HiveList.list_hives(
context=self.context,
base_config_path=self.config_path,
kernel_module_name=self.config["kernel"],
hive_offsets=None if offset is None else [offset],
):
if hive.get_name().split("\\")[-1].upper() == "SYSTEM":
syshive = hive
if hive.get_name().split("\\")[-1].upper() == "SECURITY":
sechive = hive
return renderers.TreeGrid(
[("Username", str), ("Domain", str), ("Domain name", str), ("Hash", bytes)],
self._generator(syshive, sechive),
)
+10 -633
View File
@@ -1,643 +1,20 @@
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# This file is Copyright 2025 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import binascii
import hashlib
import logging
from struct import pack, unpack
from typing import List, Optional, Tuple
from Crypto.Cipher import AES, ARC4, DES
from volatility3.framework import interfaces, renderers, exceptions, constants
from volatility3.framework.configuration import requirements
from volatility3.framework.exceptions import InvalidAddressException
from volatility3.framework.layers import registry as registrylayer
from volatility3.framework.symbols.windows.extensions import registry
from volatility3.plugins.windows.registry import hivelist
from volatility3.framework import interfaces, deprecation
from volatility3.plugins.windows.registry import hashdump
vollog = logging.getLogger(__name__)
class Hashdump(interfaces.plugins.PluginInterface):
"""Dumps user hashes from memory"""
class Hashdump(
interfaces.plugins.PluginInterface,
deprecation.PluginRenameClass,
replacement_class=hashdump.Hashdump,
removal_date="2025-09-25",
):
"""Dumps user hashes from memory (deprecated)"""
_required_framework_version = (2, 0, 0)
_version = (1, 1, 1)
@classmethod
def get_requirements(cls):
return [
requirements.ModuleRequirement(
name="kernel",
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="hivelist", component=hivelist.HiveList, version=(2, 0, 0)
),
]
odd_parity = [
1,
1,
2,
2,
4,
4,
7,
7,
8,
8,
11,
11,
13,
13,
14,
14,
16,
16,
19,
19,
21,
21,
22,
22,
25,
25,
26,
26,
28,
28,
31,
31,
32,
32,
35,
35,
37,
37,
38,
38,
41,
41,
42,
42,
44,
44,
47,
47,
49,
49,
50,
50,
52,
52,
55,
55,
56,
56,
59,
59,
61,
61,
62,
62,
64,
64,
67,
67,
69,
69,
70,
70,
73,
73,
74,
74,
76,
76,
79,
79,
81,
81,
82,
82,
84,
84,
87,
87,
88,
88,
91,
91,
93,
93,
94,
94,
97,
97,
98,
98,
100,
100,
103,
103,
104,
104,
107,
107,
109,
109,
110,
110,
112,
112,
115,
115,
117,
117,
118,
118,
121,
121,
122,
122,
124,
124,
127,
127,
128,
128,
131,
131,
133,
133,
134,
134,
137,
137,
138,
138,
140,
140,
143,
143,
145,
145,
146,
146,
148,
148,
151,
151,
152,
152,
155,
155,
157,
157,
158,
158,
161,
161,
162,
162,
164,
164,
167,
167,
168,
168,
171,
171,
173,
173,
174,
174,
176,
176,
179,
179,
181,
181,
182,
182,
185,
185,
186,
186,
188,
188,
191,
191,
193,
193,
194,
194,
196,
196,
199,
199,
200,
200,
203,
203,
205,
205,
206,
206,
208,
208,
211,
211,
213,
213,
214,
214,
217,
217,
218,
218,
220,
220,
223,
223,
224,
224,
227,
227,
229,
229,
230,
230,
233,
233,
234,
234,
236,
236,
239,
239,
241,
241,
242,
242,
244,
244,
247,
247,
248,
248,
251,
251,
253,
253,
254,
254,
]
# Permutation matrix for boot key
bootkey_perm_table = [
0x8,
0x5,
0x4,
0x2,
0xB,
0x9,
0xD,
0x3,
0x0,
0x6,
0x1,
0xC,
0xE,
0xA,
0xF,
0x7,
]
# Constants for SAM decrypt algorithm
aqwerty = b"!@#$%^&*()qwertyUIOPAzxcvbnmQQQQQQQQQQQQ)(*@&%\0"
anum = b"0123456789012345678901234567890123456789\0"
antpassword = b"NTPASSWORD\0"
almpassword = b"LMPASSWORD\0"
lmkey = b"KGS!@#$%"
empty_lm = b"\xaa\xd3\xb4\x35\xb5\x14\x04\xee\xaa\xd3\xb4\x35\xb5\x14\x04\xee"
empty_nt = b"\x31\xd6\xcf\xe0\xd1\x6a\xe9\x31\xb7\x3c\x59\xd7\xe0\xc0\x89\xc0"
@classmethod
def get_hive_key(
cls, hive: registry.RegistryHive, key: str
) -> Optional["registry.CM_KEY_NODE"]:
result = None
try:
if hive:
result = hive.get_key(key)
except (KeyError, registrylayer.RegistryException):
vollog.info(
f"Unable to load the required registry key {hive.get_name()}\\{key} from this memory image"
)
return result
@classmethod
def get_user_keys(
cls, samhive: registry.RegistryHive
) -> List[interfaces.objects.ObjectInterface]:
user_key_path = "SAM\\Domains\\Account\\Users"
user_key = cls.get_hive_key(samhive, user_key_path)
if not user_key:
return []
return [k for k in user_key.get_subkeys() if k.Name != "Names"]
@classmethod
def get_bootkey(cls, syshive: registry.RegistryHive) -> Optional[bytes]:
"""
Returns the scrambled bootkey necesary to decrypt hashes
"""
cs = 1
lsa_base = f"ControlSet{cs:03}" + "\\Control\\Lsa"
lsa_keys = ["JD", "Skew1", "GBG", "Data"]
lsa = cls.get_hive_key(syshive, lsa_base)
if not lsa:
return None
bootkey = ""
for lk in lsa_keys:
try:
key = cls.get_hive_key(syshive, lsa_base + "\\" + lk)
class_data = None
if key:
try:
class_data = syshive.read(key.Class + 4, key.ClassLength)
except exceptions.InvalidAddressException:
return None
if class_data is None:
return None
bootkey += class_data.decode("utf-16-le")
except (
InvalidAddressException,
registrylayer.RegistryException,
) as excp:
vollog.log(
constants.LOGLEVEL_VVV, f"Unable to read Lsa key {lk}: {excp}"
)
return None
bootkey_str = binascii.unhexlify(bootkey)
bootkey_scrambled = bytes(
[bootkey_str[cls.bootkey_perm_table[i]] for i in range(len(bootkey_str))]
)
return bootkey_scrambled
@classmethod
def get_hbootkey(
cls, samhive: registry.RegistryHive, bootkey: bytes
) -> Optional[bytes]:
sam_account_path = "SAM\\Domains\\Account"
if not bootkey:
return None
sam_account_key = cls.get_hive_key(samhive, sam_account_path)
if not sam_account_key:
return None
sam_data = None
for v in sam_account_key.get_values():
if v.get_name() == "F":
try:
sam_data = samhive.read(v.Data + 4, v.DataLength)
except exceptions.InvalidAddressException:
return None
if not sam_data:
return None
revision = sam_data[0x00]
if revision == 2:
md5 = hashlib.md5()
md5.update(sam_data[0x70:0x80] + cls.aqwerty + bootkey + cls.anum)
rc4_key = md5.digest()
rc4 = ARC4.new(rc4_key)
hbootkey = rc4.encrypt(
sam_data[0x80:0xA0]
) # lgtm [py/weak-cryptographic-algorithm]
return hbootkey
elif revision == 3:
# AES encrypted
iv = sam_data[0x78:0x88]
encryptedHBootKey = sam_data[0x88:0xA8]
cipher = AES.new(bootkey, AES.MODE_CBC, iv)
hbootkey = cipher.decrypt(encryptedHBootKey)
return hbootkey[:16]
return None
@classmethod
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)
cipher = AES.new(hbootkey[:16], AES.MODE_CBC, salt)
obfkey = cipher.decrypt(enc_hash)
return des1.decrypt(obfkey[:8]) + des2.decrypt(
obfkey[8:16]
) # lgtm [py/weak-cryptographic-algorithm]
@classmethod
def get_user_hashes(
cls, user: registry.CM_KEY_NODE, samhive: registry.RegistryHive, 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)
except ValueError:
return None
sam_data = None
for v in user.get_values():
if v.get_name() == "V":
try:
sam_data = samhive.read(v.Data + 4, v.DataLength)
except (
exceptions.InvalidAddressException,
registrylayer.RegistryException,
):
return None
if not sam_data:
return None
lm_offset = unpack("<L", sam_data[0x9C:0xA0])[0] + 0xCC
lm_len = unpack("<L", sam_data[0xA0:0xA4])[0]
nt_offset = unpack("<L", sam_data[0xA8:0xAC])[0] + 0xCC
nt_len = unpack("<L", sam_data[0xAC:0xB0])[0]
lm_revision = sam_data[lm_offset + 2 : lm_offset + 3]
lmhash = None
if lm_revision == b"\x01":
if lm_len == 20:
enc_lm_hash = sam_data[lm_offset + 0x04 : lm_offset + 0x14]
lmhash = cls.decrypt_single_hash(
rid, hbootkey, enc_lm_hash, cls.almpassword
)
elif lm_revision == b"\x02":
if lm_len == 56:
lm_salt = sam_data[lm_offset + 4 : lm_offset + 20]
enc_lm_hash = sam_data[lm_offset + 20 : lm_offset + 52]
lmhash = cls.decrypt_single_salted_hash(
rid, hbootkey, enc_lm_hash, cls.almpassword, lm_salt
)
# NT hash decryption
nthash = None
nt_revision = sam_data[nt_offset + 2 : nt_offset + 3]
if nt_revision == b"\x01":
if nt_len == 20:
enc_nt_hash = sam_data[nt_offset + 4 : nt_offset + 20]
nthash = cls.decrypt_single_hash(
rid, hbootkey, enc_nt_hash, cls.antpassword
)
elif nt_revision == b"\x02":
if nt_len == 56:
nt_salt = sam_data[nt_offset + 8 : nt_offset + 24]
enc_nt_hash = sam_data[nt_offset + 24 : nt_offset + 56]
nthash = cls.decrypt_single_salted_hash(
rid, hbootkey, enc_nt_hash, cls.antpassword, nt_salt
)
return lmhash, nthash
@classmethod
def sid_to_key(cls, sid: int) -> Tuple[bytes, bytes]:
"""Takes rid of a user and converts it to a key to be used by the DES cipher"""
bytestr1 = [
sid & 0xFF,
(sid >> 8) & 0xFF,
(sid >> 16) & 0xFF,
(sid >> 24) & 0xFF,
]
bytestr1 += bytestr1[0:3]
bytestr2 = [bytestr1[3]] + bytestr1[0:3]
bytestr2 += bytestr2[0:3]
return cls.sidbytes_to_key(bytes(bytestr1)), cls.sidbytes_to_key(
bytes(bytestr2)
)
@classmethod
def sidbytes_to_key(cls, s: bytes) -> bytes:
"""Builds final DES key from the strings generated in sid_to_key"""
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: 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)
md5 = hashlib.md5()
md5.update(hbootkey[:0x10] + pack("<L", rid) + lmntstr)
rc4_key = md5.digest()
rc4 = ARC4.new(rc4_key)
obfkey = rc4.encrypt(enc_hash) # lgtm [py/weak-cryptographic-algorithm]
return des1.decrypt(obfkey[:8]) + des2.decrypt(
obfkey[8:]
) # lgtm [py/weak-cryptographic-algorithm]
@classmethod
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":
try:
value = samhive.read(v.Data + 4, v.DataLength)
except exceptions.InvalidAddressException:
return None
if not value:
return None
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 = value[name_offset : name_offset + name_length]
return username
# replaces the dump_hashes method in vol2
def _generator(
self, syshive: registry.RegistryHive, samhive: registry.RegistryHive
):
if syshive is None:
vollog.debug("SYSTEM address is None: No system hive found")
if samhive is None:
vollog.debug("SAM address is None: No SAM hive found")
bootkey = self.get_bootkey(syshive)
hbootkey = self.get_hbootkey(samhive, bootkey)
if hbootkey:
for user in self.get_user_keys(samhive):
ret = self.get_user_hashes(user, samhive, hbootkey)
if ret:
lmhash, nthash = ret
## temporary fix to prevent UnicodeDecodeError backtraces
## however this can cause truncated user names as a result
name = self.get_user_name(user, samhive)
if name is None:
name = renderers.NotAvailableValue()
else:
name = str(name, "utf-16-le", errors="ignore")
lmout = str(binascii.hexlify(lmhash or self.empty_lm), "latin-1")
ntout = str(binascii.hexlify(nthash or self.empty_nt), "latin-1")
rid = int(str(user.get_name()), 16)
yield (0, (name, rid, lmout, ntout))
else:
vollog.warning("Hbootkey is not valid")
def run(self):
offset = self.config.get("offset", None)
syshive = None
samhive = None
for hive in hivelist.HiveList.list_hives(
context=self.context,
base_config_path=self.config_path,
kernel_module_name=self.config["kernel"],
hive_offsets=None if offset is None else [offset],
):
if hive.get_name().split("\\")[-1].upper() == "SYSTEM":
syshive = hive
if hive.get_name().split("\\")[-1].upper() == "SAM":
samhive = hive
return renderers.TreeGrid(
[("User", str), ("rid", int), ("lmhash", str), ("nthash", str)],
self._generator(syshive, samhive),
)
+10 -248
View File
@@ -1,258 +1,20 @@
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# This file is Copyright 2025 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
from struct import unpack
from typing import Optional
import hashlib
from Crypto.Cipher import ARC4, DES, AES
from volatility3.framework import interfaces, renderers, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.exceptions import InvalidAddressException
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
from volatility3.framework.renderers import format_hints
from volatility3.framework import interfaces, deprecation
from volatility3.plugins.windows.registry import lsadump
vollog = logging.getLogger(__name__)
class Lsadump(interfaces.plugins.PluginInterface):
"""Dumps lsa secrets from memory"""
class Lsadump(
interfaces.plugins.PluginInterface,
deprecation.PluginRenameClass,
replacement_class=lsadump.Lsadump,
removal_date="2025-09-25",
):
"""Dumps lsa secrets from memory (deprecated)"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls):
return [
requirements.ModuleRequirement(
name="kernel",
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="hashdump", component=hashdump.Hashdump, version=(1, 1, 0)
),
requirements.VersionRequirement(
name="hivelist", component=hivelist.HiveList, version=(2, 0, 0)
),
]
@classmethod
def decrypt_aes(cls, secret: bytes, key: bytes) -> bytes:
"""
Based on code from http://lab.mediaservice.net/code/cachedump.rb
"""
sha = hashlib.sha256()
sha.update(key)
for _i in range(1, 1000 + 1):
sha.update(secret[28:60])
aeskey = sha.digest()
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]
if len(buf) < 16:
buf += (16 - len(buf)) * "\00"
data += aes.decrypt(buf)
return data
@classmethod
def get_lsa_key(
cls, sechive: registry.RegistryHive, bootkey: bytes, vista_or_later: bool
) -> Optional[bytes]:
if not bootkey:
return None
if vista_or_later:
policy_key = "PolEKList"
else:
policy_key = "PolSecretEncryptionKey"
enc_reg_key = hashdump.Hashdump.get_hive_key(sechive, "Policy\\" + policy_key)
if not enc_reg_key:
return None
enc_reg_value = next(enc_reg_key.get_values(), None)
if not enc_reg_value:
return None
try:
obf_lsa_key = sechive.read(enc_reg_value.Data + 4, enc_reg_value.DataLength)
except exceptions.InvalidAddressException:
return None
if not obf_lsa_key:
return None
if not vista_or_later:
md5 = hashlib.md5()
md5.update(bootkey)
for _i in range(1000):
md5.update(obf_lsa_key[60:76])
rc4key = md5.digest()
rc4 = ARC4.new(rc4key)
lsa_key = rc4.decrypt(
obf_lsa_key[12:60]
) # lgtm [py/weak-cryptographic-algorithm]
lsa_key = lsa_key[0x10:0x20]
else:
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: registry.RegistryHive,
name: str,
lsakey: bytes,
is_vista_or_later: bool,
) -> Optional[bytes]:
enc_secret_key = hashdump.Hashdump.get_hive_key(
sechive, "Policy\\Secrets\\" + name + "\\CurrVal"
)
secret = None
if enc_secret_key:
try:
enc_secret_value = next(enc_secret_key.get_values(), None)
except (
InvalidAddressException,
registry.RegistryException,
):
enc_secret_value = None
if enc_secret_value:
try:
enc_secret = sechive.read(
enc_secret_value.Data + 4, enc_secret_value.DataLength
)
except exceptions.InvalidAddressExceptions:
return None
if enc_secret:
if not is_vista_or_later:
secret = cls.decrypt_secret(enc_secret[0xC:], lsakey)
else:
secret = cls.decrypt_aes(enc_secret, lsakey)
return secret
@classmethod
def decrypt_secret(cls, secret: bytes, key: bytes) -> bytes:
"""Python implementation of SystemFunction005.
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
for i in range(0, len(secret), 8):
enc_block = secret[i : i + 8]
block_key = key[j : j + 7]
des_key = hashdump.Hashdump.sidbytes_to_key(block_key)
des = DES.new(des_key, DES.MODE_ECB)
enc_block = enc_block + b"\x00" * int(abs(8 - len(enc_block)) % 8)
decrypted_data += des.decrypt(
enc_block
) # lgtm [py/weak-cryptographic-algorithm]
j += 7
if len(key[j : j + 7]) < 7:
j = len(key[j : j + 7])
(dec_data_len,) = unpack("<L", decrypted_data[:4])
return decrypted_data[8 : 8 + dec_data_len]
def _generator(
self, syshive: registry.RegistryHive, sechive: registry.RegistryHive
):
kernel = self.context.modules[self.config["kernel"]]
vista_or_later = versions.is_vista_or_later(
context=self.context, symbol_table=kernel.symbol_table_name
)
bootkey = hashdump.Hashdump.get_bootkey(syshive)
if not bootkey:
vollog.warning("Unable to find bootkey")
return None
lsakey = self.get_lsa_key(sechive, bootkey, vista_or_later)
if not lsakey:
vollog.warning("Unable to find lsa key")
return None
secrets_key = hashdump.Hashdump.get_hive_key(sechive, "Policy\\Secrets")
if not secrets_key:
vollog.warning("Unable to find secrets key")
return None
for key in secrets_key.get_subkeys():
sec_val_key = hashdump.Hashdump.get_hive_key(
sechive,
"Policy\\Secrets\\" + key.get_key_path().split("\\")[3] + "\\CurrVal",
)
if not sec_val_key:
continue
try:
enc_secret_value = next(sec_val_key.get_values(), None)
except (
StopIteration,
InvalidAddressException,
registry.RegistryException,
):
enc_secret_value = None
if not enc_secret_value:
continue
try:
enc_secret = sechive.read(
enc_secret_value.Data + 4, enc_secret_value.DataLength
)
except exceptions.InvalidAddressExceptions:
continue
if not vista_or_later:
secret = self.decrypt_secret(enc_secret[0xC:], lsakey)
else:
secret = self.decrypt_aes(enc_secret, lsakey)
try:
key_name = key.get_name()
except (
InvalidAddressException,
registry.RegistryException,
):
key_name = renderers.UnreadableValue()
yield (0, (key_name, format_hints.HexBytes(secret), secret))
def run(self):
offset = self.config.get("offset", None)
syshive = sechive = None
for hive in hivelist.HiveList.list_hives(
context=self.context,
base_config_path=self.config_path,
kernel_module_name=self.config["kernel"],
hive_offsets=None if offset is None else [offset],
):
if hive.get_name().split("\\")[-1].upper() == "SYSTEM":
syshive = hive
if hive.get_name().split("\\")[-1].upper() == "SECURITY":
sechive = hive
return renderers.TreeGrid(
[("Key", str), ("Secret", format_hints.HexBytes), ("Hex", bytes)],
self._generator(syshive, sechive),
)
@@ -0,0 +1,658 @@
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import dataclasses
import datetime
import enum
import itertools
import logging
from typing import Dict, Iterable, Iterator, List, Optional, Tuple, Union
from volatility3.framework import interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import registry
from volatility3.framework.renderers import conversion
from volatility3.framework.symbols.windows.extensions import registry as reg_extensions
from volatility3.plugins import timeliner
from volatility3.plugins.windows.registry import hivelist
vollog = logging.getLogger(__name__)
#######################################################################
# More information about the following enums can be found in the report
# 'Analysis of the AmCache` by Blanche Lagny, 2019
#######################################################################
class Win8FileValName(enum.Enum):
"""
An enumeration that creates a helpful mapping of opaque Windows 8 Amcache
'File' subkey value names to their human-readable equivalent.
"""
ProgramID = "100"
SHA1Hash = "101"
Product = "0"
Company = "1"
Size = "6"
SizeOfImage = "7"
PEHeaderChecksum = "9"
LastModTime = "11" # REG_QWORD FILETIME
CreateTime = "12" # REG_QWORD FILETIME
Path = "15"
LastModTime2 = "17" # REG_QWORD FILETIME
Version = "d"
CompileTime = "f" # REG_QWORD UNIX EPOCH
class Win8ProgramValName(enum.Enum):
"""
An enumeration that creates a helpful mapping of opaque Windows 8 Amcache
'Program' subkey value names to their human-readable equivalent.
"""
Product = "0"
Version = "1"
Publisher = "2"
InstallTime = "a"
MSIProductCode = "11"
MSIPackageCode = "12"
ProductCode = "f"
PackageCode = "10"
class Win10InvAppFileValName(enum.Enum):
"""
An enumeration containing the most useful Windows 10 Amcache
'InventoryApplicationFile' subkey value names.
"""
FileId = "FileId"
LinkDate = "LinkDate"
LowerCaseLongPath = "LowerCaseLongPath"
ProductName = "ProductName"
ProductVersion = "ProductVersion"
ProgramID = "ProgramId"
Publisher = "Publisher"
class Win10InvAppValName(enum.Enum):
"""
An enumeration containing the most useful Windows 10 Amcache
'InventoryApplication' subkey value names.
"""
InstallDate = "InstallDate"
Name = "Name"
Publisher = "Publisher"
RootDirPath = "RootDirPath"
Version = "Version"
class Win10DriverBinaryValName(enum.Enum):
"""
An enumeration containing the most useful Windows 10 Amcache
'InventoryDriverBinary' subkey value names.
"""
DriverId = "DriverId"
DriverName = "DriverName"
DriverCompany = "DriverCompany"
Product = "Product"
Service = "Service"
DriverTimeStamp = "DriverTimeStamp"
class AmcacheEntryType(enum.IntEnum):
Driver = 1
Program = 2
File = 3
NullableString = Union[str, None, interfaces.renderers.BaseAbsentValue]
NullableDatetime = Union[datetime.datetime, None, interfaces.renderers.BaseAbsentValue]
@dataclasses.dataclass
class _AmcacheEntry:
"""
A class containing all information about an entry from the Amcache registry hive.
Because all values could potentially be paged out of memory or malformed, they are all
a union between their expected value and `interfaces.renderers.BaseAbsentValue`.
"""
entry_type: str
path: NullableString = renderers.NotApplicableValue()
company: NullableString = renderers.NotApplicableValue()
last_modify_time: NullableDatetime = renderers.NotApplicableValue()
last_modify_time_2: NullableDatetime = renderers.NotApplicableValue()
install_time: NullableDatetime = renderers.NotApplicableValue()
compile_time: NullableDatetime = renderers.NotApplicableValue()
sha1_hash: NullableString = renderers.NotApplicableValue()
service: NullableString = renderers.NotApplicableValue()
product_name: NullableString = renderers.NotApplicableValue()
product_version: NullableString = renderers.NotApplicableValue()
def _entry_sort_key(entry_tuple: Tuple[NullableString, _AmcacheEntry]) -> str:
"""Sorts entries by program ID. This is broken out as a function here
to ensure consistency in sorting between the `group_by` and `sorted` function
invocations.
"""
program_id, _ = entry_tuple
key = program_id if isinstance(program_id, str) else ""
return key
def _get_string_value(
values: Dict[str, reg_extensions.CM_KEY_VALUE], name: str
) -> NullableString:
try:
value = values[name]
except KeyError:
return renderers.NotAvailableValue()
data = value.decode_data()
if not isinstance(data, bytes):
return renderers.UnparsableValue()
return data.decode("utf-16le", errors="replace").rstrip("\u0000")
def _get_datetime_filetime_value(
values: Dict[str, reg_extensions.CM_KEY_VALUE], name: str
) -> NullableDatetime:
try:
value = values[name]
except KeyError:
return renderers.NotAvailableValue()
data = value.decode_data()
if not isinstance(data, int):
return renderers.UnparsableValue()
return conversion.wintime_to_datetime(data)
def _get_datetime_utc_epoch_value(
values: Dict[str, reg_extensions.CM_KEY_VALUE], name: str
) -> NullableDatetime:
try:
value = values[name]
except KeyError:
return renderers.NotAvailableValue()
data = value.decode_data()
if not isinstance(data, (int, float)):
return renderers.UnparsableValue()
try:
return datetime.datetime.fromtimestamp(float(data), datetime.timezone.utc)
except (ValueError, OverflowError, OSError):
return renderers.UnparsableValue()
def _get_datetime_str_value(
values: Dict[str, reg_extensions.CM_KEY_VALUE], name: str
) -> NullableDatetime:
try:
value = values[name]
except KeyError:
return renderers.NotAvailableValue()
data = value.decode_data()
if not isinstance(data, int):
return renderers.UnparsableValue()
if isinstance(data, str):
try:
return datetime.datetime.strptime(data, "%m/%d/%Y %H:%M:%S")
except ValueError:
return renderers.UnparsableValue()
else:
return renderers.UnparsableValue()
class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Extract information on executed applications from the AmCache."""
_required_framework_version = (2, 0, 0)
# 2.0.0 - changed the signature of get_amcache_hive
_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
# Since we're calling the plugin, make sure we have the plugin's requirements
return [
requirements.ModuleRequirement(
name="kernel",
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="hivelist", component=hivelist.HiveList, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="timeliner",
component=timeliner.TimeLinerInterface,
version=(1, 0, 0),
),
]
def generate_timeline(
self,
) -> Iterator[Tuple[str, timeliner.TimeLinerType, datetime.datetime]]:
for _, entry in self._generator():
if isinstance(entry.last_modify_time, datetime.datetime):
yield f"Amcache: {entry.entry_type} {entry.path} registry key modified", timeliner.TimeLinerType.MODIFIED, entry.last_modify_time
if isinstance(entry.last_modify_time_2, datetime.datetime):
yield f"Amcache: {entry.entry_type} {entry.path} STANDARD_INFORMATION create time", timeliner.TimeLinerType.CREATED, entry.last_modify_time_2
if isinstance(entry.install_time, datetime.datetime):
yield f"Amcache: {entry.entry_type} {entry.path} installed", timeliner.TimeLinerType.CREATED, entry.install_time
if isinstance(entry.compile_time, datetime.datetime):
yield f"Amcache: {entry.entry_type} {entry.path} compiled (PE metadata)", timeliner.TimeLinerType.MODIFIED, entry.compile_time
@classmethod
def get_amcache_hive(
cls,
context: interfaces.context.ContextInterface,
config_path: str,
kernel_module_name: str,
) -> Optional[registry.RegistryHive]:
"""Retrieves the `Amcache.hve` registry hive from the kernel module, if it can be located."""
return next(
hivelist.HiveList.list_hives(
context=context,
base_config_path=interfaces.configuration.path_join(
config_path, "hivelist"
),
kernel_module_name=kernel_module_name,
filter_string="amcache",
),
None,
)
@classmethod
def parse_file_key(
cls, file_key: reg_extensions.CM_KEY_NODE
) -> Iterator[Tuple[NullableString, _AmcacheEntry]]:
"""Parses File entries from the Windows 8 `Root\\File` key.
:param programs_key: The `Root\\File` registry key.
:return: An iterator of tuples, where the first member is the program ID string for
correlating `Root\\Program` entries, and the second member is the `AmcacheEntry`.
"""
val_enum = Win8FileValName
wanted_values = [key.value for key in val_enum]
for file_entry_key in itertools.chain(
*(key.get_subkeys() for key in file_key.get_subkeys())
):
vollog.debug(f"Checking Win8 File key {file_entry_key.get_name()}")
values = {
str(value.get_name()): value
for value in file_entry_key.get_values()
if value.get_name() in wanted_values
}
program_id = _get_string_value(values, val_enum.ProgramID.value)
path = _get_string_value(values, val_enum.Path.value)
company = _get_string_value(values, val_enum.Company.value)
last_mod_time = _get_datetime_filetime_value(
values, val_enum.LastModTime.value
)
last_mod_time_2 = _get_datetime_filetime_value(
values, val_enum.LastModTime2.value
)
install_time = _get_datetime_filetime_value(
values, val_enum.CreateTime.value
)
compile_time = _get_datetime_utc_epoch_value(
values, val_enum.CompileTime.value
)
sha1_hash = _get_string_value(values, val_enum.SHA1Hash.value)
vollog.debug(f"Found sha1hash {sha1_hash}")
product_name = _get_string_value(values, val_enum.Product.value)
yield program_id, _AmcacheEntry(
AmcacheEntryType.File.name,
path=path,
company=company,
last_modify_time=last_mod_time,
last_modify_time_2=last_mod_time_2,
install_time=install_time,
compile_time=compile_time,
sha1_hash=(
sha1_hash.lstrip("0000")
if isinstance(sha1_hash, str)
else sha1_hash
),
product_name=product_name,
)
@classmethod
def parse_programs_key(
cls, programs_key: reg_extensions.CM_KEY_NODE
) -> Iterator[Tuple[str, _AmcacheEntry]]:
"""Parses Program entries from the Windows 8 `Root\\Programs` key.
:param programs_key: The `Root\\Programs` registry key.
:return: An iterator of tuples, where the first member is the program ID string for
correlating `Root\\File` entries, and the second member is the `AmcacheEntry`.
"""
val_enum = Win8ProgramValName
wanted_values = [key.value for key in val_enum]
for program_key in programs_key.get_subkeys():
values = {
str(value.get_name()): value
for value in program_key.get_values()
if value.get_name() in wanted_values
}
vollog.debug(f"Parsing Win8 Program key {program_key.get_name()}")
program_id = program_key.get_name().strip().strip("\u0000")
product = _get_string_value(values, val_enum.Product.value)
company = _get_string_value(values, val_enum.Publisher.value)
install_time = _get_datetime_utc_epoch_value(
values, val_enum.InstallTime.value
)
version = _get_string_value(values, val_enum.Version.value)
yield program_id, _AmcacheEntry(
AmcacheEntryType.Program.name,
company=company,
last_modify_time=conversion.wintime_to_datetime(
program_key.LastWriteTime.QuadPart
),
install_time=install_time,
product_name=product,
product_version=version,
)
@classmethod
def parse_inventory_app_key(
cls, inv_app_key: reg_extensions.CM_KEY_NODE
) -> Iterator[Tuple[str, _AmcacheEntry]]:
"""Parses InventoryApplication entries from the Windows 10 `Root\\InventoryApplication` key.
:param programs_key: The `Root\\InventoryApplication` registry key.
:return: An iterator of tuples, where the first member is the program ID string for
correlating `Root\\InventoryApplicationFile` entries, and the second member is the `AmcacheEntry`.
"""
val_enum = Win10InvAppValName
wanted_values = [key.value for key in val_enum]
for program_key in inv_app_key.get_subkeys():
program_id = program_key.get_name()
values = {
str(value.get_name()): value
for value in program_key.get_values()
if value.get_name() in wanted_values
}
name = _get_string_value(values, val_enum.Name.value)
version = _get_string_value(values, val_enum.Version.value)
publisher = _get_string_value(values, val_enum.Publisher.value)
path = _get_string_value(values, val_enum.RootDirPath.value)
install_date = _get_datetime_str_value(values, val_enum.InstallDate.value)
last_mod = conversion.wintime_to_datetime(
program_key.LastWriteTime.QuadPart
)
product: str = name if isinstance(name, str) else "UNKNOWN" # type: ignore
yield program_id.strip().strip("\u0000"), _AmcacheEntry(
AmcacheEntryType.Program.name,
path=path,
last_modify_time=last_mod,
install_time=install_date,
product_name=product,
company=publisher,
product_version=version,
)
@classmethod
def parse_inventory_app_file_key(
cls, inv_app_file_key: reg_extensions.CM_KEY_NODE
) -> Iterator[Tuple[NullableString, _AmcacheEntry]]:
"""Parses executable file entries from the `Root\\InventoryApplicationFile` registry key.
:param inv_app_file_key: The `Root\\InventoryApplicationFile` registry key.
:return: An iterator of tuples, where the first member is the program ID string for correlating
with it's parent `InventoryApplication` program entry, and the second member is the `Amcache` entry.
"""
val_enum = Win10InvAppFileValName
wanted_values = [key.value for key in val_enum]
for file_key in inv_app_file_key.get_subkeys():
vollog.debug(
f"Parsing Win10 InventoryApplicationFile key {file_key.get_name()}"
)
values = {
str(value.get_name()): value
for value in file_key.get_values()
if value.get_name() in wanted_values
}
last_mod = conversion.wintime_to_datetime(file_key.LastWriteTime.QuadPart)
path = _get_string_value(values, val_enum.LowerCaseLongPath.value)
linkdate = _get_datetime_str_value(values, val_enum.LinkDate.value)
sha1_hash = _get_string_value(values, val_enum.FileId.value)
publisher = _get_string_value(values, val_enum.Publisher.value)
prod_name = _get_string_value(values, val_enum.ProductName.value)
prod_ver = _get_string_value(values, val_enum.ProductVersion.value)
program_id = _get_string_value(values, val_enum.ProgramID.value)
yield program_id, _AmcacheEntry(
AmcacheEntryType.File.name,
path=path,
company=publisher,
last_modify_time=last_mod,
compile_time=linkdate,
sha1_hash=(
sha1_hash.lstrip("0000")
if isinstance(sha1_hash, str)
else sha1_hash
),
product_name=prod_name,
product_version=prod_ver,
)
@classmethod
def parse_driver_binary_key(
cls, driver_binary_key: reg_extensions.CM_KEY_NODE
) -> Iterator[_AmcacheEntry]:
"""Parses information about installed drivers from the `Root\\InventoryDriverBinary` registry key.
:param driver_binary_key: The `Root\\InventoryDriverBinary` registry key
:return: An iterator of `AmcacheEntry`s
"""
val_enum = Win10DriverBinaryValName
wanted_values = [key.value for key in val_enum]
for binary_key in driver_binary_key.get_subkeys():
values = {
str(value.get_name()): value
for value in binary_key.get_values()
if value.get_name() in wanted_values
}
# Depending on the Windows version, the key name will be either the name
# of the driver, or its SHA1 hash.
if "/" in str(binary_key.get_name()):
driver_name = str(binary_key.get_name())
sha1_hash = _get_string_value(values, val_enum.DriverId.name)
else:
sha1_hash = str(binary_key.get_name())
driver_name = _get_string_value(values, val_enum.DriverName.name)
if isinstance(sha1_hash, str):
sha1_hash = sha1_hash[4:] if sha1_hash.startswith("0000") else sha1_hash
company, product, service, last_write_time, driver_timestamp = (
_get_string_value(values, val_enum.DriverCompany.name),
_get_string_value(values, val_enum.Product.name),
_get_string_value(values, val_enum.Service.name),
conversion.wintime_to_datetime(binary_key.LastWriteTime.QuadPart),
_get_datetime_utc_epoch_value(values, val_enum.DriverTimeStamp.name),
)
yield _AmcacheEntry(
entry_type=AmcacheEntryType.Driver.name,
path=driver_name,
company=company,
last_modify_time=last_write_time,
compile_time=driver_timestamp,
sha1_hash=(
sha1_hash.lstrip("0000")
if isinstance(sha1_hash, str)
else sha1_hash
),
service=service,
product_name=product,
)
def _generator(self) -> Iterator[Tuple[int, _AmcacheEntry]]:
def indented(
entry_gen: Iterable[_AmcacheEntry], indent: int = 0
) -> Iterator[Tuple[int, _AmcacheEntry]]:
for item in entry_gen:
yield indent, item
# Building the dictionary ahead of time is much better for performance
# vs looking up each service's DLL individually.
amcache = self.get_amcache_hive(
self.context, self.config_path, self.config["kernel"]
)
if amcache is None:
return
try:
yield from indented(
self.parse_driver_binary_key(
amcache.get_key("Root\\InventoryDriverBinary") # type: ignore
)
)
except (KeyError, registry.RegistryException):
# Registry key not found
pass
try:
programs: Dict[str, _AmcacheEntry] = {
program_id: entry
for program_id, entry in self.parse_programs_key(
amcache.get_key("Root\\Programs")
) # type: ignore
}
except (KeyError, registry.RegistryException):
programs = {}
try:
files = sorted(
list(
self.parse_file_key(amcache.get_key("Root\\File")), # type: ignore
),
key=_entry_sort_key,
)
except (KeyError, registry.RegistryException):
files = []
for program_id, file_entries in itertools.groupby(
files,
key=_entry_sort_key,
):
files_indent = 0
if isinstance(program_id, str):
try:
program_entry = programs.pop(program_id.strip().strip("\u0000"))
yield (0, program_entry)
files_indent = 1
except KeyError:
# No parent program for this file entry
pass
for _, entry in file_entries:
yield files_indent, entry
for empty_program in programs.values():
yield 0, empty_program
try:
programs: Dict[str, _AmcacheEntry] = dict(
self.parse_inventory_app_key(
amcache.get_key("Root\\InventoryApplication") # type: ignore
)
)
except (KeyError, registry.RegistryException):
programs = {}
try:
files = sorted(
list(
self.parse_inventory_app_file_key(
amcache.get_key("Root\\InventoryApplicationFile")
),
# type: ignore
),
key=_entry_sort_key,
)
except (KeyError, registry.RegistryException):
files = []
for program_id, file_entries in itertools.groupby(
files,
key=_entry_sort_key,
):
files_indent = 0
if isinstance(program_id, str):
try:
program_entry = programs.pop(program_id.strip().strip("\u0000"))
yield (0, program_entry)
files_indent = 1
except KeyError:
# No parent program for this file entry
pass
for _, entry in file_entries:
yield files_indent, entry
for empty_program in programs.values():
yield 0, empty_program
def run(self):
return renderers.TreeGrid(
[
("EntryType", str),
("Path", str),
("Company", str),
("LastModifyTime", datetime.datetime),
("LastModifyTime2", datetime.datetime),
("InstallTime", datetime.datetime),
("CompileTime", datetime.datetime),
("SHA1", str),
("Service", str),
("ProductName", str),
("ProductVersion", str),
],
(
(indent, dataclasses.astuple(entry))
for indent, entry in self._generator()
),
)
@@ -0,0 +1,186 @@
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
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, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import registry
from volatility3.framework.symbols.windows import versions
from volatility3.plugins.windows.registry import hashdump, hivelist, lsadump
vollog = logging.getLogger(__name__)
class Cachedump(interfaces.plugins.PluginInterface):
"""Dumps lsa secrets from memory"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls):
return [
requirements.ModuleRequirement(
name="kernel",
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="hivelist", component=hivelist.HiveList, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="lsadump", component=lsadump.Lsadump, version=(1, 0, 0)
),
requirements.VersionRequirement(
name="hashdump", component=hashdump.Hashdump, version=(1, 1, 0)
),
]
@classmethod
def get_nlkm(
cls, 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
)
@classmethod
def decrypt_hash(cls, edata: bytes, nlkm: bytes, ch, xp: bool):
if xp:
hmac_md5 = HMAC.new(nlkm, ch)
rc4key = hmac_md5.digest()
rc4 = ARC4.new(rc4key)
data = rc4.encrypt(edata) # lgtm [py/weak-cryptographic-algorithm]
else:
# Based on code from http://lab.mediaservice.net/code/cachedump.rb
aes = AES.new(nlkm[16:32], AES.MODE_CBC, ch)
data = b""
for i in range(0, len(edata), 16):
buf = edata[i : i + 16]
if len(buf) < 16:
buf += (16 - len(buf)) * b"\00"
data += aes.decrypt(buf)
return data
@classmethod
def parse_cache_entry(cls, 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, 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)
@classmethod
def parse_decrypted_cache(
cls, 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)
domain_offset = int(uname_offset + uname_len + pad)
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].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)
def _generator(self, syshive, sechive):
if not syshive or not sechive:
if syshive is None:
vollog.warning("Unable to locate SYSTEM hive")
if sechive is None:
vollog.warning("Unable to locate SECURITY hive")
return None
bootkey = hashdump.Hashdump.get_bootkey(syshive)
if not bootkey:
vollog.warning("Unable to find bootkey")
return None
kernel = self.context.modules[self.config["kernel"]]
vista_or_later = versions.is_vista_or_later(
context=self.context, symbol_table=kernel.symbol_table_name
)
lsakey = lsadump.Lsadump.get_lsa_key(sechive, bootkey, vista_or_later)
if not lsakey:
vollog.warning("Unable to find lsa key")
return None
nlkm = self.get_nlkm(sechive, lsakey, vista_or_later)
if not nlkm:
vollog.warning("Unable to find nlkma key")
return None
cache = hashdump.Hashdump.get_hive_key(sechive, "Cache")
if not cache:
vollog.warning("Unable to find cache key")
return None
for cache_item in cache.get_values():
if cache_item.Name == "NL$Control":
continue
try:
data = sechive.read(cache_item.Data + 4, cache_item.DataLength)
except exceptions.InvalidAddressException:
continue
if not data:
continue
(
uname_len,
domain_len,
domain_name_len,
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))
def run(self):
offset = self.config.get("offset", None)
syshive = sechive = None
for hive in hivelist.HiveList.list_hives(
context=self.context,
base_config_path=self.config_path,
kernel_module_name=self.config["kernel"],
hive_offsets=None if offset is None else [offset],
):
if hive.get_name().split("\\")[-1].upper() == "SYSTEM":
syshive = hive
if hive.get_name().split("\\")[-1].upper() == "SECURITY":
sechive = hive
return renderers.TreeGrid(
[("Username", str), ("Domain", str), ("Domain name", str), ("Hash", bytes)],
self._generator(syshive, sechive),
)
@@ -0,0 +1,643 @@
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import binascii
import hashlib
import logging
from struct import pack, unpack
from typing import List, Optional, Tuple
from Crypto.Cipher import AES, ARC4, DES
from volatility3.framework import interfaces, renderers, exceptions, constants
from volatility3.framework.configuration import requirements
from volatility3.framework.exceptions import InvalidAddressException
from volatility3.framework.layers import registry as registrylayer
from volatility3.framework.symbols.windows.extensions import registry
from volatility3.plugins.windows.registry import hivelist
vollog = logging.getLogger(__name__)
class Hashdump(interfaces.plugins.PluginInterface):
"""Dumps user hashes from memory"""
_required_framework_version = (2, 0, 0)
_version = (1, 1, 1)
@classmethod
def get_requirements(cls):
return [
requirements.ModuleRequirement(
name="kernel",
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="hivelist", component=hivelist.HiveList, version=(2, 0, 0)
),
]
odd_parity = [
1,
1,
2,
2,
4,
4,
7,
7,
8,
8,
11,
11,
13,
13,
14,
14,
16,
16,
19,
19,
21,
21,
22,
22,
25,
25,
26,
26,
28,
28,
31,
31,
32,
32,
35,
35,
37,
37,
38,
38,
41,
41,
42,
42,
44,
44,
47,
47,
49,
49,
50,
50,
52,
52,
55,
55,
56,
56,
59,
59,
61,
61,
62,
62,
64,
64,
67,
67,
69,
69,
70,
70,
73,
73,
74,
74,
76,
76,
79,
79,
81,
81,
82,
82,
84,
84,
87,
87,
88,
88,
91,
91,
93,
93,
94,
94,
97,
97,
98,
98,
100,
100,
103,
103,
104,
104,
107,
107,
109,
109,
110,
110,
112,
112,
115,
115,
117,
117,
118,
118,
121,
121,
122,
122,
124,
124,
127,
127,
128,
128,
131,
131,
133,
133,
134,
134,
137,
137,
138,
138,
140,
140,
143,
143,
145,
145,
146,
146,
148,
148,
151,
151,
152,
152,
155,
155,
157,
157,
158,
158,
161,
161,
162,
162,
164,
164,
167,
167,
168,
168,
171,
171,
173,
173,
174,
174,
176,
176,
179,
179,
181,
181,
182,
182,
185,
185,
186,
186,
188,
188,
191,
191,
193,
193,
194,
194,
196,
196,
199,
199,
200,
200,
203,
203,
205,
205,
206,
206,
208,
208,
211,
211,
213,
213,
214,
214,
217,
217,
218,
218,
220,
220,
223,
223,
224,
224,
227,
227,
229,
229,
230,
230,
233,
233,
234,
234,
236,
236,
239,
239,
241,
241,
242,
242,
244,
244,
247,
247,
248,
248,
251,
251,
253,
253,
254,
254,
]
# Permutation matrix for boot key
bootkey_perm_table = [
0x8,
0x5,
0x4,
0x2,
0xB,
0x9,
0xD,
0x3,
0x0,
0x6,
0x1,
0xC,
0xE,
0xA,
0xF,
0x7,
]
# Constants for SAM decrypt algorithm
aqwerty = b"!@#$%^&*()qwertyUIOPAzxcvbnmQQQQQQQQQQQQ)(*@&%\0"
anum = b"0123456789012345678901234567890123456789\0"
antpassword = b"NTPASSWORD\0"
almpassword = b"LMPASSWORD\0"
lmkey = b"KGS!@#$%"
empty_lm = b"\xaa\xd3\xb4\x35\xb5\x14\x04\xee\xaa\xd3\xb4\x35\xb5\x14\x04\xee"
empty_nt = b"\x31\xd6\xcf\xe0\xd1\x6a\xe9\x31\xb7\x3c\x59\xd7\xe0\xc0\x89\xc0"
@classmethod
def get_hive_key(
cls, hive: registry.RegistryHive, key: str
) -> Optional["registry.CM_KEY_NODE"]:
result = None
try:
if hive:
result = hive.get_key(key)
except (KeyError, registrylayer.RegistryException):
vollog.info(
f"Unable to load the required registry key {hive.get_name()}\\{key} from this memory image"
)
return result
@classmethod
def get_user_keys(
cls, samhive: registry.RegistryHive
) -> List[interfaces.objects.ObjectInterface]:
user_key_path = "SAM\\Domains\\Account\\Users"
user_key = cls.get_hive_key(samhive, user_key_path)
if not user_key:
return []
return [k for k in user_key.get_subkeys() if k.Name != "Names"]
@classmethod
def get_bootkey(cls, syshive: registry.RegistryHive) -> Optional[bytes]:
"""
Returns the scrambled bootkey necesary to decrypt hashes
"""
cs = 1
lsa_base = f"ControlSet{cs:03}" + "\\Control\\Lsa"
lsa_keys = ["JD", "Skew1", "GBG", "Data"]
lsa = cls.get_hive_key(syshive, lsa_base)
if not lsa:
return None
bootkey = ""
for lk in lsa_keys:
try:
key = cls.get_hive_key(syshive, lsa_base + "\\" + lk)
class_data = None
if key:
try:
class_data = syshive.read(key.Class + 4, key.ClassLength)
except exceptions.InvalidAddressException:
return None
if class_data is None:
return None
bootkey += class_data.decode("utf-16-le")
except (
InvalidAddressException,
registrylayer.RegistryException,
) as excp:
vollog.log(
constants.LOGLEVEL_VVV, f"Unable to read Lsa key {lk}: {excp}"
)
return None
bootkey_str = binascii.unhexlify(bootkey)
bootkey_scrambled = bytes(
[bootkey_str[cls.bootkey_perm_table[i]] for i in range(len(bootkey_str))]
)
return bootkey_scrambled
@classmethod
def get_hbootkey(
cls, samhive: registry.RegistryHive, bootkey: bytes
) -> Optional[bytes]:
sam_account_path = "SAM\\Domains\\Account"
if not bootkey:
return None
sam_account_key = cls.get_hive_key(samhive, sam_account_path)
if not sam_account_key:
return None
sam_data = None
for v in sam_account_key.get_values():
if v.get_name() == "F":
try:
sam_data = samhive.read(v.Data + 4, v.DataLength)
except exceptions.InvalidAddressException:
return None
if not sam_data:
return None
revision = sam_data[0x00]
if revision == 2:
md5 = hashlib.md5()
md5.update(sam_data[0x70:0x80] + cls.aqwerty + bootkey + cls.anum)
rc4_key = md5.digest()
rc4 = ARC4.new(rc4_key)
hbootkey = rc4.encrypt(
sam_data[0x80:0xA0]
) # lgtm [py/weak-cryptographic-algorithm]
return hbootkey
elif revision == 3:
# AES encrypted
iv = sam_data[0x78:0x88]
encryptedHBootKey = sam_data[0x88:0xA8]
cipher = AES.new(bootkey, AES.MODE_CBC, iv)
hbootkey = cipher.decrypt(encryptedHBootKey)
return hbootkey[:16]
return None
@classmethod
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)
cipher = AES.new(hbootkey[:16], AES.MODE_CBC, salt)
obfkey = cipher.decrypt(enc_hash)
return des1.decrypt(obfkey[:8]) + des2.decrypt(
obfkey[8:16]
) # lgtm [py/weak-cryptographic-algorithm]
@classmethod
def get_user_hashes(
cls, user: registry.CM_KEY_NODE, samhive: registry.RegistryHive, 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)
except ValueError:
return None
sam_data = None
for v in user.get_values():
if v.get_name() == "V":
try:
sam_data = samhive.read(v.Data + 4, v.DataLength)
except (
exceptions.InvalidAddressException,
registrylayer.RegistryException,
):
return None
if not sam_data:
return None
lm_offset = unpack("<L", sam_data[0x9C:0xA0])[0] + 0xCC
lm_len = unpack("<L", sam_data[0xA0:0xA4])[0]
nt_offset = unpack("<L", sam_data[0xA8:0xAC])[0] + 0xCC
nt_len = unpack("<L", sam_data[0xAC:0xB0])[0]
lm_revision = sam_data[lm_offset + 2 : lm_offset + 3]
lmhash = None
if lm_revision == b"\x01":
if lm_len == 20:
enc_lm_hash = sam_data[lm_offset + 0x04 : lm_offset + 0x14]
lmhash = cls.decrypt_single_hash(
rid, hbootkey, enc_lm_hash, cls.almpassword
)
elif lm_revision == b"\x02":
if lm_len == 56:
lm_salt = sam_data[lm_offset + 4 : lm_offset + 20]
enc_lm_hash = sam_data[lm_offset + 20 : lm_offset + 52]
lmhash = cls.decrypt_single_salted_hash(
rid, hbootkey, enc_lm_hash, cls.almpassword, lm_salt
)
# NT hash decryption
nthash = None
nt_revision = sam_data[nt_offset + 2 : nt_offset + 3]
if nt_revision == b"\x01":
if nt_len == 20:
enc_nt_hash = sam_data[nt_offset + 4 : nt_offset + 20]
nthash = cls.decrypt_single_hash(
rid, hbootkey, enc_nt_hash, cls.antpassword
)
elif nt_revision == b"\x02":
if nt_len == 56:
nt_salt = sam_data[nt_offset + 8 : nt_offset + 24]
enc_nt_hash = sam_data[nt_offset + 24 : nt_offset + 56]
nthash = cls.decrypt_single_salted_hash(
rid, hbootkey, enc_nt_hash, cls.antpassword, nt_salt
)
return lmhash, nthash
@classmethod
def sid_to_key(cls, sid: int) -> Tuple[bytes, bytes]:
"""Takes rid of a user and converts it to a key to be used by the DES cipher"""
bytestr1 = [
sid & 0xFF,
(sid >> 8) & 0xFF,
(sid >> 16) & 0xFF,
(sid >> 24) & 0xFF,
]
bytestr1 += bytestr1[0:3]
bytestr2 = [bytestr1[3]] + bytestr1[0:3]
bytestr2 += bytestr2[0:3]
return cls.sidbytes_to_key(bytes(bytestr1)), cls.sidbytes_to_key(
bytes(bytestr2)
)
@classmethod
def sidbytes_to_key(cls, s: bytes) -> bytes:
"""Builds final DES key from the strings generated in sid_to_key"""
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: 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)
md5 = hashlib.md5()
md5.update(hbootkey[:0x10] + pack("<L", rid) + lmntstr)
rc4_key = md5.digest()
rc4 = ARC4.new(rc4_key)
obfkey = rc4.encrypt(enc_hash) # lgtm [py/weak-cryptographic-algorithm]
return des1.decrypt(obfkey[:8]) + des2.decrypt(
obfkey[8:]
) # lgtm [py/weak-cryptographic-algorithm]
@classmethod
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":
try:
value = samhive.read(v.Data + 4, v.DataLength)
except exceptions.InvalidAddressException:
return None
if not value:
return None
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 = value[name_offset : name_offset + name_length]
return username
# replaces the dump_hashes method in vol2
def _generator(
self, syshive: registry.RegistryHive, samhive: registry.RegistryHive
):
if syshive is None:
vollog.debug("SYSTEM address is None: No system hive found")
if samhive is None:
vollog.debug("SAM address is None: No SAM hive found")
bootkey = self.get_bootkey(syshive)
hbootkey = self.get_hbootkey(samhive, bootkey)
if hbootkey:
for user in self.get_user_keys(samhive):
ret = self.get_user_hashes(user, samhive, hbootkey)
if ret:
lmhash, nthash = ret
## temporary fix to prevent UnicodeDecodeError backtraces
## however this can cause truncated user names as a result
name = self.get_user_name(user, samhive)
if name is None:
name = renderers.NotAvailableValue()
else:
name = str(name, "utf-16-le", errors="ignore")
lmout = str(binascii.hexlify(lmhash or self.empty_lm), "latin-1")
ntout = str(binascii.hexlify(nthash or self.empty_nt), "latin-1")
rid = int(str(user.get_name()), 16)
yield (0, (name, rid, lmout, ntout))
else:
vollog.warning("Hbootkey is not valid")
def run(self):
offset = self.config.get("offset", None)
syshive = None
samhive = None
for hive in hivelist.HiveList.list_hives(
context=self.context,
base_config_path=self.config_path,
kernel_module_name=self.config["kernel"],
hive_offsets=None if offset is None else [offset],
):
if hive.get_name().split("\\")[-1].upper() == "SYSTEM":
syshive = hive
if hive.get_name().split("\\")[-1].upper() == "SAM":
samhive = hive
return renderers.TreeGrid(
[("User", str), ("rid", int), ("lmhash", str), ("nthash", str)],
self._generator(syshive, samhive),
)
@@ -0,0 +1,257 @@
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
from struct import unpack
from typing import Optional
import hashlib
from Crypto.Cipher import ARC4, DES, AES
from volatility3.framework import interfaces, renderers, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.exceptions import InvalidAddressException
from volatility3.framework.layers import registry
from volatility3.framework.symbols.windows import versions
from volatility3.plugins.windows.registry import hashdump, hivelist
from volatility3.framework.renderers import format_hints
vollog = logging.getLogger(__name__)
class Lsadump(interfaces.plugins.PluginInterface):
"""Dumps lsa secrets from memory"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls):
return [
requirements.ModuleRequirement(
name="kernel",
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="hashdump", component=hashdump.Hashdump, version=(1, 1, 0)
),
requirements.VersionRequirement(
name="hivelist", component=hivelist.HiveList, version=(2, 0, 0)
),
]
@classmethod
def decrypt_aes(cls, secret: bytes, key: bytes) -> bytes:
"""
Based on code from http://lab.mediaservice.net/code/cachedump.rb
"""
sha = hashlib.sha256()
sha.update(key)
for _i in range(1, 1000 + 1):
sha.update(secret[28:60])
aeskey = sha.digest()
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]
if len(buf) < 16:
buf += (16 - len(buf)) * "\00"
data += aes.decrypt(buf)
return data
@classmethod
def get_lsa_key(
cls, sechive: registry.RegistryHive, bootkey: bytes, vista_or_later: bool
) -> Optional[bytes]:
if not bootkey:
return None
if vista_or_later:
policy_key = "PolEKList"
else:
policy_key = "PolSecretEncryptionKey"
enc_reg_key = hashdump.Hashdump.get_hive_key(sechive, "Policy\\" + policy_key)
if not enc_reg_key:
return None
enc_reg_value = next(enc_reg_key.get_values(), None)
if not enc_reg_value:
return None
try:
obf_lsa_key = sechive.read(enc_reg_value.Data + 4, enc_reg_value.DataLength)
except exceptions.InvalidAddressException:
return None
if not obf_lsa_key:
return None
if not vista_or_later:
md5 = hashlib.md5()
md5.update(bootkey)
for _i in range(1000):
md5.update(obf_lsa_key[60:76])
rc4key = md5.digest()
rc4 = ARC4.new(rc4key)
lsa_key = rc4.decrypt(
obf_lsa_key[12:60]
) # lgtm [py/weak-cryptographic-algorithm]
lsa_key = lsa_key[0x10:0x20]
else:
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: registry.RegistryHive,
name: str,
lsakey: bytes,
is_vista_or_later: bool,
) -> Optional[bytes]:
enc_secret_key = hashdump.Hashdump.get_hive_key(
sechive, "Policy\\Secrets\\" + name + "\\CurrVal"
)
secret = None
if enc_secret_key:
try:
enc_secret_value = next(enc_secret_key.get_values(), None)
except (
InvalidAddressException,
registry.RegistryException,
):
enc_secret_value = None
if enc_secret_value:
try:
enc_secret = sechive.read(
enc_secret_value.Data + 4, enc_secret_value.DataLength
)
except exceptions.InvalidAddressExceptions:
return None
if enc_secret:
if not is_vista_or_later:
secret = cls.decrypt_secret(enc_secret[0xC:], lsakey)
else:
secret = cls.decrypt_aes(enc_secret, lsakey)
return secret
@classmethod
def decrypt_secret(cls, secret: bytes, key: bytes) -> bytes:
"""Python implementation of SystemFunction005.
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
for i in range(0, len(secret), 8):
enc_block = secret[i : i + 8]
block_key = key[j : j + 7]
des_key = hashdump.Hashdump.sidbytes_to_key(block_key)
des = DES.new(des_key, DES.MODE_ECB)
enc_block = enc_block + b"\x00" * int(abs(8 - len(enc_block)) % 8)
decrypted_data += des.decrypt(
enc_block
) # lgtm [py/weak-cryptographic-algorithm]
j += 7
if len(key[j : j + 7]) < 7:
j = len(key[j : j + 7])
(dec_data_len,) = unpack("<L", decrypted_data[:4])
return decrypted_data[8 : 8 + dec_data_len]
def _generator(
self, syshive: registry.RegistryHive, sechive: registry.RegistryHive
):
kernel = self.context.modules[self.config["kernel"]]
vista_or_later = versions.is_vista_or_later(
context=self.context, symbol_table=kernel.symbol_table_name
)
bootkey = hashdump.Hashdump.get_bootkey(syshive)
if not bootkey:
vollog.warning("Unable to find bootkey")
return None
lsakey = self.get_lsa_key(sechive, bootkey, vista_or_later)
if not lsakey:
vollog.warning("Unable to find lsa key")
return None
secrets_key = hashdump.Hashdump.get_hive_key(sechive, "Policy\\Secrets")
if not secrets_key:
vollog.warning("Unable to find secrets key")
return None
for key in secrets_key.get_subkeys():
sec_val_key = hashdump.Hashdump.get_hive_key(
sechive,
"Policy\\Secrets\\" + key.get_key_path().split("\\")[3] + "\\CurrVal",
)
if not sec_val_key:
continue
try:
enc_secret_value = next(sec_val_key.get_values(), None)
except (
StopIteration,
InvalidAddressException,
registry.RegistryException,
):
enc_secret_value = None
if not enc_secret_value:
continue
try:
enc_secret = sechive.read(
enc_secret_value.Data + 4, enc_secret_value.DataLength
)
except exceptions.InvalidAddressExceptions:
continue
if not vista_or_later:
secret = self.decrypt_secret(enc_secret[0xC:], lsakey)
else:
secret = self.decrypt_aes(enc_secret, lsakey)
try:
key_name = key.get_name()
except (
InvalidAddressException,
registry.RegistryException,
):
key_name = renderers.UnreadableValue()
yield (0, (key_name, format_hints.HexBytes(secret), secret))
def run(self):
offset = self.config.get("offset", None)
syshive = sechive = None
for hive in hivelist.HiveList.list_hives(
context=self.context,
base_config_path=self.config_path,
kernel_module_name=self.config["kernel"],
hive_offsets=None if offset is None else [offset],
):
if hive.get_name().split("\\")[-1].upper() == "SYSTEM":
syshive = hive
if hive.get_name().split("\\")[-1].upper() == "SECURITY":
sechive = hive
return renderers.TreeGrid(
[("Key", str), ("Secret", format_hints.HexBytes), ("Hex", bytes)],
self._generator(syshive, sechive),
)
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff