Parallelize URL indicator checks

src/mvt/android/artifacts/dumpsys_battery_history.py

@@ -31,38 +31,21 @@ class DumpsysBatteryHistoryArtifact(AndroidArtifact):
             if line.strip() == "":
                 break
 
-            time_parts = line.strip().split()
-            time_elapsed = time_parts[0]
-            if (
-                len(time_parts) > 1
-                and len(time_parts[0]) == 5
-                and time_parts[0][2] == "-"
-                and ":" in time_parts[1]
-            ):
-                time_elapsed = " ".join(time_parts[:2])
+            time_elapsed = line.strip().split(" ", 1)[0]
 
             event = ""
             if line.find("+job") > 0:
                 event = "start_job"
-                payload = line.split("+job=", 1)[1]
-                uid, separator, service = payload.partition(":")
-                if not separator:
-                    continue
-                service = service.strip().strip('"')
+                uid = line[line.find("+job") + 5 : line.find(":")]
+                service = line[line.find(":") + 1 :].strip('"')
                 package_name = service.split("/")[0]
             elif line.find("-job") > 0:
                 event = "end_job"
-                payload = line.split("-job=", 1)[1]
-                uid, separator, service = payload.partition(":")
-                if not separator:
-                    continue
-                service = service.strip().strip('"')
+                uid = line[line.find("-job") + 5 : line.find(":")]
+                service = line[line.find(":") + 1 :].strip('"')
                 package_name = service.split("/")[0]
             elif line.find("+running +wake_lock=") > 0:
-                payload = line.split("+running +wake_lock=", 1)[1]
-                uid, separator, _ = payload.partition(":")
-                if not separator:
-                    continue
+                uid = line[line.find("+running +wake_lock=") + 21 : line.find(":")]
                 event = "wake"
                 service = (
                     line[line.find("*walarm*:") + 9 :].split(" ")[0].strip('"').strip()

src/mvt/android/artifacts/dumpsys_dbinfo.py

@@ -29,9 +29,10 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
 
     def parse(self, output: str) -> None:
         rxp = re.compile(
-            r".*\[((?:[0-9]{4}-)?[0-9]{2}-[0-9]{2} "
-            r"[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\]\s*"
-            r"(?:\[Pid:\((\d+)\)\])?([\w-]+).*?sql=\"(.+?)\""
+            r".*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\].*\[Pid:\((\d+)\)\](\w+).*sql\=\"(.+?)\""
+        )  # pylint: disable=line-too-long
+        rxp_no_pid = re.compile(
+            r".*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\][ ]{1}(\w+).*sql\=\"(.+?)\""
         )  # pylint: disable=line-too-long
 
         pool = None
@@ -55,16 +56,29 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
                 pool = None
                 continue
 
-            match = rxp.match(line)
-            if not match:
-                continue
+            matches = rxp.findall(line)
+            if not matches:
+                matches = rxp_no_pid.findall(line)
+                if not matches:
+                    continue
 
-            result = {
-                "isodate": match.group(1),
-                "action": match.group(3),
-                "sql": match.group(4),
-                "path": pool,
-            }
-            if match.group(2):
-                result["pid"] = match.group(2)
-            self.results.append(result)
+                match = matches[0]
+                self.results.append(
+                    {
+                        "isodate": match[0],
+                        "action": match[1],
+                        "sql": match[2],
+                        "path": pool,
+                    }
+                )
+            else:
+                match = matches[0]
+                self.results.append(
+                    {
+                        "isodate": match[0],
+                        "pid": match[1],
+                        "action": match[2],
+                        "sql": match[3],
+                        "path": pool,
+                    }
+                )

src/mvt/android/artifacts/dumpsys_receivers.py

@@ -96,18 +96,6 @@ class DumpsysReceiversArtifact(AndroidArtifact):
                 self.results[intent] = []
                 continue
 
-            parts = line.strip().split(" ")
-            if len(parts) < 2:
-                # A single-token line here is not a receiver. Real dumpstate
-                # output can print an action header mis-indented (observed with
-                # 15 leading spaces instead of 6), which used to raise
-                # IndexError and abort the whole module. Treat a trailing-colon
-                # token as the next action, skip anything else.
-                if parts[0].endswith(":"):
-                    intent = parts[0][:-1]
-                    self.results.setdefault(intent, [])
-                continue
-
             # If we are not in an intent block yet, skip.
             if not intent:
                 continue
@@ -121,7 +109,7 @@ class DumpsysReceiversArtifact(AndroidArtifact):
 
             # If we got this far, we are processing receivers for the
             # activities we are interested in.
-            receiver = parts[1]
+            receiver = line.strip().split(" ")[1]
             package_name = receiver.split("/")[0]
 
             self.results[intent].append(

src/mvt/android/artifacts/tombstone_crashes.py

@@ -206,11 +206,6 @@ class TombstoneCrashArtifact(AndroidArtifact):
         return True
 
     def _load_pid_line(self, line: str, tombstone: dict) -> bool:
-        # The first pid line identifies the crashing thread. Full text tombstones
-        # contain additional pid lines for the other threads in the process.
-        if "pid" in tombstone:
-            return True
-
         try:
             parts = line.split(" >>> ") if " >>> " in line else line.split(">>>")
             process_info = parts[0]

src/mvt/android/cli.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-from zipfile import BadZipFile
 
 import click
 
@@ -213,10 +212,7 @@ def check_bugreport(
 
     log.info("Checking Android bug report at path: %s", bugreport_path)
 
-    try:
-        cmd.run()
-    except BadZipFile as exc:
-        raise click.ClickException(f"Invalid bugreport archive: {exc}") from exc
+    cmd.run()
     cmd.show_alerts_brief()
     cmd.show_support_message()
 

src/mvt/android/cmd_check_androidqf.py

@@ -296,7 +296,6 @@ class CmdAndroidCheckAndroidQF(Command):
 
             elif self.__format == "zip" and self.__zip:
                 temp_dir = tempfile.mkdtemp(prefix="mvt_intrusion_logs_")
-                temp_root = Path(temp_dir).resolve()
                 for entry in intrusion_log_files:
                     normalized = entry.replace("\\", "/")
                     idx = normalized.find("intrusion_logs/")
@@ -304,15 +303,9 @@ class CmdAndroidCheckAndroidQF(Command):
                     if not relative or relative.endswith("/"):
                         continue
 
-                    target = (temp_root / relative).resolve()
-                    if not target.is_relative_to(temp_root):
-                        self.log.warning(
-                            "Skipping unsafe intrusion log archive entry: %s", entry
-                        )
-                        continue
-
-                    target.parent.mkdir(parents=True, exist_ok=True)
-                    with self.__zip.open(entry) as src, target.open("wb") as dst:
+                    target = os.path.join(temp_dir, relative)
+                    os.makedirs(os.path.dirname(target), exist_ok=True)
+                    with self.__zip.open(entry) as src, open(target, "wb") as dst:
                         dst.write(src.read())
 
                 intrusion_logs_path = temp_dir

src/mvt/android/modules/androidqf/__init__.py

@@ -5,7 +5,6 @@
 
 from .aqf_files import AQFFiles
 from .aqf_getprop import AQFGetProp
-from .aqf_log_timestamps import AQFLogTimestamps
 from .aqf_packages import AQFPackages
 from .aqf_processes import AQFProcesses
 from .aqf_settings import AQFSettings
@@ -18,7 +17,6 @@ ANDROIDQF_MODULES = [
     AQFGetProp,
     AQFSettings,
     AQFFiles,
-    AQFLogTimestamps,
     RootBinaries,
     Mounts,
 ]

src/mvt/android/parsers/backup.py

@@ -131,23 +131,20 @@ def decrypt_backup_data(encrypted_backup, password, encryption_algo, format_vers
     if password is None:
         raise InvalidBackupPassword()
 
-    try:
-        [
-            user_salt,
-            checksum_salt,
-            pbkdf2_rounds,
-            user_iv,
-            master_key_blob,
-            encrypted_data,
-        ] = encrypted_backup.split(b"\n", 5)
+    [
+        user_salt,
+        checksum_salt,
+        pbkdf2_rounds,
+        user_iv,
+        master_key_blob,
+        encrypted_data,
+    ] = encrypted_backup.split(b"\n", 5)
 
-        user_salt = bytes.fromhex(user_salt.decode("utf-8"))
-        checksum_salt = bytes.fromhex(checksum_salt.decode("utf-8"))
-        pbkdf2_rounds = int(pbkdf2_rounds)
-        user_iv = bytes.fromhex(user_iv.decode("utf-8"))
-        master_key_blob = bytes.fromhex(master_key_blob.decode("utf-8"))
-    except (UnicodeDecodeError, ValueError) as exc:
-        raise AndroidBackupParsingError("Invalid encrypted backup header") from exc
+    user_salt = bytes.fromhex(user_salt.decode("utf-8"))
+    checksum_salt = bytes.fromhex(checksum_salt.decode("utf-8"))
+    pbkdf2_rounds = int(pbkdf2_rounds)
+    user_iv = bytes.fromhex(user_iv.decode("utf-8"))
+    master_key_blob = bytes.fromhex(master_key_blob.decode("utf-8"))
 
     # Derive decryption master key from password.
     master_key, master_iv = decrypt_master_key(
@@ -177,12 +174,10 @@ def parse_backup_file(data, password=None):
     if not data.startswith(b"ANDROID BACKUP"):
         raise AndroidBackupParsingError("Invalid file header")
 
-    try:
-        [_, version, is_compressed, encryption_algo, tar_data] = data.split(b"\n", 4)
-        version = int(version)
-        is_compressed = int(is_compressed)
-    except ValueError as exc:
-        raise AndroidBackupParsingError("Invalid file header") from exc
+    [_, version, is_compressed, encryption_algo, tar_data] = data.split(b"\n", 4)
+
+    version = int(version)
+    is_compressed = int(is_compressed)
 
     if encryption_algo != b"none":
         tar_data = decrypt_backup_data(

src/mvt/ios/modules/base.py

@@ -9,7 +9,6 @@ import os
 import shutil
 import sqlite3
 import subprocess
-import tempfile
 from pathlib import Path
 from typing import Iterator, Optional, Union
 
@@ -21,20 +20,6 @@ from mvt.common.module import (
 )
 
 
-class TemporarySQLiteConnection(sqlite3.Connection):
-    """SQLite connection that owns a temporary copy of a database."""
-
-    temporary_directory: Optional[tempfile.TemporaryDirectory] = None
-
-    def close(self) -> None:
-        try:
-            super().close()
-        finally:
-            if self.temporary_directory:
-                self.temporary_directory.cleanup()
-                self.temporary_directory = None
-
-
 class IOSExtraction(MVTModule):
     """This class provides a base for all iOS filesystem/backup extraction
     modules."""
@@ -59,8 +44,6 @@ class IOSExtraction(MVTModule):
 
         self.is_backup = False
         self.is_fs_dump = False
-        self._recovered_sqlite_paths: dict[str, str] = {}
-        self._sqlite_temp_directories: list[tempfile.TemporaryDirectory] = []
 
     def _recover_sqlite_db_if_needed(
         self, file_path: str, forced: bool = False
@@ -70,12 +53,17 @@ class IOSExtraction(MVTModule):
         :param file_path: Path to the malformed database file.
 
         """
+        # SQLite's immutable mode cannot open databases with active WAL files.
         if not forced:
-            conn = self._open_sqlite_db(file_path)
+            # If the database is open, do not use immutable
+            if os.path.isfile(file_path + "-shm"):
+                conn = sqlite3.connect(file_path)
+            else:
+                conn = self._open_sqlite_db(file_path)
             cur = conn.cursor()
 
-            recover = False
             try:
+                recover = False
                 cur.execute("SELECT name FROM sqlite_master WHERE type='table';")
             except sqlite3.DatabaseError as exc:
                 if "database disk image is malformed" in str(exc):
@@ -94,54 +82,27 @@ class IOSExtraction(MVTModule):
             raise DatabaseCorruptedError(
                 "failed to recover without sqlite3 binary: please install sqlite3!"
             )
-        temporary_directory = tempfile.TemporaryDirectory(prefix="mvt_sqlite_recover_")
-        temporary_path = Path(temporary_directory.name)
-        source_path = temporary_path / "source.db"
-        recovered_path = temporary_path / "recovered.db"
-        shutil.copy2(file_path, source_path)
-        for suffix in ("-wal", "-shm"):
-            sidecar = Path(file_path + suffix)
-            if sidecar.is_file():
-                shutil.copy2(sidecar, Path(str(source_path) + suffix))
+        if '"' in file_path:
+            raise DatabaseCorruptedError(
+                f"database at path '{file_path}' is corrupted. unable to "
+                'recover because it has a quotation mark (") in its name'
+            )
+
+        bak_path = f"{file_path}.bak"
+        shutil.move(file_path, bak_path)
 
         ret = subprocess.call(
-            ["sqlite3", str(source_path), f'.clone "{recovered_path}"'],
+            ["sqlite3", bak_path, f'.clone "{file_path}"'],
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
         )
         if ret != 0:
-            temporary_directory.cleanup()
             raise DatabaseCorruptedError("failed to recover database")
 
-        self._sqlite_temp_directories.append(temporary_directory)
-        self._recovered_sqlite_paths[file_path] = str(recovered_path)
         self.log.info("Database at path %s recovered successfully!", file_path)
 
     def _open_sqlite_db(self, file_path: str) -> sqlite3.Connection:
-        database_path = self._recovered_sqlite_paths.get(file_path, file_path)
-        if not os.path.isfile(database_path + "-wal"):
-            uri = Path(database_path).resolve().as_uri() + "?mode=ro&immutable=1"
-            return sqlite3.connect(uri, uri=True)
-
-        temporary_directory = tempfile.TemporaryDirectory(prefix="mvt_sqlite_")
-        temporary_path = Path(temporary_directory.name) / Path(database_path).name
-        shutil.copy2(database_path, temporary_path)
-        for suffix in ("-wal", "-shm"):
-            sidecar = Path(database_path + suffix)
-            if sidecar.is_file():
-                shutil.copy2(sidecar, Path(str(temporary_path) + suffix))
-
-        try:
-            conn = sqlite3.connect(
-                temporary_path.resolve().as_uri() + "?mode=ro",
-                uri=True,
-                factory=TemporarySQLiteConnection,
-            )
-        except Exception:
-            temporary_directory.cleanup()
-            raise
-        conn.temporary_directory = temporary_directory
-        return conn
+        return sqlite3.connect(f"file:{file_path}?immutable=1", uri=True)
 
     def _get_backup_files_from_manifest(
         self, relative_path: Optional[str] = None, domain: Optional[str] = None
@@ -205,11 +166,7 @@ class IOSExtraction(MVTModule):
         if not self.target_path:
             return None
         file_path = os.path.join(self.target_path, file_id[0:2], file_id)
-        if (
-            not Path(file_path)
-            .resolve()
-            .is_relative_to(Path(self.target_path).resolve())
-        ):
+        if not Path(file_path).resolve().is_relative_to(Path(self.target_path).resolve()):
             return None
         if os.path.exists(file_path):
             return file_path
@@ -240,9 +197,9 @@ class IOSExtraction(MVTModule):
         :param backup_ids: Default value = None)
 
         """
-        file_path: Optional[str] = self.file_path
+        file_path: Optional[str] = None
         # First we check if the was an explicit file path specified.
-        if not file_path:
+        if not self.file_path:
             # Type narrowing: we know self.file_path is None here, work with local file_path
             # If not, we first try with backups.
             # We construct the path to the file according to the iTunes backup

src/mvt/ios/modules/fs/analytics.py

@@ -134,8 +134,6 @@ class Analytics(IOSExtraction):
                 isodate = ""
                 data = plistlib.loads(row[1])
                 data["isodate"] = isodate
-            else:
-                continue
 
             data["artifact"] = artifact
 

src/mvt/ios/modules/mixed/safari_browserstate.py

@@ -19,17 +19,10 @@ from mvt.common.utils import convert_mactime_to_iso, keys_bytes_to_string
 
 from ..base import IOSExtraction
 
-# Safari profiles (iOS 17 and later) keep their per-profile databases under
-# Library/Safari/Profiles/<UUID>/, separate from the default profile's.
-SAFARI_BROWSER_STATE_BACKUP_RELPATHS = [
-    "Library/Safari/BrowserState.db",
-    "Library/Safari/Profiles/*/BrowserState.db",
-]
+SAFARI_BROWSER_STATE_BACKUP_RELPATH = "Library/Safari/BrowserState.db"
 SAFARI_BROWSER_STATE_ROOT_PATHS = [
     "private/var/mobile/Library/Safari/BrowserState.db",
-    "private/var/mobile/Library/Safari/Profiles/*/BrowserState.db",
     "private/var/mobile/Containers/Data/Application/*/Library/Safari/BrowserState.db",
-    "private/var/mobile/Containers/Data/Application/*/Library/Safari/Profiles/*/BrowserState.db",
 ]
 
 
@@ -181,22 +174,20 @@ class SafariBrowserState(IOSExtraction):
 
     def run(self) -> None:
         if self.is_backup:
-            for relative_path in SAFARI_BROWSER_STATE_BACKUP_RELPATHS:
-                for backup_file in self._get_backup_files_from_manifest(
-                    relative_path=relative_path
-                ):
-                    browserstate_path = self._get_backup_file_from_id(
-                        backup_file["file_id"]
-                    )
+            for backup_file in self._get_backup_files_from_manifest(
+                relative_path=SAFARI_BROWSER_STATE_BACKUP_RELPATH
+            ):
+                browserstate_path = self._get_backup_file_from_id(
+                    backup_file["file_id"]
+                )
 
-                    if not browserstate_path:
-                        continue
+                if not browserstate_path:
+                    continue
 
-                    self.log.info(
-                        "Found Safari browser state database at path: %s",
-                        browserstate_path,
-                    )
-                    self._process_browser_state_db(browserstate_path)
+                self.log.info(
+                    "Found Safari browser state database at path: %s", browserstate_path
+                )
+                self._process_browser_state_db(browserstate_path)
         elif self.is_fs_dump:
             for browserstate_path in self._get_fs_files_from_patterns(
                 SAFARI_BROWSER_STATE_ROOT_PATHS

src/mvt/ios/modules/mixed/safari_history.py

@@ -17,17 +17,10 @@ from mvt.common.utils import convert_mactime_to_datetime, convert_mactime_to_iso
 
 from ..base import IOSExtraction
 
-# Safari profiles (iOS 17 and later) each keep their own History.db under
-# Library/Safari/Profiles/<UUID>/, separate from the default profile's database.
-SAFARI_HISTORY_BACKUP_RELPATHS = [
-    "Library/Safari/History.db",
-    "Library/Safari/Profiles/*/History.db",
-]
+SAFARI_HISTORY_BACKUP_RELPATH = "Library/Safari/History.db"
 SAFARI_HISTORY_ROOT_PATHS = [
     "private/var/mobile/Library/Safari/History.db",
-    "private/var/mobile/Library/Safari/Profiles/*/History.db",
     "private/var/mobile/Containers/Data/Application/*/Library/Safari/History.db",
-    "private/var/mobile/Containers/Data/Application/*/Library/Safari/Profiles/*/History.db",
 ]
 
 
@@ -82,9 +75,6 @@ class SafariHistory(IOSExtraction):
 
             # We loop again through visits in order to find redirect record.
             for redirect in self.results:
-                if redirect["safari_history_db"] != result["safari_history_db"]:
-                    continue
-
                 if redirect["visit_id"] != result["redirect_destination"]:
                     continue
 
@@ -169,22 +159,17 @@ class SafariHistory(IOSExtraction):
 
     def run(self) -> None:
         if self.is_backup:
-            for relative_path in SAFARI_HISTORY_BACKUP_RELPATHS:
-                for history_file in self._get_backup_files_from_manifest(
-                    relative_path=relative_path
-                ):
-                    history_path = self._get_backup_file_from_id(
-                        history_file["file_id"]
-                    )
+            for history_file in self._get_backup_files_from_manifest(
+                relative_path=SAFARI_HISTORY_BACKUP_RELPATH
+            ):
+                history_path = self._get_backup_file_from_id(history_file["file_id"])
 
-                    if not history_path:
-                        continue
+                if not history_path:
+                    continue
 
-                    self.log.info(
-                        "Found Safari history database at path: %s", history_path
-                    )
+                self.log.info("Found Safari history database at path: %s", history_path)
 
-                    self._process_history_db(history_path)
+                self._process_history_db(history_path)
         elif self.is_fs_dump:
             for history_path in self._get_fs_files_from_patterns(
                 SAFARI_HISTORY_ROOT_PATHS

src/mvt/ios/modules/net_base.py

@@ -42,8 +42,7 @@ class NetBase(IOSExtraction):
         )
 
     def _extract_net_data(self):
-        assert self.file_path is not None
-        conn = self._open_sqlite_db(self.file_path)
+        conn = sqlite3.connect(self.file_path)
         conn.row_factory = sqlite3.Row
         cur = conn.cursor()
         try:
@@ -238,7 +237,9 @@ class NetBase(IOSExtraction):
                             "been truncated in the database)"
                         )
 
-                    self.alertstore.medium(msg, proc["live_isodate"], proc)
+                    self.alertstore.medium(
+                        msg, proc["live_isodate"], proc
+                    )
 
             if not proc["live_proc_id"]:
                 self.log.info(

tests/android/test_artifact_dumpsys_battery_history.py

@@ -42,22 +42,3 @@ 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,22 +40,3 @@ 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_artifact_dumpsys_receivers.py

@@ -31,44 +31,6 @@ 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")

tests/android/test_artifact_tombstones.py

@@ -42,32 +42,6 @@ class TestTombstoneCrashArtifact:
         assert len(tombstone_artifact.results) == 1
         self.validate_tombstone_result(tombstone_artifact.results[0])
 
-    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()

tests/android/test_backup_parser.py

@@ -5,10 +5,7 @@
 
 import hashlib
 
-import pytest
-
 from mvt.android.parsers.backup import (
-    AndroidBackupParsingError,
     parse_ab_header,
     parse_backup_file,
     parse_tar_for_sms,
@@ -26,14 +23,6 @@ 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,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-import os
 
 from mvt.common.indicators import Indicators
 from mvt.common.module import run_module
@@ -22,18 +21,6 @@ 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_safari_browserstate.py

@@ -4,44 +4,12 @@
 #   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 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)
+from ..utils import get_ios_backup_folder
 
 
 class TestSafariBrowserStateModule:
@@ -53,15 +21,6 @@ class TestSafariBrowserStateModule:
         assert len(m.timeline) == 1
         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())
         m.is_backup = True

tests/ios_backup/test_safari_history.py

@@ -1,165 +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 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

tests/ios_backup/test_sqlite_handling.py

@@ -1,85 +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 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,15 +6,10 @@
 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
@@ -23,9 +18,6 @@ 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")
@@ -97,25 +89,3 @@ 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,13 +18,3 @@ 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

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import os
-import sqlite3
 from pathlib import Path
 
 
@@ -38,22 +37,6 @@ 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