Merge pull request #1750 from volatilityfoundation/import_checker

Testing: Check + fix 'import from' statements

Author: ikelos

.github/workflows/check-requirements.yml .github/workflows/vol3-code-analysis.yml.github/workflows/check-requirements.yml renamed to .github/workflows/vol3-code-analysis.yml

@@ -1,4 +1,4 @@
-name: Check Volatility3 Version Requirements
+name: Volatility3 Code Analysis
 on: [push, pull_request]
 jobs:
 
@@ -21,5 +21,4 @@ jobs:
 
     - name: Testing...
       run: |
-        # Verify completeness of ConfigurableInterface requirements
-        python ./test/check_configurable_requirements.py
+        python ./test/volatility3_code_analysis.py

test/check_configurable_requirements.py test/volatility3_code_analysis.pytest/check_configurable_requirements.py renamed to test/volatility3_code_analysis.py

@@ -22,6 +22,7 @@
 import logging
 import pkgutil
 import sys
+import traceback
 import types
 from typing import Any, Iterator, List, Optional, Tuple, Type, Union
 
@@ -105,6 +106,36 @@ def __str__(self) -> str:
         )
 
 
+class DirectVolatilityImportUsage(CodeViolation):
+
+    def __init__(
+        self,
+        module: types.ModuleType,
+        node: ast.AST,
+        importing_module: str,
+        imported_item: object,
+        imported_name: str,
+    ) -> None:
+        self.imported_item = imported_item
+        self.imported_name = imported_name
+        self.importing_module = importing_module
+        super().__init__(module, node)
+
+    def __str__(self) -> str:
+        components = self.importing_module.split(".")
+        return (
+            super().__str__()
+            + ": "
+            + (
+                f"Direct import of {self.imported_name} "
+                f"({type(self.imported_item)}) "
+                f"from module {self.importing_module} - "
+                "change to "
+                f"'from {'.'.join(components[:-1])} import {components[-1]} and using {components[-1]}.{self.imported_name}"
+            )
+        )
+
+
 def is_versionable(var):
     try:
         return (
@@ -134,6 +165,46 @@ def __init__(self, module: types.ModuleType) -> None:
     def violations(self):
         return self._violations
 
+    def _check_vol3_import_from(self, node: ast.ImportFrom):
+        """
+        Ensure that the only thing imported from a volatility3 module (apart
+        from the root volatility3 module) are functions and modules. This
+        prevents re-exporting of classes and variables from modules that use
+        them.
+        """
+        if (
+            node.module
+            and node.module.startswith("volatility3.") # Give a pass to volatility3 module
+            and node.module != "volatility3.framework.constants._version" # make an exception for this
+        ):
+            for name in node.names:
+                try:
+                    item = vars(self._module)[
+                        name.asname if name.asname is not None else name.name
+                    ]
+                except KeyError:
+                    logger.debug(
+                        "Couldn't find imported name %s in module %s",
+                        name.asname or name.name,
+                        self._module.__name__,
+                    )
+                    continue
+
+                if not (isinstance(item, types.ModuleType) or inspect.isfunction(item)):
+                    self._violations.append(
+                        DirectVolatilityImportUsage(
+                            self._module,
+                            node,
+                            node.module,
+                            item,
+                            name.asname or name.name,
+                        )
+                    )
+
+    def enter_ImportFrom(self, node: ast.ImportFrom):
+        self._check_vol3_import_from(node)
+
+
     def enter_ClassDef(self, node: ast.ClassDef) -> Any:
         logger.debug("Entering class %s", node.name)
         clazz = None
@@ -304,6 +375,7 @@ def report_missing_requirements() -> Iterator[Tuple[str, UnrequiredVersionableUs
                 modname,
                 str(exc),
             )
+            traceback.print_exc()
             continue
 
         logger.info("Checking module %s", plugin_module.__name__)
@@ -344,9 +416,7 @@ def perform_review():
         print(str(usage))
 
     if found:
-        print(
-            f"Found {found} issues"
-        )
+        print(f"Found {found} issues")
         sys.exit(1)
 
     print("All configurable classes passed validation!")

volatility3/framework/exceptions.py

@@ -11,7 +11,6 @@
 from typing import Callable, Dict, Optional, Tuple
 
 from volatility3.framework import interfaces
-from volatility3.framework.interfaces.configuration import VersionableInterface
 
 
 class VolatilityException(Exception):
@@ -143,15 +142,15 @@ class VersionMismatchException(VolatilityException):
     def __init__(
         self,
         source_component: Callable,
-        target_component: VersionableInterface,
+        target_component: interfaces.configuration.VersionableInterface,
         target_version: Tuple[int, int, int],
         failure_reason: str = None,
         *args,
     ):
         """
         Args:
             source_component: The component that required the target component
-            target_component: The component that is required. Must inherit from VersionableInterface
+            target_component: The component that is required. Must inherit from interfaces.configuration.VersionableInterface
             target_version: The version of the target component that was required, and ultimately was not satisfied
             failure_reason: A detailed failure reason to enhance debugging and bug tracking
         """

volatility3/framework/interfaces/symbols.py

@@ -10,7 +10,6 @@
 from volatility3.framework import constants, exceptions, interfaces
 from volatility3.framework.configuration import requirements
 from volatility3.framework.interfaces import configuration, objects
-from volatility3.framework.interfaces.configuration import RequirementInterface
 
 
 class SymbolInterface:
@@ -347,7 +346,7 @@ def build_configuration(self) -> "configuration.HierarchicalDict":
         return config
 
     @classmethod
-    def get_requirements(cls) -> List[RequirementInterface]:
+    def get_requirements(cls) -> List[configuration.RequirementInterface]:
         return super().get_requirements() + [
             requirements.IntRequirement(
                 name="symbol_mask",

volatility3/framework/layers/elf.py

@@ -6,7 +6,7 @@
 from typing import Optional
 
 from volatility3.framework import exceptions, interfaces, constants
-from volatility3.framework.constants.linux import ELF_CLASS
+from volatility3.framework.constants import linux as linux_constants
 from volatility3.framework.layers import segmented
 from volatility3.framework.symbols import intermed
 
@@ -23,7 +23,7 @@ class Elf64Layer(segmented.SegmentedLayer):
 
     _header_struct = struct.Struct("<IBBB")
     MAGIC = 0x464C457F  # "\x7fELF"
-    ELF_CLASS = ELF_CLASS.ELFCLASS64
+    ELF_CLASS = linux_constants.ELF_CLASS.ELFCLASS64
 
     def __init__(
         self, context: interfaces.context.ContextInterface, config_path: str, name: str

volatility3/framework/layers/registry.py

@@ -7,11 +7,6 @@
 
 from volatility3.framework import constants, exceptions, interfaces, objects
 from volatility3.framework.configuration import requirements
-from volatility3.framework.configuration.requirements import (
-    IntRequirement,
-    TranslationLayerRequirement,
-)
-from volatility3.framework.exceptions import InvalidAddressException
 from volatility3.framework.layers import linear
 from volatility3.framework.symbols import intermed
 from volatility3.framework.symbols.windows import extensions
@@ -154,7 +149,7 @@ def address_mask(self) -> int:
     @property
     def root_cell_offset(self) -> int:
         """Returns the offset for the root cell in this hive."""
-        with contextlib.suppress(InvalidAddressException):
+        with contextlib.suppress(exceptions.InvalidAddressException):
             if (
                 self._base_block.Signature.cast(
                     "string", max_length=4, encoding="latin-1"
@@ -271,7 +266,7 @@ def _mask(value: int, high_bit: int, low_bit: int) -> int:
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
         return [
-            IntRequirement(
+            requirements.IntRequirement(
                 name="hive_offset",
                 description="Offset within the base layer at which the hive lives",
                 default=0,
@@ -280,7 +275,7 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
             requirements.SymbolTableRequirement(
                 name="nt_symbols", description="Windows kernel symbols"
             ),
-            TranslationLayerRequirement(
+            requirements.TranslationLayerRequirement(
                 name="base_layer",
                 description="Layer in which the registry hive lives",
                 optional=False,

volatility3/framework/layers/xen.py

@@ -5,7 +5,7 @@
 from volatility3.framework import constants, interfaces, exceptions
 from volatility3.framework.layers import elf
 from volatility3.framework.symbols import intermed
-from volatility3.framework.constants.linux import ELF_CLASS
+from volatility3.framework.constants import linux as linux_constants
 
 vollog = logging.getLogger(__name__)
 
@@ -15,7 +15,7 @@ class XenCoreDumpLayer(elf.Elf64Layer):
 
     _header_struct = struct.Struct("<IBBB")
     MAGIC = 0x464C457F  # "\x7fELF"
-    ELF_CLASS = ELF_CLASS.ELFCLASS64
+    ELF_CLASS = linux_constants.ELF_CLASS.ELFCLASS64
 
     def __init__(
         self, context: interfaces.context.ContextInterface, config_path: str, name: str

volatility3/framework/plugins/linux/bash.py

@@ -13,7 +13,7 @@
 from volatility3.framework.interfaces import plugins
 from volatility3.framework.layers import scanners
 from volatility3.framework.objects import utility
-from volatility3.framework.symbols.linux.bash import BashIntermedSymbols
+from volatility3.framework.symbols.linux import bash
 from volatility3.plugins import timeliner
 from volatility3.plugins.linux import pslist
 
@@ -70,7 +70,7 @@ def _generator(self, tasks):
             pack_format = "Q"
             bash_json_file = "bash64"
 
-        bash_table_name = BashIntermedSymbols.create(
+        bash_table_name = bash.BashIntermedSymbols.create(
             self.context, self.config_path, "linux", bash_json_file
         )
 

volatility3/framework/plugins/linux/elfs.py

@@ -14,7 +14,7 @@
 from volatility3.framework.renderers import format_hints
 from volatility3.framework.symbols import intermed
 from volatility3.framework.symbols.linux.extensions import elf
-from volatility3.framework.constants.linux import ELF_MAX_EXTRACTION_SIZE
+from volatility3.framework.constants import linux as linux_constants
 from volatility3.plugins.linux import pslist
 
 
@@ -116,7 +116,7 @@ def elf_dump(
             real_size = end - start
 
             # Check if ELF has a legitimate size
-            if real_size < 0 or real_size > ELF_MAX_EXTRACTION_SIZE:
+            if real_size < 0 or real_size > linux_constants.ELF_MAX_EXTRACTION_SIZE:
                 raise ValueError(f"The claimed size of the ELF is invalid: {real_size}")
 
             sections[start] = real_size

volatility3/framework/plugins/linux/modxview.py

@@ -6,9 +6,9 @@
 
 import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
 
-from volatility3.framework import interfaces, deprecation
+from volatility3.framework import interfaces, deprecation, renderers
 from volatility3.framework.configuration import requirements
-from volatility3.framework.renderers import format_hints, TreeGrid, NotAvailableValue
+from volatility3.framework.renderers import format_hints
 from volatility3.framework.symbols.linux import extensions
 from volatility3.framework.constants import architectures
 from volatility3.framework.symbols.linux.utilities import tainting
@@ -156,12 +156,12 @@ def _generator(self):
             yield (
                 0,
                 (
-                    module.get_name() or NotAvailableValue(),
+                    module.get_name() or renderers.NotAvailableValue(),
                     format_hints.Hex(module_offset),
                     linux_utilities_modules.ModuleGathererLsmod.name in gatherers,
                     linux_utilities_modules.ModuleGathererSysFs.name in gatherers,
                     linux_utilities_modules.ModuleGathererScanner.name in gatherers,
-                    taints or NotAvailableValue(),
+                    taints or renderers.NotAvailableValue(),
                 ),
             )
 
@@ -175,7 +175,7 @@ def run(self):
             ("Taints", str),
         ]
 
-        return TreeGrid(
+        return renderers.TreeGrid(
             columns,
             self._generator(),
         )