From 8140e350f7544915f3a6bbff24741c815dddd63d Mon Sep 17 00:00:00 2001 From: Rory Flynn <75283103+roaree@users.noreply.github.com> Date: Wed, 29 Jul 2026 13:57:34 +0200 Subject: [PATCH 1/4] Update version.py (#854) --- src/mvt/common/version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/mvt/common/version.py b/src/mvt/common/version.py index d3fc913..37d62f1 100644 --- a/src/mvt/common/version.py +++ b/src/mvt/common/version.py @@ -3,4 +3,4 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ -MVT_VERSION = "2026.5.12" +MVT_VERSION = "2026.6.29" From 53fb12aee869f2365c972d54900c6b6250f0080d Mon Sep 17 00:00:00 2001 From: Rory Flynn <75283103+roaree@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:08:42 +0200 Subject: [PATCH 2/4] Correct version.py (#855) --- src/mvt/common/version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/mvt/common/version.py b/src/mvt/common/version.py index 37d62f1..c8e7716 100644 --- a/src/mvt/common/version.py +++ b/src/mvt/common/version.py @@ -3,4 +3,4 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ -MVT_VERSION = "2026.6.29" +MVT_VERSION = "2026.7.29" From 8617e0bf54f1025de2945d8290e61d0a65292c22 Mon Sep 17 00:00:00 2001 From: besendorf Date: Wed, 5 Aug 2026 17:36:49 +0200 Subject: [PATCH 3/4] Alert on AndroidQF trusted ADB keys (#860) --- src/mvt/android/cmd_check_androidqf.py | 42 ++++++ .../modules/bugreport/dumpsys_adb_state.py | 123 ++++++++++++++++++ tests/android/test_artifact_dumpsys_adb.py | 118 +++++++++++++++++ tests/test_check_android_androidqf.py | 44 +++++++ 4 files changed, 327 insertions(+) diff --git a/src/mvt/android/cmd_check_androidqf.py b/src/mvt/android/cmd_check_androidqf.py index 319b614..6f1089c 100644 --- a/src/mvt/android/cmd_check_androidqf.py +++ b/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) diff --git a/src/mvt/android/modules/bugreport/dumpsys_adb_state.py b/src/mvt/android/modules/bugreport/dumpsys_adb_state.py index 506af30..06b29d5 100644 --- a/src/mvt/android/modules/bugreport/dumpsys_adb_state.py +++ b/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 "" + 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, + ) diff --git a/tests/android/test_artifact_dumpsys_adb.py b/tests/android/test_artifact_dumpsys_adb.py index 47dc3df..f5574a6 100644 --- a/tests/android/test_artifact_dumpsys_adb.py +++ b/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 == [] diff --git a/tests/test_check_android_androidqf.py b/tests/test_check_android_androidqf.py index 9b271be..a8a6a5a 100644 --- a/tests/test_check_android_androidqf.py +++ b/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,49 @@ class TestCheckAndroidqfCommand: result = runner.invoke(check_androidqf, [path]) assert result.exit_code == 0 + 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( From 93b7fb52322b4e98fd9f9251e5180432feb7cda5 Mon Sep 17 00:00:00 2001 From: besendorf Date: Wed, 5 Aug 2026 23:21:14 +0200 Subject: [PATCH 4/4] Store message URLs in analysis output (#856) --- docs/android/backup.md | 6 +++++ docs/ios/records.md | 9 +++++++ src/mvt/android/cmd_check_androidqf.py | 3 +++ src/mvt/android/modules/backup/sms.py | 5 ++++ src/mvt/common/command.py | 13 +++++++++- src/mvt/common/indicators.py | 15 ++++++++++- src/mvt/common/module.py | 28 +++++++++++++++++++++ src/mvt/common/module_types.py | 9 ++++++- src/mvt/ios/modules/mixed/sms.py | 5 ++++ src/mvt/ios/modules/mixed/whatsapp.py | 5 ++++ tests/common/test_command.py | 24 ++++++++++++++++++ tests/common/test_indicators.py | 9 +++++++ tests/ios_backup/test_sms.py | 8 ++++++ tests/ios_backup/test_whatsapp.py | 35 ++++++++++++++++++++++++++ tests/test_check_android_androidqf.py | 18 +++++++++++++ 15 files changed, 189 insertions(+), 3 deletions(-) create mode 100644 tests/ios_backup/test_whatsapp.py diff --git a/docs/android/backup.md b/docs/android/backup.md index 81344aa..fae0d17 100644 --- a/docs/android/backup.md +++ b/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. diff --git a/docs/ios/records.md b/docs/ios/records.md index 17be192..016c861 100644 --- a/docs/ios/records.md +++ b/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" diff --git a/src/mvt/android/cmd_check_androidqf.py b/src/mvt/android/cmd_check_androidqf.py index 6f1089c..ffa3e13 100644 --- a/src/mvt/android/cmd_check_androidqf.py +++ b/src/mvt/android/cmd_check_androidqf.py @@ -262,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: @@ -299,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 @@ -378,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 diff --git a/src/mvt/android/modules/backup/sms.py b/src/mvt/android/modules/backup/sms.py index c42e9a4..1c73ea2 100644 --- a/src/mvt/android/modules/backup/sms.py +++ b/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): diff --git a/src/mvt/common/command.py b/src/mvt/common/command.py index 8d21aa3..aab53e6 100644 --- a/src/mvt/common/command.py +++ b/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() diff --git a/src/mvt/common/indicators.py b/src/mvt/common/indicators.py index 7186ae3..94afadf 100644 --- a/src/mvt/common/indicators.py +++ b/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. diff --git a/src/mvt/common/module.py b/src/mvt/common/module.py index 14c5fe5..cd127a6 100644 --- a/src/mvt/common/module.py +++ b/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: diff --git a/src/mvt/common/module_types.py b/src/mvt/common/module_types.py index 06fdc12..a47ffe4 100644 --- a/src/mvt/common/module_types.py +++ b/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 diff --git a/src/mvt/ios/modules/mixed/sms.py b/src/mvt/ios/modules/mixed/sms.py index 596ba47..d1afd06 100644 --- a/src/mvt/ios/modules/mixed/sms.py +++ b/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) diff --git a/src/mvt/ios/modules/mixed/whatsapp.py b/src/mvt/ios/modules/mixed/whatsapp.py index cf3a49b..0a80aad 100644 --- a/src/mvt/ios/modules/mixed/whatsapp.py +++ b/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 diff --git a/tests/common/test_command.py b/tests/common/test_command.py index fa90a3e..4dbfe1a 100644 --- a/tests/common/test_command.py +++ b/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] diff --git a/tests/common/test_indicators.py b/tests/common/test_indicators.py index bbb55b2..ac50f82 100644 --- a/tests/common/test_indicators.py +++ b/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", diff --git a/tests/ios_backup/test_sms.py b/tests/ios_backup/test_sms.py index 5081a67..64646f4 100644 --- a/tests/ios_backup/test_sms.py +++ b/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): diff --git a/tests/ios_backup/test_whatsapp.py b/tests/ios_backup/test_whatsapp.py new file mode 100644 index 0000000..864fb84 --- /dev/null +++ b/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", + } + ] diff --git a/tests/test_check_android_androidqf.py b/tests/test_check_android_androidqf.py index a8a6a5a..2253a50 100644 --- a/tests/test_check_android_androidqf.py +++ b/tests/test_check_android_androidqf.py @@ -33,6 +33,24 @@ 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()