Merge pull request #782 from volatilityfoundation/feature/context-manager-exceptions

Core: Convert try/except/pass to contextlib.supress

volatility3/framework/automagic/pdbscan.py

@@ -7,10 +7,11 @@ from loaded PE files.
 This module contains a standalone scanner, and also a :class:`~volatility3.framework.interfaces.layers.ScannerInterface`
 based scanner for use within the framework by calling :func:`~volatility3.framework.interfaces.layers.DataLayerInterface.scan`.
 """
+import contextlib
 import logging
 import math
 import os
-from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Union, Callable
+from typing import Any, Callable, Dict, Iterable, List, Optional, Set, Tuple, Union
 
 from volatility3.framework import constants, exceptions, interfaces, layers
 from volatility3.framework.configuration import requirements
@@ -139,7 +140,8 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
                          vlayer: layers.intel.Intel,
                          progress_callback: constants.ProgressCallback = None) -> Optional[ValidKernelType]:
 
-        def test_virtual_kernel(physical_layer_name, virtual_layer_name: str, kernel: Dict[str, Any]) -> Optional[ValidKernelType]:
+        def test_virtual_kernel(physical_layer_name, virtual_layer_name: str, kernel: Dict[str, Any]) -> Optional[
+            ValidKernelType]:
             # It seems the kernel is loaded at a fixed mapping (presumably because the memory manager hasn't started yet)
             if kernel['mz_offset'] is None or not isinstance(kernel['mz_offset'], int):
                 # Rule out kernels that couldn't find a suitable MZ header
@@ -159,7 +161,8 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
                              vlayer: layers.intel.Intel,
                              progress_callback: constants.ProgressCallback = None) -> Optional[ValidKernelType]:
 
-        def test_physical_kernel(physical_layer_name:str , virtual_layer_name: str, kernel: Dict[str, Any]) -> Optional[ValidKernelType]:
+        def test_physical_kernel(physical_layer_name: str, virtual_layer_name: str, kernel: Dict[str, Any]) -> Optional[
+            ValidKernelType]:
             # It seems the kernel is loaded at a fixed mapping (presumably because the memory manager hasn't started yet)
             if kernel['mz_offset'] is None or not isinstance(kernel['mz_offset'], int):
                 # Rule out kernels that couldn't find a suitable MZ header
@@ -274,7 +277,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
         kernel_pdb_names = [bytes(name + ".pdb", "utf-8") for name in constants.windows.KERNEL_MODULE_NAMES]
 
         virtual_layer_name = vlayer.name
-        try:
+        with contextlib.suppress(exceptions.InvalidAddressException):
             if vlayer.read(address, 0x2) == b'MZ':
                 res = list(
                     PDBUtility.pdbname_scan(ctx = context,
@@ -286,8 +289,6 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
                                             end = address + self.max_pdb_size))
                 if res:
                     valid_kernel = (virtual_layer_name, address, res[0])
-        except exceptions.InvalidAddressException:
-            pass
         return valid_kernel
 
     # List of methods to be run, in order, to determine the valid kernels

volatility3/framework/interfaces/objects.py

@@ -6,6 +6,7 @@ interpreted values of data from a layer."""
 import abc
 import collections
 import collections.abc
+import contextlib
 import logging
 from typing import Any, Dict, List, Mapping, Optional
 
@@ -187,11 +188,9 @@ class ObjectInterface(metaclass = abc.ABCMeta):
         """
         if self.has_member(member_name):
             # noinspection PyBroadException
-            try:
+            with contextlib.suppress(Exception):
                 _ = getattr(self, member_name)
                 return True
-            except Exception:
-                pass
         return False
 
     def has_valid_members(self, member_names: List[str]) -> bool:

volatility3/framework/layers/crash.py

@@ -1,6 +1,7 @@
 # This file is Copyright 2021 Volatility Foundation and licensed under the Volatility Software License 1.0
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
+import contextlib
 import logging
 import struct
 from typing import Tuple, Optional
@@ -202,11 +203,9 @@ class WindowsCrashDumpStacker(interfaces.automagic.StackerLayerInterface):
               layer_name: str,
               progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]:
         for layer in [WindowsCrashDump32Layer, WindowsCrashDump64Layer]:
-            try:
+            with contextlib.suppress(WindowsCrashDumpFormatException):
                 layer.check_header(context.layers[layer_name])
                 new_name = context.layers.free_layer_name(layer.__name__)
                 context.config[interfaces.configuration.path_join(new_name, "base_layer")] = layer_name
                 return layer(context, new_name, new_name)
-            except WindowsCrashDumpFormatException:
-                pass
         return None

volatility3/framework/layers/registry.py

@@ -1,7 +1,7 @@
 # This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
-
+import contextlib
 import logging
 from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union
 
@@ -92,11 +92,9 @@ class RegistryHive(linear.LinearlyMappedLayer):
     @property
     def root_cell_offset(self) -> int:
         """Returns the offset for the root cell in this hive."""
-        try:
+        with contextlib.suppress(InvalidAddressException):
             if self._base_block.Signature.cast("string", max_length = 4, encoding = "latin-1") == 'regf':
                 return self._base_block.RootCell
-        except InvalidAddressException:
-            pass
         return 0x20
 
     def get_cell(self, cell_offset: int) -> 'objects.StructType':
@@ -201,11 +199,11 @@ class RegistryHive(linear.LinearlyMappedLayer):
         if offset & 0x7fffffff > self._get_hive_maxaddr(volatile):
             vollog.log(constants.LOGLEVEL_VVV,
                        "Layer {} couldn't translate offset {}, greater than {} in {} store of {}".format(
-                               self.name,
-                               hex(offset & 0x7fffffff),
-                               hex(self._get_hive_maxaddr(volatile)),
-                               "volative" if volatile else "non-volatile",
-                               self.get_name()))
+                           self.name,
+                           hex(offset & 0x7fffffff),
+                           hex(self._get_hive_maxaddr(volatile)),
+                           "volative" if volatile else "non-volatile",
+                           self.get_name()))
             raise RegistryInvalidIndex(self.name, "Mapping request for value greater than maxaddr")
 
         storage = self.hive.Storage[volatile]
@@ -252,14 +250,13 @@ class RegistryHive(linear.LinearlyMappedLayer):
 
     def is_valid(self, offset: int, length: int = 1) -> bool:
         """Returns a boolean based on whether the offset is valid or not."""
-        try:
+        with contextlib.suppress(exceptions.InvalidAddressException):
             # Pass this to the lower layers for now
             return all([
                 self.context.layers[layer].is_valid(offset, length)
                 for (_, _, offset, length, layer) in self.mapping(offset, length)
             ])
-        except exceptions.InvalidAddressException:
-            return False
+        return False
 
     @property
     def minimum_address(self) -> int:

volatility3/framework/layers/resources.py

@@ -184,14 +184,12 @@ class ResourceAccessor(object):
             stop = False
             while not stop:
                 detected = None
-                try:
+                with contextlib.suppress(AttributeError, IOError):
                     # Detect the content
                     detected = magic.detect_from_fobj(curfile)
                     IMPORTED_MAGIC = True
                     # This is because python-magic and file provide a magic module
                     # Only file's python has magic.detect_from_fobj
-                except (AttributeError, IOError):
-                    pass
 
                 if detected:
                     if detected.mime_type == 'application/x-xz':

volatility3/framework/layers/vmware.py

@@ -1,14 +1,14 @@
 # This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
-
+import contextlib
 import logging
 import struct
 from typing import Any, Dict, List, Optional
 
-from volatility3.framework import interfaces, constants, exceptions
+from volatility3.framework import constants, exceptions, interfaces
 from volatility3.framework.configuration import requirements
-from volatility3.framework.layers import physical, segmented, resources
+from volatility3.framework.layers import physical, resources, segmented
 from volatility3.framework.symbols import native
 
 vollog = logging.getLogger(__name__)
@@ -87,13 +87,13 @@ class VmwareLayer(segmented.SegmentedLayer):
                                              offset = offset + name_len + 2 + (index * index_len),
                                              layer_name = self._meta_layer))
                 data_len = flags & 0x3f
-                
+
                 if data_len in [62, 63]:  # Handle special data sizes that indicate a longer data stream
                     data_len = 4 if version == 0 else 8
                     # Read the size of the data
                     data_size = self._context.object(self._choose_type(data_len),
-                                            layer_name = self._meta_layer,
-                                            offset = offset + 2 + name_len + (indices_len * index_len))
+                                                     layer_name = self._meta_layer,
+                                                     offset = offset + 2 + name_len + (indices_len * index_len))
                     # Skip two bytes of padding (as it seems?)
                     # Read the actual data
                     data = self._context.object("vmware!bytes",
@@ -113,9 +113,9 @@ class VmwareLayer(segmented.SegmentedLayer):
         if tags[("regionsCount", ())][1] == 0:
             raise VmwareFormatException(self.name, "VMware VMEM is not split into regions")
         for region in range(tags[("regionsCount", ())][1]):
-            offset = tags[("regionPPN", (region, ))][1] * self._page_size
-            mapped_offset = tags[("regionPageNum", (region, ))][1] * self._page_size
-            length = tags[("regionSize", (region, ))][1] * self._page_size
+            offset = tags[("regionPPN", (region,))][1] * self._page_size
+            mapped_offset = tags[("regionPageNum", (region,))][1] * self._page_size
+            length = tags[("regionSize", (region,))][1] * self._page_size
             self._segments.append((offset, mapped_offset, length, length))
 
     @property
@@ -153,23 +153,19 @@ class VmwareStacker(interfaces.automagic.StackerLayerInterface):
                                                                      current_layer_name)
 
             vmss_success = False
-            try:
+            with contextlib.suppress(IOError):
                 _ = resources.ResourceAccessor().open(vmss).read(10)
                 context.config[interfaces.configuration.path_join(current_config_path, "location")] = vmss
                 context.layers.add_layer(physical.FileLayer(context, current_config_path, current_layer_name))
                 vmss_success = True
-            except IOError:
-                pass
 
             vmsn_success = False
             if not vmss_success:
-                try:
+                with contextlib.suppress(IOError):
                     _ = resources.ResourceAccessor().open(vmsn).read(10)
                     context.config[interfaces.configuration.path_join(current_config_path, "location")] = vmsn
                     context.layers.add_layer(physical.FileLayer(context, current_config_path, current_layer_name))
                     vmsn_success = True
-                except IOError:
-                    pass
 
             vollog.log(constants.LOGLEVEL_VVVV, f"Metadata found: VMSS ({vmss_success}) or VMSN ({vmsn_success})")
 

volatility3/framework/plugins/linux/check_syscall.py

@@ -3,11 +3,11 @@
 #
 """A module containing a collection of plugins that produce data typically
 found in Linux's /proc file system."""
+import contextlib
 import logging
 from typing import List
 
-from volatility3.framework import exceptions, interfaces
-from volatility3.framework import renderers, constants
+from volatility3.framework import constants, exceptions, interfaces, renderers
 from volatility3.framework.configuration import requirements
 from volatility3.framework.interfaces import plugins
 from volatility3.framework.renderers import format_hints
@@ -40,11 +40,9 @@ class Check_syscall(plugins.PluginInterface):
 
         symbol_list = []
         for sn in vmlinux.symbols:
-            try:
+            with contextlib.suppress(exceptions.SymbolError):
                 # When requesting the symbol from the module, a full resolve is performed
                 symbol_list.append((vmlinux.get_symbol(sn).address, sn))
-            except exceptions.SymbolError:
-                pass
         sorted_symbols = sorted(symbol_list)
 
         sym_address = 0

volatility3/framework/plugins/windows/dlllist.py

@@ -1,18 +1,19 @@
 # This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
+import contextlib
 import datetime
 import logging
 import ntpath
 from typing import List, Optional, Type
 
-from volatility3.framework import exceptions, renderers, interfaces, constants
+from volatility3.framework import constants, exceptions, interfaces, renderers
 from volatility3.framework.configuration import requirements
-from volatility3.framework.renderers import format_hints, conversion
+from volatility3.framework.renderers import conversion, format_hints
 from volatility3.framework.symbols import intermed
 from volatility3.framework.symbols.windows.extensions import pe
 from volatility3.plugins import timeliner
-from volatility3.plugins.windows import pslist, info
+from volatility3.plugins.windows import info, pslist
 
 vollog = logging.getLogger(__name__)
 
@@ -28,7 +29,7 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
         # 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"]),
+                                           architectures = ["Intel32", "Intel64"]),
             requirements.VersionRequirement(name = 'pslist', component = pslist.PsList, version = (2, 0, 0)),
             requirements.VersionRequirement(name = 'info', component = info.Info, version = (1, 0, 0)),
             requirements.ListRequirement(name = 'pid',
@@ -107,12 +108,10 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
             for entry in proc.load_order_modules():
 
                 BaseDllName = FullDllName = renderers.UnreadableValue()
-                try:
+                with contextlib.suppress(exceptions.InvalidAddressException):
                     BaseDllName = entry.BaseDllName.get_string()
                     # We assume that if the BaseDllName points to an invalid buffer, so will FullDllName
                     FullDllName = entry.FullDllName.get_string()
-                except exceptions.InvalidAddressException:
-                    pass
 
                 if dll_load_time_field:
                     # Versions prior to 6.1 won't have the LoadTime attribute

volatility3/framework/plugins/windows/envars.py

@@ -1,9 +1,10 @@
 # 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 contextlib
 import logging
 from typing import List
 
-from volatility3.framework import renderers, interfaces, objects, exceptions, constants
+from volatility3.framework import constants, exceptions, interfaces, objects, renderers
 from volatility3.framework.configuration import requirements
 from volatility3.framework.layers import registry
 from volatility3.plugins.windows import pslist
@@ -23,7 +24,7 @@ class Envars(interfaces.plugins.PluginInterface):
         # 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"]),
+                                           architectures = ["Intel32", "Intel64"]),
             requirements.ListRequirement(name = 'pid',
                                          description = 'Filter on specific process IDs',
                                          element_type = int,
@@ -61,13 +62,11 @@ class Envars(interfaces.plugins.PluginInterface):
                 key = hive.get_key('CurrentControlSet\\Control\\Session Manager\\Environment')
                 sys = True
             except KeyError:
-                try:
+                with contextlib.suppress(KeyError):
                     key = hive.get_key('ControlSet001\\Control\\Session Manager\\Environment')
                     sys = True
-                except KeyError:
-                    pass
             if sys:
-                try:
+                with contextlib.suppress(KeyError):
                     for node in key.get_values():
                         try:
                             value_node_name = node.get_name()
@@ -78,17 +77,13 @@ class Envars(interfaces.plugins.PluginInterface):
                                 constants.LOGLEVEL_VVV,
                                 "Error while parsing global environment variables keys (some keys might be excluded)")
                             continue
-                except KeyError:
-                    pass
 
             ## The user-specific variables
-            try:
+            with contextlib.suppress(KeyError):
                 key = hive.get_key('Environment')
                 ntuser = True
-            except KeyError:
-                pass
             if ntuser:
-                try:
+                with contextlib.suppress(KeyError):
                     for node in key.get_values():
                         try:
                             value_node_name = node.get_name()
@@ -99,8 +94,6 @@ class Envars(interfaces.plugins.PluginInterface):
                                 constants.LOGLEVEL_VVV,
                                 "Error while parsing user environment variables keys (some keys might be excluded)")
                             continue
-                except KeyError:
-                    pass
 
             ## The volatile user variables
             try:

volatility3/framework/plugins/windows/mftscan.py

@@ -1,7 +1,7 @@
 # This file is Copyright 2022 Volatility Foundation and licensed under the Volatility Software License 1.0
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
-
+import contextlib
 import datetime
 import logging
 
@@ -56,7 +56,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
         # Scan the layer for Raw MFT records and parse the fields
         for offset, _rule_name, _name, _value in layer.scan(context = self.context,
                                                             scanner = yarascan.YaraScanner(rules = rules)):
-            try:
+            with contextlib.suppress(exceptions.PagedInvalidAddressException):
                 mft_record = self.context.object(mft_object, offset = offset, layer_name = layer.name)
                 # We will update this on each pass in the next loop and use it as the new offset.
                 attr_base_offset = mft_record.FirstAttrOffset
@@ -131,9 +131,6 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                                                       offset = offset + attr_base_offset,
                                                       layer_name = layer.name)
 
-            except exceptions.PagedInvalidAddressException:
-                pass
-
     def generate_timeline(self):
         for row in self._generator():
             _depth, row_data = row

volatility3/framework/plugins/windows/registry/userassist.py

@@ -3,17 +3,18 @@
 #
 
 import codecs
+import contextlib
 import datetime
 import json
 import logging
 import os
-from typing import Any, List, Tuple, Generator
+from typing import Any, Generator, List, Tuple
 
-from volatility3.framework import exceptions, renderers, constants, interfaces
+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
-from volatility3.framework.renderers import format_hints, conversion
+from volatility3.framework.renderers import conversion, format_hints
 from volatility3.framework.symbols import intermed
 from volatility3.plugins.windows.registry import hivelist
 
@@ -38,7 +39,7 @@ class UserAssist(interfaces.plugins.PluginInterface):
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
         return [
             requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
-                                                     architectures = ["Intel32", "Intel64"]),
+                                           architectures = ["Intel32", "Intel64"]),
             requirements.IntRequirement(name = 'offset', description = "Hive Offset", default = None, optional = True),
             requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0))
         ]
@@ -126,11 +127,9 @@ class UserAssist(interfaces.plugins.PluginInterface):
         hive_name = hive.hive.cast(kernel.symbol_table_name + constants.BANG + "_CMHIVE").get_name()
 
         if self._win7 is None:
-            try:
+            with contextlib.suppress(exceptions.SymbolError):
                 self._win7 = self._win7_or_later()
-            except exceptions.SymbolError:
                 # self._win7 will be None and only registry value rawdata will be output
-                pass
 
         self._determine_userassist_type()
 
@@ -163,7 +162,6 @@ class UserAssist(interfaces.plugins.PluginInterface):
 
                 # output any subkeys under Count
                 for subkey in countkey.get_subkeys():
-
                     subkey_name = subkey.get_name()
                     result = (1, (
                         renderers.format_hints.Hex(hive.hive_offset),
@@ -185,10 +183,8 @@ class UserAssist(interfaces.plugins.PluginInterface):
                 for value in countkey.get_values():
 
                     value_name = value.get_name()
-                    try:
+                    with contextlib.suppress(UnicodeDecodeError):
                         value_name = codecs.encode(value_name, "rot_13")
-                    except UnicodeDecodeError:
-                        pass
 
                     if self._win7:
                         guid = value_name.split("\\")[0]

volatility3/framework/renderers/conversion.py

@@ -1,7 +1,7 @@
 # This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
-
+import contextlib
 import datetime
 import ipaddress
 import socket
@@ -27,10 +27,8 @@ def unixtime_to_datetime(unixtime: int) -> Union[interfaces.renderers.BaseAbsent
     ret: Union[interfaces.renderers.BaseAbsentValue, datetime.datetime] = renderers.UnparsableValue()
 
     if unixtime > 0:
-        try:
+        with contextlib.suppress(ValueError):
             ret = datetime.datetime.utcfromtimestamp(unixtime)
-        except ValueError:
-            pass
 
     return ret
 

volatility3/framework/symbols/mac/extensions/__init__.py

@@ -1,19 +1,18 @@
 # This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
-
+import contextlib
+import logging
 from typing import Generator, Iterable, Optional, Set, Tuple
 
-import logging
-
-from volatility3.framework import constants, objects, renderers
-from volatility3.framework import exceptions, interfaces
+from volatility3.framework import constants, exceptions, interfaces, objects
 from volatility3.framework.objects import utility
 from volatility3.framework.renderers import conversion
 from volatility3.framework.symbols import generic
 
 vollog = logging.getLogger(__name__)
 
+
 class proc(generic.GenericIntelProcess):
 
     def get_task(self):
@@ -29,10 +28,8 @@ class proc(generic.GenericIntelProcess):
         if not isinstance(parent_layer, interfaces.layers.TranslationLayerInterface):
             raise TypeError("Parent layer is not a translation layer, unable to construct process layer")
 
-        try:
+        with contextlib.suppress(exceptions.InvalidAddressException):
             dtb = self.get_task().map.pmap.pm_cr3
-        except exceptions.InvalidAddressException:
-            return None
 
         if preferred_name is None:
             preferred_name = self.vol.layer_name + f"_Process{self.p_pid}"
@@ -41,10 +38,8 @@ class proc(generic.GenericIntelProcess):
         return self._add_process_layer(self._context, dtb, config_prefix, preferred_name)
 
     def get_map_iter(self) -> Iterable[interfaces.objects.ObjectInterface]:
-        try:
+        with contextlib.suppress(exceptions.InvalidAddressException):
             task = self.get_task()
-        except exceptions.InvalidAddressException:
-            return
 
         try:
             current_map = task.map.hdr.links.next
@@ -55,9 +50,9 @@ class proc(generic.GenericIntelProcess):
 
         for i in range(task.map.hdr.nentries):
             if (not current_map or
-                current_map.vol.offset in seen or
-                not self._context.layers[task.vol.native_layer_name].is_valid(current_map.dereference().vol.offset, current_map.dereference().vol.size)):
-
+                    current_map.vol.offset in seen or
+                    not self._context.layers[task.vol.native_layer_name].is_valid(current_map.dereference().vol.offset,
+                                                                                  current_map.dereference().vol.size)):
                 vollog.log(constants.LOGLEVEL_VVV, "Breaking process maps iteration due to invalid state.")
                 break
 
@@ -102,10 +97,8 @@ class fileglob(objects.StructType):
         if self.has_member("fg_type"):
             ret = self.fg_type
         elif self.fg_ops != 0:
-            try:
+            with contextlib.suppress(exceptions.InvalidAddressException):
                 ret = self.fg_ops.fo_type
-            except exceptions.InvalidAddressException:
-                pass
 
         if ret:
             ret = str(ret.description).replace("DTYPE_", "")
@@ -456,7 +449,7 @@ class queue_entry(objects.StructType):
         seen = set()
 
         for attr in ['next', 'prev']:
-            try:
+            with contextlib.suppress(exceptions.InvalidAddressException):
                 n = getattr(self, attr).dereference().cast(type_name)
 
                 while n is not None and n.vol.offset != list_head:
@@ -473,9 +466,6 @@ class queue_entry(objects.StructType):
 
                     n = getattr(n.member(attr = member_name), attr).dereference().cast(type_name)
 
-            except exceptions.InvalidAddressException:
-                pass
-
 
 class ifnet(objects.StructType):
 

volatility3/framework/symbols/windows/__init__.py

@@ -1,10 +1,11 @@
 # 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 contextlib
 
 from volatility3.framework.symbols import intermed
 from volatility3.framework.symbols.windows import extensions
-from volatility3.framework.symbols.windows.extensions import registry, pool, pe
+from volatility3.framework.symbols.windows.extensions import pe, pool, registry
 
 
 class WindowsKernelIntermedSymbols(intermed.IntermediateSymbolTable):
@@ -39,26 +40,23 @@ class WindowsKernelIntermedSymbols(intermed.IntermediateSymbolTable):
         self.set_type_class('_VACB', extensions.VACB)
         self.set_type_class('_POOL_TRACKER_BIG_PAGES', pool.POOL_TRACKER_BIG_PAGES)
         self.set_type_class('_IMAGE_DOS_HEADER', pe.IMAGE_DOS_HEADER)
-        
+
         # Might not necessarily defined in every version of windows
         self.optional_set_type_class('_IMAGE_NT_HEADERS', pe.IMAGE_NT_HEADERS)
         self.optional_set_type_class('_IMAGE_NT_HEADERS64', pe.IMAGE_NT_HEADERS)
 
         # This doesn't exist in very specific versions of windows
-        try:
+        with contextlib.suppress(ValueError):
             if self.get_type("_POOL_TRACKER_BIG_PAGES").has_member("PoolType"):
                 self.set_type_class('_POOL_HEADER', pool.POOL_HEADER_VISTA)
             else:
                 self.set_type_class('_POOL_HEADER', pool.POOL_HEADER)
-        except ValueError:
-            pass
 
         # these don't exist in windows XP
         self.optional_set_type_class('_MMADDRESS_NODE', extensions.MMVAD_SHORT)
-        
+
         # these were introduced starting in windows 8
         self.optional_set_type_class('_MM_AVL_NODE', extensions.MMVAD_SHORT)
-        
+
         # these were introduced starting in windows 7
         self.optional_set_type_class('_RTL_BALANCED_NODE', extensions.MMVAD_SHORT)
-        

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

@@ -3,6 +3,7 @@
 #
 
 import collections.abc
+import contextlib
 import datetime
 import functools
 import logging
@@ -305,7 +306,7 @@ class MMVAD(MMVAD_SHORT):
 
         file_name = renderers.NotApplicableValue()
 
-        try:
+        with contextlib.suppress(exceptions.InvalidAddressException):
             # this is for xp and 2003
             if self.has_member("ControlArea"):
                 filename_obj = self.ControlArea.FilePointer.FileName
@@ -318,9 +319,6 @@ class MMVAD(MMVAD_SHORT):
             if filename_obj.Length > 0:
                 file_name = filename_obj.get_string()
 
-        except exceptions.InvalidAddressException:
-            pass
-
         return file_name
 
 
@@ -364,6 +362,7 @@ class DEVICE_OBJECT(objects.StructType, pool.ExecutiveObject):
             yield device
             device = device.AttachedDevice.dereference()
 
+
 class DRIVER_OBJECT(objects.StructType, pool.ExecutiveObject):
     """A class for kernel driver objects."""
 
@@ -374,7 +373,7 @@ class DRIVER_OBJECT(objects.StructType, pool.ExecutiveObject):
 
     def get_devices(self) -> Generator[ObjectInterface, None, None]:
         """Enumerate the driver's device objects"""
-        device =  self.DeviceObject.dereference()
+        device = self.DeviceObject.dereference()
         while device:
             yield device
             device = device.NextDevice.dereference()
@@ -413,15 +412,11 @@ class FILE_OBJECT(objects.StructType, pool.ExecutiveObject):
         # this pointer needs to be checked against native_layer_name because the object may
         # be instantiated from a primary (virtual) layer or a memory (physical) layer.
         if self._context.layers[self.vol.native_layer_name].is_valid(self.DeviceObject):
-            try:
+            with contextlib.suppress(ValueError):
                 name = f"\\Device\\{self.DeviceObject.get_device_name()}"
-            except ValueError:
-                pass
 
-        try:
+        with contextlib.suppress(TypeError, exceptions.InvalidAddressException):
             name += self.FileName.String
-        except (TypeError, exceptions.InvalidAddressException):
-            pass
 
         return name
 
@@ -1114,12 +1109,10 @@ class SHARED_CACHE_MAP(objects.StructType):
         iterval = 0
         while (iterval < full_blocks) and (full_blocks <= 4):
             vacb_obj = self.InitialVacbs[iterval]
-            try:
+            with contextlib.suppress(exceptions.InvalidAddressException):
                 # Make sure that the SharedCacheMap member of the VACB points back to the parent object.
                 if vacb_obj.SharedCacheMap == self.vol.offset:
                     self.save_vacb(vacb_obj, vacb_list)
-            except exceptions.InvalidAddressException:
-                pass
             iterval += 1
 
         # We also have to account for the spill over data that is not found in the full blocks.

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

@@ -1,12 +1,14 @@
+import contextlib
 import functools
 import logging
 import struct
-from typing import Optional, Tuple, List, Dict, Union
+from typing import Dict, List, Optional, Tuple, Union
 
-from volatility3.framework import objects, interfaces, constants, symbols, exceptions, renderers
-from volatility3.framework.renderers import conversion
 from volatility3.plugins.windows.poolscanner import PoolConstraint
 
+from volatility3.framework import constants, exceptions, interfaces, objects, renderers, symbols
+from volatility3.framework.renderers import conversion
+
 vollog = logging.getLogger(__name__)
 
 
@@ -138,7 +140,7 @@ class POOL_HEADER(objects.StructType):
                     if addr - optional_headers_length >= padding_length > addr:
                         continue
 
-                    try:
+                    with contextlib.suppress(TypeError, exceptions.InvalidAddressException):
                         mem_object = self._context.object(symbol_table_name + constants.BANG + type_name,
                                                           layer_name = self.vol.layer_name,
                                                           offset = addr + body_offset + start_offset,
@@ -147,15 +149,13 @@ class POOL_HEADER(objects.StructType):
                         if mem_object.is_valid():
                             yield mem_object
 
-                    except (TypeError, exceptions.InvalidAddressException):
-                        pass
-
             # use the bottom up approach for windows 7 and earlier
             else:
                 type_size = self._context.symbol_space.get_type(symbol_table_name + constants.BANG + type_name).size
                 if constraint.additional_structures:
                     for additional_structure in constraint.additional_structures:
-                        type_size += self._context.symbol_space.get_type(symbol_table_name + constants.BANG + additional_structure).size
+                        type_size += self._context.symbol_space.get_type(
+                            symbol_table_name + constants.BANG + additional_structure).size
 
                 rounded_size = conversion.round(type_size, alignment, up = True)
 
@@ -164,11 +164,9 @@ class POOL_HEADER(objects.StructType):
                                                   offset = self.vol.offset + self.BlockSize * alignment - rounded_size,
                                                   native_layer_name = native_layer_name)
 
-                try:
+                with contextlib.suppress(TypeError, exceptions.InvalidAddressException):
                     if mem_object.is_valid():
                         yield mem_object
-                except (TypeError, exceptions.InvalidAddressException):
-                    pass
 
     @classmethod
     @functools.lru_cache()
@@ -177,20 +175,18 @@ class POOL_HEADER(objects.StructType):
         headers = []
         sizes = []
         for header in [
-                'CREATOR_INFO', 'NAME_INFO', 'HANDLE_INFO', 'QUOTA_INFO', 'PROCESS_INFO', 'AUDIT_INFO', 'EXTENDED_INFO',
-                'HANDLE_REVOCATION_INFO', 'PADDING_INFO'
+            'CREATOR_INFO', 'NAME_INFO', 'HANDLE_INFO', 'QUOTA_INFO', 'PROCESS_INFO', 'AUDIT_INFO', 'EXTENDED_INFO',
+            'HANDLE_REVOCATION_INFO', 'PADDING_INFO'
         ]:
-            try:
+            with contextlib.suppress(AttributeError, exceptions.SymbolError):
                 type_name = f"{symbol_table_name}{constants.BANG}_OBJECT_HEADER_{header}"
                 header_type = context.symbol_space.get_type(type_name)
                 headers.append(header)
                 sizes.append(header_type.size)
-            except (AttributeError, exceptions.SymbolError):
                 # Some of these may not exist, for example:
                 #   if build < 9200: PADDING_INFO else: AUDIT_INFO
                 #   if build == 10586: HANDLE_REVOCATION_INFO else EXTENDED_INFO
                 # based on what's present and what's not, this list should be the right order and the right length
-                pass
         return headers, sizes
 
     def is_free_pool(self):

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

@@ -1,7 +1,7 @@
 # This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
-
+import contextlib
 import enum
 import logging
 import struct
@@ -75,12 +75,10 @@ class CMHIVE(objects.StructType):
         """
 
         for attr in ["FileFullPath", "FileUserName", "HiveRootPath"]:
-            try:
+            with contextlib.suppress(AttributeError, exceptions.InvalidAddressException):
                 name = getattr(self, attr)
                 if name.Length > 0:
                     return name.get_string()
-            except (AttributeError, exceptions.InvalidAddressException):
-                pass
 
         return None