Parallelize URL indicator checks

* fix: terminate dumpsys adb multiline values at structural lines

* fix dumpsys ADB multiline boundaries

---------

Co-authored-by: Janik Besendorf <janik@besendorf.org>

docs/iocs.md

@@ -37,8 +37,10 @@ export MVT_STIX2="/home/user/IOC1.stix2:/home/user/IOC2.stix2"
 ## Network Access
 
 When checking URL indicators, MVT follows recognized shortened URLs with an
-HTTP `HEAD` request. The following environment variables control these
-requests:
+HTTP `HEAD` request. URL checks are deduplicated and run concurrently, with at
+most 20 requests in progress at a time. Redirects within an individual URL
+chain are still followed sequentially. The following environment variables
+control these requests:
 
 - `MVT_NETWORK_ACCESS_ALLOWED` enables or disables network requests. It defaults
   to `true`. Set it to `false` to prevent MVT from attempting to resolve
@@ -73,4 +75,3 @@ You can automaticallly download the latest public indicator files with the comma
 
 Please [open an issue](https://github.com/mvt-project/mvt/issues/) to suggest new sources of STIX-formatted IOCs.
 
-

src/mvt/android/artifacts/dumpsys_adb.py

@@ -13,6 +13,13 @@ from .artifact import AndroidArtifact
 class DumpsysADBArtifact(AndroidArtifact):
     multiline_fields = ["user_keys", "keystore"]
 
+    @staticmethod
+    def is_structural_line(key: str, vals: list) -> bool:
+        if key == "}":
+            return True
+        # XML keystore continuations also split on "=", but never into an identifier.
+        return len(vals) == 2 and key.isidentifier()
+
     def indented_dump_parser(self, dump_data):
         """
         Parse the indented dumpsys output, generated by DualDumpOutputStream in Android.
@@ -41,10 +48,18 @@ class DumpsysADBArtifact(AndroidArtifact):
                 if key == "":
                     # If the line is empty, it's the terminator for the multiline value
                     in_multiline = False
-                    stack.pop()
-                else:
+                    if isinstance(stack[-1], list):
+                        stack.pop()
+                    continue
+
+                if not self.is_structural_line(key, vals):
                     current_dict.append(line.lstrip())
-                continue
+                    continue
+
+                in_multiline = False
+                if isinstance(stack[-1], list):
+                    stack.pop()
+                current_dict = stack[-1]
 
             if key == "}":
                 stack.pop()

src/mvt/android/modules/androidqf/__init__.py

@@ -10,7 +10,6 @@ from .aqf_processes import AQFProcesses
 from .aqf_settings import AQFSettings
 from .mounts import Mounts
 from .root_binaries import RootBinaries
-from .sms import SMS
 
 ANDROIDQF_MODULES = [
     AQFPackages,
@@ -18,7 +17,6 @@ ANDROIDQF_MODULES = [
     AQFGetProp,
     AQFSettings,
     AQFFiles,
-    SMS,
     RootBinaries,
     Mounts,
 ]

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

@@ -1,105 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 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 typing import Optional
-
-from mvt.android.modules.backup.helpers import prompt_or_load_android_backup_password
-from mvt.android.parsers.backup import (
-    AndroidBackupParsingError,
-    InvalidBackupPassword,
-    parse_ab_header,
-    parse_backup_file,
-    parse_tar_for_sms,
-)
-
-from .base import AndroidQFModule
-
-
-class SMS(AndroidQFModule):
-    """
-    This module analyse SMS file in backup
-    """
-
-    def __init__(
-        self,
-        file_path: Optional[str] = None,
-        target_path: Optional[str] = None,
-        results_path: Optional[str] = None,
-        module_options: Optional[dict] = None,
-        log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None,
-    ) -> None:
-        super().__init__(
-            file_path=file_path,
-            target_path=target_path,
-            results_path=results_path,
-            module_options=module_options,
-            log=log,
-            results=results,
-        )
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for message in self.results:
-            if "body" not in message:
-                continue
-
-            ioc_match = self.indicators.check_domains(message.get("links", []))
-            if ioc_match:
-                self.alertstore.critical(
-                    ioc_match.message, "", message, matched_indicator=ioc_match.ioc
-                )
-
-    def parse_backup(self, data):
-        header = parse_ab_header(data)
-        if not header["backup"]:
-            self.log.warning("Invalid backup format, backup.ab was not analysed")
-            return
-
-        password = None
-        if header["encryption"] != "none":
-            password = prompt_or_load_android_backup_password(
-                self.log, self.module_options
-            )
-            if not password:
-                self.log.critical("No backup password provided.")
-                return
-
-        try:
-            tardata = parse_backup_file(data, password=password)
-        except InvalidBackupPassword:
-            self.log.critical("Invalid backup password")
-            return
-        except AndroidBackupParsingError:
-            self.log.warning(
-                "Impossible to parse this backup file, please use"
-                " Android Backup Extractor instead"
-            )
-            return
-
-        if not tardata:
-            return
-
-        try:
-            self.results = parse_tar_for_sms(tardata)
-        except AndroidBackupParsingError:
-            self.log.info(
-                "Impossible to read SMS from the Android Backup, "
-                "please extract the SMS and try extracting it with "
-                "Android Backup Extractor"
-            )
-            return
-
-    def run(self) -> None:
-        files = self._get_files_by_pattern("*/backup.ab")
-        if not files:
-            self.log.info("No backup data found")
-            return
-
-        self.parse_backup(self._get_file_content(files[0]))
-        self.log.info("Identified %d SMS in backup data", len(self.results))

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

@@ -4,9 +4,8 @@
 #   https://license.mvt.re/1.1/
 
 
-from rich.prompt import Prompt
-
 from mvt.common.config import settings
+from mvt.common.password import prompt_password
 
 MVT_ANDROID_BACKUP_PASSWORD = "MVT_ANDROID_BACKUP_PASSWORD"
 
@@ -49,7 +48,7 @@ def prompt_or_load_android_backup_password(log, module_options):
 
     # The default is to allow interactivity
     elif module_options.get("interactive", True):
-        backup_password = Prompt.ask(prompt="Enter backup password", password=True)
+        backup_password = prompt_password("Enter backup password: ")
     else:
         log.critical(
             "Cannot decrypt backup because interactivity"

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

@@ -36,6 +36,8 @@ class SMS(BackupModule):
         if not self.indicators:
             return
 
+        messages = []
+        url_batches = []
         for message in self.results:
             if "body" not in message:
                 continue
@@ -44,12 +46,16 @@ class SMS(BackupModule):
             if message_links == []:
                 message_links = check_for_links(message.get("text", ""))
 
-            ioc_match = self.indicators.check_urls(message_links)
+            messages.append(message)
+            url_batches.append(message_links)
+
+        for message, ioc_match in zip(
+            messages, self.indicators.check_url_batches(url_batches)
+        ):
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", message, matched_indicator=ioc_match.ioc
                 )
-                continue
 
     def run(self) -> None:
         sms_path = "apps/com.android.providers.telephony/d_f/*_sms_backup"

src/mvt/common/indicators.py

@@ -7,9 +7,10 @@ import glob
 import json
 import logging
 import os
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
 from functools import lru_cache
-from typing import Any, Dict, Iterator, List, Optional
+from typing import Any, Dict, Iterator, List, Optional, Sequence
 
 import ahocorasick
 from appdirs import user_data_dir
@@ -22,6 +23,8 @@ MVT_INDICATORS_FOLDER = os.path.join(MVT_DATA_FOLDER, "indicators")
 
 logger = logging.getLogger(__name__)
 
+URL_CHECK_MAX_WORKERS = 20
+
 
 @dataclass
 class Indicator:
@@ -490,12 +493,41 @@ class Indicators:
         if not urls:
             return None
 
-        for url in urls:
-            check = self.check_url(url)
-            if check:
-                return check
+        return self.check_url_batches([urls])[0]
 
-        return None
+    def check_url_batches(
+        self, url_batches: Sequence[Optional[Sequence[str]]]
+    ) -> List[Optional[IndicatorMatch]]:
+        """Check batches of URLs concurrently while preserving batch order.
+
+        URLs are deduplicated across batches before checking. Each returned item
+        is the first indicator match from the corresponding input batch, using
+        the original URL order.
+        """
+        batches = [list(urls) if urls else [] for urls in url_batches]
+        unique_urls = list(
+            dict.fromkeys(url for urls in batches for url in urls)
+        )
+
+        if not unique_urls:
+            return [None] * len(batches)
+
+        if settings.NETWORK_ACCESS_ALLOWED and len(unique_urls) > 1:
+            worker_count = min(URL_CHECK_MAX_WORKERS, len(unique_urls))
+            with ThreadPoolExecutor(max_workers=worker_count) as executor:
+                url_matches = dict(
+                    zip(unique_urls, executor.map(self.check_url, unique_urls))
+                )
+        else:
+            url_matches = {url: self.check_url(url) for url in unique_urls}
+
+        return [
+            next(
+                (url_matches[url] for url in urls if url_matches[url] is not None),
+                None,
+            )
+            for urls in batches
+        ]
 
     def check_process(self, process: str) -> Optional[IndicatorMatch]:
         """Check the provided process name against the list of process

src/mvt/common/password.py

@@ -0,0 +1,143 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 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/
+
+"""Password prompts with masked keyboard feedback.
+
+This is based on the ``echo_char`` support added to :mod:`getpass` in Python
+3.14. MVT supports Python 3.10 and later, so it cannot use that API directly.
+"""
+
+import contextlib
+import io
+import os
+import sys
+import warnings
+from typing import TextIO
+
+
+def prompt_password(prompt: str) -> str:
+    """Return a password while displaying an asterisk for each character."""
+    try:
+        import termios
+
+        return _unix_getpass(prompt, termios)
+    except ImportError:
+        try:
+            import msvcrt
+
+            return _windows_getpass(prompt, msvcrt)
+        except ImportError:
+            return _fallback_getpass(prompt)
+
+
+def _unix_getpass(prompt: str, termios) -> str:
+    password = None
+    input_stream: TextIO
+    output_stream: TextIO
+    with contextlib.ExitStack() as stack:
+        try:
+            fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
+            tty = io.FileIO(fd, "w+")
+            stack.enter_context(tty)
+            input_stream = io.TextIOWrapper(tty)
+            stack.enter_context(input_stream)
+            output_stream = input_stream
+        except OSError:
+            stack.close()
+            try:
+                fd = sys.stdin.fileno()
+            except (AttributeError, ValueError):
+                return _fallback_getpass(prompt)
+            input_stream = sys.stdin
+            output_stream = sys.stderr
+
+        try:
+            old = termios.tcgetattr(fd)
+            new = old[:]
+            new[3] &= ~termios.ECHO
+            new[3] &= ~termios.ICANON
+            try:
+                termios.tcsetattr(fd, termios.TCSAFLUSH, new)
+                password = _readline_with_asterisks(output_stream, input_stream, prompt)
+            finally:
+                termios.tcsetattr(fd, termios.TCSAFLUSH, old)
+                output_stream.flush()
+        except termios.error:
+            if password is not None:
+                raise
+            password = _fallback_getpass(prompt)
+
+        output_stream.write("\n")
+        return password
+
+
+def _windows_getpass(prompt: str, msvcrt) -> str:
+    if sys.stdin is not sys.__stdin__:
+        return _fallback_getpass(prompt)
+
+    for char in prompt:
+        msvcrt.putwch(char)
+
+    password = ""
+    while True:
+        char = msvcrt.getwch()
+        if char in ("\r", "\n"):
+            break
+        if char == "\x03":
+            raise KeyboardInterrupt
+        if char == "\b":
+            if password:
+                msvcrt.putwch("\b")
+                msvcrt.putwch(" ")
+                msvcrt.putwch("\b")
+            password = password[:-1]
+        else:
+            password += char
+            msvcrt.putwch("*")
+
+    msvcrt.putwch("\r")
+    msvcrt.putwch("\n")
+    return password
+
+
+def _fallback_getpass(prompt: str) -> str:
+    warnings.warn(
+        "Can not control echo on the terminal.",
+        category=UserWarning,
+        stacklevel=2,
+    )
+    print("Warning: Password input may be echoed.", file=sys.stderr)
+    return input(prompt)
+
+
+def _readline_with_asterisks(
+    output_stream: TextIO, input_stream: TextIO, prompt: str
+) -> str:
+    output_stream.write(prompt)
+    output_stream.flush()
+
+    password = ""
+    eof_pressed = False
+    while True:
+        char = input_stream.read(1)
+        if char in ("\n", "\r"):
+            break
+        if char == "\x03":
+            raise KeyboardInterrupt
+        if char in ("\x7f", "\b"):
+            if password:
+                output_stream.write("\b \b")
+                output_stream.flush()
+            password = password[:-1]
+        elif char == "\x04":
+            if eof_pressed:
+                break
+            eof_pressed = True
+        elif char != "\x00":
+            password += char
+            output_stream.write("*")
+            output_stream.flush()
+            eof_pressed = False
+    return password

src/mvt/common/url.py

@@ -5,6 +5,7 @@
 
 import logging
 from typing import Optional
+from urllib.parse import urlparse
 
 import requests
 from tld import get_tld
@@ -373,6 +374,10 @@ class URL:
         :rtype: bool
 
         """
+        parsed_url = urlparse(self.url if "://" in self.url else f"//{self.url}")
+        if self.domain.lower() == "goo.gl" and parsed_url.path.startswith("/maps/"):
+            return False
+
         if self.domain.lower() in SHORTENER_DOMAINS:
             self.is_shortened = True
 

src/mvt/ios/cli.py

@@ -8,8 +8,6 @@ import logging
 import os
 
 import click
-from rich.prompt import Prompt
-
 from mvt.common.cmd_check_iocs import CmdCheckIOCS
 from mvt.common.completion import (
     SUPPORTED_SHELLS,
@@ -49,6 +47,7 @@ from mvt.common.help import (
     HELP_MSG_COMPLETION,
 )
 from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
+from mvt.common.password import prompt_password
 from .cmd_check_backup import CmdIOSCheckBackup
 from .cmd_check_fs import CmdIOSCheckFS
 from .decrypt import DecryptBackup
@@ -201,7 +200,7 @@ def decrypt_backup(ctx, destination, password, key_file, hashes, backup_path):
         log.info("Using password from %s environment variable", MVT_IOS_BACKUP_PASSWORD)
         backup.decrypt_with_password(os.environ[MVT_IOS_BACKUP_PASSWORD])
     else:
-        sekrit = Prompt.ask("Enter backup password", password=True)
+        sekrit = prompt_password("Enter backup password: ")
         backup.decrypt_with_password(sekrit)
 
     if not backup.can_process():
@@ -253,7 +252,7 @@ def extract_key(password, key_file, backup_path):
         log.info("Using password from %s environment variable", MVT_IOS_BACKUP_PASSWORD)
         password = os.environ[MVT_IOS_BACKUP_PASSWORD]
     else:
-        password = Prompt.ask("Enter backup password", password=True)
+        password = prompt_password("Enter backup password: ")
 
     backup.decrypt_with_password(password)
     backup.get_key()

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

@@ -76,8 +76,10 @@ class Shortcuts(IOSExtraction):
         if not self.indicators:
             return
 
-        for result in self.results:
-            ioc_match = self.indicators.check_urls(result["action_urls"])
+        url_batches = [result["action_urls"] for result in self.results]
+        for result, ioc_match in zip(
+            self.results, self.indicators.check_url_batches(url_batches)
+        ):
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc

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

@@ -85,12 +85,17 @@ class SMS(IOSExtraction):
         if not self.indicators:
             return
 
+        url_batches = []
         for result in self.results:
             message_links = result.get("links", [])
             # Making sure not link was ignored
             if message_links == []:
                 message_links = check_for_links(result.get("text", ""))
-            ioc_match = self.indicators.check_urls(message_links)
+            url_batches.append(message_links)
+
+        for result, ioc_match in zip(
+            self.results, self.indicators.check_url_batches(url_batches)
+        ):
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc

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

@@ -70,6 +70,8 @@ class WebkitSessionResourceLog(IOSExtraction):
         if not self.indicators:
             return
 
+        records = []
+        url_batches = []
         for _, entries in self.results.items():
             for entry in entries:
                 source_domains = self._extract_domains(entry["redirect_source"])
@@ -94,36 +96,42 @@ class WebkitSessionResourceLog(IOSExtraction):
                     )
                 )
 
-                ioc_match = self.indicators.check_urls(all_origins)
-                if ioc_match:
-                    self.alertstore.critical(
-                        ioc_match.message, "", entry, matched_indicator=ioc_match.ioc
-                    )
+                records.append((entry, source_domains, destination_domains))
+                url_batches.append(all_origins)
 
-                    redirect_path = ""
-                    if len(source_domains) > 0:
-                        redirect_path += "SOURCE: "
-                        for idx, item in enumerate(source_domains):
-                            source_domains[idx] = f'"{item}"'
+        for record, ioc_match in zip(
+            records, self.indicators.check_url_batches(url_batches)
+        ):
+            if ioc_match:
+                entry, source_domains, destination_domains = record
+                self.alertstore.critical(
+                    ioc_match.message, "", entry, matched_indicator=ioc_match.ioc
+                )
 
-                        redirect_path += ", ".join(source_domains)
-                        redirect_path += " -> "
+                redirect_path = ""
+                if len(source_domains) > 0:
+                    redirect_path += "SOURCE: "
+                    for idx, item in enumerate(source_domains):
+                        source_domains[idx] = f'"{item}"'
 
-                    redirect_path += f'ORIGIN: "{entry["origin"]}"'
+                    redirect_path += ", ".join(source_domains)
+                    redirect_path += " -> "
 
-                    if len(destination_domains) > 0:
-                        redirect_path += " -> "
-                        redirect_path += "DESTINATION: "
-                        for idx, item in enumerate(destination_domains):
-                            destination_domains[idx] = f'"{item}"'
+                redirect_path += f'ORIGIN: "{entry["origin"]}"'
 
-                        redirect_path += ", ".join(destination_domains)
+                if len(destination_domains) > 0:
+                    redirect_path += " -> "
+                    redirect_path += "DESTINATION: "
+                    for idx, item in enumerate(destination_domains):
+                        destination_domains[idx] = f'"{item}"'
 
-                    self.alertstore.high(
-                        f"Found HTTP redirect between suspicious domains: {redirect_path}",
-                        "",
-                        entry,
-                    )
+                    redirect_path += ", ".join(destination_domains)
+
+                self.alertstore.high(
+                    f"Found HTTP redirect between suspicious domains: {redirect_path}",
+                    "",
+                    entry,
+                )
 
     def _extract_browsing_stats(self, log_path):
         items = []

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

@@ -61,8 +61,10 @@ class Whatsapp(IOSExtraction):
         if not self.indicators:
             return
 
-        for result in self.results:
-            ioc_match = self.indicators.check_urls(result.get("links", []))
+        url_batches = [result.get("links", []) for result in self.results]
+        for result, ioc_match in zip(
+            self.results, self.indicators.check_url_batches(url_batches)
+        ):
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc

tests/android/test_artifact_dumpsys_adb.py

@@ -30,6 +30,66 @@ class TestDumpsysADBArtifact:
         )
         assert user_key["user"] == "user@linux"
 
+    def test_parsing_adb_wifi(self):
+        da_adb = DumpsysADBArtifact()
+        file = get_artifact("android_data/dumpsys_adb_wifi.txt")
+        with open(file, "rb") as f:
+            data = f.read()
+
+        da_adb.parse(data)
+
+        assert len(da_adb.results) == 1
+        adb_data = da_adb.results[0]
+        assert "user_keys" in adb_data
+        assert len(adb_data["user_keys"]) == 1
+
+        user_key = adb_data["user_keys"][0]
+        assert (
+            user_key["fingerprint"] == "F0:A1:3D:8C:B3:F4:7B:09:9F:EE:8B:D8:38:2E:BD:C6"
+        )
+        assert user_key["user"] == "user@linux"
+
+        # The adb_wifi block following the keystore is not part of the keystore.
+        assert b"adb_wifi" not in adb_data["keystore"]
+
+    def test_parsing_multiline_terminated_by_structural_line(self):
+        dump_data = (
+            b"debugging_manager={\n"
+            b"  keystore=ABX\x00\x0bkeyStore\x00\x02\x11\n"
+            b"  connected_to_adb=true\n"
+            b"  adb_wifi={\n"
+            b"    enabled=false\n"
+            b"    tls_port=0\n"
+            b"  }\n"
+        )
+
+        parsed = DumpsysADBArtifact().indented_dump_parser(dump_data)
+
+        debugging_manager = parsed["debugging_manager"]
+        assert debugging_manager["keystore"] == [b"ABX\x00\x0bkeyStore\x00\x02\x11"]
+        assert debugging_manager["connected_to_adb"] == b"true"
+        assert debugging_manager["adb_wifi"] == {
+            "enabled": b"false",
+            "tls_port": b"0",
+        }
+
+    def test_parsing_multiline_terminated_by_closing_brace(self):
+        dump_data = (
+            b"debugging_manager={\n"
+            b"  keystore=ABX\x00\x0bkeyStore\x00\x02\x11\n"
+            b"}\n"
+            b"other={\n"
+            b"  value=true\n"
+            b"}\n"
+        )
+
+        parsed = DumpsysADBArtifact().indented_dump_parser(dump_data)
+
+        assert parsed["debugging_manager"]["keystore"] == [
+            b"ABX\x00\x0bkeyStore\x00\x02\x11"
+        ]
+        assert parsed["other"] == {"value": b"true"}
+
     def test_parsing_adb_xml(self):
         da_adb = DumpsysADBArtifact()
         file = get_artifact("android_data/dumpsys_adb_xml.txt")

tests/android_androidqf/test_sms.py

@@ -1,91 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-import logging
-import os
-from pathlib import Path
-
-from mvt.android.modules.androidqf.sms import SMS
-from mvt.common.module import run_module
-
-from ..utils import get_android_androidqf, get_artifact_folder, list_files
-
-TEST_BACKUP_PASSWORD = "123456"
-
-
-class TestAndroidqfSMSAnalysis:
-    def test_androidqf_sms(self):
-        data_path = get_android_androidqf()
-        m = SMS(target_path=data_path, log=logging)
-        files = list_files(data_path)
-        parent_path = Path(data_path).absolute().parent.as_posix()
-        m.from_dir(parent_path, files)
-        run_module(m)
-        assert len(m.results) == 2
-        assert len(m.timeline) == 0
-        assert len(m.alertstore.alerts) == 0
-
-    def test_androidqf_sms_encrypted_password_valid(self):
-        data_path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
-        m = SMS(
-            target_path=data_path,
-            log=logging,
-            module_options={"backup_password": TEST_BACKUP_PASSWORD},
-        )
-        files = list_files(data_path)
-        parent_path = Path(data_path).absolute().parent.as_posix()
-        m.from_dir(parent_path, files)
-        run_module(m)
-        assert len(m.results) == 1
-
-    def test_androidqf_sms_encrypted_password_prompt(self, mocker):
-        data_path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
-        prompt_mock = mocker.patch(
-            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
-        )
-        m = SMS(
-            target_path=data_path,
-            log=logging,
-            module_options={},
-        )
-        files = list_files(data_path)
-        parent_path = Path(data_path).absolute().parent.as_posix()
-        m.from_dir(parent_path, files)
-        run_module(m)
-        assert prompt_mock.call_count == 1
-        assert len(m.results) == 1
-
-    def test_androidqf_sms_encrypted_password_invalid(self, caplog):
-        data_path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
-        with caplog.at_level(logging.CRITICAL):
-            m = SMS(
-                target_path=data_path,
-                log=logging,
-                module_options={"backup_password": "invalid_password"},
-            )
-            files = list_files(data_path)
-            parent_path = Path(data_path).absolute().parent.as_posix()
-            m.from_dir(parent_path, files)
-            run_module(m)
-        assert len(m.results) == 0
-        assert "Invalid backup password" in caplog.text
-
-    def test_androidqf_sms_encrypted_no_interactive(self, caplog):
-        data_path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
-        with caplog.at_level(logging.CRITICAL):
-            m = SMS(
-                target_path=data_path,
-                log=logging,
-                module_options={"interactive": False},
-            )
-            files = list_files(data_path)
-            parent_path = Path(data_path).absolute().parent.as_posix()
-            m.from_dir(parent_path, files)
-            run_module(m)
-        assert len(m.results) == 0
-        assert (
-            "Cannot decrypt backup because interactivity was disabled and the password was not supplied"
-            in caplog.text
-        )

tests/common/test_indicators.py

@@ -5,7 +5,9 @@
 
 import logging
 import os
+import threading
 
+import requests
 
 from mvt.common.config import settings
 from mvt.common.indicators import Indicators
@@ -80,6 +82,127 @@ class TestIndicators:
         assert ind.check_url("https://198.51.100.1:8080/")
         assert ind.check_url("https://1.1.1.1/") is None
 
+    def test_google_maps_short_url_is_not_resolved(self, indicator_file, mocker):
+        head_request = mocker.patch("mvt.common.url.requests.head")
+        ind = Indicators(log=logging)
+        ind.load_indicators_files([indicator_file], load_default=False)
+
+        assert ind.check_url("https://goo.gl/maps/example") is None
+        head_request.assert_not_called()
+
+    def test_check_url_batches_preserves_order(self, indicator_file):
+        ind = Indicators(log=logging)
+        ind.load_indicators_files([indicator_file], load_default=False)
+
+        matches = ind.check_url_batches(
+            [
+                [
+                    "https://github.com",
+                    "http://example.com/thisisbad",
+                    "https://www.example.org/foobar",
+                ],
+                ["https://github.com", "https://www.example.org/foobar"],
+                [],
+                None,
+            ]
+        )
+
+        assert matches[0]
+        assert matches[0].ioc.value == "http://example.com/thisisbad"
+        assert matches[1]
+        assert matches[1].ioc.value == "example.org"
+        assert matches[2] is None
+        assert matches[3] is None
+
+    def test_check_url_batches_deduplicates_and_limits_workers(
+        self, indicator_file, mocker
+    ):
+        ind = Indicators(log=logging)
+        ind.load_indicators_files([indicator_file], load_default=False)
+        mocker.patch("mvt.common.indicators.URL_CHECK_MAX_WORKERS", 2)
+
+        barrier = threading.Barrier(2)
+        lock = threading.Lock()
+        calls = []
+        active = 0
+        max_active = 0
+
+        def head_request(url, timeout):
+            nonlocal active, max_active
+            with lock:
+                calls.append(url)
+                active += 1
+                max_active = max(max_active, active)
+                call_number = len(calls)
+
+            try:
+                if call_number <= 2:
+                    barrier.wait(timeout=5)
+                return mocker.Mock(status_code=200, headers={})
+            finally:
+                with lock:
+                    active -= 1
+
+        mocker.patch("mvt.common.url.requests.head", side_effect=head_request)
+        urls = [
+            "https://bit.ly/one",
+            "https://tinyurl.com/two",
+            "https://t.co/three",
+        ]
+
+        assert ind.check_url_batches([urls, [urls[0]]]) == [None, None]
+        assert sorted(calls) == sorted(urls)
+        assert max_active == 2
+
+    def test_check_url_batches_respects_disabled_network(
+        self, indicator_file, mocker
+    ):
+        ind = Indicators(log=logging)
+        ind.load_indicators_files([indicator_file], load_default=False)
+        mocker.patch("mvt.common.indicators.settings.NETWORK_ACCESS_ALLOWED", False)
+        head_request = mocker.patch("mvt.common.url.requests.head")
+
+        assert ind.check_url_batches([["https://bit.ly/example"]]) == [None]
+        head_request.assert_not_called()
+
+    def test_check_url_batches_handles_nested_redirects_and_request_failures(
+        self, indicator_file, mocker
+    ):
+        ind = Indicators(log=logging)
+        ind.load_indicators_files([indicator_file], load_default=False)
+
+        def head_request(url, timeout):
+            if url == "https://bit.ly/failure":
+                raise requests.Timeout()
+            if url == "https://tinyurl.com/nested":
+                return mocker.Mock(
+                    status_code=301,
+                    headers={"Location": "https://t.co/nested"},
+                )
+            if url == "https://t.co/nested":
+                return mocker.Mock(
+                    status_code=302,
+                    headers={"Location": "https://www.example.org/landing"},
+                )
+            raise AssertionError(f"Unexpected URL: {url}")
+
+        head = mocker.patch(
+            "mvt.common.url.requests.head", side_effect=head_request
+        )
+
+        matches = ind.check_url_batches(
+            [["https://bit.ly/failure"], ["https://tinyurl.com/nested"]]
+        )
+
+        assert matches[0] is None
+        assert matches[1]
+        assert matches[1].ioc.value == "example.org"
+        assert {call.args[0] for call in head.call_args_list} == {
+            "https://bit.ly/failure",
+            "https://tinyurl.com/nested",
+            "https://t.co/nested",
+        }
+
     def test_check_file_hash(self, indicator_file):
         ind = Indicators(log=logging)
         ind.load_indicators_files([indicator_file], load_default=False)

tests/common/test_password.py

@@ -0,0 +1,28 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2026 The MVT Authors.
+# Use of this software is governed by the MVT License 1.1 that can be found at
+# https://license.mvt.re/1.1/
+
+from io import StringIO
+
+from mvt.common.password import _readline_with_asterisks
+
+
+def test_readline_with_asterisks():
+    output = StringIO()
+
+    password = _readline_with_asterisks(
+        output, StringIO("pass\x7fword\n"), "Enter backup password: "
+    )
+
+    assert password == "pasword"
+    assert output.getvalue() == "Enter backup password: ****\b \b****"
+
+
+def test_readline_with_asterisks_ignores_nul_and_handles_eof():
+    output = StringIO()
+
+    password = _readline_with_asterisks(output, StringIO("a\x00b\x04\x04"), "")
+
+    assert password == "ab"
+    assert output.getvalue() == "**"

tests/common/test_url.py

@@ -0,0 +1,24 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2023 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 pytest
+
+from mvt.common.url import URL
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://goo.gl/maps/example",
+        "http://goo.gl/maps/example?entry=message",
+        "goo.gl/maps/example",
+    ],
+)
+def test_google_maps_url_is_not_shortened(url):
+    assert URL(url).check_if_shortened() is False
+
+
+def test_other_google_short_url_is_shortened():
+    assert URL("https://goo.gl/example").check_if_shortened() is True

tests/ios_backup/test_sms.py

@@ -29,3 +29,31 @@ class TestSMSModule:
         m.indicators = ind
         run_module(m)
         assert len(m.alertstore.alerts) == 1
+
+    def test_detection_batches_urls_and_preserves_event(self, indicator_file, mocker):
+        results = [
+            {
+                "text": "first",
+                "links": ["http://example.com/thisisbad"],
+            },
+            {
+                "text": "second",
+                "links": ["https://github.com"],
+            },
+        ]
+        m = SMS(results=results)
+        ind = Indicators(log=logging.getLogger())
+        ind.parse_stix2(indicator_file)
+        batch_check = mocker.spy(ind, "check_url_batches")
+        m.indicators = ind
+
+        m.check_indicators()
+
+        batch_check.assert_called_once_with(
+            [
+                ["http://example.com/thisisbad"],
+                ["https://github.com"],
+            ]
+        )
+        assert len(m.alertstore.alerts) == 1
+        assert m.alertstore.alerts[0].event is results[0]

tests/test_check_android_androidqf.py

@@ -27,21 +27,24 @@ class TestCheckAndroidqfCommand:
     def test_check_encrypted_backup_prompt_valid(self, mocker):
         """Prompt for password on CLI"""
         prompt_mock = mocker.patch(
-            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+            "mvt.android.modules.backup.helpers.prompt_password",
+            return_value=TEST_BACKUP_PASSWORD,
         )
 
         runner = CliRunner()
         path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
         result = runner.invoke(check_androidqf, [path])
 
-        # Called twice, once in AnroidQF SMS module and once in Backup SMS module
-        assert prompt_mock.call_count == 2
+        # The password entered for the AndroidQF SMS module is reused by the
+        # nested backup command.
+        assert prompt_mock.call_count == 1
         assert result.exit_code == 0
 
     def test_check_encrypted_backup_cli(self, mocker):
         """Provide password as CLI argument"""
         prompt_mock = mocker.patch(
-            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+            "mvt.android.modules.backup.helpers.prompt_password",
+            return_value=TEST_BACKUP_PASSWORD,
         )
 
         runner = CliRunner()
@@ -56,7 +59,8 @@ class TestCheckAndroidqfCommand:
     def test_check_encrypted_backup_env(self, mocker):
         """Provide password as environment variable"""
         prompt_mock = mocker.patch(
-            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+            "mvt.android.modules.backup.helpers.prompt_password",
+            return_value=TEST_BACKUP_PASSWORD,
         )
 
         os.environ["MVT_ANDROID_BACKUP_PASSWORD"] = TEST_BACKUP_PASSWORD

tests/test_check_android_backup.py

@@ -20,7 +20,8 @@ class TestCheckAndroidBackupCommand:
     def test_check_encrypted_backup_prompt_valid(self, mocker):
         """Prompt for password on CLI"""
         prompt_mock = mocker.patch(
-            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+            "mvt.android.modules.backup.helpers.prompt_password",
+            return_value=TEST_BACKUP_PASSWORD,
         )
         runner = CliRunner()
         path = os.path.join(get_artifact_folder(), "androidqf_encrypted/backup.ab")
@@ -32,7 +33,8 @@ class TestCheckAndroidBackupCommand:
     def test_check_encrypted_backup_cli(self, mocker):
         """Provide password as CLI argument"""
         prompt_mock = mocker.patch(
-            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+            "mvt.android.modules.backup.helpers.prompt_password",
+            return_value=TEST_BACKUP_PASSWORD,
         )
 
         runner = CliRunner()
@@ -60,7 +62,8 @@ class TestCheckAndroidBackupCommand:
     def test_check_encrypted_backup_env(self, mocker):
         """Provide password as environment variable"""
         prompt_mock = mocker.patch(
-            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+            "mvt.android.modules.backup.helpers.prompt_password",
+            return_value=TEST_BACKUP_PASSWORD,
         )
 
         os.environ["MVT_ANDROID_BACKUP_PASSWORD"] = TEST_BACKUP_PASSWORD