Merge branch 'main' into fix/dumpsys-battery-daily-order

tests/android/test_artifact_dumpsys_accessibility.py

@@ -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")
@@ -51,4 +65,12 @@ class TestDumpsysAccessibilityArtifact:
         da.indicators = ind
         assert len(da.alertstore.alerts) == 0
         da.check_indicators()
-        assert len(da.alertstore.alerts) == 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"

tests/android/test_intrusion_logs.py

@@ -6,12 +6,14 @@
 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):
@@ -143,6 +145,58 @@ def test_check_intrusion_logs_parses_core_and_unknown_security_events(
     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", [])
 
@@ -152,3 +206,82 @@ def test_check_intrusion_logs_cli_lists_modules(tmp_path):
     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
+
+
+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)

tests/android_androidqf/test_settings.py

@@ -21,4 +21,5 @@ class TestSettingsModule:
         run_module(m)
         assert len(m.results) == 1
         assert "random" in m.results.keys()
-        assert len(m.alertstore.alerts) == 0
+        assert len(m.alertstore.alerts) == 1
+        assert "samsung_errorlog_agree" in m.alertstore.alerts[0].message

tests/common/test_command.py

@@ -4,11 +4,59 @@
 #   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 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(
@@ -21,3 +69,66 @@ class TestCommand:
 
         alerts = json.loads((tmp_path / "alerts.json").read_text())
         assert alerts[0]["event"]["payload"] == "\\xa8\\xa9"
+
+    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

tests/ios_backup/test_webkit_resource_load_statistics.py

@@ -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,
@@ -19,3 +21,50 @@ class TestWebkitResourceLoadStatisticsModule:
         assert len(m.results) == 2
         assert len(m.timeline) == 2
         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