CalvinBackup/mvt
1
0
Fork 0
mirror of https://github.com/mvt-project/mvt.git synced 2026-07-18 02:17:22 +02:00
Code Issues Packages Projects Releases Wiki Activity

Deduplicate AndroidQF SMS analysis (#837)

Browse Source
This commit is contained in:
besendorf
2026-07-15 09:09:04 +02:00
committed by GitHub
parent 911115b0c8
commit afcfda4720
9 changed files with 191 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mvt/android/modules/androidqf/__init__.py
-2
View File
@@ -10,7 +10,6 @@ from .aqf_processes import AQFProcesses
from .aqf_settings import AQFSettings
from .mounts import Mounts
from .root_binaries import RootBinaries
from .sms import SMS
ANDROIDQF_MODULES = [
AQFPackages,
@@ -18,7 +17,6 @@ ANDROIDQF_MODULES = [
AQFGetProp,
AQFSettings,
AQFFiles,
SMS,
RootBinaries,
Mounts,
]
src/mvt/android/modules/androidqf/sms.py
-105
View File
@@ -1,105 +0,0 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2023 The MVT Authors.
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1
import logging
from typing import Optional
from mvt.android.modules.backup.helpers import prompt_or_load_android_backup_password
from mvt.android.parsers.backup import (
AndroidBackupParsingError,
InvalidBackupPassword,
parse_ab_header,
parse_backup_file,
parse_tar_for_sms,
)
from .base import AndroidQFModule
class SMS(AndroidQFModule):
"""
This module analyse SMS file in backup
"""
def __init__(
self,
file_path: Optional[str] = None,
target_path: Optional[str] = None,
results_path: Optional[str] = None,
module_options: Optional[dict] = None,
log: logging.Logger = logging.getLogger(__name__),
results: Optional[list] = None,
) -> None:
super().__init__(
file_path=file_path,
target_path=target_path,
results_path=results_path,
module_options=module_options,
log=log,
results=results,
)
def check_indicators(self) -> None:
if not self.indicators:
return
for message in self.results:
if "body" not in message:
continue
ioc_match = self.indicators.check_domains(message.get("links", []))
if ioc_match:
self.alertstore.critical(
ioc_match.message, "", message, matched_indicator=ioc_match.ioc
)
def parse_backup(self, data):
header = parse_ab_header(data)
if not header["backup"]:
self.log.warning("Invalid backup format, backup.ab was not analysed")
return
password = None
if header["encryption"] != "none":
password = prompt_or_load_android_backup_password(
self.log, self.module_options
)
if not password:
self.log.critical("No backup password provided.")
return
try:
tardata = parse_backup_file(data, password=password)
except InvalidBackupPassword:
self.log.critical("Invalid backup password")
return
except AndroidBackupParsingError:
self.log.warning(
"Impossible to parse this backup file, please use"
" Android Backup Extractor instead"
)
return
if not tardata:
return
try:
self.results = parse_tar_for_sms(tardata)
except AndroidBackupParsingError:
self.log.info(
"Impossible to read SMS from the Android Backup, "
"please extract the SMS and try extracting it with "
"Android Backup Extractor"
)
return
def run(self) -> None:
files = self._get_files_by_pattern("*/backup.ab")
if not files:
self.log.info("No backup data found")
return
self.parse_backup(self._get_file_content(files[0]))
self.log.info("Identified %d SMS in backup data", len(self.results))
src/mvt/android/modules/backup/helpers.py
+2 -3
View File
@@ -4,9 +4,8 @@
# https://license.mvt.re/1.1/
from rich.prompt import Prompt
from mvt.common.config import settings
from mvt.common.password import prompt_password
MVT_ANDROID_BACKUP_PASSWORD = "MVT_ANDROID_BACKUP_PASSWORD"
@@ -49,7 +48,7 @@ def prompt_or_load_android_backup_password(log, module_options):
# The default is to allow interactivity
elif module_options.get("interactive", True):
backup_password = Prompt.ask(prompt="Enter backup password", password=True)
backup_password = prompt_password("Enter backup password: ")
else:
log.critical(
"Cannot decrypt backup because interactivity"
src/mvt/common/password.py
+143
View File
@@ -0,0 +1,143 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2026 The MVT Authors.
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
"""Password prompts with masked keyboard feedback.
This is based on the ``echo_char`` support added to :mod:`getpass` in Python
3.14. MVT supports Python 3.10 and later, so it cannot use that API directly.
"""
import contextlib
import io
import os
import sys
import warnings
from typing import TextIO
def prompt_password(prompt: str) -> str:
"""Return a password while displaying an asterisk for each character."""
try:
import termios
return _unix_getpass(prompt, termios)
except ImportError:
try:
import msvcrt
return _windows_getpass(prompt, msvcrt)
except ImportError:
return _fallback_getpass(prompt)
def _unix_getpass(prompt: str, termios) -> str:
password = None
input_stream: TextIO
output_stream: TextIO
with contextlib.ExitStack() as stack:
try:
fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
tty = io.FileIO(fd, "w+")
stack.enter_context(tty)
input_stream = io.TextIOWrapper(tty)
stack.enter_context(input_stream)
output_stream = input_stream
except OSError:
stack.close()
try:
fd = sys.stdin.fileno()
except (AttributeError, ValueError):
return _fallback_getpass(prompt)
input_stream = sys.stdin
output_stream = sys.stderr
try:
old = termios.tcgetattr(fd)
new = old[:]
new[3] &= ~termios.ECHO
new[3] &= ~termios.ICANON
try:
termios.tcsetattr(fd, termios.TCSAFLUSH, new)
password = _readline_with_asterisks(output_stream, input_stream, prompt)
finally:
termios.tcsetattr(fd, termios.TCSAFLUSH, old)
output_stream.flush()
except termios.error:
if password is not None:
raise
password = _fallback_getpass(prompt)
output_stream.write("\n")
return password
def _windows_getpass(prompt: str, msvcrt) -> str:
if sys.stdin is not sys.__stdin__:
return _fallback_getpass(prompt)
for char in prompt:
msvcrt.putwch(char)
password = ""
while True:
char = msvcrt.getwch()
if char in ("\r", "\n"):
break
if char == "\x03":
raise KeyboardInterrupt
if char == "\b":
if password:
msvcrt.putwch("\b")
msvcrt.putwch(" ")
msvcrt.putwch("\b")
password = password[:-1]
else:
password += char
msvcrt.putwch("*")
msvcrt.putwch("\r")
msvcrt.putwch("\n")
return password
def _fallback_getpass(prompt: str) -> str:
warnings.warn(
"Can not control echo on the terminal.",
category=UserWarning,
stacklevel=2,
)
print("Warning: Password input may be echoed.", file=sys.stderr)
return input(prompt)
def _readline_with_asterisks(
output_stream: TextIO, input_stream: TextIO, prompt: str
) -> str:
output_stream.write(prompt)
output_stream.flush()
password = ""
eof_pressed = False
while True:
char = input_stream.read(1)
if char in ("\n", "\r"):
break
if char == "\x03":
raise KeyboardInterrupt
if char in ("\x7f", "\b"):
if password:
output_stream.write("\b \b")
output_stream.flush()
password = password[:-1]
elif char == "\x04":
if eof_pressed:
break
eof_pressed = True
elif char != "\x00":
password += char
output_stream.write("*")
output_stream.flush()
eof_pressed = False
return password
src/mvt/ios/cli.py
+3 -4
View File
@@ -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()
tests/android_androidqf/test_sms.py
-91
View File
@@ -1,91 +0,0 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2023 The MVT Authors.
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import logging
import os
from pathlib import Path
from mvt.android.modules.androidqf.sms import SMS
from mvt.common.module import run_module
from ..utils import get_android_androidqf, get_artifact_folder, list_files
TEST_BACKUP_PASSWORD = "123456"
class TestAndroidqfSMSAnalysis:
def test_androidqf_sms(self):
data_path = get_android_androidqf()
m = SMS(target_path=data_path, log=logging)
files = list_files(data_path)
parent_path = Path(data_path).absolute().parent.as_posix()
m.from_dir(parent_path, files)
run_module(m)
assert len(m.results) == 2
assert len(m.timeline) == 0
assert len(m.alertstore.alerts) == 0
def test_androidqf_sms_encrypted_password_valid(self):
data_path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
m = SMS(
target_path=data_path,
log=logging,
module_options={"backup_password": TEST_BACKUP_PASSWORD},
)
files = list_files(data_path)
parent_path = Path(data_path).absolute().parent.as_posix()
m.from_dir(parent_path, files)
run_module(m)
assert len(m.results) == 1
def test_androidqf_sms_encrypted_password_prompt(self, mocker):
data_path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
prompt_mock = mocker.patch(
"rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
)
m = SMS(
target_path=data_path,
log=logging,
module_options={},
)
files = list_files(data_path)
parent_path = Path(data_path).absolute().parent.as_posix()
m.from_dir(parent_path, files)
run_module(m)
assert prompt_mock.call_count == 1
assert len(m.results) == 1
def test_androidqf_sms_encrypted_password_invalid(self, caplog):
data_path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
with caplog.at_level(logging.CRITICAL):
m = SMS(
target_path=data_path,
log=logging,
module_options={"backup_password": "invalid_password"},
)
files = list_files(data_path)
parent_path = Path(data_path).absolute().parent.as_posix()
m.from_dir(parent_path, files)
run_module(m)
assert len(m.results) == 0
assert "Invalid backup password" in caplog.text
def test_androidqf_sms_encrypted_no_interactive(self, caplog):
data_path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
with caplog.at_level(logging.CRITICAL):
m = SMS(
target_path=data_path,
log=logging,
module_options={"interactive": False},
)
files = list_files(data_path)
parent_path = Path(data_path).absolute().parent.as_posix()
m.from_dir(parent_path, files)
run_module(m)
assert len(m.results) == 0
assert (
"Cannot decrypt backup because interactivity was disabled and the password was not supplied"
in caplog.text
)
tests/common/test_password.py
+28
View File
@@ -0,0 +1,28 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2026 The MVT Authors.
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
from io import StringIO
from mvt.common.password import _readline_with_asterisks
def test_readline_with_asterisks():
output = StringIO()
password = _readline_with_asterisks(
output, StringIO("pass\x7fword\n"), "Enter backup password: "
)
assert password == "pasword"
assert output.getvalue() == "Enter backup password: ****\b \b****"
def test_readline_with_asterisks_ignores_nul_and_handles_eof():
output = StringIO()
password = _readline_with_asterisks(output, StringIO("a\x00b\x04\x04"), "")
assert password == "ab"
assert output.getvalue() == "**"
tests/test_check_android_androidqf.py
+9 -5
View File
@@ -27,21 +27,24 @@ class TestCheckAndroidqfCommand:
def test_check_encrypted_backup_prompt_valid(self, mocker):
"""Prompt for password on CLI"""
prompt_mock = mocker.patch(
"rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
"mvt.android.modules.backup.helpers.prompt_password",
return_value=TEST_BACKUP_PASSWORD,
)
runner = CliRunner()
path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
result = runner.invoke(check_androidqf, [path])
# Called twice, once in AnroidQF SMS module and once in Backup SMS module
assert prompt_mock.call_count == 2
# The password entered for the AndroidQF SMS module is reused by the
# nested backup command.
assert prompt_mock.call_count == 1
assert result.exit_code == 0
def test_check_encrypted_backup_cli(self, mocker):
"""Provide password as CLI argument"""
prompt_mock = mocker.patch(
"rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
"mvt.android.modules.backup.helpers.prompt_password",
return_value=TEST_BACKUP_PASSWORD,
)
runner = CliRunner()
@@ -56,7 +59,8 @@ class TestCheckAndroidqfCommand:
def test_check_encrypted_backup_env(self, mocker):
"""Provide password as environment variable"""
prompt_mock = mocker.patch(
"rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
"mvt.android.modules.backup.helpers.prompt_password",
return_value=TEST_BACKUP_PASSWORD,
)
os.environ["MVT_ANDROID_BACKUP_PASSWORD"] = TEST_BACKUP_PASSWORD
tests/test_check_android_backup.py
+6 -3
View File
@@ -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