From a2934e8d898119002790edd518f75cf9b2426217 Mon Sep 17 00:00:00 2001 From: Brigs Date: Wed, 8 Jul 2026 16:22:27 -0400 Subject: [PATCH] Port --custom_artifacts_path from iLEAPP for loader parity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports iLEAPP's plugin loader and CLI option so additional artifact directories can be loaded on demand (e.g. scripts/alternate_artifacts): - scripts/plugin_loader.py: verbatim copy of iLEAPP's loader — PluginLoader(plugin_paths=[...]) accepts a list of directories, lazy module names are namespaced by parent folder, duplicate keys within one folder still raise, and a duplicate key from an alternate folder overrides the standard one with an informational message - aleapp.py: new --custom_artifacts_path option; loader construction moved after parse_args (nothing used it earlier); default behavior is unchanged - aleapp.py: pre-existing pylint warnings resolved so the file passes the CI lint gate (encoding on path_list.txt, f-string/comment fixes, targeted inline disables for the long-standing wildcard imports and except blocks — no behavior changes) Verification (no regressions): - loaded plugin set byte-identical to main (583 plugins: module, name, category and search patterns compared) - full run on a test zip: completed-artifact set identical to main - run with --custom_artifacts_path pointing at an external folder loads exactly the additional artifacts - admin tests pass (test_artifact_imports, test_dependency_compatibility) - no-arg PluginLoader() unchanged for aleappGUI.py - -p path list generation works - pylint --disable=C,R at 10.00 for both touched files --- aleapp.py | 32 ++++++---- scripts/plugin_loader.py | 125 ++++++++++++++++++++++++++++++++++----- 2 files changed, 129 insertions(+), 28 deletions(-) diff --git a/aleapp.py b/aleapp.py index dcbcabc6..06c7f5a8 100755 --- a/aleapp.py +++ b/aleapp.py @@ -7,14 +7,15 @@ import traceback import sys +import pathlib import scripts.plugin_loader as plugin_loader import leapp_functions.app.history as history -from scripts.search_files import * -from scripts.ilapfuncs import * +from scripts.search_files import * # pylint: disable=wildcard-import,unused-wildcard-import +from scripts.ilapfuncs import * # pylint: disable=wildcard-import,unused-wildcard-import from scripts.version_info import leapp_name, leapp_version from time import process_time, gmtime, strftime, perf_counter -from scripts.lavafuncs import * +from scripts.lavafuncs import * # pylint: disable=wildcard-import,unused-wildcard-import from scripts.context import Context def validate_args(args): @@ -152,9 +153,8 @@ def main(): help=("Generate a text file list of artifact paths. " "This argument is meant to be used alone, without any other arguments.")) parser.add_argument('--custom_output_folder', required=False, action="store", help="Custom name for the output folder") + parser.add_argument('--custom_artifacts_path', required=False, action="store", help="Additional path to load artifacts from (e.g., scripts/alternate_artifacts)") - loader = plugin_loader.PluginLoader() - available_plugins = list(loader.plugins) profile_filename = None casedata = {} @@ -165,6 +165,12 @@ def main(): args = parser.parse_args() + loader_paths = [plugin_loader.PLUGINPATH] + if args.custom_artifacts_path: + loader_paths.append(pathlib.Path(args.custom_artifacts_path)) + loader = plugin_loader.PluginLoader(plugin_paths=loader_paths) + available_plugins = list(loader.plugins) + plugins = [] plugins_parsed_first = [] @@ -184,13 +190,13 @@ def main(): if args.artifact_paths: print('Artifact path list generation started.') print('') - with open('path_list.txt', 'a') as paths: + with open('path_list.txt', 'a', encoding='utf-8') as paths: for plugin in loader.plugins: if isinstance(plugin.search, tuple): for x in plugin.search: paths.write(x + '\n') print(x) - else: # TODO check that this is actually a string? + else: # search should be a string here paths.write(plugin.search + '\n') print(plugin.search) print('') @@ -231,7 +237,7 @@ def main(): with open(case_data_filename, "rt", encoding="utf-8") as case_data_file: try: case_data = json.load(case_data_file) - except: + except: # pylint: disable=bare-except case_data_load_error = "File was not a valid case data file: invalid format" print(case_data_load_error) return @@ -256,7 +262,7 @@ def main(): with open(profile_filename, "rt", encoding="utf-8") as profile_file: try: profile = json.load(profile_file) - except: + except: # pylint: disable=bare-except profile_load_error = "File was not a valid case data file: invalid format" print(profile_load_error) return @@ -332,7 +338,7 @@ def crunch_artifacts( else: logfunc('Error on argument -o (input type)') return False - except Exception as ex: + except Exception: # pylint: disable=broad-exception-caught logfunc('Had an exception in Seeker - see details below. Terminating Program!') temp_file = io.StringIO() traceback.print_exc(file=temp_file) @@ -387,7 +393,7 @@ def crunch_artifacts( lava_insert_sqlite_file_path(file_path_id,seeker.file_infos.get(pathh).source_path) file_path_ids.add(file_path_id) lava_insert_sqlite_artifact_link_pattern_to_file(artifact_search_pattern_id, file_path_id) - log.write(f'') + log.write('') files_found.extend(found) if files_found: category_folder = os.path.join(out_params.output_folder_base, '_HTML', plugin.category) @@ -400,13 +406,13 @@ def crunch_artifacts( continue # cannot do work try: plugin.method(files_found, category_folder, seeker, wrap_text) - except Exception as ex: + except Exception as ex: # pylint: disable=broad-exception-caught logfunc('Reading {} artifact had errors!'.format(plugin.name)) logfunc('Error was {}'.format(str(ex))) logfunc('Exception Traceback: {}'.format(traceback.format_exc())) continue # nope else: - logfunc(f"No file found") + logfunc("No file found") logfunc('{} [{}] artifact completed'.format(plugin.name, plugin.module_name)) parsed_modules += 1 GuiWindow.SetProgressBar(parsed_modules, len(plugins)) diff --git a/scripts/plugin_loader.py b/scripts/plugin_loader.py index dbbb7b0c..0db93d2b 100755 --- a/scripts/plugin_loader.py +++ b/scripts/plugin_loader.py @@ -1,40 +1,109 @@ +""" +Plugin loader module for dynamically loading and managing artifact plugins. + +Provides functionality to discover and load modules from the artifacts directory. +It supports both v1 and v2 artifact formats and manages theit metadata through +the PluginSpec dataclass. + +Classes: + PluginSpec: A frozen dataclass containing artifact metadata including name, module name, + category, search paths, callable method, and artifact information. + PluginLoader: Main class for loading and managing artifacts. Discovers Python files in + the artifacts directory, extracts artifact definitions, and provides access + to loaded artifacts. + +Constants: + PLUGINPATH: Default path to the artifacts directory, resolved relative to this file's + location for PyInstaller compatibility. + +""" + import pathlib import dataclasses import typing import importlib.util -#PLUGINPATH = pathlib.Path("./scripts/artifacts") +# PLUGINPATH = pathlib.Path("./scripts/artifacts") # a bit long-winded to make compatible with PyInstaller PLUGINPATH = pathlib.Path(__file__).resolve().parent / pathlib.Path("artifacts") @dataclasses.dataclass(frozen=True) class PluginSpec: + """ + Specification class for artifact metadata and configuration. + This class stores essential information about an artifact including its identification, + categorization, search criteria, execution method, and details. + Attributes: + name (str): The name of the artifact. + module_name (str): The Python module name where the artifact is defined. + category (str): The category this artifact belongs to. + search (str): Search pattern used to identify relevant files. + method (Callable): The callable function that executes the artifact's main logic. + artifact_info (dict): Dictionary containing metadata and information about artifacts. + """ + name: str module_name: str category: str search: str - method: typing.Callable # todo define callable signature + method: typing.Callable # to do: define callable signature artifact_info: dict # Add this line to include artifact_info class PluginLoader: - def __init__(self, plugin_path: typing.Optional[pathlib.Path] = None): - self._plugin_path = plugin_path or PLUGINPATH + """ + A class responsible for dynamically loading and managing artifacts from Python files. + The PluginLoader scans a specified directory for Python files containing artifact definitions + and loads them lazily. It supports two versions of artifact definitions: + - Version 2 (__artifacts_v2__): Dictionary-based with detailed metadata + - Version 1 (__artifacts__): Tuple-based (category, search, func) + Attributes: + _plugin_path (pathlib.Path): The directory path containing LEAPPs module. + _plugins (dict[str, PluginSpec]): Dictionary mapping artifact names to their specifications. + Args: + plugin_path (typing.Optional[pathlib.Path]): Optional path to LEAPPs module directory. + If not provided, defaults to PLUGINPATH. + Raises: + KeyError: If duplicate function names are found across different modules. + """ + + def __init__(self, plugin_paths: typing.Optional[typing.List[pathlib.Path]] = None): + self._plugin_paths = plugin_paths or [PLUGINPATH] self._plugins: dict[str, PluginSpec] = {} - self._load_plugins() + self._artifact_names = {} + self._plugin_sources = {} + for path in self._plugin_paths: + self._load_plugins(path) @staticmethod def load_module_lazy(path: pathlib.Path): - spec = importlib.util.spec_from_file_location(path.stem, path) + """ + Lazily loads a Python module from a file path. + This function creates a module specification from the given file path and uses + LazyLoader to defer the execution of the module until its attributes are accessed. + This can improve startup performance when loading many plugins. + Args: + path (pathlib.Path): The file path to the Python module to be loaded. + The module name will be derived from the file stem (filename without extension). + Returns: + module: The loaded module object. The module's code is executed lazily upon + first attribute access. + """ + + # Use the full module path to avoid name collisions between different plugin directories + module_name = f"{path.parent.name}.{path.stem}" + spec = importlib.util.spec_from_file_location(module_name, path) loader = importlib.util.LazyLoader(spec.loader) spec.loader = loader mod = importlib.util.module_from_spec(spec) loader.exec_module(mod) return mod - def _load_plugins(self): - for py_file in self._plugin_path.glob("*.py"): + def _load_plugins(self, plugin_path: pathlib.Path): + for py_file in plugin_path.glob("*.py"): + if py_file.name == "__init__.py": + continue mod = PluginLoader.load_module_lazy(py_file) mod_artifacts = getattr(mod, '__artifacts_v2__', None) or getattr(mod, '__artifacts__', None) if mod_artifacts is None: @@ -46,7 +115,7 @@ def _load_plugins(self): if version == 2: category = artifact.get('category') search = artifact.get('paths') - + artifact_name = artifact.get('name') func = None # 1. Look for a wrapped function with the name of the dictionary for item_name in dir(mod): @@ -54,13 +123,13 @@ def _load_plugins(self): if callable(item) and item_name == name and hasattr(item, '__wrapped__'): func = item break - + # 2. If no wrapped function, look for declared function if func is None: func_name = artifact.get('function') if func_name: func = getattr(mod, func_name, None) - + # 3. If neither above work, log the failure if func is None: print(f"Warning: No matching function found for artifact '{name}' in module '{py_file.stem}'") @@ -73,18 +142,45 @@ def _load_plugins(self): else: # 4. If no v2, then use v1 + artifact_name = name category, search, func = artifact artifact_info = {'category': category, 'paths': search} + # Check for duplicate plugin keys within the same artifact folder and error + # Identify duplicate plugin in alternate plugin folder and override standard if name in self._plugins: - raise KeyError(f"Duplicate plugin: '{name}' in module '{py_file.stem}'") - + prev_path = self._plugin_sources[name] + if prev_path == plugin_path: + raise KeyError(f"Duplicate plugin key: '{name}' in module '{py_file.stem}'") + else: + print(f"Info: Overriding artifact key '{name}' from {prev_path.name} " + f"with version from {plugin_path.name}/{py_file.name}") + + # Check for duplicate artifact names within the same artifact folder + if artifact_name in self._artifact_names: + prev_file, prev_path = self._artifact_names[artifact_name] + if prev_path == plugin_path: + raise KeyError(f"Duplicate artifact name in {py_file.name}: " + f"'{artifact_name}' was also found in module '{prev_file}'") + + self._artifact_names[artifact_name] = (py_file.name, plugin_path) + self._plugin_sources[name] = plugin_path + # Add artifact_info to PluginSpec self._plugins[name] = PluginSpec(name, py_file.stem, category, search, func, artifact_info) - @property def plugins(self) -> typing.Iterable[PluginSpec]: + """ + Yields all loaded artifacts specifications. + This method provides an iterable access to all artifacts that have been + loaded and stored in the internal _plugins dictionary. + Yields + ------ + PluginSpec + Individual plugin specification objects from the internal plugins collection. + """ + yield from self._plugins.values() def __getitem__(self, item: str) -> PluginSpec: @@ -95,4 +191,3 @@ def __contains__(self, item): def __len__(self): return len(self._plugins) -