Record the source of loaded modules for auditability

Now that installed module packages load automatically, record where every
module came from:

- --list-modules groups the available modules by source, one line per
  source with the modules comma-separated: MVT itself with its version,
  each installed package with its version and VCS commit when recorded
  (PEP 610 direct_url.json), and each --load-module/MVT_CUSTOM_MODULES
  file with its SHA-256 hash.
- Commands log one line per module source with its version or hash and
  the modules loaded from it, so command.log records exactly which
  modules ran and where they came from.
- Make init_logging() idempotent: a loaded module package importing an
  MVT CLI module would previously add a second console handler and
  duplicate every console log line.

docs/development.md

@@ -160,6 +160,20 @@ For a `pipx` installation of MVT, inject the package into MVT's environment:
 pipx inject mvt my-mvt-modules
 ```
 
+## Auditing loaded modules
+
+Because installed module packages load automatically, MVT records where every
+module came from:
+
+- `--list-modules` groups the available modules by source: MVT itself
+  (with its version), each installed package (with its version and, when
+  installed directly from a repository, the commit), and each file loaded
+  with `--load-module` or `MVT_CUSTOM_MODULES` (with the SHA-256 hash of the
+  file).
+- When a command runs with an `--output` folder, the `command.log` file
+  records one line per module source with the source's version or hash and
+  the list of modules loaded from it.
+
 ## Profiling
 
 Some MVT modules extract and process significant amounts of data during the analysis process or while checking results against known indicators. Care must be

src/mvt/common/command.py

@@ -19,7 +19,7 @@ from .alerts import AlertLevel, AlertStore
 from .config import settings
 from .indicators import Indicators
 from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
-from .module_loader import module_supports_command
+from .module_loader import ModuleOrigin, get_module_origin, module_supports_command
 from .module_types import ModuleTimeline, URLResult
 from .utils import (
     CustomJSONEncoder,
@@ -210,10 +210,34 @@ class Command:
         for file in generate_hashes_from_path(self.target_path, self.log):
             self.hash_values.append(file)
 
+    @staticmethod
+    def _modules_by_origin(
+        modules: list[type[MVTModule]],
+    ) -> dict[ModuleOrigin, list[str]]:
+        origins: dict[ModuleOrigin, list[str]] = {}
+        for module in modules:
+            origins.setdefault(get_module_origin(module), []).append(module.__name__)
+        return origins
+
     def list_modules(self) -> None:
         self.log.info("Following is the list of available %s modules:", self.name)
-        for module in self._available_modules():
+        for origin, module_names in self._modules_by_origin(
-            self.log.info(" - %s", module.__name__)
+            self._available_modules()
+        ).items():
+            self.log.info(
+                " - Modules from %s: %s", origin.label, ", ".join(module_names)
+            )
+
+    def _log_loaded_modules(self, modules: list[type[MVTModule]]) -> None:
+        """Record the loaded modules and their origins for auditability."""
+        for origin, module_names in self._modules_by_origin(modules).items():
+            self.log.info(
+                "Loaded %d %s modules from %s: %s",
+                len(module_names),
+                self.name,
+                origin.label,
+                ", ".join(module_names),
+            )
 
     def _available_modules(self) -> list[type[MVTModule]]:
         modules = list(self.modules)
@@ -360,6 +384,8 @@ class Command:
         if ordered_modules is None:
             return
 
+        self._log_loaded_modules(ordered_modules)
+
         try:
             self.init()
         except NotImplementedError:

src/mvt/common/help.py

@@ -8,7 +8,7 @@ HELP_MSG_VERSION = "Show the currently installed version of MVT"
 HELP_MSG_OUTPUT = "Specify a path to a folder where you want to store JSON results"
 HELP_MSG_IOC = "Path to indicators file (can be invoked multiple time)"
 HELP_MSG_FAST = "Avoid running time/resource consuming features"
-HELP_MSG_LIST_MODULES = "Print list of available modules and exit"
+HELP_MSG_LIST_MODULES = "Print list of available modules and their source, then exit"
 HELP_MSG_MODULE = "Name of a single module you would like to run instead of all"
 HELP_MSG_LOAD_MODULE = (
     "Load custom MVT module(s) from a Python file or folder "

src/mvt/common/module_loader.py

@@ -7,17 +7,22 @@ import hashlib
 import importlib.metadata
 import importlib.util
 import inspect
+import json
 import logging
 import os
 import sys
+from dataclasses import dataclass
+from functools import lru_cache
 from pathlib import Path
 from types import ModuleType
 from typing import Iterable, Optional
 
 from .module import MVTModule
+from .version import MVT_VERSION
 
 MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES"
 MODULES_ENTRY_POINT_GROUP = "mvt.modules"
+_ORIGIN_ATTRIBUTE = "_mvt_module_origin"
 log = logging.getLogger(__name__)
 
 
@@ -25,6 +30,34 @@ class CustomModuleLoadError(Exception):
     pass
 
 
+@dataclass(frozen=True)
+class ModuleOrigin:
+    """Describes where a loaded module came from, for auditability.
+
+    ``kind`` is one of ``builtin`` (shipped with MVT), ``package`` (loaded
+    from an installed package) or ``path`` (loaded from a file passed with
+    ``--load-module`` or the environment variable).
+    """
+
+    kind: str
+    name: str
+    version: Optional[str] = None
+    commit: Optional[str] = None
+    file_sha256: Optional[str] = None
+
+    @property
+    def label(self) -> str:
+        label = self.name
+        if self.version:
+            label += f"@{self.version}"
+        label = f"'{label}'"
+        if self.commit:
+            label += f" (commit {self.commit})"
+        if self.file_sha256:
+            label += f" (sha256: {self.file_sha256})"
+        return label
+
+
 def _module_name_for_path(path: Path) -> str:
     digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
     return f"_mvt_custom_module_{path.stem}_{digest}"
@@ -90,12 +123,17 @@ def load_custom_modules_from_path(path: str) -> list[type[MVTModule]]:
     resolved_path = Path(path).expanduser().resolve()
 
     for module_file in _iter_module_files(resolved_path):
+        file_sha256 = hashlib.sha256(module_file.read_bytes()).hexdigest()
         loaded_module = _load_python_file(module_file)
+        origin = ModuleOrigin(
+            kind="path", name=str(module_file), file_sha256=file_sha256
+        )
         for module_class in discover_mvt_modules(loaded_module):
             key = (str(module_file), module_class.__qualname__)
             if key in seen:
                 continue
             seen.add(key)
+            setattr(module_class, _ORIGIN_ATTRIBUTE, origin)
             custom_modules.append(module_class)
 
     return custom_modules
@@ -109,6 +147,76 @@ def _module_key(module_class: type[MVTModule]) -> tuple[str, str]:
     return (source, module_class.__qualname__)
 
 
+def _distribution_commit(dist: importlib.metadata.Distribution) -> Optional[str]:
+    """Return the VCS commit a distribution was installed from, if recorded.
+
+    Packages installed directly from a repository (``pip install git+...``)
+    record the commit in ``direct_url.json`` (PEP 610).
+    """
+    try:
+        direct_url_text = dist.read_text("direct_url.json")
+        if not direct_url_text:
+            return None
+        commit = json.loads(direct_url_text).get("vcs_info", {}).get("commit_id")
+        return commit if isinstance(commit, str) else None
+    except Exception:
+        return None
+
+
+def _entry_point_origin(entry_point: importlib.metadata.EntryPoint) -> ModuleOrigin:
+    name = entry_point.name
+    version = None
+    commit = None
+    # Manually constructed entry points have no associated distribution.
+    dist = getattr(entry_point, "dist", None)
+    if dist is not None:
+        try:
+            name = dist.name or name
+            version = dist.version
+        except Exception:
+            pass
+        commit = _distribution_commit(dist)
+    return ModuleOrigin(kind="package", name=name, version=version, commit=commit)
+
+
+@lru_cache(maxsize=1)
+def _packages_distributions() -> dict[str, list[str]]:
+    try:
+        return dict(importlib.metadata.packages_distributions())
+    except Exception:
+        return {}
+
+
+def get_module_origin(module_class: type[MVTModule]) -> ModuleOrigin:
+    """Return the origin of a module class for auditing purposes."""
+    origin = module_class.__dict__.get(_ORIGIN_ATTRIBUTE)
+    if isinstance(origin, ModuleOrigin):
+        return origin
+
+    top_level = module_class.__module__.partition(".")[0]
+    if top_level == "mvt":
+        return ModuleOrigin(kind="builtin", name="mvt", version=MVT_VERSION)
+
+    distributions = _packages_distributions().get(top_level)
+    if distributions:
+        name = distributions[0]
+        version = None
+        commit = None
+        try:
+            dist = importlib.metadata.distribution(name)
+            version = dist.version
+            commit = _distribution_commit(dist)
+        except Exception:
+            pass
+        return ModuleOrigin(kind="package", name=name, version=version, commit=commit)
+
+    try:
+        source = str(Path(inspect.getfile(module_class)).resolve())
+    except (OSError, TypeError):
+        source = module_class.__module__
+    return ModuleOrigin(kind="path", name=source)
+
+
 def load_installed_modules() -> list[type[MVTModule]]:
     """Load MVT modules registered by installed packages.
 
@@ -146,6 +254,7 @@ def load_installed_modules() -> list[type[MVTModule]]:
             )
             continue
 
+        origin = _entry_point_origin(entry_point)
         for module_class in module_classes:
             if not (
                 isinstance(module_class, type) and issubclass(module_class, MVTModule)
@@ -158,6 +267,7 @@ def load_installed_modules() -> list[type[MVTModule]]:
                     module_class,
                 )
                 continue
+            setattr(module_class, _ORIGIN_ATTRIBUTE, origin)
             installed_modules.append(module_class)
 
     return installed_modules

src/mvt/common/utils.py

@@ -239,6 +239,13 @@ def init_logging(verbose: bool = False):
     """
     log = logging.getLogger("mvt")
     log.setLevel(logging.DEBUG)
+
+    # Importing an MVT CLI module calls init_logging() at import time, and
+    # loaded module packages may import one indirectly. Keep this idempotent
+    # so console log lines are not duplicated by a second handler.
+    if any(isinstance(handler, MVTLogHandler) for handler in log.handlers):
+        return
+
     consoleHandler = MVTLogHandler()
     consoleHandler.setFormatter(logging.Formatter("%(message)s"))
     if verbose:

tests/common/test_utils.py

@@ -8,6 +8,7 @@ import logging
 import os
 from datetime import datetime
 
+from mvt.common.log import MVTLogHandler
 from mvt.common.utils import (
     CustomJSONEncoder,
     convert_datetime_to_iso,
@@ -16,6 +17,7 @@ from mvt.common.utils import (
     convert_unix_to_utc_datetime,
     generate_hashes_from_path,
     get_sha256_from_file_path,
+    init_logging,
 )
 
 from ..utils import get_artifact_folder
@@ -103,3 +105,20 @@ class TestCustomJSONEncoder:
             json.dumps({"name": "家".encode()}, cls=CustomJSONEncoder)
             == '{"name": "\\u5bb6"}'
         )
+
+
+class TestInitLogging:
+    def test__init_logging_is_idempotent(self):
+        # Loaded module packages may import an MVT CLI module, which calls
+        # init_logging() again at import time. A second call must not add
+        # a duplicate console handler.
+        log = logging.getLogger("mvt")
+        init_logging()
+        handler_count = sum(
+            isinstance(handler, MVTLogHandler) for handler in log.handlers
+        )
+        init_logging()
+        assert (
+            sum(isinstance(handler, MVTLogHandler) for handler in log.handlers)
+            == handler_count
+        )

tests/test_custom_modules.py

@@ -1,4 +1,6 @@
+import hashlib
 import importlib.metadata
+import json
 
 from click.testing import CliRunner
 
@@ -9,6 +11,7 @@ from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
 from mvt.android.cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
 from mvt.common import module_loader
 from mvt.common.module import MVTModule
+from mvt.common.version import MVT_VERSION
 from mvt.ios.cli import check_backup, check_fs
 
 
@@ -165,6 +168,94 @@ def test_entry_point_module_deduplicated_against_paths(monkeypatch, tmp_path):
     ]
 
 
+def test_list_modules_shows_module_sources(tmp_path, caplog):
+    module_path = _write_custom_module(
+        tmp_path / "custom.py",
+        "SourcedBackupModule",
+        (("ios", "check-backup"),),
+    )
+    file_sha256 = hashlib.sha256(module_path.read_bytes()).hexdigest()
+    custom_modules = module_loader.load_custom_modules([str(module_path)])
+
+    from mvt.ios.cmd_check_backup import CmdIOSCheckBackup
+
+    cmd = CmdIOSCheckBackup(target_path=str(tmp_path), custom_modules=custom_modules)
+    cmd.list_modules()
+
+    assert f" - Modules from 'mvt@{MVT_VERSION}':" in caplog.text
+    assert (
+        f" - Modules from '{module_path}' (sha256: {file_sha256}): SourcedBackupModule"
+        in caplog.text
+    )
+
+
+def test_builtin_module_origin():
+    from mvt.ios.modules.backup import BACKUP_MODULES
+
+    origin = module_loader.get_module_origin(BACKUP_MODULES[0])
+
+    assert origin.kind == "builtin"
+    assert origin.name == "mvt"
+    assert origin.version == MVT_VERSION
+
+
+def test_installed_module_origin(monkeypatch):
+    _fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
+
+    modules = module_loader.load_custom_modules()
+
+    origin = module_loader.get_module_origin(modules[0])
+    assert origin.kind == "package"
+    assert origin.name == "test-modules"
+
+
+def test_distribution_commit_read_from_direct_url():
+    class FakeDistribution:
+        def read_text(self, filename):
+            assert filename == "direct_url.json"
+            return json.dumps(
+                {
+                    "url": "https://github.com/example/example-modules",
+                    "vcs_info": {"commit_id": "abc1234", "vcs": "git"},
+                }
+            )
+
+    assert module_loader._distribution_commit(FakeDistribution()) == "abc1234"
+
+
+def test_command_log_records_loaded_modules(tmp_path):
+    (tmp_path / "Manifest.db").touch()
+    (tmp_path / "Info.plist").touch()
+    module_path = _write_custom_module(
+        tmp_path / "custom.py",
+        "AuditedRunModule",
+        (("ios", "check-backup"),),
+        slug="audited_run_module",
+    )
+    file_sha256 = hashlib.sha256(module_path.read_bytes()).hexdigest()
+    output_path = tmp_path / "out"
+
+    result = CliRunner().invoke(
+        check_backup,
+        [
+            "--module",
+            "AuditedRunModule",
+            "--load-module",
+            str(module_path),
+            "--output",
+            str(output_path),
+            str(tmp_path),
+        ],
+    )
+
+    assert result.exit_code == 0
+    command_log = (output_path / "command.log").read_text(encoding="utf-8")
+    assert (
+        f"Loaded 1 check-backup modules from '{module_path}' "
+        f"(sha256: {file_sha256}): AuditedRunModule" in command_log
+    )
+
+
 class NestedBugreportModule(MVTModule):
     supported_commands = (("android", "check-bugreport"),)