From afcfda47202f0dc65bc029fe5fe0ce1a30b29481 Mon Sep 17 00:00:00 2001 From: besendorf Date: Wed, 15 Jul 2026 09:09:04 +0200 Subject: [PATCH] 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