Merge main and fix encrypted backup extraction

This commit is contained in:
Codex
2026-08-16 22:39:08 +05:30
232 changed files with 12251 additions and 4033 deletions
@@ -5,6 +5,7 @@
import logging
from mvt.android.artifacts.dumpsys_accessibility import DumpsysAccessibilityArtifact
from mvt.common.alerts import AlertLevel
from mvt.common.indicators import Indicators
from ..utils import get_artifact
@@ -38,6 +39,19 @@ class TestDumpsysAccessibilityArtifact:
assert da.results[0]["package_name"] == "com.malware.accessibility"
assert da.results[0]["service"] == "com.malware.service.malwareservice"
def test_accessibility_service_alert(self):
da = DumpsysAccessibilityArtifact()
file = get_artifact("android_data/dumpsys_accessibility_v14_or_later.txt")
with open(file) as f:
data = f.read()
da.parse(data)
da.check_indicators()
assert len(da.alertstore.alerts) == 1
assert da.alertstore.alerts[0].level == AlertLevel.MEDIUM
assert da.alertstore.alerts[0].event == da.results[0]
def test_ioc_check(self, indicator_file):
da = DumpsysAccessibilityArtifact()
file = get_artifact("android_data/dumpsys_accessibility.txt")
@@ -49,6 +63,14 @@ class TestDumpsysAccessibilityArtifact:
ind.parse_stix2(indicator_file)
ind.ioc_collections[0]["app_ids"].append("com.sec.android.app.camera")
da.indicators = ind
assert len(da.detected) == 0
assert len(da.alertstore.alerts) == 0
da.check_indicators()
assert len(da.detected) == 1
assert len(da.alertstore.alerts) == len(da.results)
assert da.alertstore.count(AlertLevel.MEDIUM) == 3
assert da.alertstore.count(AlertLevel.CRITICAL) == 1
critical_alert = next(
alert
for alert in da.alertstore.alerts
if alert.level == AlertLevel.CRITICAL
)
assert critical_alert.event["package_name"] == "com.sec.android.app.camera"
+178
View File
@@ -4,6 +4,8 @@
# https://license.mvt.re/1.1/
from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
from mvt.android.modules.bugreport.dumpsys_adb_state import DumpsysADBState
from mvt.common.alerts import AlertLevel
from ..utils import get_artifact
@@ -30,6 +32,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")
@@ -54,3 +116,119 @@ class TestDumpsysADBArtifact:
assert key_store_entry["user"] == "user@laptop"
assert key_store_entry["fingerprint"] == expected_fingerprint
assert key_store_entry["last_connected"] == "1628501829898"
class TestDumpsysADBStateAlerts:
def test_no_androidqf_context_preserves_existing_behavior(self):
module = DumpsysADBState(
results=[
{
"user_keys": [
{
"key": b"QUJDRA==",
"user": "host@example",
"fingerprint": "fingerprint",
}
]
}
]
)
module.check_indicators()
assert module.alertstore.alerts == []
def test_androidqf_trusted_keys_create_expected_alerts(self):
module = DumpsysADBState(
module_options={
"androidqf_acquisition": {
"started": "2025-06-20T18:00:00Z",
"adb_host_public_key": "QUJDRA== acquisition@host",
}
},
results=[
{
"user_keys": [
{
"key": b"QUJDRA==",
"user": "acquisition@host",
"fingerprint": "acquisition-fingerprint",
},
{
"key": b"RUZHSA==",
"user": "other@host",
"fingerprint": "other-fingerprint",
},
{
"key": b"not-base64",
"user": "invalid@host",
"fingerprint": "",
},
],
"keystore": [
{
"key": b"QUJDRA==",
"user": "acquisition@host",
"fingerprint": "acquisition-fingerprint",
"last_connected": "1750266000000",
}
],
}
],
)
module.check_indicators()
assert [alert.level for alert in module.alertstore.alerts] == [
AlertLevel.INFORMATIONAL,
AlertLevel.LOW,
AlertLevel.LOW,
]
informational, different, invalid = module.alertstore.alerts
assert "at least one day before" in informational.message
assert informational.event_time == "2025-06-18 17:00:00.000000"
assert "different from the AndroidQF acquisition host" in different.message
assert "invalid trusted ADB host key" in invalid.message
def test_missing_androidqf_host_key_creates_low_alert(self):
trusted_key = {
"key": b"QUJDRA==",
"user": "host@example",
"fingerprint": "fingerprint",
}
module = DumpsysADBState(
module_options={"androidqf_acquisition": {}},
results=[{"user_keys": [trusted_key]}],
)
module.check_indicators()
assert len(module.alertstore.alerts) == 1
assert module.alertstore.alerts[0].level == AlertLevel.LOW
assert "does not include its host key" in module.alertstore.alerts[0].message
def test_recent_acquisition_host_key_does_not_create_alert(self):
module = DumpsysADBState(
module_options={
"androidqf_acquisition": {
"started": "2025-06-20T18:00:00Z",
"adb_host_public_key": "QUJDRA== acquisition@host",
}
},
results=[
{
"keystore": [
{
"key": b"QUJDRA==",
"user": "acquisition@host",
"fingerprint": "fingerprint",
"last_connected": "1750438800000",
}
]
}
],
)
module.check_indicators()
assert module.alertstore.alerts == []
@@ -42,22 +42,25 @@ class TestDumpsysAppopsArtifact:
ind.parse_stix2(indicator_file)
ind.ioc_collections[0]["app_ids"].append("com.facebook.katana")
da.indicators = ind
assert len(da.detected) == 0
assert len(da.alertstore.alerts) == 0
da.check_indicators()
detected_by_ioc = [
detected for detected in da.detected if detected.get("matched_indicator")
alert
for alert in da.alertstore.alerts
if alert.matched_indicator is not None
]
detected_by_permission_heuristic = [
detected
for detected in da.detected
alert
for alert in da.alertstore.alerts
if all(
[
perm["name"] == "REQUEST_INSTALL_PACKAGES"
for perm in detected["permissions"]
for perm in alert.event["permissions"]
]
)
]
assert len(da.detected) == 3
assert len(da.alertstore.alerts) == 3
assert len(detected_by_ioc) == 1
assert detected_by_ioc[0].matched_indicator is not None
assert len(detected_by_permission_heuristic) == 2
@@ -5,6 +5,7 @@
import logging
from mvt.android.artifacts.dumpsys_battery_daily import DumpsysBatteryDailyArtifact
from mvt.common.alerts import AlertLevel
from mvt.common.indicators import Indicators
from ..utils import get_artifact
@@ -32,6 +33,101 @@ class TestDumpsysBatteryDailyArtifact:
ind.parse_stix2(indicator_file)
ind.ioc_collections[0]["app_ids"].append("com.facebook.system")
dba.indicators = ind
assert len(dba.detected) == 0
assert len(dba.alertstore.alerts) == 0
dba.check_indicators()
assert len(dba.detected) == 1
assert len(dba.alertstore.alerts) == 1
def test_uninstall_and_downgrade_create_medium_alerts(self):
dba = DumpsysBatteryDailyArtifact()
dba.parse(
"""
Daily from 2022-08-16-15-56-39 to 2022-08-17-01-15-45:
Update com.example.app vers=10
Update com.example.removed vers=0
Daily from 2022-08-17-15-56-39 to 2022-08-18-01-15-45:
Update com.example.app vers=9
"""
)
assert len(dba.results) == 3
assert len(dba.alertstore.alerts) == 2
uninstall_alert, downgrade_alert = dba.alertstore.alerts
assert uninstall_alert.level == AlertLevel.MEDIUM
assert uninstall_alert.message == (
"Detected uninstall of package com.example.removed (vers 0)"
)
assert uninstall_alert.event_time == "2022-08-16"
assert uninstall_alert.event["package_name"] == "com.example.removed"
assert uninstall_alert.event["vers"] == "0"
assert downgrade_alert.level == AlertLevel.MEDIUM
assert downgrade_alert.message == (
"Detected downgrade of package com.example.app from vers 10 to vers 9"
)
assert downgrade_alert.event_time == "2022-08-17"
assert downgrade_alert.event["package_name"] == "com.example.app"
assert downgrade_alert.event["action"] == "downgrade"
assert downgrade_alert.event["previous_vers"] == "10"
def test_newest_first_update_is_not_reported_as_downgrade(self):
dba = DumpsysBatteryDailyArtifact()
dba.parse(
"""
Daily from 2026-01-10 to 2026-01-11:
Update com.example.app vers=102
Daily from 2026-01-05 to 2026-01-06:
Update com.example.app vers=101
"""
)
assert len(dba.results) == 2
assert len(dba.alertstore.alerts) == 0
assert all(result["action"] == "update" for result in dba.results)
def test_newest_first_downgrade_creates_medium_alert(self):
dba = DumpsysBatteryDailyArtifact()
dba.parse(
"""
Daily from 2026-01-10 to 2026-01-11:
Update com.example.app vers=101
Daily from 2026-01-05 to 2026-01-06:
Update com.example.app vers=102
"""
)
assert len(dba.results) == 2
assert len(dba.alertstore.alerts) == 1
downgrade_alert = dba.alertstore.alerts[0]
assert downgrade_alert.level == AlertLevel.MEDIUM
assert downgrade_alert.message == (
"Detected downgrade of package com.example.app from vers 102 to vers 101"
)
assert downgrade_alert.event_time == "2026-01-10"
assert downgrade_alert.event["package_name"] == "com.example.app"
assert downgrade_alert.event["action"] == "downgrade"
assert downgrade_alert.event["previous_vers"] == "102"
def test_reinstall_after_uninstall_is_not_reported_as_downgrade(self):
dba = DumpsysBatteryDailyArtifact()
dba.parse(
"""
Daily from 2026-01-15 to 2026-01-16:
Update com.example.app vers=10
Daily from 2026-01-10 to 2026-01-11:
Update com.example.app vers=0
Daily from 2026-01-05 to 2026-01-06:
Update com.example.app vers=102
"""
)
assert len(dba.results) == 3
assert len(dba.alertstore.alerts) == 1
uninstall_alert = dba.alertstore.alerts[0]
assert uninstall_alert.level == AlertLevel.MEDIUM
assert uninstall_alert.message == (
"Detected uninstall of package com.example.app (vers 0)"
)
assert uninstall_alert.event_time == "2026-01-10"
@@ -39,6 +39,25 @@ class TestDumpsysBatteryHistoryArtifact:
ind.parse_stix2(indicator_file)
ind.ioc_collections[0]["app_ids"].append("com.samsung.android.app.reminder")
dba.indicators = ind
assert len(dba.detected) == 0
assert len(dba.alertstore.alerts) == 0
dba.check_indicators()
assert len(dba.detected) == 2
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"
+21 -2
View File
@@ -37,6 +37,25 @@ class TestDumpsysDBinfoArtifact:
ind.parse_stix2(indicator_file)
ind.ioc_collections[0]["app_ids"].append("com.wssyncmldm")
dbi.indicators = ind
assert len(dbi.detected) == 0
assert len(dbi.alertstore.alerts) == 0
dbi.check_indicators()
assert len(dbi.detected) == 5
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",
}
]
@@ -39,6 +39,6 @@ class TestDumpsysPackageActivitiesArtifact:
ind.parse_stix2(indicator_file)
ind.ioc_collections[0]["app_ids"].append("com.google.android.gms")
dpa.indicators = ind
assert len(dpa.detected) == 0
assert len(dpa.alertstore.alerts) == 0
dpa.check_indicators()
assert len(dpa.detected) == 1
assert len(dpa.alertstore.alerts) == 1
@@ -25,6 +25,25 @@ class TestDumpsysPackagesArtifact:
== "com.samsung.android.provider.filterprovider"
)
assert dpa.results[0]["version_name"] == "5.0.07"
assert dpa.results[0]["first_install_time"] == "2008-12-31 16:00:00"
assert dpa.results[0]["system"] is True
def test_parsing_system_flag(self):
system_details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
" pkgFlags=[ SYSTEM HAS_CODE ALLOW_CLEAR_USER_DATA ]"
)
third_party_details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
" pkgFlags=[ HAS_CODE ALLOW_BACKUP ]"
)
missing_flag_details = (
DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
" versionName=1.0"
)
)
assert system_details["system"] is True
assert third_party_details["system"] is False
assert missing_flag_details["system"] is False
def test_ioc_check(self, indicator_file):
dpa = DumpsysPackagesArtifact()
@@ -37,6 +56,46 @@ class TestDumpsysPackagesArtifact:
ind.parse_stix2(indicator_file)
ind.ioc_collections[0]["app_ids"].append("com.sec.android.app.DataCreate")
dpa.indicators = ind
assert len(dpa.detected) == 0
assert len(dpa.alertstore.alerts) == 0
dpa.check_indicators()
assert len(dpa.detected) == 1
assert len(dpa.alertstore.alerts) == 1
def test_per_user_fields_use_primary_user(self):
details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
""" User 0: installed=true
firstInstallTime=2024-01-10 09:19:39
runtime permissions:
android.permission.CAMERA: granted=true
User 95: installed=false
firstInstallTime=1970-01-01 01:00:00
runtime permissions:
android.permission.CAMERA: granted=false
android.permission.RECORD_AUDIO: granted=false
"""
)
assert details["first_install_time"] == "2024-01-10 09:19:39"
runtime_permissions = [
permission
for permission in details["permissions"]
if permission["type"] == "runtime"
]
assert runtime_permissions == [
{
"name": "android.permission.CAMERA",
"granted": True,
"type": "runtime",
}
]
def test_per_user_fields_fall_back_when_user_zero_is_missing(self):
details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
""" User 10: installed=true
firstInstallTime=2024-02-10 09:19:39
runtime permissions:
android.permission.CAMERA: granted=true
"""
)
assert details["first_install_time"] == "2024-02-10 09:19:39"
assert details["permissions"][-1]["name"] == "android.permission.CAMERA"
@@ -35,6 +35,6 @@ class TestDumpsysPlatformCompatArtifact:
ind.ioc_collections[0]["app_ids"].append("org.torproject.torbrowser")
ind.ioc_collections[0]["app_ids"].append("org.article19.circulo.next")
dbi.indicators = ind
assert len(dbi.detected) == 0
assert len(dbi.alertstore.alerts) == 0
dbi.check_indicators()
assert len(dbi.detected) == 2
assert len(dbi.alertstore.alerts) == 2
@@ -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")
@@ -42,6 +80,6 @@ class TestDumpsysReceiversArtifact:
ind.parse_stix2(indicator_file)
ind.ioc_collections[0]["app_ids"].append("com.android.storagemanager")
dr.indicators = ind
assert len(dr.detected) == 0
assert len(dr.alertstore.alerts) == 0
dr.check_indicators()
assert len(dr.detected) == 1
assert len(dr.alertstore.alerts) == 1
+2 -2
View File
@@ -36,6 +36,6 @@ class TestGetPropArtifact:
"dalvik.vm.appimageformat"
)
gp.indicators = ind
assert len(gp.detected) == 0
assert len(gp.alertstore.alerts) == 0
gp.check_indicators()
assert len(gp.detected) == 1
assert len(gp.alertstore.alerts) == 1
+2 -2
View File
@@ -33,6 +33,6 @@ class TestProcessesArtifact:
ind.parse_stix2(indicator_file)
ind.ioc_collections[0]["processes"].append("lru-add-drain")
p.indicators = ind
assert len(p.detected) == 0
assert len(p.alertstore.alerts) == 0
p.check_indicators()
assert len(p.detected) == 1
assert len(p.alertstore.alerts) == 1
+68
View File
@@ -8,6 +8,7 @@ import datetime
import pytest
from mvt.android.artifacts.tombstone_crashes import TombstoneCrashArtifact
from mvt.android.parsers.proto.tombstone import Tombstone
from ..utils import get_artifact
@@ -42,6 +43,73 @@ class TestTombstoneCrashArtifact:
assert len(tombstone_artifact.results) == 1
self.validate_tombstone_result(tombstone_artifact.results[0])
def test_text_tombstone_preserves_abort_message(self):
tombstone_artifact = TombstoneCrashArtifact()
artifact_path = "android_data/bugreport/FS/data/tombstones/tombstone_00"
file = get_artifact(artifact_path)
with open(file, "rb") as f:
data = f.read()
tombstone_artifact.parse(
os.path.basename(artifact_path),
datetime.datetime(2021, 9, 29, 17, 43, 49),
data,
)
assert tombstone_artifact.results[0]["abort_message"] == (
"Check failed: payload.size() <= bytes_left "
"(payload.size()=99, bytes_left=51) "
)
def test_protobuf_tombstone_preserves_abort_message_and_causes(self):
tombstone_artifact = TombstoneCrashArtifact()
artifact_path = "android_data/tombstone_process.pb"
file = get_artifact(artifact_path)
with open(file, "rb") as f:
tombstone = Tombstone().parse(f.read())
tombstone.abort_message = "synthetic abort reason"
tombstone_artifact.parse_protobuf(
os.path.basename(artifact_path),
datetime.datetime(2023, 4, 12, 12, 32, 40, 518290),
bytes(tombstone),
)
result = tombstone_artifact.results[0]
assert result["abort_message"] == "synthetic abort reason"
assert result["causes"] == [
{
"human_readable": "null pointer dereference",
"memory_error": None,
}
]
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()
+24 -2
View File
@@ -5,12 +5,35 @@
import hashlib
from mvt.android.parsers.backup import parse_backup_file, parse_tar_for_sms
import pytest
from mvt.android.parsers.backup import (
AndroidBackupParsingError,
parse_ab_header,
parse_backup_file,
parse_tar_for_sms,
)
from ..utils import get_artifact
class TestBackupParsing:
def test_parse_incomplete_header(self):
assert parse_ab_header(b"ANDROID BACKUP\n") == {
"backup": False,
"compression": None,
"version": None,
"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:
@@ -60,7 +83,6 @@ class TestBackupParsing:
== "33e73df2ede9798dcb3a85c06200ee41c8f52dd2f2e50ffafcceb0407bc13e3a"
)
sms = parse_tar_for_sms(ddata)
print(sms)
assert isinstance(sms, list)
assert len(sms) == 1
assert len(sms[0]["links"]) == 1
+333
View File
@@ -0,0 +1,333 @@
# 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 json
import logging
import pytest
from click.testing import CliRunner
from mvt.android.cli import check_intrusion_logs
from mvt.android.cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
from mvt.android.modules.intrusion_logs.base import IntrusionLogsModule
from mvt.android.modules.intrusion_logs.security_event import SecurityEvent
from mvt.common.alerts import AlertLevel
def _write_ndjson(path, records):
path.write_text(
"\n".join(json.dumps(record) for record in records),
encoding="utf-8",
)
def test_load_all_events_preserves_unknown_top_level_event(tmp_path):
_write_ndjson(
tmp_path / "intrusion.txt",
[
{
"future_event": {
"event_time": 1_700_000_000_000,
"field": "value",
}
}
],
)
module = IntrusionLogsModule(target_path=str(tmp_path))
events = module.load_all_events(str(tmp_path))
assert events == {
"future_event": [
{
"event_time": 1_700_000_000_000,
"field": "value",
}
]
}
def test_check_intrusion_logs_warns_about_unknown_top_level_event_type(
tmp_path, caplog
):
_write_ndjson(
tmp_path / "intrusion.txt",
[
{
"future_event": {
"event_time": 1_700_000_000_000,
"field": "value",
}
}
],
)
with caplog.at_level(logging.WARNING):
cmd = CmdAndroidCheckIntrusionLogs(target_path=str(tmp_path))
cmd.run()
assert "Found unknown intrusion logging event type(s): future_event" in caplog.text
assert "Please open an issue on GitHub" in caplog.text
def test_check_intrusion_logs_parses_core_and_unknown_security_events(
tmp_path, caplog
):
_write_ndjson(
tmp_path / "intrusion.txt",
[
{
"dns_event": {
"event_time": 1_700_000_000_000,
"hostname": "example.com",
"package_name": "com.example.app",
"ip_addresses": ["/1.2.3.4"],
}
},
{
"connect_event": {
"event_time": 1_700_000_001_000,
"ip_address": "/5.6.7.8",
"port": 443,
"package_name": "com.example.app",
}
},
{
"security_event": {
"event_time": 1_700_000_002_000_000_000,
"app_process_start": {
"process": "com.example.app",
"uid": 10_000,
"pid": 1234,
},
}
},
{
"security_event": {
"event_time": 1_700_000_003_000_000_000,
"future_google_event": {
"field": "value",
},
}
},
],
)
with caplog.at_level(logging.WARNING):
cmd = CmdAndroidCheckIntrusionLogs(target_path=str(tmp_path))
cmd.run()
assert [module.__class__.__name__ for module in cmd.executed] == [
"DnsEvent",
"ConnectEvent",
"SecurityEvent",
]
assert [len(module.results) for module in cmd.executed] == [1, 1, 2]
security_module = next(
module for module in cmd.executed if isinstance(module, SecurityEvent)
)
assert security_module.event_type_counts["app_process_start"] == 1
assert security_module.event_type_counts["future_google_event"] == 1
future_timeline_events = [
event for event in cmd.timeline if event["event"] == "future_google_event"
]
assert len(future_timeline_events) == 1
assert "future_google_event" in future_timeline_events[0]["data"]
assert "field" in future_timeline_events[0]["data"]
assert (
"Found unknown intrusion logging security event type(s): future_google_event"
in caplog.text
)
assert "Please open an issue on GitHub" in caplog.text
def test_check_intrusion_logs_treats_event_id_as_security_event_metadata(
tmp_path, caplog
):
_write_ndjson(
tmp_path / "intrusion.txt",
[
{
"security_event": {
"event_id": 191,
"event_time": 1_700_000_002_000_000_000,
"keyguard_dismiss_auth_attempt": {
"success": True,
"method_strength": 0,
},
}
},
{
"security_event": {
"event_id": 192,
"event_time": 1_700_000_003_000_000_000,
"keyguard_dismissed": {},
}
},
],
)
with caplog.at_level(logging.WARNING):
cmd = CmdAndroidCheckIntrusionLogs(target_path=str(tmp_path))
cmd.run()
security_module = next(
module for module in cmd.executed if isinstance(module, SecurityEvent)
)
assert security_module.event_type_counts == {
"keyguard_dismiss_auth_attempt": 1,
"keyguard_dismissed": 1,
}
assert [event["event_id"] for event in security_module.results] == [191, 192]
keyguard_events = {
event["event"]: event
for event in cmd.timeline
if event["event"]
in {"keyguard_dismiss_auth_attempt", "keyguard_dismissed"}
}
assert "Auth attempt: Success" in keyguard_events[
"keyguard_dismiss_auth_attempt"
]["data"]
assert keyguard_events["keyguard_dismissed"]["data"] == "Keyguard dismissed"
assert "unknown intrusion logging security event type(s): event_id" not in caplog.text
def test_check_intrusion_logs_cli_lists_modules(tmp_path):
_write_ndjson(tmp_path / "intrusion.txt", [])
result = CliRunner().invoke(check_intrusion_logs, ["--list-modules", str(tmp_path)])
assert result.exit_code == 0
assert "DnsEvent" in result.output
assert "ConnectEvent" in result.output
assert "SecurityEvent" in result.output
def _run_security_heuristics(results):
# No indicators loaded: heuristic alerts must still fire.
module = SecurityEvent(results=results)
module.check_indicators()
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(
[
{
"timestamp": "2024-01-01 00:00:00.000",
"cert_authority_installed": {
"subject": "CN=Unexpected Root CA",
"success": True,
},
}
]
)
assert len(alerts) == 1
assert alerts[0].level == AlertLevel.MEDIUM
assert "Certificate authority installed" in alerts[0].message
assert "Unexpected Root CA" in alerts[0].message
# Exported logs encode success as a JSON bool, raw SecurityLog as int 0/1.
@pytest.mark.parametrize("success", [False, 0])
def test_failed_cert_authority_install_does_not_alert(success, caplog):
with caplog.at_level(logging.WARNING):
alerts = _run_security_heuristics(
[
{
"timestamp": "2024-01-01 00:00:00.000",
"cert_authority_installed": {
"subject": "CN=Unexpected Root CA",
"success": success,
},
}
]
)
assert alerts == []
assert "Failed certificate authority install attempt" in caplog.text
assert "Unexpected Root CA" in caplog.text
def test_cert_validation_failure_raises_medium_alert_without_indicators():
alerts = _run_security_heuristics(
[
{
"timestamp": "2024-01-01 00:00:00.000",
"cert_validation_failure": "chain validation failed",
}
]
)
assert len(alerts) == 1
assert alerts[0].level == AlertLevel.MEDIUM
assert "Certificate validation failure" in alerts[0].message
def test_security_heuristics_fire_when_no_indicators_loaded():
# check_indicators() previously returned early with no indicators loaded,
# so none of the heuristic alerts fired on a default run.
alerts = _run_security_heuristics(
[
{"timestamp": "2024-01-01 00:00:00.000", "wipe_failure": {"reason": "x"}},
{
"timestamp": "2024-01-01 00:00:00.000",
"key_integrity_violation": {"key_id": "k1"},
},
]
)
assert len(alerts) == 2
assert all(alert.level == AlertLevel.MEDIUM for alert in alerts)
+1 -1
View File
@@ -22,4 +22,4 @@ class TestAndroidqfFilesAnalysis:
run_module(m)
assert len(m.results) == 3
assert len(m.timeline) == 6
assert len(m.detected) == 0
assert len(m.alertstore.alerts) == 0
+4 -4
View File
@@ -26,7 +26,7 @@ class TestAndroidqfGetpropAnalysis:
assert m.results[0]["name"] == "dalvik.vm.appimageformat"
assert m.results[0]["value"] == "lz4"
assert len(m.timeline) == 0
assert len(m.detected) == 0
assert len(m.alertstore.alerts) == 0
def test_getprop_parsing_zip(self):
fpath = get_artifact("androidqf.zip")
@@ -38,7 +38,7 @@ class TestAndroidqfGetpropAnalysis:
assert m.results[0]["name"] == "dalvik.vm.appimageformat"
assert m.results[0]["value"] == "lz4"
assert len(m.timeline) == 0
assert len(m.detected) == 0
assert len(m.alertstore.alerts) == 0
def test_androidqf_getprop_detection(self, indicator_file):
data_path = get_android_androidqf()
@@ -52,5 +52,5 @@ class TestAndroidqfGetpropAnalysis:
m.indicators = ind
run_module(m)
assert len(m.results) == 10
assert len(m.detected) == 1
assert m.detected[0]["name"] == "dalvik.vm.heapmaxfree"
assert len(m.alertstore.alerts) == 1
assert m.alertstore.alerts[0].event["name"] == "dalvik.vm.heapmaxfree"
+36 -1
View File
@@ -6,6 +6,7 @@
import logging
from pathlib import Path
from mvt.common.indicators import Indicator, IndicatorMatch
from mvt.common.module import run_module
from ..utils import get_android_androidqf, list_files
@@ -72,6 +73,38 @@ class TestAndroidqfMountsArtifact:
(("by-name/data" in s or "/data" in s) and "rw" in s) for s in concatenated
), f"No data-like tokens (data + rw) found in parsed results: {concatenated}"
def test_mount_ioc_alert_uses_indicator(self):
from mvt.android.artifacts.mounts import Mounts as MountsArtifact
indicator = Indicator(
value="/system",
type="file_path",
name="TestMalware",
stix2_file_name="indicators.stix2",
)
m = MountsArtifact()
m.indicators = type(
"MountIndicators",
(),
{
"check_file_path": lambda self, path: IndicatorMatch(
ioc=indicator, message="matched file path"
)
if path == "/system"
else None
},
)()
m.parse("/dev/block/by-name/system on /system type ext4 (rw,seclabel)")
m.check_indicators()
indicator_alerts = [
alert for alert in m.alertstore.alerts if alert.matched_indicator
]
assert len(indicator_alerts) == 1
assert indicator_alerts[0].matched_indicator == indicator
assert indicator_alerts[0].message == "matched file path"
class TestAndroidqfMountsModule:
def test_androidqf_module_no_mounts_file(self):
@@ -94,4 +127,6 @@ class TestAndroidqfMountsModule:
assert len(m.results) == 0, (
f"Expected no results when mounts.json is absent, got: {m.results}"
)
assert len(m.detected) == 0, f"Expected no detections, got: {m.detected}"
assert len(m.alertstore.alerts) == 0, (
f"Expected no detections, got: {m.alertstore.alerts}"
)
+87 -31
View File
@@ -7,8 +7,11 @@ import logging
from pathlib import Path
import pytest
from click.testing import CliRunner
from mvt.android.cli import check_androidqf
from mvt.android.modules.androidqf.aqf_packages import AQFPackages
from mvt.android.modules.androidqf import aqf_packages as aqf_packages_module
from mvt.common.module import run_module
from ..utils import get_android_androidqf, list_files
@@ -47,38 +50,35 @@ class TestAndroidqfPackages:
def test_non_appstore_warnings(self, caplog, module):
run_module(module)
assert len(module.detected) == 4
assert len(module.alertstore.alerts) == 5
# Not a super test to be searching logs for this but heuristic detections not yet formalised
assert (
'Found a non-system package installed via adb or another method: "com.whatsapp"'
in caplog.text
)
adb_message = "Found a non-system package installed via adb or another method:"
whatsapp_detected = [
pkg for pkg in module.detected if pkg["name"] == "com.whatsapp"
alert
for alert in module.alertstore.alerts
if alert.event["name"] == "com.whatsapp"
]
assert len(whatsapp_detected) == 1
assert adb_message in whatsapp_detected[0].message
assert (
'Found a package installed via a browser (installer="com.google.android.packageinstaller"): '
'"app.revanced.manager.flutter"' in caplog.text
)
browser_message = 'Found a package installed via a browser (installer="com.google.android.packageinstaller"): '
revanced_detected = [
pkg
for pkg in module.detected
if pkg["name"] == "app.revanced.manager.flutter"
alert
for alert in module.alertstore.alerts
if alert.event["name"] == "app.revanced.manager.flutter"
]
assert len(revanced_detected) == 1
assert browser_message in revanced_detected[0].message
assert (
'Found a package installed via a third party store (installer="org.fdroid.fdroid"): "org.nuclearfog.apollo"'
in caplog.text
)
# We do not currently flag a third party store as a detection, we only flag the app in the logs.
third_party_message = 'Found a package installed via a third party store (installer="org.fdroid.fdroid")'
appollo_detected = [
pkg for pkg in module.detected if pkg["name"] == "org.nuclearfog.apollo"
alert
for alert in module.alertstore.alerts
if alert.event["name"] == "org.nuclearfog.apollo"
]
assert len(appollo_detected) == 0
assert len(appollo_detected) == 1
assert third_party_message in appollo_detected[0].message
def test_packages_ioc_package_names(self, module, indicators_factory):
module.indicators = indicators_factory(app_ids=["com.malware.blah"])
@@ -86,13 +86,13 @@ class TestAndroidqfPackages:
run_module(module)
possible_detected_app = [
pkg for pkg in module.detected if pkg["name"] == "com.malware.blah"
alert
for alert in module.alertstore.alerts
if alert.event["name"] == "com.malware.blah"
]
assert len(possible_detected_app) == 1
assert possible_detected_app[0]["name"] == "com.malware.blah"
assert (
possible_detected_app[0]["matched_indicator"]["value"] == "com.malware.blah"
)
assert possible_detected_app[0].event["name"] == "com.malware.blah"
assert possible_detected_app[0].matched_indicator.value == "com.malware.blah"
def test_packages_ioc_sha256(self, module, indicators_factory):
module.indicators = indicators_factory(
@@ -104,12 +104,14 @@ class TestAndroidqfPackages:
run_module(module)
possible_detected_app = [
pkg for pkg in module.detected if pkg["name"] == "com.malware.muahaha"
alert
for alert in module.alertstore.alerts
if alert.event["name"] == "com.malware.muahaha"
]
assert len(possible_detected_app) == 1
assert possible_detected_app[0]["name"] == "com.malware.muahaha"
assert possible_detected_app[0].event["name"] == "com.malware.muahaha"
assert (
possible_detected_app[0]["matched_indicator"]["value"]
possible_detected_app[0].matched_indicator.value
== "31037a27af59d4914906c01ad14a318eee2f3e31d48da8954dca62a99174e3fa"
)
@@ -123,11 +125,65 @@ class TestAndroidqfPackages:
run_module(module)
possible_detected_app = [
pkg for pkg in module.detected if pkg["name"] == "com.malware.muahaha"
alert
for alert in module.alertstore.alerts
if alert.event["name"] == "com.malware.muahaha"
]
assert len(possible_detected_app) == 1
assert possible_detected_app[0]["name"] == "com.malware.muahaha"
assert possible_detected_app[0].event["name"] == "com.malware.muahaha"
assert (
possible_detected_app[0]["matched_indicator"]["value"]
possible_detected_app[0].matched_indicator.value
== "c7e56178748be1441370416d4c10e34817ea0c961eb636c8e9d98e0fd79bf730"
)
def test_virustotal_delays_after_missing_result(self, monkeypatch):
lookups = []
sleeps = []
def fake_virustotal_lookup(file_hash):
lookups.append(file_hash)
if file_hash == "missing_hash":
return None
return {
"attributes": {
"last_analysis_stats": {"malicious": 1},
"last_analysis_results": {"engine": {}},
}
}
monkeypatch.setattr(
aqf_packages_module, "virustotal_lookup", fake_virustotal_lookup
)
monkeypatch.setattr(aqf_packages_module.time, "sleep", sleeps.append)
module = AQFPackages(
module_options={"virustotal": True, "virustotal_delay": 16},
results=[
{
"name": "org.example",
"installer": "com.android.vending",
"disabled": False,
"system": False,
"files": [
{"path": "/data/app/missing.apk", "sha256": "missing_hash"},
{"path": "/data/app/found.apk", "sha256": "found_hash"},
],
}
],
)
module.check_indicators()
assert lookups == ["missing_hash", "found_hash"]
assert sleeps == [16]
assert module.results[0]["files"][1]["virustotal"] == "1/1"
assert len(module.alertstore.alerts) == 1
def test_check_androidqf_rejects_negative_virustotal_delay(data_path):
runner = CliRunner()
result = runner.invoke(check_androidqf, ["--delay", "-1", data_path])
assert result.exit_code == 2
assert "Invalid value for '--delay'" in result.output
+1 -1
View File
@@ -22,4 +22,4 @@ class TestAndroidqfProcessesAnalysis:
run_module(m)
assert len(m.results) == 15
assert len(m.timeline) == 0
assert len(m.detected) == 0
assert len(m.alertstore.alerts) == 0
@@ -42,7 +42,7 @@ class TestAndroidqfRootBinaries:
# Should find 4 root binaries from the test file
assert len(module.results) == 4
assert len(module.detected) == 4
assert len(module.alertstore.alerts) == 4
# Check that all results are detected as indicators
binary_paths = [result["path"] for result in module.results]
@@ -113,4 +113,4 @@ class TestAndroidqfRootBinaries:
run_module(m)
assert len(m.results) == 0
assert len(m.detected) == 0
assert len(m.alertstore.alerts) == 0
+2 -1
View File
@@ -21,4 +21,5 @@ class TestSettingsModule:
run_module(m)
assert len(m.results) == 1
assert "random" in m.results.keys()
assert len(m.detected) == 0
assert len(m.alertstore.alerts) == 1
assert "samsung_errorlog_agree" in m.alertstore.alerts[0].message
-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.detected) == 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
)
+36
View File
@@ -0,0 +1,36 @@
# 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 mvt.common.indicators import Indicators
from mvt.common.module import run_module
from mvt.ios.modules.mixed.tcc import TCC
from ..utils import get_ios_backup_folder
class TestTCCModule:
def test_tcc(self):
m = TCC(target_path=get_ios_backup_folder())
run_module(m)
assert len(m.results) == 11
assert len(m.timeline) == 11
assert len(m.alertstore.alerts) == 0
assert m.results[0]["service"] == "kTCCServiceUbiquity"
assert m.results[0]["client"] == "com.apple.Preferences"
assert m.results[0]["auth_value"] == "allowed"
def test_tcc_detection(self, indicator_file):
m = TCC(target_path=get_ios_backup_folder())
ind = Indicators(log=logging.getLogger())
ind.parse_stix2(indicator_file)
m.indicators = ind
run_module(m)
assert len(m.results) == 11
assert len(m.timeline) == 11
assert len(m.alertstore.alerts) == 1
assert m.alertstore.alerts[0].event["service"] == "kTCCServiceLiverpool"
assert m.alertstore.alerts[0].event["client"] == "Launch"
+34 -2
View File
@@ -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
@@ -36,9 +37,13 @@ class TestBugreportAnalysis:
assert len(m.timeline) == 16
detected_by_ioc = [
detected for detected in m.detected if detected.get("matched_indicator")
detected
for detected in m.alertstore.alerts
if detected.event.get("matched_indicator")
]
assert len(m.detected) == 1 # Hueristic detection for suspicious permissions
assert (
len(m.alertstore.alerts) == 1
) # Hueristic detection for suspicious permissions
assert len(detected_by_ioc) == 0
def test_packages_module(self):
@@ -49,6 +54,8 @@ class TestBugreportAnalysis:
== "com.samsung.android.provider.filterprovider"
)
assert m.results[1]["package_name"] == "com.instagram.android"
assert m.results[0]["installer"] == ""
assert m.results[1]["installer"] == "com.android.vending"
assert len(m.results[0]["permissions"]) == 4
assert len(m.results[1]["permissions"]) == 32
@@ -56,6 +63,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
Binary file not shown.
+49
View File
@@ -0,0 +1,49 @@
# 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 mvt.common.alerts import Alert, AlertLevel, AlertStore
def test_as_json_promotes_nested_matched_indicator():
indicator = {"value": "com.apple.weather", "type": "processes"}
alertstore = AlertStore()
alertstore.add(
Alert(
level=AlertLevel.CRITICAL,
module="datausage",
message="Matched indicator",
event_time="2026-01-01 00:00:00",
event={
"proc_name": "WeatherWidget/com.apple.weather",
"matched_indicator": indicator,
},
)
)
alert = alertstore.as_json()[0]
assert alert["matched_indicator"] == indicator
assert "matched_indicator" not in alert["event"]
def test_as_json_removes_nested_matched_indicator_when_parent_exists():
event_indicator = {"value": "nested", "type": "processes"}
alert_indicator = {"value": "parent", "type": "processes"}
alertstore = AlertStore()
alertstore.add(
Alert(
level=AlertLevel.CRITICAL,
module="manifest",
message="Matched indicator",
event_time="2026-01-01 00:00:00",
event={"path": "/tmp/example", "matched_indicator": event_indicator},
matched_indicator=alert_indicator,
)
)
alert = alertstore.as_json()[0]
assert alert["matched_indicator"] == alert_indicator
assert "matched_indicator" not in alert["event"]
+371
View File
@@ -0,0 +1,371 @@
from types import SimpleNamespace
import click
from click.testing import CliRunner
from mvt.common.cli_plugins import (
ANDROID_CLI_PLUGIN_GROUP,
IOS_CLI_PLUGIN_GROUP,
BrokenPluginCommand,
load_cli_commands_option,
register_cli_commands_from_path,
register_cli_plugins,
register_installed_cli_commands,
)
COMMAND_TEMPLATE = """
import click
@click.command({name!r})
@click.pass_context
def cli(ctx):
click.echo({message!r})
if ctx.obj:
click.echo(ctx.obj.get("marker", ""))
"""
def _write_command(path, name, message="command ran"):
path.write_text(
COMMAND_TEMPLATE.format(name=name, message=message),
encoding="utf-8",
)
return path
def _make_group():
@click.group()
@load_cli_commands_option
@click.pass_context
def group(ctx):
ctx.ensure_object(dict)
ctx.obj["marker"] = "parent context"
return group
def _entry_point(name, value, command=None, exception=None, distribution="plugin"):
def load():
if exception is not None:
raise exception
return command
dist = SimpleNamespace(metadata={"Name": distribution}, version="1.0")
return SimpleNamespace(name=name, value=value, load=load, dist=dist)
def test_load_command_option_registers_command_before_resolution(tmp_path):
command_path = _write_command(tmp_path / "hello.py", "hello")
group = _make_group()
result = CliRunner().invoke(
group,
["--load-command", str(command_path), "hello"],
)
assert result.exit_code == 0
assert "command ran" in result.output
assert "parent context" in result.output
def test_load_command_option_supports_folders_and_repeated_paths(tmp_path):
folder = tmp_path / "commands"
folder.mkdir()
_write_command(folder / "b.py", "second")
_write_command(folder / "a.py", "first")
_write_command(folder / ".hidden.py", "hidden")
_write_command(folder / "__init__.py", "init")
other = _write_command(tmp_path / "third.py", "third")
group = _make_group()
result = CliRunner().invoke(
group,
[
"--load-command",
str(folder),
"--load-command",
str(other),
"--load-command",
str(other),
"--help",
],
)
assert result.exit_code == 0
assert "first" in result.output
assert "second" in result.output
assert "third" in result.output
assert "hidden" not in result.output
assert "init" not in result.output
def test_loaded_command_participates_in_shell_completion(tmp_path):
command_path = _write_command(tmp_path / "hello.py", "hello")
group = _make_group()
words = f"group --load-command {command_path} he"
result = CliRunner().invoke(
group,
[],
env={
"_GROUP_COMPLETE": "bash_complete",
"COMP_WORDS": words,
"COMP_CWORD": "3",
},
)
assert result.exit_code == 0
assert "plain,hello" in result.output
def test_explicit_command_import_and_contract_failures_are_usage_errors(tmp_path):
broken_path = tmp_path / "broken.py"
broken_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
missing_cli_path = tmp_path / "missing_cli.py"
missing_cli_path.write_text("value = 1", encoding="utf-8")
broken_result = CliRunner().invoke(
_make_group(),
["--load-command", str(broken_path), "broken"],
)
missing_cli_result = CliRunner().invoke(
_make_group(),
["--load-command", str(missing_cli_path), "missing-cli"],
)
assert broken_result.exit_code == 2
assert "broken import" in broken_result.output
assert missing_cli_result.exit_code == 2
assert "must export a Click command or group named 'cli'" in (
missing_cli_result.output
)
def test_environment_command_failure_gets_broken_placeholder(tmp_path):
command_path = tmp_path / "broken_command.py"
command_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
group = click.Group()
registered = register_cli_commands_from_path(group, command_path)
assert registered == ["broken-command"]
assert isinstance(group.commands["broken-command"], BrokenPluginCommand)
result = CliRunner().invoke(group, ["broken-command"])
assert result.exit_code == 1
assert "broken import" in result.output
assert str(command_path) in result.output
def test_environment_command_system_exit_gets_broken_placeholder(tmp_path):
command_path = tmp_path / "exiting_command.py"
command_path.write_text("raise SystemExit(7)", encoding="utf-8")
group = click.Group()
registered = register_cli_commands_from_path(group, command_path)
assert registered == ["exiting-command"]
assert isinstance(group.commands["exiting-command"], BrokenPluginCommand)
result = CliRunner().invoke(group, ["exiting-command"])
assert result.exit_code == 1
assert "Unable to import custom command" in result.output
assert result.output.rstrip().endswith(": 7")
def test_installed_entry_point_name_is_the_command_name(monkeypatch):
@click.command("internal-name")
def command():
click.echo("installed command ran")
entry_point = _entry_point(
"external-name",
"example_plugin:cli",
command=command,
)
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [entry_point],
)
group = click.Group()
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert registered == ["external-name"]
assert "internal-name" not in group.commands
result = CliRunner().invoke(group, ["external-name"])
assert result.exit_code == 0
assert result.output == "installed command ran\n"
def test_broken_installed_plugin_does_not_break_cli(monkeypatch):
broken = _entry_point(
"broken",
"broken_plugin:cli",
exception=RuntimeError("missing dependency"),
distribution="broken-plugin",
)
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [broken],
)
group = click.Group()
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
help_result = CliRunner().invoke(group, ["--help"])
assert help_result.exit_code == 0
assert "Warning: external command could not be loaded." in help_result.output
result = CliRunner().invoke(group, ["broken"])
assert result.exit_code == 1
assert "broken-plugin 1.0 (broken_plugin:cli)" in result.output
assert "RuntimeError: missing dependency" in result.output
def test_installed_plugin_system_exit_does_not_break_cli(monkeypatch):
exiting = _entry_point(
"exiting",
"exiting_plugin:cli",
exception=SystemExit(7),
distribution="exiting-plugin",
)
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [exiting],
)
group = click.Group()
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
help_result = CliRunner().invoke(group, ["--help"])
assert help_result.exit_code == 0
result = CliRunner().invoke(group, ["exiting"])
assert result.exit_code == 1
assert "SystemExit: 7" in result.output
def test_non_click_entry_point_gets_broken_placeholder(monkeypatch):
invalid = _entry_point("invalid", "plugin:value", command=object())
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [invalid],
)
group = click.Group()
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert isinstance(group.commands["invalid"], BrokenPluginCommand)
result = CliRunner().invoke(group, ["invalid"])
assert result.exit_code == 1
assert "must resolve to a Click command or group" in result.output
def test_entry_point_discovery_failure_does_not_break_group(monkeypatch, caplog):
def fail_discovery(**kwargs):
raise RuntimeError("invalid package metadata")
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
fail_discovery,
)
group = click.Group()
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert registered == []
assert not group.commands
assert "Unable to discover external commands" in caplog.text
assert "invalid package metadata" in caplog.text
def test_builtin_and_first_external_command_win_collisions(monkeypatch, caplog):
@click.command("version")
def core_version():
pass
@click.command()
def first():
pass
@click.command()
def second():
pass
entry_points = [
_entry_point("duplicate", "z_plugin:cli", command=second, distribution="z"),
_entry_point("version", "plugin:version", command=first),
_entry_point("duplicate", "a_plugin:cli", command=first, distribution="a"),
]
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: entry_points,
)
group = click.Group(commands={"version": core_version})
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert registered == ["duplicate"]
assert group.commands["version"] is core_version
assert group.commands["duplicate"] is first
assert "the command name is already registered" in caplog.text
def test_explicit_command_cannot_replace_existing_command(tmp_path):
command_path = _write_command(tmp_path / "version.py", "version")
group = _make_group()
@group.command("version")
def core_version():
pass
result = CliRunner().invoke(
group,
["--load-command", str(command_path), "version"],
)
assert result.exit_code == 2
assert "the command name is already registered" in result.output
def test_platform_entry_point_groups_and_environment_paths_are_separate(
tmp_path, monkeypatch
):
ios_path = _write_command(tmp_path / "ios.py", "ios-file")
android_path = _write_command(tmp_path / "android.py", "android-file")
@click.command()
def ios_package():
pass
@click.command()
def android_package():
pass
def entry_points(*, group):
if group == IOS_CLI_PLUGIN_GROUP:
return [_entry_point("ios-package", "ios_plugin:cli", ios_package)]
return [_entry_point("android-package", "android_plugin:cli", android_package)]
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
entry_points,
)
monkeypatch.setenv("TEST_IOS_COMMANDS", str(ios_path))
monkeypatch.setenv("TEST_ANDROID_COMMANDS", str(android_path))
ios_group = click.Group()
android_group = click.Group()
register_cli_plugins(
ios_group,
entry_point_group=IOS_CLI_PLUGIN_GROUP,
environment_variable="TEST_IOS_COMMANDS",
)
register_cli_plugins(
android_group,
entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
environment_variable="TEST_ANDROID_COMMANDS",
)
assert set(ios_group.commands) == {"ios-file", "ios-package"}
assert set(android_group.commands) == {"android-file", "android-package"}
+222
View File
@@ -0,0 +1,222 @@
# 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 json
import logging
from mvt.common.command import Command
from mvt.common.module import MVTModule
class RecordingModule(MVTModule):
run_order: list[str] = []
def run(self):
self.run_order.append(self.__class__.__name__)
def check_indicators(self):
pass
class FirstModule(RecordingModule):
def run(self):
super().run()
self.results = ["first"]
class SecondModule(RecordingModule):
dependencies = (FirstModule,)
def run(self):
super().run()
self.results = self.get_dependency_results(FirstModule) + ["second"]
class ThirdModule(RecordingModule):
dependencies = (SecondModule,)
class IndependentModule(RecordingModule):
pass
class URLRecordingModule(RecordingModule):
def collect_url_results(self):
self.add_url_result(
"https://example.org/message",
"2026-07-29 12:00:00.000000",
"test-chat",
)
class CustomIOSBackupModule(RecordingModule):
supported_commands = (("ios", "check-backup"),)
class CustomIOSFSModule(RecordingModule):
supported_commands = (("ios", "check-fs"),)
class UnscopedCustomModule(RecordingModule):
pass
class CustomDependsOnBuiltin(RecordingModule):
supported_commands = (("ios", "check-backup"),)
dependencies = (FirstModule,)
class RecordingCommand(Command):
def init(self):
self.initialized = True
def module_init(self, module):
pass
def finish(self):
pass
class TestCommand:
def setup_method(self):
RecordingModule.run_order = []
def test_store_alerts_handles_bytes(self, tmp_path):
cmd = Command(results_path=str(tmp_path))
cmd.alertstore.medium(
"bytes event",
"",
{"payload": b"\xa8\xa9"},
)
cmd._store_alerts()
alerts = json.loads((tmp_path / "alerts.json").read_text())
assert alerts[0]["event"]["payload"] == "\\xa8\\xa9"
def test_stores_collected_urls(self, tmp_path):
cmd = RecordingCommand(results_path=str(tmp_path))
cmd.modules = [URLRecordingModule]
cmd.run()
assert json.loads((tmp_path / "urls.json").read_text()) == [
{
"url": "https://example.org/message",
"expanded_url": None,
"timestamp": "2026-07-29 12:00:00.000000",
"source": "test-chat",
}
]
def test_modules_run_in_stable_topological_order(self):
cmd = RecordingCommand()
cmd.modules = [ThirdModule, IndependentModule, SecondModule, FirstModule]
cmd.run()
assert RecordingModule.run_order == [
"IndependentModule",
"FirstModule",
"SecondModule",
"ThirdModule",
]
second = next(module for module in cmd.executed if isinstance(module, SecondModule))
assert second.results == ["first", "second"]
def test_selected_module_runs_transitive_dependencies(self):
cmd = RecordingCommand(module_name="ThirdModule")
cmd.modules = [ThirdModule, SecondModule, FirstModule, IndependentModule]
cmd.run()
assert RecordingModule.run_order == [
"FirstModule",
"SecondModule",
"ThirdModule",
]
def test_circular_dependency_warns_and_stops(self, caplog):
class CircularOne(RecordingModule):
pass
class CircularTwo(RecordingModule):
dependencies = (CircularOne,)
CircularOne.dependencies = (CircularTwo,)
cmd = RecordingCommand()
cmd.modules = [CircularOne, CircularTwo]
with caplog.at_level(logging.WARNING):
cmd.run()
assert RecordingModule.run_order == []
assert not hasattr(cmd, "initialized")
assert "Circular module dependency detected" in caplog.text
def test_unavailable_dependency_warns_and_stops(self, caplog):
class UnavailableModule(RecordingModule):
pass
class DependentModule(RecordingModule):
dependencies = (UnavailableModule,)
cmd = RecordingCommand()
cmd.modules = [DependentModule]
with caplog.at_level(logging.WARNING):
cmd.run()
assert RecordingModule.run_order == []
assert not hasattr(cmd, "initialized")
assert "depends on unavailable module UnavailableModule" in caplog.text
def test_custom_modules_are_filtered_before_ordering(self):
cmd = RecordingCommand()
cmd.platform = "ios"
cmd.name = "check-backup"
cmd.modules = [FirstModule]
cmd.custom_modules = [
CustomIOSBackupModule,
CustomIOSFSModule,
UnscopedCustomModule,
]
assert [module.__name__ for module in cmd._ordered_modules()] == [
"FirstModule",
"CustomIOSBackupModule",
]
def test_selected_custom_module_runs(self):
cmd = RecordingCommand(module_name="CustomIOSBackupModule")
cmd.platform = "ios"
cmd.name = "check-backup"
cmd.custom_modules = [CustomIOSBackupModule]
cmd.run()
assert RecordingModule.run_order == ["CustomIOSBackupModule"]
def test_selected_unsupported_custom_module_does_not_run(self):
cmd = RecordingCommand(module_name="CustomIOSFSModule")
cmd.platform = "ios"
cmd.name = "check-backup"
cmd.custom_modules = [CustomIOSFSModule]
cmd.run()
assert RecordingModule.run_order == []
def test_custom_module_dependencies_use_topological_order(self):
cmd = RecordingCommand(module_name="CustomDependsOnBuiltin")
cmd.platform = "ios"
cmd.name = "check-backup"
cmd.modules = [SecondModule, FirstModule]
cmd.custom_modules = [CustomDependsOnBuiltin]
cmd.run()
assert RecordingModule.run_order == ["FirstModule", "CustomDependsOnBuiltin"]
+132
View File
@@ -5,7 +5,9 @@
import logging
import os
import threading
import requests
from mvt.common.config import settings
from mvt.common.indicators import Indicators
@@ -80,6 +82,136 @@ class TestIndicators:
assert ind.check_url("https://198.51.100.1:8080/")
assert ind.check_url("https://1.1.1.1/") is None
def test_google_maps_short_url_is_not_resolved(self, indicator_file, mocker):
head_request = mocker.patch("mvt.common.url.requests.head")
ind = Indicators(log=logging)
ind.load_indicators_files([indicator_file], load_default=False)
assert ind.check_url("https://goo.gl/maps/example") is None
head_request.assert_not_called()
def test_check_url_batches_preserves_order(self, indicator_file):
ind = Indicators(log=logging)
ind.load_indicators_files([indicator_file], load_default=False)
matches = ind.check_url_batches(
[
[
"https://github.com",
"http://example.com/thisisbad",
"https://www.example.org/foobar",
],
["https://github.com", "https://www.example.org/foobar"],
[],
None,
]
)
assert matches[0]
assert matches[0].ioc.value == "http://example.com/thisisbad"
assert matches[1]
assert matches[1].ioc.value == "example.org"
assert matches[2] is None
assert matches[3] is None
def test_check_url_batches_deduplicates_and_limits_workers(
self, indicator_file, mocker
):
ind = Indicators(log=logging)
ind.load_indicators_files([indicator_file], load_default=False)
mocker.patch("mvt.common.indicators.URL_CHECK_MAX_WORKERS", 2)
barrier = threading.Barrier(2)
lock = threading.Lock()
calls = []
active = 0
max_active = 0
def head_request(url, timeout):
nonlocal active, max_active
with lock:
calls.append(url)
active += 1
max_active = max(max_active, active)
call_number = len(calls)
try:
if call_number <= 2:
barrier.wait(timeout=5)
return mocker.Mock(status_code=200, headers={})
finally:
with lock:
active -= 1
mocker.patch("mvt.common.url.requests.head", side_effect=head_request)
urls = [
"https://bit.ly/one",
"https://tinyurl.com/two",
"https://t.co/three",
]
assert ind.check_url_batches([urls, [urls[0]]]) == [None, None]
assert sorted(calls) == sorted(urls)
assert max_active == 2
def test_check_url_batches_respects_disabled_network(
self, indicator_file, mocker
):
ind = Indicators(log=logging)
ind.load_indicators_files([indicator_file], load_default=False)
mocker.patch("mvt.common.indicators.settings.NETWORK_ACCESS_ALLOWED", False)
head_request = mocker.patch("mvt.common.url.requests.head")
assert ind.check_url_batches([["https://bit.ly/example"]]) == [None]
head_request.assert_not_called()
def test_check_url_batches_handles_nested_redirects_and_request_failures(
self, indicator_file, mocker
):
ind = Indicators(log=logging)
ind.load_indicators_files([indicator_file], load_default=False)
def head_request(url, timeout):
if url == "https://bit.ly/failure":
raise requests.Timeout()
if url == "https://tinyurl.com/nested":
return mocker.Mock(
status_code=301,
headers={"Location": "https://t.co/nested"},
)
if url == "https://t.co/nested":
return mocker.Mock(
status_code=302,
headers={"Location": "https://www.example.org/landing"},
)
raise AssertionError(f"Unexpected URL: {url}")
head = mocker.patch(
"mvt.common.url.requests.head", side_effect=head_request
)
matches = ind.check_url_batches(
[["https://bit.ly/failure"], ["https://tinyurl.com/nested"]]
)
assert matches[0] is None
assert matches[1]
assert matches[1].ioc.value == "example.org"
assert (
ind.get_expanded_url("https://tinyurl.com/nested")
== "https://www.example.org/landing"
)
assert (
ind.get_expanded_url("https://t.co/nested")
== "https://www.example.org/landing"
)
assert ind.get_expanded_url("https://bit.ly/failure") is None
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)
+146
View File
@@ -0,0 +1,146 @@
import pytest
from mvt.common.module import MVTModule
from mvt.common.module_loader import (
CustomModuleLoadError,
load_custom_modules,
load_custom_modules_from_path,
module_supports_command,
)
MODULE_TEMPLATE = """
from mvt.common.module import MVTModule
class {name}(MVTModule):
supported_commands = {supported_commands!r}
def run(self):
pass
def check_indicators(self):
pass
def serialize(self, result):
return None
"""
def _write_module(path, name, supported_commands=()):
path.write_text(
MODULE_TEMPLATE.format(
name=name,
supported_commands=supported_commands,
),
encoding="utf-8",
)
return path
def test_load_custom_modules_from_python_file(tmp_path):
module_path = _write_module(tmp_path / "custom.py", "FileModule")
modules = load_custom_modules_from_path(str(module_path))
assert [module.__name__ for module in modules] == ["FileModule"]
assert issubclass(modules[0], MVTModule)
def test_load_custom_modules_from_folder_in_sorted_order(tmp_path):
_write_module(tmp_path / "b_module.py", "BModule")
_write_module(tmp_path / "a_module.py", "AModule")
_write_module(tmp_path / ".hidden.py", "HiddenModule")
_write_module(tmp_path / "__init__.py", "InitModule")
nested = tmp_path / "nested"
nested.mkdir()
_write_module(nested / "nested_module.py", "NestedModule")
modules = load_custom_modules_from_path(str(tmp_path))
assert [module.__name__ for module in modules] == ["AModule", "BModule"]
def test_discovery_ignores_imported_base_and_unrelated_classes(tmp_path):
module_path = tmp_path / "custom.py"
module_path.write_text(
"""
from mvt.common.module import MVTModule
class Unrelated:
pass
class DiscoveredModule(MVTModule):
def run(self):
pass
def check_indicators(self):
pass
def serialize(self, result):
return None
""",
encoding="utf-8",
)
modules = load_custom_modules_from_path(str(module_path))
assert [module.__name__ for module in modules] == ["DiscoveredModule"]
def test_load_custom_modules_deduplicates_same_class(tmp_path):
module_path = _write_module(tmp_path / "custom.py", "DuplicateModule")
modules = load_custom_modules([str(module_path), str(module_path)])
assert [module.__name__ for module in modules] == ["DuplicateModule"]
def test_load_custom_modules_raises_for_missing_path(tmp_path):
with pytest.raises(CustomModuleLoadError, match="does not exist"):
load_custom_modules_from_path(str(tmp_path / "missing.py"))
def test_load_custom_modules_raises_for_import_error(tmp_path):
module_path = tmp_path / "broken.py"
module_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
with pytest.raises(CustomModuleLoadError, match="broken import"):
load_custom_modules_from_path(str(module_path))
def test_load_custom_modules_loads_env_folder_first(tmp_path, monkeypatch):
env_folder = tmp_path / "env"
env_folder.mkdir()
cli_folder = tmp_path / "cli"
cli_folder.mkdir()
_write_module(env_folder / "env_module.py", "EnvModule")
_write_module(cli_folder / "cli_module.py", "CliModule")
monkeypatch.setenv("MVT_CUSTOM_MODULES", str(env_folder))
modules = load_custom_modules([str(cli_folder)])
assert [module.__name__ for module in modules] == ["EnvModule", "CliModule"]
def test_module_supports_command_requires_explicit_declaration(tmp_path, caplog):
module_path = _write_module(tmp_path / "custom.py", "DefaultModule")
module = load_custom_modules_from_path(str(module_path))[0]
assert not module_supports_command(module, "ios", "check-backup")
assert not module_supports_command(module, "android", "check-bugreport")
assert "DefaultModule has no supported_commands" in caplog.text
def test_module_supports_command_honors_supported_commands(tmp_path):
module_path = _write_module(
tmp_path / "custom.py",
"SpecificModule",
(("ios", "check-backup"),),
)
module = load_custom_modules_from_path(str(module_path))[0]
assert module_supports_command(module, "ios", "check-backup")
assert not module_supports_command(module, "ios", "check-fs")
+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() == "**"
+24
View File
@@ -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
+2
View File
@@ -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)
+15 -2
View File
@@ -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
@@ -18,7 +19,19 @@ class TestCalendarModule:
run_module(m)
assert len(m.results) == 1
assert len(m.timeline) == 4
assert len(m.detected) == 0
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):
@@ -30,4 +43,4 @@ class TestCalendarModule:
run_module(m)
assert len(m.results) == 1
assert len(m.timeline) == 4
assert len(m.detected) == 1
assert len(m.alertstore.alerts) == 1
+20 -2
View File
@@ -7,6 +7,7 @@ import logging
from mvt.common.indicators import Indicators
from mvt.common.module import run_module
from mvt.common.alerts import AlertLevel
from mvt.ios.modules.mixed.net_datausage import Datausage
from ..utils import get_ios_backup_folder
@@ -19,7 +20,9 @@ class TestDatausageModule:
assert m.results[0]["isodate"][0:19] == "2019-08-27 15:08:09"
assert len(m.results) == 42
assert len(m.timeline) == 60
assert len(m.detected) == 0
assert (
len(m.alertstore.alerts) == 1
) # We now have a detection for missing processes.
def test_detection(self, indicator_file):
m = Datausage(target_path=get_ios_backup_folder())
@@ -29,4 +32,19 @@ class TestDatausageModule:
ind.ioc_collections[0]["processes"].append("CumulativeUsageTracker")
m.indicators = ind
run_module(m)
assert len(m.detected) == 2
critical_alerts = [
alert for alert in m.alertstore.alerts if alert.level == AlertLevel.CRITICAL
]
assert len(critical_alerts) == 2
assert all(
"matched_indicator" not in alert.event for alert in critical_alerts
)
serialized_alerts = [
alert
for alert in m.alertstore.as_json()
if alert["matched_indicator"] is not None
]
assert len(serialized_alerts) == 2
assert all(
"matched_indicator" not in alert["event"] for alert in serialized_alerts
)
+125
View File
@@ -0,0 +1,125 @@
# 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 pathlib import Path
from Crypto.Cipher import AES
from mvt.ios.decrypt import DecryptBackup, MVTEncryptedBackup
def _encrypted_file(backup_path, file_id, key, plaintext):
padding_length = AES.block_size - (len(plaintext) % AES.block_size)
padded = plaintext + bytes([padding_length]) * padding_length
encrypted = AES.new(key, AES.MODE_CBC, iv=b"\x00" * AES.block_size).encrypt(
padded
)
source_path = backup_path / file_id[:2] / file_id
source_path.parent.mkdir(parents=True)
source_path.write_bytes(encrypted)
def test_extract_file_by_id_preserves_bytes_with_wrong_manifest_size(
mocker, tmp_path
):
file_id = "ab" + "1" * 38
plaintext = b"complete decrypted content"
inner_key = b"k" * 32
_encrypted_file(tmp_path, file_id, inner_key, plaintext)
file_plist = mocker.Mock(
encryption_key=b"wrapped-key",
protection_class=1,
filesize=1,
mtime=None,
)
mocker.patch("mvt.ios.decrypt.FilePlist", return_value=file_plist)
backup = MVTEncryptedBackup(
backup_directory=str(tmp_path), derived_key=b"d" * 32
)
mocker.patch.object(backup, "_read_and_unlock_keybag", return_value=True)
backup._keybag = mocker.Mock()
backup._keybag.unwrapKeyForClass.return_value = inner_key
streaming_decrypt = mocker.spy(backup, "_decrypt_file_to_disk")
output_path = tmp_path / "output"
backup.extract_file_by_id(
file_id=file_id,
file_bplist=b"plist",
output_filename=str(output_path),
)
assert output_path.read_bytes() == plaintext
streaming_decrypt.assert_called_once()
def test_extract_file_by_id_copies_unencrypted_files(mocker, tmp_path):
file_id = "cd" + "2" * 38
source_path = tmp_path / file_id[:2] / file_id
source_path.parent.mkdir(parents=True)
source_path.write_bytes(b"plain content")
file_plist = mocker.Mock(encryption_key=None)
mocker.patch("mvt.ios.decrypt.FilePlist", return_value=file_plist)
backup = MVTEncryptedBackup(
backup_directory=str(tmp_path), derived_key=b"d" * 32
)
mocker.patch.object(backup, "_read_and_unlock_keybag", return_value=True)
output_path = tmp_path / "output"
backup.extract_file_by_id(
file_id=file_id,
file_bplist=b"plist",
output_filename=str(output_path),
)
assert output_path.read_bytes() == b"plain content"
def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_path):
backup_path = tmp_path / "backup"
destination = tmp_path / "destination"
outside = tmp_path / "outside"
backup_path.mkdir()
destination.mkdir()
outside.mkdir()
safe_file_id = "ef" + "3" * 38
unsafe_file_id = "../../outside-file"
symlink_file_id = "ab" + "4" * 38
for file_id in (safe_file_id, symlink_file_id):
source_path = backup_path / file_id[:2] / file_id
source_path.parent.mkdir(parents=True, exist_ok=True)
source_path.write_bytes(b"encrypted")
(destination / "ab").symlink_to(outside, target_is_directory=True)
cursor = mocker.MagicMock()
cursor.__iter__.return_value = iter(
[
(safe_file_id, "Domain", "safe", b"plist"),
(unsafe_file_id, "Domain", "unsafe", b"plist"),
(symlink_file_id, "Domain", "symlink", b"plist"),
]
)
cursor_context = mocker.MagicMock()
cursor_context.__enter__.return_value = cursor
backup = mocker.MagicMock()
backup.manifest_db_cursor.return_value = cursor_context
def extract_file_by_id(*, output_filename, **kwargs):
Path(output_filename).write_bytes(b"decrypted")
backup.extract_file_by_id.side_effect = extract_file_by_id
decryptor = DecryptBackup(str(backup_path), str(destination))
decryptor._backup = backup
decryptor.process_backup()
assert (destination / safe_file_id[:2] / safe_file_id).read_bytes() == b"decrypted"
assert not (outside / symlink_file_id).exists()
backup.extract_file_by_id.assert_called_once()
assert backup.extract_file_by_id.call_args.kwargs["file_id"] == safe_file_id
+7 -1
View File
@@ -4,6 +4,7 @@
# https://license.mvt.re/1.1/
from mvt.common.module import run_module
from mvt.common.alerts import AlertLevel
from mvt.ios.modules.mixed.global_preferences import GlobalPreferences
from ..utils import get_ios_backup_folder
@@ -15,6 +16,11 @@ class TestGlobalPreferencesModule:
run_module(m)
assert len(m.results) == 16
assert len(m.timeline) == 0
assert len(m.detected) == 0
assert len(m.alertstore.alerts) == 1
lockdown_mode_alert = m.alertstore.alerts[0]
assert lockdown_mode_alert.message == "Lockdown mode enabled"
assert lockdown_mode_alert.level == AlertLevel.INFORMATIONAL
assert m.results[0]["entry"] == "WebKitShowLinkPreviews"
assert m.results[0]["value"] is False
+23 -2
View File
@@ -3,22 +3,43 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import gc
import logging
import warnings
from mvt.common.indicators import Indicators
from mvt.common.module import run_module
from mvt.ios.modules.base import IOSExtraction
from mvt.ios.modules.backup.manifest import Manifest
from ..utils import get_ios_backup_folder
class TestIOSExtraction:
def test_get_backup_files_from_manifest_closes_connection(self):
m = IOSExtraction(target_path=get_ios_backup_folder())
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always", ResourceWarning)
files = list(m._get_backup_files_from_manifest(domain="CameraRollDomain"))
gc.collect()
assert files
assert not [
warning
for warning in caught
if issubclass(warning.category, ResourceWarning)
and "unclosed database" in str(warning.message)
]
class TestManifestModule:
def test_manifest(self):
m = Manifest(target_path=get_ios_backup_folder())
run_module(m)
assert len(m.results) == 3721
assert len(m.timeline) == 5881
assert len(m.detected) == 0
assert len(m.alertstore.alerts) == 0
def test_detection(self, indicator_file):
m = Manifest(target_path=get_ios_backup_folder())
@@ -27,4 +48,4 @@ class TestManifestModule:
ind.ioc_collections[0]["file_names"].append("com.apple.CoreBrightness.plist")
m.indicators = ind
run_module(m)
assert len(m.detected) == 1
assert len(m.alertstore.alerts) == 1
+44 -3
View File
@@ -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:
@@ -19,7 +51,16 @@ class TestSafariBrowserStateModule:
run_module(m)
assert len(m.results) == 1
assert len(m.timeline) == 1
assert len(m.detected) == 0
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())
@@ -30,6 +71,6 @@ class TestSafariBrowserStateModule:
ind.ioc_collections[0]["domains"].append("en.wikipedia.org")
m.indicators = ind
run_module(m)
assert len(m.detected) == 1
assert len(m.alertstore.alerts) == 1
assert len(m.results) == 1
assert m.results[0]["tab_url"] == "https://en.wikipedia.org/wiki/NSO_Group"
+165
View File
@@ -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
+38 -2
View File
@@ -18,7 +18,15 @@ class TestSMSModule:
run_module(m)
assert len(m.results) == 1
assert len(m.timeline) == 2
assert len(m.detected) == 0
assert m.url_results == [
{
"url": "https://badbadbad.example.org/",
"expanded_url": None,
"timestamp": "2019-08-29 23:13:30.000000",
"source": "sms",
}
]
assert len(m.alertstore.alerts) == 0
def test_detection(self, indicator_file):
m = SMS(target_path=get_ios_backup_folder())
@@ -28,4 +36,32 @@ class TestSMSModule:
ind.ioc_collections[0]["domains"].append("badbadbad.example.org")
m.indicators = ind
run_module(m)
assert len(m.detected) == 1
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]
+85
View File
@@ -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"
+4 -4
View File
@@ -18,7 +18,7 @@ class TestTCCModule:
run_module(m)
assert len(m.results) == 11
assert len(m.timeline) == 11
assert len(m.detected) == 0
assert len(m.alertstore.alerts) == 0
assert m.results[0]["service"] == "kTCCServiceUbiquity"
assert m.results[0]["client"] == "com.apple.Preferences"
assert m.results[0]["auth_value"] == "allowed"
@@ -31,6 +31,6 @@ class TestTCCModule:
run_module(m)
assert len(m.results) == 11
assert len(m.timeline) == 11
assert len(m.detected) == 1
assert m.detected[0]["service"] == "kTCCServiceLiverpool"
assert m.detected[0]["client"] == "Launch"
assert len(m.alertstore.alerts) == 1
assert m.alertstore.alerts[0].event["service"] == "kTCCServiceLiverpool"
assert m.alertstore.alerts[0].event["client"] == "Launch"
@@ -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 sqlite3
from mvt.common.module import run_module
from mvt.ios.modules.mixed.webkit_resource_load_statistics import (
WebkitResourceLoadStatistics,
@@ -18,4 +20,51 @@ class TestWebkitResourceLoadStatisticsModule:
run_module(m)
assert len(m.results) == 2
assert len(m.timeline) == 2
assert len(m.detected) == 0
assert len(m.alertstore.alerts) == 0
results = {result["registrable_domain"]: result for result in m.results}
assert results["google.com"]["most_recent_user_interaction_time"] > 0
assert "most_recent_user_interaction_time_isodate" in results["google.com"]
assert results["gstatic.com"]["most_recent_user_interaction_time"] == -1.0
assert (
"most_recent_user_interaction_time_isodate"
not in results["gstatic.com"]
)
assert all(
"most_recent_web_push_interaction_time" not in result
for result in m.results
)
def test_webkit_full_timestamp_schema(self, tmp_path):
db_path = tmp_path / "observations.db"
conn = sqlite3.connect(db_path)
conn.execute(
"""
CREATE TABLE ObservedDomains (
domainID INTEGER PRIMARY KEY,
registrableDomain TEXT NOT NULL,
lastSeen REAL NOT NULL,
hadUserInteraction INTEGER NOT NULL,
mostRecentUserInteractionTime REAL NOT NULL,
mostRecentWebPushInteractionTime REAL NOT NULL
);
"""
)
conn.execute(
"""
INSERT INTO ObservedDomains VALUES (?, ?, ?, ?, ?, ?);
""",
(1, "example.com", 1634560250.0, 1, 1634560030.0, -1.0),
)
conn.commit()
conn.close()
m = WebkitResourceLoadStatistics(target_path=str(tmp_path))
m._process_observations_db(str(db_path), "", "observations.db")
assert len(m.results) == 1
result = m.results[0]
assert result["most_recent_user_interaction_time"] == 1634560030.0
assert "most_recent_user_interaction_time_isodate" in result
assert result["most_recent_web_push_interaction_time"] == -1.0
assert "most_recent_web_push_interaction_time_isodate" not in result
+35
View File
@@ -0,0 +1,35 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-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/
import logging
from mvt.common.indicators import Indicators
from mvt.ios.modules.mixed.whatsapp import Whatsapp
def test_collect_url_results_includes_expansion():
module = Whatsapp(
results=[
{
"links": ["https://bit.ly/message"],
"isodate": "2026-07-29 12:00:00.000000",
}
]
)
module.indicators = Indicators(log=logging.getLogger())
module.indicators.resolved_urls["https://bit.ly/message"] = (
"https://example.org/landing"
)
module.collect_url_results()
assert module.url_results == [
{
"url": "https://bit.ly/message",
"expanded_url": "https://example.org/landing",
"timestamp": "2026-07-29 12:00:00.000000",
"source": "whatsapp",
}
]
+37
View File
@@ -0,0 +1,37 @@
# 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 mvt.common.indicators import Indicators
from mvt.ios.modules.fs.cache_files import CacheFiles
class TestCacheFiles:
def test_detection(self, indicator_file):
m = CacheFiles(
results={
"Library/Caches/example/Cache.db": [
{
"entry_id": 1,
"version": 1,
"hash_value": 123,
"storage_policy": 0,
"url": "http://example.com/thisisbad",
"isodate": "2026-01-01 00:00:00.000000",
}
]
}
)
ind = Indicators(log=logging.getLogger())
ind.parse_stix2(indicator_file)
m.indicators = ind
m.check_indicators()
assert len(m.alertstore.alerts) == 1
alert = m.alertstore.alerts[0]
assert alert.event["cache_file"] == "Library/Caches/example/Cache.db"
assert alert.event["url"] == "http://example.com/thisisbad"
assert alert.matched_indicator is not None
+2 -2
View File
@@ -17,7 +17,7 @@ class TestFilesystem:
run_module(m)
assert len(m.results) == 15
assert len(m.timeline) == 15
assert len(m.detected) == 0
assert len(m.alertstore.alerts) == 0
def test_detection(self, indicator_file):
m = Filesystem(target_path=get_ios_backup_folder())
@@ -31,4 +31,4 @@ class TestFilesystem:
run_module(m)
assert len(m.results) == 15
assert len(m.timeline) == 15
assert len(m.detected) == 1
assert len(m.alertstore.alerts) == 1
+59
View File
@@ -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
+16
View File
@@ -0,0 +1,16 @@
# 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 click.testing import CliRunner
from mvt.android.cli import check_adb
class TestCheckAndroidADBRemovedCommand:
def test_check_adb_exits_nonzero(self):
runner = CliRunner()
result = runner.invoke(check_adb)
assert result.exit_code == 1
+118 -5
View File
@@ -3,11 +3,19 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import json
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
@@ -16,30 +24,97 @@ 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")
result = runner.invoke(check_androidqf, [path])
assert result.exit_code == 0
def test_check_stores_nested_sms_urls(self, tmp_path):
runner = CliRunner()
path = os.path.join(get_artifact_folder(), "androidqf")
result = runner.invoke(check_androidqf, ["--output", str(tmp_path), path])
assert result.exit_code == 0
urls = json.loads((tmp_path / "urls.json").read_text())
assert {entry["url"] for entry in urls} == {
"http://google.com",
"https://google.com/",
}
assert all(
set(entry) == {"url", "expanded_url", "timestamp", "source"}
for entry in urls
)
assert all(entry["source"] == "sms" for entry in urls)
def test_acquisition_context_is_passed_to_bugreport(self, tmp_path, mocker):
data_path = tmp_path / "androidqf"
data_path.mkdir()
(data_path / "acquisition.json").write_text(
json.dumps(
{
"started": "2025-06-20T18:00:00Z",
"adb_host_public_key": "QUJDRA== acquisition@host",
}
)
)
with zipfile.ZipFile(data_path / "bugreport.zip", "w"):
pass
nested_command = mocker.patch(
"mvt.android.cmd_check_androidqf.CmdAndroidCheckBugreport"
)
nested_command.return_value.timeline = []
nested_command.return_value.alertstore.alerts = []
command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
command.init()
assert command.run_bugreport_cmd() is True
assert nested_command.call_args.kwargs["module_options"][
"androidqf_acquisition"
] == {
"started": "2025-06-20T18:00:00Z",
"adb_host_public_key": "QUJDRA== acquisition@host",
}
def test_acquisition_context_falls_back_to_public_key_file(self, tmp_path):
data_path = tmp_path / "androidqf"
data_path.mkdir()
(data_path / "adb_host_key.pub").write_text("QUJDRA== acquisition@host\n")
command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
command.init()
assert command.module_options["androidqf_acquisition"] == {
"adb_host_public_key": "QUJDRA== acquisition@host\n"
}
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()
@@ -54,7 +129,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
@@ -68,3 +144,40 @@ class TestCheckAndroidqfCommand:
assert result.exit_code == 0
del os.environ["MVT_ANDROID_BACKUP_PASSWORD"]
settings.__init__() # Reset settings
def test_check_malformed_backup_skips_backup_modules(self, tmp_path, caplog):
path = tmp_path / "androidqf"
shutil.copytree(os.path.join(get_artifact_folder(), "androidqf"), path)
(path / "backup.ab").write_bytes(b"")
runner = CliRunner()
with caplog.at_level(logging.WARNING):
result = runner.invoke(check_androidqf, [str(path)])
assert result.exit_code == 0
assert "Skipping backup modules as backup.ab is malformed" in caplog.text
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
+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
+10
View File
@@ -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
+17
View File
@@ -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
+57
View File
@@ -0,0 +1,57 @@
from click.testing import CliRunner
from mvt.ios.cli import check_sysdiagnose
CUSTOM_MODULE = """
from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
class CustomSysdiagnoseModule(SysdiagnoseExtraction):
supported_commands = (("ios", "check-sysdiagnose"),)
slug = "custom_sysdiagnose_module"
def run(self):
file_path = self._get_files_by_pattern("*/artifact.txt")[0]
self.results = [{"content": self._get_file_content(file_path).decode("utf-8")}]
def check_indicators(self):
pass
def serialize(self, result):
return None
"""
def _create_sysdiagnose_folder(tmp_path):
folder = tmp_path / "sysdiagnose"
folder.mkdir()
(folder / "artifact.txt").write_text("artifact", encoding="utf-8")
return folder
def test_check_sysdiagnose_runs_explicitly_scoped_custom_module(tmp_path):
module_path = tmp_path / "custom_sysdiagnose.py"
module_path.write_text(CUSTOM_MODULE, encoding="utf-8")
output_path = tmp_path / "output"
result = CliRunner().invoke(
check_sysdiagnose,
[
"--load-module",
str(module_path),
"--output",
str(output_path),
str(_create_sysdiagnose_folder(tmp_path)),
],
)
assert result.exit_code == 0
assert (output_path / "custom_sysdiagnose_module.json").exists()
def test_check_sysdiagnose_requires_an_explicitly_scoped_module(tmp_path):
result = CliRunner().invoke(check_sysdiagnose, [str(_create_sysdiagnose_folder(tmp_path))])
assert result.exit_code != 0
assert "No custom modules support mvt-ios check-sysdiagnose" in result.output
+117
View File
@@ -0,0 +1,117 @@
import io
import tarfile
from datetime import timedelta
from pathlib import Path
from mvt.ios.cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
class SysdiagnoseTestModule(SysdiagnoseExtraction):
supported_commands = (("ios", "check-sysdiagnose"),)
def run(self):
file_path = self._get_files_by_pattern("*/artifact.txt")[0]
self.results = [
{
"content": self._get_file_content(file_path).decode("utf-8"),
"timezone_offset": self._extract_timezone().utcoffset(None).seconds,
}
]
def check_indicators(self):
pass
def serialize(self, result):
return None
def _create_sysdiagnose_folder(tmp_path):
folder = tmp_path / "sysdiagnose"
folder.mkdir()
(folder / "artifact.txt").write_text("artifact", encoding="utf-8")
(folder / "sysdiagnose.log").write_text(
"sysdiagnose_2024.01.02_03-04-05+0200.tar.gz", encoding="utf-8"
)
(folder / "report.ips").write_text('{"bug_type": 210}\nbody', encoding="utf-8")
return folder
def _create_sysdiagnose_archive(tmp_path, folder):
archive_path = tmp_path / "sysdiagnose.tar.gz"
with tarfile.open(archive_path, "w:gz") as archive:
for path in folder.iterdir():
archive.add(path, arcname=f"sysdiagnose/{path.name}")
return archive_path
def _run_command(path):
command = CmdIOSCheckSysdiagnose(
target_path=str(path), custom_modules=[SysdiagnoseTestModule]
)
command.run()
return command
def test_check_sysdiagnose_from_folder(tmp_path):
command = _run_command(_create_sysdiagnose_folder(tmp_path))
assert command.executed[0].results == [
{"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
]
assert command.executed[0].ips_files == [
{"file_path": str(tmp_path / "sysdiagnose" / "report.ips"), "bug_type": 210}
]
def test_check_sysdiagnose_from_archive_closes_archive(tmp_path):
folder = _create_sysdiagnose_folder(tmp_path)
command = _run_command(_create_sysdiagnose_archive(tmp_path, folder))
assert command.executed[0].results == [
{"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
]
assert command.executed[0].ips_files == [
{
"file_path": str(
Path(command.extracted_sysdiagnose_path) / "report.ips"
),
"bug_type": 210,
}
]
assert command.sysdiagnose_archive is None
def test_archive_is_extracted_once_and_unsafe_members_are_skipped(tmp_path):
archive_path = tmp_path / "sysdiagnose.tar.gz"
escaped_path = tmp_path / "escaped.txt"
content = b"test content"
member = tarfile.TarInfo("sysdiagnose/artifact.txt")
member.size = len(content)
with tarfile.open(archive_path, "w:gz") as archive:
archive.addfile(member, io.BytesIO(content))
escaped = tarfile.TarInfo(f"sysdiagnose/../../{escaped_path.name}")
escaped.size = len(content)
archive.addfile(escaped, io.BytesIO(content))
link = tarfile.TarInfo("sysdiagnose/link")
link.type = tarfile.SYMTYPE
link.linkname = "/etc/hostname"
archive.addfile(link)
command = CmdIOSCheckSysdiagnose(target_path=str(archive_path))
try:
command.init()
extracted_path = Path(command.extracted_sysdiagnose_path)
assert (extracted_path / "artifact.txt").read_bytes() == content
assert not escaped_path.exists()
assert not (extracted_path / "link").exists()
module = SysdiagnoseExtraction()
command.module_init(module)
assert module.tar is None
assert module.parent_path == str(extracted_path.parent)
finally:
command.finish()
assert not extracted_path.exists()
+78
View File
@@ -0,0 +1,78 @@
# 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 click.testing import CliRunner
from mvt.android.cli import cli as android_cli
from mvt.ios.cli import cli as ios_cli
class TestCompletionCommand:
def test_completion_prints_instructions_by_default(self):
runner = CliRunner()
result = runner.invoke(ios_cli, ["completion"])
assert result.exit_code == 0
assert "Shell completion for mvt-ios" in result.output
assert "mvt-ios completion bash > ~/.mvt-ios-complete.bash" in result.output
assert "Mobile Verification Toolkit" not in result.output
def test_completion_prints_bash_script(self):
runner = CliRunner()
result = runner.invoke(ios_cli, ["completion", "bash"])
assert result.exit_code == 0
assert "_MVT_IOS_COMPLETE=bash_complete" in result.output
assert "complete -o nosort" in result.output
assert "mvt-ios" in result.output
assert "Mobile Verification Toolkit" not in result.output
def test_completion_prints_fish_script(self):
runner = CliRunner()
result = runner.invoke(android_cli, ["completion", "fish"])
assert result.exit_code == 0
assert "_MVT_ANDROID_COMPLETE=fish_complete" in result.output
assert "complete --no-files --command mvt-android" in result.output
assert "Mobile Verification Toolkit" not in result.output
def test_completion_install_updates_bashrc_once(self, tmp_path, monkeypatch):
monkeypatch.setenv("HOME", str(tmp_path))
runner = CliRunner()
result = runner.invoke(ios_cli, ["completion", "bash", "--install"])
assert result.exit_code == 0
script_path = tmp_path / ".mvt-ios-complete.bash"
bashrc_path = tmp_path / ".bashrc"
assert script_path.exists()
assert "_MVT_IOS_COMPLETE=bash_complete" in script_path.read_text(
encoding="utf-8"
)
bashrc = bashrc_path.read_text(encoding="utf-8")
assert "[ -f" in bashrc
assert ".mvt-ios-complete.bash" in bashrc
result = runner.invoke(ios_cli, ["completion", "bash", "--install"])
assert result.exit_code == 0
assert bashrc_path.read_text(encoding="utf-8") == bashrc
def test_completion_install_fish_does_not_update_shell_rc(
self, tmp_path, monkeypatch
):
monkeypatch.setenv("HOME", str(tmp_path))
runner = CliRunner()
result = runner.invoke(android_cli, ["completion", "fish", "--install"])
assert result.exit_code == 0
script_path = (
tmp_path / ".config" / "fish" / "completions" / "mvt-android.fish"
)
assert script_path.exists()
assert "_MVT_ANDROID_COMPLETE=fish_complete" in script_path.read_text(
encoding="utf-8"
)
assert not (tmp_path / ".fishrc").exists()
+195
View File
@@ -0,0 +1,195 @@
from click.testing import CliRunner
from mvt.android.cli import check_bugreport
from mvt.android.cmd_check_androidqf import CmdAndroidCheckAndroidQF
from mvt.android.cmd_check_backup import CmdAndroidCheckBackup
from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
from mvt.android.cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
from mvt.common.module import MVTModule
from mvt.ios.cli import check_backup, check_fs
CUSTOM_MODULE = """
from mvt.common.module import MVTModule
class {name}(MVTModule):
supported_commands = {supported_commands!r}
slug = "{slug}"
def run(self):
self.results = [{{"message": "custom module ran"}}]
def check_indicators(self):
pass
def serialize(self, result):
return None
"""
def _write_custom_module(path, name, supported_commands, slug=None):
path.write_text(
CUSTOM_MODULE.format(
name=name,
supported_commands=supported_commands,
slug=slug or name.lower(),
),
encoding="utf-8",
)
return path
def test_load_module_appears_only_for_supported_cli_command(tmp_path):
module_path = _write_custom_module(
tmp_path / "custom.py",
"IOSBackupOnlyModule",
(("ios", "check-backup"),),
)
backup_result = CliRunner().invoke(
check_backup,
["--list-modules", "--load-module", str(module_path), str(tmp_path)],
)
fs_result = CliRunner().invoke(
check_fs,
["--list-modules", "--load-module", str(module_path), str(tmp_path)],
)
assert backup_result.exit_code == 0
assert "IOSBackupOnlyModule" in backup_result.output
assert fs_result.exit_code == 0
assert "IOSBackupOnlyModule" not in fs_result.output
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",
(("ios", "check-backup"),),
slug="custom_run_module",
)
output_path = tmp_path / "out"
result = CliRunner().invoke(
check_backup,
[
"--module",
"CustomRunModule",
"--load-module",
str(module_path),
"--output",
str(output_path),
str(tmp_path),
],
)
assert result.exit_code == 0
assert (output_path / "custom_run_module.json").exists()
def test_custom_modules_load_from_environment_without_cli_flag(tmp_path, monkeypatch):
custom_modules_path = tmp_path / "custom_modules"
custom_modules_path.mkdir()
_write_custom_module(
custom_modules_path / "env_module.py",
"EnvBugreportModule",
(("android", "check-bugreport"),),
)
monkeypatch.setenv("MVT_CUSTOM_MODULES", str(custom_modules_path))
result = CliRunner().invoke(check_bugreport, ["--list-modules", str(tmp_path)])
assert result.exit_code == 0
assert "EnvBugreportModule" in result.output
class NestedBugreportModule(MVTModule):
supported_commands = (("android", "check-bugreport"),)
class NestedBackupModule(MVTModule):
supported_commands = (("android", "check-backup"),)
class NestedIntrusionLogsModule(MVTModule):
supported_commands = (("android", "check-intrusion-logs"),)
class NestedAndroidQFModule(MVTModule):
supported_commands = (("android", "check-androidqf"),)
class DummyZip:
def close(self):
pass
def test_androidqf_propagates_custom_modules_to_nested_commands(tmp_path, monkeypatch):
records = {}
custom_modules = [
NestedBugreportModule,
NestedBackupModule,
NestedIntrusionLogsModule,
NestedAndroidQFModule,
]
cmd = CmdAndroidCheckAndroidQF(
target_path=str(tmp_path),
custom_modules=custom_modules,
)
def record_available(name):
def _record(command):
records[name] = [
module.__name__
for module in command._available_modules()
if module.__name__.startswith("Nested")
]
return _record
monkeypatch.setattr(cmd, "load_bugreport", lambda: DummyZip())
monkeypatch.setattr(
CmdAndroidCheckBugreport,
"from_zip",
lambda self, bugreport: None,
)
monkeypatch.setattr(
CmdAndroidCheckBugreport,
"run",
record_available("bugreport"),
)
monkeypatch.setattr(cmd, "load_backup", lambda: b"")
monkeypatch.setattr(CmdAndroidCheckBackup, "from_ab", lambda self, backup: None)
monkeypatch.setattr(
CmdAndroidCheckBackup,
"run",
record_available("backup"),
)
intrusion_logs_path = tmp_path / "intrusion_logs"
intrusion_logs_path.mkdir()
setattr(cmd, "_CmdAndroidCheckAndroidQF__format", "dir")
setattr(
cmd,
"_CmdAndroidCheckAndroidQF__files",
["androidqf/intrusion_logs/security.txt"],
)
monkeypatch.setattr(cmd, "_read_device_timezone", lambda: None)
monkeypatch.setattr(
CmdAndroidCheckIntrusionLogs,
"run",
record_available("intrusion_logs"),
)
assert cmd.run_bugreport_cmd()
assert cmd.run_backup_cmd()
assert cmd.run_intrusion_logs_cmd()
assert records == {
"bugreport": ["NestedBugreportModule"],
"backup": ["NestedBackupModule"],
"intrusion_logs": ["NestedIntrusionLogsModule"],
}
+17
View File
@@ -4,6 +4,7 @@
# https://license.mvt.re/1.1/
import os
import sqlite3
from pathlib import Path
@@ -37,6 +38,22 @@ 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()
# Checkpoint the test change so the fixture does not retain SQLite sidecars.
conn.execute("PRAGMA journal_mode=DELETE;")
conn.close()
def delete_tmp_db_files(file_path):
"""
Remove Sqlite temporary files that appear on some platforms