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()