Store message URLs in analysis output

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/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"

src/mvt/android/cmd_check_androidqf.py

@@ -262,6 +262,7 @@ def run_bugreport_cmd(self) -> bool:
             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 @@ def run_backup_cmd(self) -> bool:
         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 @@ def run_intrusion_logs_cmd(self) -> bool:
             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 @@ def check_indicators(self) -> None:
                     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/common/command.py

@@ -20,7 +20,7 @@
 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 @@ def __init__(
         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 @@ def _store_alerts(self) -> None:
         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 @@ def run(self) -> None:
             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 @@ def run(self) -> None:
         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 @@ def __init__(self, log=logger) -> None:
         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 @@ def check_url(self, url: str) -> Optional[IndicatorMatch]:
                         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 @@ def check_url(self, url: str) -> Optional[IndicatorMatch]:
 
         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 @@
     ModuleResults,
     ModuleSerializedResult,
     ModuleTimeline,
+    URLResult,
 )
 from .utils import CustomJSONEncoder, exec_or_profile
 
@@ -82,6 +83,7 @@ def __init__(
 
         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 @@ def get_slug(cls) -> str:
     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 @@
 ModuleResults = Any
 
 
+class URLResult(TypedDict):
+    url: str
+    expanded_url: Optional[str]
+    timestamp: Optional[str]
+    source: str
+
+
 @dataclass
 class ModuleAtomicTimeline:
     timestamp: str

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

@@ -101,6 +101,11 @@ def check_indicators(self) -> None:
                     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 @@ def check_indicators(self) -> None:
                     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/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 @@ def test_store_alerts_handles_bytes(self, tmp_path):
         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 @@ def head_request(url, timeout):
         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 @@ def test_sms(self):
         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

@@ -33,6 +33,24 @@ def test_check(self):
         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()