Fix module audit findings (#850)

* Fix module audit findings

* Always parse paired tombstones

tests/android/test_artifact_dumpsys_battery_history.py

@@ -42,3 +42,22 @@ class TestDumpsysBatteryHistoryArtifact:
         assert len(dba.alertstore.alerts) == 0
         dba.check_indicators()
         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"

tests/android/test_artifact_dumpsys_dbinfo.py

@@ -40,3 +40,22 @@ class TestDumpsysDBinfoArtifact:
         assert len(dbi.alertstore.alerts) == 0
         dbi.check_indicators()
         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",
+            }
+        ]

tests/android/test_backup_parser.py

@@ -5,7 +5,10 @@
 
 import hashlib
 
+import pytest
+
 from mvt.android.parsers.backup import (
+    AndroidBackupParsingError,
     parse_ab_header,
     parse_backup_file,
     parse_tar_for_sms,
@@ -23,6 +26,14 @@ class TestBackupParsing:
             "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:

tests/ios_backup/test_calendar.py

@@ -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
@@ -21,6 +22,18 @@ class TestCalendarModule:
         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):
         m = Calendar(target_path=get_ios_backup_folder())
         ind = Indicators(log=logging.getLogger())

tests/ios_backup/test_sqlite_handling.py

@@ -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"

tests/test_check_android_androidqf.py

@@ -6,10 +6,15 @@
 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
@@ -18,6 +23,9 @@ 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")
@@ -89,3 +97,25 @@ class TestCheckAndroidqfCommand:
         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

tests/test_check_android_bugreport.py

@@ -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

tests/utils.py

@@ -49,8 +49,7 @@ def add_backup_manifest_entry(backup_path, file_id, domain, relative_path):
         (file_id, domain, relative_path, b""),
     )
     conn.commit()
-    # MVT opens backup databases with immutable=1, which ignores any -wal file,
-    # so the new row has to be checkpointed into Manifest.db itself.
+    # Checkpoint the test change so the fixture does not retain SQLite sidecars.
     conn.execute("PRAGMA journal_mode=DELETE;")
     conn.close()