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()