Allow custom modules to replace built-in modules

A custom module which extends a built-in one ran alongside it, and both
wrote to the same results file when they shared a slug, with the run
order deciding the surviving content.

Custom modules can now name the module class they supersede in a
`replaces` attribute. When both are available to a command, the named
module is dropped from the run and the substitution is logged with the
origin of the replacement, so it is recorded in command.log. Only the
replacements which are applied are reported: a declaration from a
disabled module is ignored, so that replacing a module cannot silently
disable it, and modules which replace each other in a cycle all keep
running and replace nothing, with a warning naming every one of them.

Dependencies are remapped along with the modules themselves: a module
depending on a replaced class is ordered against, and receives the
results of, the module which took its place. Replacing a module which
others depend on therefore does not make that dependency unavailable.

The remapping applies wherever dependencies are read, so a replacement
which cannot run is skipped like any other module with an unavailable
dependency, and takes the modules depending on the module it replaced
with it. Those warnings name the dependency the author declared as well
as the module which replaces it.

A replacement does not have to keep the class name of the module it
replaces, so `--module` now falls back to the name of a replaced module
and runs its replacement. A name which matches no module at all stops
the run with a warning instead of silently analyzing nothing, as does a
selection left with nothing to run once skipped modules are dropped.

Sharing a slug outside a replacement stays possible and is now reported.
Two modules writing to the same results file is a forensic-integrity
problem rather than an error, so both still run and a warning names them,
where each came from, and the file the later one overwrites. Taking over
the slug of a replaced module is not reported, because that module is no
longer part of the run.

docs/development/index.md

@@ -176,6 +176,59 @@ 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, so a
+replacement checks the indicators of its own results only if it also declares
+the `("ios", "check-iocs")` pair; otherwise the built-in module it replaced
+re-checks the file. It also matches `--module` on the class name only, so 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/common/command.py

@@ -72,6 +72,10 @@ 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] = []
@@ -257,7 +261,174 @@ class Command:
             if module not in deduplicated:
                 deduplicated.append(module)
 
-        return deduplicated
+        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]
 
     def init(self) -> None:
         raise NotImplementedError
@@ -317,6 +488,80 @@ 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]],
@@ -349,15 +594,18 @@ class Command:
                 if module in skipped:
                     continue
 
-                for dependency in module.dependencies:
+                for declared, dependency in self._module_dependencies(module):
                     if dependency not in module_indexes:
-                        skipped[module] = (module, dependency)
+                        # 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)
                         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__,
-                            dependency.__name__,
+                            declared.__name__,
                             remainder,
                         )
                         break
@@ -372,7 +620,7 @@ class Command:
                                 "module %s, itself skipped for depending on "
                                 "unavailable module %s.%s",
                                 module.__name__,
-                                dependency.__name__,
+                                self._dependency_name(declared, dependency),
                                 missing.__name__,
                                 remainder,
                             )
@@ -383,7 +631,7 @@ class Command:
                                 "module %s, which depends on unavailable "
                                 "module %s.%s",
                                 module.__name__,
-                                dependency.__name__,
+                                self._dependency_name(declared, dependency),
                                 root.__name__,
                                 missing.__name__,
                                 remainder,
@@ -397,12 +645,9 @@ class Command:
         modules = self._available_modules()
         module_indexes = {module: index for index, module in enumerate(modules)}
 
-        if self.module_name:
+        selected = self._selected_modules(modules)
-            selected = [
+        if selected is None:
-                module for module in modules if module.__name__ == self.module_name
+            return None
-            ]
-        else:
-            selected = [module for module in modules if module.enabled]
 
         required: set[type[MVTModule]] = set()
         pending = list(selected)
@@ -411,7 +656,7 @@ class Command:
             if module in required:
                 continue
             required.add(module)
-            for dependency in module.dependencies:
+            for _, dependency in self._module_dependencies(module):
                 # Unavailable dependencies are reported by _skipped_modules().
                 if dependency in module_indexes:
                     pending.append(dependency)
@@ -424,13 +669,14 @@ 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 module.dependencies:
+            for _, dependency in self._module_dependencies(module):
                 if dependency not in indegree:
                     continue
                 dependents[dependency].append(module)
@@ -486,9 +732,12 @@ 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[dependency]
+                dependency: executed_by_type[resolved]
-                for dependency in module.dependencies
+                for dependency, resolved in self._module_dependencies(module)
             }
 
             if self.iocs.total_ioc_count:

src/mvt/common/module.py

@@ -46,6 +46,11 @@ 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,

tests/common/test_command.py

@@ -68,6 +68,123 @@ 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
@@ -303,3 +420,376 @@ 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