Add custom module loading

2026-07-08 22:17:53 +02:00 · 2026-06-17 20:18:28 +02:00
parent b9f13b8146
commit 26bece59ee
18 changed files with 793 additions and 10 deletions
@@ -42,6 +42,19 @@ class IndependentModule(RecordingModule):
    pass


+class CustomIOSBackupModule(RecordingModule):
+    supported_commands = (("ios", "check-backup"),)
+
+
+class CustomIOSFSModule(RecordingModule):
+    supported_commands = (("ios", "check-fs"),)
+
+
+class CustomDependsOnBuiltin(RecordingModule):
+    supported_commands = (("ios", "check-backup"),)
+    dependencies = (FirstModule,)
+
+
 class RecordingCommand(Command):
    def init(self):
        self.initialized = True
@@ -132,3 +145,46 @@ class TestCommand:
        assert RecordingModule.run_order == []
        assert not hasattr(cmd, "initialized")
        assert "depends on unavailable module UnavailableModule" in caplog.text
+
+    def test_custom_modules_are_filtered_before_ordering(self):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule]
+        cmd.custom_modules = [CustomIOSBackupModule, CustomIOSFSModule]
+
+        assert [module.__name__ for module in cmd._ordered_modules()] == [
+            "FirstModule",
+            "CustomIOSBackupModule",
+        ]
+
+    def test_selected_custom_module_runs(self):
+        cmd = RecordingCommand(module_name="CustomIOSBackupModule")
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.custom_modules = [CustomIOSBackupModule]
+
+        cmd.run()
+
+        assert RecordingModule.run_order == ["CustomIOSBackupModule"]
+
+    def test_selected_unsupported_custom_module_does_not_run(self):
+        cmd = RecordingCommand(module_name="CustomIOSFSModule")
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.custom_modules = [CustomIOSFSModule]
+
+        cmd.run()
+
+        assert RecordingModule.run_order == []
+
+    def test_custom_module_dependencies_use_topological_order(self):
+        cmd = RecordingCommand(module_name="CustomDependsOnBuiltin")
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [SecondModule, FirstModule]
+        cmd.custom_modules = [CustomDependsOnBuiltin]
+
+        cmd.run()
+
+        assert RecordingModule.run_order == ["FirstModule", "CustomDependsOnBuiltin"]
@@ -0,0 +1,145 @@
+import pytest
+
+from mvt.common.module import MVTModule
+from mvt.common.module_loader import (
+    CustomModuleLoadError,
+    load_custom_modules,
+    load_custom_modules_from_path,
+    module_supports_command,
+)
+
+
+MODULE_TEMPLATE = """
+from mvt.common.module import MVTModule
+
+
+class {name}(MVTModule):
+    supported_commands = {supported_commands!r}
+
+    def run(self):
+        pass
+
+    def check_indicators(self):
+        pass
+
+    def serialize(self, result):
+        return None
+"""
+
+
+def _write_module(path, name, supported_commands=()):
+    path.write_text(
+        MODULE_TEMPLATE.format(
+            name=name,
+            supported_commands=supported_commands,
+        ),
+        encoding="utf-8",
+    )
+    return path
+
+
+def test_load_custom_modules_from_python_file(tmp_path):
+    module_path = _write_module(tmp_path / "custom.py", "FileModule")
+
+    modules = load_custom_modules_from_path(str(module_path))
+
+    assert [module.__name__ for module in modules] == ["FileModule"]
+    assert issubclass(modules[0], MVTModule)
+
+
+def test_load_custom_modules_from_folder_in_sorted_order(tmp_path):
+    _write_module(tmp_path / "b_module.py", "BModule")
+    _write_module(tmp_path / "a_module.py", "AModule")
+    _write_module(tmp_path / ".hidden.py", "HiddenModule")
+    _write_module(tmp_path / "__init__.py", "InitModule")
+    nested = tmp_path / "nested"
+    nested.mkdir()
+    _write_module(nested / "nested_module.py", "NestedModule")
+
+    modules = load_custom_modules_from_path(str(tmp_path))
+
+    assert [module.__name__ for module in modules] == ["AModule", "BModule"]
+
+
+def test_discovery_ignores_imported_base_and_unrelated_classes(tmp_path):
+    module_path = tmp_path / "custom.py"
+    module_path.write_text(
+        """
+from mvt.common.module import MVTModule
+
+
+class Unrelated:
+    pass
+
+
+class DiscoveredModule(MVTModule):
+    def run(self):
+        pass
+
+    def check_indicators(self):
+        pass
+
+    def serialize(self, result):
+        return None
+""",
+        encoding="utf-8",
+    )
+
+    modules = load_custom_modules_from_path(str(module_path))
+
+    assert [module.__name__ for module in modules] == ["DiscoveredModule"]
+
+
+def test_load_custom_modules_deduplicates_same_class(tmp_path):
+    module_path = _write_module(tmp_path / "custom.py", "DuplicateModule")
+
+    modules = load_custom_modules([str(module_path), str(module_path)])
+
+    assert [module.__name__ for module in modules] == ["DuplicateModule"]
+
+
+def test_load_custom_modules_raises_for_missing_path(tmp_path):
+    with pytest.raises(CustomModuleLoadError, match="does not exist"):
+        load_custom_modules_from_path(str(tmp_path / "missing.py"))
+
+
+def test_load_custom_modules_raises_for_import_error(tmp_path):
+    module_path = tmp_path / "broken.py"
+    module_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
+
+    with pytest.raises(CustomModuleLoadError, match="broken import"):
+        load_custom_modules_from_path(str(module_path))
+
+
+def test_load_custom_modules_loads_env_folder_first(tmp_path, monkeypatch):
+    env_folder = tmp_path / "env"
+    env_folder.mkdir()
+    cli_folder = tmp_path / "cli"
+    cli_folder.mkdir()
+    _write_module(env_folder / "env_module.py", "EnvModule")
+    _write_module(cli_folder / "cli_module.py", "CliModule")
+    monkeypatch.setenv("MVT_CUSTOM_MODULES", str(env_folder))
+
+    modules = load_custom_modules([str(cli_folder)])
+
+    assert [module.__name__ for module in modules] == ["EnvModule", "CliModule"]
+
+
+def test_module_supports_command_defaults_to_all_commands(tmp_path):
+    module_path = _write_module(tmp_path / "custom.py", "DefaultModule")
+    module = load_custom_modules_from_path(str(module_path))[0]
+
+    assert module_supports_command(module, "ios", "check-backup")
+    assert module_supports_command(module, "android", "check-bugreport")
+
+
+def test_module_supports_command_honors_supported_commands(tmp_path):
+    module_path = _write_module(
+        tmp_path / "custom.py",
+        "SpecificModule",
+        (("ios", "check-backup"),),
+    )
+    module = load_custom_modules_from_path(str(module_path))[0]
+
+    assert module_supports_command(module, "ios", "check-backup")
+    assert not module_supports_command(module, "ios", "check-fs")
@@ -0,0 +1,193 @@
+from click.testing import CliRunner
+
+from mvt.android.cli import check_bugreport
+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.module import MVTModule
+from mvt.ios.cli import check_backup, check_fs
+
+
+CUSTOM_MODULE = """
+from mvt.common.module import MVTModule
+
+
+class {name}(MVTModule):
+    supported_commands = {supported_commands!r}
+    slug = "{slug}"
+
+    def run(self):
+        self.results = [{{"message": "custom module ran"}}]
+
+    def check_indicators(self):
+        pass
+
+    def serialize(self, result):
+        return None
+"""
+
+
+def _write_custom_module(path, name, supported_commands, slug=None):
+    path.write_text(
+        CUSTOM_MODULE.format(
+            name=name,
+            supported_commands=supported_commands,
+            slug=slug or name.lower(),
+        ),
+        encoding="utf-8",
+    )
+    return path
+
+
+def test_load_module_appears_only_for_supported_cli_command(tmp_path):
+    module_path = _write_custom_module(
+        tmp_path / "custom.py",
+        "IOSBackupOnlyModule",
+        (("ios", "check-backup"),),
+    )
+
+    backup_result = CliRunner().invoke(
+        check_backup,
+        ["--list-modules", "--load-module", str(module_path), str(tmp_path)],
+    )
+    fs_result = CliRunner().invoke(
+        check_fs,
+        ["--list-modules", "--load-module", str(module_path), str(tmp_path)],
+    )
+
+    assert backup_result.exit_code == 0
+    assert "IOSBackupOnlyModule" in backup_result.output
+    assert fs_result.exit_code == 0
+    assert "IOSBackupOnlyModule" not in fs_result.output
+
+
+def test_module_option_runs_supported_custom_module(tmp_path):
+    module_path = _write_custom_module(
+        tmp_path / "custom.py",
+        "CustomRunModule",
+        (("ios", "check-backup"),),
+        slug="custom_run_module",
+    )
+    output_path = tmp_path / "out"
+
+    result = CliRunner().invoke(
+        check_backup,
+        [
+            "--module",
+            "CustomRunModule",
+            "--load-module",
+            str(module_path),
+            "--output",
+            str(output_path),
+            str(tmp_path),
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert (output_path / "custom_run_module.json").exists()
+
+
+def test_custom_modules_load_from_environment_without_cli_flag(tmp_path, monkeypatch):
+    custom_modules_path = tmp_path / "custom_modules"
+    custom_modules_path.mkdir()
+    _write_custom_module(
+        custom_modules_path / "env_module.py",
+        "EnvBugreportModule",
+        (("android", "check-bugreport"),),
+    )
+    monkeypatch.setenv("MVT_CUSTOM_MODULES", str(custom_modules_path))
+
+    result = CliRunner().invoke(check_bugreport, ["--list-modules", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "EnvBugreportModule" in result.output
+
+
+class NestedBugreportModule(MVTModule):
+    supported_commands = (("android", "check-bugreport"),)
+
+
+class NestedBackupModule(MVTModule):
+    supported_commands = (("android", "check-backup"),)
+
+
+class NestedIntrusionLogsModule(MVTModule):
+    supported_commands = (("android", "check-intrusion-logs"),)
+
+
+class NestedAndroidQFModule(MVTModule):
+    supported_commands = (("android", "check-androidqf"),)
+
+
+class DummyZip:
+    def close(self):
+        pass
+
+
+def test_androidqf_propagates_custom_modules_to_nested_commands(tmp_path, monkeypatch):
+    records = {}
+    custom_modules = [
+        NestedBugreportModule,
+        NestedBackupModule,
+        NestedIntrusionLogsModule,
+        NestedAndroidQFModule,
+    ]
+    cmd = CmdAndroidCheckAndroidQF(
+        target_path=str(tmp_path),
+        custom_modules=custom_modules,
+    )
+
+    def record_available(name):
+        def _record(command):
+            records[name] = [
+                module.__name__
+                for module in command._available_modules()
+                if module.__name__.startswith("Nested")
+            ]
+
+        return _record
+
+    monkeypatch.setattr(cmd, "load_bugreport", lambda: DummyZip())
+    monkeypatch.setattr(
+        CmdAndroidCheckBugreport,
+        "from_zip",
+        lambda self, bugreport: None,
+    )
+    monkeypatch.setattr(
+        CmdAndroidCheckBugreport,
+        "run",
+        record_available("bugreport"),
+    )
+
+    monkeypatch.setattr(cmd, "load_backup", lambda: b"")
+    monkeypatch.setattr(CmdAndroidCheckBackup, "from_ab", lambda self, backup: None)
+    monkeypatch.setattr(
+        CmdAndroidCheckBackup,
+        "run",
+        record_available("backup"),
+    )
+
+    intrusion_logs_path = tmp_path / "intrusion_logs"
+    intrusion_logs_path.mkdir()
+    setattr(cmd, "_CmdAndroidCheckAndroidQF__format", "dir")
+    setattr(
+        cmd,
+        "_CmdAndroidCheckAndroidQF__files",
+        ["androidqf/intrusion_logs/security.txt"],
+    )
+    monkeypatch.setattr(cmd, "_read_device_timezone", lambda: None)
+    monkeypatch.setattr(
+        CmdAndroidCheckIntrusionLogs,
+        "run",
+        record_available("intrusion_logs"),
+    )
+
+    assert cmd.run_bugreport_cmd()
+    assert cmd.run_backup_cmd()
+    assert cmd.run_intrusion_logs_cmd()
+    assert records == {
+        "bugreport": ["NestedBugreportModule"],
+        "backup": ["NestedBackupModule"],
+        "intrusion_logs": ["NestedIntrusionLogsModule"],
+    }