Merge branch 'main' into feature/build-arm64-docker-images

README.md

@@ -84,7 +84,7 @@ details.
 
 Users can also add top-level commands to `mvt-ios` and `mvt-android` from
 installed Python packages or local files and folders. See the
-[custom CLI command documentation](https://docs.mvt.re/en/latest/development/custom_commands/)
+[custom CLI command documentation](https://docs.mvt.re/en/latest/custom_commands/)
 for the plugin entry points and `--load-command` interface.
 
 

docs/custom_commands.md

@@ -1,7 +1,7 @@
 # Custom CLI Commands
 
 MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
-Custom commands are different from [custom forensic modules](index.md#custom-modules):
+Custom commands are different from [custom forensic modules](development.md#custom-modules):
 commands add new CLI operations, while modules add analysis steps to existing
 `check-*` commands.
 
@@ -56,10 +56,6 @@ pipx inject mvt my-mvt-plugin
 When MVT is installed in an active virtual environment, install the plugin with
 `pip` in that environment.
 
-Command packages that need their own settings, such as an API key, should store
-them in a namespaced [plugin configuration file](plugin_configuration.md)
-rather than in MVT's own `config.yaml`.
-
 ## Load a Command File
 
 For local commands that are not packaged, create a Python file that exports one

docs/development.md

@@ -117,97 +117,6 @@ class DependentCustomModule(MVTModule):
         self.results = [{"manifest_entries": len(manifest_results)}]
 ```
 
-## Installed module packages
-
-Python packages can register modules so they load automatically in every
-module-running `check-*` command, without `--load-module` or
-`MVT_CUSTOM_MODULES`. Register an entry point in the `mvt.modules` group in
-the package's `pyproject.toml`:
-
-```toml
-[project.entry-points."mvt.modules"]
-mvt-plugin-amnesty-custom = "mvt_plugin_amnesty_custom:get_modules"
-```
-
-The entry point must resolve to an iterable of `MVTModule` subclasses, or to
-a callable returning one:
-
-```python
-from mvt.common.module import MVTModule
-
-
-class PackagedModule(MVTModule):
-    supported_commands = (("ios", "check-backup"),)
-
-    def run(self):
-        self.results = [{"message": "packaged module ran"}]
-
-
-def get_modules() -> list[type[MVTModule]]:
-    return [PackagedModule]
-```
-
-Installed modules follow the same rules as other custom modules: each module
-must declare `supported_commands`, and dependencies are resolved with the
-standard ordering logic. A broken entry point is skipped with a warning and
-does not prevent MVT from running. As with custom commands, installed module
-packages run as trusted code inside the MVT process, so install only packages
-from sources you trust.
-
-For a `pipx` installation of MVT, inject the package into MVT's environment:
-
-```bash
-pipx inject mvt mvt-plugin-amnesty-custom
-```
-
-Module packages that need their own settings, such as an API key, should store
-them in a namespaced [plugin configuration file](plugin_configuration.md)
-rather than in MVT's own `config.yaml`.
-
-### Naming module packages
-
-Name module packages `mvt-plugin-<name>` (import package `mvt_plugin_<name>`),
-and include the name of the publishing organization or author so packages from
-different groups do not collide: for example, Amnesty International's custom
-modules would be distributed as `mvt-plugin-amnesty-custom` with the import
-package `mvt_plugin_amnesty_custom`.
-
-The prefix makes module packages easy to find on PyPI and keeps their import
-names from clashing with unrelated Python packages. It is a convention, not a
-technical requirement: modules load through the `mvt.modules` entry point
-regardless of what the package is called, and existing packages with other
-names keep working. Note that the prefix is also not a mark of authenticity —
-anyone can publish a package with any available name, so vet a module package
-and its publisher before installing it, whatever it is called.
-
-### Module logging
-
-Modules log through `self.log`, and MVT names the logger for where the module
-came from. MVT's own modules log under their dotted path (for example
-`mvt.ios.modules.mixed.whatsapp`). Everything external is namespaced under
-`mvt.ext` to keep it visually distinct from built-in modules and isolated from
-MVT's internal logger tree:
-
-- Installed packages log under `mvt.ext.<package>`, with the `mvt_plugin_`
-  prefix stripped: modules in `mvt_plugin_amnesty_custom` log as
-  `mvt.ext.amnesty_custom.*`.
-- Files loaded with `--load-module` or `MVT_CUSTOM_MODULES` log as
-  `mvt.ext.<file name>`.
-
-## 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

docs/development/plugin_configuration.md

@@ -1,149 +0,0 @@
-# Plugin Configuration
-
-Plugin packages that add [custom CLI commands](custom_commands.md) or
-[modules](index.md#installed-module-packages) often need to store their
-own settings, such as an API key, a server URL or the timestamp of the last
-synchronization. MVT provides a namespaced settings base class so each plugin
-keeps its configuration in its own file, and a data folder for anything else a
-plugin needs to keep on disk.
-
-!!! warning
-
-    Do not write plugin settings to MVT's own `config.yaml`. MVT rewrites that
-    file with the settings it knows about every time it starts, so any other
-    section is deleted.
-
-## Where Settings Are Stored
-
-Each plugin gets one YAML file in a `plugins` folder next to MVT's own
-configuration:
-
-```
-~/.config/mvt/plugins/<plugin name>.yaml
-```
-
-The exact parent folder follows the platform convention used for MVT's
-`config.yaml` (for example `~/Library/Application Support/mvt` on macOS). Use
-`mvt.common.plugin_config.plugin_config_path()` instead of building the path by
-hand.
-
-Plugin names must be lowercase and may only contain letters, digits and dashes,
-matching the `mvt-plugin-<name>` package naming convention. MVT creates the
-`plugins` folder with `0700` permissions and writes the settings files with
-`0600` permissions, because they commonly hold credentials. Files are written
-through a temporary file and moved into place, so an interrupted save never
-leaves a partially written settings file behind.
-
-## Plugin Data Folder
-
-Everything else a plugin keeps on disk, such as a cache, a downloaded artifact
-or synchronization state, belongs in the folder returned by the `data_folder()`
-class method of the plugin's settings class, or by
-`mvt.common.plugin_config.plugin_data_folder()` called with the plugin name if
-the plugin has no settings class:
-
-```
-~/.local/share/mvt/plugin-data/<plugin name>/                # Linux
-~/Library/Application Support/mvt/plugin-data/<plugin name>/ # macOS
-```
-
-The folder sits beside MVT's own data, such as the downloaded indicators. It is
-created if it is missing, with `0700` permissions. Asking for it again returns
-the same path and leaves the contents alone, so a plugin can ask for it every
-time it needs the folder. `ExamplePluginSettings` below is the settings class
-defined in the next section:
-
-```python
-import os
-
-
-def cache_path() -> str:
-    folder = ExamplePluginSettings.data_folder()
-    return os.path.join(folder, "virustotal_lookups_cache.json")
-```
-
-A plugin which has no settings class calls
-`plugin_data_folder("example-plugin")` instead.
-
-Do not fall back on a path of your own such as `~/.cache/example-plugin`: it
-is a Linux-only convention, and MVT will not create it for you.
-
-## Defining Plugin Settings
-
-Subclass `MVTPluginSettings`, set `plugin_name` and declare typed fields with
-defaults:
-
-```python
-from typing import Optional
-
-from mvt.common.plugin_config import MVTPluginSettings
-
-
-class ExamplePluginSettings(MVTPluginSettings):
-    plugin_name = "example-plugin"
-
-    API_KEY: Optional[str] = None
-    MAX_RESULTS: int = 25
-    LAST_SYNC: Optional[str] = None
-```
-
-`load()` returns the current settings and `save()` writes them back:
-
-```python
-from datetime import datetime, timezone
-
-import click
-
-
-def sync():
-    settings = ExamplePluginSettings.load()
-    if not settings.API_KEY:
-        raise click.ClickException(
-            "No API key configured. Set MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY or "
-            "run 'example-plugin configure'."
-        )
-
-    settings.LAST_SYNC = datetime.now(timezone.utc).isoformat()
-    settings.save()
-```
-
-A missing settings file is not an error: the plugin then runs on the field
-defaults and on whatever the environment provides. `save()` only persists the
-values that differ from the defaults, and it never touches MVT's `config.yaml`.
-A settings file that cannot be parsed, or that does not hold a mapping of
-setting names to values, raises a `PluginConfigLoadError` naming the file.
-
-Every subclass that sets its own `plugin_name` gets its own file and its own
-environment namespace. A subclass that does not redefine `plugin_name` inherits
-it, and therefore shares the file and the environment variables of its parent
-class.
-
-## Environment Variables
-
-Every field can also be set with an environment variable. The prefix is
-`MVT_PLUGIN_`, followed by the plugin name upper-cased with dashes replaced by
-underscores, followed by the field name. For the example above:
-
-```bash
-export MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY=...
-export MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS=50
-```
-
-Settings resolve in this order, from highest to lowest priority:
-
-1. Arguments passed to the settings class directly, such as
-   `ExamplePluginSettings(API_KEY="...")`
-2. Environment variables
-3. The plugin's YAML file
-4. The field defaults declared on the settings class
-
-!!! tip
-
-    On shared or multi-user machines, prefer passing API keys through
-    environment variables rather than saving them to the plugin file. `save()`
-    skips every value that the environment currently supplies, so a credential
-    provided that way is not copied into the settings file when a plugin saves
-    an unrelated setting.
-
-Unknown keys in a plugin's YAML file are ignored, so a settings file written by
-a newer version of a plugin does not break an older one.

docs/ios/records.md

@@ -417,21 +417,7 @@ If indicators are provided through the command-line, they are checked against th
     Backup: :material-check:
     Full filesystem dump: :material-check:
 
-This JSON file is created by mvt-ios' `WhatsApp` module. The module extracts a list of WhatsApp messages from the SQLite database located at *private/var/mobile/Containers/Shared/AppGroup/\*/ChatStorage.sqlite*, along with one record per chat session (marked with `"record_type": "chat_session"`) containing the first and last interaction dates of each conversation. Chat sessions produce `chat_first_message` and `chat_last_message` timeline events, and group chats additionally produce a `group_created` event. A chat session's last-message date can postdate its newest stored message when the most recent messages in the chat were deleted.
-
-Recent WhatsApp versions key 1:1 chat sessions by an opaque LID identifier rather than the contact's phone number. The module resolves these using the `ZWAPHONENUMBERLIDPAIR` table from the *LID.sqlite* database in the same app group (or from *ChatStorage.sqlite* itself in versions that store it there), populating `partner_resolved_phone_number` on chat session records and using the phone number in timeline events. Each LID-phone number pair is also extracted as a record (`"record_type": "lid_phone_number_pair"`) and produces a `lid_pair_recorded` timeline event marking when WhatsApp learned the association.
+This JSON file is created by mvt-ios' `WhatsApp` module. The module extracts a list of WhatsApp messages from the SQLite database located at *private/var/mobile/Containers/Shared/AppGroup/\*/ChatStorage.sqlite*.
 
 If indicators are provided through the command-line, they are checked against the extracted HTTP links. Any matches are stored in *whatsapp_detected.json*.
 
----
-
-### `whatsapp_contacts.json`
-
-!!! info "Availability"
-    Backup: :material-check:
-    Full filesystem dump: :material-check:
-
-This JSON file is created by mvt-ios' `WhatsappContacts` module. The module extracts WhatsApp contact records from the SQLite database located at *private/var/mobile/Containers/Shared/AppGroup/\*/ContactsV2.sqlite*, including each contact's phone number, WhatsApp and LID identifiers, and the per-contact disappearing messages timer, which is not recorded in *ChatStorage.sqlite*. Each timestamp stored on a contact record produces a timeline event: `disappearing_mode_set` (when the disappearing messages timer was last changed), `about_changed` (when the contact last changed their "about" text), `about_expiration` (when a timed "about" is scheduled to expire) and `contact_last_updated` (when the contact record was last updated).
-
-This database is often missing from incremental backups. When it cannot be found, the module logs a warning and produces no results, in which case the disappearing messages state of chats cannot be determined from the backup.
-

mkdocs.yml

@@ -31,6 +31,7 @@ nav:
     - Introduction: "introduction.md"
     - Installation: "install.md"
     - Command Completion: "command_completion.md"
+    - Custom CLI Commands: "custom_commands.md"
     - Using Docker: "docker.md"
     - MVT for iOS:
         - iOS Forensic Methodology: "ios/methodology.md"
@@ -49,8 +50,5 @@ nav:
         - Check an Android Backup (SMS messages): "android/backup.md"
         - Check Android Intrusion Logs: "android/intrusion_logs.md"
     - Indicators of Compromise: "iocs.md"
-    - Development:
-        - Development Instructions: "development/index.md"
-        - Custom CLI Commands: "development/custom_commands.md"
-        - Plugin Configuration: "development/plugin_configuration.md"
+    - Development: "development.md"
     - License: "license.md"

src/mvt/common/cmd_check_iocs.py

@@ -9,7 +9,6 @@ from typing import Optional
 
 from mvt.common.command import Command
 from mvt.common.module import MVTModule
-from mvt.common.module_loader import get_module_logger
 from mvt.common.utils import exec_or_profile
 
 log = logging.getLogger(__name__)
@@ -77,7 +76,7 @@ class CmdCheckIOCS(Command):
                 )
 
                 m = iocs_module.from_json(
-                    file_path, log=get_module_logger(iocs_module)
+                    file_path, log=logging.getLogger(iocs_module.__module__)
                 )
                 if not m:
                     log.warning("No result from this module, skipping it")

src/mvt/common/command.py

@@ -19,12 +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 (
-    ModuleOrigin,
-    get_module_logger,
-    get_module_origin,
-    module_supports_command,
-)
+from .module_loader import module_supports_command
 from .module_types import ModuleTimeline, URLResult
 from .utils import (
     CustomJSONEncoder,
@@ -215,34 +210,10 @@ 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 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),
-            )
+        for module in self._available_modules():
+            self.log.info(" - %s", module.__name__)
 
     def _available_modules(self) -> list[type[MVTModule]]:
         modules = list(self.modules)
@@ -389,8 +360,6 @@ class Command:
         if ordered_modules is None:
             return
 
-        self._log_loaded_modules(ordered_modules)
-
         try:
             self.init()
         except NotImplementedError:
@@ -399,7 +368,7 @@ class Command:
         executed_by_type: dict[type[MVTModule], MVTModule] = {}
         for module in ordered_modules:
 
-            module_logger = get_module_logger(module)
+            module_logger = logging.getLogger(module.__module__)
 
             m = module(
                 target_path=self.target_path,

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 their source, then exit"
+HELP_MSG_LIST_MODULES = "Print list of available modules and 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

@@ -4,28 +4,18 @@
 #   https://license.mvt.re/1.1/
 
 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"
-EXTERNAL_LOGGER_NAMESPACE = "mvt.ext"
-PLUGIN_PACKAGE_PREFIX = "mvt_plugin_"
-_ORIGIN_ATTRIBUTE = "_mvt_module_origin"
-_PATH_MODULE_PREFIX = "_mvt_custom_module_"
 log = logging.getLogger(__name__)
 
 
@@ -33,74 +23,15 @@ 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"{_PATH_MODULE_PREFIX}{path.stem}_{digest}"
-
-
-def get_module_logger(module_class: type[MVTModule]) -> logging.Logger:
-    """Return the logger a module's records should be emitted through.
-
-    Modules loaded from installed packages or file paths live outside the
-    "mvt" logger hierarchy, so their records would never reach the handlers
-    attached to the "mvt" logger and instead fall through to
-    logging.lastResort (which prints bare messages and drops anything below
-    WARNING). Their loggers are parented under the "mvt.ext" namespace,
-    keeping external module names from colliding with MVT's own logger
-    tree. File-path modules are named after their file instead of the
-    mangled internal import name, and packages following the recommended
-    "mvt_plugin_<name>" naming convention log under "mvt.ext.<name>".
-    """
-    name = module_class.__module__
-    if name == "mvt" or name.startswith("mvt."):
-        return logging.getLogger(name)
-
-    if name.startswith(_PATH_MODULE_PREFIX):
-        name = Path(get_module_origin(module_class).name).stem
-    else:
-        top_level, separator, rest = name.partition(".")
-        if top_level.startswith(PLUGIN_PACKAGE_PREFIX) and len(top_level) > len(
-            PLUGIN_PACKAGE_PREFIX
-        ):
-            name = top_level[len(PLUGIN_PACKAGE_PREFIX) :] + separator + rest
-
-    return logging.getLogger(f"{EXTERNAL_LOGGER_NAMESPACE}.{name}")
+    return f"_mvt_custom_module_{path.stem}_{digest}"
 
 
 def _iter_module_files(path: Path) -> Iterable[Path]:
     if path.is_file():
         if path.suffix != ".py":
-            raise CustomModuleLoadError(
-                f"Custom module file is not a Python file: {path}"
-            )
+            raise CustomModuleLoadError(f"Custom module file is not a Python file: {path}")
         yield path
         return
 
@@ -128,9 +59,7 @@ def _load_python_file(path: Path) -> ModuleType:
     try:
         spec.loader.exec_module(module)
     except Exception as exc:
-        raise CustomModuleLoadError(
-            f"Unable to import custom module {path}: {exc}"
-        ) from exc
+        raise CustomModuleLoadError(f"Unable to import custom module {path}: {exc}") from exc
 
     return module
 
@@ -155,156 +84,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
 
 
-def _module_key(module_class: type[MVTModule]) -> tuple[str, str]:
-    try:
-        source = str(Path(inspect.getfile(module_class)).resolve())
-    except (OSError, TypeError):
-        source = module_class.__module__
-    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.
-
-    Packages register modules in the ``mvt.modules`` entry-point group. Each
-    entry point must resolve to an iterable of MVTModule subclasses, or to a
-    callable which returns one. A broken entry point is skipped with a
-    warning so that a faulty plugin package cannot break MVT.
-    """
-    try:
-        entry_points = importlib.metadata.entry_points(group=MODULES_ENTRY_POINT_GROUP)
-    except Exception as exc:
-        log.warning(
-            "Unable to discover installed module packages in entry-point group %s: %s",
-            MODULES_ENTRY_POINT_GROUP,
-            exc,
-        )
-        return []
-
-    installed_modules: list[type[MVTModule]] = []
-    ordered_entry_points = sorted(
-        entry_points, key=lambda entry_point: (entry_point.name, entry_point.value)
-    )
-    for entry_point in ordered_entry_points:
-        try:
-            loaded = entry_point.load()
-            if callable(loaded) and not isinstance(loaded, type):
-                loaded = loaded()
-            module_classes = list(loaded)
-        except (Exception, SystemExit) as exc:
-            log.warning(
-                "Unable to load modules from entry point %s (%s): %s",
-                entry_point.name,
-                entry_point.value,
-                exc,
-            )
-            continue
-
-        origin = _entry_point_origin(entry_point)
-        for module_class in module_classes:
-            if not (
-                isinstance(module_class, type) and issubclass(module_class, MVTModule)
-            ):
-                log.warning(
-                    "Entry point %s (%s) provided %r which is not an "
-                    "MVTModule subclass",
-                    entry_point.name,
-                    entry_point.value,
-                    module_class,
-                )
-                continue
-            setattr(module_class, _ORIGIN_ATTRIBUTE, origin)
-            installed_modules.append(module_class)
-
-    return installed_modules
-
-
 def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTModule]]:
     search_paths: list[str] = []
     env_path = os.environ.get(MVT_CUSTOM_MODULES_ENV)
@@ -315,17 +105,10 @@ def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTM
 
     custom_modules: list[type[MVTModule]] = []
     seen: set[tuple[str, str]] = set()
-
-    for module_class in load_installed_modules():
-        key = _module_key(module_class)
-        if key in seen:
-            continue
-        seen.add(key)
-        custom_modules.append(module_class)
-
     for path in search_paths:
         for module_class in load_custom_modules_from_path(path):
-            key = _module_key(module_class)
+            source = Path(inspect.getfile(module_class)).resolve()
+            key = (str(source), module_class.__qualname__)
             if key in seen:
                 continue
             seen.add(key)

src/mvt/common/plugin_config.py

@@ -1,299 +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 contextlib
-import json
-import os
-import re
-import tempfile
-from typing import Any, ClassVar, Dict, List, Tuple, Type, TypeVar
-
-import yaml
-from appdirs import user_config_dir, user_data_dir
-from pydantic import ValidationError
-from pydantic_settings import (
-    BaseSettings,
-    EnvSettingsSource,
-    PydanticBaseSettingsSource,
-    SettingsConfigDict,
-    YamlConfigSettingsSource,
-)
-
-PLUGIN_CONFIG_FOLDER_NAME = "plugins"
-# Not "plugins": on macOS the configuration and data folders are the same
-# directory, and that name already holds the settings files.
-PLUGIN_DATA_FOLDER_NAME = "plugin-data"
-PLUGIN_ENV_PREFIX = "MVT_PLUGIN_"
-
-PLUGIN_NAME_PATTERN = re.compile(r"[a-z0-9][a-z0-9-]*")
-
-PluginSettingsType = TypeVar("PluginSettingsType", bound="MVTPluginSettings")
-
-
-class PluginConfigLoadError(Exception):
-    pass
-
-
-def validate_plugin_name(plugin_name: Any) -> str:
-    """
-    Check that a plugin name is safe to use in a file name and an env variable.
-
-    :param plugin_name: Name to validate.
-    :returns: The validated plugin name.
-    """
-    if not isinstance(plugin_name, str):
-        raise TypeError(
-            f"Plugin name must be a string, not {type(plugin_name).__name__}"
-        )
-    if not PLUGIN_NAME_PATTERN.fullmatch(plugin_name):
-        raise ValueError(
-            f"Invalid plugin name {plugin_name!r}: plugin names must start with a "
-            "lowercase letter or a digit and may only contain lowercase letters, "
-            "digits and dashes"
-        )
-    return plugin_name
-
-
-def plugin_config_folder() -> str:
-    """
-    Return the folder where plugins store their configuration files.
-
-    The path is resolved on every call so it always reflects the current
-    environment.
-    """
-    return os.path.join(user_config_dir("mvt"), PLUGIN_CONFIG_FOLDER_NAME)
-
-
-def plugin_config_path(plugin_name: str) -> str:
-    """
-    Return the path of the configuration file of a given plugin.
-
-    :param plugin_name: Name of the plugin.
-    """
-    return os.path.join(
-        plugin_config_folder(), f"{validate_plugin_name(plugin_name)}.yaml"
-    )
-
-
-def plugin_data_folder(plugin_name: str) -> str:
-    """
-    Return the folder where a given plugin stores its data, creating it.
-
-    Plugins should keep whatever they persist, such as caches or downloaded
-    artifacts, in this folder. It is created with owner-only permissions. The
-    path is resolved on every call so it always reflects the current
-    environment. A plugin with a settings class calls
-    `MVTPluginSettings.data_folder()` instead, which passes `plugin_name` here.
-
-    :param plugin_name: Name of the plugin.
-    :returns: The path of the data folder of the plugin.
-    """
-    # Validate the name before anything is created, so an unsafe name cannot
-    # leave a folder behind.
-    name = validate_plugin_name(plugin_name)
-
-    # makedirs() applies its mode only to the last folder of the path, so
-    # MVT's own data folder keeps the default permissions while the two
-    # plugin folders are private.
-    data_folder = os.path.join(user_data_dir("mvt"), PLUGIN_DATA_FOLDER_NAME)
-    os.makedirs(data_folder, mode=0o700, exist_ok=True)
-
-    folder = os.path.join(data_folder, name)
-    os.makedirs(folder, mode=0o700, exist_ok=True)
-    return folder
-
-
-def plugin_env_prefix(plugin_name: str) -> str:
-    """
-    Return the environment variable prefix used by a given plugin.
-
-    Dashes are replaced by underscores. Plugin names cannot contain underscores,
-    so two different plugin names never share an environment namespace.
-
-    :param plugin_name: Name of the plugin.
-    """
-    name = validate_plugin_name(plugin_name).upper().replace("-", "_")
-    return f"{PLUGIN_ENV_PREFIX}{name}_"
-
-
-def _settings_plugin_name(settings_cls: Type[BaseSettings]) -> str:
-    plugin_name = getattr(settings_cls, "plugin_name", None)
-    if plugin_name is None:
-        raise TypeError(
-            f"{settings_cls.__name__} must set a 'plugin_name' class attribute to "
-            "namespace its configuration file and environment variables"
-        )
-    return validate_plugin_name(plugin_name)
-
-
-def _plugin_yaml_source(
-    settings_cls: Type[BaseSettings], config_path: str
-) -> YamlConfigSettingsSource:
-    """
-    Build the YAML settings source of a plugin, reporting unusable files.
-
-    A missing file is not an error, but a file which cannot be parsed or which
-    does not hold a mapping of setting names is reported with its path.
-    """
-    try:
-        return YamlConfigSettingsSource(settings_cls, config_path)
-    except yaml.YAMLError as exc:
-        raise PluginConfigLoadError(
-            f"Invalid plugin configuration file {config_path}: {exc}"
-        ) from exc
-    except (TypeError, ValueError) as exc:
-        raise PluginConfigLoadError(
-            f"Invalid plugin configuration file {config_path}: the file must "
-            "contain a mapping of setting names to values"
-        ) from exc
-
-
-class MVTPluginSettings(BaseSettings):
-    """
-    Base class for plugin-namespaced settings.
-
-    Subclass with typed fields and set `plugin_name`. Values resolve from
-    constructor arguments, then environment variables (MVT_PLUGIN_<NAME>_*),
-    then the plugin's YAML file (~/.config/mvt/plugins/<name>.yaml), then field
-    defaults.
-
-    Plugins must not store their settings in MVT's own configuration file: MVT
-    rewrites it with the fields it knows about, dropping anything else.
-
-    `data_folder()` returns the folder the plugin keeps its data in.
-    """
-
-    model_config = SettingsConfigDict(extra="ignore")
-
-    plugin_name: ClassVar[str]
-
-    def __init_subclass__(cls, **kwargs: Any) -> None:
-        super().__init_subclass__(**kwargs)
-        plugin_name = _settings_plugin_name(cls)
-        # Namespace the environment variables of this plugin. Each pydantic
-        # model gets its own configuration dictionary, so this does not leak
-        # into other plugins.
-        cls.model_config["env_prefix"] = plugin_env_prefix(plugin_name)
-
-    @classmethod
-    def settings_customise_sources(
-        cls,
-        settings_cls: Type[BaseSettings],
-        init_settings: PydanticBaseSettingsSource,
-        env_settings: PydanticBaseSettingsSource,
-        dotenv_settings: PydanticBaseSettingsSource,
-        file_secret_settings: PydanticBaseSettingsSource,
-    ) -> Tuple[PydanticBaseSettingsSource, ...]:
-        config_path = plugin_config_path(_settings_plugin_name(settings_cls))
-        yaml_source = _plugin_yaml_source(settings_cls, config_path)
-        # Explicit arguments take precedence over environment variables, which
-        # in turn take precedence over the configuration file.
-        return (init_settings, env_settings, yaml_source)
-
-    @classmethod
-    def load(cls: Type[PluginSettingsType]) -> PluginSettingsType:
-        """
-        Load the settings of the plugin.
-
-        A missing configuration file is not an error: the settings then come
-        from the environment and from the field defaults.
-        """
-        return cls()
-
-    @classmethod
-    def data_folder(cls) -> str:
-        """
-        Return the data folder of the plugin, creating it.
-
-        The folder is the one plugin_data_folder() returns for `plugin_name`,
-        so a plugin with a settings class does not repeat its name.
-        """
-        return plugin_data_folder(_settings_plugin_name(cls))
-
-    def _environment_values(self) -> Dict[str, Any]:
-        """
-        Return the settings values currently supplied by the environment.
-
-        Values are validated by the model, so they can be compared with the
-        values held by this instance.
-        """
-        settings_cls = type(self)
-        raw_values = EnvSettingsSource(settings_cls)()
-        names = [name for name in raw_values if name in settings_cls.model_fields]
-        if not names:
-            return {}
-
-        # Fall back on the current values for the fields the environment does
-        # not set, so that required fields do not fail validation here.
-        current_values = json.loads(self.model_dump_json())
-        try:
-            from_environment = settings_cls.model_validate(
-                {**current_values, **raw_values}
-            )
-        except ValidationError:
-            # An environment variable which the model cannot validate must not
-            # stop the other environment values from being recognised, or a
-            # credential would be written to the configuration file.
-            return self._environment_values_by_field(current_values, raw_values, names)
-        return {name: getattr(from_environment, name) for name in names}
-
-    def _environment_values_by_field(
-        self,
-        current_values: Dict[str, Any],
-        raw_values: Dict[str, Any],
-        names: List[str],
-    ) -> Dict[str, Any]:
-        """
-        Validate each environment value on its own, skipping unusable ones.
-
-        :param current_values: Serialized values held by this instance.
-        :param raw_values: Values supplied by the environment.
-        :param names: Names of the fields set by the environment.
-        """
-        settings_cls = type(self)
-        values = {}
-        for name in names:
-            try:
-                from_environment = settings_cls.model_validate(
-                    {**current_values, name: raw_values[name]}
-                )
-            except ValidationError:
-                continue
-            values[name] = getattr(from_environment, name)
-        return values
-
-    def save(self) -> None:
-        """
-        Save the current settings to the configuration file of the plugin.
-
-        Only values which differ from the field defaults are persisted. Values
-        which come from the environment are not written to disk, so credentials
-        passed as environment variables stay out of the configuration file.
-        MVT's own configuration file is never modified.
-        """
-        config_folder = plugin_config_folder()
-        if not os.path.isdir(config_folder):
-            os.makedirs(config_folder, mode=0o700, exist_ok=True)
-
-        values = json.loads(self.model_dump_json(exclude_defaults=True))
-        for name, environment_value in self._environment_values().items():
-            if name in values and getattr(self, name, None) == environment_value:
-                del values[name]
-
-        # Settings files can hold credentials, so write them through a private
-        # temporary file and move it in place. The file is then never partially
-        # written and never briefly readable by other users.
-        config_path = plugin_config_path(self.plugin_name)
-        descriptor, temporary_path = tempfile.mkstemp(
-            dir=config_folder, prefix=f".{self.plugin_name}-", suffix=".yaml"
-        )
-        try:
-            with os.fdopen(descriptor, "w") as config_file:
-                config_file.write(yaml.dump(values, default_flow_style=False))
-            os.replace(temporary_path, config_path)
-        except BaseException:
-            with contextlib.suppress(OSError):
-                os.unlink(temporary_path)
-            raise

src/mvt/common/utils.py

@@ -239,13 +239,6 @@ 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:

src/mvt/ios/modules/mixed/__init__.py

@@ -26,7 +26,6 @@ from .tcc import TCC
 from .webkit_resource_load_statistics import WebkitResourceLoadStatistics
 from .webkit_session_resource_log import WebkitSessionResourceLog
 from .whatsapp import Whatsapp
-from .whatsapp_contacts import WhatsappContacts
 
 MIXED_MODULES = [
     Calls,
@@ -48,7 +47,6 @@ MIXED_MODULES = [
     WebkitResourceLoadStatistics,
     WebkitSessionResourceLog,
     Whatsapp,
-    WhatsappContacts,
     Shortcuts,
     Applications,
     Calendar,

src/mvt/ios/modules/mixed/interactionc.py

@@ -3,11 +3,9 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-import datetime
 import logging
-import re
 import sqlite3
-from typing import Optional, Tuple
+from typing import Optional
 
 from mvt.common.module_types import (
     ModuleAtomicResult,
@@ -17,7 +15,6 @@ from mvt.common.module_types import (
 from mvt.common.utils import convert_mactime_to_iso
 
 from ..base import IOSExtraction
-from .whatsapp_contacts import WhatsappContacts
 
 INTERACTIONC_BACKUP_IDS = [
     "1f5a521220a3ad80ebfdc196978df8e7a2e49dee",
@@ -25,49 +22,6 @@ INTERACTIONC_BACKUP_IDS = [
 INTERACTIONC_ROOT_PATHS = [
     "private/var/mobile/Library/CoreDuet/People/interactionC.db",
 ]
-
-# The interaction record's creation date normally trails its start date by
-# milliseconds: emitting it as a timeline event only duplicates the start
-# date event. A large divergence, however, indicates the record was
-# backfilled (sync, restore, or tampering) and is worth surfacing.
-CREATION_DATE_DIVERGENCE_THRESHOLD = 3600.0
-
-# Per-contact aggregate dates from ZCONTACTS are repeated on every
-# interaction row of the same contact. They are serialized with a
-# contact-centric data string so that timeline de-duplication collapses
-# them into one event per contact.
-CONTACT_EVENT_TEMPLATES = {
-    "contacts_creation_date": "Contact {party} first recorded in interactionC",
-    "first_incoming_sender_date": "First incoming interaction from {party}",
-    "last_incoming_sender_date": "Last incoming interaction from {party}",
-    "first_incoming_recipient_date": (
-        "First incoming interaction where {party} was a recipient"
-    ),
-    "last_incoming_recipient_date": (
-        "Last incoming interaction where {party} was a recipient"
-    ),
-    "first_outgoing_recipient_date": (
-        "First outgoing interaction to {party}"
-    ),
-    "last_outgoing_recipient_date": (
-        "Last outgoing interaction to {party}"
-    ),
-}
-
-
-def _parse_iso(timestamp) -> Optional[datetime.datetime]:
-    try:
-        return datetime.datetime.strptime(
-            timestamp, "%Y-%m-%d %H:%M:%S.%f"
-        )
-    except (TypeError, ValueError):
-        return None
-
-
-def _describe_delta(seconds: float) -> str:
-    if seconds >= 86400:
-        return f"{seconds / 86400:.0f} days"
-    return f"{seconds / 3600:.0f} hours"
 # Taken from APOLLO
 # https://github.com/mac4n6/APOLLO/blob/master/modules/interaction_contact_interactions.txt
 QUERIES = [
@@ -80,7 +34,7 @@ QUERIES = [
             CASE ZINTERACTIONS.ZDIRECTION
                 WHEN '0' THEN 'INCOMING'
                 WHEN '1' THEN 'OUTGOING'
-            END AS "direction",
+            END 'DIRECTION' AS "direction",
             ZCONTACTS.ZDISPLAYNAME AS "sender_display_name",
             ZCONTACTS.ZIDENTIFIER AS "sender_identifier",
             ZCONTACTS.ZPERSONID AS "sender_personid",
@@ -135,7 +89,7 @@ QUERIES = [
             CASE ZINTERACTIONS.ZDIRECTION
                 WHEN '0' THEN 'INCOMING'
                 WHEN '1' THEN 'OUTGOING'
-            END AS "direction",
+            END 'DIRECTION' AS "direction",
             ZCONTACTS.ZDISPLAYNAME AS "sender_display_name",
             ZCONTACTS.ZIDENTIFIER AS "sender_identifier",
             ZCONTACTS.ZPERSONID AS "sender_personid",
@@ -163,7 +117,7 @@ QUERIES = [
             CASE ZCONTACTS.ZLASTINCOMINGRECIPIENTDATE
                 WHEN '0' THEN '0'
                 ELSE ZCONTACTS.ZLASTINCOMINGRECIPIENTDATE
-         END AS "last_incoming_recipient_date",
+         END 'LAST INCOMING RECIPIENT DATE' AS "last_incoming_recipient_date",
          ZCONTACTS.ZLASTOUTGOINGRECIPIENTDATE AS "last_outgoing_recipient_date",
          ZCONTACTS.ZCUSTOMIDENTIFIER AS "custom_id",
          ZINTERACTIONS.ZCONTENTURL AS "interaction_content_url",
@@ -264,16 +218,9 @@ QUERIES = [
 ]
 
 
-WHATSAPP_BUNDLE_ID = "net.whatsapp.WhatsApp"
-
-
 class InteractionC(IOSExtraction):
     """This module extracts data from InteractionC db."""
 
-    # WhatsApp identifies chat peers by LID in interactionC.db, which only the
-    # WhatsApp contacts database can map back to a phone number and name.
-    dependencies = [WhatsappContacts]
-
     def __init__(
         self,
         file_path: Optional[str] = None,
@@ -305,51 +252,10 @@ class InteractionC(IOSExtraction):
             "last_outgoing_recipient_date",
         ]
 
-    @staticmethod
-    def _describe_party(record: ModuleAtomicResult, prefix: str) -> Optional[str]:
-        name = record.get(f"{prefix}_display_name") or record.get(
-            f"{prefix}_resolved_name"
-        )
-        identifier = record.get(f"{prefix}_resolved_phone_number") or record.get(
-            f"{prefix}_identifier"
-        )
-        if name and identifier:
-            # A display name that is just a formatted copy of the phone
-            # number adds no information.
-            name_digits = re.sub(r"\D", "", name)
-            if name_digits and name_digits == re.sub(r"\D", "", identifier):
-                return identifier
-            return f"{name} ({identifier})"
-        return name or identifier or None
-
     def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
-        sender = self._describe_party(record, "sender")
-        # The chat peer from the domain identifier stands in when the
-        # recipient was not recorded (or the recipient join is unavailable).
-        recipient = self._describe_party(record, "recipient") or self._describe_party(
-            record, "domain"
-        )
-        direction = record.get("direction")
-        if not sender and direction == "OUTGOING":
-            sender = "local user"
-        if not recipient and direction == "INCOMING":
-            recipient = "local user"
-
-        header = f"[{record['bundle_id']}]"
-        if record.get("account"):
-            header += f" {record['account']}"
-        if direction:
-            header += f" {direction}"
-
-        data = f"{header} from {sender or 'unknown'} to {recipient or 'unknown'}"
-        if record.get("group_name"):
-            data += f" (group: {record['group_name']})"
-        if record.get("content"):
-            data += f": {record['content']}"
-
         records = []
         processed = []
-        for timestamp in ("start_date", "end_date"):
+        for timestamp in self.timestamps:
             # Check if the record has the current timestamp.
             if timestamp not in record or not record[timestamp]:
                 continue
@@ -363,142 +269,16 @@ class InteractionC(IOSExtraction):
                     "timestamp": record[timestamp],
                     "module": self.__class__.__name__,
                     "event": timestamp,
-                    "data": data,
+                    "data": f"[{record['bundle_id']}] {record['account']} - "
+                    f"from {record['sender_display_name']} ({record['sender_identifier']}) "
+                    f"to {record.get('recipient_display_name', '')} ({record.get('recipient_identifier', '')}):"
+                    f" {record.get('content', '')}",
                 }
             )
             processed.append(record[timestamp])
 
-        creation_event = self._serialize_creation_date(record, data)
-        if creation_event:
-            records.append(creation_event)
-
-        # Contact-level aggregates describe the sender's contact record.
-        party = self._describe_party(record, "sender")
-        if party:
-            for field, template in CONTACT_EVENT_TEMPLATES.items():
-                if not record.get(field):
-                    continue
-                records.append(
-                    {
-                        "timestamp": record[field],
-                        "module": self.__class__.__name__,
-                        "event": field,
-                        "data": template.format(party=party),
-                    }
-                )
-
         return records
 
-    def _serialize_creation_date(
-        self, record: ModuleAtomicResult, data: str
-    ) -> Optional[dict]:
-        """Serialize the interaction record's creation date only when it
-        diverges from the start date enough to indicate the record was
-        backfilled."""
-        creation = record.get("interactions_creation_date")
-        if not creation:
-            return None
-
-        event = {
-            "timestamp": creation,
-            "module": self.__class__.__name__,
-            "event": "interactions_creation_date",
-            "data": data,
-        }
-
-        start = _parse_iso(record.get("start_date"))
-        creation_parsed = _parse_iso(creation)
-        if not start or not creation_parsed:
-            # Without a start date the creation date is the only anchor.
-            return event
-
-        delta = (creation_parsed - start).total_seconds()
-        if abs(delta) < CREATION_DATE_DIVERGENCE_THRESHOLD:
-            return None
-
-        direction = "after" if delta > 0 else "before"
-        event["data"] = (
-            f"Interaction record created {_describe_delta(abs(delta))} "
-            f"{direction} the event: {data}"
-        )
-        return event
-
-    def _whatsapp_contact_maps(self) -> Tuple[dict, dict]:
-        """Build LID and phone-digit lookup maps from the WhatsappContacts
-        module results, when available."""
-        by_lid: dict = {}
-        by_phone: dict = {}
-        contacts_module = self.dependency_modules.get(WhatsappContacts)
-        if not contacts_module:
-            return by_lid, by_phone
-
-        for contact in contacts_module.results:
-            name = contact.get("full_name") or contact.get("given_name")
-            phone = contact.get("phone_number")
-            entry = (phone, name)
-            if contact.get("lid"):
-                by_lid[contact["lid"]] = entry
-            if phone:
-                by_phone[re.sub(r"\D", "", phone)] = entry
-            whatsapp_id = contact.get("whatsapp_id")
-            if whatsapp_id and "@" in whatsapp_id:
-                by_phone.setdefault(whatsapp_id.split("@")[0], entry)
-
-        return by_lid, by_phone
-
-    @staticmethod
-    def _resolve_whatsapp_identifier(
-        value, by_lid: dict, by_phone: dict
-    ) -> Tuple[Optional[str], Optional[str]]:
-        """Resolve a WhatsApp identifier (LID, JID or phone number) to a
-        (phone_number, contact_name) tuple."""
-        if not value:
-            return None, None
-
-        value = str(value)
-        if value.endswith("@lid"):
-            return by_lid.get(value, (None, None))
-        if value.endswith("@g.us"):
-            return None, None
-        if value.endswith("@s.whatsapp.net"):
-            digits = value.split("@")[0]
-            phone, name = by_phone.get(digits, (None, None))
-            return phone or f"+{digits}", name
-        if value.startswith("+"):
-            _, name = by_phone.get(re.sub(r"\D", "", value), (None, None))
-            return None, name
-
-        return None, None
-
-    def _postprocess_results(self) -> None:
-        by_lid, by_phone = self._whatsapp_contact_maps()
-
-        for entry in self.results:
-            # The fallback queries return ZDIRECTION raw instead of labelled.
-            if entry.get("direction") in (0, "0"):
-                entry["direction"] = "INCOMING"
-            elif entry.get("direction") in (1, "1"):
-                entry["direction"] = "OUTGOING"
-
-            if entry.get("bundle_id") != WHATSAPP_BUNDLE_ID:
-                continue
-
-            candidates = {
-                "sender": [entry.get("sender_identifier"), entry.get("custom_id")],
-                "recipient": [entry.get("recipient_identifier")],
-                "domain": [entry.get("domain_identifier")],
-            }
-            for prefix, values in candidates.items():
-                phone = name = None
-                for value in values:
-                    phone, name = self._resolve_whatsapp_identifier(
-                        value, by_lid, by_phone
-                    )
-                    if phone or name:
-                        break
-                entry[f"{prefix}_resolved_phone_number"] = phone
-                entry[f"{prefix}_resolved_name"] = name
-
     def run(self) -> None:
         self._find_ios_database(
             backup_ids=INTERACTIONC_BACKUP_IDS, root_paths=INTERACTIONC_ROOT_PATHS
@@ -545,6 +325,4 @@ class InteractionC(IOSExtraction):
             cur.close()
             conn.close()
 
-        self._postprocess_results()
-
         self.log.info("Extracted a total of %d InteractionC events", len(self.results))

src/mvt/ios/modules/mixed/whatsapp.py

@@ -4,9 +4,7 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-import os
-import sqlite3
-from typing import Dict, Optional
+from typing import Optional
 
 from mvt.common.module_types import (
     ModuleAtomicResult,
@@ -24,71 +22,9 @@ WHATSAPP_ROOT_PATHS = [
     "private/var/mobile/Containers/Shared/AppGroup/*/ChatStorage.sqlite",
 ]
 
-WHATSAPP_LID_BACKUP_IDS = [
-    # SHA-1 of "AppDomainGroup-group.net.whatsapp.WhatsApp.shared-LID.sqlite"
-    "e794f6ffcc3c222535f47684a63d5178da3c4500",
-]
-WHATSAPP_LID_ROOT_PATHS = [
-    "private/var/mobile/Containers/Shared/AppGroup/*/LID.sqlite",
-]
-
-# WhatsApp records the mapping between a contact's LID and phone number
-# identifiers in the ZWAPHONENUMBERLIDPAIR table. Depending on the WhatsApp
-# version this lives in a dedicated LID.sqlite database or in
-# ChatStorage.sqlite itself.
-LID_PAIRS_QUERY = """
-    SELECT
-        ZLID AS "lid",
-        ZPHONENUMBER AS "phone_number",
-        ZTIMESTAMP AS "pair_timestamp"
-    FROM ZWAPHONENUMBERLIDPAIR;
-"""
-
-CHAT_SESSIONS_QUERY = """
-    SELECT
-        ZWACHATSESSION.Z_PK AS "session_pk",
-        ZWACHATSESSION.ZCONTACTJID AS "contact_jid",
-        ZWACHATSESSION.ZPARTNERNAME AS "partner_name",
-        ZWACHATSESSION.ZSESSIONTYPE AS "session_type",
-        ZWACHATSESSION.ZARCHIVED AS "archived",
-        ZWACHATSESSION.ZREMOVED AS "removed",
-        ZWACHATSESSION.ZMESSAGECOUNTER AS "message_counter",
-        ZWACHATSESSION.ZLASTMESSAGEDATE AS "last_message_date",
-        ZWAGROUPINFO.ZCREATIONDATE AS "group_creation_date",
-        MIN(ZWAMESSAGE.ZMESSAGEDATE) AS "first_stored_message_date",
-        MAX(ZWAMESSAGE.ZMESSAGEDATE) AS "last_stored_message_date",
-        COUNT(ZWAMESSAGE.Z_PK) AS "stored_message_count"
-    FROM ZWACHATSESSION
-    LEFT JOIN ZWAGROUPINFO
-        ON ZWACHATSESSION.ZGROUPINFO = ZWAGROUPINFO.Z_PK
-    LEFT JOIN ZWAMESSAGE
-        ON ZWAMESSAGE.ZCHATSESSION = ZWACHATSESSION.Z_PK
-    GROUP BY ZWACHATSESSION.Z_PK;
-"""
-
-CHAT_SESSION_DATE_FIELDS = [
-    "last_message_date",
-    "group_creation_date",
-    "first_stored_message_date",
-    "last_stored_message_date",
-]
-
-
-def _describe_chat(record: ModuleAtomicResult) -> str:
-    jid = record.get("contact_jid") or "unknown"
-    name = record.get("partner_name")
-    identifier = record.get("partner_resolved_phone_number") or jid
-    label = f"'{name}' ({identifier})" if name else identifier
-    is_group = jid.endswith("@g.us") or record.get("group_creation_date")
-    if is_group:
-        return f"WhatsApp group chat {label}"
-    return f"WhatsApp chat with {label}"
-
 
 class Whatsapp(IOSExtraction):
-    """This module extracts all WhatsApp messages containing links, as well
-    as per-chat records with the first and last interaction dates of each
-    conversation."""
+    """This module extracts all WhatsApp messages containing links."""
 
     def __init__(
         self,
@@ -109,21 +45,6 @@ class Whatsapp(IOSExtraction):
         )
 
     def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
-        if record.get("record_type") == "chat_session":
-            return self._serialize_chat_session(record)
-        if record.get("record_type") == "lid_phone_number_pair":
-            if not record.get("pair_timestamp"):
-                return []
-            return {
-                "timestamp": record["pair_timestamp"],
-                "module": self.__class__.__name__,
-                "event": "lid_pair_recorded",
-                "data": (
-                    f"WhatsApp associated LID {record.get('lid')} with "
-                    f"phone number {record.get('phone_number')}"
-                ),
-            }
-
         text = record.get("ZTEXT", "").replace("\n", "\\n")
         links_text = ""
         if record.get("links"):
@@ -136,49 +57,6 @@ class Whatsapp(IOSExtraction):
             "data": f"'{text}' from {record.get('ZFROMJID', 'Unknown')}{links_text}",
         }
 
-    def _serialize_chat_session(
-        self, record: ModuleAtomicResult
-    ) -> ModuleSerializedResult:
-        records = []
-        chat = _describe_chat(record)
-
-        if record.get("group_creation_date"):
-            records.append(
-                {
-                    "timestamp": record["group_creation_date"],
-                    "module": self.__class__.__name__,
-                    "event": "group_created",
-                    "data": f"{chat} was created",
-                }
-            )
-
-        if record.get("first_stored_message_date"):
-            records.append(
-                {
-                    "timestamp": record["first_stored_message_date"],
-                    "module": self.__class__.__name__,
-                    "event": "chat_first_message",
-                    "data": f"First stored message in {chat}",
-                }
-            )
-
-        # The chat session's own last-message date is authoritative: it can
-        # postdate the newest stored message if that message was deleted.
-        last_message_date = record.get("last_message_date") or record.get(
-            "last_stored_message_date"
-        )
-        if last_message_date:
-            records.append(
-                {
-                    "timestamp": last_message_date,
-                    "module": self.__class__.__name__,
-                    "event": "chat_last_message",
-                    "data": f"Last message in {chat}",
-                }
-            )
-
-        return records
-
     def check_indicators(self) -> None:
         if not self.indicators:
             return
@@ -268,126 +146,7 @@ class Whatsapp(IOSExtraction):
                 message["links"] = list(set(filtered_links))
             self.results.append(message)
 
-        total_messages = len(self.results)
-        lid_map = self._extract_lid_pairs(cur)
-        total_sessions = self._extract_chat_sessions(cur, lid_map)
-
         cur.close()
         conn.close()
 
-        self.log.info(
-            "Extracted a total of %d WhatsApp messages, %d chat sessions "
-            "and %d LID-phone number pairs",
-            total_messages,
-            total_sessions,
-            len(lid_map),
-        )
-
-    def _find_lid_db_path(self) -> Optional[str]:
-        for backup_id in WHATSAPP_LID_BACKUP_IDS:
-            file_path = self._get_backup_file_from_id(backup_id)
-            if file_path and os.path.exists(file_path):
-                return file_path
-        for found_path in self._get_fs_files_from_patterns(
-            WHATSAPP_LID_ROOT_PATHS
-        ):
-            return found_path
-        return None
-
-    def _extract_lid_pairs(self, chat_cur: sqlite3.Cursor) -> Dict[str, str]:
-        """Extract the LID to phone number mapping from the dedicated
-        LID.sqlite database, falling back to the same table in
-        ChatStorage.sqlite. Returns a map of LID digits to phone number
-        digits."""
-        rows = []
-        lid_db_path = self._find_lid_db_path()
-        if lid_db_path:
-            self.log.info(
-                "Found WhatsApp LID database at path: %s", lid_db_path
-            )
-            lid_conn = self._open_sqlite_db(lid_db_path)
-            try:
-                lid_cur = lid_conn.cursor()
-                lid_cur.execute(LID_PAIRS_QUERY)
-                rows = lid_cur.fetchall()
-                lid_cur.close()
-            except sqlite3.DatabaseError as exc:
-                self.log.warning(
-                    "Unable to extract WhatsApp LID-phone number pairs: %s",
-                    exc,
-                )
-            finally:
-                lid_conn.close()
-        else:
-            try:
-                chat_cur.execute(LID_PAIRS_QUERY)
-                rows = chat_cur.fetchall()
-            except sqlite3.OperationalError:
-                self.log.info(
-                    "No WhatsApp LID database found in this backup or "
-                    "filesystem dump: LID chat identifiers cannot be "
-                    "resolved to phone numbers"
-                )
-
-        lid_map: Dict[str, str] = {}
-        for lid, phone_number, pair_timestamp in rows:
-            record = {
-                "record_type": "lid_phone_number_pair",
-                "lid": lid,
-                "phone_number": phone_number,
-                "pair_timestamp": (
-                    convert_mactime_to_iso(pair_timestamp) or None
-                )
-                if pair_timestamp
-                else None,
-            }
-            self.results.append(record)
-
-            if lid and phone_number:
-                lid_digits = str(lid).split("@")[0]
-                phone_digits = str(phone_number).split("@")[0].lstrip("+")
-                lid_map[lid_digits] = phone_digits
-
-        return lid_map
-
-    def _extract_chat_sessions(
-        self, cur: sqlite3.Cursor, lid_map: Dict[str, str]
-    ) -> int:
-        """Extract one record per chat session with the first and last
-        interaction dates of each conversation."""
-        try:
-            cur.execute(CHAT_SESSIONS_QUERY)
-        except sqlite3.OperationalError as exc:
-            self.log.warning(
-                "Unable to extract WhatsApp chat sessions: %s", exc
-            )
-            return 0
-
-        names = [description[0] for description in cur.description]
-
-        total_sessions = 0
-        for row in cur.fetchall():
-            session = dict(zip(names, row))
-            session["record_type"] = "chat_session"
-
-            session["partner_resolved_phone_number"] = None
-            jid = session.get("contact_jid") or ""
-            if jid.endswith("@lid"):
-                phone_digits = lid_map.get(jid.split("@")[0])
-                if phone_digits:
-                    session["partner_resolved_phone_number"] = (
-                        f"+{phone_digits}"
-                    )
-
-            for field in CHAT_SESSION_DATE_FIELDS:
-                if session.get(field):
-                    session[field] = (
-                        convert_mactime_to_iso(session[field]) or None
-                    )
-                else:
-                    session[field] = None
-
-            self.results.append(session)
-            total_sessions += 1
-
-        return total_sessions
+        self.log.info("Extracted a total of %d WhatsApp messages", len(self.results))

src/mvt/ios/modules/mixed/whatsapp_contacts.py

@@ -1,308 +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 logging
-import sqlite3
-from typing import Optional
-
-from mvt.common.module import DatabaseNotFoundError
-from mvt.common.module_types import (
-    ModuleAtomicResult,
-    ModuleResults,
-    ModuleSerializedResult,
-)
-from mvt.common.utils import convert_mactime_to_iso
-
-from ..base import IOSExtraction
-
-WHATSAPP_CONTACTS_BACKUP_IDS = [
-    # SHA-1 of "AppDomainGroup-group.net.whatsapp.WhatsApp.shared-ContactsV2.sqlite"
-    "b8548dc30aa1030df0ce18ef08b882cf7ab5212f",
-]
-WHATSAPP_CONTACTS_ROOT_PATHS = [
-    "private/var/mobile/Containers/Shared/AppGroup/*/ContactsV2.sqlite",
-]
-
-# WhatsApp's standard disappearing-messages timer values, in seconds.
-DISAPPEARING_DURATION_LABELS = {
-    86400: "24 hours",
-    604800: "7 days",
-    1209600: "14 days",
-    2592000: "30 days",
-    7776000: "90 days",
-}
-
-# Output field -> candidate columns in ZWAADDRESSBOOKCONTACT, in order of
-# preference. WhatsApp renames columns across versions, so the query is built
-# from the columns actually present in the database.
-COLUMN_CANDIDATES = {
-    "whatsapp_id": ["ZWHATSAPPID"],
-    "lid": ["ZLID"],
-    "phone_number": ["ZPHONENUMBER"],
-    "localized_phone_number": ["ZLOCALIZEDPHONENUMBER"],
-    "full_name": ["ZFULLNAME"],
-    "given_name": ["ZGIVENNAME"],
-    "last_name": ["ZLASTNAME"],
-    "user_name": ["ZUSERNAME"],
-    "business_name": ["ZBUSINESSNAME"],
-    "about_text": ["ZABOUTTEXT"],
-    "about_emoji": ["ZABOUTEMOJI"],
-    "notes": ["ZNOTES"],
-    "disappearing_mode_duration": ["ZDISAPPEARINGMODEDURATION"],
-    "disappearing_mode_timestamp": ["ZDISAPPEARINGMODETIMESTAMP"],
-    "about_timestamp": ["ZABOUTTIMESTAMP"],
-    "about_expiration_timestamp": ["ZABOUTEXPIRATIONTIMESTAMP"],
-    "last_updated": ["ZLASTUPDATED"],
-    "phone_status": ["ZPHONESTATUS", "ZPHONENUMBERSTATUS"],
-    "sync_policy": ["ZSYNCPOLICY"],
-}
-
-STRING_FIELDS = [
-    "whatsapp_id",
-    "lid",
-    "phone_number",
-    "localized_phone_number",
-    "full_name",
-    "given_name",
-    "last_name",
-    "user_name",
-    "business_name",
-    "about_text",
-    "about_emoji",
-    "notes",
-]
-
-DATE_FIELDS = [
-    "disappearing_mode_timestamp",
-    "about_timestamp",
-    "about_expiration_timestamp",
-    "last_updated",
-]
-
-
-def _decode_string(value) -> Optional[str]:
-    # CoreData stores string attributes as UTF-8 blobs in some WhatsApp
-    # versions, so values can arrive as either bytes or str.
-    if value is None:
-        return None
-    if isinstance(value, bytes):
-        return value.decode("utf-8", "replace")
-    return str(value)
-
-
-def _label_duration(duration) -> str:
-    if not duration:
-        return "off"
-    return DISAPPEARING_DURATION_LABELS.get(
-        int(duration), f"{int(duration)} seconds"
-    )
-
-
-def _describe_contact(record: ModuleAtomicResult) -> str:
-    contact = (
-        record.get("whatsapp_id")
-        or record.get("lid")
-        or record.get("phone_number")
-        or "unknown"
-    )
-    full_name = record.get("full_name")
-    if full_name:
-        contact = f"{contact} ({full_name})"
-    return contact
-
-
-class WhatsappContacts(IOSExtraction):
-    """This module extracts WhatsApp contact records and per-contact
-    disappearing-messages settings from ContactsV2.sqlite.
-
-    ChatStorage.sqlite does not record the disappearing-messages state of 1:1
-    chats: the authoritative timer is stored on each contact record in this
-    database, alongside the mapping between a contact's LID and phone number
-    identifiers.
-    """
-
-    def __init__(
-        self,
-        file_path: Optional[str] = None,
-        target_path: Optional[str] = None,
-        results_path: Optional[str] = None,
-        module_options: Optional[dict] = None,
-        log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[ModuleResults] = None,
-    ) -> None:
-        super().__init__(
-            file_path=file_path,
-            target_path=target_path,
-            results_path=results_path,
-            module_options=module_options,
-            log=log,
-            results=results,
-        )
-
-    def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
-        records = []
-        contact = _describe_contact(record)
-
-        if record.get("disappearing_mode_timestamp"):
-            records.append(
-                {
-                    "timestamp": record["disappearing_mode_timestamp"],
-                    "module": self.__class__.__name__,
-                    "event": "disappearing_mode_set",
-                    "data": (
-                        f"WhatsApp disappearing messages timer set to "
-                        f"'{record.get('disappearing_mode_label')}' "
-                        f"for {contact}"
-                    ),
-                }
-            )
-
-        if record.get("about_timestamp"):
-            data = f"WhatsApp about text of {contact} changed"
-            about_text = record.get("about_text")
-            if about_text:
-                data += f' to "{about_text}"'
-            records.append(
-                {
-                    "timestamp": record["about_timestamp"],
-                    "module": self.__class__.__name__,
-                    "event": "about_changed",
-                    "data": data,
-                }
-            )
-
-        if record.get("about_expiration_timestamp"):
-            records.append(
-                {
-                    "timestamp": record["about_expiration_timestamp"],
-                    "module": self.__class__.__name__,
-                    "event": "about_expiration",
-                    "data": (
-                        f"WhatsApp about text of {contact} scheduled "
-                        f"to expire"
-                    ),
-                }
-            )
-
-        if record.get("last_updated"):
-            records.append(
-                {
-                    "timestamp": record["last_updated"],
-                    "module": self.__class__.__name__,
-                    "event": "contact_last_updated",
-                    "data": f"WhatsApp contact record for {contact} updated",
-                }
-            )
-
-        return records
-
-    def run(self) -> None:
-        try:
-            self._find_ios_database(
-                backup_ids=WHATSAPP_CONTACTS_BACKUP_IDS,
-                root_paths=WHATSAPP_CONTACTS_ROOT_PATHS,
-            )
-        except DatabaseNotFoundError:
-            self.log.warning(
-                "Unable to find the WhatsApp ContactsV2.sqlite database in "
-                "this backup or filesystem dump. WhatsApp disappearing "
-                "messages settings and contact records cannot be extracted. "
-                "This database is often missing from incremental backups."
-            )
-            return
-
-        self.log.info(
-            "Found WhatsApp contacts database at path: %s", self.file_path
-        )
-
-        assert self.file_path is not None
-        conn = self._open_sqlite_db(self.file_path)
-        cur = conn.cursor()
-        try:
-            try:
-                cur.execute("PRAGMA table_info(ZWAADDRESSBOOKCONTACT)")
-                available_columns = {row[1] for row in cur.fetchall()}
-            except sqlite3.DatabaseError as exc:
-                self.log.error(
-                    "Unable to read the ZWAADDRESSBOOKCONTACT table schema: %s",
-                    exc,
-                )
-                return
-
-            if not available_columns:
-                self.log.warning(
-                    "The WhatsApp contacts database does not contain a "
-                    "ZWAADDRESSBOOKCONTACT table"
-                )
-                return
-
-            selected = {}
-            for field, candidates in COLUMN_CANDIDATES.items():
-                for candidate in candidates:
-                    if candidate in available_columns:
-                        selected[field] = candidate
-                        break
-
-            # A record with no duration column is "unknown", not "off": the
-            # timer state cannot be determined from this database version.
-            has_duration = "disappearing_mode_duration" in selected
-            if not has_duration:
-                self.log.warning(
-                    "The ZDISAPPEARINGMODEDURATION column is not present in "
-                    "this WhatsApp contacts database: disappearing messages "
-                    "state is unknown"
-                )
-
-            columns = ["Z_PK"] + list(selected.values())
-            cur.execute(
-                f"SELECT {', '.join(columns)} FROM ZWAADDRESSBOOKCONTACT;"
-            )
-            fields = ["row_pk"] + list(selected.keys())
-
-            for row in cur:
-                record = dict(zip(fields, row))
-
-                for field in STRING_FIELDS:
-                    if field in record:
-                        record[field] = _decode_string(record[field])
-                    else:
-                        record[field] = None
-
-                for field in DATE_FIELDS:
-                    if record.get(field) is not None:
-                        record[field] = (
-                            convert_mactime_to_iso(record[field]) or None
-                        )
-                    else:
-                        record[field] = None
-
-                duration = record.get("disappearing_mode_duration")
-                if has_duration:
-                    record["disappearing_mode_is_on"] = bool(duration)
-                    record["disappearing_mode_label"] = _label_duration(
-                        duration
-                    )
-                else:
-                    record["disappearing_mode_duration"] = None
-                    record["disappearing_mode_is_on"] = None
-                    record["disappearing_mode_label"] = None
-
-                record.setdefault("phone_status", None)
-                record.setdefault("sync_policy", None)
-
-                self.results.append(record)
-        finally:
-            cur.close()
-            conn.close()
-
-        total_ephemeral = sum(
-            1 for record in self.results if record["disappearing_mode_is_on"]
-        )
-        self.log.info(
-            "Extracted a total of %d WhatsApp contacts (%d with disappearing "
-            "messages enabled)",
-            len(self.results),
-            total_ephemeral,
-        )

tests/common/test_module_loader.py

@@ -3,12 +3,10 @@ import pytest
 from mvt.common.module import MVTModule
 from mvt.common.module_loader import (
     CustomModuleLoadError,
-    get_module_logger,
     load_custom_modules,
     load_custom_modules_from_path,
     module_supports_command,
 )
-from mvt.ios.modules.mixed.whatsapp import Whatsapp
 
 
 MODULE_TEMPLATE = """
@@ -146,44 +144,3 @@ def test_module_supports_command_honors_supported_commands(tmp_path):
 
     assert module_supports_command(module, "ios", "check-backup")
     assert not module_supports_command(module, "ios", "check-fs")
-
-
-def test_get_module_logger_keeps_builtin_names():
-    assert get_module_logger(Whatsapp).name == "mvt.ios.modules.mixed.whatsapp"
-
-
-def test_get_module_logger_parents_package_modules_under_mvt_ext():
-    class PackageModule(MVTModule):
-        pass
-
-    PackageModule.__module__ = "some_plugin_package.ios.custom"
-
-    assert (
-        get_module_logger(PackageModule).name
-        == "mvt.ext.some_plugin_package.ios.custom"
-    )
-
-
-def test_get_module_logger_strips_the_plugin_package_prefix():
-    class PluginModule(MVTModule):
-        pass
-
-    PluginModule.__module__ = "mvt_plugin_amnesty_custom.ios.custom"
-
-    assert get_module_logger(PluginModule).name == "mvt.ext.amnesty_custom.ios.custom"
-
-
-def test_get_module_logger_only_strips_the_prefix_from_the_top_level():
-    class NestedModule(MVTModule):
-        pass
-
-    NestedModule.__module__ = "other_package.mvt_plugin_sub"
-
-    assert get_module_logger(NestedModule).name == "mvt.ext.other_package.mvt_plugin_sub"
-
-
-def test_get_module_logger_names_path_modules_after_their_file(tmp_path):
-    module_path = _write_module(tmp_path / "my_custom_module.py", "PathModule")
-    module = load_custom_modules_from_path(str(module_path))[0]
-
-    assert get_module_logger(module).name == "mvt.ext.my_custom_module"

tests/common/test_plugin_config.py

@@ -1,391 +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 os
-import stat
-import sys
-from typing import Optional
-
-import pytest
-import yaml
-
-from mvt.common.plugin_config import (
-    MVTPluginSettings,
-    PluginConfigLoadError,
-    plugin_config_folder,
-    plugin_config_path,
-    plugin_data_folder,
-    plugin_env_prefix,
-)
-
-
-class ExamplePluginSettings(MVTPluginSettings):
-    plugin_name = "example-plugin"
-
-    API_KEY: Optional[str] = None
-    CACHE_FOLDER: str = "cache"
-    MAX_RESULTS: int = 25
-
-
-class OtherPluginSettings(MVTPluginSettings):
-    plugin_name = "other-plugin"
-
-    API_KEY: Optional[str] = None
-
-
-@pytest.fixture
-def config_folder(tmp_path, monkeypatch):
-    monkeypatch.setattr(
-        "mvt.common.plugin_config.user_config_dir",
-        lambda *args, **kwargs: str(tmp_path),
-    )
-    return tmp_path
-
-
-@pytest.fixture
-def data_folder(tmp_path, monkeypatch):
-    folder = tmp_path / "data"
-    monkeypatch.setattr(
-        "mvt.common.plugin_config.user_data_dir",
-        lambda *args, **kwargs: str(folder),
-    )
-    return folder
-
-
-def _write_plugin_file(plugin_name, values):
-    config_path = plugin_config_path(plugin_name)
-    os.makedirs(os.path.dirname(config_path), exist_ok=True)
-    content = values if isinstance(values, str) else yaml.dump(values)
-    with open(config_path, "w") as config_file:
-        config_file.write(content)
-    return config_path
-
-
-def test_plugin_paths_and_prefixes_are_namespaced(config_folder):
-    assert plugin_config_folder() == str(config_folder / "plugins")
-    assert plugin_config_path("example-plugin") == str(
-        config_folder / "plugins" / "example-plugin.yaml"
-    )
-    assert plugin_env_prefix("example-plugin") == "MVT_PLUGIN_EXAMPLE_PLUGIN_"
-    assert plugin_env_prefix("other-plugin") == "MVT_PLUGIN_OTHER_PLUGIN_"
-
-
-def test_defaults_are_used_without_file_or_environment(config_folder):
-    settings = ExamplePluginSettings.load()
-
-    assert settings.API_KEY is None
-    assert settings.CACHE_FOLDER == "cache"
-    assert settings.MAX_RESULTS == 25
-    assert not os.path.exists(plugin_config_path("example-plugin"))
-
-
-def test_values_are_loaded_from_the_plugin_file(config_folder):
-    _write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
-
-    settings = ExamplePluginSettings.load()
-
-    assert settings.API_KEY == "from-file"
-    assert settings.MAX_RESULTS == 5
-    assert settings.CACHE_FOLDER == "cache"
-
-
-def test_environment_overrides_the_plugin_file(config_folder, monkeypatch):
-    _write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
-    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "from-environment")
-
-    settings = ExamplePluginSettings.load()
-
-    assert settings.API_KEY == "from-environment"
-    assert settings.MAX_RESULTS == 5
-
-
-def test_arguments_override_the_environment_and_the_plugin_file(
-    config_folder, monkeypatch
-):
-    _write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
-    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "from-environment")
-    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS", "10")
-
-    settings = ExamplePluginSettings(API_KEY="from-argument")
-
-    assert settings.API_KEY == "from-argument"
-    assert settings.MAX_RESULTS == 10
-
-
-def test_save_and_load_round_trip(config_folder):
-    settings = ExamplePluginSettings.load()
-    settings.API_KEY = "saved-key"
-    settings.MAX_RESULTS = 100
-
-    settings.save()
-
-    config_path = plugin_config_path("example-plugin")
-    assert os.path.isfile(config_path)
-    with open(config_path) as config_file:
-        assert yaml.safe_load(config_file) == {
-            "API_KEY": "saved-key",
-            "MAX_RESULTS": 100,
-        }
-
-    reloaded = ExamplePluginSettings.load()
-    assert reloaded.API_KEY == "saved-key"
-    assert reloaded.MAX_RESULTS == 100
-    assert reloaded.CACHE_FOLDER == "cache"
-
-
-@pytest.mark.skipif(
-    sys.platform == "win32", reason="POSIX file permissions are not available"
-)
-def test_saved_file_is_only_readable_by_the_user(config_folder):
-    settings = ExamplePluginSettings.load()
-    settings.API_KEY = "saved-key"
-
-    settings.save()
-
-    config_path = plugin_config_path("example-plugin")
-    assert stat.S_IMODE(os.stat(config_path).st_mode) == 0o600
-    folder_mode = stat.S_IMODE(os.stat(plugin_config_folder()).st_mode)
-    assert folder_mode & 0o077 == 0
-
-
-@pytest.mark.skipif(
-    sys.platform == "win32", reason="POSIX file permissions are not available"
-)
-def test_save_restricts_the_permissions_of_an_existing_file(config_folder):
-    config_path = _write_plugin_file("example-plugin", {"API_KEY": "from-file"})
-    os.chmod(config_path, 0o644)
-
-    settings = ExamplePluginSettings.load()
-    settings.MAX_RESULTS = 100
-    settings.save()
-
-    assert stat.S_IMODE(os.stat(config_path).st_mode) == 0o600
-    assert os.listdir(plugin_config_folder()) == ["example-plugin.yaml"]
-
-
-def test_save_only_persists_non_default_values(config_folder):
-    settings = ExamplePluginSettings.load()
-    settings.CACHE_FOLDER = "another-cache"
-
-    settings.save()
-
-    with open(plugin_config_path("example-plugin")) as config_file:
-        assert yaml.safe_load(config_file) == {"CACHE_FOLDER": "another-cache"}
-
-
-def test_save_does_not_persist_values_coming_from_the_environment(
-    config_folder, monkeypatch
-):
-    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
-
-    settings = ExamplePluginSettings.load()
-    assert settings.API_KEY == "environment-secret"
-    settings.MAX_RESULTS = 100
-    settings.save()
-
-    with open(plugin_config_path("example-plugin")) as config_file:
-        assert yaml.safe_load(config_file) == {"MAX_RESULTS": 100}
-
-
-def test_an_invalid_environment_variable_still_protects_the_other_values(
-    config_folder, monkeypatch
-):
-    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
-    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS", "not-an-int")
-
-    settings = ExamplePluginSettings(MAX_RESULTS=100)
-    assert settings.API_KEY == "environment-secret"
-    settings.save()
-
-    with open(plugin_config_path("example-plugin")) as config_file:
-        saved_values = yaml.safe_load(config_file)
-    assert saved_values == {"MAX_RESULTS": 100}
-    assert "API_KEY" not in saved_values
-
-
-def test_save_persists_values_which_differ_from_the_environment(
-    config_folder, monkeypatch
-):
-    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
-
-    settings = ExamplePluginSettings.load()
-    settings.API_KEY = "chosen-key"
-    settings.save()
-
-    with open(plugin_config_path("example-plugin")) as config_file:
-        assert yaml.safe_load(config_file) == {"API_KEY": "chosen-key"}
-
-
-def test_save_does_not_write_the_mvt_configuration_file(config_folder):
-    settings = ExamplePluginSettings.load()
-    settings.API_KEY = "saved-key"
-
-    settings.save()
-
-    assert os.listdir(config_folder) == ["plugins"]
-
-
-def test_unknown_keys_in_the_plugin_file_are_ignored(config_folder):
-    _write_plugin_file(
-        "example-plugin",
-        {"API_KEY": "from-file", "UNKNOWN_SETTING": "ignored"},
-    )
-
-    settings = ExamplePluginSettings.load()
-
-    assert settings.API_KEY == "from-file"
-    assert not hasattr(settings, "UNKNOWN_SETTING")
-
-
-def test_unparsable_plugin_file_is_reported_with_its_path(config_folder):
-    config_path = _write_plugin_file("example-plugin", "API_KEY: [unclosed\n")
-
-    with pytest.raises(PluginConfigLoadError) as raised:
-        ExamplePluginSettings.load()
-
-    assert config_path in str(raised.value)
-
-
-def test_plugin_file_which_is_not_a_mapping_is_reported_with_its_path(config_folder):
-    config_path = _write_plugin_file("example-plugin", "- one\n- two\n")
-
-    with pytest.raises(PluginConfigLoadError) as raised:
-        ExamplePluginSettings.load()
-
-    assert config_path in str(raised.value)
-    assert "mapping of setting names" in str(raised.value)
-
-
-def test_plugins_do_not_interfere_with_each_other(config_folder, monkeypatch):
-    _write_plugin_file("other-plugin", {"API_KEY": "other-file-key"})
-    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "example-environment-key")
-
-    example_settings = ExamplePluginSettings.load()
-    other_settings = OtherPluginSettings.load()
-
-    assert example_settings.API_KEY == "example-environment-key"
-    assert other_settings.API_KEY == "other-file-key"
-
-    example_settings.MAX_RESULTS = 100
-    example_settings.save()
-    assert sorted(os.listdir(plugin_config_folder())) == [
-        "example-plugin.yaml",
-        "other-plugin.yaml",
-    ]
-    with open(plugin_config_path("other-plugin")) as config_file:
-        assert yaml.safe_load(config_file) == {"API_KEY": "other-file-key"}
-
-
-def test_subclass_without_plugin_name_is_rejected():
-    with pytest.raises(TypeError, match="plugin_name"):
-
-        class MissingNameSettings(MVTPluginSettings):
-            API_KEY: Optional[str] = None
-
-
-def test_subclass_with_invalid_plugin_name_is_rejected():
-    with pytest.raises(ValueError, match="Invalid plugin name"):
-
-        class InvalidNameSettings(MVTPluginSettings):
-            plugin_name = "Bad/Name"
-
-
-def test_underscores_are_not_allowed_in_plugin_names():
-    # Underscores are replaced by dashes in the environment prefix, so allowing
-    # both would let two plugin names share one environment namespace.
-    with pytest.raises(ValueError, match="Invalid plugin name"):
-
-        class UnderscoreNameSettings(MVTPluginSettings):
-            plugin_name = "under_score"
-
-    with pytest.raises(ValueError, match="Invalid plugin name"):
-        plugin_config_path("under_score")
-    with pytest.raises(ValueError, match="Invalid plugin name"):
-        plugin_env_prefix("under_score")
-
-
-@pytest.mark.parametrize(
-    "plugin_name", ["../escape", "folder/name", "UPPER", "-dash", ""]
-)
-def test_unsafe_plugin_names_have_no_configuration_path(plugin_name):
-    with pytest.raises(ValueError, match="Invalid plugin name"):
-        plugin_config_path(plugin_name)
-
-
-def test_data_folder_is_namespaced_and_created(data_folder):
-    folder = plugin_data_folder("example-plugin")
-
-    assert folder == str(data_folder / "plugin-data" / "example-plugin")
-    assert os.path.isdir(folder)
-
-
-def test_data_folder_can_be_requested_repeatedly(data_folder):
-    folder = plugin_data_folder("example-plugin")
-    with open(os.path.join(folder, "kept.json"), "w") as data_file:
-        data_file.write("{}")
-
-    assert plugin_data_folder("example-plugin") == folder
-    assert os.listdir(folder) == ["kept.json"]
-
-
-@pytest.mark.skipif(
-    sys.platform == "win32", reason="POSIX file permissions are not available"
-)
-def test_data_folder_is_only_accessible_by_the_user(data_folder):
-    folder = plugin_data_folder("example-plugin")
-
-    assert stat.S_IMODE(os.stat(folder).st_mode) & 0o077 == 0
-    parent_mode = stat.S_IMODE(os.stat(os.path.dirname(folder)).st_mode)
-    assert parent_mode & 0o077 == 0
-
-
-def test_plugins_get_their_own_data_folder(data_folder):
-    example_folder = plugin_data_folder("example-plugin")
-    other_folder = plugin_data_folder("other-plugin")
-
-    assert example_folder != other_folder
-    assert sorted(os.listdir(data_folder / "plugin-data")) == [
-        "example-plugin",
-        "other-plugin",
-    ]
-
-
-def test_data_folder_does_not_touch_the_configuration_folder(
-    config_folder, data_folder
-):
-    plugin_data_folder("example-plugin")
-
-    assert not os.path.exists(plugin_config_folder())
-
-
-@pytest.mark.parametrize(
-    "plugin_name", ["../escape", "folder/name", "UPPER", "-dash", ""]
-)
-def test_unsafe_plugin_names_have_no_data_folder(data_folder, plugin_name):
-    with pytest.raises(ValueError, match="Invalid plugin name"):
-        plugin_data_folder(plugin_name)
-
-    assert not os.path.exists(data_folder)
-
-
-def test_settings_class_knows_its_data_folder(data_folder):
-    folder = ExamplePluginSettings.data_folder()
-
-    assert folder == plugin_data_folder("example-plugin")
-    assert os.path.isdir(folder)
-    assert OtherPluginSettings.data_folder() != folder
-
-
-def test_settings_instance_uses_the_same_data_folder(config_folder, data_folder):
-    settings = ExamplePluginSettings.load()
-
-    assert settings.data_folder() == ExamplePluginSettings.data_folder()
-
-
-def test_subclass_without_its_own_name_shares_the_data_folder(data_folder):
-    class InheritingSettings(ExamplePluginSettings):
-        pass
-
-    assert InheritingSettings.data_folder() == ExamplePluginSettings.data_folder()

tests/common/test_utils.py

@@ -8,7 +8,6 @@ import logging
 import os
 from datetime import datetime
 
-from mvt.common.log import MVTLogHandler
 from mvt.common.utils import (
     CustomJSONEncoder,
     convert_datetime_to_iso,
@@ -17,7 +16,6 @@ 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
@@ -105,20 +103,3 @@ 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/ios_backup/test_interactionc.py

@@ -1,105 +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/
-
-from mvt.common.module import run_module
-from mvt.ios.modules.mixed.interactionc import InteractionC
-from mvt.ios.modules.mixed.whatsapp_contacts import WhatsappContacts
-
-from ..utils import get_ios_backup_folder
-
-
-class TestInteractionCModule:
-    def test_extraction_with_whatsapp_contacts(self):
-        contacts = WhatsappContacts(target_path=get_ios_backup_folder())
-        run_module(contacts)
-
-        m = InteractionC(target_path=get_ios_backup_folder())
-        m.dependency_modules = {WhatsappContacts: contacts}
-        run_module(m)
-
-        assert len(m.results) == 3
-
-        incoming = next(
-            r for r in m.results if r["sender_identifier"] == "100000000000001@lid"
-        )
-        assert incoming["direction"] == "INCOMING"
-        assert incoming["sender_resolved_phone_number"] == "+14155550100"
-        assert incoming["sender_resolved_name"] == "Alice Example"
-
-        outgoing = next(
-            r for r in m.results if r["direction"] == "OUTGOING"
-        )
-        assert outgoing["recipient_identifier"] == "+14155550100"
-        assert outgoing["recipient_resolved_name"] == "Alice Example"
-        assert outgoing["domain_resolved_phone_number"] == "+14155550100"
-        assert outgoing["domain_resolved_name"] == "Alice Example"
-
-        sms = next(
-            r for r in m.results if r["bundle_id"] == "com.apple.MobileSMS"
-        )
-        assert sms.get("sender_resolved_name") is None
-        assert sms["sender_display_name"] == "Bob Example"
-
-        events = [entry["data"] for entry in m.timeline]
-        assert (
-            "[net.whatsapp.WhatsApp] INCOMING from "
-            "Alice Example (+14155550100) to local user" in events
-        )
-        assert (
-            "[net.whatsapp.WhatsApp] OUTGOING from local user to "
-            "Alice Example (+14155550100)" in events
-        )
-        assert (
-            "[com.apple.MobileSMS] INCOMING from "
-            "Bob Example (+14155550101) to local user" in events
-        )
-
-        # The creation date is only serialized when it diverges from the
-        # start date; the SMS record was created 90 days after the event.
-        creation_events = [
-            entry
-            for entry in m.timeline
-            if entry["event"] == "interactions_creation_date"
-        ]
-        assert len(creation_events) == 1
-        assert creation_events[0]["timestamp"] == "2025-12-09 12:26:40.000000"
-        assert creation_events[0]["data"] == (
-            "Interaction record created 90 days after the event: "
-            "[com.apple.MobileSMS] INCOMING from "
-            "Bob Example (+14155550101) to local user"
-        )
-
-        # Per-contact aggregate dates use contact-centric data strings.
-        first_seen = [
-            entry
-            for entry in m.timeline
-            if entry["event"] == "first_incoming_sender_date"
-        ]
-        assert len(first_seen) == 1
-        assert first_seen[0]["timestamp"] == "2025-09-03 13:46:40.000000"
-        assert first_seen[0]["data"] == (
-            "First incoming interaction from Bob Example (+14155550101)"
-        )
-        assert (
-            "Last incoming interaction from Bob Example (+14155550101)"
-            in events
-        )
-
-    def test_extraction_without_whatsapp_contacts(self):
-        # Without the WhatsappContacts dependency the module still runs, and
-        # unresolvable LIDs are shown as-is.
-        m = InteractionC(target_path=get_ios_backup_folder())
-        run_module(m)
-
-        assert len(m.results) == 3
-        events = [entry["data"] for entry in m.timeline]
-        assert (
-            "[net.whatsapp.WhatsApp] INCOMING from "
-            "100000000000001@lid to local user" in events
-        )
-        assert (
-            "[net.whatsapp.WhatsApp] OUTGOING from local user to "
-            "+14155550100" in events
-        )

tests/ios_backup/test_whatsapp.py

@@ -6,79 +6,8 @@
 import logging
 
 from mvt.common.indicators import Indicators
-from mvt.common.module import run_module
 from mvt.ios.modules.mixed.whatsapp import Whatsapp
 
-from ..utils import get_ios_backup_folder
-
-
-def test_extraction():
-    m = Whatsapp(target_path=get_ios_backup_folder())
-    run_module(m)
-
-    messages = [r for r in m.results if "ZTEXT" in r]
-    sessions = [r for r in m.results if r.get("record_type") == "chat_session"]
-    pairs = [
-        r for r in m.results
-        if r.get("record_type") == "lid_phone_number_pair"
-    ]
-    assert len(messages) == 3
-    assert len(sessions) == 2
-    assert len(pairs) == 1
-
-    assert pairs[0]["lid"] == "100000000000001"
-    assert pairs[0]["phone_number"] == "14155550100"
-    assert pairs[0]["pair_timestamp"] == "2025-08-25 07:33:20.000000"
-
-    linked = next(r for r in messages if r.get("links"))
-    assert linked["links"] == ["https://example.org/news"]
-
-    alice = next(s for s in sessions if s["partner_name"] == "Alice Example")
-    assert alice["contact_jid"] == "100000000000001@lid"
-    assert alice["partner_resolved_phone_number"] == "+14155550100"
-    assert alice["first_stored_message_date"] == "2025-08-27 15:06:40.000000"
-    assert alice["last_message_date"] == "2025-08-28 18:53:20.000000"
-    assert alice["group_creation_date"] is None
-    assert alice["stored_message_count"] == 2
-
-    group = next(s for s in sessions if s["partner_name"] == "Example Group")
-    assert group["group_creation_date"] == "2025-08-21 20:13:20.000000"
-    assert group["first_stored_message_date"] == "2025-08-29 22:40:00.000000"
-    # The last stored message predates the session's own last-message date:
-    # the newest message in this chat was deleted.
-    assert group["last_stored_message_date"] == "2025-08-29 22:40:00.000000"
-    assert group["last_message_date"] == "2025-08-31 02:26:40.000000"
-
-    # 3 message events, first/last per chat, the group creation and the
-    # LID-phone number pair.
-    assert len(m.timeline) == 9
-    events = {
-        (entry["event"], entry["timestamp"]): entry["data"]
-        for entry in m.timeline
-    }
-    # Alice's session is keyed by LID but labelled with the phone number
-    # resolved through LID.sqlite.
-    assert events[("chat_first_message", "2025-08-27 15:06:40.000000")] == (
-        "First stored message in WhatsApp chat with "
-        "'Alice Example' (+14155550100)"
-    )
-    assert events[("chat_last_message", "2025-08-28 18:53:20.000000")] == (
-        "Last message in WhatsApp chat with "
-        "'Alice Example' (+14155550100)"
-    )
-    assert events[("lid_pair_recorded", "2025-08-25 07:33:20.000000")] == (
-        "WhatsApp associated LID 100000000000001 with "
-        "phone number 14155550100"
-    )
-    assert events[("group_created", "2025-08-21 20:13:20.000000")] == (
-        "WhatsApp group chat 'Example Group' "
-        "(120000000000000001@g.us) was created"
-    )
-    assert ("chat_first_message", "2025-08-29 22:40:00.000000") in events
-    assert ("chat_last_message", "2025-08-31 02:26:40.000000") in events
-
-    assert len(m.alertstore.alerts) == 0
-
 
 def test_collect_url_results_includes_expansion():
     module = Whatsapp(

tests/ios_backup/test_whatsapp_contacts.py

@@ -1,83 +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/
-
-from mvt.common.module import run_module
-from mvt.ios.modules.mixed.whatsapp_contacts import WhatsappContacts
-
-from ..utils import get_ios_backup_folder
-
-
-class TestWhatsappContactsModule:
-    def test_extraction(self):
-        m = WhatsappContacts(target_path=get_ios_backup_folder())
-        run_module(m)
-        assert len(m.results) == 2
-
-        alice = next(r for r in m.results if r["given_name"] == "Alice")
-        assert alice["full_name"] == "Alice Example"
-        assert alice["phone_number"] == "+14155550100"
-        assert alice["whatsapp_id"] == "14155550100@s.whatsapp.net"
-        assert alice["lid"] == "100000000000001@lid"
-        assert alice["user_name"] == "alice.example"
-        assert alice["disappearing_mode_duration"] == 86400.0
-        assert alice["disappearing_mode_is_on"] is True
-        assert alice["disappearing_mode_label"] == "24 hours"
-        assert alice["disappearing_mode_timestamp"] == "2025-07-23 21:46:40.000000"
-        assert alice["about_timestamp"] == "2025-07-12 08:00:00.000000"
-        assert alice["about_expiration_timestamp"] == "2025-08-16 01:20:00.000000"
-        assert alice["last_updated"] == "2025-08-04 11:33:20.000000"
-
-        bob = next(r for r in m.results if r["given_name"] == "Bob")
-        assert bob["lid"] is None
-        assert bob["disappearing_mode_duration"] is None
-        assert bob["disappearing_mode_is_on"] is False
-        assert bob["disappearing_mode_label"] == "off"
-        assert bob["disappearing_mode_timestamp"] is None
-
-        # Alice: disappearing_mode_set, about_changed, about_expiration and
-        # contact_last_updated. Bob: contact_last_updated only.
-        assert len(m.timeline) == 5
-
-        events = {
-            (entry["event"], entry["timestamp"]): entry["data"]
-            for entry in m.timeline
-        }
-        assert (
-            "24 hours"
-            in events[("disappearing_mode_set", "2025-07-23 21:46:40.000000")]
-        )
-        assert (
-            "14155550100@s.whatsapp.net (Alice Example)"
-            in events[("disappearing_mode_set", "2025-07-23 21:46:40.000000")]
-        )
-        assert (
-            'changed to "Hey there! I am using WhatsApp."'
-            in events[("about_changed", "2025-07-12 08:00:00.000000")]
-        )
-        assert (
-            "scheduled to expire"
-            in events[("about_expiration", "2025-08-16 01:20:00.000000")]
-        )
-
-        updated = [
-            entry["data"]
-            for entry in m.timeline
-            if entry["event"] == "contact_last_updated"
-        ]
-        assert len(updated) == 2
-        assert all(
-            entry["timestamp"] == "2025-08-04 11:33:20.000000"
-            for entry in m.timeline
-            if entry["event"] == "contact_last_updated"
-        )
-        assert any("14155550101@s.whatsapp.net (Bob Example)" in d for d in updated)
-
-        assert len(m.alertstore.alerts) == 0
-
-    def test_missing_database(self, tmp_path):
-        m = WhatsappContacts(target_path=str(tmp_path))
-        run_module(m)
-        assert m.results == []
-        assert len(m.alertstore.alerts) == 0

tests/ios_fs/test_filesystem.py

@@ -15,8 +15,8 @@ class TestFilesystem:
     def test_filesystem(self):
         m = Filesystem(target_path=get_ios_backup_folder())
         run_module(m)
-        assert len(m.results) == 23
-        assert len(m.timeline) == 23
+        assert len(m.results) == 15
+        assert len(m.timeline) == 15
         assert len(m.alertstore.alerts) == 0
 
     def test_detection(self, indicator_file):
@@ -29,6 +29,6 @@ class TestFilesystem:
         )
         m.indicators = ind
         run_module(m)
-        assert len(m.results) == 23
-        assert len(m.timeline) == 23
+        assert len(m.results) == 15
+        assert len(m.timeline) == 15
         assert len(m.alertstore.alerts) == 1

tests/test_custom_modules.py

@@ -1,7 +1,3 @@
-import hashlib
-import importlib.metadata
-import json
-
 from click.testing import CliRunner
 
 from mvt.android.cli import check_bugreport
@@ -9,9 +5,7 @@ from mvt.android.cmd_check_androidqf import CmdAndroidCheckAndroidQF
 from mvt.android.cmd_check_backup import CmdAndroidCheckBackup
 from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
 from mvt.android.cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
-from mvt.common 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
 
 
@@ -112,150 +106,6 @@ def test_custom_modules_load_from_environment_without_cli_flag(tmp_path, monkeyp
     assert "EnvBugreportModule" in result.output
 
 
-class InstalledPackageModule(MVTModule):
-    supported_commands = (("ios", "check-backup"),)
-
-
-def get_installed_package_modules():
-    return [InstalledPackageModule]
-
-
-def _fake_entry_points(monkeypatch, value, name="test-modules"):
-    entry_point = importlib.metadata.EntryPoint(
-        name=name, value=value, group=module_loader.MODULES_ENTRY_POINT_GROUP
-    )
-
-    def fake_entry_points(*, group):
-        assert group == module_loader.MODULES_ENTRY_POINT_GROUP
-        return [entry_point]
-
-    monkeypatch.setattr(
-        module_loader.importlib.metadata, "entry_points", fake_entry_points
-    )
-
-
-def test_installed_module_package_loads_from_entry_point(monkeypatch):
-    _fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
-
-    modules = module_loader.load_custom_modules()
-
-    assert modules == [InstalledPackageModule]
-
-
-def test_broken_module_entry_point_is_skipped(monkeypatch, caplog):
-    _fake_entry_points(monkeypatch, "nonexistent_module_xyz:get_modules")
-
-    with caplog.at_level("WARNING"):
-        modules = module_loader.load_custom_modules()
-
-    assert modules == []
-    assert "Unable to load modules from entry point" in caplog.text
-
-
-def test_entry_point_module_deduplicated_against_paths(monkeypatch, tmp_path):
-    _fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
-    module_path = _write_custom_module(
-        tmp_path / "custom.py",
-        "PathLoadedModule",
-        (("ios", "check-backup"),),
-    )
-
-    modules = module_loader.load_custom_modules([str(module_path)])
-
-    assert [module.__name__ for module in modules] == [
-        "InstalledPackageModule",
-        "PathLoadedModule",
-    ]
-
-
-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"),)