Fix iOS update workflow and refresh dependencies

* Speed up compressed sysdiagnose analysis

* ci: retrigger Ruff check

.github/workflows/update-ios-data.yml

@@ -11,10 +11,12 @@ jobs:
   update-ios-version:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
-      - uses: actions/setup-python@v4
+      - uses: actions/checkout@v6
+      - uses: actions/setup-python@v6
+      - name: Install script dependencies
+        run: python -m pip install "packaging==26.3"
       - name: Run script to fetch latest iOS releases from Apple RSS feed.
-        run: python3 .github/workflows/scripts/update-ios-releases.py
+        run: python .github/workflows/scripts/update-ios-releases.py
       - name: Create Pull Request
         uses: peter-evans/create-pull-request@v8
         with:

README.md

@@ -82,6 +82,11 @@ Module-running `check-*` commands can load custom Python modules with
 [development documentation](https://docs.mvt.re/en/latest/development/) for
 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/custom_commands/)
+for the plugin entry points and `--load-command` interface.
+
 
 ## License
 

docs/android/backup.md

@@ -57,3 +57,9 @@ If the backup is encrypted, ABE will prompt you to enter the password.
 Alternatively, [ab-decrypt](https://github.com/joernheissler/ab-decrypt) can be used for that purpose.
 
 You can then extract SMSs with MVT by passing the folder path as parameter instead of the `.ab` file: `mvt-android check-backup --output /path/to/results/ /path/to/backup/` (the path to backup given should be the folder containing the `apps` folder).
+
+When an output folder is specified, URLs extracted from SMS and MMS messages
+are also written to `urls.json`. Each entry contains the URL, its expanded
+destination when MVT resolved a shortened URL during indicator checking, the
+message timestamp, and the `sms` source. The same file is created by
+`check-androidqf` when its nested Android backup contains messages with URLs.

docs/custom_commands.md

@@ -0,0 +1,107 @@
+# Custom CLI Commands
+
+MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
+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.
+
+!!! warning
+
+    Custom commands run as trusted Python code inside the MVT process. Install
+    or load commands only from sources you trust. MVT does not sandbox
+    third-party commands, and the MVT maintainers do not maintain them.
+
+## Install a Command Package
+
+Python packages can register a Click command or group for either MVT CLI. A
+minimal package can expose this command from `my_mvt_plugin.py`:
+
+```python
+import click
+
+
+@click.command()
+@click.argument("path", type=click.Path(exists=True))
+def summarize(path):
+    """Summarize an acquisition."""
+    click.echo(f"Summarizing {path}")
+```
+
+Register the object in the package's `pyproject.toml`. The entry-point name is
+the command users invoke:
+
+```toml
+[project.entry-points."mvt.ios.cli_plugins"]
+summarize = "my_mvt_plugin:summarize"
+
+[project.entry-points."mvt.android.cli_plugins"]
+summarize = "my_mvt_plugin:summarize"
+```
+
+Use only the iOS or Android group if the command is platform-specific. After
+installing the package in the same environment as MVT, it appears directly in
+the appropriate CLI:
+
+```bash
+mvt-ios summarize ./ios-backup
+mvt-android summarize ./androidqf-output
+```
+
+For a `pipx` installation of MVT, inject the plugin into MVT's environment:
+
+```bash
+pipx inject mvt my-mvt-plugin
+```
+
+When MVT is installed in an active virtual environment, install the plugin with
+`pip` in that environment.
+
+## Load a Command File
+
+For local commands that are not packaged, create a Python file that exports one
+Click command or group named `cli`:
+
+```python
+import click
+
+
+@click.command("case-summary")
+@click.argument("path", type=click.Path(exists=True))
+def cli(path):
+    """Summarize a case directory."""
+    click.echo(f"Summarizing {path}")
+```
+
+Pass the file before the custom command name:
+
+```bash
+mvt-ios --load-command ./case_summary.py case-summary ./ios-backup
+```
+
+`--load-command` can be repeated and also accepts a folder. MVT loads
+non-hidden top-level `*.py` files in sorted order and skips `__init__.py`.
+Every loaded file must export one `cli` object.
+
+To load a file or folder on every invocation, set the platform-specific
+environment variable:
+
+```bash
+export MVT_IOS_CUSTOM_COMMANDS=./ios_commands
+export MVT_ANDROID_CUSTOM_COMMANDS=./android_commands
+```
+
+## Naming and Errors
+
+Built-in MVT commands cannot be replaced. External command names must also be
+unique; when installed packages or environment paths collide, MVT keeps the
+first command and logs a warning. A collision from an explicit
+`--load-command` is a usage error.
+
+A package entry point or environment command that cannot be imported appears
+as a marked broken command without preventing other MVT commands from working.
+Invoke that command to see its package or file source and the underlying error.
+An invalid command supplied explicitly with `--load-command` fails immediately
+with a usage error.
+
+Installed command packages use the entry-point name as the CLI command name.
+The entry point must resolve to a `click.Command` or `click.Group`.

docs/ios/records.md

@@ -312,6 +312,15 @@ If indicators are provided through the command-line, they are checked against th
 
 ---
 
+### `urls.json`
+
+This JSON file collects URLs extracted from SMS, iMessage, and WhatsApp
+messages. Each entry contains the original URL, its expanded destination when
+MVT resolved a shortened URL during indicator checking, the message timestamp,
+and its `sms` or `whatsapp` source.
+
+---
+
 ### `sms_attachments.json`
 
 !!! info "Availability"

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"

pyproject.toml

@@ -17,25 +17,25 @@ classifiers = [
     "Programming Language :: Python",
 ]
 dependencies = [
-    "click==8.3.3",
-    "rich==14.3.3",
+    "click==8.4.2",
+    "rich==15.0.0",
     "tld==0.13.2",
-    "requests==2.33.1",
-    "simplejson==3.20.2",
-    "packaging==26.2",
+    "requests==2.34.2",
+    "simplejson==4.1.1",
+    "packaging==26.3",
     "appdirs==1.4.4",
     "iOSbackup==0.9.925",
     "adb-shell[usb]==0.4.4",
-    "libusb1==3.3.1",
-    "cryptography==47.0.0",
+    "libusb1==3.4.0",
+    "cryptography==50.0.0",
     "PyYAML>=6.0.2",
     "pyahocorasick==2.3.1",
-    "betterproto2==0.9.1",
-    "pydantic==2.13.3",
-    "pydantic-settings==2.14.0",
+    "betterproto2==0.10.0",
+    "pydantic==2.13.4",
+    "pydantic-settings==2.15.0",
     "NSKeyedUnArchiver==1.5.2",
     "python-dateutil==2.9.0.post0",
-    "tzdata==2026.2",
+    "tzdata==2026.3",
 ]
 requires-python = ">= 3.10"
 
@@ -65,9 +65,9 @@ dev = [
 docs = [
     "mkdocs==1.6.1",
     "mkdocs-autorefs==1.4.4",
-    "mkdocs-material==9.7.6",
+    "mkdocs-material==9.7.7",
     "mkdocs-material-extensions==1.3.1",
-    "mkdocstrings==1.0.4",
+    "mkdocstrings==1.0.6",
 ]
 
 [build-system]

src/mvt/android/artifacts/tombstone_crashes.py

@@ -34,6 +34,7 @@ TOMBSTONE_TEXT_KEY_MAPPINGS = {
     "signal": "signal_info",
     "code": "code",
     "Cause": "cause",
+    "Abort message": "abort_message",
 }
 
 
@@ -67,6 +68,8 @@ class TombstoneCrashResult(pydantic.BaseModel):
     uid: int
     signal_info: SignalInfo
     cause: Optional[str] = None
+    causes: Optional[List[dict]] = None
+    abort_message: Optional[str] = None
     extra: Optional[str] = None
 
 

src/mvt/android/cli.py

@@ -8,6 +8,12 @@ from zipfile import BadZipFile
 
 import click
 
+from mvt.common.cli_plugins import (
+    ANDROID_CLI_PLUGIN_GROUP,
+    MVT_ANDROID_CUSTOM_COMMANDS_ENV,
+    load_cli_commands_option,
+    register_cli_plugins,
+)
 from mvt.common.cmd_check_iocs import CmdCheckIOCS
 from mvt.common.completion import (
     SUPPORTED_SHELLS,
@@ -82,6 +88,7 @@ def _load_custom_modules(load_module):
 # Main
 # ==============================================================================
 @click.group(invoke_without_command=False)
+@load_cli_commands_option
 @click.option(
     "--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
 )
@@ -505,3 +512,10 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
 def download_indicators():
     ioc_updates = IndicatorsUpdates()
     ioc_updates.update()
+
+
+register_cli_plugins(
+    cli,
+    entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
+    environment_variable=MVT_ANDROID_CUSTOM_COMMANDS_ENV,
+)

src/mvt/android/cmd_check_androidqf.py

@@ -3,6 +3,7 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
+import json
 import logging
 import os
 import shutil
@@ -94,6 +95,47 @@ class CmdAndroidCheckAndroidQF(Command):
             self.__zip = zipfile.ZipFile(self.target_path)
             self.__files = self.__zip.namelist()
 
+        self._load_acquisition_context()
+
+    def _load_acquisition_context(self) -> None:
+        """Pass AndroidQF acquisition metadata to nested commands and modules."""
+        context = {}
+        metadata_files = [
+            file_path
+            for file_path in self.__files
+            if file_path.replace("\\", "/").rsplit("/", 1)[-1] == "acquisition.json"
+        ]
+        for file_path in metadata_files:
+            try:
+                metadata = json.loads(self._get_file_content(file_path))
+                if isinstance(metadata, dict):
+                    context["started"] = metadata.get("started")
+                    context["adb_host_public_key"] = metadata.get("adb_host_public_key")
+                    break
+            except (json.JSONDecodeError, OSError, TypeError, UnicodeDecodeError):
+                self.log.warning(
+                    'Unable to read AndroidQF acquisition metadata "%s"', file_path
+                )
+
+        if not context.get("adb_host_public_key"):
+            key_files = [
+                file_path
+                for file_path in self.__files
+                if file_path.replace("\\", "/").rsplit("/", 1)[-1] == "adb_host_key.pub"
+            ]
+            for file_path in key_files:
+                try:
+                    context["adb_host_public_key"] = self._get_file_content(
+                        file_path
+                    ).decode("utf-8")
+                    break
+                except (OSError, UnicodeDecodeError):
+                    self.log.warning(
+                        'Unable to read AndroidQF ADB host key "%s"', file_path
+                    )
+
+        self.module_options["androidqf_acquisition"] = context
+
     def module_init(self, module: AndroidQFModule) -> None:  # type: ignore[override]
         if self.__format == "zip" and self.__zip:
             module.from_zip(self.__zip, self.__files)
@@ -220,6 +262,7 @@ class CmdAndroidCheckAndroidQF(Command):
             cmd.run()
 
             self.timeline.extend(cmd.timeline)
+            self.url_results.extend(cmd.url_results)
             self.alertstore.extend(cmd.alertstore.alerts)
         finally:
             if bugreport:
@@ -257,6 +300,7 @@ class CmdAndroidCheckAndroidQF(Command):
         cmd.run()
 
         self.timeline.extend(cmd.timeline)
+        self.url_results.extend(cmd.url_results)
         self.alertstore.extend(cmd.alertstore.alerts)
         return True
 
@@ -336,6 +380,7 @@ class CmdAndroidCheckAndroidQF(Command):
             cmd.run()
 
             self.timeline.extend(cmd.timeline)
+            self.url_results.extend(cmd.url_results)
             self.alertstore.extend(cmd.alertstore.alerts)
             return True
 

src/mvt/android/modules/backup/sms.py

@@ -57,6 +57,11 @@ class SMS(BackupModule):
                     ioc_match.message, "", message, matched_indicator=ioc_match.ioc
                 )
 
+    def collect_url_results(self) -> None:
+        for message in self.results:
+            for url in message.get("links", []):
+                self.add_url_result(url, message.get("isodate"), "sms")
+
     def run(self) -> None:
         sms_path = "apps/com.android.providers.telephony/d_f/*_sms_backup"
         for file in self._get_files_by_pattern(sms_path):

src/mvt/android/modules/bugreport/dumpsys_adb_state.py

@@ -3,11 +3,15 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
+import base64
+import binascii
+import datetime
 import logging
 from typing import Optional
 
 from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
 from mvt.common.module_types import ModuleResults
+from mvt.common.utils import convert_datetime_to_iso
 
 from .base import BugReportModule
 
@@ -53,3 +57,122 @@ class DumpsysADBState(DumpsysADBArtifact, BugReportModule):
                 "Identified a total of %d trusted ADB keys",
                 len(self.results[0].get("user_keys", [])),
             )
+
+    @staticmethod
+    def _key_material(public_key: object) -> str:
+        if isinstance(public_key, bytes):
+            public_key = public_key.decode("utf-8", errors="replace")
+        if not isinstance(public_key, str):
+            return ""
+        return public_key.strip().split(" ", 1)[0]
+
+    @staticmethod
+    def _is_valid_key(public_key: str) -> bool:
+        if not public_key:
+            return False
+        try:
+            return bool(base64.b64decode(public_key, validate=True))
+        except (binascii.Error, ValueError):
+            return False
+
+    @staticmethod
+    def _parse_acquisition_time(value: object) -> Optional[datetime.datetime]:
+        if not isinstance(value, str):
+            return None
+        try:
+            timestamp = datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
+        except ValueError:
+            return None
+        if not timestamp.tzinfo:
+            timestamp = timestamp.replace(tzinfo=datetime.timezone.utc)
+        return timestamp.astimezone(datetime.timezone.utc)
+
+    @staticmethod
+    def _parse_last_connected(value: object) -> Optional[datetime.datetime]:
+        try:
+            return datetime.datetime.fromtimestamp(
+                int(str(value)) / 1000,
+                tz=datetime.timezone.utc,
+            )
+        except (OSError, OverflowError, TypeError, ValueError):
+            return None
+
+    def _trusted_keys(self) -> list[dict]:
+        """Return unique trusted keys, preferring keystore connection metadata."""
+        trusted_keys = []
+        seen = set()
+        for result in self.results:
+            keystore = result.get("keystore", [])
+            candidates = keystore if isinstance(keystore, list) else []
+            candidates = [*candidates, *result.get("user_keys", [])]
+            for candidate in candidates:
+                if not isinstance(candidate, dict):
+                    continue
+                key = self._key_material(candidate.get("key"))
+                identity = key or repr(candidate)
+                if identity in seen:
+                    continue
+                seen.add(identity)
+                trusted_keys.append(candidate)
+        return trusted_keys
+
+    def check_indicators(self) -> None:
+        if "androidqf_acquisition" not in self.module_options:
+            return super().check_indicators()
+
+        context = self.module_options.get("androidqf_acquisition")
+        if not isinstance(context, dict):
+            context = {}
+        acquisition_key = self._key_material(context.get("adb_host_public_key"))
+        if acquisition_key and not self._is_valid_key(acquisition_key):
+            acquisition_key = ""
+        acquisition_time = self._parse_acquisition_time(context.get("started"))
+        cutoff = (
+            acquisition_time - datetime.timedelta(days=1) if acquisition_time else None
+        )
+
+        for trusted_key in self._trusted_keys():
+            key = self._key_material(trusted_key.get("key"))
+            fingerprint = trusted_key.get("fingerprint") or "<unknown key>"
+            user = trusted_key.get("user") or "unknown user"
+            description = f"{fingerprint} ({user})"
+            last_connected = self._parse_last_connected(
+                trusted_key.get("last_connected")
+            )
+            event_time = (
+                convert_datetime_to_iso(last_connected) if last_connected else ""
+            )
+
+            if not self._is_valid_key(key):
+                self.alertstore.low(
+                    f"Found an invalid trusted ADB host key: {description}",
+                    event_time,
+                    trusted_key,
+                )
+                continue
+
+            if not acquisition_key:
+                self.alertstore.low(
+                    "Found a trusted ADB host key, but the AndroidQF acquisition "
+                    f"does not include its host key: {description}",
+                    event_time,
+                    trusted_key,
+                )
+                continue
+
+            if key != acquisition_key:
+                self.alertstore.low(
+                    "Found a trusted ADB host key different from the AndroidQF "
+                    f"acquisition host: {description}",
+                    event_time,
+                    trusted_key,
+                )
+                continue
+
+            if cutoff and last_connected and last_connected <= cutoff:
+                self.alertstore.info(
+                    "Found a trusted ADB host key last connected at least one day "
+                    f"before the AndroidQF acquisition: {description}",
+                    event_time,
+                    trusted_key,
+                )

src/mvt/common/cli_plugins.py

@@ -0,0 +1,297 @@
+# 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 hashlib
+import importlib.metadata
+import importlib.util
+import logging
+import os
+import sys
+from pathlib import Path
+from types import ModuleType
+from typing import Iterable
+
+import click
+
+IOS_CLI_PLUGIN_GROUP = "mvt.ios.cli_plugins"
+ANDROID_CLI_PLUGIN_GROUP = "mvt.android.cli_plugins"
+MVT_IOS_CUSTOM_COMMANDS_ENV = "MVT_IOS_CUSTOM_COMMANDS"
+MVT_ANDROID_CUSTOM_COMMANDS_ENV = "MVT_ANDROID_CUSTOM_COMMANDS"
+
+log = logging.getLogger(__name__)
+
+
+class CustomCommandLoadError(Exception):
+    pass
+
+
+class BrokenPluginCommand(click.Command):
+    """A placeholder for an installed or configured command that failed to load."""
+
+    def __init__(self, name: str, source: str, exception: BaseException):
+        super().__init__(
+            name,
+            help=(
+                f"Unable to load external command from {source}.\n\n"
+                f"{type(exception).__name__}: {exception}"
+            ),
+            short_help="Warning: external command could not be loaded.",
+        )
+        self.source = source
+        self.exception = exception
+
+    def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
+        return args
+
+    def invoke(self, ctx: click.Context) -> None:
+        raise click.ClickException(
+            f"Unable to load external command '{self.name}' from {self.source}: "
+            f"{type(self.exception).__name__}: {self.exception}"
+        )
+
+
+def _module_name_for_path(path: Path) -> str:
+    digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
+    return f"_mvt_custom_command_{path.stem}_{digest}"
+
+
+def _iter_command_files(path: Path) -> Iterable[Path]:
+    if path.is_file():
+        if path.suffix != ".py":
+            raise CustomCommandLoadError(
+                f"Custom command file is not a Python file: {path}"
+            )
+        yield path
+        return
+
+    if path.is_dir():
+        for child in sorted(path.iterdir()):
+            if child.name.startswith(".") or child.name == "__init__.py":
+                continue
+            if child.is_file() and child.suffix == ".py":
+                yield child
+        return
+
+    raise CustomCommandLoadError(f"Custom command path does not exist: {path}")
+
+
+def _load_python_file(path: Path) -> ModuleType:
+    module_name = _module_name_for_path(path)
+    spec = importlib.util.spec_from_file_location(module_name, path)
+    if spec is None or spec.loader is None:
+        raise CustomCommandLoadError(f"Unable to load custom command file: {path}")
+
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    try:
+        spec.loader.exec_module(module)
+    except (Exception, SystemExit) as exc:
+        raise CustomCommandLoadError(
+            f"Unable to import custom command {path}: {exc}"
+        ) from exc
+
+    return module
+
+
+def load_cli_command_file(path: Path) -> click.Command:
+    module = _load_python_file(path)
+    command = getattr(module, "cli", None)
+    if not isinstance(command, click.Command):
+        raise CustomCommandLoadError(
+            f"Custom command {path} must export a Click command or group named 'cli'"
+        )
+    if command.name is None:
+        raise CustomCommandLoadError(f"Custom command {path} has no command name")
+    return command
+
+
+def _register_command(
+    group: click.Group,
+    command: click.Command,
+    *,
+    name: str,
+    source: str,
+    collision_is_error: bool,
+) -> bool:
+    if name in group.commands:
+        registered_sources = getattr(group, "_mvt_external_command_sources", {})
+        if registered_sources.get(name) == source:
+            return False
+        message = (
+            f"Unable to register external command '{name}' from {source}: "
+            "the command name is already registered"
+        )
+        if collision_is_error:
+            raise CustomCommandLoadError(message)
+        log.warning(message)
+        return False
+
+    group.add_command(command, name=name)
+    registered_sources = getattr(group, "_mvt_external_command_sources", {})
+    registered_sources[name] = source
+    setattr(group, "_mvt_external_command_sources", registered_sources)
+    return True
+
+
+def register_cli_commands_from_path(
+    group: click.Group,
+    path: str | Path,
+    *,
+    collision_is_error: bool = False,
+    failures_are_errors: bool = False,
+) -> list[str]:
+    resolved_path = Path(path).expanduser().resolve()
+    registered: list[str] = []
+
+    try:
+        command_files = list(_iter_command_files(resolved_path))
+    except CustomCommandLoadError as exc:
+        if failures_are_errors:
+            raise
+        name = resolved_path.stem.replace("_", "-") or "custom-command"
+        if _register_command(
+            group,
+            BrokenPluginCommand(name, str(resolved_path), exc),
+            name=name,
+            source=str(resolved_path),
+            collision_is_error=collision_is_error,
+        ):
+            registered.append(name)
+        return registered
+
+    for command_file in command_files:
+        try:
+            command = load_cli_command_file(command_file)
+        except CustomCommandLoadError as exc:
+            if failures_are_errors:
+                raise
+            name = command_file.stem.replace("_", "-")
+            command = BrokenPluginCommand(name, str(command_file), exc)
+
+        command_name = command.name
+        if command_name is None:
+            raise CustomCommandLoadError(
+                f"Custom command {command_file} has no command name"
+            )
+        if _register_command(
+            group,
+            command,
+            name=command_name,
+            source=str(command_file),
+            collision_is_error=collision_is_error,
+        ):
+            registered.append(command_name)
+
+    return registered
+
+
+def _entry_point_source(entry_point: importlib.metadata.EntryPoint) -> str:
+    try:
+        distribution = getattr(entry_point, "dist", None)
+        if distribution is None:
+            return f"entry point {entry_point.value}"
+
+        name = distribution.metadata.get("Name", "unknown distribution")
+        version = getattr(distribution, "version", None)
+        if version:
+            return f"{name} {version} ({entry_point.value})"
+        return f"{name} ({entry_point.value})"
+    except Exception:
+        return f"entry point {entry_point.value}"
+
+
+def register_installed_cli_commands(
+    group: click.Group,
+    entry_point_group: str,
+) -> list[str]:
+    try:
+        entry_points = importlib.metadata.entry_points(group=entry_point_group)
+    except Exception as exc:
+        log.warning(
+            "Unable to discover external commands in entry-point group %s: %s",
+            entry_point_group,
+            exc,
+        )
+        return []
+    ordered_entry_points = sorted(
+        entry_points,
+        key=lambda entry_point: (
+            entry_point.name,
+            _entry_point_source(entry_point),
+            entry_point.value,
+        ),
+    )
+    registered: list[str] = []
+
+    for entry_point in ordered_entry_points:
+        source = _entry_point_source(entry_point)
+        try:
+            command = entry_point.load()
+            if not isinstance(command, click.Command):
+                raise TypeError(
+                    f"entry point must resolve to a Click command or group, "
+                    f"not {type(command).__name__}"
+                )
+        except (Exception, SystemExit) as exc:
+            command = BrokenPluginCommand(entry_point.name, source, exc)
+
+        if _register_command(
+            group,
+            command,
+            name=entry_point.name,
+            source=source,
+            collision_is_error=False,
+        ):
+            registered.append(entry_point.name)
+
+    return registered
+
+
+def register_cli_plugins(
+    group: click.Group,
+    *,
+    entry_point_group: str,
+    environment_variable: str,
+) -> None:
+    environment_path = os.environ.get(environment_variable)
+    if environment_path:
+        register_cli_commands_from_path(group, environment_path)
+    register_installed_cli_commands(group, entry_point_group)
+
+
+def _load_command_option_callback(
+    ctx: click.Context,
+    param: click.Parameter,
+    paths: tuple[str, ...],
+) -> tuple[str, ...]:
+    if not isinstance(ctx.command, click.Group):
+        raise click.ClickException("--load-command requires a Click command group")
+
+    for path in paths:
+        try:
+            register_cli_commands_from_path(
+                ctx.command,
+                path,
+                collision_is_error=True,
+                failures_are_errors=True,
+            )
+        except CustomCommandLoadError as exc:
+            raise click.BadParameter(str(exc), ctx=ctx, param=param) from exc
+    return paths
+
+
+load_cli_commands_option = click.option(
+    "--load-command",
+    "load_commands",
+    type=click.Path(exists=True),
+    multiple=True,
+    expose_value=False,
+    is_eager=True,
+    callback=_load_command_option_callback,
+    help=(
+        "Load a custom CLI command from a Python file or folder "
+        "(can be invoked multiple times)"
+    ),
+)

src/mvt/common/command.py

@@ -20,7 +20,7 @@ from .config import settings
 from .indicators import Indicators
 from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
 from .module_loader import module_supports_command
-from .module_types import ModuleTimeline
+from .module_types import ModuleTimeline, URLResult
 from .utils import (
     CustomJSONEncoder,
     convert_datetime_to_iso,
@@ -73,6 +73,7 @@ class Command:
         self.hashes = hashes
         self.hash_values: list[dict[str, Any]] = []
         self.timeline: ModuleTimeline = []
+        self.url_results: list[URLResult] = []
 
         # Load IOCs
         self._create_storage()
@@ -150,6 +151,14 @@ class Command:
         with open(alerts_path, "w+", encoding="utf-8") as handle:
             json.dump(alerts, handle, indent=4, cls=CustomJSONEncoder)
 
+    def _store_urls(self) -> None:
+        if not self.results_path or not self.url_results:
+            return
+
+        urls_path = os.path.join(self.results_path, "urls.json")
+        with open(urls_path, "w", encoding="utf-8") as handle:
+            json.dump(self.url_results, handle, indent=4, cls=CustomJSONEncoder)
+
     def _store_alerts_timeline(self) -> None:
         if not self.results_path:
             return
@@ -396,6 +405,7 @@ class Command:
             self.executed.append(m)
             executed_by_type[module] = m
             self.timeline.extend(m.timeline)
+            self.url_results.extend(m.url_results)
             self.alertstore.extend(m.alertstore.alerts)
 
         try:
@@ -410,4 +420,5 @@ class Command:
         self._store_timeline()
         self._store_alerts_timeline()
         self._store_alerts()
+        self._store_urls()
         self._store_info()

src/mvt/common/indicators.py

@@ -49,6 +49,7 @@ class Indicators:
         self.log = log
         self.ioc_collections: List[Dict[str, Any]] = []
         self.total_ioc_count = 0
+        self.resolved_urls: Dict[str, str] = {}
 
     def _load_downloaded_indicators(self) -> None:
         if not os.path.isdir(MVT_INDICATORS_FOLDER):
@@ -439,9 +440,14 @@ class Indicators:
                         orig_url.url,
                         dest_url.url,
                     )
-                    return self.check_url(dest_url.url)
+                    match = self.check_url(dest_url.url)
+                    self.resolved_urls[url] = self.resolved_urls.get(
+                        dest_url.url, dest_url.url
+                    )
+                    return match
 
                 final_url = dest_url
+                self.resolved_urls[url] = final_url.url
             else:
                 # If it's not shortened, we just use the original URL object.
                 final_url = orig_url
@@ -482,6 +488,13 @@ class Indicators:
 
         return None
 
+    def get_expanded_url(self, url: str) -> Optional[str]:
+        """Return the final URL recorded while checking a shortened URL."""
+        expanded_url = self.resolved_urls.get(url)
+        if expanded_url and expanded_url != url:
+            return expanded_url
+        return None
+
     def check_urls(self, urls: list) -> Optional[IndicatorMatch]:
         """Check a list of URLs against the provided list of domain indicators.
 

src/mvt/common/module.py

@@ -18,6 +18,7 @@ from .module_types import (
     ModuleResults,
     ModuleSerializedResult,
     ModuleTimeline,
+    URLResult,
 )
 from .utils import CustomJSONEncoder, exec_or_profile
 
@@ -82,6 +83,7 @@ class MVTModule:
 
         self.results: ModuleResults = results if results is not None else []
         self.timeline: ModuleTimeline = []
+        self.url_results: list[URLResult] = []
         self.dependency_modules: Dict[type["MVTModule"], "MVTModule"] = {}
 
     def get_dependency_results(
@@ -110,6 +112,23 @@ class MVTModule:
     def check_indicators(self) -> None:
         raise NotImplementedError
 
+    def collect_url_results(self) -> None:
+        """Collect URL records exposed by this module."""
+
+    def add_url_result(self, url: str, timestamp: Optional[str], source: str) -> None:
+        expanded_url = None
+        if self.indicators:
+            expanded_url = self.indicators.get_expanded_url(url)
+
+        self.url_results.append(
+            {
+                "url": url,
+                "expanded_url": expanded_url,
+                "timestamp": timestamp,
+                "source": source,
+            }
+        )
+
     def save_to_json(self) -> None:
         if not self.results_path:
             return
@@ -249,6 +268,15 @@ def run_module(module: MVTModule) -> None:
                     "The %s module produced no detections!", module.__class__.__name__
                 )
 
+        try:
+            module.collect_url_results()
+        except Exception as exc:
+            module.log.exception(
+                "Error when collecting URLs from module %s: %s",
+                module.__class__.__name__,
+                exc,
+            )
+
         try:
             module.to_timeline()
         except NotImplementedError:

src/mvt/common/module_types.py

@@ -4,7 +4,7 @@
 #   https://license.mvt.re/1.1/
 
 from dataclasses import dataclass
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, Optional, TypedDict, Union
 
 
 # ModuleAtomicResult is a flexible dictionary that can contain any data.
@@ -22,6 +22,13 @@ ModuleAtomicResult = Dict[str, Any]
 ModuleResults = Any
 
 
+class URLResult(TypedDict):
+    url: str
+    expanded_url: Optional[str]
+    timestamp: Optional[str]
+    source: str
+
+
 @dataclass
 class ModuleAtomicTimeline:
     timestamp: str

src/mvt/ios/cli.py

@@ -8,6 +8,12 @@ import logging
 import os
 
 import click
+from mvt.common.cli_plugins import (
+    IOS_CLI_PLUGIN_GROUP,
+    MVT_IOS_CUSTOM_COMMANDS_ENV,
+    load_cli_commands_option,
+    register_cli_plugins,
+)
 from mvt.common.cmd_check_iocs import CmdCheckIOCS
 from mvt.common.completion import (
     SUPPORTED_SHELLS,
@@ -86,6 +92,7 @@ def _load_custom_modules(load_module):
 # Main
 # ==============================================================================
 @click.group(invoke_without_command=False)
+@load_cli_commands_option
 @click.option(
     "--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
 )
@@ -523,3 +530,10 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
 def download_iocs():
     ioc_updates = IndicatorsUpdates()
     ioc_updates.update()
+
+
+register_cli_plugins(
+    cli,
+    entry_point_group=IOS_CLI_PLUGIN_GROUP,
+    environment_variable=MVT_IOS_CUSTOM_COMMANDS_ENV,
+)

src/mvt/ios/cmd_check_sysdiagnose.py

@@ -6,8 +6,10 @@
 import json
 import logging
 import os
+import shutil
 import tarfile
-from pathlib import Path
+from pathlib import Path, PurePosixPath
+from tempfile import TemporaryDirectory
 from typing import Any, Optional
 
 from mvt.common.command import Command
@@ -54,6 +56,8 @@ class CmdIOSCheckSysdiagnose(Command):
         self.sysdiagnose_archive: Optional[tarfile.TarFile] = None
         self.sysdiagnose_files: list[str] = []
         self.ips_files: list[dict[str, Any]] = []
+        self.temp_sysdiagnose_dir: Optional[TemporaryDirectory[str]] = None
+        self.extracted_sysdiagnose_path: Optional[str] = None
 
     @staticmethod
     def _parse_bugtype_header(data: bytes) -> Optional[int]:
@@ -91,21 +95,68 @@ class CmdIOSCheckSysdiagnose(Command):
         self.log.info("Parsing sysdiagnose archive. This might take a while...")
         self.sysdiagnose_format = "tar"
         self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz")
-        for member in self.sysdiagnose_archive:
-            self.sysdiagnose_files.append(member.name)
-            if member.isfile() and member.name.endswith(".ips"):
-                archive_handle = self.sysdiagnose_archive.extractfile(member)
-                if archive_handle is not None:
-                    with archive_handle:
-                        self._add_ips_file(member.name, archive_handle.read())
+        self._extract_sysdiagnose_archive()
+
+    def _extract_sysdiagnose_archive(self) -> None:
+        archive = self.sysdiagnose_archive
+        if archive is None:
+            raise RuntimeError("Sysdiagnose archive has not been initialized")
+
+        self.temp_sysdiagnose_dir = TemporaryDirectory()
+        extraction_root = Path(self.temp_sysdiagnose_dir.name).resolve()
+        archive_roots = set()
+
+        for member in archive:
+            member_path = PurePosixPath(member.name.replace("\\", "/"))
+            if member_path.is_absolute() or ".." in member_path.parts:
+                self.log.warning("Skipping unsafe sysdiagnose path %r", member.name)
+                continue
+
+            destination = extraction_root.joinpath(*member_path.parts).resolve()
+            if not destination.is_relative_to(extraction_root):
+                self.log.warning("Skipping unsafe sysdiagnose path %r", member.name)
+                continue
+
+            if not member_path.parts:
+                continue
+            archive_roots.add(member_path.parts[0])
+
+            if member.isdir():
+                destination.mkdir(parents=True, exist_ok=True)
+                continue
+
+            # Modules only need directories and regular files. Do not materialize
+            # links or device nodes from an untrusted sysdiagnose archive.
+            if not member.isfile():
+                self.log.warning("Skipping unsafe sysdiagnose member %r", member.name)
+                continue
+
+            normalized_name = member_path.as_posix()
+            self.sysdiagnose_files.append(normalized_name)
+
+            source = archive.extractfile(member)
+            if source is None:
+                continue
+
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            with source, destination.open("wb") as output:
+                shutil.copyfileobj(source, output)
+
+            if normalized_name.endswith(".ips"):
+                self._add_ips_file(str(destination), destination.read_bytes())
+
+        if len(archive_roots) != 1:
+            raise ValueError("Sysdiagnose archive must contain one top-level directory")
+
+        self.extracted_sysdiagnose_path = str(extraction_root / archive_roots.pop())
 
     def module_init(self, module) -> None:
         module.ips_files = self.ips_files
         if self.sysdiagnose_format == "tar":
-            if self.sysdiagnose_archive is None:
-                raise RuntimeError("Sysdiagnose archive has not been initialized")
-            module.from_sysdiagnose_tar(
-                self.sysdiagnose_archive, self.sysdiagnose_files
+            if self.extracted_sysdiagnose_path is None:
+                raise RuntimeError("Sysdiagnose archive has not been extracted")
+            module.from_sysdiagnose_folder(
+                self.extracted_sysdiagnose_path, self.sysdiagnose_files
             )
             return
         if self.sysdiagnose_format == "dir" and self.target_path:
@@ -117,3 +168,6 @@ class CmdIOSCheckSysdiagnose(Command):
         if self.sysdiagnose_archive is not None:
             self.sysdiagnose_archive.close()
             self.sysdiagnose_archive = None
+        if self.temp_sysdiagnose_dir is not None:
+            self.temp_sysdiagnose_dir.cleanup()
+            self.temp_sysdiagnose_dir = None

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

@@ -73,7 +73,9 @@ class Calls(IOSExtraction):
                     "duration": row[1],
                     "location": row[2],
                     "number": (
-                        row[3].decode("utf-8") if row[3] and row[3] is bytes else row[3]
+                        row[3].decode("utf-8")
+                        if isinstance(row[3], bytes)
+                        else row[3]
                     ),
                     "provider": row[4],
                 }

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

@@ -101,6 +101,11 @@ class SMS(IOSExtraction):
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc
                 )
 
+    def collect_url_results(self) -> None:
+        for message in self.results:
+            for url in message.get("links", []):
+                self.add_url_result(url, message.get("isodate"), "sms")
+
     def run(self) -> None:
         self._find_ios_database(backup_ids=SMS_BACKUP_IDS, root_paths=SMS_ROOT_PATHS)
         self.log.info("Found SMS database at path: %s", self.file_path)

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

@@ -70,6 +70,11 @@ class Whatsapp(IOSExtraction):
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc
                 )
 
+    def collect_url_results(self) -> None:
+        for message in self.results:
+            for url in message.get("links", []):
+                self.add_url_result(url, message.get("isodate"), "whatsapp")
+
     def run(self) -> None:
         self._find_ios_database(
             backup_ids=WHATSAPP_BACKUP_IDS, root_paths=WHATSAPP_ROOT_PATHS

tests/android/test_artifact_dumpsys_adb.py

@@ -4,6 +4,8 @@
 #   https://license.mvt.re/1.1/
 
 from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
+from mvt.android.modules.bugreport.dumpsys_adb_state import DumpsysADBState
+from mvt.common.alerts import AlertLevel
 
 from ..utils import get_artifact
 
@@ -114,3 +116,119 @@ class TestDumpsysADBArtifact:
         assert key_store_entry["user"] == "user@laptop"
         assert key_store_entry["fingerprint"] == expected_fingerprint
         assert key_store_entry["last_connected"] == "1628501829898"
+
+
+class TestDumpsysADBStateAlerts:
+    def test_no_androidqf_context_preserves_existing_behavior(self):
+        module = DumpsysADBState(
+            results=[
+                {
+                    "user_keys": [
+                        {
+                            "key": b"QUJDRA==",
+                            "user": "host@example",
+                            "fingerprint": "fingerprint",
+                        }
+                    ]
+                }
+            ]
+        )
+
+        module.check_indicators()
+
+        assert module.alertstore.alerts == []
+
+    def test_androidqf_trusted_keys_create_expected_alerts(self):
+        module = DumpsysADBState(
+            module_options={
+                "androidqf_acquisition": {
+                    "started": "2025-06-20T18:00:00Z",
+                    "adb_host_public_key": "QUJDRA== acquisition@host",
+                }
+            },
+            results=[
+                {
+                    "user_keys": [
+                        {
+                            "key": b"QUJDRA==",
+                            "user": "acquisition@host",
+                            "fingerprint": "acquisition-fingerprint",
+                        },
+                        {
+                            "key": b"RUZHSA==",
+                            "user": "other@host",
+                            "fingerprint": "other-fingerprint",
+                        },
+                        {
+                            "key": b"not-base64",
+                            "user": "invalid@host",
+                            "fingerprint": "",
+                        },
+                    ],
+                    "keystore": [
+                        {
+                            "key": b"QUJDRA==",
+                            "user": "acquisition@host",
+                            "fingerprint": "acquisition-fingerprint",
+                            "last_connected": "1750266000000",
+                        }
+                    ],
+                }
+            ],
+        )
+
+        module.check_indicators()
+
+        assert [alert.level for alert in module.alertstore.alerts] == [
+            AlertLevel.INFORMATIONAL,
+            AlertLevel.LOW,
+            AlertLevel.LOW,
+        ]
+        informational, different, invalid = module.alertstore.alerts
+        assert "at least one day before" in informational.message
+        assert informational.event_time == "2025-06-18 17:00:00.000000"
+        assert "different from the AndroidQF acquisition host" in different.message
+        assert "invalid trusted ADB host key" in invalid.message
+
+    def test_missing_androidqf_host_key_creates_low_alert(self):
+        trusted_key = {
+            "key": b"QUJDRA==",
+            "user": "host@example",
+            "fingerprint": "fingerprint",
+        }
+        module = DumpsysADBState(
+            module_options={"androidqf_acquisition": {}},
+            results=[{"user_keys": [trusted_key]}],
+        )
+
+        module.check_indicators()
+
+        assert len(module.alertstore.alerts) == 1
+        assert module.alertstore.alerts[0].level == AlertLevel.LOW
+        assert "does not include its host key" in module.alertstore.alerts[0].message
+
+    def test_recent_acquisition_host_key_does_not_create_alert(self):
+        module = DumpsysADBState(
+            module_options={
+                "androidqf_acquisition": {
+                    "started": "2025-06-20T18:00:00Z",
+                    "adb_host_public_key": "QUJDRA== acquisition@host",
+                }
+            },
+            results=[
+                {
+                    "keystore": [
+                        {
+                            "key": b"QUJDRA==",
+                            "user": "acquisition@host",
+                            "fingerprint": "fingerprint",
+                            "last_connected": "1750438800000",
+                        }
+                    ]
+                }
+            ],
+        )
+
+        module.check_indicators()
+
+        assert module.alertstore.alerts == []

tests/android/test_artifact_tombstones.py

@@ -8,6 +8,7 @@ import datetime
 import pytest
 
 from mvt.android.artifacts.tombstone_crashes import TombstoneCrashArtifact
+from mvt.android.parsers.proto.tombstone import Tombstone
 
 from ..utils import get_artifact
 
@@ -42,6 +43,47 @@ class TestTombstoneCrashArtifact:
         assert len(tombstone_artifact.results) == 1
         self.validate_tombstone_result(tombstone_artifact.results[0])
 
+    def test_text_tombstone_preserves_abort_message(self):
+        tombstone_artifact = TombstoneCrashArtifact()
+        artifact_path = "android_data/bugreport/FS/data/tombstones/tombstone_00"
+        file = get_artifact(artifact_path)
+        with open(file, "rb") as f:
+            data = f.read()
+
+        tombstone_artifact.parse(
+            os.path.basename(artifact_path),
+            datetime.datetime(2021, 9, 29, 17, 43, 49),
+            data,
+        )
+
+        assert tombstone_artifact.results[0]["abort_message"] == (
+            "Check failed: payload.size() <= bytes_left "
+            "(payload.size()=99, bytes_left=51) "
+        )
+
+    def test_protobuf_tombstone_preserves_abort_message_and_causes(self):
+        tombstone_artifact = TombstoneCrashArtifact()
+        artifact_path = "android_data/tombstone_process.pb"
+        file = get_artifact(artifact_path)
+        with open(file, "rb") as f:
+            tombstone = Tombstone().parse(f.read())
+
+        tombstone.abort_message = "synthetic abort reason"
+        tombstone_artifact.parse_protobuf(
+            os.path.basename(artifact_path),
+            datetime.datetime(2023, 4, 12, 12, 32, 40, 518290),
+            bytes(tombstone),
+        )
+
+        result = tombstone_artifact.results[0]
+        assert result["abort_message"] == "synthetic abort reason"
+        assert result["causes"] == [
+            {
+                "human_readable": "null pointer dereference",
+                "memory_error": None,
+            }
+        ]
+
     def test_text_tombstone_keeps_crashing_thread(self):
         tombstone_artifact = TombstoneCrashArtifact()
         artifact_path = "android_data/tombstone_process.txt"

tests/common/test_cli_plugins.py

@@ -0,0 +1,371 @@
+from types import SimpleNamespace
+
+import click
+from click.testing import CliRunner
+
+from mvt.common.cli_plugins import (
+    ANDROID_CLI_PLUGIN_GROUP,
+    IOS_CLI_PLUGIN_GROUP,
+    BrokenPluginCommand,
+    load_cli_commands_option,
+    register_cli_commands_from_path,
+    register_cli_plugins,
+    register_installed_cli_commands,
+)
+
+
+COMMAND_TEMPLATE = """
+import click
+
+
+@click.command({name!r})
+@click.pass_context
+def cli(ctx):
+    click.echo({message!r})
+    if ctx.obj:
+        click.echo(ctx.obj.get("marker", ""))
+"""
+
+
+def _write_command(path, name, message="command ran"):
+    path.write_text(
+        COMMAND_TEMPLATE.format(name=name, message=message),
+        encoding="utf-8",
+    )
+    return path
+
+
+def _make_group():
+    @click.group()
+    @load_cli_commands_option
+    @click.pass_context
+    def group(ctx):
+        ctx.ensure_object(dict)
+        ctx.obj["marker"] = "parent context"
+
+    return group
+
+
+def _entry_point(name, value, command=None, exception=None, distribution="plugin"):
+    def load():
+        if exception is not None:
+            raise exception
+        return command
+
+    dist = SimpleNamespace(metadata={"Name": distribution}, version="1.0")
+    return SimpleNamespace(name=name, value=value, load=load, dist=dist)
+
+
+def test_load_command_option_registers_command_before_resolution(tmp_path):
+    command_path = _write_command(tmp_path / "hello.py", "hello")
+    group = _make_group()
+
+    result = CliRunner().invoke(
+        group,
+        ["--load-command", str(command_path), "hello"],
+    )
+
+    assert result.exit_code == 0
+    assert "command ran" in result.output
+    assert "parent context" in result.output
+
+
+def test_load_command_option_supports_folders_and_repeated_paths(tmp_path):
+    folder = tmp_path / "commands"
+    folder.mkdir()
+    _write_command(folder / "b.py", "second")
+    _write_command(folder / "a.py", "first")
+    _write_command(folder / ".hidden.py", "hidden")
+    _write_command(folder / "__init__.py", "init")
+    other = _write_command(tmp_path / "third.py", "third")
+    group = _make_group()
+
+    result = CliRunner().invoke(
+        group,
+        [
+            "--load-command",
+            str(folder),
+            "--load-command",
+            str(other),
+            "--load-command",
+            str(other),
+            "--help",
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert "first" in result.output
+    assert "second" in result.output
+    assert "third" in result.output
+    assert "hidden" not in result.output
+    assert "init" not in result.output
+
+
+def test_loaded_command_participates_in_shell_completion(tmp_path):
+    command_path = _write_command(tmp_path / "hello.py", "hello")
+    group = _make_group()
+    words = f"group --load-command {command_path} he"
+
+    result = CliRunner().invoke(
+        group,
+        [],
+        env={
+            "_GROUP_COMPLETE": "bash_complete",
+            "COMP_WORDS": words,
+            "COMP_CWORD": "3",
+        },
+    )
+
+    assert result.exit_code == 0
+    assert "plain,hello" in result.output
+
+
+def test_explicit_command_import_and_contract_failures_are_usage_errors(tmp_path):
+    broken_path = tmp_path / "broken.py"
+    broken_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
+    missing_cli_path = tmp_path / "missing_cli.py"
+    missing_cli_path.write_text("value = 1", encoding="utf-8")
+
+    broken_result = CliRunner().invoke(
+        _make_group(),
+        ["--load-command", str(broken_path), "broken"],
+    )
+    missing_cli_result = CliRunner().invoke(
+        _make_group(),
+        ["--load-command", str(missing_cli_path), "missing-cli"],
+    )
+
+    assert broken_result.exit_code == 2
+    assert "broken import" in broken_result.output
+    assert missing_cli_result.exit_code == 2
+    assert "must export a Click command or group named 'cli'" in (
+        missing_cli_result.output
+    )
+
+
+def test_environment_command_failure_gets_broken_placeholder(tmp_path):
+    command_path = tmp_path / "broken_command.py"
+    command_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
+    group = click.Group()
+
+    registered = register_cli_commands_from_path(group, command_path)
+
+    assert registered == ["broken-command"]
+    assert isinstance(group.commands["broken-command"], BrokenPluginCommand)
+    result = CliRunner().invoke(group, ["broken-command"])
+    assert result.exit_code == 1
+    assert "broken import" in result.output
+    assert str(command_path) in result.output
+
+
+def test_environment_command_system_exit_gets_broken_placeholder(tmp_path):
+    command_path = tmp_path / "exiting_command.py"
+    command_path.write_text("raise SystemExit(7)", encoding="utf-8")
+    group = click.Group()
+
+    registered = register_cli_commands_from_path(group, command_path)
+
+    assert registered == ["exiting-command"]
+    assert isinstance(group.commands["exiting-command"], BrokenPluginCommand)
+    result = CliRunner().invoke(group, ["exiting-command"])
+    assert result.exit_code == 1
+    assert "Unable to import custom command" in result.output
+    assert result.output.rstrip().endswith(": 7")
+
+
+def test_installed_entry_point_name_is_the_command_name(monkeypatch):
+    @click.command("internal-name")
+    def command():
+        click.echo("installed command ran")
+
+    entry_point = _entry_point(
+        "external-name",
+        "example_plugin:cli",
+        command=command,
+    )
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        lambda **kwargs: [entry_point],
+    )
+    group = click.Group()
+
+    registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
+
+    assert registered == ["external-name"]
+    assert "internal-name" not in group.commands
+    result = CliRunner().invoke(group, ["external-name"])
+    assert result.exit_code == 0
+    assert result.output == "installed command ran\n"
+
+
+def test_broken_installed_plugin_does_not_break_cli(monkeypatch):
+    broken = _entry_point(
+        "broken",
+        "broken_plugin:cli",
+        exception=RuntimeError("missing dependency"),
+        distribution="broken-plugin",
+    )
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        lambda **kwargs: [broken],
+    )
+    group = click.Group()
+
+    register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
+
+    help_result = CliRunner().invoke(group, ["--help"])
+    assert help_result.exit_code == 0
+    assert "Warning: external command could not be loaded." in help_result.output
+
+    result = CliRunner().invoke(group, ["broken"])
+    assert result.exit_code == 1
+    assert "broken-plugin 1.0 (broken_plugin:cli)" in result.output
+    assert "RuntimeError: missing dependency" in result.output
+
+
+def test_installed_plugin_system_exit_does_not_break_cli(monkeypatch):
+    exiting = _entry_point(
+        "exiting",
+        "exiting_plugin:cli",
+        exception=SystemExit(7),
+        distribution="exiting-plugin",
+    )
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        lambda **kwargs: [exiting],
+    )
+    group = click.Group()
+
+    register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
+
+    help_result = CliRunner().invoke(group, ["--help"])
+    assert help_result.exit_code == 0
+    result = CliRunner().invoke(group, ["exiting"])
+    assert result.exit_code == 1
+    assert "SystemExit: 7" in result.output
+
+
+def test_non_click_entry_point_gets_broken_placeholder(monkeypatch):
+    invalid = _entry_point("invalid", "plugin:value", command=object())
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        lambda **kwargs: [invalid],
+    )
+    group = click.Group()
+
+    register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
+
+    assert isinstance(group.commands["invalid"], BrokenPluginCommand)
+    result = CliRunner().invoke(group, ["invalid"])
+    assert result.exit_code == 1
+    assert "must resolve to a Click command or group" in result.output
+
+
+def test_entry_point_discovery_failure_does_not_break_group(monkeypatch, caplog):
+    def fail_discovery(**kwargs):
+        raise RuntimeError("invalid package metadata")
+
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        fail_discovery,
+    )
+    group = click.Group()
+
+    registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
+
+    assert registered == []
+    assert not group.commands
+    assert "Unable to discover external commands" in caplog.text
+    assert "invalid package metadata" in caplog.text
+
+
+def test_builtin_and_first_external_command_win_collisions(monkeypatch, caplog):
+    @click.command("version")
+    def core_version():
+        pass
+
+    @click.command()
+    def first():
+        pass
+
+    @click.command()
+    def second():
+        pass
+
+    entry_points = [
+        _entry_point("duplicate", "z_plugin:cli", command=second, distribution="z"),
+        _entry_point("version", "plugin:version", command=first),
+        _entry_point("duplicate", "a_plugin:cli", command=first, distribution="a"),
+    ]
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        lambda **kwargs: entry_points,
+    )
+    group = click.Group(commands={"version": core_version})
+
+    registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
+
+    assert registered == ["duplicate"]
+    assert group.commands["version"] is core_version
+    assert group.commands["duplicate"] is first
+    assert "the command name is already registered" in caplog.text
+
+
+def test_explicit_command_cannot_replace_existing_command(tmp_path):
+    command_path = _write_command(tmp_path / "version.py", "version")
+    group = _make_group()
+
+    @group.command("version")
+    def core_version():
+        pass
+
+    result = CliRunner().invoke(
+        group,
+        ["--load-command", str(command_path), "version"],
+    )
+
+    assert result.exit_code == 2
+    assert "the command name is already registered" in result.output
+
+
+def test_platform_entry_point_groups_and_environment_paths_are_separate(
+    tmp_path, monkeypatch
+):
+    ios_path = _write_command(tmp_path / "ios.py", "ios-file")
+    android_path = _write_command(tmp_path / "android.py", "android-file")
+
+    @click.command()
+    def ios_package():
+        pass
+
+    @click.command()
+    def android_package():
+        pass
+
+    def entry_points(*, group):
+        if group == IOS_CLI_PLUGIN_GROUP:
+            return [_entry_point("ios-package", "ios_plugin:cli", ios_package)]
+        return [_entry_point("android-package", "android_plugin:cli", android_package)]
+
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        entry_points,
+    )
+    monkeypatch.setenv("TEST_IOS_COMMANDS", str(ios_path))
+    monkeypatch.setenv("TEST_ANDROID_COMMANDS", str(android_path))
+    ios_group = click.Group()
+    android_group = click.Group()
+
+    register_cli_plugins(
+        ios_group,
+        entry_point_group=IOS_CLI_PLUGIN_GROUP,
+        environment_variable="TEST_IOS_COMMANDS",
+    )
+    register_cli_plugins(
+        android_group,
+        entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
+        environment_variable="TEST_ANDROID_COMMANDS",
+    )
+
+    assert set(ios_group.commands) == {"ios-file", "ios-package"}
+    assert set(android_group.commands) == {"android-file", "android-package"}

tests/common/test_command.py

@@ -42,6 +42,15 @@ class IndependentModule(RecordingModule):
     pass
 
 
+class URLRecordingModule(RecordingModule):
+    def collect_url_results(self):
+        self.add_url_result(
+            "https://example.org/message",
+            "2026-07-29 12:00:00.000000",
+            "test-chat",
+        )
+
+
 class CustomIOSBackupModule(RecordingModule):
     supported_commands = (("ios", "check-backup"),)
 
@@ -87,6 +96,21 @@ class TestCommand:
         alerts = json.loads((tmp_path / "alerts.json").read_text())
         assert alerts[0]["event"]["payload"] == "\\xa8\\xa9"
 
+    def test_stores_collected_urls(self, tmp_path):
+        cmd = RecordingCommand(results_path=str(tmp_path))
+        cmd.modules = [URLRecordingModule]
+
+        cmd.run()
+
+        assert json.loads((tmp_path / "urls.json").read_text()) == [
+            {
+                "url": "https://example.org/message",
+                "expanded_url": None,
+                "timestamp": "2026-07-29 12:00:00.000000",
+                "source": "test-chat",
+            }
+        ]
+
     def test_modules_run_in_stable_topological_order(self):
         cmd = RecordingCommand()
         cmd.modules = [ThirdModule, IndependentModule, SecondModule, FirstModule]

tests/common/test_indicators.py

@@ -197,6 +197,15 @@ class TestIndicators:
         assert matches[0] is None
         assert matches[1]
         assert matches[1].ioc.value == "example.org"
+        assert (
+            ind.get_expanded_url("https://tinyurl.com/nested")
+            == "https://www.example.org/landing"
+        )
+        assert (
+            ind.get_expanded_url("https://t.co/nested")
+            == "https://www.example.org/landing"
+        )
+        assert ind.get_expanded_url("https://bit.ly/failure") is None
         assert {call.args[0] for call in head.call_args_list} == {
             "https://bit.ly/failure",
             "https://tinyurl.com/nested",

tests/ios_backup/test_sms.py

@@ -18,6 +18,14 @@ class TestSMSModule:
         run_module(m)
         assert len(m.results) == 1
         assert len(m.timeline) == 2
+        assert m.url_results == [
+            {
+                "url": "https://badbadbad.example.org/",
+                "expanded_url": None,
+                "timestamp": "2019-08-29 23:13:30.000000",
+                "source": "sms",
+            }
+        ]
         assert len(m.alertstore.alerts) == 0
 
     def test_detection(self, indicator_file):

tests/ios_backup/test_whatsapp.py

@@ -0,0 +1,35 @@
+# 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
+
+from mvt.common.indicators import Indicators
+from mvt.ios.modules.mixed.whatsapp import Whatsapp
+
+
+def test_collect_url_results_includes_expansion():
+    module = Whatsapp(
+        results=[
+            {
+                "links": ["https://bit.ly/message"],
+                "isodate": "2026-07-29 12:00:00.000000",
+            }
+        ]
+    )
+    module.indicators = Indicators(log=logging.getLogger())
+    module.indicators.resolved_urls["https://bit.ly/message"] = (
+        "https://example.org/landing"
+    )
+
+    module.collect_url_results()
+
+    assert module.url_results == [
+        {
+            "url": "https://bit.ly/message",
+            "expanded_url": "https://example.org/landing",
+            "timestamp": "2026-07-29 12:00:00.000000",
+            "source": "whatsapp",
+        }
+    ]

tests/test_check_android_androidqf.py

@@ -3,6 +3,7 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
+import json
 import logging
 import os
 import shutil
@@ -32,6 +33,67 @@ class TestCheckAndroidqfCommand:
         result = runner.invoke(check_androidqf, [path])
         assert result.exit_code == 0
 
+    def test_check_stores_nested_sms_urls(self, tmp_path):
+        runner = CliRunner()
+        path = os.path.join(get_artifact_folder(), "androidqf")
+
+        result = runner.invoke(check_androidqf, ["--output", str(tmp_path), path])
+
+        assert result.exit_code == 0
+        urls = json.loads((tmp_path / "urls.json").read_text())
+        assert {entry["url"] for entry in urls} == {
+            "http://google.com",
+            "https://google.com/",
+        }
+        assert all(
+            set(entry) == {"url", "expanded_url", "timestamp", "source"}
+            for entry in urls
+        )
+        assert all(entry["source"] == "sms" for entry in urls)
+
+    def test_acquisition_context_is_passed_to_bugreport(self, tmp_path, mocker):
+        data_path = tmp_path / "androidqf"
+        data_path.mkdir()
+        (data_path / "acquisition.json").write_text(
+            json.dumps(
+                {
+                    "started": "2025-06-20T18:00:00Z",
+                    "adb_host_public_key": "QUJDRA== acquisition@host",
+                }
+            )
+        )
+        with zipfile.ZipFile(data_path / "bugreport.zip", "w"):
+            pass
+
+        nested_command = mocker.patch(
+            "mvt.android.cmd_check_androidqf.CmdAndroidCheckBugreport"
+        )
+        nested_command.return_value.timeline = []
+        nested_command.return_value.alertstore.alerts = []
+        command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
+        command.init()
+
+        assert command.run_bugreport_cmd() is True
+
+        assert nested_command.call_args.kwargs["module_options"][
+            "androidqf_acquisition"
+        ] == {
+            "started": "2025-06-20T18:00:00Z",
+            "adb_host_public_key": "QUJDRA== acquisition@host",
+        }
+
+    def test_acquisition_context_falls_back_to_public_key_file(self, tmp_path):
+        data_path = tmp_path / "androidqf"
+        data_path.mkdir()
+        (data_path / "adb_host_key.pub").write_text("QUJDRA== acquisition@host\n")
+        command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
+
+        command.init()
+
+        assert command.module_options["androidqf_acquisition"] == {
+            "adb_host_public_key": "QUJDRA== acquisition@host\n"
+        }
+
     def test_check_encrypted_backup_prompt_valid(self, mocker):
         """Prompt for password on CLI"""
         prompt_mock = mocker.patch(

tests/test_cmd_check_sysdiagnose.py

@@ -1,5 +1,7 @@
+import io
 import tarfile
 from datetime import timedelta
+from pathlib import Path
 
 from mvt.ios.cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
 from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
@@ -70,6 +72,46 @@ def test_check_sysdiagnose_from_archive_closes_archive(tmp_path):
         {"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
     ]
     assert command.executed[0].ips_files == [
-        {"file_path": "sysdiagnose/report.ips", "bug_type": 210}
+        {
+            "file_path": str(
+                Path(command.extracted_sysdiagnose_path) / "report.ips"
+            ),
+            "bug_type": 210,
+        }
     ]
     assert command.sysdiagnose_archive is None
+
+
+def test_archive_is_extracted_once_and_unsafe_members_are_skipped(tmp_path):
+    archive_path = tmp_path / "sysdiagnose.tar.gz"
+    escaped_path = tmp_path / "escaped.txt"
+    content = b"test content"
+    member = tarfile.TarInfo("sysdiagnose/artifact.txt")
+    member.size = len(content)
+
+    with tarfile.open(archive_path, "w:gz") as archive:
+        archive.addfile(member, io.BytesIO(content))
+        escaped = tarfile.TarInfo(f"sysdiagnose/../../{escaped_path.name}")
+        escaped.size = len(content)
+        archive.addfile(escaped, io.BytesIO(content))
+        link = tarfile.TarInfo("sysdiagnose/link")
+        link.type = tarfile.SYMTYPE
+        link.linkname = "/etc/hostname"
+        archive.addfile(link)
+
+    command = CmdIOSCheckSysdiagnose(target_path=str(archive_path))
+    try:
+        command.init()
+        extracted_path = Path(command.extracted_sysdiagnose_path)
+        assert (extracted_path / "artifact.txt").read_bytes() == content
+        assert not escaped_path.exists()
+        assert not (extracted_path / "link").exists()
+
+        module = SysdiagnoseExtraction()
+        command.module_init(module)
+        assert module.tar is None
+        assert module.parent_path == str(extracted_path.parent)
+    finally:
+        command.finish()
+
+    assert not extracted_path.exists()