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

@@ -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

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)

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"),)