Merge branch 'main' into feature/custom-cli-commands

This commit is contained in:
besendorf
2026-08-05 23:21:38 +02:00
committed by GitHub
18 changed files with 517 additions and 4 deletions
+118
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
@@ -114,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 == []
+24
View File
@@ -42,6 +42,15 @@ 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"),)
@@ -87,6 +96,21 @@ class TestCommand:
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]
+9
View File
@@ -197,6 +197,15 @@ class TestIndicators:
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",
+8
View File
@@ -18,6 +18,14 @@ class TestSMSModule:
run_module(m)
assert len(m.results) == 1
assert len(m.timeline) == 2
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):
+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",
}
]
+62
View File
@@ -3,6 +3,7 @@
# 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
@@ -32,6 +33,67 @@ class TestCheckAndroidqfCommand:
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(