make ruff happy

development/mac-kdk/parse_pbzx2.py

@@ -7,6 +7,7 @@
 # Cleaned up C version (as the basis for my code) here, thanks to Pepijn Bruienne / @bruienne
 # https://gist.github.com/bruienne/029494bbcfb358098b41
 
+import os
 import struct
 import sys
 
@@ -22,7 +23,7 @@ def seekread(f, offset = None, length = 0, relative = True):
 
 def parse_pbzx(pbzx_path):
     section = 0
-    xar_out_path = '%s.part%02d.cpio.xz' % (pbzx_path, section)
+    xar_out_path = f'{pbzx_path}.part{section:02d}.cpio.xz'
     with open(pbzx_path, 'rb') as f:
         # pbzx = f.read()
         # f.close()
@@ -50,12 +51,12 @@ def parse_pbzx(pbzx_path):
                     # ... and split it out ...
                     f_content = seekread(f, length = f_length)
                     section += 1
-                    decomp_out = '%s.part%02d.cpio' % (pbzx_path, section)
+                    decomp_out = f'{pbzx_path}.part{section:02d}.cpio'
                     with open(decomp_out, 'wb') as g:
                         g.write(f_content)
                     # Now to start the next section, which should hopefully be .xz (we'll just assume it is ...)
                     section += 1
-                    xar_out_path = '%s.part%02d.cpio.xz' % (pbzx_path, section)
+                    xar_out_path = f'{pbzx_path}.part{section:02d}.cpio.xz'
                 else:
                     f_length -= 6
                     # This part needs buffering

development/pdbparse-to-json.py

@@ -32,12 +32,12 @@ def retreive_pdb(self, guid: str, file_name: str) -> Optional[str]:
             result = None
             for suffix in [file_name[:-1] + '_', file_name]:
                 try:
-                    logger.debug(f"Attempting to retrieve {url + suffix}")
+                    logger.debug("Attempting to retrieve %s", url + suffix)
                     result, _ = request.urlretrieve(url + suffix)
                 except request.HTTPError as excp:
-                    logger.debug(f"Failed with {excp}")
+                    logger.debug("Failed with %s", excp)
             if result:
-                logger.debug(f"Successfully written to {result}")
+                logger.debug("Successfully written to %s", result)
                 break
         return result
 

development/schema_validate.py

@@ -6,7 +6,7 @@
 # TODO: Rather nasty hack, when volatility's actually installed this would be unnecessary
 sys.path += ".."
 
-import logging
+import logging  # noqa: E402
 
 console = logging.StreamHandler()
 console.setLevel(logging.DEBUG)
@@ -17,7 +17,7 @@
 logger.addHandler(console)
 logger.setLevel(logging.DEBUG)
 
-from volatility3 import schemas
+from volatility3 import schemas  # noqa: E402
 
 if __name__ == '__main__':
     parser = argparse.ArgumentParser("Validates ")

doc/source/conf.py

@@ -19,6 +19,8 @@
 
 import sphinx.ext.apidoc
 
+from importlib.util import find_spec
+
 
 def setup(app):
     volatility_directory = os.path.abspath(
@@ -124,7 +126,7 @@ def setup(app):
 # documentation root, use os.path.abspath to make it absolute, like shown here.
 sys.path.insert(0, os.path.abspath("../.."))
 
-from volatility3.framework import constants
+from volatility3.framework import constants  # noqa: E402
 
 # -- General configuration ------------------------------------------------
 
@@ -147,13 +149,9 @@ def setup(app):
 
 autosectionlabel_prefix_document = True
 
-try:
-    import sphinx_autodoc_typehints
-
+if find_spec("sphinx_autodoc_typehints") is not None:
     extensions.append("sphinx_autodoc_typehints")
-except ImportError:
-    # If the autodoc typehints extension isn't available, carry on regardless
-    pass
+# If the autodoc typehints extension isn't available, carry on regardless
 
 # Add any paths that contain templates here, relative to this directory.
 # templates_path = ['tools/templates']

volatility3/cli/__init__.py

@@ -879,7 +879,7 @@ def populate_requirements_argparse(
                 volatility3.framework.configuration.requirements.ListRequirement,
             ):
                 # Allow a list of integers, specified with the convenient 0x hexadecimal format
-                if requirement.element_type == int:
+                if requirement.element_type is int:
                     additional["type"] = lambda x: int(x, 0)
                 else:
                     additional["type"] = requirement.element_type

volatility3/framework/__init__.py

@@ -22,14 +22,14 @@
         )
     )
 
-import importlib
-import inspect
-import logging
-import os
-import traceback
-from typing import Any, Dict, Generator, List, Tuple, Type, TypeVar
+import importlib  # noqa: E402
+import inspect  # noqa: E402
+import logging  # noqa: E402
+import os  # noqa: E402
+import traceback  # noqa: E402
+from typing import Any, Dict, Generator, List, Tuple, Type, TypeVar  # noqa: E402
 
-from volatility3.framework import constants, interfaces
+from volatility3.framework import constants, interfaces  # noqa: E402
 
 
 # ##
@@ -74,7 +74,7 @@ def __init__(self, value: Any, cls: Type) -> None:
         self.cls = cls
 
     def __get__(self, obj: Any, get_type: Type = None) -> Any:
-        if type == self.cls:
+        if type is self.cls:
             if hasattr(self.default_value, "__get__"):
                 return self.default_value.__get__(obj, get_type)
             return self.default_value

volatility3/framework/configuration/__init__.py

@@ -3,3 +3,5 @@
 #
 
 from volatility3.framework.configuration import requirements
+
+__all__ = ["requirements"]

volatility3/framework/constants/__init__.py

@@ -14,7 +14,7 @@
 from typing import Callable, Optional
 
 import volatility3.framework.constants.linux
-import volatility3.framework.constants.windows
+import volatility3.framework.constants.windows  # noqa: F401
 from volatility3.framework.constants._version import (
     PACKAGE_VERSION,
     VERSION_MAJOR,
@@ -141,3 +141,36 @@ def __getattr__(name):
         return globals()[f"{deprecated_tag}{name}"]
 
     return getattr(__import__(__name__), name)
+
+
+__all__ = [
+    "PACKAGE_VERSION",
+    "VERSION_MAJOR",
+    "VERSION_MINOR",
+    "VERSION_PATCH",
+    "VERSION_SUFFIX",
+    "PLUGINS_PATH",
+    "SYMBOL_BASEPATHS",
+    "ISF_EXTENSIONS",
+    "BANG",
+    "AUTOMAGIC_CONFIG_PATH",
+    "LOGLEVEL_INFO",
+    "LOGLEVEL_DEBUG",
+    "LOGLEVEL_V",
+    "LOGLEVEL_VV",
+    "LOGLEVEL_VVV",
+    "LOGLEVEL_VVVV",
+    "CACHE_PATH",
+    "SQLITE_CACHE_PERIOD",
+    "IDENTIFIERS_FILENAME",
+    "CACHE_SQLITE_SCHEMA_VERSION",
+    "BUG_URL",
+    "ProgressCallback",
+    "OS_CATEGORIES",
+    "Parallelism",
+    "PARALLELISM",
+    "ISF_MINIMUM_SUPPORTED",
+    "ISF_MINIMUM_DEPRECATED",
+    "OFFLINE",
+    "REMOTE_ISF_URL",
+]

volatility3/framework/interfaces/__init__.py

@@ -22,3 +22,14 @@
     symbols,
     automagic,
 )
+
+__all__ = [
+    "renderers",
+    "configuration",
+    "context",
+    "layers",
+    "objects",
+    "plugins",
+    "symbols",
+    "automagic",
+]

volatility3/framework/layers/resources.py

@@ -29,7 +29,7 @@
 
 try:
     # Import so that the handler is found by the framework.class_subclasses callc
-    import smb.SMBHandler  # lgtm [py/unused-import]
+    import smb.SMBHandler  # lgtm [py/unused-import]  # noqa: F401
 except ImportError:
     # If we fail to import this, it means that SMB handling won't be available
     pass

volatility3/framework/layers/scanners/__init__.py

@@ -136,3 +136,6 @@ def search(self, haystack: bytes) -> Generator[Tuple[int, bytes], None, None]:
             )
         for match in re.finditer(self._regex, haystack):
             yield match.start(0), match.group()
+
+
+__all__ = ["multiregexp", "BytesScanner", "RegExScanner", "MultiStringScanner"]

volatility3/framework/objects/__init__.py

@@ -35,13 +35,13 @@ def convert_data_to_value(
     data_format: DataFormatInfo,
 ) -> TUnion[int, float, bytes, str, bool]:
     """Converts a series of bytes to a particular type of value."""
-    if struct_type == int:
+    if struct_type is int:
         return int.from_bytes(
             data, byteorder=data_format.byteorder, signed=data_format.signed
         )
-    if struct_type == bool:
+    if struct_type is bool:
         struct_format = "?"
-    elif struct_type == float:
+    elif struct_type is float:
         float_vals = "zzezfzzzd"
         if (
             data_format.length > len(float_vals)
@@ -70,17 +70,17 @@ def convert_value_to_data(
             f"Written value is not of the correct type for {struct_type.__name__}"
         )
 
-    if struct_type == int and isinstance(value, int):
+    if struct_type is int and isinstance(value, int):
         # Doubling up on the isinstance is for mypy
         return int.to_bytes(
             value,
             length=data_format.length,
             byteorder=data_format.byteorder,
             signed=data_format.signed,
         )
-    if struct_type == bool:
+    if struct_type is bool:
         struct_format = "?"
-    elif struct_type == float:
+    elif struct_type is float:
         float_vals = "zzezfzzzd"
         if (
             data_format.length > len(float_vals)

volatility3/framework/plugins/isfinfo.py

@@ -7,6 +7,7 @@
 import pathlib
 import zipfile
 from typing import Generator, List
+from importlib.util import find_spec
 
 from volatility3 import schemas, symbols
 from volatility3.framework import constants, interfaces, renderers
@@ -96,16 +97,12 @@ def _generator(self):
                     if filter_item in isf_file:
                         filtered_list.append(isf_file)
 
-        try:
-            import jsonschema
-
-            if not self.config["validate"]:
-                raise ImportError  # Act as if we couldn't import if validation is turned off
+        if find_spec("jsonschema") and self.config["validate"]:
 
             def check_valid(data):
                 return "True" if schemas.validate(data, True) else "False"
 
-        except ImportError:
+        else:
 
             def check_valid(data):
                 return "Unknown"

volatility3/framework/plugins/linux/kmsg.py

@@ -149,7 +149,7 @@ def nsec_to_sec_str(self, nsec: int) -> str:
         #   This might seem insignificant but it could cause some issues
         #   when compared with userland tool results or when used in
         #   timelines.
-        return "%lu.%06lu" % (nsec / 1000000000, (nsec % 1000000000) / 1000)
+        return f"{nsec / 1000000000:lu}.{(nsec % 1000000000) / 1000:06lu}"
 
     def get_timestamp_in_sec_str(self, obj) -> str:
         # obj could be log, printk_log or printk_info
@@ -166,7 +166,7 @@ def get_caller(self, obj):
 
     def get_caller_text(self, caller_id):
         caller_name = "CPU" if caller_id & 0x80000000 else "Task"
-        caller = "%s(%u)" % (caller_name, caller_id & ~0x80000000)
+        caller = f"{caller_name}({caller_id & ~0x80000000:u})"
         return caller
 
     def get_prefix(self, obj) -> Tuple[int, int, str, str]:

volatility3/framework/plugins/mac/check_sysctl.py

@@ -60,7 +60,7 @@ def _parse_global_variable_sysctls(self, kernel, name):
         return var_str
 
     def _process_sysctl_list(self, kernel, sysctl_list, recursive=0):
-        if type(sysctl_list) == volatility3.framework.objects.Pointer:
+        if type(sysctl_list) is volatility3.framework.objects.Pointer:
             sysctl_list = sysctl_list.dereference().cast("sysctl_oid_list")
 
         sysctl = sysctl_list.slh_first

volatility3/framework/plugins/mac/kauth_scopes.py

@@ -80,7 +80,7 @@ def _generator(self):
                 (
                     identifier,
                     format_hints.Hex(scope.ks_idata),
-                    len([l for l in scope.get_listeners()]),
+                    len([listener for listener in scope.get_listeners()]),
                     format_hints.Hex(callback),
                     module_name,
                     symbol_name,

volatility3/framework/plugins/windows/netscan.py

@@ -161,7 +161,7 @@ def determine_tcpip_version(
             raise NotImplementedError(
                 "Kernel Debug Structure version format not supported!"
             )
-        except:
+        except:  # noqa: E722
             # unsure what to raise here. Also, it might be useful to add some kind of fallback,
             # either to a user-provided version or to another method to determine tcpip.sys's version
             raise exceptions.VolatilityException(

volatility3/framework/plugins/windows/psxview.py

@@ -218,7 +218,7 @@ def _generator(self):
             name = self._proc_name_to_string(proc)
 
             exit_time = proc.get_exit_time()
-            if type(exit_time) != datetime.datetime:
+            if type(exit_time) is not datetime.datetime:
                 exit_time = ""
             else:
                 exit_time = str(exit_time)

volatility3/framework/plugins/windows/shimcachemem.py

@@ -146,7 +146,7 @@ def find_shimcache_win_xp(
             context, layer_name, kernel_symbol_table
         ):
             pid = process.UniqueProcessId
-            vollog.debug("checking process %d" % pid)
+            vollog.debug("checking process %d", pid)
             for vad in vadinfo.VadInfo.list_vads(
                 process, lambda x: x.get_tag() == b"Vad " and x.Protection == 4
             ):

volatility3/framework/renderers/__init__.py

@@ -430,10 +430,10 @@ def __call__(self, values: List[Any]) -> Any:
                 value = datetime.datetime.min
             elif self._type in [int, float]:
                 value = -1
-            elif self._type == bool:
+            elif self._type is bool:
                 value = False
             elif self._type in [str, renderers.Disassembly]:
                 value = "-"
-            elif self._type == bytes:
+            elif self._type is bytes:
                 value = b""
         return value

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

@@ -237,7 +237,7 @@ def get_special_path(self):
     def get_path(self, context, config_prefix):
         node = self.get_vnode(context, config_prefix)
 
-        if type(node) == str and node == "sub_map":
+        if type(node) is str and node == "sub_map":
             ret = node
         elif node:
             path = []