Merge pull request #1500 from volatilityfoundation/issue_1469_cachedump_exception

Exception handling in registry-related code

volatility3/framework/constants/_version.py

@@ -1,6 +1,6 @@
 # We use the SemVer 2.0.0 versioning scheme
 VERSION_MAJOR = 2  # Number of releases of the library with a breaking change
-VERSION_MINOR = 14  # Number of changes that only add to the interface
+VERSION_MINOR = 15  # Number of changes that only add to the interface
 VERSION_PATCH = 0  # Number of changes that do not change the interface
 VERSION_SUFFIX = ""
 

volatility3/framework/plugins/windows/amcache.py

@@ -543,7 +543,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                     amcache.get_key("Root\\InventoryDriverBinary")  # type: ignore
                 )
             )
-        except KeyError:
+        except (KeyError, registry.RegistryFormatException):
             # Registry key not found
             pass
 
@@ -554,7 +554,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                     amcache.get_key("Root\\Programs")
                 )  # type: ignore
             }
-        except KeyError:
+        except (KeyError, registry.RegistryFormatException):
             programs = {}
 
         try:
@@ -564,7 +564,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                 ),
                 key=_entry_sort_key,
             )
-        except KeyError:
+        except (KeyError, registry.RegistryFormatException):
             files = []
 
         for program_id, file_entries in itertools.groupby(
@@ -593,7 +593,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                     amcache.get_key("Root\\InventoryApplication")  # type: ignore
                 )
             )
-        except KeyError:
+        except (KeyError, registry.RegistryFormatException):
             programs = {}
 
         try:
@@ -603,7 +603,7 @@ class Amcache(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                 ),
                 key=_entry_sort_key,
             )
-        except KeyError:
+        except (KeyError, registry.RegistryFormatException):
             files = []
 
         for program_id, file_entries in itertools.groupby(

volatility3/framework/plugins/windows/cachedump.py

@@ -8,7 +8,7 @@ from typing import Tuple
 from Crypto.Cipher import ARC4, AES
 from Crypto.Hash import HMAC
 
-from volatility3.framework import interfaces, renderers
+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
@@ -140,9 +140,14 @@ class Cachedump(interfaces.plugins.PluginInterface):
             if cache_item.Name == "NL$Control":
                 continue
 
-            data = sechive.read(cache_item.Data + 4, cache_item.DataLength)
-            if data is None:
+            try:
+                data = sechive.read(cache_item.Data + 4, cache_item.DataLength)
+            except exceptions.InvalidAddressException:
                 continue
+
+            if not data:
+                continue
+
             (
                 uname_len,
                 domain_len,

volatility3/framework/plugins/windows/hashdump.py

@@ -332,7 +332,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
         try:
             if hive:
                 result = hive.get_key(key)
-        except KeyError:
+        except (KeyError, registry.RegistryFormatException):
             vollog.info(
                 f"Unable to load the required registry key {hive.get_name()}\\{key} from this memory image"
             )

volatility3/framework/plugins/windows/lsadump.py

@@ -8,7 +8,7 @@ from typing import Optional
 from Crypto.Cipher import ARC4, DES, AES
 from Crypto.Hash import MD5, SHA256
 
-from volatility3.framework import interfaces, renderers
+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
@@ -81,7 +81,10 @@ class Lsadump(interfaces.plugins.PluginInterface):
         if not enc_reg_value:
             return None
 
-        obf_lsa_key = sechive.read(enc_reg_value.Data + 4, enc_reg_value.DataLength)
+        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

volatility3/framework/symbols/windows/extensions/registry.py

@@ -159,14 +159,20 @@ class CM_KEY_NODE(objects.StructType):
     """Extension to allow traversal of registry keys."""
 
     def get_volatile(self) -> bool:
+        """
+        Returns a bool indicating whether or not the key is volatile.
+
+        Raises TypeError if the key was not instantiated on a RegistryHive layer
+        """
         if not isinstance(self._context.layers[self.vol.layer_name], RegistryHive):
-            raise ValueError(
-                "Cannot determine volatility of registry key without an offset in a RegistryHive layer"
-            )
+            raise TypeError("CM_KEY_NODE was not instantiated on a RegistryHive layer")
         return bool(self.vol.offset & 0x80000000)
 
     def get_subkeys(self) -> Iterator["CM_KEY_NODE"]:
-        """Returns a list of the key nodes."""
+        """Returns a list of the key nodes.
+
+        Raises TypeError if the key was not instantiated on a RegistryHive layer
+        """
         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")
@@ -222,7 +228,10 @@ class CM_KEY_NODE(objects.StructType):
                     yield from self._get_subkeys_recursive(hive, subnode)
 
     def get_values(self) -> Iterator["CM_KEY_VALUE"]:
-        """Returns a list of the Value nodes for a key."""
+        """Returns a list of the Value nodes for a key.
+
+        Raises TypeError if the key was not instantiated on a RegistryHive layer
+        """
         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")
@@ -251,6 +260,11 @@ class CM_KEY_NODE(objects.StructType):
         return self.Name.cast("string", max_length=namelength, encoding="latin-1")
 
     def get_key_path(self) -> str:
+        """
+        Returns the full path to this registry key.
+
+        Raises TypeError if the key was not instantiated on a RegistryHive layer
+        """
         reg = self._context.layers[self.vol.layer_name]
         if not isinstance(reg, RegistryHive):
             raise TypeError("Key was not instantiated on a RegistryHive layer")
@@ -276,7 +290,16 @@ class CM_KEY_VALUE(objects.StructType):
         return RegValueTypes(self.Type)
 
     def decode_data(self) -> Union[int, bytes]:
-        """Properly decodes the data associated with the value node"""
+        """
+        Properly decodes the data associated with the value node.
+
+        If an InvalidAddressException occurs when reading data from the
+        underlying RegistryHive layer, the data will be padded with null bytes
+        of the same length.
+
+        Raises ValueError if the data cannot be read
+        Raises TypeError if the class was not instantiated on a RegistryHive layer
+        """
         # Determine if the data is stored inline
         datalen = self.DataLength
         data = b""
@@ -310,14 +333,26 @@ class CM_KEY_VALUE(objects.StructType):
                     and block_offset < layer.maximum_address
                 ):
                     amount = min(BIG_DATA_MAXLEN, datalen)
-                    data += layer.read(
-                        offset=layer.get_cell(block_offset).vol.offset, length=amount
-                    )
+                    try:
+                        data += layer.read(
+                            offset=layer.get_cell(block_offset).vol.offset,
+                            length=amount,
+                        )
+                    except exceptions.InvalidAddressException:
+                        vollog.debug(
+                            f"Failed to read {amount:x} bytes of data, padding with {amount:x}"
+                        )
                     datalen -= amount
         else:
             # Suspect Data actually points to a Cell,
             # but the length at the start could be negative so just adding 4 to jump past it
-            data = layer.read(self.Data + 4, datalen)
+            try:
+                data = layer.read(self.Data + 4, datalen)
+            except exceptions.InvalidAddressException:
+                vollog.debug(
+                    f"Failed to read {datalen:x} bytes of data, returning {datalen:x} null bytes"
+                )
+                data = b"\x00" * datalen
 
         if self.get_type() == RegValueTypes.REG_DWORD:
             if len(data) != struct.calcsize("<L"):