mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-23 15:48:24 +02:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
cc27b40aa0 | ||
|
|
911115b0c8 | ||
|
|
c806fd8d61 | ||
|
|
516ba06cf7 | ||
|
|
1532578b39 |
No files matched your search
@@ -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,
|
||||
]
|
||||
@@ -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))
|
||||
@@ -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"
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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]:
|
||||
|
||||
@@ -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
|
||||
+3
-4
@@ -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()
|
||||
|
||||
@@ -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())
|
||||
@@ -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(
|
||||
[
|
||||
|
||||
@@ -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
|
||||
)
|
||||
@@ -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
|
||||
|
||||
@@ -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() == "**"
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user