From 2ad1536b4e154f601f28247e15640ccc4bc0ad84 Mon Sep 17 00:00:00 2001 From: David McDonald Date: Mon, 24 Mar 2025 11:59:38 -0500 Subject: [PATCH 01/11] Testing: Verify `VersionRequirement`s This adds a script and GitHub action to the `test` directory that dynamically imports all modules in `volatility3`, searches for usages of `VersionableInterface` objects within classes that inherit from `ConfigurableInterface` but don't enumerate the used component as a requirement in `get_requirements()`, and returns -1 if any violations are found. Fixes --- .github/workflows/check-requirements.yml | 25 ++ pyproject.toml | 2 + test/check_configurable_requirements.py | 300 +++++++++++++++++++++++ 3 files changed, 327 insertions(+) create mode 100644 .github/workflows/check-requirements.yml create mode 100644 test/check_configurable_requirements.py diff --git a/.github/workflows/check-requirements.yml b/.github/workflows/check-requirements.yml new file mode 100644 index 000000000..9892d7b94 --- /dev/null +++ b/.github/workflows/check-requirements.yml @@ -0,0 +1,25 @@ +name: Check Volatility3 Version Requirements +on: [push, pull_request] +jobs: + + build: + runs-on: ubuntu-22.04 + strategy: + matrix: + python-version: ["3.8"] + steps: + - uses: actions/checkout@v4 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install .[test] + + - name: Testing... + run: | + # Verify completeness of ConfigurableInterface requirements + python ./test/check_configurable_requirements.py diff --git a/pyproject.toml b/pyproject.toml index abd2e79f7..8bc3693a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,6 +46,8 @@ test = [ "volatility3[dev]", "pytest>=8.3.3,<9", "yara-x>=0.10.0,<1", + "tree-sitter==0.21.3", + "tree-sitter-python==0.21.0", ] docs = [ diff --git a/test/check_configurable_requirements.py b/test/check_configurable_requirements.py new file mode 100644 index 000000000..e21df18ac --- /dev/null +++ b/test/check_configurable_requirements.py @@ -0,0 +1,300 @@ +import importlib +import inspect +import pkgutil +import sys +import traceback +import types +from textwrap import dedent +from typing import Dict, Iterator, List, NamedTuple, Optional, Tuple, Type + +from tree_sitter import Language, Node, Parser +from tree_sitter_python import language as python_language + +from volatility3.framework import configuration, interfaces + + +class UnrequiredVersionableUsage(NamedTuple): + versionable_item_class: str + """ + The name of the VersionableInterface class + """ + + consuming_class: str + """ + The name of the class that is using the imported VersionableInterface class + """ + + methodname: Optional[str] + """ + The name of the invoked method or attribute, if one is used or referenced + """ + + node: Node + """ + The tree-sitter node encapsulating the used module component. + """ + + def __str__(self) -> str: + return ( + f"Found usage of {self.versionable_item_class} " + f"in class {self.consuming_class} that is not declared " + f"in {self.consuming_class}'s `get_requirements()` classmethod" + ) + + +class RequirementValidator: + language = Language(python_language(), "python") + + def __init__(self, plugin_module: types.ModuleType) -> None: + if plugin_module.__file__ is None: + raise ValueError("Attempting to validate a module without a file") + + self._module = plugin_module + + # See which classes in *this* module are configurable (can have requirements declared) + self._configurable_classes = get_configurable_classes(plugin_module) + + # Get a mapping of class names to configurable classes that they declare in their requirements + self._versioned_item_mapping = get_versioned_item_mapping( + self._configurable_classes + ) + + # Get a mapping of module name -> versionable classes within the namespace of each module + self._imported_mod_classes = get_versionable_import_mapping( + get_imported_modules(plugin_module) + ) + + with open(plugin_module.__file__, "rb") as f: + source = f.read() + + self._parser = Parser() + self._parser.set_language(self.language) + self._tree = self._parser.parse(source) + + def enumerate_unrequired_usages( + self, + clazz: Type[interfaces.configuration.ConfigurableInterface], + class_node: Node, + ): + + # This query is designed to look for three different identifier usages: + # simple identifiers: PsList + # module attrs: pslist.PsList + # method calls: pslist.PsList.list_processes + obj_query = self.language.query( + dedent( + """ + [ + (identifier) + (attribute + object: (identifier) + attribute: (identifier)) + (attribute + object: (attribute + object: (identifier) + attribute: (identifier)) + ) + ] @ident + """ + ) + ) + + containing_name = class_node.child_by_field_name("name").text.decode("utf-8") + + valid_types = self._versioned_item_mapping[containing_name] + for _, match in obj_query.matches(class_node): + if "ident" not in match: + continue + + # Get the raw text of the match. This could be something like + # - PsList + # - pslist.PsList + # - pslist.PsList.list_processes + ident_text = match["ident"].text.decode("utf-8") + + # split the attributes + components = ident_text.split(".") + try: + # See if the first attribute is in the module namespace. + item = vars(self._module)[components[0]] + except KeyError: + # If it's not, it's likely a variable in a smaller scope and we + # can ignore it. + continue + + # If it's in the module namespace and is a module... + if isinstance(item, types.ModuleType): + try: + # We try getting attributes from it until we + # find one that is a versionable class + + # Ideally, we shouldn't have to look further than + # two levels + item = getattr(item, components[1]) + if not is_versionable(item): + item = getattr(item, components[2]) + if not is_versionable(item): + continue + + except (IndexError, AttributeError): + # we ran out of attributes to check + continue + + elif is_versionable(item): + # The versionable thing was at the top level. This + # goes against our preferred style, but is possible. + pass + else: + # This isn't something we care about. + continue + + if ( + item in valid_types + or item is clazz + or inspect.isabstract(item) + or item + is interfaces.configuration.VersionableInterface # Avoid checking the interface itself + ): + continue + + yield UnrequiredVersionableUsage( + item, + containing_name, + components[1] if len(components) > 1 else None, + match["ident"], + ) + + def find_class_nodes( + self, + ) -> Iterator[Tuple[Type[interfaces.configuration.ConfigurableInterface], Node]]: + """ + Yields an iterator of (classname, node) tuples, where the node is the subtree containing + the entire class definition. + """ + class_query = self.language.query("(class_definition) @classdef") + + matches = class_query.captures(self._tree.root_node) + for node, _ in matches: + classname = node.child_by_field_name("name").text.decode("utf-8") + if classname not in self._configurable_classes: + continue + + yield self._configurable_classes[classname], node + + +def is_versionable(var): + try: + return issubclass(var, interfaces.configuration.VersionableInterface) + except TypeError: + return False + + +def is_configurable(var): + try: + return issubclass(var, interfaces.configuration.ConfigurableInterface) + except TypeError: + return False + + +def get_imported_modules( + plugin_module: types.ModuleType, +) -> List[Tuple[str, types.ModuleType]]: + return [ + (name, var) + for name, var in vars(plugin_module).items() + if isinstance(var, types.ModuleType) + ] + + +def get_configurable_classes( + plugin_module: types.ModuleType, +) -> Dict[str, Type[interfaces.configuration.ConfigurableInterface]]: + return { + name: clazz + for name, clazz in vars(plugin_module).items() + if is_configurable(clazz) + } + + +def get_versioned_item_mapping( + configurable_classes: Dict[ + str, Type[interfaces.configuration.ConfigurableInterface] + ] +) -> Dict[str, List[Type[interfaces.configuration.VersionableInterface]]]: + return { + name: [ + req._component + for req in clazz.get_requirements() + if isinstance(req, configuration.requirements.VersionRequirement) + ] + for name, clazz in configurable_classes.items() + } + + +def get_versionable_import_mapping( + imported_modules: List[Tuple[str, types.ModuleType]] +) -> Dict[str, List[str]]: + return { + modname: [name for name, var in vars(module).items() if is_versionable(var)] + for modname, module in imported_modules + } + + +def report_missing_requirements() -> Iterator[Tuple[str, UnrequiredVersionableUsage]]: + vol3 = importlib.import_module("volatility3") + + for _, module_name, _ in pkgutil.walk_packages( + vol3.__path__, vol3.__name__ + ".", onerror=lambda _: None + ): + try: + # import the module that we want to check + modname = module_name.replace( + "volatility3.framework.plugins", "volatility3.plugins" + ) + plugin_module = importlib.import_module(modname) + + except ImportError: + continue + except Exception: + continue + + if plugin_module.__file__ is None: + continue + + try: + # construct a validator for the module + try: + validator = RequirementValidator(plugin_module) + except Exception: + traceback.print_stack() + continue + for clazz, node in validator.find_class_nodes(): + for item in validator.enumerate_unrequired_usages(clazz, node): + yield module_name, item + except Exception as exc: + traceback.print_exc() + print( + f"Failed to create validator for source code from {plugin_module.__file__}: {exc}" + ) + sys.exit(1) + + +def perform_review(): + found = 0 + for mod, usage in report_missing_requirements(): + found += 1 + print( + f"Violation in module {mod} (line {usage.node.start_point[0]}): {str(usage)}" + ) + + if found: + print( + f"Found {found} uses of versionable components not declared in get_requirements()" + ) + sys.exit(1) + + print("All configurable classes passed validation!") + + +if __name__ == "__main__": + perform_review() From 0f73686364392dbba36a4a20fff2a13f8df04c31 Mon Sep 17 00:00:00 2001 From: David McDonald Date: Thu, 27 Mar 2025 11:07:58 -0500 Subject: [PATCH 02/11] Framework: Fix remaining missing requirements This adds all of the missing requirements discovered via the new code analysis script. --- volatility3/cli/volshell/generic.py | 7 ++++++ volatility3/cli/volshell/linux.py | 5 ++++ volatility3/cli/volshell/mac.py | 5 ++++ volatility3/cli/volshell/windows.py | 5 ++++ volatility3/framework/automagic/pdbscan.py | 23 +++++++++++++++---- .../framework/automagic/symbol_finder.py | 7 +++++- volatility3/framework/layers/qemu.py | 11 +++++++++ .../framework/layers/scanners/__init__.py | 5 ++++ volatility3/framework/plugins/banners.py | 7 +++++- volatility3/framework/plugins/linux/bash.py | 10 ++++++++ .../framework/plugins/linux/check_modules.py | 5 ++++ .../framework/plugins/linux/hidden_modules.py | 5 ++++ volatility3/framework/plugins/linux/psscan.py | 5 ++++ .../framework/plugins/linux/vmaregexscan.py | 5 ++++ volatility3/framework/plugins/mac/bash.py | 10 ++++++++ .../framework/plugins/mac/list_files.py | 5 ++++ volatility3/framework/plugins/regexscan.py | 5 ++++ volatility3/framework/plugins/vmscan.py | 7 ++++++ .../framework/plugins/windows/cmdscan.py | 5 ++++ .../framework/plugins/windows/consoles.py | 5 ++++ .../framework/plugins/windows/mbrscan.py | 5 ++++ .../framework/plugins/windows/poolscanner.py | 7 ++++++ .../plugins/windows/skeleton_key_check.py | 5 ++++ .../framework/plugins/windows/svclist.py | 5 ++++ .../framework/plugins/windows/svcscan.py | 5 ++++ .../framework/plugins/windows/vadregexscan.py | 5 ++++ .../framework/plugins/windows/verinfo.py | 5 ++++ volatility3/framework/plugins/yarascan.py | 7 +++++- 28 files changed, 179 insertions(+), 7 deletions(-) diff --git a/volatility3/cli/volshell/generic.py b/volatility3/cli/volshell/generic.py index 3a5d514fe..1b3ae59d1 100644 --- a/volatility3/cli/volshell/generic.py +++ b/volatility3/cli/volshell/generic.py @@ -38,6 +38,8 @@ class Volshell(interfaces.plugins.PluginInterface): _required_framework_version = (2, 0, 0) + _version = (1, 0, 0) + DEFAULT_NUM_DISPLAY_BYTES = 128 def __init__(self, *args, **kwargs): @@ -61,6 +63,11 @@ class Volshell(interfaces.plugins.PluginInterface): default=None, optional=True, ), + requirements.VersionRequirement( + name="regex_scanner", + component=scanners.RegExScanner, + version=(1, 0, 0), + ), requirements.BooleanRequirement( name="script-only", description="Exit volshell after the script specified in --script completes", diff --git a/volatility3/cli/volshell/linux.py b/volatility3/cli/volshell/linux.py index 27c630614..761b64084 100644 --- a/volatility3/cli/volshell/linux.py +++ b/volatility3/cli/volshell/linux.py @@ -36,6 +36,11 @@ class Volshell(generic.Volshell): requirements.IntRequirement( name="pid", description="Process ID", optional=True ), + requirements.VersionRequirement( + name="generic_volshell", + component=generic.Volshell, + version=(1, 0, 0), + ), ] def change_task(self, pid=None): diff --git a/volatility3/cli/volshell/mac.py b/volatility3/cli/volshell/mac.py index 393eff20b..fcb45e124 100644 --- a/volatility3/cli/volshell/mac.py +++ b/volatility3/cli/volshell/mac.py @@ -25,6 +25,11 @@ class Volshell(generic.Volshell): requirements.IntRequirement( name="pid", description="Process ID", optional=True ), + requirements.VersionRequirement( + name="generic_volshell", + component=generic.Volshell, + version=(1, 0, 0), + ), ] def change_task(self, pid=None): diff --git a/volatility3/cli/volshell/windows.py b/volatility3/cli/volshell/windows.py index ce5995648..a8c7af5b3 100644 --- a/volatility3/cli/volshell/windows.py +++ b/volatility3/cli/volshell/windows.py @@ -23,6 +23,11 @@ class Volshell(generic.Volshell): requirements.IntRequirement( name="pid", description="Process ID", optional=True ), + requirements.VersionRequirement( + name="generic_volshell", + component=generic.Volshell, + version=(1, 0, 0), + ), ] def change_process(self, pid=None): diff --git a/volatility3/framework/automagic/pdbscan.py b/volatility3/framework/automagic/pdbscan.py index dd2ad0683..55b9b81e1 100644 --- a/volatility3/framework/automagic/pdbscan.py +++ b/volatility3/framework/automagic/pdbscan.py @@ -17,7 +17,7 @@ from volatility3.framework import constants, exceptions, interfaces, layers from volatility3.framework.configuration import requirements from volatility3.framework.layers import intel, scanners from volatility3.framework.symbols import native -from volatility3.framework.symbols.windows.pdbutil import PDBUtility +from volatility3.framework.symbols.windows import pdbutil if __name__ == "__main__": import sys @@ -50,6 +50,21 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): max_pdb_size = 0x400000 exclusion_list = ["linux", "mac"] + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + return [ + requirements.VersionRequirement( + name="pdb_utility", + component=pdbutil.PDBUtility, + version=(1, 0, 1), + ), + requirements.VersionRequirement( + name="bytes_scanner", + component=scanners.BytesScanner, + version=(1, 0, 0), + ), + ] + def find_virtual_layers_from_req( self, context: interfaces.context.ContextInterface, @@ -120,7 +135,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): ): raise TypeError("PDB name or GUID not a string value") - PDBUtility.load_windows_symbol_table( + pdbutil.PDBUtility.load_windows_symbol_table( context=context, guid=kernel["GUID"], age=kernel["age"], @@ -259,7 +274,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): bytes(name + ".pdb", "utf-8") for name in constants.windows.KERNEL_MODULE_NAMES ] - kernels = PDBUtility.pdbname_scan( + kernels = pdbutil.PDBUtility.pdbname_scan( ctx=context, layer_name=layer_to_scan, start=start_scan_address, @@ -362,7 +377,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface): with contextlib.suppress(exceptions.InvalidAddressException): if vlayer.read(address, 0x2) == b"MZ": res = list( - PDBUtility.pdbname_scan( + pdbutil.PDBUtility.pdbname_scan( ctx=context, layer_name=vlayer.name, page_size=vlayer.page_size, diff --git a/volatility3/framework/automagic/symbol_finder.py b/volatility3/framework/automagic/symbol_finder.py index 1d30f3f51..d7c6a22c1 100644 --- a/volatility3/framework/automagic/symbol_finder.py +++ b/volatility3/framework/automagic/symbol_finder.py @@ -40,7 +40,12 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface): name="SQLiteCache", component=symbol_cache.SqliteCache, version=(1, 0, 0), - ) + ), + requirements.VersionRequirement( + name="multi_string_scanner", + component=scanners.MultiStringScanner, + version=(1, 0, 0), + ), ] @property diff --git a/volatility3/framework/layers/qemu.py b/volatility3/framework/layers/qemu.py index a8127e954..eb44de347 100644 --- a/volatility3/framework/layers/qemu.py +++ b/volatility3/framework/layers/qemu.py @@ -9,6 +9,7 @@ import struct from typing import Any, Dict, List, Optional, Set, Tuple from volatility3.framework import constants, exceptions, interfaces +from volatility3.framework.configuration import requirements from volatility3.framework.layers import scanners, segmented from volatility3.framework.symbols import intermed @@ -99,6 +100,16 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer): context=context, config_path=config_path, name=name, metadata=metadata ) + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + return [ + requirements.VersionRequirement( + name="regex_scanner", + component=scanners.RegExScanner, + version=(1, 0, 0), + ), + ] + @classmethod def _check_header( cls, base_layer: interfaces.layers.DataLayerInterface, name: str = "" diff --git a/volatility3/framework/layers/scanners/__init__.py b/volatility3/framework/layers/scanners/__init__.py index be9f1c39a..f07849f42 100644 --- a/volatility3/framework/layers/scanners/__init__.py +++ b/volatility3/framework/layers/scanners/__init__.py @@ -11,6 +11,8 @@ from volatility3.framework.layers.scanners import multiregexp as multiregexp class BytesScanner(layers.ScannerInterface): thread_safe = True + _version = (1, 0, 0) + _required_framework_version = (2, 0, 0) def __init__(self, needle: bytes) -> None: @@ -38,6 +40,8 @@ class RegExScanner(layers.ScannerInterface): thread_safe = True + _version = (1, 0, 0) + _required_framework_version = (2, 0, 0) def __init__(self, pattern: bytes, flags: int = re.DOTALL) -> None: @@ -57,6 +61,7 @@ class RegExScanner(layers.ScannerInterface): class MultiStringScanner(layers.ScannerInterface): thread_safe = True + _version = (1, 0, 0) _required_framework_version = (2, 0, 0) def __init__(self, patterns: List[bytes]) -> None: diff --git a/volatility3/framework/plugins/banners.py b/volatility3/framework/plugins/banners.py index b3c2fd3a5..d4e6e2aa8 100644 --- a/volatility3/framework/plugins/banners.py +++ b/volatility3/framework/plugins/banners.py @@ -22,7 +22,12 @@ class Banners(interfaces.plugins.PluginInterface): return [ requirements.TranslationLayerRequirement( name="primary", description="Memory layer to scan" - ) + ), + requirements.VersionRequirement( + name="regex_scanner", + component=scanners.RegExScanner, + version=(1, 0, 0), + ), ] def _generator(self): diff --git a/volatility3/framework/plugins/linux/bash.py b/volatility3/framework/plugins/linux/bash.py index 2a63ac329..382b66194 100644 --- a/volatility3/framework/plugins/linux/bash.py +++ b/volatility3/framework/plugins/linux/bash.py @@ -40,6 +40,16 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface): component=timeliner.TimeLinerInterface, version=(1, 0, 0), ), + requirements.VersionRequirement( + name="multi_string_scanner", + component=scanners.MultiStringScanner, + version=(1, 0, 0), + ), + requirements.VersionRequirement( + name="bytes_scanner", + component=scanners.BytesScanner, + version=(1, 0, 0), + ), requirements.ListRequirement( name="pid", element_type=int, diff --git a/volatility3/framework/plugins/linux/check_modules.py b/volatility3/framework/plugins/linux/check_modules.py index ec6f0b73d..7805bbd8a 100644 --- a/volatility3/framework/plugins/linux/check_modules.py +++ b/volatility3/framework/plugins/linux/check_modules.py @@ -46,6 +46,11 @@ class Check_modules(plugins.PluginInterface): @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: return [ + requirements.VersionRequirement( + name="modules", + component=linux_utilities_modules.Modules, + version=(3, 0, 1), + ), requirements.VersionRequirement( name="linux_utilities_modules_module_display_plugin", component=linux_utilities_modules.ModuleDisplayPlugin, diff --git a/volatility3/framework/plugins/linux/hidden_modules.py b/volatility3/framework/plugins/linux/hidden_modules.py index 136aafdd9..dcd602c5d 100644 --- a/volatility3/framework/plugins/linux/hidden_modules.py +++ b/volatility3/framework/plugins/linux/hidden_modules.py @@ -93,6 +93,11 @@ class Hidden_modules(plugins.PluginInterface): component=linux_utilities_modules.ModuleDisplayPlugin, version=(1, 0, 0), ), + requirements.VersionRequirement( + name="linux_utilities_modules", + component=linux_utilities_modules.Modules, + version=(3, 0, 1), + ), ] + linux_utilities_modules.ModuleDisplayPlugin.get_requirements() @staticmethod diff --git a/volatility3/framework/plugins/linux/psscan.py b/volatility3/framework/plugins/linux/psscan.py index 0813cebed..0013bc1d8 100644 --- a/volatility3/framework/plugins/linux/psscan.py +++ b/volatility3/framework/plugins/linux/psscan.py @@ -41,6 +41,11 @@ class PsScan(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="pslist", component=pslist.PsList, version=(4, 0, 0) ), + requirements.VersionRequirement( + name="multi_string_scanner", + component=scanners.MultiStringScanner, + version=(1, 0, 0), + ), ] def _generator(self): diff --git a/volatility3/framework/plugins/linux/vmaregexscan.py b/volatility3/framework/plugins/linux/vmaregexscan.py index 4c8ef5b8f..37a1a5940 100644 --- a/volatility3/framework/plugins/linux/vmaregexscan.py +++ b/volatility3/framework/plugins/linux/vmaregexscan.py @@ -46,6 +46,11 @@ class VmaRegExScan(plugins.PluginInterface): requirements.StringRequirement( name="pattern", description="RegEx pattern", optional=False ), + requirements.VersionRequirement( + name="regex_scanner", + component=scanners.RegExScanner, + version=(1, 0, 0), + ), requirements.IntRequirement( name="maxsize", description="Maximum size in bytes for displayed context", diff --git a/volatility3/framework/plugins/mac/bash.py b/volatility3/framework/plugins/mac/bash.py index 4cbade1cf..5ad6facd0 100644 --- a/volatility3/framework/plugins/mac/bash.py +++ b/volatility3/framework/plugins/mac/bash.py @@ -38,6 +38,16 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface): component=timeliner.TimeLinerInterface, version=(1, 0, 0), ), + requirements.VersionRequirement( + name="multi_string_scanner", + component=scanners.MultiStringScanner, + version=(1, 0, 0), + ), + requirements.VersionRequirement( + name="bytes_scanner", + component=scanners.BytesScanner, + version=(1, 0, 0), + ), requirements.ListRequirement( name="pid", description="Filter on specific process IDs", diff --git a/volatility3/framework/plugins/mac/list_files.py b/volatility3/framework/plugins/mac/list_files.py index bf3dcfce6..423e2e0da 100644 --- a/volatility3/framework/plugins/mac/list_files.py +++ b/volatility3/framework/plugins/mac/list_files.py @@ -31,6 +31,11 @@ class List_Files(plugins.PluginInterface): requirements.VersionRequirement( name="mount", component=mount.Mount, version=(2, 0, 0) ), + requirements.VersionRequirement( + name="mac_utilities", + component=mac.MacUtilities, + version=(1, 3, 0), + ), ] @classmethod diff --git a/volatility3/framework/plugins/regexscan.py b/volatility3/framework/plugins/regexscan.py index c526b1697..343753e92 100644 --- a/volatility3/framework/plugins/regexscan.py +++ b/volatility3/framework/plugins/regexscan.py @@ -39,6 +39,11 @@ class RegExScan(plugins.PluginInterface): default=cls.MAXSIZE_DEFAULT, optional=True, ), + requirements.VersionRequirement( + name="regex_scanner", + component=scanners.RegExScanner, + version=(1, 0, 0), + ), ] def _generator(self, regex_pattern): diff --git a/volatility3/framework/plugins/vmscan.py b/volatility3/framework/plugins/vmscan.py index 64377d7d8..5322456b5 100644 --- a/volatility3/framework/plugins/vmscan.py +++ b/volatility3/framework/plugins/vmscan.py @@ -26,6 +26,8 @@ class VMCSTest(enum.IntFlag): class PageStartScanner(interfaces.layers.ScannerInterface): + _version = (1, 0, 0) + def __init__(self, signatures: List[bytes], page_size: int = 0x1000): super().__init__() if not len(signatures): @@ -69,6 +71,11 @@ class Vmscan(plugins.PluginInterface): requirements.TranslationLayerRequirement( name="primary", description="Physical base memory layer" ), + requirements.VersionRequirement( + name="page_start_scanner", + component=PageStartScanner, + version=(1, 0, 0), + ), requirements.IntRequirement( name="log-threshold", description="Number of criteria failed to log to debug output", diff --git a/volatility3/framework/plugins/windows/cmdscan.py b/volatility3/framework/plugins/windows/cmdscan.py index 8c477b57d..676050b65 100644 --- a/volatility3/framework/plugins/windows/cmdscan.py +++ b/volatility3/framework/plugins/windows/cmdscan.py @@ -41,6 +41,11 @@ class CmdScan(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="consoles", component=consoles.Consoles, version=(3, 0, 0) ), + requirements.VersionRequirement( + name="bytes_scanner", + component=scanners.BytesScanner, + version=(1, 0, 0), + ), requirements.BooleanRequirement( name="no_registry", description="Don't search the registry for possible values of CommandHistorySize", diff --git a/volatility3/framework/plugins/windows/consoles.py b/volatility3/framework/plugins/windows/consoles.py index 6cfc6d588..efc03ad1b 100644 --- a/volatility3/framework/plugins/windows/consoles.py +++ b/volatility3/framework/plugins/windows/consoles.py @@ -54,6 +54,11 @@ class Consoles(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="hivelist", component=hivelist.HiveList, version=(2, 0, 0) ), + requirements.VersionRequirement( + name="bytes_scanner", + component=scanners.BytesScanner, + version=(1, 0, 0), + ), requirements.BooleanRequirement( name="no_registry", description="Don't search the registry for possible values of CommandHistorySize and HistoryBufferMax", diff --git a/volatility3/framework/plugins/windows/mbrscan.py b/volatility3/framework/plugins/windows/mbrscan.py index 4d5198181..541aae60d 100644 --- a/volatility3/framework/plugins/windows/mbrscan.py +++ b/volatility3/framework/plugins/windows/mbrscan.py @@ -37,6 +37,11 @@ class MBRScan(interfaces.plugins.PluginInterface): default=False, optional=True, ), + requirements.VersionRequirement( + name="multi_string_scanner", + component=scanners.MultiStringScanner, + version=(1, 0, 0), + ), ] @classmethod diff --git a/volatility3/framework/plugins/windows/poolscanner.py b/volatility3/framework/plugins/windows/poolscanner.py index 975ed2326..7929b70e4 100644 --- a/volatility3/framework/plugins/windows/poolscanner.py +++ b/volatility3/framework/plugins/windows/poolscanner.py @@ -55,6 +55,8 @@ class PoolConstraint: class PoolHeaderScanner(interfaces.layers.ScannerInterface): + _version = (1, 0, 0) + def __init__( self, module: interfaces.context.ModuleInterface, @@ -142,6 +144,11 @@ class PoolScanner(plugins.PluginInterface): requirements.VersionRequirement( name="handles", component=handles.Handles, version=(3, 0, 0) ), + requirements.VersionRequirement( + name="pool_header_scanner", + component=PoolHeaderScanner, + version=(1, 0, 0), + ), ] def _generator(self): diff --git a/volatility3/framework/plugins/windows/skeleton_key_check.py b/volatility3/framework/plugins/windows/skeleton_key_check.py index 4831362fd..6071a2a39 100644 --- a/volatility3/framework/plugins/windows/skeleton_key_check.py +++ b/volatility3/framework/plugins/windows/skeleton_key_check.py @@ -63,6 +63,11 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="pe_symbols", component=pe_symbols.PESymbols, version=(3, 0, 0) ), + requirements.VersionRequirement( + name="bytes_scanner", + component=scanners.BytesScanner, + version=(1, 0, 0), + ), ] def _check_for_skeleton_key_vad( diff --git a/volatility3/framework/plugins/windows/svclist.py b/volatility3/framework/plugins/windows/svclist.py index 24ac2278f..963b7fc71 100644 --- a/volatility3/framework/plugins/windows/svclist.py +++ b/volatility3/framework/plugins/windows/svclist.py @@ -42,6 +42,11 @@ class SvcList(svcscan.SvcScan): description="Windows kernel", architectures=["Intel32", "Intel64"], ), + requirements.VersionRequirement( + name="bytes_scanner", + component=scanners.BytesScanner, + version=(1, 0, 0), + ), ] @classmethod diff --git a/volatility3/framework/plugins/windows/svcscan.py b/volatility3/framework/plugins/windows/svcscan.py index 94ce02897..5f0e4761e 100644 --- a/volatility3/framework/plugins/windows/svcscan.py +++ b/volatility3/framework/plugins/windows/svcscan.py @@ -56,6 +56,11 @@ class SvcScan(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="hivelist", component=hivelist.HiveList, version=(2, 0, 0) ), + requirements.VersionRequirement( + name="bytes_scanner", + component=scanners.BytesScanner, + version=(1, 0, 0), + ), ] @classmethod diff --git a/volatility3/framework/plugins/windows/vadregexscan.py b/volatility3/framework/plugins/windows/vadregexscan.py index 9b666cbcb..5ead4e453 100644 --- a/volatility3/framework/plugins/windows/vadregexscan.py +++ b/volatility3/framework/plugins/windows/vadregexscan.py @@ -41,6 +41,11 @@ class VadRegExScan(plugins.PluginInterface): element_type=int, optional=True, ), + requirements.VersionRequirement( + name="regex_scanner", + component=scanners.RegExScanner, + version=(1, 0, 0), + ), requirements.StringRequirement( name="pattern", description="RegEx pattern", optional=False ), diff --git a/volatility3/framework/plugins/windows/verinfo.py b/volatility3/framework/plugins/windows/verinfo.py index b5eba7ec6..d2722418b 100644 --- a/volatility3/framework/plugins/windows/verinfo.py +++ b/volatility3/framework/plugins/windows/verinfo.py @@ -48,6 +48,11 @@ class VerInfo(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="modules", component=modules.Modules, version=(3, 0, 0) ), + requirements.VersionRequirement( + name="page_start_scanner", + component=scanners.BytesScanner, + version=(1, 0, 0), + ), requirements.BooleanRequirement( name="extensive", description="Search physical layer for version information", diff --git a/volatility3/framework/plugins/yarascan.py b/volatility3/framework/plugins/yarascan.py index 38c8b6085..bb86ab6f1 100644 --- a/volatility3/framework/plugins/yarascan.py +++ b/volatility3/framework/plugins/yarascan.py @@ -118,7 +118,12 @@ class YaraScan(plugins.PluginInterface): name="primary", description="Memory layer for the kernel", architectures=["Intel32", "Intel64"], - ) + ), + requirements.VersionRequirement( + name="yarascanner", + component=YaraScanner, + version=(2, 1, 1), + ), ] @classmethod From 68116556a8fbe01d63e7d8866d8b9330902d0bed Mon Sep 17 00:00:00 2001 From: David McDonald Date: Thu, 27 Mar 2025 12:54:08 -0500 Subject: [PATCH 03/11] Add calls to super().get_requirements() on inherited classes --- volatility3/cli/volshell/linux.py | 2 +- volatility3/cli/volshell/mac.py | 2 +- volatility3/cli/volshell/windows.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/volatility3/cli/volshell/linux.py b/volatility3/cli/volshell/linux.py index 761b64084..8b9e236b4 100644 --- a/volatility3/cli/volshell/linux.py +++ b/volatility3/cli/volshell/linux.py @@ -41,7 +41,7 @@ class Volshell(generic.Volshell): component=generic.Volshell, version=(1, 0, 0), ), - ] + ] + super().get_requirements() def change_task(self, pid=None): """Change the current process and layer, based on a process ID""" diff --git a/volatility3/cli/volshell/mac.py b/volatility3/cli/volshell/mac.py index fcb45e124..190c7b9f7 100644 --- a/volatility3/cli/volshell/mac.py +++ b/volatility3/cli/volshell/mac.py @@ -30,7 +30,7 @@ class Volshell(generic.Volshell): component=generic.Volshell, version=(1, 0, 0), ), - ] + ] + super().get_requirements() def change_task(self, pid=None): """Change the current process and layer, based on a process ID""" diff --git a/volatility3/cli/volshell/windows.py b/volatility3/cli/volshell/windows.py index a8c7af5b3..e7e37ed61 100644 --- a/volatility3/cli/volshell/windows.py +++ b/volatility3/cli/volshell/windows.py @@ -28,7 +28,7 @@ class Volshell(generic.Volshell): component=generic.Volshell, version=(1, 0, 0), ), - ] + ] + super().get_requirements() def change_process(self, pid=None): """Change the current process and layer, based on a process ID""" From 9f024cf0f485cae6c03a4fb980687c1930f97a51 Mon Sep 17 00:00:00 2001 From: David McDonald Date: Thu, 27 Mar 2025 17:30:17 -0500 Subject: [PATCH 04/11] Refactor: use builtin ast lib instead of treesitter Instead of using the tree-sitter third party library, this uses Python's `ast` module to parse the source code and traverse the tree with a visitor pattern. This is preferred because it's native to the language itself, and Python developers are more likely to be familiar with it. The traversal also handles nested scopes better than the prior implementation. For example, classes that are declared inside of other classes can now be looked up even though they don't exist at the top level of the module namespace, since any time a class definition is entered, that class is pushed to the top of a stack that can be examined when visiting inner classes. This also adds lots of log messages at different levels, plus a command line argument for specifying verbosity, which should help with debugging down the line. --- pyproject.toml | 2 - test/check_configurable_requirements.py | 464 +++++++++++++----------- 2 files changed, 253 insertions(+), 213 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 8bc3693a2..abd2e79f7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,8 +46,6 @@ test = [ "volatility3[dev]", "pytest>=8.3.3,<9", "yara-x>=0.10.0,<1", - "tree-sitter==0.21.3", - "tree-sitter-python==0.21.0", ] docs = [ diff --git a/test/check_configurable_requirements.py b/test/check_configurable_requirements.py index e21df18ac..d532d98b3 100644 --- a/test/check_configurable_requirements.py +++ b/test/check_configurable_requirements.py @@ -1,16 +1,57 @@ +import argparse +import ast import importlib import inspect +import logging import pkgutil import sys -import traceback import types -from textwrap import dedent -from typing import Dict, Iterator, List, NamedTuple, Optional, Tuple, Type - -from tree_sitter import Language, Node, Parser -from tree_sitter_python import language as python_language +from typing import Any, Iterator, NamedTuple, Optional, Tuple, Type, Union from volatility3.framework import configuration, interfaces +from volatility3.framework.deprecation import PluginRenameClass + +logging.basicConfig(format="%(levelname)s: %(message)s") +logger = logging.getLogger(__name__) + + +class NodeVisitor: + def visit(self, node): + """Visit a node.""" + method = "visit_" + node.__class__.__name__ + visitor = getattr(self, method, self.generic_visit) + self.enter(node) + result = visitor(node) + self.leave(node) + return result + + def enter(self, node): + """Called when entering a node.""" + method = "enter_" + node.__class__.__name__ + visitor = getattr(self, method, self.generic_enter) + return visitor(node) + + def leave(self, node): + """Called when leaving a node.""" + method = "leave_" + node.__class__.__name__ + visitor = getattr(self, method, self.generic_leave) + return visitor(node) + + def generic_visit(self, node): + """Called if no explicit visitor function exists for a node.""" + for _, value in ast.iter_fields(node): + if isinstance(value, list): + for item in value: + if isinstance(item, ast.AST): + self.visit(item) + elif isinstance(value, ast.AST): + self.visit(value) + + def generic_enter(self, node): + """Default enter behavior.""" + + def generic_leave(self, node): + """Default leave behavior.""" class UnrequiredVersionableUsage(NamedTuple): @@ -24,12 +65,7 @@ class UnrequiredVersionableUsage(NamedTuple): The name of the class that is using the imported VersionableInterface class """ - methodname: Optional[str] - """ - The name of the invoked method or attribute, if one is used or referenced - """ - - node: Node + node: Union[ast.Name, ast.Attribute] """ The tree-sitter node encapsulating the used module component. """ @@ -42,149 +78,13 @@ class UnrequiredVersionableUsage(NamedTuple): ) -class RequirementValidator: - language = Language(python_language(), "python") - - def __init__(self, plugin_module: types.ModuleType) -> None: - if plugin_module.__file__ is None: - raise ValueError("Attempting to validate a module without a file") - - self._module = plugin_module - - # See which classes in *this* module are configurable (can have requirements declared) - self._configurable_classes = get_configurable_classes(plugin_module) - - # Get a mapping of class names to configurable classes that they declare in their requirements - self._versioned_item_mapping = get_versioned_item_mapping( - self._configurable_classes - ) - - # Get a mapping of module name -> versionable classes within the namespace of each module - self._imported_mod_classes = get_versionable_import_mapping( - get_imported_modules(plugin_module) - ) - - with open(plugin_module.__file__, "rb") as f: - source = f.read() - - self._parser = Parser() - self._parser.set_language(self.language) - self._tree = self._parser.parse(source) - - def enumerate_unrequired_usages( - self, - clazz: Type[interfaces.configuration.ConfigurableInterface], - class_node: Node, - ): - - # This query is designed to look for three different identifier usages: - # simple identifiers: PsList - # module attrs: pslist.PsList - # method calls: pslist.PsList.list_processes - obj_query = self.language.query( - dedent( - """ - [ - (identifier) - (attribute - object: (identifier) - attribute: (identifier)) - (attribute - object: (attribute - object: (identifier) - attribute: (identifier)) - ) - ] @ident - """ - ) - ) - - containing_name = class_node.child_by_field_name("name").text.decode("utf-8") - - valid_types = self._versioned_item_mapping[containing_name] - for _, match in obj_query.matches(class_node): - if "ident" not in match: - continue - - # Get the raw text of the match. This could be something like - # - PsList - # - pslist.PsList - # - pslist.PsList.list_processes - ident_text = match["ident"].text.decode("utf-8") - - # split the attributes - components = ident_text.split(".") - try: - # See if the first attribute is in the module namespace. - item = vars(self._module)[components[0]] - except KeyError: - # If it's not, it's likely a variable in a smaller scope and we - # can ignore it. - continue - - # If it's in the module namespace and is a module... - if isinstance(item, types.ModuleType): - try: - # We try getting attributes from it until we - # find one that is a versionable class - - # Ideally, we shouldn't have to look further than - # two levels - item = getattr(item, components[1]) - if not is_versionable(item): - item = getattr(item, components[2]) - if not is_versionable(item): - continue - - except (IndexError, AttributeError): - # we ran out of attributes to check - continue - - elif is_versionable(item): - # The versionable thing was at the top level. This - # goes against our preferred style, but is possible. - pass - else: - # This isn't something we care about. - continue - - if ( - item in valid_types - or item is clazz - or inspect.isabstract(item) - or item - is interfaces.configuration.VersionableInterface # Avoid checking the interface itself - ): - continue - - yield UnrequiredVersionableUsage( - item, - containing_name, - components[1] if len(components) > 1 else None, - match["ident"], - ) - - def find_class_nodes( - self, - ) -> Iterator[Tuple[Type[interfaces.configuration.ConfigurableInterface], Node]]: - """ - Yields an iterator of (classname, node) tuples, where the node is the subtree containing - the entire class definition. - """ - class_query = self.language.query("(class_definition) @classdef") - - matches = class_query.captures(self._tree.root_node) - for node, _ in matches: - classname = node.child_by_field_name("name").text.decode("utf-8") - if classname not in self._configurable_classes: - continue - - yield self._configurable_classes[classname], node - - def is_versionable(var): try: - return issubclass(var, interfaces.configuration.VersionableInterface) + return ( + issubclass(var, interfaces.configuration.VersionableInterface) + and var is not interfaces.configuration.VersionableInterface + and not inspect.isabstract(var) + ) except TypeError: return False @@ -196,48 +96,162 @@ def is_configurable(var): return False -def get_imported_modules( - plugin_module: types.ModuleType, -) -> List[Tuple[str, types.ModuleType]]: - return [ - (name, var) - for name, var in vars(plugin_module).items() - if isinstance(var, types.ModuleType) - ] +class ModuleVisitor(NodeVisitor): + def __init__(self, module: types.ModuleType) -> None: + self._module = module + self._scopes = [] + self._violations = [] + + @property + def violations(self): + return self._violations + + def enter_ClassDef(self, node: ast.ClassDef) -> Any: + logger.debug("Entering class %s", node.name) + clazz = None + try: + clazz = vars(self._module)[str(node.name)] + except KeyError: + logger.debug( + "Failed to get %s from module scope: (%s)", + node.name, + self._module.__name__, + ) + if self._scopes: + try: + logger.debug( + "Attempting to get class %s from scope of %s", + node.name, + self._scopes[-1].__name__, + ) + clazz = getattr(self._scopes[-1], node.name) + except AttributeError: + logger.debug( + "Class not found in scope of %s", self._scopes[-1].__name__ + ) + if clazz: + self._scopes.append(clazz) + + if clazz and is_configurable(clazz): + logger.info("Checking configurable class %s", clazz.__name__) + visitor = ConfigurableClassVisitor(self._module, clazz) + visitor.visit(node) + self._violations += visitor.violations + + self.generic_visit(node) + + def leave_ClassDef(self, node: ast.ClassDef): + logger.debug("Leaving class %s", node.name) + try: + scoped_class = next( + scope for scope in self._scopes if scope.__name__ == node.name + ) + self._scopes.remove(scoped_class) + except StopIteration: + logger.debug("%s not found in scope list", node.name) -def get_configurable_classes( - plugin_module: types.ModuleType, -) -> Dict[str, Type[interfaces.configuration.ConfigurableInterface]]: - return { - name: clazz - for name, clazz in vars(plugin_module).items() - if is_configurable(clazz) - } +class ConfigurableClassVisitor(NodeVisitor): + def __init__( + self, + module: types.ModuleType, + clazz: Optional[Type[interfaces.configuration.ConfigurableInterface]], + ) -> None: + self._module = module + self._current_object = None + self._clazz = clazz + self._seen = set() + self._violations = [] + @property + def versioned_classes(self): + return ( + [ + req._component + for req in self._clazz.get_requirements() + if isinstance(req, configuration.requirements.VersionRequirement) + ] + if self._clazz is not None + else [] + ) -def get_versioned_item_mapping( - configurable_classes: Dict[ - str, Type[interfaces.configuration.ConfigurableInterface] - ] -) -> Dict[str, List[Type[interfaces.configuration.VersionableInterface]]]: - return { - name: [ - req._component - for req in clazz.get_requirements() - if isinstance(req, configuration.requirements.VersionRequirement) - ] - for name, clazz in configurable_classes.items() - } + def check_item(self, item: Type, node: Union[ast.Name, ast.Attribute]): + if ( + is_versionable(item) + and self._clazz is not None + and item not in self.versioned_classes + and item is not self._clazz + and not issubclass(self._clazz, PluginRenameClass) + ): + logger.info( + "Found versionable item %s, checking against %s", + str(item), + str(self.versioned_classes), + ) + result = UnrequiredVersionableUsage( + item.__name__, self._clazz.__name__, node + ) + self._violations.append(result) + @property + def violations(self): + return self._violations -def get_versionable_import_mapping( - imported_modules: List[Tuple[str, types.ModuleType]] -) -> Dict[str, List[str]]: - return { - modname: [name for name, var in vars(module).items() if is_versionable(var)] - for modname, module in imported_modules - } + def visit_Name(self, node: ast.Name): + try: + logger.debug( + "Checking module %s for name %s", self._module.__name__, node.id + ) + item = vars(self._module)[str(node.id)] + logger.debug("Found %s in %s namespace", node.id, self._module.__name__) + except KeyError: + return + + self.check_item(item, node) + + def visit_Attribute( + self, node: ast.Attribute + ) -> Optional[UnrequiredVersionableUsage]: + if self._clazz is None: + self.generic_visit(node) + return + + if (node.lineno, node.col_offset) in self._seen: + return + + self._seen.add((node.lineno, node.col_offset)) + + stack = [] + root = node + while True: + stack.append(node.attr) + if isinstance(node.value, ast.Attribute): + node = node.value + elif isinstance(node.value, ast.Name): + stack.append(node.value.id) + break + else: + break + + current = None + logger.debug("Checking %s", ".".join(stack[::-1])) + for item in stack[::-1]: + try: + current = ( + vars(self._module)[item] + if current is None + else getattr(current, item) + ) + except (KeyError, AttributeError) as exc: + logger.debug( + "Failed to get attribute %s (%s)%s", + item, + exc.__class__.__name__, + (" on" + str(current)) if current is not None else "", + ) + break + + self.check_item(current, root) def report_missing_requirements() -> Iterator[Tuple[str, UnrequiredVersionableUsage]]: @@ -246,46 +260,60 @@ def report_missing_requirements() -> Iterator[Tuple[str, UnrequiredVersionableUs for _, module_name, _ in pkgutil.walk_packages( vol3.__path__, vol3.__name__ + ".", onerror=lambda _: None ): + modname = module_name.replace( + "volatility3.framework.plugins", "volatility3.plugins" + ) try: # import the module that we want to check - modname = module_name.replace( - "volatility3.framework.plugins", "volatility3.plugins" - ) plugin_module = importlib.import_module(modname) - except ImportError: + except ImportError as exc: + logger.warning("Failed to import %s: %s", modname, str(exc)) continue - except Exception: + except Exception as exc: + logger.warning( + "An unexpected exception occurred while importing %s: %s", + modname, + str(exc), + ) continue + logger.info("Checking module %s", plugin_module.__name__) if plugin_module.__file__ is None: + logger.warning("Plugin module %s has no source file", modname) continue try: - # construct a validator for the module - try: - validator = RequirementValidator(plugin_module) - except Exception: - traceback.print_stack() - continue - for clazz, node in validator.find_class_nodes(): - for item in validator.enumerate_unrequired_usages(clazz, node): - yield module_name, item - except Exception as exc: - traceback.print_exc() - print( - f"Failed to create validator for source code from {plugin_module.__file__}: {exc}" + with open(plugin_module.__file__, "rb") as f: + source = f.read() + except OSError: + logger.warning( + "Failed to read file contents for %s", plugin_module.__file__ + ) + continue + + try: + module_ast_root = ast.parse(source) + except (SyntaxError, ValueError) as exc: + logger.warning( + "Failed to parse source for %s: %s", plugin_module.__file__, str(exc) + ) + raise + + mod_visitor = ModuleVisitor(plugin_module) + mod_visitor.visit(module_ast_root) + + if mod_visitor.violations: + yield from ( + (plugin_module.__name__, res) for res in iter(mod_visitor.violations) ) - sys.exit(1) def perform_review(): found = 0 for mod, usage in report_missing_requirements(): found += 1 - print( - f"Violation in module {mod} (line {usage.node.start_point[0]}): {str(usage)}" - ) + print(f"Violation in module {mod} (line {usage.node.lineno}): {str(usage)}") if found: print( @@ -296,5 +324,19 @@ def perform_review(): print("All configurable classes passed validation!") +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("-v", "--verbose", action="count", dest="verbosity", default=0) + return parser.parse_args() + + if __name__ == "__main__": + args = parse_args() + if args.verbosity == 0: + logger.setLevel(logging.WARNING) + elif args.verbosity == 1: + logger.setLevel(logging.INFO) + elif args.verbosity > 1: + logger.setLevel(logging.DEBUG) + perform_review() From 03c647790206267ecf4c3d4921b2a0164ea4dcd1 Mon Sep 17 00:00:00 2001 From: David McDonald Date: Fri, 28 Mar 2025 13:04:24 -0500 Subject: [PATCH 05/11] Volshell: Attempt to resolve requirement conflicts This change sets the `script`, `script-only`, and `primary` requirements to only apply to the `generic.Volshell` class. `regex-scanner` is okay to be shared between the base and inherited classes, but `script` and `script-only` have to be generic-only in order to avoid conflicts when populating the argparse parser. `primary` must be generic-only in order to avoid ending up unsatisfied when superclass requirements require a module, suppressing construction of the `primary` layer. --- volatility3/cli/volshell/generic.py | 25 +++++++++++++------------ 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/volatility3/cli/volshell/generic.py b/volatility3/cli/volshell/generic.py index 1b3ae59d1..39e4fc963 100644 --- a/volatility3/cli/volshell/generic.py +++ b/volatility3/cli/volshell/generic.py @@ -54,20 +54,24 @@ class Volshell(interfaces.plugins.PluginInterface): @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: - reqs: List[interfaces.configuration.RequirementInterface] = [] + reqs: List[interfaces.configuration.RequirementInterface] = [ + requirements.VersionRequirement( + name="regex_scanner", + component=scanners.RegExScanner, + version=(1, 0, 0), + ), + ] if cls == Volshell: - reqs = [ + reqs += [ + requirements.TranslationLayerRequirement( + name="primary", description="Memory layer for the kernel" + ), requirements.URIRequirement( name="script", description="File to load and execute at start", default=None, optional=True, ), - requirements.VersionRequirement( - name="regex_scanner", - component=scanners.RegExScanner, - version=(1, 0, 0), - ), requirements.BooleanRequirement( name="script-only", description="Exit volshell after the script specified in --script completes", @@ -75,11 +79,8 @@ class Volshell(interfaces.plugins.PluginInterface): optional=True, ), ] - return reqs + [ - requirements.TranslationLayerRequirement( - name="primary", description="Memory layer for the kernel" - ), - ] + + return reqs def run( self, additional_locals: Dict[str, Any] = {} From 27e59263a6825886cb5d8f1524f2e8d48955b57f Mon Sep 17 00:00:00 2001 From: David McDonald Date: Fri, 28 Mar 2025 13:17:38 -0500 Subject: [PATCH 06/11] Docstring: explain version-checking script This documents the general behavior and expectations of the version-checking CI script. --- test/check_configurable_requirements.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/test/check_configurable_requirements.py b/test/check_configurable_requirements.py index d532d98b3..f856c80b7 100644 --- a/test/check_configurable_requirements.py +++ b/test/check_configurable_requirements.py @@ -1,3 +1,19 @@ +""" +This script performs syntax analysis on the volatility3 source tree through a combination of AST analysis and import-time introspection of classes. + +The current checks it implements are: + 1. Ensure that classes derived from `ConfigurableInterface` properly + declare all `VersionableInterface` classes that they make use of in their + `get_requirements()` classmethod. + + :WARNING: a notable exception to this are classes defined within factory + functions. Because these classes are not created until the factory function + is called, they therefore do no exist at import time and cannot be checked + by this script. It is important to keep in mind during code review that + this is a best-effort check and does not make guarantees about the + completeness of declared requirements. +""" + import argparse import ast import importlib From d0a1daf82c1a118b1b33958e4cd6dec2243df248 Mon Sep 17 00:00:00 2001 From: David McDonald Date: Fri, 28 Mar 2025 13:24:51 -0500 Subject: [PATCH 07/11] ModuleExtract: Add missing requirement --- volatility3/framework/plugins/linux/module_extract.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/volatility3/framework/plugins/linux/module_extract.py b/volatility3/framework/plugins/linux/module_extract.py index d7c875523..97824aca0 100644 --- a/volatility3/framework/plugins/linux/module_extract.py +++ b/volatility3/framework/plugins/linux/module_extract.py @@ -36,6 +36,11 @@ class ModuleExtract(interfaces.plugins.PluginInterface): description="Base virtual address to reconstruct an ELF file", optional=False, ), + requirements.VersionRequirement( + name="linux_utilities_module_extract", + version=(1, 0, 0), + component=linux_utilities_module_extract.ModuleExtract, + ), ] def _generator(self): From 196556eab3ed0abbffd20bcec169d2f131535426 Mon Sep 17 00:00:00 2001 From: David McDonald Date: Fri, 28 Mar 2025 15:18:22 -0500 Subject: [PATCH 08/11] Test: Allow for other types of coding style violations --- test/check_configurable_requirements.py | 51 +++++++++++++++---------- 1 file changed, 31 insertions(+), 20 deletions(-) diff --git a/test/check_configurable_requirements.py b/test/check_configurable_requirements.py index f856c80b7..89e06e751 100644 --- a/test/check_configurable_requirements.py +++ b/test/check_configurable_requirements.py @@ -14,6 +14,7 @@ The current checks it implements are: completeness of declared requirements. """ +import abc import argparse import ast import importlib @@ -22,7 +23,7 @@ import logging import pkgutil import sys import types -from typing import Any, Iterator, NamedTuple, Optional, Tuple, Type, Union +from typing import Any, Iterator, List, Optional, Tuple, Type, Union from volatility3.framework import configuration, interfaces from volatility3.framework.deprecation import PluginRenameClass @@ -70,27 +71,37 @@ class NodeVisitor: """Default leave behavior.""" -class UnrequiredVersionableUsage(NamedTuple): - versionable_item_class: str - """ - The name of the VersionableInterface class - """ +class CodeViolation(metaclass=abc.ABCMeta): + def __init__(self, module: types.ModuleType, node: ast.AST) -> None: + self.module = module + self.node = node - consuming_class: str - """ - The name of the class that is using the imported VersionableInterface class - """ + def __str__(self): + return f"Code violation in module {self.module.__name__}: line {self.node.lineno}, col {self.node.col_offset}" - node: Union[ast.Name, ast.Attribute] - """ - The tree-sitter node encapsulating the used module component. - """ + +class UnrequiredVersionableUsage(CodeViolation): + + def __init__( + self, + module: types.ModuleType, + node: ast.AST, + consuming_class: str, + versionable_item_class: str, + ) -> None: + super().__init__(module, node) + self.consuming_class = consuming_class + self.versionable_item_class = versionable_item_class def __str__(self) -> str: return ( - f"Found usage of {self.versionable_item_class} " - f"in class {self.consuming_class} that is not declared " - f"in {self.consuming_class}'s `get_requirements()` classmethod" + super().__str__() + + ": " + + ( + f"Found usage of {self.versionable_item_class} " + f"in class {self.consuming_class} that is not declared " + f"in {self.consuming_class}'s `get_requirements()` classmethod" + ) ) @@ -177,7 +188,7 @@ class ConfigurableClassVisitor(NodeVisitor): self._current_object = None self._clazz = clazz self._seen = set() - self._violations = [] + self._violations: List[CodeViolation] = [] @property def versioned_classes(self): @@ -205,7 +216,7 @@ class ConfigurableClassVisitor(NodeVisitor): str(self.versioned_classes), ) result = UnrequiredVersionableUsage( - item.__name__, self._clazz.__name__, node + self._module, node, self._clazz.__name__, item.__name__ ) self._violations.append(result) @@ -333,7 +344,7 @@ def perform_review(): if found: print( - f"Found {found} uses of versionable components not declared in get_requirements()" + f"Found {found} coding standards violations" ) sys.exit(1) From 46e3b8ffdb4e9c4b536e2a6fc8217f2be3d77c4c Mon Sep 17 00:00:00 2001 From: David McDonald Date: Fri, 28 Mar 2025 15:21:54 -0500 Subject: [PATCH 09/11] Check for 'hidden' attribute when determining classes to validate --- test/check_configurable_requirements.py | 1 + 1 file changed, 1 insertion(+) diff --git a/test/check_configurable_requirements.py b/test/check_configurable_requirements.py index 89e06e751..ce65aca48 100644 --- a/test/check_configurable_requirements.py +++ b/test/check_configurable_requirements.py @@ -111,6 +111,7 @@ def is_versionable(var): issubclass(var, interfaces.configuration.VersionableInterface) and var is not interfaces.configuration.VersionableInterface and not inspect.isabstract(var) + and not (hasattr(var, "hidden") and getattr(var, "hidden") is True) ) except TypeError: return False From d7695ab9cfb507b3bb3e791f00bcd082f5de3afc Mon Sep 17 00:00:00 2001 From: David McDonald Date: Fri, 28 Mar 2025 15:36:54 -0500 Subject: [PATCH 10/11] Simplify error message output --- test/check_configurable_requirements.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/check_configurable_requirements.py b/test/check_configurable_requirements.py index ce65aca48..b643facb5 100644 --- a/test/check_configurable_requirements.py +++ b/test/check_configurable_requirements.py @@ -341,7 +341,7 @@ def perform_review(): found = 0 for mod, usage in report_missing_requirements(): found += 1 - print(f"Violation in module {mod} (line {usage.node.lineno}): {str(usage)}") + print(str(usage)) if found: print( From 6452fc18bd6cb614e8eaa747bc2d5be36a224e52 Mon Sep 17 00:00:00 2001 From: David McDonald Date: Fri, 28 Mar 2025 15:39:09 -0500 Subject: [PATCH 11/11] Tone down language severity in messages --- test/check_configurable_requirements.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/check_configurable_requirements.py b/test/check_configurable_requirements.py index b643facb5..864c3e89e 100644 --- a/test/check_configurable_requirements.py +++ b/test/check_configurable_requirements.py @@ -77,7 +77,7 @@ class CodeViolation(metaclass=abc.ABCMeta): self.node = node def __str__(self): - return f"Code violation in module {self.module.__name__}: line {self.node.lineno}, col {self.node.col_offset}" + return f"Issue in module {self.module.__name__}: line {self.node.lineno}, col {self.node.col_offset}" class UnrequiredVersionableUsage(CodeViolation): @@ -345,7 +345,7 @@ def perform_review(): if found: print( - f"Found {found} coding standards violations" + f"Found {found} issues" ) sys.exit(1)