From e77008b49af2a02877a511b41e81971f8fb3e014 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Wed, 19 Aug 2026 21:20:11 +0200 Subject: [PATCH] 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 | 14 ++++ src/mvt/common/command.py | 32 +++++++++- src/mvt/common/help.py | 2 +- src/mvt/common/module_loader.py | 110 ++++++++++++++++++++++++++++++++ src/mvt/common/utils.py | 7 ++ tests/common/test_utils.py | 19 ++++++ tests/test_custom_modules.py | 91 ++++++++++++++++++++++++++ 7 files changed, 271 insertions(+), 4 deletions(-) diff --git a/docs/development.md b/docs/development.md index 8949dff..8cde546 100644 --- a/docs/development.md +++ b/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 diff --git a/src/mvt/common/command.py b/src/mvt/common/command.py index aab53e6..54a59e2 100644 --- a/src/mvt/common/command.py +++ b/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(): - self.log.info(" - %s", module.__name__) + for origin, module_names in self._modules_by_origin( + 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: diff --git a/src/mvt/common/help.py b/src/mvt/common/help.py index 5101f93..7514113 100644 --- a/src/mvt/common/help.py +++ b/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 " diff --git a/src/mvt/common/module_loader.py b/src/mvt/common/module_loader.py index 63260fb..893e9b8 100644 --- a/src/mvt/common/module_loader.py +++ b/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 diff --git a/src/mvt/common/utils.py b/src/mvt/common/utils.py index 30de159..ad3b394 100644 --- a/src/mvt/common/utils.py +++ b/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: diff --git a/tests/common/test_utils.py b/tests/common/test_utils.py index 4dbe5c0..b8791e1 100644 --- a/tests/common/test_utils.py +++ b/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 + ) diff --git a/tests/test_custom_modules.py b/tests/test_custom_modules.py index f66a3b2..43ab621 100644 --- a/tests/test_custom_modules.py +++ b/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"),)