From 1532578b39ef4169864b629d2a8727eaeda68977 Mon Sep 17 00:00:00 2001 From: besendorf Date: Tue, 14 Jul 2026 21:58:46 +0200 Subject: [PATCH 01/12] Validate iOS backup path before checks (#825) * Validate iOS backup path before checks * Fix iOS backup path mypy typing * Fix custom module test backup fixture --- src/mvt/ios/cli.py | 5 +++- src/mvt/ios/cmd_check_backup.py | 48 +++++++++++++++++++++++++++++++++ tests/test_check_ios_backup.py | 17 ++++++++++++ tests/test_custom_modules.py | 2 ++ 4 files changed, 71 insertions(+), 1 deletion(-) diff --git a/src/mvt/ios/cli.py b/src/mvt/ios/cli.py index 176b036..e15f8db 100644 --- a/src/mvt/ios/cli.py +++ b/src/mvt/ios/cli.py @@ -323,7 +323,10 @@ def check_backup( cmd.list_modules() return - log.info("Checking iTunes backup located at: %s", backup_path) + if not cmd.resolve_backup_path(): + ctx.exit(1) + + log.info("Checking iTunes backup located at: %s", cmd.target_path) cmd.run() cmd.show_alerts_brief() diff --git a/src/mvt/ios/cmd_check_backup.py b/src/mvt/ios/cmd_check_backup.py index 4264a7f..1b90584 100644 --- a/src/mvt/ios/cmd_check_backup.py +++ b/src/mvt/ios/cmd_check_backup.py @@ -4,6 +4,7 @@ # https://license.mvt.re/1.1/ import logging +import os from typing import Optional from mvt.common.command import Command @@ -16,6 +17,12 @@ from .modules.mixed import MIXED_MODULES log = logging.getLogger(__name__) +def is_ios_backup_folder(path: str) -> bool: + return os.path.isfile(os.path.join(path, "Manifest.db")) and os.path.isfile( + os.path.join(path, "Info.plist") + ) + + class CmdIOSCheckBackup(Command): def __init__( self, @@ -52,5 +59,46 @@ class CmdIOSCheckBackup(Command): self.name = "check-backup" self.modules = BACKUP_MODULES + MIXED_MODULES + def resolve_backup_path(self) -> bool: + target_path = getattr(self, "target_path", None) + if not isinstance(target_path, str) or not target_path: + return False + + if is_ios_backup_folder(target_path): + return True + + if not os.path.isdir(target_path): + self.log.critical( + "%s does not appear to be an iTunes backup folder. " + "Expected Manifest.db and Info.plist.", + target_path, + ) + return False + + candidates = [] + for entry_name in sorted(os.listdir(target_path)): + entry_path = os.path.join(target_path, entry_name) + if os.path.isdir(entry_path) and is_ios_backup_folder(entry_path): + candidates.append(entry_path) + + if len(candidates) == 1: + self.log.info("Found iTunes backup in subfolder: %s", candidates[0]) + self.target_path = candidates[0] + return True + + if candidates: + self.log.critical( + "Found multiple iTunes backups in %s. Please specify one backup folder.", + target_path, + ) + return False + + self.log.critical( + "%s does not appear to be an iTunes backup folder. " + "Expected Manifest.db and Info.plist.", + target_path, + ) + return False + def module_init(self, module): module.is_backup = True diff --git a/tests/test_check_ios_backup.py b/tests/test_check_ios_backup.py index a8ad6b2..eb10b13 100644 --- a/tests/test_check_ios_backup.py +++ b/tests/test_check_ios_backup.py @@ -3,6 +3,8 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +import shutil + from click.testing import CliRunner from mvt.ios.cli import check_backup @@ -16,3 +18,18 @@ class TestCheckBackupCommand: path = get_ios_backup_folder() result = runner.invoke(check_backup, [path]) assert result.exit_code == 0 + + def test_check_finds_backup_in_subfolder(self, tmp_path, caplog): + runner = CliRunner() + backup_path = tmp_path / "MobileSync" / "Backup" / "device-id" + shutil.copytree(get_ios_backup_folder(), backup_path) + + result = runner.invoke(check_backup, [str(backup_path.parent)]) + assert result.exit_code == 0 + assert f"Found iTunes backup in subfolder: {backup_path}" in caplog.text + + def test_check_rejects_non_backup_folder(self, tmp_path, caplog): + runner = CliRunner() + result = runner.invoke(check_backup, [str(tmp_path)]) + assert result.exit_code == 1 + assert "does not appear to be an iTunes backup folder" in caplog.text diff --git a/tests/test_custom_modules.py b/tests/test_custom_modules.py index dcbbbc9..a00faea 100644 --- a/tests/test_custom_modules.py +++ b/tests/test_custom_modules.py @@ -63,6 +63,8 @@ def test_load_module_appears_only_for_supported_cli_command(tmp_path): def test_module_option_runs_supported_custom_module(tmp_path): + (tmp_path / "Manifest.db").touch() + (tmp_path / "Info.plist").touch() module_path = _write_custom_module( tmp_path / "custom.py", "CustomRunModule", From 516ba06cf7a26581cee6980513238ce9c710cce7 Mon Sep 17 00:00:00 2001 From: besendorf Date: Tue, 14 Jul 2026 23:20:53 +0200 Subject: [PATCH 02/12] Suppress benign PinStorage key generation warning (#835) --- .../modules/intrusion_logs/security_event.py | 16 ++++++- tests/android/test_intrusion_logs.py | 46 +++++++++++++++++++ 2 files changed, 60 insertions(+), 2 deletions(-) diff --git a/src/mvt/android/modules/intrusion_logs/security_event.py b/src/mvt/android/modules/intrusion_logs/security_event.py index 1415ce8..c2dea5d 100644 --- a/src/mvt/android/modules/intrusion_logs/security_event.py +++ b/src/mvt/android/modules/intrusion_logs/security_event.py @@ -270,6 +270,11 @@ SECURITY_EVENT_METADATA_KEYS = { "timestamp", } +# Known platform failures that remain in the timeline but do not need a warning. +KNOWN_BENIGN_KEY_GENERATION_FAILURES = { + ("PinStorage_crossReboot_key", 1001), +} + class SecurityEvent(IntrusionLogsModule): """This module analyzes security events from intrusion logs.""" @@ -400,10 +405,17 @@ class SecurityEvent(IntrusionLogsModule): independent of any loaded indicators.""" # Flag failed cryptographic operations as potentially suspicious if "key_generated" in result: - if not result["key_generated"].get("success", True): + key_info = result["key_generated"] + key_id = key_info.get("key_id", "unknown") + uid = key_info.get("uid") + is_known_benign_failure = ( + key_id, + uid, + ) in KNOWN_BENIGN_KEY_GENERATION_FAILURES + if not key_info.get("success", True) and not is_known_benign_failure: self.log.warning( "Failed key generation detected for key_id: %s", - result["key_generated"].get("key_id", "unknown"), + key_id, ) # Flag certificate validation failures diff --git a/tests/android/test_intrusion_logs.py b/tests/android/test_intrusion_logs.py index ae31d78..2e68397 100644 --- a/tests/android/test_intrusion_logs.py +++ b/tests/android/test_intrusion_logs.py @@ -215,6 +215,52 @@ def _run_security_heuristics(results): return module.alertstore.alerts +@pytest.mark.parametrize("success", [False, 0]) +def test_known_pinstorage_key_generation_failure_does_not_warn(success, caplog): + record = { + "timestamp": "2026-06-17 15:31:02.014", + "key_generated": { + "success": success, + "key_id": "PinStorage_crossReboot_key", + "uid": 1001, + }, + } + + with caplog.at_level(logging.WARNING): + _run_security_heuristics([record]) + + assert "Failed key generation detected" not in caplog.text + + timeline_event = SecurityEvent().serialize(record) + assert timeline_event["event"] == "key_generated" + assert "Key generation failed: PinStorage_crossReboot_key" in timeline_event["data"] + + +@pytest.mark.parametrize( + ("key_id", "uid"), + [ + ("PinStorage_crossReboot_key", 10_000), + ("another_key", 1001), + ], +) +def test_other_key_generation_failures_still_warn(key_id, uid, caplog): + with caplog.at_level(logging.WARNING): + _run_security_heuristics( + [ + { + "timestamp": "2026-06-17 15:31:02.014", + "key_generated": { + "success": False, + "key_id": key_id, + "uid": uid, + }, + } + ] + ) + + assert f"Failed key generation detected for key_id: {key_id}" in caplog.text + + def test_cert_authority_installed_raises_medium_alert_without_indicators(): alerts = _run_security_heuristics( [ From c806fd8d616af235d4890ec0c49bd0eeb14b2fa5 Mon Sep 17 00:00:00 2001 From: besendorf Date: Tue, 14 Jul 2026 23:27:44 +0200 Subject: [PATCH 03/12] Support rotated iOS shutdown logs (#834) --- src/mvt/ios/modules/fs/shutdownlog.py | 21 +++++++--- tests/conftest.py | 2 + tests/ios_fs/test_shutdownlog.py | 59 +++++++++++++++++++++++++++ 3 files changed, 76 insertions(+), 6 deletions(-) create mode 100644 tests/ios_fs/test_shutdownlog.py diff --git a/src/mvt/ios/modules/fs/shutdownlog.py b/src/mvt/ios/modules/fs/shutdownlog.py index 6c1fca7..3136177 100644 --- a/src/mvt/ios/modules/fs/shutdownlog.py +++ b/src/mvt/ios/modules/fs/shutdownlog.py @@ -6,6 +6,7 @@ import logging from typing import Optional +from mvt.common.module import DatabaseNotFoundError from mvt.common.module_types import ( ModuleAtomicResult, ModuleResults, @@ -17,6 +18,7 @@ from ..base import IOSExtraction SHUTDOWN_LOG_PATH = [ "private/var/db/diagnostics/shutdown.log", + "private/var/db/diagnostics/shutdown.*.log", ] @@ -132,10 +134,17 @@ class ShutdownLog(IOSExtraction): self.results = sorted(self.results, key=lambda entry: entry["isodate"]) def run(self) -> None: - self._find_ios_database(root_paths=SHUTDOWN_LOG_PATH) - self.log.info("Found shutdown log at path: %s", self.file_path) + if self.file_path: + shutdown_log_paths = [self.file_path] + else: + shutdown_log_paths = sorted( + self._get_fs_files_from_patterns(SHUTDOWN_LOG_PATH) + ) - if not self.file_path: - return - with open(self.file_path, "r", encoding="utf-8") as handle: - self.process_shutdownlog(handle.read()) + if not shutdown_log_paths: + raise DatabaseNotFoundError("unable to find any shutdown log files") + + for shutdown_log_path in shutdown_log_paths: + self.log.info("Found shutdown log at path: %s", shutdown_log_path) + with open(shutdown_log_path, "r", encoding="utf-8") as handle: + self.process_shutdownlog(handle.read()) diff --git a/tests/conftest.py b/tests/conftest.py index fb5cabf..c89f629 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -35,6 +35,7 @@ def indicators_factory(indicator_file): domains=[], emails=[], file_names=[], + file_paths=[], processes=[], app_ids=[], app_cert_hashes=[], @@ -47,6 +48,7 @@ def indicators_factory(indicator_file): ind.ioc_collections[0]["domains"].extend(domains) ind.ioc_collections[0]["emails"].extend(emails) ind.ioc_collections[0]["file_names"].extend(file_names) + ind.ioc_collections[0]["file_paths"].extend(file_paths) ind.ioc_collections[0]["processes"].extend(processes) ind.ioc_collections[0]["app_ids"].extend(app_ids) ind.ioc_collections[0]["android_property_names"].extend(android_property_names) diff --git a/tests/ios_fs/test_shutdownlog.py b/tests/ios_fs/test_shutdownlog.py new file mode 100644 index 0000000..e3a5016 --- /dev/null +++ b/tests/ios_fs/test_shutdownlog.py @@ -0,0 +1,59 @@ +# 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/ + +from mvt.common.module import run_module +from mvt.ios.modules.fs.shutdownlog import ShutdownLog + + +def _shutdown_log_entry(pid: int, client: str, timestamp: int) -> str: + return ( + f"remaining client pid: {pid} ({client})\n" + f"SIGTERM: [{timestamp}]\n" + ) + + +class TestShutdownLog: + def test_discovers_rotated_shutdown_logs(self, tmp_path): + diagnostics_path = tmp_path / "private/var/db/diagnostics" + diagnostics_path.mkdir(parents=True) + (diagnostics_path / "shutdown.log").write_text( + _shutdown_log_entry(100, "/usr/libexec/first", 1_700_000_000), + encoding="utf-8", + ) + (diagnostics_path / "shutdown.0.log").write_text( + _shutdown_log_entry(200, "/usr/libexec/second", 1_700_000_001), + encoding="utf-8", + ) + + module = ShutdownLog(target_path=str(tmp_path)) + run_module(module) + + assert {result["client"] for result in module.results} == { + "/usr/libexec/first", + "/usr/libexec/second", + } + + def test_file_path_indicator_matches_client_with_trailing_uuid( + self, indicators_factory + ): + executable_path = "/usr/sbin/filecoordinationd" + client = f"{executable_path}/123e4567-e89b-12d3-a456-426614174000" + module = ShutdownLog( + results=[ + { + "isodate": "2023-11-14 22:13:20.000000", + "pid": "100", + "client": client, + "delay": 0.0, + "times_delayed": 0, + } + ] + ) + module.indicators = indicators_factory(file_paths=[executable_path]) + + module.check_indicators() + + assert len(module.alertstore.alerts) == 1 + assert module.alertstore.alerts[0].matched_indicator.value == executable_path From 911115b0c8e7769e7f4bcaf505012cb2380ef9b5 Mon Sep 17 00:00:00 2001 From: besendorf Date: Wed, 15 Jul 2026 08:32:13 +0200 Subject: [PATCH 04/12] Match receiver indicators by package ID (#836) --- .../modules/bugreport/dumpsys_receivers.py | 18 ------------- src/mvt/common/indicators.py | 23 ---------------- tests/android_bugreport/test_bugreport.py | 26 +++++++++++++++++++ 3 files changed, 26 insertions(+), 41 deletions(-) diff --git a/src/mvt/android/modules/bugreport/dumpsys_receivers.py b/src/mvt/android/modules/bugreport/dumpsys_receivers.py index ded9069..8907d5e 100644 --- a/src/mvt/android/modules/bugreport/dumpsys_receivers.py +++ b/src/mvt/android/modules/bugreport/dumpsys_receivers.py @@ -35,24 +35,6 @@ class DumpsysReceivers(DumpsysReceiversArtifact, BugReportModule): self.results = results if results else {} - def check_indicators(self) -> None: - for result in self.results: - if self.indicators: - receiver_name = self.results[result][0]["receiver"] - - # return IoC if the stix2 process name a substring of the receiver name - ioc_match = self.indicators.check_receiver_prefix(receiver_name) - if ioc_match: - self.alertstore.critical( - ioc_match.message, - "", - self.results[result][0], - matched_indicator=ioc_match.ioc, - ) - continue - - - def run(self) -> None: content = self._get_dumpstate_file() if not content: diff --git a/src/mvt/common/indicators.py b/src/mvt/common/indicators.py index 39334a6..215c1ef 100644 --- a/src/mvt/common/indicators.py +++ b/src/mvt/common/indicators.py @@ -727,29 +727,6 @@ class Indicators: return None - def check_receiver_prefix( - self, receiver_name: str - ) -> Optional[IndicatorMatch]: - """Check the provided receiver name against the list of indicators. - An IoC match is detected when a substring of the receiver matches the indicator. - - :param receiver_name: Receiver name to check against app ID indicators - :type receiver_name: str - :returns: IndicatorMatch if matched, otherwise None - - """ - if not receiver_name: - return None - - for ioc in self.get_iocs("app_ids"): - if ioc.value.lower() in receiver_name.lower(): - return IndicatorMatch( - ioc=ioc, - message=f'Found a known suspicious receiver with name "{receiver_name}" matching indicators from "{ioc.name}"', - ) - - return None - def check_android_property_name( self, property_name: str ) -> Optional[IndicatorMatch]: diff --git a/tests/android_bugreport/test_bugreport.py b/tests/android_bugreport/test_bugreport.py index a2c5e37..2bfaef5 100644 --- a/tests/android_bugreport/test_bugreport.py +++ b/tests/android_bugreport/test_bugreport.py @@ -9,6 +9,7 @@ from pathlib import Path from mvt.android.modules.bugreport.dumpsys_appops import DumpsysAppops from mvt.android.modules.bugreport.dumpsys_getprop import DumpsysGetProp from mvt.android.modules.bugreport.dumpsys_packages import DumpsysPackages +from mvt.android.modules.bugreport.dumpsys_receivers import DumpsysReceivers from mvt.android.modules.bugreport.tombstones import Tombstones from mvt.common.module import run_module @@ -60,6 +61,31 @@ class TestBugreportAnalysis: m = self.launch_bug_report_module(DumpsysGetProp) assert len(m.results) == 0 + def test_receivers_match_exact_package_name(self, indicators_factory): + intent = "android.intent.action.PHONE_STATE" + false_positive = { + "package_name": "com.android.phone", + "receiver": ( + "com.android.phone/" + "com.android.services.telephony.sip.SipIncomingCallReceiver" + ), + } + malicious_receiver = { + "package_name": "com.android.services", + "receiver": "com.android.services/com.example.SomeReceiver", + } + module = DumpsysReceivers( + results={intent: [false_positive, malicious_receiver]} + ) + module.indicators = indicators_factory(app_ids=["com.android.services"]) + + module.check_indicators() + + assert len(module.alertstore.alerts) == 1 + alert = module.alertstore.alerts[0] + assert alert.event == {intent: malicious_receiver} + assert alert.matched_indicator.value == "com.android.services" + def test_tombstones_modules(self): m = self.launch_bug_report_module(Tombstones) assert len(m.results) == 2 From afcfda47202f0dc65bc029fe5fe0ce1a30b29481 Mon Sep 17 00:00:00 2001 From: besendorf Date: Wed, 15 Jul 2026 09:09:04 +0200 Subject: [PATCH 05/12] Deduplicate AndroidQF SMS analysis (#837) --- src/mvt/android/modules/androidqf/__init__.py | 2 - src/mvt/android/modules/androidqf/sms.py | 105 ------------- src/mvt/android/modules/backup/helpers.py | 5 +- src/mvt/common/password.py | 143 ++++++++++++++++++ src/mvt/ios/cli.py | 7 +- tests/android_androidqf/test_sms.py | 91 ----------- tests/common/test_password.py | 28 ++++ tests/test_check_android_androidqf.py | 14 +- tests/test_check_android_backup.py | 9 +- 9 files changed, 191 insertions(+), 213 deletions(-) delete mode 100644 src/mvt/android/modules/androidqf/sms.py create mode 100644 src/mvt/common/password.py delete mode 100644 tests/android_androidqf/test_sms.py create mode 100644 tests/common/test_password.py diff --git a/src/mvt/android/modules/androidqf/__init__.py b/src/mvt/android/modules/androidqf/__init__.py index 9009f52..1803949 100644 --- a/src/mvt/android/modules/androidqf/__init__.py +++ b/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, ] diff --git a/src/mvt/android/modules/androidqf/sms.py b/src/mvt/android/modules/androidqf/sms.py deleted file mode 100644 index 64aaab6..0000000 --- a/src/mvt/android/modules/androidqf/sms.py +++ /dev/null @@ -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)) diff --git a/src/mvt/android/modules/backup/helpers.py b/src/mvt/android/modules/backup/helpers.py index 3e48078..4c16778 100644 --- a/src/mvt/android/modules/backup/helpers.py +++ b/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" diff --git a/src/mvt/common/password.py b/src/mvt/common/password.py new file mode 100644 index 0000000..b59ae49 --- /dev/null +++ b/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 diff --git a/src/mvt/ios/cli.py b/src/mvt/ios/cli.py index e15f8db..3f4beef 100644 --- a/src/mvt/ios/cli.py +++ b/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() diff --git a/tests/android_androidqf/test_sms.py b/tests/android_androidqf/test_sms.py deleted file mode 100644 index 116d5b2..0000000 --- a/tests/android_androidqf/test_sms.py +++ /dev/null @@ -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 - ) diff --git a/tests/common/test_password.py b/tests/common/test_password.py new file mode 100644 index 0000000..32845fb --- /dev/null +++ b/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() == "**" diff --git a/tests/test_check_android_androidqf.py b/tests/test_check_android_androidqf.py index c4c3410..a2ed952 100644 --- a/tests/test_check_android_androidqf.py +++ b/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 diff --git a/tests/test_check_android_backup.py b/tests/test_check_android_backup.py index 71c0586..ac0680f 100644 --- a/tests/test_check_android_backup.py +++ b/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 From 5ed8b3c1a5b0aaca925d590552852e756a22c65a Mon Sep 17 00:00:00 2001 From: Felix <24938145+Nxtmaster10@users.noreply.github.com> Date: Fri, 17 Jul 2026 18:25:47 +0200 Subject: [PATCH 06/12] fix: terminate dumpsys adb multiline values at structural lines (#842) * fix: terminate dumpsys adb multiline values at structural lines * fix dumpsys ADB multiline boundaries --------- Co-authored-by: Janik Besendorf --- src/mvt/android/artifacts/dumpsys_adb.py | 21 +++++- tests/android/test_artifact_dumpsys_adb.py | 60 ++++++++++++++++++ .../android_data/dumpsys_adb_wifi.txt | Bin 0 -> 2037 bytes 3 files changed, 78 insertions(+), 3 deletions(-) create mode 100644 tests/artifacts/android_data/dumpsys_adb_wifi.txt diff --git a/src/mvt/android/artifacts/dumpsys_adb.py b/src/mvt/android/artifacts/dumpsys_adb.py index d2a33fb..e5438c9 100644 --- a/src/mvt/android/artifacts/dumpsys_adb.py +++ b/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() diff --git a/tests/android/test_artifact_dumpsys_adb.py b/tests/android/test_artifact_dumpsys_adb.py index 1d3d76b..47dc3df 100644 --- a/tests/android/test_artifact_dumpsys_adb.py +++ b/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") diff --git a/tests/artifacts/android_data/dumpsys_adb_wifi.txt b/tests/artifacts/android_data/dumpsys_adb_wifi.txt new file mode 100644 index 0000000000000000000000000000000000000000..ecd7f159c378d80eb525055a5dc35632d44033ac GIT binary patch literal 2037 zcmeH|*^c8x6oxZ{1Sl5>-oTe2)5L3HTZ%N^5+_dL@t#CN!Ap6IV<+C-s2>gSNW210 zx*6b}D>S82sq~*kRnqzOJw5%WYb==CkC3u}bWyWbya+kI%ZyopgSPCBU5Xl{8>}Hh z-~09DTK45V{sS}qbqqm1aO*IPv+#6rGA9JZ{H{WfmuDI9D&U{0{4}J`tD**vNa~c; z=?wZ)0Xz^lz@IM=hD8}%U?`s9aArX;Gmb%MhF&nD$WR<3Pz+8nn8Jk+9 zF6CThhxO6qN-IvJp(T)US2XI{&&U<5!aoIKbjY@+_`;`C}B zkwF+``ihT=5HY)*l6ZW#J&?KMR%B|g4n(wuW3-8H-VpcJ3%0R|37ufAcsec!hv08Fm2Q;e+v9Ot|v9?!2u4a zimGOs)6kJKJwCUdS)243mfD-hXO`p}=eAHAzv{BEjupA?cd%9@OAbDo%GB`$u+$0; z!-|GYFf}$NslP*7cON?~h7G@==_MNb#4g^g+r*IjRKl|wH@l_>jagwOrn(o2NJ!qW zQl_9xogno?eql%jGPl!E4>_0)HJ(3_R(hlCZqJvET~nu*|Q7 zQP}3Kg%{mNeaj>Mes7uHwz=wAj6~fgWT})~t_{ShDIw|i%6@$fRXQV5Kakih)$hC6 z&alquF-}>Zbz7t8x3!*%JI^w5bY+=ZiNjYsZIsLGZ0e$z5w?KSm$23vN+l0djf^&d zBr7zLOH-+FLx7OvRo{dJMcYh^)Ik-#){p2m>i=pQIaQ|$N&-VBq*yDAohjaAJ9U%NG zULF>BH>kgKhoAA$dwewd?%|`s3#T>?;x`&W5Xc-i9as}MY5B?_2+Ba!=EeEsmSybE zA9a6=|J360G~lbe)9kJ0Uwx{yJYDmmnj?=MFT?pVei?s4G=%(wAZQ7-P6<^Jfc&~} ns^RHS9t`(SME!Aiq(6`paw-PG5R7<2@Fxs~P?SMt42Jy)NcfnA literal 0 HcmV?d00001 From 123c9081eda5cb70cc0b8d7a1b70b12962ff1540 Mon Sep 17 00:00:00 2001 From: besendorf Date: Sun, 19 Jul 2026 17:13:57 +0200 Subject: [PATCH 07/12] Skip resolving Google Maps short URLs (#843) --- src/mvt/common/url.py | 5 +++++ tests/common/test_indicators.py | 8 ++++++++ tests/common/test_url.py | 24 ++++++++++++++++++++++++ 3 files changed, 37 insertions(+) create mode 100644 tests/common/test_url.py diff --git a/src/mvt/common/url.py b/src/mvt/common/url.py index 95de5c7..426d64b 100644 --- a/src/mvt/common/url.py +++ b/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 diff --git a/tests/common/test_indicators.py b/tests/common/test_indicators.py index 00c7276..df6a28e 100644 --- a/tests/common/test_indicators.py +++ b/tests/common/test_indicators.py @@ -80,6 +80,14 @@ 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_file_hash(self, indicator_file): ind = Indicators(log=logging) ind.load_indicators_files([indicator_file], load_default=False) diff --git a/tests/common/test_url.py b/tests/common/test_url.py new file mode 100644 index 0000000..ce0fb63 --- /dev/null +++ b/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 From a6c3a805d82886dafdb413d235a455e14be63db4 Mon Sep 17 00:00:00 2001 From: besendorf Date: Mon, 27 Jul 2026 15:03:02 +0200 Subject: [PATCH 08/12] Parallelize URL indicator checks (#844) --- docs/iocs.md | 7 +- src/mvt/android/modules/backup/sms.py | 10 +- src/mvt/common/indicators.py | 44 ++++++- src/mvt/ios/modules/mixed/shortcuts.py | 6 +- src/mvt/ios/modules/mixed/sms.py | 7 +- .../mixed/webkit_session_resource_log.py | 56 +++++---- src/mvt/ios/modules/mixed/whatsapp.py | 6 +- tests/common/test_indicators.py | 115 ++++++++++++++++++ tests/ios_backup/test_sms.py | 28 +++++ 9 files changed, 239 insertions(+), 40 deletions(-) diff --git a/docs/iocs.md b/docs/iocs.md index be9ca4e..c7586de 100644 --- a/docs/iocs.md +++ b/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. - diff --git a/src/mvt/android/modules/backup/sms.py b/src/mvt/android/modules/backup/sms.py index 6f8306a..c42e9a4 100644 --- a/src/mvt/android/modules/backup/sms.py +++ b/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" diff --git a/src/mvt/common/indicators.py b/src/mvt/common/indicators.py index 215c1ef..7186ae3 100644 --- a/src/mvt/common/indicators.py +++ b/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 diff --git a/src/mvt/ios/modules/mixed/shortcuts.py b/src/mvt/ios/modules/mixed/shortcuts.py index f8899a4..eb749ee 100644 --- a/src/mvt/ios/modules/mixed/shortcuts.py +++ b/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 diff --git a/src/mvt/ios/modules/mixed/sms.py b/src/mvt/ios/modules/mixed/sms.py index 76e0048..596ba47 100644 --- a/src/mvt/ios/modules/mixed/sms.py +++ b/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 diff --git a/src/mvt/ios/modules/mixed/webkit_session_resource_log.py b/src/mvt/ios/modules/mixed/webkit_session_resource_log.py index d8b9a85..eb34311 100644 --- a/src/mvt/ios/modules/mixed/webkit_session_resource_log.py +++ b/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 = [] diff --git a/src/mvt/ios/modules/mixed/whatsapp.py b/src/mvt/ios/modules/mixed/whatsapp.py index 2a72c6f..cf3a49b 100644 --- a/src/mvt/ios/modules/mixed/whatsapp.py +++ b/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 diff --git a/tests/common/test_indicators.py b/tests/common/test_indicators.py index df6a28e..bbb55b2 100644 --- a/tests/common/test_indicators.py +++ b/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 @@ -88,6 +90,119 @@ class TestIndicators: 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) diff --git a/tests/ios_backup/test_sms.py b/tests/ios_backup/test_sms.py index 03e4606..5081a67 100644 --- a/tests/ios_backup/test_sms.py +++ b/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] From 84df51c518d987852700ccc0c080de6db4ac55f4 Mon Sep 17 00:00:00 2001 From: Volodymyr Styran Date: Mon, 27 Jul 2026 16:06:44 +0300 Subject: [PATCH 09/12] Scan Safari profile databases for history and browser state (#846) * fix(ios): scan Safari profile databases for history and browser state Safari profiles (iOS 17 and later) keep their own databases under Library/Safari/Profiles//, but SafariHistory and SafariBrowserState only ever looked at the default profile's Library/Safari/History.db and Library/Safari/BrowserState.db. On a device where browsing happens inside a profile, MVT silently skipped that history and still reported no detections, so an indicator only ever visited within a profile went unnoticed. Both modules now also match Library/Safari/Profiles/*/ in backups and in filesystem dumps. No helper changes were needed: the Manifest.db lookup already translates "*" into a SQL LIKE wildcard, and the filesystem lookup already globs. Found while examining an encrypted iOS 26.5.2 backup that contained 14 per-profile History.db files under AppDomain-com.apple.mobilesafari::Library/Safari/Profiles//. Co-Authored-By: Claude Opus 5 * fix(ios): scope Safari redirects to history database --------- Co-authored-by: Claude Opus 5 Co-authored-by: Janik Besendorf --- .../ios/modules/mixed/safari_browserstate.py | 35 ++-- src/mvt/ios/modules/mixed/safari_history.py | 33 +++- tests/ios_backup/test_safari_browserstate.py | 43 ++++- tests/ios_backup/test_safari_history.py | 165 ++++++++++++++++++ tests/utils.py | 18 ++ 5 files changed, 271 insertions(+), 23 deletions(-) create mode 100644 tests/ios_backup/test_safari_history.py diff --git a/src/mvt/ios/modules/mixed/safari_browserstate.py b/src/mvt/ios/modules/mixed/safari_browserstate.py index 48998f1..e67b7d3 100644 --- a/src/mvt/ios/modules/mixed/safari_browserstate.py +++ b/src/mvt/ios/modules/mixed/safari_browserstate.py @@ -19,10 +19,17 @@ from mvt.common.utils import convert_mactime_to_iso, keys_bytes_to_string from ..base import IOSExtraction -SAFARI_BROWSER_STATE_BACKUP_RELPATH = "Library/Safari/BrowserState.db" +# Safari profiles (iOS 17 and later) keep their per-profile databases under +# Library/Safari/Profiles//, separate from the default profile's. +SAFARI_BROWSER_STATE_BACKUP_RELPATHS = [ + "Library/Safari/BrowserState.db", + "Library/Safari/Profiles/*/BrowserState.db", +] SAFARI_BROWSER_STATE_ROOT_PATHS = [ "private/var/mobile/Library/Safari/BrowserState.db", + "private/var/mobile/Library/Safari/Profiles/*/BrowserState.db", "private/var/mobile/Containers/Data/Application/*/Library/Safari/BrowserState.db", + "private/var/mobile/Containers/Data/Application/*/Library/Safari/Profiles/*/BrowserState.db", ] @@ -174,20 +181,22 @@ class SafariBrowserState(IOSExtraction): def run(self) -> None: if self.is_backup: - for backup_file in self._get_backup_files_from_manifest( - relative_path=SAFARI_BROWSER_STATE_BACKUP_RELPATH - ): - browserstate_path = self._get_backup_file_from_id( - backup_file["file_id"] - ) + for relative_path in SAFARI_BROWSER_STATE_BACKUP_RELPATHS: + for backup_file in self._get_backup_files_from_manifest( + relative_path=relative_path + ): + browserstate_path = self._get_backup_file_from_id( + backup_file["file_id"] + ) - if not browserstate_path: - continue + if not browserstate_path: + continue - self.log.info( - "Found Safari browser state database at path: %s", browserstate_path - ) - self._process_browser_state_db(browserstate_path) + self.log.info( + "Found Safari browser state database at path: %s", + browserstate_path, + ) + self._process_browser_state_db(browserstate_path) elif self.is_fs_dump: for browserstate_path in self._get_fs_files_from_patterns( SAFARI_BROWSER_STATE_ROOT_PATHS diff --git a/src/mvt/ios/modules/mixed/safari_history.py b/src/mvt/ios/modules/mixed/safari_history.py index 9ba32ad..4a50f02 100644 --- a/src/mvt/ios/modules/mixed/safari_history.py +++ b/src/mvt/ios/modules/mixed/safari_history.py @@ -17,10 +17,17 @@ from mvt.common.utils import convert_mactime_to_datetime, convert_mactime_to_iso from ..base import IOSExtraction -SAFARI_HISTORY_BACKUP_RELPATH = "Library/Safari/History.db" +# Safari profiles (iOS 17 and later) each keep their own History.db under +# Library/Safari/Profiles//, separate from the default profile's database. +SAFARI_HISTORY_BACKUP_RELPATHS = [ + "Library/Safari/History.db", + "Library/Safari/Profiles/*/History.db", +] SAFARI_HISTORY_ROOT_PATHS = [ "private/var/mobile/Library/Safari/History.db", + "private/var/mobile/Library/Safari/Profiles/*/History.db", "private/var/mobile/Containers/Data/Application/*/Library/Safari/History.db", + "private/var/mobile/Containers/Data/Application/*/Library/Safari/Profiles/*/History.db", ] @@ -75,6 +82,9 @@ class SafariHistory(IOSExtraction): # We loop again through visits in order to find redirect record. for redirect in self.results: + if redirect["safari_history_db"] != result["safari_history_db"]: + continue + if redirect["visit_id"] != result["redirect_destination"]: continue @@ -159,17 +169,22 @@ class SafariHistory(IOSExtraction): def run(self) -> None: if self.is_backup: - for history_file in self._get_backup_files_from_manifest( - relative_path=SAFARI_HISTORY_BACKUP_RELPATH - ): - history_path = self._get_backup_file_from_id(history_file["file_id"]) + for relative_path in SAFARI_HISTORY_BACKUP_RELPATHS: + for history_file in self._get_backup_files_from_manifest( + relative_path=relative_path + ): + history_path = self._get_backup_file_from_id( + history_file["file_id"] + ) - if not history_path: - continue + if not history_path: + continue - self.log.info("Found Safari history database at path: %s", history_path) + self.log.info( + "Found Safari history database at path: %s", history_path + ) - self._process_history_db(history_path) + self._process_history_db(history_path) elif self.is_fs_dump: for history_path in self._get_fs_files_from_patterns( SAFARI_HISTORY_ROOT_PATHS diff --git a/tests/ios_backup/test_safari_browserstate.py b/tests/ios_backup/test_safari_browserstate.py index fd7e62e..fa2149d 100644 --- a/tests/ios_backup/test_safari_browserstate.py +++ b/tests/ios_backup/test_safari_browserstate.py @@ -4,12 +4,44 @@ # https://license.mvt.re/1.1/ import logging +import shutil + +import pytest from mvt.common.indicators import Indicators from mvt.common.module import run_module from mvt.ios.modules.mixed.safari_browserstate import SafariBrowserState -from ..utils import get_ios_backup_folder +from ..utils import add_backup_manifest_entry, get_ios_backup_folder + +# fileID of HomeDomain::Library/Safari/BrowserState.db in the test backup. +DEFAULT_BROWSER_STATE_FILE_ID = "3a47b0981ed7c10f3e2800aa66bac96a3b5db28e" +PROFILE_UUID = "00000000-0000-4000-A000-000000000001" +PROFILE_BROWSER_STATE_FILE_ID = "bb00000000000000000000000000000000000001" + + +@pytest.fixture +def backup_with_safari_profile(tmp_path): + """An iTunes backup where a Safari profile has its own browser state.""" + backup_path = tmp_path / "backup" + shutil.copytree(get_ios_backup_folder(), backup_path) + + profile_db = ( + backup_path / PROFILE_BROWSER_STATE_FILE_ID[:2] / PROFILE_BROWSER_STATE_FILE_ID + ) + profile_db.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile( + backup_path / DEFAULT_BROWSER_STATE_FILE_ID[:2] / DEFAULT_BROWSER_STATE_FILE_ID, + profile_db, + ) + add_backup_manifest_entry( + backup_path, + PROFILE_BROWSER_STATE_FILE_ID, + "AppDomain-com.apple.mobilesafari", + f"Library/Safari/Profiles/{PROFILE_UUID}/BrowserState.db", + ) + + return str(backup_path) class TestSafariBrowserStateModule: @@ -21,6 +53,15 @@ class TestSafariBrowserStateModule: assert len(m.timeline) == 1 assert len(m.alertstore.alerts) == 0 + def test_parsing_backup_with_profile(self, backup_with_safari_profile): + m = SafariBrowserState(target_path=backup_with_safari_profile) + m.is_backup = True + run_module(m) + + # Both the default profile and the named profile are extracted. + assert len(m.results) == 2 + assert len({result["safari_browser_state_db"] for result in m.results}) == 2 + def test_detection(self, indicator_file): m = SafariBrowserState(target_path=get_ios_backup_folder()) m.is_backup = True diff --git a/tests/ios_backup/test_safari_history.py b/tests/ios_backup/test_safari_history.py new file mode 100644 index 0000000..258b352 --- /dev/null +++ b/tests/ios_backup/test_safari_history.py @@ -0,0 +1,165 @@ +# 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 shutil +import sqlite3 +from pathlib import Path + +import pytest + +from mvt.common.indicators import Indicators +from mvt.common.module import run_module +from mvt.ios.modules.mixed.safari_history import SafariHistory + +from ..utils import add_backup_manifest_entry, get_ios_backup_folder + +# fileID of HomeDomain::Library/Safari/History.db in the test backup. +DEFAULT_HISTORY_FILE_ID = "1a0e7afc19d307da602ccdcece51af33afe92c53" +PROFILE_UUID = "00000000-0000-4000-A000-000000000001" +PROFILE_HISTORY_FILE_ID = "aa00000000000000000000000000000000000001" + +# example.org is already a test indicator, so use domains that do not match. +DEFAULT_URL = "https://default.example.net/visited-page" +PROFILE_URL = "https://profile.example.net/visited-page" + + +def create_history_db(path, url): + """Create a minimal Safari History.db holding a single visit.""" + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(path) + conn.executescript( + """ + CREATE TABLE history_items (id INTEGER PRIMARY KEY, url TEXT); + CREATE TABLE history_visits ( + id INTEGER PRIMARY KEY, + history_item INTEGER, + visit_time REAL, + redirect_source INTEGER, + redirect_destination INTEGER + ); + """ + ) + conn.execute("INSERT INTO history_items VALUES (1, ?);", (url,)) + conn.execute("INSERT INTO history_visits VALUES (1, 1, 726100000.0, NULL, NULL);") + conn.commit() + conn.close() + + +@pytest.fixture +def backup_with_safari_profile(tmp_path): + """An iTunes backup where Safari has both a default and a named profile.""" + backup_path = tmp_path / "backup" + shutil.copytree(get_ios_backup_folder(), backup_path) + + # The default profile's database ships empty, so give it a visit to make + # sure the profile lookup does not replace the pre-existing one. + create_history_db( + backup_path / DEFAULT_HISTORY_FILE_ID[:2] / DEFAULT_HISTORY_FILE_ID, + DEFAULT_URL, + ) + + create_history_db( + backup_path / PROFILE_HISTORY_FILE_ID[:2] / PROFILE_HISTORY_FILE_ID, + PROFILE_URL, + ) + add_backup_manifest_entry( + backup_path, + PROFILE_HISTORY_FILE_ID, + "AppDomain-com.apple.mobilesafari", + f"Library/Safari/Profiles/{PROFILE_UUID}/History.db", + ) + + return str(backup_path) + + +@pytest.fixture +def fs_dump_with_safari_profile(tmp_path): + """A filesystem dump where Safari has both a default and a named profile.""" + safari_path = tmp_path / "private" / "var" / "mobile" / "Library" / "Safari" + profile_path = safari_path / "Profiles" / PROFILE_UUID + profile_path.mkdir(parents=True) + + create_history_db(safari_path / "History.db", DEFAULT_URL) + create_history_db(profile_path / "History.db", PROFILE_URL) + + return str(tmp_path) + + +class TestSafariHistoryModule: + def test_parsing(self): + m = SafariHistory(target_path=get_ios_backup_folder()) + m.is_backup = True + run_module(m) + assert len(m.results) == 0 + assert len(m.alertstore.alerts) == 0 + + def test_parsing_backup_with_profile(self, backup_with_safari_profile): + m = SafariHistory(target_path=backup_with_safari_profile) + m.is_backup = True + run_module(m) + + # Both the default profile and the named profile are extracted. + assert len(m.results) == 2 + assert {result["url"] for result in m.results} == {DEFAULT_URL, PROFILE_URL} + assert len({result["safari_history_db"] for result in m.results}) == 2 + + def test_parsing_fs_dump_with_profile(self, fs_dump_with_safari_profile): + m = SafariHistory(target_path=fs_dump_with_safari_profile) + m.is_fs_dump = True + run_module(m) + + assert len(m.results) == 2 + assert {result["url"] for result in m.results} == {DEFAULT_URL, PROFILE_URL} + + def test_redirect_ids_are_scoped_to_database(self, fs_dump_with_safari_profile): + safari_path = ( + Path(fs_dump_with_safari_profile) + / "private" + / "var" + / "mobile" + / "Library" + / "Safari" + ) + with sqlite3.connect(safari_path / "History.db") as conn: + conn.execute( + "UPDATE history_items SET url = ? WHERE id = 1;", + ("http://safe.example.com/start",), + ) + conn.execute( + "INSERT INTO history_items VALUES (2, ?);", + ("https://safe.example.com/end",), + ) + conn.execute( + "UPDATE history_visits SET redirect_destination = 2 WHERE id = 1;" + ) + conn.execute( + "INSERT INTO history_visits VALUES (2, 2, 726100000.1, 1, NULL);" + ) + + profile_db = safari_path / "Profiles" / PROFILE_UUID / "History.db" + with sqlite3.connect(profile_db) as conn: + # Visit IDs are local to each database and commonly overlap. + conn.execute("UPDATE history_visits SET id = 2 WHERE id = 1;") + + m = SafariHistory(target_path=fs_dump_with_safari_profile) + m.is_fs_dump = True + run_module(m) + + assert len(m.results) == 3 + assert len(m.alertstore.alerts) == 0 + + def test_detection_in_profile(self, backup_with_safari_profile, indicator_file): + """An indicator only visited inside a Safari profile still alerts.""" + m = SafariHistory(target_path=backup_with_safari_profile) + m.is_backup = True + ind = Indicators(log=logging.getLogger()) + ind.parse_stix2(indicator_file) + ind.ioc_collections[0]["domains"].append("profile.example.net") + m.indicators = ind + run_module(m) + + assert len(m.alertstore.alerts) == 1 + assert m.alertstore.alerts[0].event["url"] == PROFILE_URL diff --git a/tests/utils.py b/tests/utils.py index b29e069..9ae433d 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -4,6 +4,7 @@ # https://license.mvt.re/1.1/ import os +import sqlite3 from pathlib import Path @@ -37,6 +38,23 @@ def get_indicator_file(): print("PYTEST env", os.getenv("PYTEST_CURRENT_TEST")) +def add_backup_manifest_entry(backup_path, file_id, domain, relative_path): + """ + Register an extra file in a test backup's Manifest.db + """ + conn = sqlite3.connect(os.path.join(backup_path, "Manifest.db")) + conn.execute( + "INSERT INTO Files (fileID, domain, relativePath, flags, file) " + "VALUES (?, ?, ?, 1, ?);", + (file_id, domain, relative_path, b""), + ) + conn.commit() + # MVT opens backup databases with immutable=1, which ignores any -wal file, + # so the new row has to be checkpointed into Manifest.db itself. + conn.execute("PRAGMA journal_mode=DELETE;") + conn.close() + + def delete_tmp_db_files(file_path): """ Remove Sqlite temporary files that appear on some platforms From 797411e1e50f53e1f4ad2e55ec1e2451dd354381 Mon Sep 17 00:00:00 2001 From: besendorf Date: Mon, 27 Jul 2026 17:58:34 +0200 Subject: [PATCH 10/12] Fix text tombstone crashing thread parsing (#848) --- .../android/artifacts/tombstone_crashes.py | 5 ++++ tests/android/test_artifact_tombstones.py | 26 +++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/src/mvt/android/artifacts/tombstone_crashes.py b/src/mvt/android/artifacts/tombstone_crashes.py index 9a878aa..d4888ef 100644 --- a/src/mvt/android/artifacts/tombstone_crashes.py +++ b/src/mvt/android/artifacts/tombstone_crashes.py @@ -206,6 +206,11 @@ class TombstoneCrashArtifact(AndroidArtifact): return True def _load_pid_line(self, line: str, tombstone: dict) -> bool: + # The first pid line identifies the crashing thread. Full text tombstones + # contain additional pid lines for the other threads in the process. + if "pid" in tombstone: + return True + try: parts = line.split(" >>> ") if " >>> " in line else line.split(">>>") process_info = parts[0] diff --git a/tests/android/test_artifact_tombstones.py b/tests/android/test_artifact_tombstones.py index d1d5682..e4e51cf 100644 --- a/tests/android/test_artifact_tombstones.py +++ b/tests/android/test_artifact_tombstones.py @@ -42,6 +42,32 @@ class TestTombstoneCrashArtifact: assert len(tombstone_artifact.results) == 1 self.validate_tombstone_result(tombstone_artifact.results[0]) + def test_text_tombstone_keeps_crashing_thread(self): + tombstone_artifact = TombstoneCrashArtifact() + artifact_path = "android_data/tombstone_process.txt" + file = get_artifact(artifact_path) + with open(file, "rb") as f: + data = f.read() + + data += ( + b"\npid: 25541, tid: 31896, name: worker-thread" + b" >>> /vendor/bin/other <<<\n" + ) + tombstone_artifact.parse( + os.path.basename(artifact_path), + datetime.datetime(2023, 4, 12, 12, 32, 40, 518290), + data, + ) + + result = tombstone_artifact.results[0] + assert result["pid"] == 25541 + assert result["tid"] == 21307 + assert result["process_name"] == "mtk.ape.decoder" + assert ( + result["binary_path"] + == "/vendor/bin/hw/android.hardware.media.c2@1.2-mediatek" + ) + @pytest.mark.skip(reason="Not implemented yet") def test_tombtone_kernel_parsing(self): tombstone_artifact = TombstoneCrashArtifact() From 3eff0c550de2cc5a9e016eaf207d3589e8d9c05b Mon Sep 17 00:00:00 2001 From: besendorf Date: Tue, 28 Jul 2026 18:34:13 +0200 Subject: [PATCH 11/12] Handle mis-indented dumpsys receiver actions (#852) --- .../android/artifacts/dumpsys_receivers.py | 14 ++++++- .../test_artifact_dumpsys_receivers.py | 38 +++++++++++++++++++ 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/src/mvt/android/artifacts/dumpsys_receivers.py b/src/mvt/android/artifacts/dumpsys_receivers.py index c7130ed..b437930 100644 --- a/src/mvt/android/artifacts/dumpsys_receivers.py +++ b/src/mvt/android/artifacts/dumpsys_receivers.py @@ -96,6 +96,18 @@ class DumpsysReceiversArtifact(AndroidArtifact): self.results[intent] = [] continue + parts = line.strip().split(" ") + if len(parts) < 2: + # A single-token line here is not a receiver. Real dumpstate + # output can print an action header mis-indented (observed with + # 15 leading spaces instead of 6), which used to raise + # IndexError and abort the whole module. Treat a trailing-colon + # token as the next action, skip anything else. + if parts[0].endswith(":"): + intent = parts[0][:-1] + self.results.setdefault(intent, []) + continue + # If we are not in an intent block yet, skip. if not intent: continue @@ -109,7 +121,7 @@ class DumpsysReceiversArtifact(AndroidArtifact): # If we got this far, we are processing receivers for the # activities we are interested in. - receiver = line.strip().split(" ")[1] + receiver = parts[1] package_name = receiver.split("/")[0] self.results[intent].append( diff --git a/tests/android/test_artifact_dumpsys_receivers.py b/tests/android/test_artifact_dumpsys_receivers.py index e4bed62..7875a52 100644 --- a/tests/android/test_artifact_dumpsys_receivers.py +++ b/tests/android/test_artifact_dumpsys_receivers.py @@ -31,6 +31,44 @@ class TestDumpsysReceiversArtifact: == "com.android.storagemanager" ) + def test_parsing_misindented_action(self): + dr = DumpsysReceiversArtifact() + data = """\ +Receiver Resolver Table: + Non-Data Actions: + android.app.action.ENTER_CAR_MODE: + b5b40f6 com.google.android.projection.gearhead/.CarModeBroadcastReceiver + android.intent.action.MY_PACKAGE_REPLACED: + e7706c1 com.psycatgames.nhiegame/.ScheduledNotificationBootReceiver +""" + + dr.parse(data) + + assert ( + dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][ + "package_name" + ] + == "com.psycatgames.nhiegame" + ) + + def test_parsing_misindented_first_action(self): + dr = DumpsysReceiversArtifact() + data = """\ +Receiver Resolver Table: + Non-Data Actions: + android.intent.action.MY_PACKAGE_REPLACED: + e7706c1 com.psycatgames.nhiegame/.ScheduledNotificationBootReceiver +""" + + dr.parse(data) + + assert ( + dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][ + "package_name" + ] + == "com.psycatgames.nhiegame" + ) + def test_ioc_check(self, indicator_file): dr = DumpsysReceiversArtifact() file = get_artifact("android_data/dumpsys_packages.txt") From 2dfe3cbcb19e40bf49ffa3b836e1696f9212327c Mon Sep 17 00:00:00 2001 From: besendorf Date: Tue, 28 Jul 2026 18:58:46 +0200 Subject: [PATCH 12/12] Fix module audit findings (#850) * Fix module audit findings * Always parse paired tombstones --- .../artifacts/dumpsys_battery_history.py | 29 +++++-- src/mvt/android/artifacts/dumpsys_dbinfo.py | 44 ++++------ src/mvt/android/cli.py | 6 +- src/mvt/android/cmd_check_androidqf.py | 13 ++- src/mvt/android/modules/androidqf/__init__.py | 2 + src/mvt/android/parsers/backup.py | 39 +++++---- src/mvt/ios/modules/base.py | 83 +++++++++++++----- src/mvt/ios/modules/fs/analytics.py | 2 + src/mvt/ios/modules/net_base.py | 7 +- .../test_artifact_dumpsys_battery_history.py | 19 +++++ tests/android/test_artifact_dumpsys_dbinfo.py | 19 +++++ tests/android/test_backup_parser.py | 11 +++ tests/ios_backup/test_calendar.py | 13 +++ tests/ios_backup/test_sqlite_handling.py | 85 +++++++++++++++++++ tests/test_check_android_androidqf.py | 30 +++++++ tests/test_check_android_bugreport.py | 10 +++ tests/utils.py | 3 +- 17 files changed, 333 insertions(+), 82 deletions(-) create mode 100644 tests/ios_backup/test_sqlite_handling.py diff --git a/src/mvt/android/artifacts/dumpsys_battery_history.py b/src/mvt/android/artifacts/dumpsys_battery_history.py index fb42903..77d9d3c 100644 --- a/src/mvt/android/artifacts/dumpsys_battery_history.py +++ b/src/mvt/android/artifacts/dumpsys_battery_history.py @@ -31,21 +31,38 @@ class DumpsysBatteryHistoryArtifact(AndroidArtifact): if line.strip() == "": break - time_elapsed = line.strip().split(" ", 1)[0] + time_parts = line.strip().split() + time_elapsed = time_parts[0] + if ( + len(time_parts) > 1 + and len(time_parts[0]) == 5 + and time_parts[0][2] == "-" + and ":" in time_parts[1] + ): + time_elapsed = " ".join(time_parts[:2]) event = "" if line.find("+job") > 0: event = "start_job" - uid = line[line.find("+job") + 5 : line.find(":")] - service = line[line.find(":") + 1 :].strip('"') + payload = line.split("+job=", 1)[1] + uid, separator, service = payload.partition(":") + if not separator: + continue + service = service.strip().strip('"') package_name = service.split("/")[0] elif line.find("-job") > 0: event = "end_job" - uid = line[line.find("-job") + 5 : line.find(":")] - service = line[line.find(":") + 1 :].strip('"') + payload = line.split("-job=", 1)[1] + uid, separator, service = payload.partition(":") + if not separator: + continue + service = service.strip().strip('"') package_name = service.split("/")[0] elif line.find("+running +wake_lock=") > 0: - uid = line[line.find("+running +wake_lock=") + 21 : line.find(":")] + payload = line.split("+running +wake_lock=", 1)[1] + uid, separator, _ = payload.partition(":") + if not separator: + continue event = "wake" service = ( line[line.find("*walarm*:") + 9 :].split(" ")[0].strip('"').strip() diff --git a/src/mvt/android/artifacts/dumpsys_dbinfo.py b/src/mvt/android/artifacts/dumpsys_dbinfo.py index a3a4eb1..050582d 100644 --- a/src/mvt/android/artifacts/dumpsys_dbinfo.py +++ b/src/mvt/android/artifacts/dumpsys_dbinfo.py @@ -29,10 +29,9 @@ class DumpsysDBInfoArtifact(AndroidArtifact): def parse(self, output: str) -> None: rxp = re.compile( - r".*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\].*\[Pid:\((\d+)\)\](\w+).*sql\=\"(.+?)\"" - ) # pylint: disable=line-too-long - rxp_no_pid = re.compile( - r".*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\][ ]{1}(\w+).*sql\=\"(.+?)\"" + r".*\[((?:[0-9]{4}-)?[0-9]{2}-[0-9]{2} " + r"[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\]\s*" + r"(?:\[Pid:\((\d+)\)\])?([\w-]+).*?sql=\"(.+?)\"" ) # pylint: disable=line-too-long pool = None @@ -56,29 +55,16 @@ class DumpsysDBInfoArtifact(AndroidArtifact): pool = None continue - matches = rxp.findall(line) - if not matches: - matches = rxp_no_pid.findall(line) - if not matches: - continue + match = rxp.match(line) + if not match: + continue - match = matches[0] - self.results.append( - { - "isodate": match[0], - "action": match[1], - "sql": match[2], - "path": pool, - } - ) - else: - match = matches[0] - self.results.append( - { - "isodate": match[0], - "pid": match[1], - "action": match[2], - "sql": match[3], - "path": pool, - } - ) + result = { + "isodate": match.group(1), + "action": match.group(3), + "sql": match.group(4), + "path": pool, + } + if match.group(2): + result["pid"] = match.group(2) + self.results.append(result) diff --git a/src/mvt/android/cli.py b/src/mvt/android/cli.py index 1123742..2830c98 100644 --- a/src/mvt/android/cli.py +++ b/src/mvt/android/cli.py @@ -4,6 +4,7 @@ # https://license.mvt.re/1.1/ import logging +from zipfile import BadZipFile import click @@ -212,7 +213,10 @@ def check_bugreport( log.info("Checking Android bug report at path: %s", bugreport_path) - cmd.run() + try: + cmd.run() + except BadZipFile as exc: + raise click.ClickException(f"Invalid bugreport archive: {exc}") from exc cmd.show_alerts_brief() cmd.show_support_message() diff --git a/src/mvt/android/cmd_check_androidqf.py b/src/mvt/android/cmd_check_androidqf.py index 41ad802..319b614 100644 --- a/src/mvt/android/cmd_check_androidqf.py +++ b/src/mvt/android/cmd_check_androidqf.py @@ -296,6 +296,7 @@ class CmdAndroidCheckAndroidQF(Command): elif self.__format == "zip" and self.__zip: temp_dir = tempfile.mkdtemp(prefix="mvt_intrusion_logs_") + temp_root = Path(temp_dir).resolve() for entry in intrusion_log_files: normalized = entry.replace("\\", "/") idx = normalized.find("intrusion_logs/") @@ -303,9 +304,15 @@ class CmdAndroidCheckAndroidQF(Command): if not relative or relative.endswith("/"): continue - target = os.path.join(temp_dir, relative) - os.makedirs(os.path.dirname(target), exist_ok=True) - with self.__zip.open(entry) as src, open(target, "wb") as dst: + target = (temp_root / relative).resolve() + if not target.is_relative_to(temp_root): + self.log.warning( + "Skipping unsafe intrusion log archive entry: %s", entry + ) + continue + + target.parent.mkdir(parents=True, exist_ok=True) + with self.__zip.open(entry) as src, target.open("wb") as dst: dst.write(src.read()) intrusion_logs_path = temp_dir diff --git a/src/mvt/android/modules/androidqf/__init__.py b/src/mvt/android/modules/androidqf/__init__.py index 1803949..1d8d619 100644 --- a/src/mvt/android/modules/androidqf/__init__.py +++ b/src/mvt/android/modules/androidqf/__init__.py @@ -5,6 +5,7 @@ from .aqf_files import AQFFiles from .aqf_getprop import AQFGetProp +from .aqf_log_timestamps import AQFLogTimestamps from .aqf_packages import AQFPackages from .aqf_processes import AQFProcesses from .aqf_settings import AQFSettings @@ -17,6 +18,7 @@ ANDROIDQF_MODULES = [ AQFGetProp, AQFSettings, AQFFiles, + AQFLogTimestamps, RootBinaries, Mounts, ] diff --git a/src/mvt/android/parsers/backup.py b/src/mvt/android/parsers/backup.py index 8c9bb5c..1f8da9b 100644 --- a/src/mvt/android/parsers/backup.py +++ b/src/mvt/android/parsers/backup.py @@ -131,20 +131,23 @@ def decrypt_backup_data(encrypted_backup, password, encryption_algo, format_vers if password is None: raise InvalidBackupPassword() - [ - user_salt, - checksum_salt, - pbkdf2_rounds, - user_iv, - master_key_blob, - encrypted_data, - ] = encrypted_backup.split(b"\n", 5) + try: + [ + user_salt, + checksum_salt, + pbkdf2_rounds, + user_iv, + master_key_blob, + encrypted_data, + ] = encrypted_backup.split(b"\n", 5) - user_salt = bytes.fromhex(user_salt.decode("utf-8")) - checksum_salt = bytes.fromhex(checksum_salt.decode("utf-8")) - pbkdf2_rounds = int(pbkdf2_rounds) - user_iv = bytes.fromhex(user_iv.decode("utf-8")) - master_key_blob = bytes.fromhex(master_key_blob.decode("utf-8")) + user_salt = bytes.fromhex(user_salt.decode("utf-8")) + checksum_salt = bytes.fromhex(checksum_salt.decode("utf-8")) + pbkdf2_rounds = int(pbkdf2_rounds) + user_iv = bytes.fromhex(user_iv.decode("utf-8")) + master_key_blob = bytes.fromhex(master_key_blob.decode("utf-8")) + except (UnicodeDecodeError, ValueError) as exc: + raise AndroidBackupParsingError("Invalid encrypted backup header") from exc # Derive decryption master key from password. master_key, master_iv = decrypt_master_key( @@ -174,10 +177,12 @@ def parse_backup_file(data, password=None): if not data.startswith(b"ANDROID BACKUP"): raise AndroidBackupParsingError("Invalid file header") - [_, version, is_compressed, encryption_algo, tar_data] = data.split(b"\n", 4) - - version = int(version) - is_compressed = int(is_compressed) + try: + [_, version, is_compressed, encryption_algo, tar_data] = data.split(b"\n", 4) + version = int(version) + is_compressed = int(is_compressed) + except ValueError as exc: + raise AndroidBackupParsingError("Invalid file header") from exc if encryption_algo != b"none": tar_data = decrypt_backup_data( diff --git a/src/mvt/ios/modules/base.py b/src/mvt/ios/modules/base.py index 121bad5..a5c3354 100644 --- a/src/mvt/ios/modules/base.py +++ b/src/mvt/ios/modules/base.py @@ -9,6 +9,7 @@ import os import shutil import sqlite3 import subprocess +import tempfile from pathlib import Path from typing import Iterator, Optional, Union @@ -20,6 +21,20 @@ from mvt.common.module import ( ) +class TemporarySQLiteConnection(sqlite3.Connection): + """SQLite connection that owns a temporary copy of a database.""" + + temporary_directory: Optional[tempfile.TemporaryDirectory] = None + + def close(self) -> None: + try: + super().close() + finally: + if self.temporary_directory: + self.temporary_directory.cleanup() + self.temporary_directory = None + + class IOSExtraction(MVTModule): """This class provides a base for all iOS filesystem/backup extraction modules.""" @@ -44,6 +59,8 @@ class IOSExtraction(MVTModule): self.is_backup = False self.is_fs_dump = False + self._recovered_sqlite_paths: dict[str, str] = {} + self._sqlite_temp_directories: list[tempfile.TemporaryDirectory] = [] def _recover_sqlite_db_if_needed( self, file_path: str, forced: bool = False @@ -53,17 +70,12 @@ class IOSExtraction(MVTModule): :param file_path: Path to the malformed database file. """ - # SQLite's immutable mode cannot open databases with active WAL files. if not forced: - # If the database is open, do not use immutable - if os.path.isfile(file_path + "-shm"): - conn = sqlite3.connect(file_path) - else: - conn = self._open_sqlite_db(file_path) + conn = self._open_sqlite_db(file_path) cur = conn.cursor() + recover = False try: - recover = False cur.execute("SELECT name FROM sqlite_master WHERE type='table';") except sqlite3.DatabaseError as exc: if "database disk image is malformed" in str(exc): @@ -82,27 +94,54 @@ class IOSExtraction(MVTModule): raise DatabaseCorruptedError( "failed to recover without sqlite3 binary: please install sqlite3!" ) - if '"' in file_path: - raise DatabaseCorruptedError( - f"database at path '{file_path}' is corrupted. unable to " - 'recover because it has a quotation mark (") in its name' - ) - - bak_path = f"{file_path}.bak" - shutil.move(file_path, bak_path) + temporary_directory = tempfile.TemporaryDirectory(prefix="mvt_sqlite_recover_") + temporary_path = Path(temporary_directory.name) + source_path = temporary_path / "source.db" + recovered_path = temporary_path / "recovered.db" + shutil.copy2(file_path, source_path) + for suffix in ("-wal", "-shm"): + sidecar = Path(file_path + suffix) + if sidecar.is_file(): + shutil.copy2(sidecar, Path(str(source_path) + suffix)) ret = subprocess.call( - ["sqlite3", bak_path, f'.clone "{file_path}"'], + ["sqlite3", str(source_path), f'.clone "{recovered_path}"'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) if ret != 0: + temporary_directory.cleanup() raise DatabaseCorruptedError("failed to recover database") + self._sqlite_temp_directories.append(temporary_directory) + self._recovered_sqlite_paths[file_path] = str(recovered_path) self.log.info("Database at path %s recovered successfully!", file_path) def _open_sqlite_db(self, file_path: str) -> sqlite3.Connection: - return sqlite3.connect(f"file:{file_path}?immutable=1", uri=True) + database_path = self._recovered_sqlite_paths.get(file_path, file_path) + if not os.path.isfile(database_path + "-wal"): + uri = Path(database_path).resolve().as_uri() + "?mode=ro&immutable=1" + return sqlite3.connect(uri, uri=True) + + temporary_directory = tempfile.TemporaryDirectory(prefix="mvt_sqlite_") + temporary_path = Path(temporary_directory.name) / Path(database_path).name + shutil.copy2(database_path, temporary_path) + for suffix in ("-wal", "-shm"): + sidecar = Path(database_path + suffix) + if sidecar.is_file(): + shutil.copy2(sidecar, Path(str(temporary_path) + suffix)) + + try: + conn = sqlite3.connect( + temporary_path.resolve().as_uri() + "?mode=ro", + uri=True, + factory=TemporarySQLiteConnection, + ) + except Exception: + temporary_directory.cleanup() + raise + conn.temporary_directory = temporary_directory + return conn def _get_backup_files_from_manifest( self, relative_path: Optional[str] = None, domain: Optional[str] = None @@ -166,7 +205,11 @@ class IOSExtraction(MVTModule): if not self.target_path: return None file_path = os.path.join(self.target_path, file_id[0:2], file_id) - if not Path(file_path).resolve().is_relative_to(Path(self.target_path).resolve()): + if ( + not Path(file_path) + .resolve() + .is_relative_to(Path(self.target_path).resolve()) + ): return None if os.path.exists(file_path): return file_path @@ -197,9 +240,9 @@ class IOSExtraction(MVTModule): :param backup_ids: Default value = None) """ - file_path: Optional[str] = None + file_path: Optional[str] = self.file_path # First we check if the was an explicit file path specified. - if not self.file_path: + if not file_path: # Type narrowing: we know self.file_path is None here, work with local file_path # If not, we first try with backups. # We construct the path to the file according to the iTunes backup diff --git a/src/mvt/ios/modules/fs/analytics.py b/src/mvt/ios/modules/fs/analytics.py index 0a6e50c..2bf29e4 100644 --- a/src/mvt/ios/modules/fs/analytics.py +++ b/src/mvt/ios/modules/fs/analytics.py @@ -134,6 +134,8 @@ class Analytics(IOSExtraction): isodate = "" data = plistlib.loads(row[1]) data["isodate"] = isodate + else: + continue data["artifact"] = artifact diff --git a/src/mvt/ios/modules/net_base.py b/src/mvt/ios/modules/net_base.py index 99a851d..eb13fdf 100644 --- a/src/mvt/ios/modules/net_base.py +++ b/src/mvt/ios/modules/net_base.py @@ -42,7 +42,8 @@ class NetBase(IOSExtraction): ) def _extract_net_data(self): - conn = sqlite3.connect(self.file_path) + assert self.file_path is not None + conn = self._open_sqlite_db(self.file_path) conn.row_factory = sqlite3.Row cur = conn.cursor() try: @@ -237,9 +238,7 @@ class NetBase(IOSExtraction): "been truncated in the database)" ) - self.alertstore.medium( - msg, proc["live_isodate"], proc - ) + self.alertstore.medium(msg, proc["live_isodate"], proc) if not proc["live_proc_id"]: self.log.info( diff --git a/tests/android/test_artifact_dumpsys_battery_history.py b/tests/android/test_artifact_dumpsys_battery_history.py index 03d7d2e..dd0b120 100644 --- a/tests/android/test_artifact_dumpsys_battery_history.py +++ b/tests/android/test_artifact_dumpsys_battery_history.py @@ -42,3 +42,22 @@ class TestDumpsysBatteryHistoryArtifact: assert len(dba.alertstore.alerts) == 0 dba.check_indicators() assert len(dba.alertstore.alerts) == 2 + + def test_parsing_absolute_timestamps(self): + dba = DumpsysBatteryHistoryArtifact() + dba.parse( + """07-15 20:27:39.431 (2) 100 +job=u0a123:"com.example/.ExampleJob" +07-15 20:27:40.431 (2) 100 -job=u0a123:"com.example/.ExampleJob" +""" + ) + + assert len(dba.results) == 2 + assert dba.results[0] == { + "time_elapsed": "07-15 20:27:39.431", + "event": "start_job", + "uid": "u0a123", + "package_name": "com.example", + "service": "com.example/.ExampleJob", + } + assert dba.results[1]["event"] == "end_job" + assert dba.results[1]["uid"] == "u0a123" diff --git a/tests/android/test_artifact_dumpsys_dbinfo.py b/tests/android/test_artifact_dumpsys_dbinfo.py index 2becf65..691d43c 100644 --- a/tests/android/test_artifact_dumpsys_dbinfo.py +++ b/tests/android/test_artifact_dumpsys_dbinfo.py @@ -40,3 +40,22 @@ class TestDumpsysDBinfoArtifact: assert len(dbi.alertstore.alerts) == 0 dbi.check_indicators() assert len(dbi.alertstore.alerts) == 5 + + def test_parsing_month_day_timestamp_without_pid(self): + dbi = DumpsysDBInfoArtifact() + dbi.parse( + """ +Connection pool for /data/user/0/com.example/databases/current.db: + Most recently executed operations: + 0: [07-15 20:27:39.431] executeForCursorWindow took 1ms - succeeded, sql="SELECT 1" +""" + ) + + assert dbi.results == [ + { + "isodate": "07-15 20:27:39.431", + "action": "executeForCursorWindow", + "sql": "SELECT 1", + "path": "/data/user/0/com.example/databases/current.db", + } + ] diff --git a/tests/android/test_backup_parser.py b/tests/android/test_backup_parser.py index 6e5be72..c0f867d 100644 --- a/tests/android/test_backup_parser.py +++ b/tests/android/test_backup_parser.py @@ -5,7 +5,10 @@ import hashlib +import pytest + from mvt.android.parsers.backup import ( + AndroidBackupParsingError, parse_ab_header, parse_backup_file, parse_tar_for_sms, @@ -23,6 +26,14 @@ class TestBackupParsing: "encryption": None, } + def test_parse_truncated_encrypted_header(self): + with pytest.raises( + AndroidBackupParsingError, match="Invalid encrypted backup header" + ): + parse_backup_file( + b"ANDROID BACKUP\n5\n0\nAES-256\ntruncated", password="password" + ) + def test_parsing_noencryption(self): file = get_artifact("android_backup/backup.ab") with open(file, "rb") as f: diff --git a/tests/ios_backup/test_calendar.py b/tests/ios_backup/test_calendar.py index bf931f4..5f1035d 100644 --- a/tests/ios_backup/test_calendar.py +++ b/tests/ios_backup/test_calendar.py @@ -4,6 +4,7 @@ # https://license.mvt.re/1.1/ import logging +import os from mvt.common.indicators import Indicators from mvt.common.module import run_module @@ -21,6 +22,18 @@ class TestCalendarModule: assert len(m.alertstore.alerts) == 0 assert m.results[0]["summary"] == "Super interesting meeting" + def test_calendar_with_explicit_file_path(self): + backup_path = get_ios_backup_folder() + database_path = os.path.join( + backup_path, "20", "2041457d5fe04d39d0ab481178355df6781e6858" + ) + m = Calendar(file_path=database_path) + + run_module(m) + + assert len(m.results) == 1 + assert m.results[0]["summary"] == "Super interesting meeting" + def test_calendar_detection(self, indicator_file): m = Calendar(target_path=get_ios_backup_folder()) ind = Indicators(log=logging.getLogger()) diff --git a/tests/ios_backup/test_sqlite_handling.py b/tests/ios_backup/test_sqlite_handling.py new file mode 100644 index 0000000..f780640 --- /dev/null +++ b/tests/ios_backup/test_sqlite_handling.py @@ -0,0 +1,85 @@ +# 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 hashlib +import os +import plistlib +import shutil +import sqlite3 + +from mvt.ios.modules.base import IOSExtraction +from mvt.ios.modules.fs.analytics import Analytics + + +def _sha256(path): + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def test_open_sqlite_reads_wal_without_modifying_evidence(tmp_path): + live_path = tmp_path / "live.db" + evidence_path = tmp_path / "evidence.db" + conn = sqlite3.connect(live_path) + conn.execute("PRAGMA journal_mode=WAL;") + conn.execute("PRAGMA wal_autocheckpoint=0;") + conn.execute("CREATE TABLE records (value TEXT);") + conn.commit() + conn.execute("INSERT INTO records VALUES ('from wal');") + conn.commit() + + shutil.copy2(live_path, evidence_path) + shutil.copy2(str(live_path) + "-wal", str(evidence_path) + "-wal") + conn.close() + + evidence_hash = _sha256(evidence_path) + wal_hash = _sha256(tmp_path / "evidence.db-wal") + module = IOSExtraction(file_path=str(evidence_path)) + read_conn = module._open_sqlite_db(str(evidence_path)) + rows = read_conn.execute("SELECT value FROM records;").fetchall() + read_conn.close() + + assert rows == [("from wal",)] + assert _sha256(evidence_path) == evidence_hash + assert _sha256(tmp_path / "evidence.db-wal") == wal_hash + assert not os.path.exists(str(evidence_path) + "-shm") + + +def test_recovery_preserves_source_database(tmp_path): + database_path = tmp_path / "source.db" + conn = sqlite3.connect(database_path) + conn.execute("CREATE TABLE records (value TEXT);") + conn.execute("INSERT INTO records VALUES ('preserved');") + conn.commit() + conn.close() + source_hash = _sha256(database_path) + + module = IOSExtraction(file_path=str(database_path)) + module._recover_sqlite_db_if_needed(str(database_path), forced=True) + recovered_conn = module._open_sqlite_db(str(database_path)) + rows = recovered_conn.execute("SELECT value FROM records;").fetchall() + recovered_conn.close() + + assert rows == [("preserved",)] + assert _sha256(database_path) == source_hash + assert not os.path.exists(str(database_path) + ".bak") + + +def test_analytics_skips_empty_rows_and_continues(tmp_path): + database_path = tmp_path / "analytics.db" + conn = sqlite3.connect(database_path) + for table in ("hard_failures", "soft_failures", "all_events"): + conn.execute(f"CREATE TABLE {table} (timestamp REAL, data BLOB);") + conn.execute("INSERT INTO hard_failures VALUES (NULL, NULL);") + conn.execute( + "INSERT INTO soft_failures VALUES (?, ?);", + (1.0, plistlib.dumps({"event": "valid"})), + ) + conn.commit() + conn.close() + + module = Analytics(file_path=str(database_path)) + module._extract_analytics_data() + + assert len(module.results) == 1 + assert module.results[0]["event"] == "valid" diff --git a/tests/test_check_android_androidqf.py b/tests/test_check_android_androidqf.py index a2ed952..9b271be 100644 --- a/tests/test_check_android_androidqf.py +++ b/tests/test_check_android_androidqf.py @@ -6,10 +6,15 @@ import logging import os import shutil +import tempfile +import zipfile from click.testing import CliRunner from mvt.android.cli import check_androidqf +from mvt.android.cmd_check_androidqf import CmdAndroidCheckAndroidQF +from mvt.android.modules.androidqf import ANDROIDQF_MODULES +from mvt.android.modules.androidqf.aqf_log_timestamps import AQFLogTimestamps from mvt.common.config import settings from .utils import get_artifact_folder @@ -18,6 +23,9 @@ TEST_BACKUP_PASSWORD = "123456" class TestCheckAndroidqfCommand: + def test_log_timestamps_module_is_registered(self): + assert AQFLogTimestamps in ANDROIDQF_MODULES + def test_check(self): runner = CliRunner() path = os.path.join(get_artifact_folder(), "androidqf") @@ -89,3 +97,25 @@ class TestCheckAndroidqfCommand: assert not any( record.levelname in {"CRITICAL", "FATAL"} for record in caplog.records ) + + def test_intrusion_log_zip_rejects_path_traversal(self, tmp_path, mocker, caplog): + escaped_name = f"mvt-escaped-{tmp_path.name}.txt" + escaped_path = os.path.join(tempfile.gettempdir(), escaped_name) + archive_path = tmp_path / "androidqf.zip" + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr(f"intrusion_logs/../{escaped_name}", "unsafe") + archive.writestr("intrusion_logs/safe.txt", "safe") + + nested_command = mocker.patch( + "mvt.android.cmd_check_androidqf.CmdAndroidCheckIntrusionLogs" + ) + nested_command.return_value.timeline = [] + nested_command.return_value.alertstore.alerts = [] + command = CmdAndroidCheckAndroidQF(target_path=str(archive_path)) + command.init() + + with caplog.at_level(logging.WARNING): + assert command.run_intrusion_logs_cmd() is True + + assert not os.path.exists(escaped_path) + assert "Skipping unsafe intrusion log archive entry" in caplog.text diff --git a/tests/test_check_android_bugreport.py b/tests/test_check_android_bugreport.py index b13c38d..bff4709 100644 --- a/tests/test_check_android_bugreport.py +++ b/tests/test_check_android_bugreport.py @@ -18,3 +18,13 @@ class TestCheckBugreportCommand: path = os.path.join(get_artifact_folder(), "android_data/bugreport/") result = runner.invoke(check_bugreport, [path]) assert result.exit_code == 0 + + def test_invalid_zip_reports_clean_error(self, tmp_path): + path = tmp_path / "invalid.zip" + path.write_bytes(b"not a zip archive") + + result = CliRunner().invoke(check_bugreport, [str(path)]) + + assert result.exit_code == 1 + assert "Invalid bugreport archive" in result.output + assert "Traceback" not in result.output diff --git a/tests/utils.py b/tests/utils.py index 9ae433d..d84dc2f 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -49,8 +49,7 @@ def add_backup_manifest_entry(backup_path, file_id, domain, relative_path): (file_id, domain, relative_path, b""), ) conn.commit() - # MVT opens backup databases with immutable=1, which ignores any -wal file, - # so the new row has to be checkpointed into Manifest.db itself. + # Checkpoint the test change so the fixture does not retain SQLite sidecars. conn.execute("PRAGMA journal_mode=DELETE;") conn.close()