#1476 - introduce RegistryException for simpler exception handling

This commit is contained in:
Dave Lassalle
2025-03-11 16:04:06 -05:00
parent 7f7295cdb7
commit 95aee9a664
13 changed files with 69 additions and 99 deletions
+7 -3
View File
@@ -19,11 +19,15 @@ from volatility3.plugins.windows import pslist
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."""
@@ -142,7 +146,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})"
)
@@ -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(
@@ -73,13 +73,11 @@ class Envars(interfaces.plugins.PluginInterface):
)
except (
KeyError,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
with contextlib.suppress(
KeyError,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
sys = hive.get_key(
"ControlSet001\\Control\\Session Manager\\Environment"
@@ -87,8 +85,7 @@ class Envars(interfaces.plugins.PluginInterface):
if sys:
with contextlib.suppress(
KeyError,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
for node in sys.get_values():
try:
@@ -97,8 +94,7 @@ class Envars(interfaces.plugins.PluginInterface):
values.append(value_node_name)
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
vollog.log(
constants.LOGLEVEL_VVV,
@@ -110,15 +106,13 @@ class Envars(interfaces.plugins.PluginInterface):
## The user-specific variables
with contextlib.suppress(
KeyError,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
ntuser = hive.get_key("Environment")
if ntuser:
with contextlib.suppress(
KeyError,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
for node in ntuser.get_values():
try:
@@ -127,8 +121,7 @@ class Envars(interfaces.plugins.PluginInterface):
values.append(value_node_name)
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
vollog.log(
constants.LOGLEVEL_VVV,
@@ -141,8 +134,7 @@ class Envars(interfaces.plugins.PluginInterface):
key = hive.get_key("Volatile Environment")
except (
KeyError,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
continue
try:
@@ -153,8 +145,7 @@ class Envars(interfaces.plugins.PluginInterface):
values.append(value_node_name)
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
vollog.log(
constants.LOGLEVEL_VVV,
@@ -88,16 +88,14 @@ class GetServiceSIDs(interfaces.plugins.PluginInterface):
except (
KeyError,
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
try:
services = hive.get_key(r"ControlSet001\Services")
except (
KeyError,
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
continue
@@ -107,8 +105,7 @@ class GetServiceSIDs(interfaces.plugins.PluginInterface):
sid_name = s.get_name()
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
continue
@@ -116,8 +116,7 @@ class GetSIDs(interfaces.plugins.PluginInterface):
sid = str(subkey.get_name())
except (
exceptions.InvalidAddressException,
layers.registry.RegistryFormatException,
layers.registry.RegistryInvalidIndex,
layers.registry.RegistryException,
):
continue
@@ -127,8 +126,7 @@ class GetSIDs(interfaces.plugins.PluginInterface):
value_node_name = node.get_name() or "(Default)"
except (
exceptions.InvalidAddressException,
layers.registry.RegistryFormatException,
layers.registry.RegistryInvalidIndex,
layers.registry.RegistryException,
):
continue
try:
@@ -162,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
@@ -12,6 +12,7 @@ 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
@@ -334,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"
)
@@ -382,8 +383,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
bootkey += class_data.decode("utf-16-le")
except (
InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registrylayer.RegistryException,
) as excp:
vollog.log(
constants.LOGLEVEL_VVV, f"Unable to read Lsa key {lk}: {excp}"
@@ -468,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, registry.RegistryHive):
except (
exceptions.InvalidAddressException,
registrylayer.RegistryException,
):
return None
if not sam_data:
@@ -125,8 +125,7 @@ class Lsadump(interfaces.plugins.PluginInterface):
enc_secret_value = next(enc_secret_key.get_values(), None)
except (
InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
enc_secret_value = None
@@ -209,8 +208,7 @@ class Lsadump(interfaces.plugins.PluginInterface):
except (
StopIteration,
InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
enc_secret_value = None
@@ -233,8 +231,7 @@ class Lsadump(interfaces.plugins.PluginInterface):
key_name = key.get_name()
except (
InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
key_name = renderers.UnreadableValue()
@@ -12,7 +12,7 @@ from volatility3.framework.layers.registry import (
RegistryHive,
RegistryFormatException,
InvalidAddressException,
RegistryInvalidIndex,
RegistryException,
)
from volatility3.framework.renderers import TreeGrid, conversion, format_hints
from volatility3.framework.symbols.windows.extensions.registry import RegValueTypes
@@ -88,8 +88,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
key_path_names.append(k.get_name())
except (
InvalidAddressException,
RegistryFormatException,
RegistryInvalidIndex,
RegistryException,
):
key_path_names.append("-")
key_path = "\\".join([k for k in key_path_names])
@@ -117,8 +116,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
key_node.get_name()
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryInvalidIndex,
RegistryException,
) as excp:
vollog.debug(excp)
continue
@@ -168,8 +166,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
key_node_name = node.get_name()
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryInvalidIndex,
RegistryException,
) as excp:
vollog.debug(excp)
key_node_name = renderers.UnreadableValue()
@@ -196,8 +193,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
value_node_name = node.get_name() or "(Default)"
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryInvalidIndex,
RegistryException,
) as excp:
vollog.debug(excp)
value_node_name = renderers.UnreadableValue()
@@ -206,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()
@@ -241,7 +237,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
except (
ValueError,
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryException,
) as excp:
vollog.debug(excp)
value_data = renderers.UnreadableValue()
@@ -283,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(
@@ -15,8 +15,7 @@ from volatility3.framework.configuration import requirements
from volatility3.framework.layers.physical import BufferDataLayer
from volatility3.framework.layers.registry import (
RegistryHive,
RegistryFormatException,
RegistryInvalidIndex,
RegistryException,
)
from volatility3.framework.renderers import conversion, format_hints
from volatility3.framework.symbols import intermed
@@ -176,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}"
)
@@ -246,8 +245,7 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
subkey_name = subkey.get_name()
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryInvalidIndex,
RegistryException,
):
subkey_name = renderers.UnreadableValue()
@@ -276,8 +274,7 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
value_name = value.get_name()
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryInvalidIndex,
RegistryException,
):
value_name = renderers.UnreadableValue()
@@ -313,8 +313,7 @@ def _build_guid_name_map(key: reg_extensions.CM_KEY_NODE) -> Dict[str, str]:
break
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
):
continue
@@ -334,8 +333,7 @@ def _build_guid_name_map(key: reg_extensions.CM_KEY_NODE) -> Dict[str, str]:
)
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryInvalidIndex,
registry.RegistryException,
) as excp:
vollog.debug(f"Exception occurred while decoding id_str: {excp}")
@@ -1221,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
@@ -1243,8 +1241,7 @@ information about triggers, actions, run times, and creation times."""
name = str(value.get_name())
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryFormatException,
registry.RegistryException,
):
continue
@@ -1255,8 +1252,7 @@ information about triggers, actions, run times, and creation times."""
key_name = str(key.get_name())
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryFormatException,
registry.RegistryException,
):
key_name = None
@@ -1264,8 +1260,7 @@ information about triggers, actions, run times, and creation times."""
task_name = guid_mapping.get(key_name, renderers.NotAvailableValue())
except (
exceptions.InvalidAddressException,
registry.RegistryFormatException,
registry.RegistryFormatException,
registry.RegistryException,
):
task_name = renderers.NotAvailableValue()
@@ -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,
@@ -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, RegistryInvalidIndex
AttributeError, exceptions.InvalidAddressException, RegistryException
):
name = getattr(self, attr)
if name.Length > 0:
@@ -201,8 +200,7 @@ class CM_KEY_NODE(objects.StructType):
signature = node.cast("string", max_length=2, encoding="latin-1")
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryInvalidIndex,
RegistryException,
):
return None
@@ -231,8 +229,7 @@ class CM_KEY_NODE(objects.StructType):
subnode = hive.get_node(subnode_offset)
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryInvalidIndex,
RegistryException,
):
vollog.log(
constants.LOGLEVEL_VVV,
@@ -258,11 +255,7 @@ class CM_KEY_NODE(objects.StructType):
if v != 0:
try:
node = hive.get_node(v)
except (
RegistryInvalidIndex,
RegistryFormatException,
RegistryInvalidIndex,
) as excp:
except (RegistryException,) as excp:
vollog.debug(f"Invalid address {excp}")
continue
if isinstance(node, CM_KEY_VALUE):
@@ -270,8 +263,7 @@ class CM_KEY_NODE(objects.StructType):
except (
exceptions.InvalidAddressException,
RegistryFormatException,
RegistryInvalidIndex,
RegistryException,
) as excp:
vollog.debug(f"Invalid address in get_values iteration: {excp}")
return None
@@ -361,7 +353,7 @@ class CM_KEY_VALUE(objects.StructType):
offset=layer.get_cell(block_offset).vol.offset,
length=amount,
)
except (exceptions.InvalidAddressException, RegistryInvalidIndex):
except (exceptions.InvalidAddressException, RegistryException):
vollog.debug(
f"Failed to read {amount:x} bytes of data, padding with {amount:x}"
)
@@ -371,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, RegistryInvalidIndex):
except (exceptions.InvalidAddressException, RegistryException):
vollog.debug(
f"Failed to read {datalen:x} bytes of data, returning {datalen:x} null bytes"
)
@@ -80,7 +80,7 @@ class Certificates(interfaces.plugins.PluginInterface):
]:
with contextlib.suppress(
KeyError,
registry.RegistryFormatException,
registry.RegistryException,
exceptions.InvalidAddressException,
):
# Walk it