WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-11 04:07:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1701 from volatilityfoundation/1476-all-calls-to-get_name-on-registry-keys-need-auditing

Browse Source 
1476 all calls to get name on registry keys need auditing
This commit is contained in:
ikelos
2025-03-11 22:53:13 +00:00
committed by GitHub
parent aff84858ca 95aee9a664
commit e2d48b8018
13 changed files with 222 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/framework/layers/registry.py
+8 -4
View File
@@ -19,11 +19,15 @@ from volatility3.framework.symbols.windows import extensions
vollog = logging.getLogger(__name__)
class RegistryFormatException(exceptions.LayerException):
class RegistryException(exceptions.LayerException):
"""Base Registry Exception class for catching Registry layer errors."""
class RegistryFormatException(RegistryException):
"""Thrown when an error occurs with the underlying Registry file format."""
class RegistryInvalidIndex(exceptions.LayerException):
class RegistryInvalidIndex(RegistryException):
"""Thrown when an index that doesn't exist or can't be found occurs."""
@@ -176,7 +180,7 @@ class RegistryHive(linear.LinearlyMappedLayer):
cell = self.get_cell(cell_offset)
try:
signature = cell.cast("string", max_length=2, encoding="latin-1")
except (RegistryInvalidIndex, exceptions.InvalidAddressException):
except (RegistryException, exceptions.InvalidAddressException):
vollog.debug(
f"Failed to get cell signature for cell (0x{cell.vol.offset:x})"
)
@@ -296,7 +300,7 @@ class RegistryHive(linear.LinearlyMappedLayer):
self.name,
hex(offset & 0x7FFFFFFF),
hex(self._get_hive_maxaddr(volatile)),
"volative" if volatile else "non-volatile",
"volatile" if volatile else "non-volatile",
self.get_name(),
),
)
volatility3/framework/plugins/windows/amcache.py
+5 -5
View File
@@ -544,7 +544,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
amcache.get_key("Root\\InventoryDriverBinary") # type: ignore
)
)
except (KeyError, registry.RegistryFormatException):
except (KeyError, registry.RegistryException):
# Registry key not found
pass
@@ -555,7 +555,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
amcache.get_key("Root\\Programs")
) # type: ignore
}
except (KeyError, registry.RegistryFormatException):
except (KeyError, registry.RegistryException):
programs = {}
try:
@@ -565,7 +565,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
),
key=_entry_sort_key,
)
except (KeyError, registry.RegistryFormatException):
except (KeyError, registry.RegistryException):
files = []
for program_id, file_entries in itertools.groupby(
@@ -594,7 +594,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
amcache.get_key("Root\\InventoryApplication") # type: ignore
)
)
except (KeyError, registry.RegistryFormatException):
except (KeyError, registry.RegistryException):
programs = {}
try:
@@ -604,7 +604,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
),
key=_entry_sort_key,
)
except (KeyError, registry.RegistryFormatException):
except (KeyError, registry.RegistryException):
files = []
for program_id, file_entries in itertools.groupby(
volatility3/framework/plugins/windows/envars.py
+27 -9
View File
@@ -71,13 +71,22 @@ class Envars(interfaces.plugins.PluginInterface):
sys = hive.get_key(
"CurrentControlSet\\Control\\Session Manager\\Environment"
)
except (KeyError, registry.RegistryFormatException):
with contextlib.suppress(KeyError, registry.RegistryFormatException):
except (
KeyError,
registry.RegistryException,
):
with contextlib.suppress(
KeyError,
registry.RegistryException,
):
sys = hive.get_key(
"ControlSet001\\Control\\Session Manager\\Environment"
)
if sys:
with contextlib.suppress(KeyError, registry.RegistryFormatException):
with contextlib.suppress(
KeyError,
registry.RegistryException,
):
for node in sys.get_values():
try:
value_node_name = node.get_name()
@@ -85,7 +94,7 @@ class Envars(interfaces.plugins.PluginInterface):
values.append(value_node_name)
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryException,
):
vollog.log(
constants.LOGLEVEL_VVV,
@@ -95,10 +104,16 @@ class Envars(interfaces.plugins.PluginInterface):
ntuser = None
## The user-specific variables
with contextlib.suppress(KeyError, registry.RegistryFormatException):
with contextlib.suppress(
KeyError,
registry.RegistryException,
):
ntuser = hive.get_key("Environment")
if ntuser:
with contextlib.suppress(KeyError, registry.RegistryFormatException):
with contextlib.suppress(
KeyError,
registry.RegistryException,
):
for node in ntuser.get_values():
try:
value_node_name = node.get_name()
@@ -106,7 +121,7 @@ class Envars(interfaces.plugins.PluginInterface):
values.append(value_node_name)
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryException,
):
vollog.log(
constants.LOGLEVEL_VVV,
@@ -117,7 +132,10 @@ class Envars(interfaces.plugins.PluginInterface):
## The volatile user variables
try:
key = hive.get_key("Volatile Environment")
except (KeyError, registry.RegistryFormatException):
except (
KeyError,
registry.RegistryException,
):
continue
try:
for node in key.get_values():
@@ -127,7 +145,7 @@ class Envars(interfaces.plugins.PluginInterface):
values.append(value_node_name)
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryException,
):
vollog.log(
constants.LOGLEVEL_VVV,
volatility3/framework/plugins/windows/getservicesids.py
+13 -5
View File
@@ -88,22 +88,30 @@ class GetServiceSIDs(interfaces.plugins.PluginInterface):
except (
KeyError,
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryException,
):
try:
services = hive.get_key(r"ControlSet001\Services")
except (
KeyError,
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryException,
):
continue
if services:
for s in services.get_subkeys():
if s.get_name() not in self.servicesids.values():
sid = createservicesid(s.get_name())
yield (0, (sid, s.get_name()))
try:
sid_name = s.get_name()
except (
exceptions.InvalidAddressException,
registry.RegistryException,
):
continue
if sid_name not in self.servicesids.values():
sid = createservicesid(sid_name)
yield (0, (sid, sid_name))
def run(self):
return renderers.TreeGrid([("SID", str), ("Service", str)], self._generator())
volatility3/framework/plugins/windows/getsids.py
+11 -4
View File
@@ -112,14 +112,21 @@ class GetSIDs(interfaces.plugins.PluginInterface):
):
try:
for subkey in hive.get_key(key).get_subkeys():
sid = str(subkey.get_name())
try:
sid = str(subkey.get_name())
except (
exceptions.InvalidAddressException,
layers.registry.RegistryException,
):
continue
path = ""
for node in subkey.get_values():
try:
value_node_name = node.get_name() or "(Default)"
except (
exceptions.InvalidAddressException,
layers.registry.RegistryFormatException,
layers.registry.RegistryException,
):
continue
try:
@@ -153,13 +160,13 @@ class GetSIDs(interfaces.plugins.PluginInterface):
except (
ValueError,
exceptions.InvalidAddressException,
layers.registry.RegistryFormatException,
layers.registry.RegistryException,
):
continue
except (
KeyError,
exceptions.InvalidAddressException,
layers.registry.RegistryFormatException,
layers.registry.RegistryException,
):
continue
volatility3/framework/plugins/windows/hashdump.py
+26 -13
View File
@@ -9,8 +9,10 @@ from typing import List, Optional, Tuple
from Crypto.Cipher import AES, ARC4, DES
from volatility3.framework import interfaces, renderers, exceptions
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
@@ -333,7 +335,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
try:
if hive:
result = hive.get_key(key)
except (KeyError, registry.RegistryFormatException):
except (KeyError, registrylayer.RegistryException):
vollog.info(
f"Unable to load the required registry key {hive.get_name()}\\{key} from this memory image"
)
@@ -361,24 +363,32 @@ class Hashdump(interfaces.plugins.PluginInterface):
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:
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
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:
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 += class_data.decode("utf-16-le")
bootkey_str = binascii.unhexlify(bootkey)
bootkey_scrambled = bytes(
@@ -458,7 +468,10 @@ class Hashdump(interfaces.plugins.PluginInterface):
if v.get_name() == "V":
try:
sam_data = samhive.read(v.Data + 4, v.DataLength)
except exceptions.InvalidAddressException:
except (
exceptions.InvalidAddressException,
registrylayer.RegistryException,
):
return None
if not sam_data:
volatility3/framework/plugins/windows/lsadump.py
+28 -3
View File
@@ -10,6 +10,8 @@ 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
@@ -119,7 +121,14 @@ class Lsadump(interfaces.plugins.PluginInterface):
secret = None
if enc_secret_key:
enc_secret_value = next(enc_secret_key.get_values(), None)
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(
@@ -194,7 +203,15 @@ class Lsadump(interfaces.plugins.PluginInterface):
if not sec_val_key:
continue
enc_secret_value = next(sec_val_key.get_values(), None)
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
@@ -210,7 +227,15 @@ class Lsadump(interfaces.plugins.PluginInterface):
else:
secret = self.decrypt_aes(enc_secret, lsakey)
yield (0, (key.get_name(), format_hints.HexBytes(secret), secret))
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)
volatility3/framework/plugins/windows/registry/printkey.py
+27 -9
View File
@@ -8,7 +8,12 @@ from typing import List, Optional, Sequence, Iterable, Tuple, Union
from volatility3.framework import objects, renderers, exceptions, interfaces, constants
from volatility3.framework.configuration import requirements
from volatility3.framework.layers.registry import RegistryHive, RegistryFormatException
from volatility3.framework.layers.registry import (
RegistryHive,
RegistryFormatException,
InvalidAddressException,
RegistryException,
)
from volatility3.framework.renderers import TreeGrid, conversion, format_hints
from volatility3.framework.symbols.windows.extensions.registry import RegValueTypes
from volatility3.plugins.windows.registry import hivelist
@@ -77,7 +82,17 @@ class PrintKey(interfaces.plugins.PluginInterface):
return None
node = node_path[-1]
key_path_items = [hive] + node_path[1:]
key_path = "\\".join([k.get_name() for k in key_path_items])
key_path_names = []
for k in key_path_items:
try:
key_path_names.append(k.get_name())
except (
InvalidAddressException,
RegistryException,
):
key_path_names.append("-")
key_path = "\\".join([k for k in key_path_names])
if node.vol.type_name.endswith(constants.BANG + "_CELL_DATA"):
raise RegistryFormatException(
hive.name, "Encountered _CELL_DATA instead of _CM_KEY_NODE"
@@ -99,7 +114,10 @@ class PrintKey(interfaces.plugins.PluginInterface):
if key_node.vol.offset not in [x.vol.offset for x in node_path]:
try:
key_node.get_name()
except exceptions.InvalidAddressException as excp:
except (
exceptions.InvalidAddressException,
RegistryException,
) as excp:
vollog.debug(excp)
continue
@@ -148,7 +166,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
key_node_name = node.get_name()
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryException,
) as excp:
vollog.debug(excp)
key_node_name = renderers.UnreadableValue()
@@ -175,7 +193,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
value_node_name = node.get_name() or "(Default)"
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryException,
) as excp:
vollog.debug(excp)
value_node_name = renderers.UnreadableValue()
@@ -184,7 +202,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
value_type = RegValueTypes(node.Type).name
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryException,
) as excp:
vollog.debug(excp)
value_type = renderers.UnreadableValue()
@@ -219,7 +237,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
except (
ValueError,
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryException,
) as excp:
vollog.debug(excp)
value_data = renderers.UnreadableValue()
@@ -261,13 +279,13 @@ class PrintKey(interfaces.plugins.PluginInterface):
except (
exceptions.InvalidAddressException,
KeyError,
RegistryFormatException,
RegistryException,
) as excp:
if isinstance(excp, KeyError):
vollog.debug(
f"Key '{key}' not found in Hive at offset {hex(hive.hive_offset)}."
)
elif isinstance(excp, RegistryFormatException):
elif isinstance(excp, RegistryException):
vollog.debug(excp)
elif isinstance(excp, exceptions.InvalidAddressException):
vollog.debug(
volatility3/framework/plugins/windows/registry/userassist.py
+21 -4
View File
@@ -13,7 +13,10 @@ from typing import Any, Generator, List, Tuple
from volatility3.framework import constants, exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.layers.physical import BufferDataLayer
from volatility3.framework.layers.registry import RegistryHive, RegistryFormatException
from volatility3.framework.layers.registry import (
RegistryHive,
RegistryException,
)
from volatility3.framework.renderers import conversion, format_hints
from volatility3.framework.symbols import intermed
from volatility3.plugins.windows.registry import hivelist
@@ -172,7 +175,7 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
"software\\microsoft\\windows\\currentversion\\explorer\\userassist",
return_list=True,
)
except RegistryFormatException as e:
except RegistryException as e:
vollog.warning(
f"Error accessing UserAssist key in {hive_name} at {hive.hive_offset:#x}: {e}"
)
@@ -238,7 +241,14 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
# output any subkeys under Count
for subkey in countkey.get_subkeys():
subkey_name = subkey.get_name()
try:
subkey_name = subkey.get_name()
except (
exceptions.InvalidAddressException,
RegistryException,
):
subkey_name = renderers.UnreadableValue()
result = (
1,
(
@@ -260,7 +270,14 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
# output any values under Count
for value in countkey.get_values():
value_name = value.get_name()
try:
value_name = value.get_name()
except (
exceptions.InvalidAddressException,
RegistryException,
):
value_name = renderers.UnreadableValue()
with contextlib.suppress(UnicodeDecodeError):
value_name = codecs.encode(value_name, "rot_13")
volatility3/framework/plugins/windows/scheduled_tasks.py
+36 -10
View File
@@ -311,7 +311,10 @@ def _build_guid_name_map(key: reg_extensions.CM_KEY_NODE) -> Dict[str, str]:
if value.get_name() == "Id":
task_id_value = value
break
except exceptions.InvalidAddressException:
except (
exceptions.InvalidAddressException,
registry.RegistryException,
):
continue
if (
@@ -323,10 +326,16 @@ def _build_guid_name_map(key: reg_extensions.CM_KEY_NODE) -> Dict[str, str]:
except exceptions.InvalidAddressException:
id_str = None
if isinstance(id_str, bytes):
mapping[id_str.decode("utf-16le", errors="replace").rstrip(NULL)] = str(
key.get_name()
)
try:
if isinstance(id_str, bytes):
mapping[id_str.decode("utf-16le", errors="replace").rstrip(NULL)] = str(
key.get_name()
)
except (
exceptions.InvalidAddressException,
registry.RegistryException,
) as excp:
vollog.debug(f"Exception occurred while decoding id_str: {excp}")
for subkey in key.get_subkeys():
mapping.update(_build_guid_name_map(subkey))
@@ -1210,14 +1219,14 @@ information about triggers, actions, run times, and creation times."""
task_key = software_hive.get_key(
"Microsoft\\Windows NT\\CurrentVersion\\Schedule\\TaskCache\\Tasks"
)
except (KeyError, registry.RegistryFormatException):
except (KeyError, registry.RegistryException):
task_key = None
try:
task_tree = software_hive.get_key(
"Microsoft\\Windows NT\\CurrentVersion\\Schedule\\TaskCache\\Tree"
)
except (KeyError, registry.RegistryFormatException):
except (KeyError, registry.RegistryException):
task_tree = None
return (task_key, task_tree) # type: ignore
@@ -1230,13 +1239,30 @@ information about triggers, actions, run times, and creation times."""
for value in key.get_values():
try:
name = str(value.get_name())
except exceptions.InvalidAddressException:
except (
exceptions.InvalidAddressException,
registry.RegistryException,
):
continue
if name in ["Actions", "Triggers", "DynamicInfo"]:
values[name] = value
task_name = guid_mapping.get(str(key.get_name()), renderers.NotAvailableValue())
try:
key_name = str(key.get_name())
except (
exceptions.InvalidAddressException,
registry.RegistryException,
):
key_name = None
try:
task_name = guid_mapping.get(key_name, renderers.NotAvailableValue())
except (
exceptions.InvalidAddressException,
registry.RegistryException,
):
task_name = renderers.NotAvailableValue()
try:
action_set = cls.parse_actions_value(values["Actions"])
@@ -1351,7 +1377,7 @@ information about triggers, actions, run times, and creation times."""
else renderers.NotAvailableValue()
),
working_directory,
str(key.get_name()),
key_name or renderers.NotAvailableValue(),
)
def _generator(self) -> Iterator[Tuple[int, _ScheduledTaskEntry]]:
volatility3/framework/plugins/windows/svcscan.py
+2 -2
View File
@@ -162,7 +162,7 @@ class SvcScan(interfaces.plugins.PluginInterface):
except (
KeyError,
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryException,
):
try:
return cast(
@@ -171,7 +171,7 @@ class SvcScan(interfaces.plugins.PluginInterface):
except (
KeyError,
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryException,
):
vollog.log(
constants.LOGLEVEL_VVVV,
volatility3/framework/symbols/windows/extensions/registry.py
+17 -11
View File
@@ -9,9 +9,8 @@ from typing import Iterator, Optional, Union, cast
from volatility3.framework import constants, exceptions, interfaces, objects
from volatility3.framework.layers.registry import (
RegistryFormatException,
RegistryException,
RegistryHive,
RegistryInvalidIndex,
)
vollog = logging.getLogger(__name__)
@@ -103,7 +102,7 @@ class CMHIVE(objects.StructType):
for attr in ["FileFullPath", "FileUserName", "HiveRootPath"]:
with contextlib.suppress(
AttributeError, exceptions.InvalidAddressException
AttributeError, exceptions.InvalidAddressException, RegistryException
):
name = getattr(self, attr)
if name.Length > 0:
@@ -199,7 +198,10 @@ class CM_KEY_NODE(objects.StructType):
# We could change the array type to a struct with both parts
try:
signature = node.cast("string", max_length=2, encoding="latin-1")
except (exceptions.InvalidAddressException, RegistryFormatException):
except (
exceptions.InvalidAddressException,
RegistryException,
):
return None
listjump = None
@@ -227,7 +229,7 @@ class CM_KEY_NODE(objects.StructType):
subnode = hive.get_node(subnode_offset)
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryException,
):
vollog.log(
constants.LOGLEVEL_VVV,
@@ -244,21 +246,25 @@ class CM_KEY_NODE(objects.StructType):
hive = self._context.layers[self.vol.layer_name]
if not isinstance(hive, RegistryHive):
raise TypeError("CM_KEY_NODE was not instantiated on a RegistryHive layer")
child_list = hive.get_cell(self.ValueList.List).u.KeyList
child_list.count = self.ValueList.Count
try:
child_list = hive.get_cell(self.ValueList.List).u.KeyList
child_list.count = self.ValueList.Count
for v in child_list:
if v != 0:
try:
node = hive.get_node(v)
except (RegistryInvalidIndex, RegistryFormatException) as excp:
except (RegistryException,) as excp:
vollog.debug(f"Invalid address {excp}")
continue
if isinstance(node, CM_KEY_VALUE):
yield node
except (exceptions.InvalidAddressException, RegistryFormatException) as excp:
except (
exceptions.InvalidAddressException,
RegistryException,
) as excp:
vollog.debug(f"Invalid address in get_values iteration: {excp}")
return None
@@ -347,7 +353,7 @@ class CM_KEY_VALUE(objects.StructType):
offset=layer.get_cell(block_offset).vol.offset,
length=amount,
)
except exceptions.InvalidAddressException:
except (exceptions.InvalidAddressException, RegistryException):
vollog.debug(
f"Failed to read {amount:x} bytes of data, padding with {amount:x}"
)
@@ -357,7 +363,7 @@ class CM_KEY_VALUE(objects.StructType):
# but the length at the start could be negative so just adding 4 to jump past it
try:
data = layer.read(self.Data + 4, datalen)
except exceptions.InvalidAddressException:
except (exceptions.InvalidAddressException, RegistryException):
vollog.debug(
f"Failed to read {datalen:x} bytes of data, returning {datalen:x} null bytes"
)
volatility3/plugins/windows/registry/certificates.py
+1 -1
View File
@@ -80,7 +80,7 @@ class Certificates(interfaces.plugins.PluginInterface):
]:
with contextlib.suppress(
KeyError,
registry.RegistryFormatException,
registry.RegistryException,
exceptions.InvalidAddressException,
):
# Walk it