From cb25137423e296bf048bb9017177ad365c341245 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Wed, 19 Aug 2026 14:03:09 +0200 Subject: [PATCH] Load installed module packages via entry points Python packages can already register custom CLI commands which load automatically, but custom modules still require --load-module or the MVT_CUSTOM_MODULES environment variable on every invocation. Add an mvt.modules entry-point group so installed packages can register forensic modules which load automatically into every module-running check-* command. An entry point resolves to an iterable of MVTModule subclasses, or a callable returning one. Broken entry points are skipped with a warning so a faulty package cannot break MVT. --- docs/development.md | 43 +++++++++++++++++ src/mvt/common/module_loader.py | 83 +++++++++++++++++++++++++++++++-- tests/test_custom_modules.py | 59 +++++++++++++++++++++++ 3 files changed, 181 insertions(+), 4 deletions(-) diff --git a/docs/development.md b/docs/development.md index ffe1265..8949dff 100644 --- a/docs/development.md +++ b/docs/development.md @@ -117,6 +117,49 @@ class DependentCustomModule(MVTModule): self.results = [{"manifest_entries": len(manifest_results)}] ``` +## Installed module packages + +Python packages can register modules so they load automatically in every +module-running `check-*` command, without `--load-module` or +`MVT_CUSTOM_MODULES`. Register an entry point in the `mvt.modules` group in +the package's `pyproject.toml`: + +```toml +[project.entry-points."mvt.modules"] +my-mvt-modules = "my_mvt_modules:get_modules" +``` + +The entry point must resolve to an iterable of `MVTModule` subclasses, or to +a callable returning one: + +```python +from mvt.common.module import MVTModule + + +class PackagedModule(MVTModule): + supported_commands = (("ios", "check-backup"),) + + def run(self): + self.results = [{"message": "packaged module ran"}] + + +def get_modules() -> list[type[MVTModule]]: + return [PackagedModule] +``` + +Installed modules follow the same rules as other custom modules: each module +must declare `supported_commands`, and dependencies are resolved with the +standard ordering logic. A broken entry point is skipped with a warning and +does not prevent MVT from running. As with custom commands, installed module +packages run as trusted code inside the MVT process, so install only packages +from sources you trust. + +For a `pipx` installation of MVT, inject the package into MVT's environment: + +```bash +pipx inject mvt my-mvt-modules +``` + ## Profiling Some MVT modules extract and process significant amounts of data during the analysis process or while checking results against known indicators. Care must be diff --git a/src/mvt/common/module_loader.py b/src/mvt/common/module_loader.py index da94826..63260fb 100644 --- a/src/mvt/common/module_loader.py +++ b/src/mvt/common/module_loader.py @@ -4,6 +4,7 @@ # https://license.mvt.re/1.1/ import hashlib +import importlib.metadata import importlib.util import inspect import logging @@ -16,6 +17,7 @@ from typing import Iterable, Optional from .module import MVTModule MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES" +MODULES_ENTRY_POINT_GROUP = "mvt.modules" log = logging.getLogger(__name__) @@ -31,7 +33,9 @@ def _module_name_for_path(path: Path) -> str: def _iter_module_files(path: Path) -> Iterable[Path]: if path.is_file(): if path.suffix != ".py": - raise CustomModuleLoadError(f"Custom module file is not a Python file: {path}") + raise CustomModuleLoadError( + f"Custom module file is not a Python file: {path}" + ) yield path return @@ -59,7 +63,9 @@ def _load_python_file(path: Path) -> ModuleType: try: spec.loader.exec_module(module) except Exception as exc: - raise CustomModuleLoadError(f"Unable to import custom module {path}: {exc}") from exc + raise CustomModuleLoadError( + f"Unable to import custom module {path}: {exc}" + ) from exc return module @@ -95,6 +101,68 @@ def load_custom_modules_from_path(path: str) -> list[type[MVTModule]]: return custom_modules +def _module_key(module_class: type[MVTModule]) -> tuple[str, str]: + try: + source = str(Path(inspect.getfile(module_class)).resolve()) + except (OSError, TypeError): + source = module_class.__module__ + return (source, module_class.__qualname__) + + +def load_installed_modules() -> list[type[MVTModule]]: + """Load MVT modules registered by installed packages. + + Packages register modules in the ``mvt.modules`` entry-point group. Each + entry point must resolve to an iterable of MVTModule subclasses, or to a + callable which returns one. A broken entry point is skipped with a + warning so that a faulty plugin package cannot break MVT. + """ + try: + entry_points = importlib.metadata.entry_points(group=MODULES_ENTRY_POINT_GROUP) + except Exception as exc: + log.warning( + "Unable to discover installed module packages in entry-point group %s: %s", + MODULES_ENTRY_POINT_GROUP, + exc, + ) + return [] + + installed_modules: list[type[MVTModule]] = [] + ordered_entry_points = sorted( + entry_points, key=lambda entry_point: (entry_point.name, entry_point.value) + ) + for entry_point in ordered_entry_points: + try: + loaded = entry_point.load() + if callable(loaded) and not isinstance(loaded, type): + loaded = loaded() + module_classes = list(loaded) + except (Exception, SystemExit) as exc: + log.warning( + "Unable to load modules from entry point %s (%s): %s", + entry_point.name, + entry_point.value, + exc, + ) + continue + + for module_class in module_classes: + if not ( + isinstance(module_class, type) and issubclass(module_class, MVTModule) + ): + log.warning( + "Entry point %s (%s) provided %r which is not an " + "MVTModule subclass", + entry_point.name, + entry_point.value, + module_class, + ) + continue + installed_modules.append(module_class) + + return installed_modules + + def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTModule]]: search_paths: list[str] = [] env_path = os.environ.get(MVT_CUSTOM_MODULES_ENV) @@ -105,10 +173,17 @@ def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTM custom_modules: list[type[MVTModule]] = [] seen: set[tuple[str, str]] = set() + + for module_class in load_installed_modules(): + key = _module_key(module_class) + if key in seen: + continue + seen.add(key) + custom_modules.append(module_class) + for path in search_paths: for module_class in load_custom_modules_from_path(path): - source = Path(inspect.getfile(module_class)).resolve() - key = (str(source), module_class.__qualname__) + key = _module_key(module_class) if key in seen: continue seen.add(key) diff --git a/tests/test_custom_modules.py b/tests/test_custom_modules.py index a00faea..f66a3b2 100644 --- a/tests/test_custom_modules.py +++ b/tests/test_custom_modules.py @@ -1,3 +1,5 @@ +import importlib.metadata + from click.testing import CliRunner from mvt.android.cli import check_bugreport @@ -5,6 +7,7 @@ from mvt.android.cmd_check_androidqf import CmdAndroidCheckAndroidQF from mvt.android.cmd_check_backup import CmdAndroidCheckBackup from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport from mvt.android.cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs +from mvt.common import module_loader from mvt.common.module import MVTModule from mvt.ios.cli import check_backup, check_fs @@ -106,6 +109,62 @@ def test_custom_modules_load_from_environment_without_cli_flag(tmp_path, monkeyp assert "EnvBugreportModule" in result.output +class InstalledPackageModule(MVTModule): + supported_commands = (("ios", "check-backup"),) + + +def get_installed_package_modules(): + return [InstalledPackageModule] + + +def _fake_entry_points(monkeypatch, value, name="test-modules"): + entry_point = importlib.metadata.EntryPoint( + name=name, value=value, group=module_loader.MODULES_ENTRY_POINT_GROUP + ) + + def fake_entry_points(*, group): + assert group == module_loader.MODULES_ENTRY_POINT_GROUP + return [entry_point] + + monkeypatch.setattr( + module_loader.importlib.metadata, "entry_points", fake_entry_points + ) + + +def test_installed_module_package_loads_from_entry_point(monkeypatch): + _fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules") + + modules = module_loader.load_custom_modules() + + assert modules == [InstalledPackageModule] + + +def test_broken_module_entry_point_is_skipped(monkeypatch, caplog): + _fake_entry_points(monkeypatch, "nonexistent_module_xyz:get_modules") + + with caplog.at_level("WARNING"): + modules = module_loader.load_custom_modules() + + assert modules == [] + assert "Unable to load modules from entry point" in caplog.text + + +def test_entry_point_module_deduplicated_against_paths(monkeypatch, tmp_path): + _fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules") + module_path = _write_custom_module( + tmp_path / "custom.py", + "PathLoadedModule", + (("ios", "check-backup"),), + ) + + modules = module_loader.load_custom_modules([str(module_path)]) + + assert [module.__name__ for module in modules] == [ + "InstalledPackageModule", + "PathLoadedModule", + ] + + class NestedBugreportModule(MVTModule): supported_commands = (("android", "check-bugreport"),)