Document how to write MVT plugins

The custom modules page now leads with plugin packages. Loading module
files with --load-module and MVT_CUSTOM_MODULES moves to a section on
developing a module locally.

A new "Writing a module" section shows a module which subclasses
IOSExtraction. It lists each base class, the command pair it serves and the
helpers it provides. "Depending on a built-in module" says to import a
built-in class from its family package.

"Importing from MVT" says what mvt.plugin exports and what importing from
it means. The custom commands page shows a Command subclass which lists its
own modules. The sysdiagnose and plugin configuration pages import from
mvt.plugin.

docs/development/custom_commands.md

@@ -73,7 +73,11 @@ rather than in MVT's own `config.yaml`.
 
 ### Commands on `mvt`
 
-A MVT plugin command can also add sub-commands to the base `mvt` command. This can be used for commands which are not tied to a particular forensic platform:
+The `mvt` command hosts what belongs to neither platform: `version`,
+`completion`, `plugins` and `download-iocs`. A plugin command which is not
+about the acquisition of one platform, such as one which configures the plugin
+or synchronizes the indicators it uses, belongs there too, in the
+`mvt.cli_plugins` group:
 
 ```toml
 [project.entry-points."mvt.cli_plugins"]

docs/development/index.md

@@ -78,9 +78,7 @@ MVT logs a warning. The nine pairs are:
 
 `check-iocs` re-checks stored results rather than an acquisition. It matches
 every `<slug>.json` file in the results folder to the module with that slug.
-It then runs that module's `check_indicators()` again. A module which
-implements `check_indicators()` is included in `check-iocs` for its platform.
-It does not need to declare the `check-iocs` pair.
+It then runs that module's `check_indicators()` again.
 
 ### Writing a module
 
@@ -178,61 +176,6 @@ not, MVT skips the module with a warning.
 They are the same records it writes to `<slug>.json`. Typed results per module
 are planned.
 
-### Replacing a built-in module
-
-A custom module which extends a built-in one runs alongside it, and when the
-two share a slug they write to the same results file. MVT warns about that
-collision, naming both modules, where each came from and the file the later one
-overwrites, but it runs them both. Set `replaces` to the class the module
-supersedes to take that module's place instead: when both are available to a
-command, the named module is dropped from the run.
-
-```python
-from mvt.ios.modules.backup import Manifest as BuiltinManifest
-
-
-class Manifest(BuiltinManifest):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = BuiltinManifest
-```
-
-Keeping the class name of the replaced module, as above, keeps the name
-`--module` selects the module by and the slug its results file is named after.
-A replacement with a different class name writes to a file named after its own
-slug, unless it sets `slug` to the slug of the module it replaces, and
-`--module` still selects it by the name of the replaced module, which MVT logs.
-Taking over the slug of a replaced module is not a collision and is not warned
-about, because that module is no longer part of the run.
-
-Modules which depend on the replaced module receive the replacement instead, so
-`get_dependency_results(BuiltinManifest)` returns the replacement's results. A
-module must not depend on the module it replaces: that module does not run, so
-the replacement has to produce the data itself, and MVT logs a warning when a
-module declares both.
-
-A replacement takes on the obligations of the module it replaces. If it
-declares a dependency the command does not provide, it is skipped like any
-other module with an unavailable dependency, and the module it replaced stays
-out of the run: neither of them produces results, and the modules depending on
-the replaced module are skipped as well.
-
-Subclassing the replaced module is not required, but a replacement which is not
-a subclass is logged with a warning, because its results may not be what the
-modules depending on the replaced module expect. Naming a module the command
-does not run has no effect, and modules which replace each other in a cycle all
-keep running and replace nothing. Every applied substitution is logged, so it
-is recorded in `command.log` when the command runs with an `--output` folder.
-
-Every command resolves replacements on its own. `check-iocs` matches stored
-results files against the slugs of the modules available for that command. A
-replacement which subclasses the module it replaces inherits its
-`check_indicators()`. It is then part of `check-iocs` for its platform and
-re-checks the results file named after its slug. A replacement with no
-`check_indicators()` is not part of `check-iocs`. The built-in module it
-replaced re-checks the file. `check-iocs` matches `--module` on the class name
-only. Pass a differently named replacement's own name there, not the name of
-the module it replaces.
-
 ### Importing from MVT
 
 Import from `mvt.plugin` if it has what you need. The names it exports are

src/mvt/android/artifacts/tombstone_crashes.py

@@ -191,14 +191,9 @@ class TombstoneCrashArtifact(AndroidArtifact):
     def _load_key_value_line(
         self, line: str, key: str, destination_key: str, tombstone: dict
     ) -> bool:
-        # The caller matched the key as a bare prefix, so a longer word starting
-        # with it arrives here: `Caused by: …` inside an abort message reaches
-        # the `Cause` key. That is a different line, not a broken file — say so
-        # by declining it, and let the remaining keys have their turn. Raising
-        # here discarded the whole tombstone, crash and stack trace included.
-        line_key, separator, value = line.partition(":")
-        if not separator or line_key != key:
-            return False
+        line_key, value = line.split(":", 1)
+        if line_key != key:
+            raise ValueError(f"Expected key {key}, got {line_key}")
 
         value_clean = value.strip().strip("'")
         if destination_key == "uid":

src/mvt/android/cli.py

@@ -14,6 +14,7 @@ from mvt.common.cli_plugins import (
     load_cli_commands_option,
     register_cli_plugins,
 )
+from mvt.common.cmd_check_iocs import CmdCheckIOCS
 from mvt.common.help import (
     HELP_MSG_ANDROID_BACKUP_PASSWORD,
     HELP_MSG_CHECK_ADB_REMOVED,
@@ -39,12 +40,17 @@ from mvt.common.help import (
     HELP_MSG_VERSION,
     HELP_MSG_VIRUS_TOTAL,
 )
+from mvt.common.logo import logo
+from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
+from mvt.common.updates import IndicatorsUpdates
 from mvt.common.utils import init_logging, set_verbose_logging
 
-# The commands import what they run only when they are invoked. This module is
-# imported at every start of mvt-android, including by shell completion on
-# every keystroke, so importing it must do no more than build the command tree:
-# the forensic modules, the backup parsers and the update checks stay out of it.
+from .cmd_check_androidqf import CmdAndroidCheckAndroidQF
+from .cmd_check_backup import CmdAndroidCheckBackup
+from .cmd_check_bugreport import CmdAndroidCheckBugreport
+from .cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
+from .command_modules import ANDROID_CHECK_IOCS_MODULES
+from .modules.backup.helpers import cli_load_android_backup_password
 
 init_logging()
 log = logging.getLogger("mvt")
@@ -68,8 +74,6 @@ def _get_verbose(ctx):
 
 
 def _load_custom_modules(load_module):
-    from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
-
     try:
         return load_custom_modules(load_module)
     except CustomModuleLoadError as exc:
@@ -97,9 +101,6 @@ def cli(ctx, disable_update_check, disable_indicator_update_check, verbose):
     ctx.obj["disable_indicator_check"] = disable_indicator_update_check
     ctx.obj["verbose"] = verbose
     set_verbose_logging(verbose)
-
-    from mvt.common.logo import logo
-
     logo(
         disable_version_check=disable_update_check,
         disable_indicator_check=disable_indicator_update_check,
@@ -163,8 +164,6 @@ def check_bugreport(
     verbose,
     bugreport_path,
 ):
-    from .cmd_check_bugreport import CmdAndroidCheckBugreport
-
     set_verbose_logging(verbose or _get_verbose(ctx))
     custom_modules = _load_custom_modules(load_module)
     # Always generate hashes as bug reports are small.
@@ -234,9 +233,6 @@ def check_backup(
     verbose,
     backup_path,
 ):
-    from .cmd_check_backup import CmdAndroidCheckBackup
-    from .modules.backup.helpers import cli_load_android_backup_password
-
     set_verbose_logging(verbose or _get_verbose(ctx))
     custom_modules = _load_custom_modules(load_module)
 
@@ -315,9 +311,6 @@ def check_androidqf(
     verbose,
     androidqf_path,
 ):
-    from .cmd_check_androidqf import CmdAndroidCheckAndroidQF
-    from .modules.backup.helpers import cli_load_android_backup_password
-
     set_verbose_logging(verbose or _get_verbose(ctx))
     custom_modules = _load_custom_modules(load_module)
 
@@ -400,8 +393,6 @@ def check_intrusion_logs(
     verbose,
     logs_path,
 ):
-    from .cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
-
     set_verbose_logging(verbose or _get_verbose(ctx))
     custom_modules = _load_custom_modules(load_module)
 
@@ -455,10 +446,6 @@ def check_intrusion_logs(
 @click.argument("FOLDER", type=click.Path(exists=True))
 @click.pass_context
 def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
-    from mvt.common.cmd_check_iocs import CmdCheckIOCS
-
-    from .command_modules import ANDROID_CHECK_IOCS_MODULES
-
     custom_modules = _load_custom_modules(load_module)
     cmd = CmdCheckIOCS(
         target_path=folder,
@@ -485,8 +472,6 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
 # ==============================================================================
 @cli.command("download-iocs", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_STIX2)
 def download_indicators():
-    from mvt.common.updates import IndicatorsUpdates
-
     ioc_updates = IndicatorsUpdates()
     ioc_updates.update()
 

src/mvt/android/cmd_check_androidqf.py

@@ -292,7 +292,9 @@ class CmdAndroidCheckAndroidQF(Command):
         try:
             cmd.from_ab(backup)
         except InvalidAndroidBackup as exc:
-            self.log.warning("Skipping backup modules: %s", exc)
+            self.log.warning(
+                "Skipping backup modules as backup.ab is malformed: %s", exc
+            )
             return False
 
         cmd.run()

src/mvt/android/cmd_check_backup.py

@@ -87,15 +87,11 @@ class CmdAndroidCheckBackup(Command):
         if header["encryption"] != "none":
             password = prompt_or_load_android_backup_password(log, self.module_options)
             if not password:
-                if self.sub_command:
-                    raise InvalidAndroidBackup("No backup password provided")
                 log.critical("No backup password provided.")
                 sys.exit(1)
         try:
             tardata = parse_backup_file(ab_file_bytes, password=password)
         except InvalidBackupPassword:
-            if self.sub_command:
-                raise InvalidAndroidBackup("Invalid backup password")
             log.critical("Invalid backup password")
             sys.exit(1)
         except AndroidBackupParsingError as exc:

src/mvt/cli.py

@@ -47,8 +47,10 @@ CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 def cli(ctx, disable_update_check, disable_indicator_update_check, verbose):
     """Mobile Verification Toolkit.
 
-    The 'mvt-ios' and 'mvt-android' CLI commands are used to perform
-    forensic analysis on IOS and Android devices.
+    mvt-ios and mvt-android run the forensic analysis of an acquisition: each
+    provides the check-* commands of its platform. This command hosts what
+    belongs to neither platform; run it without a command to see the installed
+    version and the list of what it offers.
     """
     ctx.ensure_object(dict)
     ctx.obj["disable_version_check"] = disable_update_check

src/mvt/common/cli_plugins.py

@@ -15,9 +15,7 @@ from typing import Iterable
 
 import click
 
-# This module is imported by every CLI at start-up, and by shell completion on
-# every keystroke, so it must stay cheap: nothing here may import the module
-# machinery (mvt.common.module_loader and what it pulls in).
+from .module_loader import CUSTOM_COMMAND_MODULE_PREFIX
 
 IOS_CLI_PLUGIN_GROUP = "mvt.ios.cli_plugins"
 ANDROID_CLI_PLUGIN_GROUP = "mvt.android.cli_plugins"
@@ -26,9 +24,6 @@ NEUTRAL_CLI_PLUGIN_GROUP = "mvt.cli_plugins"
 MVT_CUSTOM_COMMANDS_ENV = "MVT_CUSTOM_COMMANDS"
 MVT_IOS_CUSTOM_COMMANDS_ENV = "MVT_IOS_CUSTOM_COMMANDS"
 MVT_ANDROID_CUSTOM_COMMANDS_ENV = "MVT_ANDROID_CUSTOM_COMMANDS"
-# Prefix of the import name given to a command file loaded from a path. Shared
-# with module_loader, which recognises such files when naming their loggers.
-CUSTOM_COMMAND_MODULE_PREFIX = "_mvt_custom_command_"
 
 log = logging.getLogger(__name__)
 

src/mvt/common/command.py

@@ -72,10 +72,6 @@ class Command:
         # down a password to decrypt a backup or flags which are need by some modules.
         self.module_options = module_options if module_options else {}
 
-        # This dictionary maps the modules which were replaced by a module
-        # declaring `replaces` to the module which took their place.
-        self.module_replacements: dict[type[MVTModule], type[MVTModule]] = {}
-
         # This list will contain all executed modules.
         # We can use this to reference e.g. self.executed[0].results.
         self.executed: list[MVTModule] = []
@@ -261,174 +257,7 @@ class Command:
             if module not in deduplicated:
                 deduplicated.append(module)
 
-        available = self._apply_replacements(deduplicated)
-        self._warn_about_slug_collisions(available)
-
-        return available
-
-    def _warn_about_slug_collisions(self, modules: list[type[MVTModule]]) -> None:
-        """Report modules writing their results to the same file.
-
-        Results are stored in a file named after the module slug, so two
-        modules sharing one silently overwrite each other. Replacements are
-        already resolved here, so a module deliberately taking over the slug
-        of the module it replaces is not reported: the module it replaced is
-        no longer part of the run.
-        """
-        modules_by_slug: dict[str, type[MVTModule]] = {}
-        for module in modules:
-            slug = module.get_slug()
-            first = modules_by_slug.setdefault(slug, module)
-            if first is module:
-                continue
-
-            self.log.warning(
-                "Modules %s from %s and %s from %s both use the slug %s. If "
-                "both run, whichever runs last overwrites the results of the "
-                "other in %s.json.",
-                first.__name__,
-                get_module_origin(first).label,
-                module.__name__,
-                get_module_origin(module).label,
-                slug,
-                slug,
-            )
-
-    def _declared_replacements(
-        self, modules: list[type[MVTModule]]
-    ) -> dict[type[MVTModule], type[MVTModule]]:
-        """Return the replacements declared by the given modules."""
-        replacements: dict[type[MVTModule], type[MVTModule]] = {}
-        for module in modules:
-            replaced = module.replaces
-            if replaced is None or replaced is module:
-                continue
-
-            if not module.enabled:
-                # Replacing a module must not disable it, as a disabled
-                # replacement never runs in its place.
-                self.log.debug(
-                    "Module %s is disabled and does not replace module %s.",
-                    module.__name__,
-                    replaced.__name__,
-                )
-                continue
-
-            if replaced not in modules:
-                # A module can support several commands while the module it
-                # replaces is only available in some of them.
-                self.log.debug(
-                    "Module %s replaces module %s, which is not available "
-                    "for the %s command.",
-                    module.__name__,
-                    replaced.__name__,
-                    self.name,
-                )
-                continue
-
-            if replaced in replacements:
-                self.log.warning(
-                    "Modules %s and %s both replace module %s. Both of them "
-                    "will run, %s will not, and modules depending on %s will "
-                    "use the results of %s. Replacements which share the slug "
-                    "of %s overwrite each other's results file.",
-                    replacements[replaced].__name__,
-                    module.__name__,
-                    replaced.__name__,
-                    replaced.__name__,
-                    replaced.__name__,
-                    replacements[replaced].__name__,
-                    replaced.__name__,
-                )
-                continue
-
-            replacements[replaced] = module
-
-        return replacements
-
-    def _drop_replacement_cycles(
-        self, replacements: dict[type[MVTModule], type[MVTModule]]
-    ) -> None:
-        """Undo the replacements between modules which replace each other."""
-        cyclic: set[type[MVTModule]] = set()
-        for replaced in replacements:
-            walked: list[type[MVTModule]] = []
-            module = replaced
-            while module in replacements and module not in cyclic:
-                if module in walked:
-                    cyclic.update(walked[walked.index(module) :])
-                    break
-                walked.append(module)
-                module = replacements[module]
-
-        if not cyclic:
-            return
-
-        self.log.warning(
-            "Modules %s replace each other in a cycle. None of them replaces "
-            "anything and all of them will run.",
-            ", ".join(sorted(module.__name__ for module in cyclic)),
-        )
-        for module in cyclic:
-            replacements.pop(module, None)
-
-    def _log_replacement(
-        self,
-        replaced: type[MVTModule],
-        module: type[MVTModule],
-        replacement: type[MVTModule],
-    ) -> None:
-        """Report an applied replacement, and any problem with it."""
-        if not issubclass(module, replaced):
-            self.log.warning(
-                "Module %s replaces module %s but is not a subclass of it. "
-                "Its results might not be compatible with what modules "
-                "depending on %s expect.",
-                module.__name__,
-                replaced.__name__,
-                replaced.__name__,
-            )
-
-        if replaced in module.dependencies:
-            self.log.warning(
-                "Module %s depends on module %s, which it also replaces. The "
-                "dependency cannot be satisfied: a replacement has to produce "
-                "that data itself.",
-                module.__name__,
-                replaced.__name__,
-            )
-
-        self.log.info(
-            "Module %s from %s replaces module %s from %s.",
-            replacement.__name__,
-            get_module_origin(replacement).label,
-            replaced.__name__,
-            get_module_origin(replaced).label,
-        )
-
-    def _apply_replacements(
-        self, modules: list[type[MVTModule]]
-    ) -> list[type[MVTModule]]:
-        """Drop the modules superseded by a module declaring `replaces`."""
-        declared = self._declared_replacements(modules)
-        self._drop_replacement_cycles(declared)
-
-        replacements: dict[type[MVTModule], type[MVTModule]] = {}
-        for replaced, module in declared.items():
-            # Follow chains of replacements, so that a dependency on a
-            # replaced module always resolves to a module which is part of
-            # the run.
-            replacement = module
-            while replacement in declared:
-                replacement = declared[replacement]
-
-            # Only the replacements which are applied are reported, so that
-            # the record matches the modules which actually run.
-            self._log_replacement(replaced, module, replacement)
-            replacements[replaced] = replacement
-
-        self.module_replacements = replacements
-        return [module for module in modules if module not in replacements]
+        return deduplicated
 
     def init(self) -> None:
         raise NotImplementedError
@@ -488,80 +317,6 @@ class Command:
         console.print("")
         console.print(panel)
 
-    def _module_dependencies(
-        self, module: type[MVTModule]
-    ) -> list[tuple[type[MVTModule], type[MVTModule]]]:
-        """Return the (declared, resolved) dependencies of a module.
-
-        A dependency on a module which was replaced is resolved to the module
-        which took its place. A module which replaces one of its own
-        dependencies is not made to depend on itself, while a module which
-        declares itself as a dependency is left alone and still fails the
-        circular dependency check.
-        """
-        dependencies = []
-        for dependency in module.dependencies:
-            resolved = self.module_replacements.get(dependency, dependency)
-            if resolved is module and dependency is not module:
-                continue
-            dependencies.append((dependency, resolved))
-
-        return dependencies
-
-    @staticmethod
-    def _dependency_name(
-        declared: type[MVTModule], resolved: type[MVTModule]
-    ) -> str:
-        """Return how a dependency is named in the messages about it.
-
-        A dependency is named as the module which declared it wrote it, and,
-        when that module was replaced, as the module which runs in its place.
-        """
-        if declared is resolved:
-            return declared.__name__
-
-        return f"{resolved.__name__} (replacing module {declared.__name__})"
-
-    def _selected_modules(
-        self, modules: list[type[MVTModule]]
-    ) -> Optional[list[type[MVTModule]]]:
-        """Return the modules explicitly requested, or all the enabled ones.
-
-        Returns None when a module was requested by name and no module of
-        that name can be run.
-        """
-        if not self.module_name:
-            return [module for module in modules if module.enabled]
-
-        selected = [
-            module for module in modules if module.__name__ == self.module_name
-        ]
-
-        # A module replacing another one does not have to keep its name, so
-        # the name of a replaced module selects its replacement.
-        if not selected:
-            for replaced, replacement in self.module_replacements.items():
-                if replaced.__name__ != self.module_name or replacement in selected:
-                    continue
-                self.log.info(
-                    "Module %s was replaced by module %s, which is run "
-                    "in its place.",
-                    replaced.__name__,
-                    replacement.__name__,
-                )
-                selected.append(replacement)
-
-        if not selected:
-            self.log.warning(
-                "No module named %s is available for the %s command. "
-                "No modules will be run.",
-                self.module_name,
-                self.name,
-            )
-            return None
-
-        return selected
-
     def _skipped_modules(
         self,
         required: list[type[MVTModule]],
@@ -594,18 +349,15 @@ class Command:
                 if module in skipped:
                     continue
 
-                for declared, dependency in self._module_dependencies(module):
+                for dependency in module.dependencies:
                     if dependency not in module_indexes:
-                        # A replaced dependency always resolves to a module of
-                        # this command, so an unavailable one is always the
-                        # module class the author declared.
-                        skipped[module] = (module, declared)
+                        skipped[module] = (module, dependency)
                         changed = True
                         self.log.warning(
                             "Module %s will be SKIPPED: it depends on module "
                             "%s, which is not available in this command.%s",
                             module.__name__,
-                            declared.__name__,
+                            dependency.__name__,
                             remainder,
                         )
                         break
@@ -620,7 +372,7 @@ class Command:
                                 "module %s, itself skipped for depending on "
                                 "unavailable module %s.%s",
                                 module.__name__,
-                                self._dependency_name(declared, dependency),
+                                dependency.__name__,
                                 missing.__name__,
                                 remainder,
                             )
@@ -631,7 +383,7 @@ class Command:
                                 "module %s, which depends on unavailable "
                                 "module %s.%s",
                                 module.__name__,
-                                self._dependency_name(declared, dependency),
+                                dependency.__name__,
                                 root.__name__,
                                 missing.__name__,
                                 remainder,
@@ -645,9 +397,12 @@ class Command:
         modules = self._available_modules()
         module_indexes = {module: index for index, module in enumerate(modules)}
 
-        selected = self._selected_modules(modules)
-        if selected is None:
-            return None
+        if self.module_name:
+            selected = [
+                module for module in modules if module.__name__ == self.module_name
+            ]
+        else:
+            selected = [module for module in modules if module.enabled]
 
         required: set[type[MVTModule]] = set()
         pending = list(selected)
@@ -656,7 +411,7 @@ class Command:
             if module in required:
                 continue
             required.add(module)
-            for _, dependency in self._module_dependencies(module):
+            for dependency in module.dependencies:
                 # Unavailable dependencies are reported by _skipped_modules().
                 if dependency in module_indexes:
                     pending.append(dependency)
@@ -669,14 +424,13 @@ class Command:
                 "Every selected module was skipped for an unavailable "
                 "dependency. No modules will be run."
             )
-            return None
 
         dependents: dict[type[MVTModule], list[type[MVTModule]]] = {
             module: [] for module in runnable
         }
         indegree = {module: 0 for module in runnable}
         for module in runnable:
-            for _, dependency in self._module_dependencies(module):
+            for dependency in module.dependencies:
                 if dependency not in indegree:
                     continue
                 dependents[dependency].append(module)
@@ -732,12 +486,9 @@ class Command:
                 module_options=self.module_options,
                 log=module_logger,
             )
-            # Dependencies are keyed by the module class they declare, even
-            # when it was replaced, so that a module asking for the results
-            # of a replaced module receives those of its replacement.
             m.dependency_modules = {
-                dependency: executed_by_type[resolved]
-                for dependency, resolved in self._module_dependencies(module)
+                dependency: executed_by_type[dependency]
+                for dependency in module.dependencies
             }
 
             if self.iocs.total_ioc_count:

src/mvt/common/module.py

@@ -46,11 +46,6 @@ class MVTModule:
     slug: Optional[str] = None
     dependencies: Sequence[type["MVTModule"]] = ()
     supported_commands: Sequence[tuple[str, str]] = ()
-    # A custom module can name a module class it supersedes, usually a
-    # built-in one. When both are available to a command, the named module is
-    # dropped from the run and this module takes its place, including in the
-    # dependencies of any other module.
-    replaces: Optional[type["MVTModule"]] = None
 
     def __init__(
         self,

src/mvt/common/module_loader.py

@@ -18,7 +18,6 @@ from pathlib import Path
 from types import ModuleType
 from typing import Iterable, Optional
 
-from .cli_plugins import CUSTOM_COMMAND_MODULE_PREFIX
 from .module import MVTModule
 from .version import MVT_VERSION
 
@@ -28,6 +27,8 @@ EXTERNAL_LOGGER_NAMESPACE = "mvt.ext"
 PLUGIN_PACKAGE_PREFIX = "mvt_plugin_"
 _ORIGIN_ATTRIBUTE = "_mvt_module_origin"
 _PATH_MODULE_PREFIX = "_mvt_custom_module_"
+# Shared with cli_plugins, which names a loaded command file this way.
+CUSTOM_COMMAND_MODULE_PREFIX = "_mvt_custom_command_"
 _LOADED_FILE_DIGEST = re.compile(r"_[0-9a-f]{16}$")
 log = logging.getLogger(__name__)
 
@@ -401,14 +402,4 @@ def module_supports_command(
         )
         return False
 
-    pairs = {tuple(entry) for entry in supported_commands}
-    if (platform, command) in pairs:
-        return True
-
-    # A module which implements check_indicators() is re-checked by check-iocs
-    # for its platform. It does not need to declare the check-iocs pair.
-    return (
-        command == "check-iocs"
-        and platform in {entry[0] for entry in pairs if entry}
-        and module_class.check_indicators is not MVTModule.check_indicators
-    )
+    return (platform, command) in {tuple(entry) for entry in supported_commands}

src/mvt/common/utils.py

@@ -14,6 +14,7 @@ from dataclasses import asdict, is_dataclass
 from typing import Any, Iterator, Union
 
 from .log import MVTLogHandler
+from mvt.common.config import settings
 
 
 class CustomJSONEncoder(json.JSONEncoder):
@@ -271,10 +272,6 @@ def set_verbose_logging(verbose: bool = False):
 
 def exec_or_profile(module, globals, locals):
     """Hook for profiling MVT modules"""
-    # Imported here so that the CLI modules, which import this one at start-up,
-    # do not load the settings (and pydantic) before a command runs.
-    from .config import settings
-
     if settings.PROFILE:
         cProfile.runctx(module, globals, locals)
     else:

src/mvt/ios/cli.py

@@ -14,7 +14,10 @@ from mvt.common.cli_plugins import (
     load_cli_commands_option,
     register_cli_plugins,
 )
+from mvt.common.cmd_check_iocs import CmdCheckIOCS
+from mvt.common.logo import logo
 from mvt.common.options import MutuallyExclusiveOption
+from mvt.common.updates import IndicatorsUpdates
 from mvt.common.utils import (
     generate_hashes_from_path,
     init_logging,
@@ -44,12 +47,13 @@ from mvt.common.help import (
     HELP_MSG_DISABLE_UPDATE_CHECK,
     HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
 )
+from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
 from mvt.common.password import prompt_password
-
-# The commands import what they run only when they are invoked. This module is
-# imported at every start of mvt-ios, including by shell completion on every
-# keystroke, so importing it must do no more than build the command tree: the
-# forensic modules, the backup decryption and the update checks stay out of it.
+from .cmd_check_backup import CmdIOSCheckBackup
+from .cmd_check_fs import CmdIOSCheckFS
+from .cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
+from .decrypt import DecryptBackup
+from .command_modules import IOS_CHECK_IOCS_MODULES
 
 init_logging()
 log = logging.getLogger("mvt")
@@ -75,8 +79,6 @@ def _get_verbose(ctx):
 
 
 def _load_custom_modules(load_module):
-    from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
-
     try:
         return load_custom_modules(load_module)
     except CustomModuleLoadError as exc:
@@ -104,9 +106,6 @@ def cli(ctx, disable_update_check, disable_indicator_update_check, verbose):
     ctx.obj["disable_indicator_check"] = disable_indicator_update_check
     ctx.obj["verbose"] = verbose
     set_verbose_logging(verbose)
-
-    from mvt.common.logo import logo
-
     logo(
         disable_version_check=disable_update_check,
         disable_indicator_check=disable_indicator_update_check,
@@ -147,8 +146,6 @@ def version():
 @click.argument("BACKUP_PATH", type=click.Path(exists=True))
 @click.pass_context
 def decrypt_backup(ctx, destination, password, key_file, hashes, backup_path):
-    from .decrypt import DecryptBackup
-
     backup = DecryptBackup(backup_path, destination)
 
     if key_file:
@@ -212,8 +209,6 @@ def decrypt_backup(ctx, destination, password, key_file, hashes, backup_path):
 )
 @click.argument("BACKUP_PATH", type=click.Path(exists=True))
 def extract_key(password, key_file, backup_path):
-    from .decrypt import DecryptBackup
-
     backup = DecryptBackup(backup_path)
 
     if password:
@@ -281,8 +276,6 @@ def check_backup(
     verbose,
     backup_path,
 ):
-    from .cmd_check_backup import CmdIOSCheckBackup
-
     set_verbose_logging(verbose or _get_verbose(ctx))
     module_options = {"fast_mode": fast}
     custom_modules = _load_custom_modules(load_module)
@@ -352,8 +345,6 @@ def check_fs(
     verbose,
     dump_path,
 ):
-    from .cmd_check_fs import CmdIOSCheckFS
-
     set_verbose_logging(verbose or _get_verbose(ctx))
     module_options = {"fast_mode": fast}
     custom_modules = _load_custom_modules(load_module)
@@ -422,8 +413,6 @@ def check_sysdiagnose(
     verbose,
     sysdiagnose_path,
 ):
-    from .cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
-
     set_verbose_logging(verbose or _get_verbose(ctx))
     custom_modules = _load_custom_modules(load_module)
     cmd = CmdIOSCheckSysdiagnose(
@@ -478,10 +467,6 @@ def check_sysdiagnose(
 @click.argument("FOLDER", type=click.Path(exists=True))
 @click.pass_context
 def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
-    from mvt.common.cmd_check_iocs import CmdCheckIOCS
-
-    from .command_modules import IOS_CHECK_IOCS_MODULES
-
     custom_modules = _load_custom_modules(load_module)
     cmd = CmdCheckIOCS(
         target_path=folder,
@@ -508,8 +493,6 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
 # ==============================================================================
 @cli.command("download-iocs", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_STIX2)
 def download_iocs():
-    from mvt.common.updates import IndicatorsUpdates
-
     ioc_updates = IndicatorsUpdates()
     ioc_updates.update()
 

tests/android/test_artifact_tombstone_caused_by.py

@@ -1,54 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-"""A `Caused by:` line must not discard the whole text tombstone.
-
-Keys are matched as bare prefixes, so `Caused by: …` — an ordinary line inside
-an abort message — reached the `Cause` key, failed the key comparison and
-raised, which `Tombstones.run()` logged while dropping the entire crash record.
-Seen on a 1.6 MB tombstone whose protobuf twin was zero bytes: the crash then
-had no representation at all.
-"""
-
-import datetime
-
-from mvt.android.artifacts.tombstone_crashes import TombstoneCrashArtifact
-
-TOMBSTONE = b"""\
-*** *** *** *** *** *** *** *** *** *** *** *** *** *** *** ***
-Build fingerprint: 'Xiaomi/vili_eea/vili:13/TKQ1.220829.002/V14.0.10.0:user/release-keys'
-Revision: '0'
-ABI: 'arm64'
-Timestamp: 2023-08-24 14:54:47.999124034+0300
-Process uptime: 12199s
-Cmdline: com.example.game
-pid: 8044, tid: 26222, name: UnityMain  >>> com.example.game <<<
-uid: 10235
-signal 6 (SIGABRT), code -1 (SI_QUEUE), fault addr --------
-Abort message: 'No pending exception expected: java.lang.SecurityException: listen
-  at void android.os.Parcel.readException() (Parcel.java:2920)
-Caused by: android.os.RemoteException: Remote stack trace:
-\tat com.android.server.TelephonyRegistry.listen(TelephonyRegistry.java:1096)
-"""
-
-WITH_CAUSE = TOMBSTONE + b"Cause: null pointer dereference\n"
-
-
-class TestTombstoneCausedBy:
-    def _parse(self, content):
-        artifact = TombstoneCrashArtifact()
-        artifact.results = []
-        artifact.parse("tombstone_23", datetime.datetime(2023, 8, 24), content)
-        return artifact.results
-
-    def test_caused_by_line_does_not_discard_the_tombstone(self):
-        results = self._parse(TOMBSTONE)
-        assert len(results) == 1
-        assert results[0]["pid"] == 8044
-        assert results[0]["process_name"] == "UnityMain"
-        assert results[0]["uid"] == 10235
-
-    def test_the_real_cause_key_is_still_parsed(self):
-        results = self._parse(WITH_CAUSE)
-        assert results[0]["cause"] == "null pointer dereference"

tests/android/test_check_backup_optional_failure.py

@@ -1,39 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-"""An encrypted backup.ab must not take the whole check-androidqf run with it.
-
-`CmdAndroidCheckBackup.from_ab()` already raises `InvalidAndroidBackup` instead
-of exiting when it runs as a sub-command (`check-androidqf` catches that and
-skips the backup modules), for a wrong file format and for a parse error. The
-password branches used to call `sys.exit(1)` unconditionally, which ends the
-parent run inside `finish()` — before the intrusion-logs command and before the
-timeline, alerts, urls, info and run-manifest are written.
-"""
-
-import pytest
-
-from mvt.android.cmd_check_backup import CmdAndroidCheckBackup, InvalidAndroidBackup
-
-ENCRYPTED_AB_HEADER = b"ANDROID BACKUP\n5\n0\nAES-256\n" + b"\x00" * 64
-
-
-class TestCheckBackupOptionalFailure:
-    def _cmd(self, tmp_path, sub_command):
-        return CmdAndroidCheckBackup(
-            target_path=None,
-            results_path=str(tmp_path),
-            module_options={"interactive": False},
-            sub_command=sub_command,
-        )
-
-    def test_missing_password_raises_when_nested(self, tmp_path):
-        cmd = self._cmd(tmp_path, sub_command=True)
-        with pytest.raises(InvalidAndroidBackup):
-            cmd.from_ab(ENCRYPTED_AB_HEADER)
-
-    def test_missing_password_still_exits_on_its_own_command(self, tmp_path):
-        cmd = self._cmd(tmp_path, sub_command=False)
-        with pytest.raises(SystemExit):
-            cmd.from_ab(ENCRYPTED_AB_HEADER)

tests/common/test_cmd_check_iocs.py

@@ -15,7 +15,6 @@ from mvt.common.cmd_check_iocs import CmdCheckIOCS
 from mvt.common.module import MVTModule
 from mvt.ios.cli import cli as ios_cli
 from mvt.ios.command_modules import IOS_CHECK_IOCS_MODULES
-from mvt.ios.modules.backup.manifest import Manifest
 
 # Keep the banner of the group callback from checking for updates online.
 OFFLINE = ["--disable-update-check", "--disable-indicator-update-check"]
@@ -40,69 +39,34 @@ class CustomResultsModule(MVTModule):
         self.checked.append(list(self.results))
 
 
-class BackupCheckerModule(MVTModule):
-    """An iOS module which implements check_indicators() without declaring check-iocs."""
-
-    slug = "backup_checker"
-    supported_commands = (("ios", "check-backup"),)
-
-    checked: list = []
-
-    def run(self) -> None:
-        pass
-
-    def check_indicators(self) -> None:
-        self.checked.append(list(self.results))
-
-
-class BugReportCheckerModule(MVTModule):
-    """The same for Android."""
-
-    slug = "bugreport_checker"
-    supported_commands = (("android", "check-bugreport"),)
-
-    checked: list = []
-
-    def run(self) -> None:
-        pass
-
-    def check_indicators(self) -> None:
-        self.checked.append(list(self.results))
-
-
 class BackupOnlyModule(MVTModule):
-    """A custom module which does not implement check_indicators()."""
+    """A custom module which does not declare check-iocs."""
 
     slug = "backup_only"
     supported_commands = (("ios", "check-backup"),)
 
-    def run(self) -> None:
-        pass
+    def check_indicators(self) -> None:
+        raise AssertionError("must not be re-checked")
 
 
 @pytest.mark.parametrize(
-    "platform, builtin_modules, checker_module",
-    [
-        ("ios", IOS_CHECK_IOCS_MODULES, BackupCheckerModule),
-        ("android", ANDROID_CHECK_IOCS_MODULES, BugReportCheckerModule),
-    ],
+    "platform, builtin_modules",
+    [("ios", IOS_CHECK_IOCS_MODULES), ("android", ANDROID_CHECK_IOCS_MODULES)],
 )
 def test_check_iocs_rechecks_the_stored_results_of_custom_modules(
-    platform, builtin_modules, checker_module, tmp_path, caplog
+    platform, builtin_modules, tmp_path, caplog
 ):
     # check-iocs matches every <slug>.json in the results folder to the module
     # with that slug, custom modules included, and runs its check_indicators()
     # again over the stored results.
     results = [{"domain": "example.org"}]
     (tmp_path / "custom_results.json").write_text(json.dumps(results))
-    (tmp_path / f"{checker_module.slug}.json").write_text(json.dumps(results))
     (tmp_path / "backup_only.json").write_text(json.dumps(results))
     CustomResultsModule.checked.clear()
-    checker_module.checked.clear()
 
     cmd = CmdCheckIOCS(
         target_path=str(tmp_path),
-        custom_modules=[CustomResultsModule, checker_module, BackupOnlyModule],
+        custom_modules=[CustomResultsModule, BackupOnlyModule],
         platform=platform,
     )
     cmd.modules = builtin_modules
@@ -110,79 +74,29 @@ def test_check_iocs_rechecks_the_stored_results_of_custom_modules(
     with caplog.at_level(logging.INFO):
         cmd.run()
 
-    # A module which declares the check-iocs pair is re-checked.
     assert CustomResultsModule.checked == [results]
     assert (
         'Loading results from "custom_results.json" with module CustomResultsModule'
         in caplog.text
     )
-    # So is a module which only implements check_indicators().
-    assert checker_module.checked == [results]
-    # A module which does neither is not part of check-iocs.
+    # A module declaring only check-backup is not part of check-iocs.
     assert "backup_only.json" not in caplog.text
 
 
-@pytest.mark.parametrize(
-    "platform, builtin_modules, listed, not_listed",
-    [
-        (
-            "ios",
-            IOS_CHECK_IOCS_MODULES,
-            "BackupCheckerModule",
-            "BugReportCheckerModule",
-        ),
-        (
-            "android",
-            ANDROID_CHECK_IOCS_MODULES,
-            "BugReportCheckerModule",
-            "BackupCheckerModule",
-        ),
-    ],
-)
-def test_check_iocs_lists_the_custom_modules_it_runs(
-    platform, builtin_modules, listed, not_listed, caplog
-):
+def test_check_iocs_lists_custom_modules_declaring_the_command(caplog):
     cmd = CmdCheckIOCS(
-        custom_modules=[
-            CustomResultsModule,
-            BackupCheckerModule,
-            BugReportCheckerModule,
-            BackupOnlyModule,
-        ],
-        platform=platform,
+        custom_modules=[CustomResultsModule, BackupOnlyModule],
+        platform="ios",
     )
-    cmd.modules = builtin_modules
+    cmd.modules = IOS_CHECK_IOCS_MODULES
 
     with caplog.at_level(logging.INFO):
         cmd.list_modules()
 
     assert "CustomResultsModule" in caplog.text
-    # The module which implements check_indicators() for this platform is listed.
-    assert listed in caplog.text
-    # The one for the other platform is not, and neither is BackupOnlyModule.
-    assert not_listed not in caplog.text
     assert "BackupOnlyModule" not in caplog.text
 
 
-class ReplacementManifest(Manifest):
-    """A replacement for a built-in module which does not declare check-iocs."""
-
-    supported_commands = (("ios", "check-backup"),)
-    replaces = Manifest
-
-
-def test_check_iocs_uses_a_replacement_of_a_built_in_module():
-    # A replacement which subclasses a built-in module inherits its
-    # check_indicators(). check-iocs then runs it in place of that module.
-    cmd = CmdCheckIOCS(custom_modules=[ReplacementManifest], platform="ios")
-    cmd.modules = IOS_CHECK_IOCS_MODULES
-
-    available = cmd._available_modules()
-
-    assert ReplacementManifest in available
-    assert Manifest not in available
-
-
 LOADED_MODULE = '''
 from mvt.common.module import MVTModule
 

tests/common/test_command.py

@@ -68,123 +68,6 @@ class CustomDependsOnBuiltin(RecordingModule):
     dependencies = (FirstModule,)
 
 
-class ReplacementModule(FirstModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = FirstModule
-
-    def run(self):
-        super().run()
-        self.results = ["replacement"]
-
-
-class OtherReplacementModule(FirstModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = FirstModule
-
-
-class UnrelatedReplacementModule(RecordingModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = IndependentModule
-
-
-class ReplacementOfReplacementModule(ReplacementModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = ReplacementModule
-
-    def run(self):
-        super().run()
-        self.results = ["replacement of replacement"]
-
-
-class ReplacesOwnDependencyModule(FirstModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = FirstModule
-    dependencies = (FirstModule,)
-
-
-class DisabledReplacementModule(FirstModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = FirstModule
-    enabled = False
-
-
-class MutualReplacementOne(RecordingModule):
-    supported_commands = (("ios", "check-backup"),)
-
-
-class MutualReplacementTwo(RecordingModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = MutualReplacementOne
-
-
-MutualReplacementOne.replaces = MutualReplacementTwo
-
-
-class CycleOneModule(RecordingModule):
-    supported_commands = (("ios", "check-backup"),)
-
-
-class CycleTwoModule(RecordingModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = CycleOneModule
-
-
-class CycleThreeModule(RecordingModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = CycleTwoModule
-
-
-CycleOneModule.replaces = CycleThreeModule
-
-
-class UnavailableDependencyModule(RecordingModule):
-    supported_commands = (("ios", "check-backup"),)
-
-
-class ReplacementMissingDependency(FirstModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = FirstModule
-    dependencies = (UnavailableDependencyModule,)
-
-
-class SharedSlugModule(RecordingModule):
-    supported_commands = (("ios", "check-backup"),)
-    slug = "shared_slug"
-
-
-class OtherSharedSlugModule(RecordingModule):
-    supported_commands = (("ios", "check-backup"),)
-    slug = "shared_slug"
-
-
-class ReplacementKeepingTheSlug(FirstModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = FirstModule
-    slug = "first_module"
-
-
-class SameNameReplacementModule(FirstModule):
-    supported_commands = (("ios", "check-backup"),)
-    replaces = FirstModule
-
-    def run(self):
-        super().run()
-        self.results = ["replacement"]
-
-
-# Modules replacing a built-in one usually keep its class name, which is the
-# name `--module` matches on.
-SameNameReplacementModule.__name__ = "FirstModule"
-
-
-def logged_substitutions(caplog) -> list[str]:
-    return [
-        record.getMessage()
-        for record in caplog.records
-        if record.levelno == logging.INFO and "replaces module" in record.getMessage()
-    ]
-
-
 class RecordingCommand(Command):
     def init(self):
         self.initialized = True
@@ -420,376 +303,3 @@ class TestCommand:
         cmd.run()
 
         assert RecordingModule.run_order == ["FirstModule", "CustomDependsOnBuiltin"]
-
-    def test_custom_module_replaces_builtin(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule, IndependentModule]
-        cmd.custom_modules = [ReplacementModule]
-
-        with caplog.at_level(logging.INFO):
-            cmd.run()
-
-        assert RecordingModule.run_order == ["IndependentModule", "ReplacementModule"]
-        assert cmd.module_replacements == {FirstModule: ReplacementModule}
-        assert (
-            "Module ReplacementModule from" in caplog.text
-            and "replaces module FirstModule" in caplog.text
-        )
-
-    def test_modules_without_replaces_all_run(self):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule]
-        cmd.custom_modules = [CustomIOSBackupModule]
-
-        cmd.run()
-
-        assert RecordingModule.run_order == ["FirstModule", "CustomIOSBackupModule"]
-        assert cmd.module_replacements == {}
-
-    def test_unavailable_replaced_module_is_ignored(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [IndependentModule]
-        cmd.custom_modules = [ReplacementModule]
-
-        with caplog.at_level(logging.INFO):
-            cmd.run()
-
-        assert RecordingModule.run_order == ["IndependentModule", "ReplacementModule"]
-        assert cmd.module_replacements == {}
-        assert "replaces module FirstModule" not in caplog.text
-
-    def test_dependencies_are_resolved_to_the_replacement(self):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [SecondModule, FirstModule]
-        cmd.custom_modules = [ReplacementModule]
-
-        cmd.run()
-
-        assert RecordingModule.run_order == ["ReplacementModule", "SecondModule"]
-        second = next(
-            module for module in cmd.executed if isinstance(module, SecondModule)
-        )
-        assert isinstance(second.dependency_modules[FirstModule], ReplacementModule)
-        assert second.results == ["replacement", "second"]
-
-    def test_replacement_with_unavailable_dependency_skips_its_dependents(
-        self, caplog
-    ):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule, SecondModule, IndependentModule]
-        cmd.custom_modules = [ReplacementMissingDependency]
-
-        with caplog.at_level(logging.INFO):
-            cmd.run()
-
-        # The replacement cannot run, and the module it replaced was dropped
-        # from the run by the replacement, so neither of them produces
-        # results and the module depending on the replaced one is skipped.
-        assert RecordingModule.run_order == ["IndependentModule"]
-        assert (
-            "Module ReplacementMissingDependency will be SKIPPED: it depends "
-            "on module UnavailableDependencyModule, which is not available in "
-            "this command." in caplog.text
-        )
-        # The skipped dependent is told which module it actually depends on,
-        # and the module class its author declared.
-        assert (
-            "Module SecondModule will be SKIPPED: it depends on module "
-            "ReplacementMissingDependency (replacing module FirstModule), "
-            "itself skipped for depending on unavailable module "
-            "UnavailableDependencyModule." in caplog.text
-        )
-
-    def test_selected_replacement_with_unavailable_dependency_runs_nothing(
-        self, caplog, tmp_path
-    ):
-        cmd = RecordingCommand(module_name="FirstModule", results_path=str(tmp_path))
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule, IndependentModule]
-        cmd.custom_modules = [ReplacementMissingDependency]
-
-        with caplog.at_level(logging.INFO):
-            cmd.run()
-
-        assert RecordingModule.run_order == []
-        assert not hasattr(cmd, "initialized")
-        assert (
-            "Module FirstModule was replaced by module "
-            "ReplacementMissingDependency, which is run in its place."
-            in caplog.text
-        )
-        assert "Module ReplacementMissingDependency will be SKIPPED" in caplog.text
-        assert "No modules will be run" in caplog.text
-        # Nothing else was selected, so the warnings must not promise that the
-        # analysis continues right before saying that it does not.
-        assert "The rest of the analysis will still run" not in caplog.text
-        # No module ran, so no results were stored next to the command log.
-        assert [path.name for path in tmp_path.iterdir()] == ["command.log"]
-
-    def test_chained_replacements_are_resolved(self):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [SecondModule, FirstModule]
-        cmd.custom_modules = [ReplacementModule, ReplacementOfReplacementModule]
-
-        cmd.run()
-
-        assert RecordingModule.run_order == [
-            "ReplacementOfReplacementModule",
-            "SecondModule",
-        ]
-        second = next(
-            module for module in cmd.executed if isinstance(module, SecondModule)
-        )
-        assert second.results == ["replacement of replacement", "second"]
-
-    def test_replacement_which_is_not_a_subclass_warns(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [IndependentModule]
-        cmd.custom_modules = [UnrelatedReplacementModule]
-
-        with caplog.at_level(logging.WARNING):
-            cmd.run()
-
-        assert RecordingModule.run_order == ["UnrelatedReplacementModule"]
-        assert "is not a subclass of it" in caplog.text
-
-    def test_multiple_modules_replacing_the_same_module_warn(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule]
-        cmd.custom_modules = [ReplacementModule, OtherReplacementModule]
-
-        with caplog.at_level(logging.WARNING):
-            cmd.run()
-
-        assert RecordingModule.run_order == [
-            "ReplacementModule",
-            "OtherReplacementModule",
-        ]
-        assert cmd.module_replacements == {FirstModule: ReplacementModule}
-        assert (
-            "Modules ReplacementModule and OtherReplacementModule both replace "
-            "module FirstModule. Both of them will run, FirstModule will not, "
-            "and modules depending on FirstModule will use the results of "
-            "ReplacementModule." in caplog.text
-        )
-        assert "overwrite each other's results file" in caplog.text
-
-    def test_module_replacing_its_own_dependency_runs(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule]
-        cmd.custom_modules = [ReplacesOwnDependencyModule]
-
-        with caplog.at_level(logging.WARNING):
-            cmd.run()
-
-        assert RecordingModule.run_order == ["ReplacesOwnDependencyModule"]
-        assert (
-            "Module ReplacesOwnDependencyModule depends on module FirstModule, "
-            "which it also replaces" in caplog.text
-        )
-
-    def test_literal_self_dependency_is_still_circular(self, caplog):
-        class SelfDependentModule(RecordingModule):
-            pass
-
-        SelfDependentModule.dependencies = (SelfDependentModule,)
-
-        cmd = RecordingCommand()
-        cmd.modules = [SelfDependentModule]
-
-        with caplog.at_level(logging.WARNING):
-            cmd.run()
-
-        assert RecordingModule.run_order == []
-        assert not hasattr(cmd, "initialized")
-        assert "Circular module dependency detected" in caplog.text
-
-    def test_disabled_replacement_keeps_the_replaced_module(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule]
-        cmd.custom_modules = [DisabledReplacementModule]
-
-        with caplog.at_level(logging.INFO):
-            cmd.run()
-
-        assert RecordingModule.run_order == ["FirstModule"]
-        assert cmd.module_replacements == {}
-        assert logged_substitutions(caplog) == []
-
-    def test_modules_replacing_each_other_all_run(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.custom_modules = [MutualReplacementOne, MutualReplacementTwo]
-
-        with caplog.at_level(logging.INFO):
-            cmd.run()
-
-        assert RecordingModule.run_order == [
-            "MutualReplacementOne",
-            "MutualReplacementTwo",
-        ]
-        assert cmd.module_replacements == {}
-        assert (
-            "Modules MutualReplacementOne, MutualReplacementTwo replace each "
-            "other in a cycle" in caplog.text
-        )
-        assert logged_substitutions(caplog) == []
-
-    def test_replacement_cycle_of_three_modules_all_run(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.custom_modules = [CycleOneModule, CycleTwoModule, CycleThreeModule]
-
-        with caplog.at_level(logging.INFO):
-            cmd.run()
-
-        assert RecordingModule.run_order == [
-            "CycleOneModule",
-            "CycleTwoModule",
-            "CycleThreeModule",
-        ]
-        assert cmd.module_replacements == {}
-        assert (
-            "Modules CycleOneModule, CycleThreeModule, CycleTwoModule replace "
-            "each other in a cycle" in caplog.text
-        )
-        assert logged_substitutions(caplog) == []
-
-    def test_selected_replaced_module_name_runs_the_replacement(self, caplog):
-        cmd = RecordingCommand(module_name="FirstModule")
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule]
-        cmd.custom_modules = [ReplacementModule]
-
-        with caplog.at_level(logging.INFO):
-            cmd.run()
-
-        assert RecordingModule.run_order == ["ReplacementModule"]
-        assert (
-            "Module FirstModule was replaced by module ReplacementModule"
-            in caplog.text
-        )
-
-    def test_unknown_selected_module_warns_and_stops(self, caplog):
-        cmd = RecordingCommand(module_name="NoSuchModule")
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule]
-
-        with caplog.at_level(logging.WARNING):
-            cmd.run()
-
-        assert RecordingModule.run_order == []
-        assert not hasattr(cmd, "initialized")
-        assert (
-            "No module named NoSuchModule is available for the check-backup "
-            "command" in caplog.text
-        )
-
-    def test_selected_module_name_matches_the_replacement(self):
-        cmd = RecordingCommand(module_name="FirstModule")
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule]
-        cmd.custom_modules = [SameNameReplacementModule]
-
-        cmd.run()
-
-        assert len(cmd.executed) == 1
-        assert isinstance(cmd.executed[0], SameNameReplacementModule)
-        assert cmd.executed[0].results == ["replacement"]
-
-    def test_list_modules_reflects_replacements(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule, IndependentModule]
-        cmd.custom_modules = [ReplacementModule]
-
-        with caplog.at_level(logging.INFO):
-            cmd.list_modules()
-
-        listed = [
-            record.getMessage()
-            for record in caplog.records
-            if "Modules from" in record.getMessage()
-        ]
-        assert any("ReplacementModule" in message for message in listed)
-        assert not any("FirstModule" in message for message in listed)
-
-    def test_modules_sharing_a_slug_are_reported(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.custom_modules = [SharedSlugModule, OtherSharedSlugModule]
-
-        with caplog.at_level(logging.WARNING):
-            cmd.run()
-
-        assert RecordingModule.run_order == [
-            "SharedSlugModule",
-            "OtherSharedSlugModule",
-        ]
-        collisions = [
-            record.getMessage()
-            for record in caplog.records
-            if "both use the slug" in record.getMessage()
-        ]
-        assert len(collisions) == 1
-        assert "Modules SharedSlugModule from" in collisions[0]
-        assert "and OtherSharedSlugModule from" in collisions[0]
-        assert "both use the slug shared_slug" in collisions[0]
-        assert "overwrites the results of the other in shared_slug.json" in (
-            collisions[0]
-        )
-
-    def test_replacement_keeping_the_replaced_slug_is_not_reported(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule]
-        cmd.custom_modules = [ReplacementKeepingTheSlug]
-
-        with caplog.at_level(logging.WARNING):
-            cmd.run()
-
-        # The replaced module is no longer part of the run, so taking over its
-        # slug is what the replacement is for, not a collision.
-        assert RecordingModule.run_order == ["ReplacementKeepingTheSlug"]
-        assert ReplacementKeepingTheSlug.get_slug() == FirstModule.get_slug()
-        assert "both use the slug" not in caplog.text
-
-    def test_modules_with_distinct_slugs_are_not_reported(self, caplog):
-        cmd = RecordingCommand()
-        cmd.platform = "ios"
-        cmd.name = "check-backup"
-        cmd.modules = [FirstModule, IndependentModule]
-        cmd.custom_modules = [CustomIOSBackupModule]
-
-        with caplog.at_level(logging.WARNING):
-            cmd.run()
-
-        assert "both use the slug" not in caplog.text

tests/test_check_android_androidqf.py

@@ -155,7 +155,7 @@ class TestCheckAndroidqfCommand:
             result = runner.invoke(check_androidqf, [str(path)])
 
         assert result.exit_code == 0
-        assert "Skipping backup modules: Invalid backup format" in caplog.text
+        assert "Skipping backup modules as backup.ab is malformed" in caplog.text
         assert not any(
             record.levelname in {"CRITICAL", "FATAL"} for record in caplog.records
         )

tests/test_cli.py

@@ -29,6 +29,7 @@ class TestMvtCommand:
         assert result.exit_code == 0
         assert "mvt-ios" in result.output
         assert "mvt-android" in result.output
+        assert "check-*" in result.output
 
     def test_version_prints_the_installed_version(self):
         result = CliRunner().invoke(cli, [*OFFLINE, "version"])

tests/test_cli_startup.py

@@ -1,28 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2026 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-import pytest
-
-from .plugin_fixtures import run_isolated_python
-
-# Importing a platform CLI must only build its command tree: the console
-# scripts import it before Click can answer a shell completion request, which
-# the completion scripts make on every keystroke. Every command imports what
-# it runs when it is invoked. Each of these costs tens of milliseconds to
-# import and is the sign that a command implementation is imported too early.
-HEAVY_MODULES = ("pydantic", "requests", "Crypto", "mvt.common.module")
-
-
-@pytest.mark.parametrize("cli_module", ("mvt.ios.cli", "mvt.android.cli"))
-def test_importing_a_cli_does_not_import_the_module_machinery(cli_module, tmp_path):
-    result = run_isolated_python(
-        "import sys\n"
-        f"import {cli_module}\n"
-        f"print(','.join(name for name in {HEAVY_MODULES!r} if name in sys.modules))\n",
-        home=tmp_path / "home",
-    )
-
-    assert result.returncode == 0, result.stderr
-    assert result.stdout.strip() == "", f"{cli_module} imported {result.stdout.strip()}"