Files
mvt/tests/android/test_intrusion_logs.py
T

334 lines
10 KiB
Python

# 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)