From 2dfe3cbcb19e40bf49ffa3b836e1696f9212327c Mon Sep 17 00:00:00 2001 From: besendorf Date: Tue, 28 Jul 2026 18:58:46 +0200 Subject: [PATCH] Fix module audit findings (#850) * Fix module audit findings * Always parse paired tombstones --- .../artifacts/dumpsys_battery_history.py | 29 +++++-- src/mvt/android/artifacts/dumpsys_dbinfo.py | 44 ++++------ src/mvt/android/cli.py | 6 +- src/mvt/android/cmd_check_androidqf.py | 13 ++- src/mvt/android/modules/androidqf/__init__.py | 2 + src/mvt/android/parsers/backup.py | 39 +++++---- src/mvt/ios/modules/base.py | 83 +++++++++++++----- src/mvt/ios/modules/fs/analytics.py | 2 + src/mvt/ios/modules/net_base.py | 7 +- .../test_artifact_dumpsys_battery_history.py | 19 +++++ tests/android/test_artifact_dumpsys_dbinfo.py | 19 +++++ tests/android/test_backup_parser.py | 11 +++ tests/ios_backup/test_calendar.py | 13 +++ tests/ios_backup/test_sqlite_handling.py | 85 +++++++++++++++++++ tests/test_check_android_androidqf.py | 30 +++++++ tests/test_check_android_bugreport.py | 10 +++ tests/utils.py | 3 +- 17 files changed, 333 insertions(+), 82 deletions(-) create mode 100644 tests/ios_backup/test_sqlite_handling.py diff --git a/src/mvt/android/artifacts/dumpsys_battery_history.py b/src/mvt/android/artifacts/dumpsys_battery_history.py index fb42903..77d9d3c 100644 --- a/src/mvt/android/artifacts/dumpsys_battery_history.py +++ b/src/mvt/android/artifacts/dumpsys_battery_history.py @@ -31,21 +31,38 @@ class DumpsysBatteryHistoryArtifact(AndroidArtifact): if line.strip() == "": break - time_elapsed = line.strip().split(" ", 1)[0] + 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]) event = "" if line.find("+job") > 0: event = "start_job" - uid = line[line.find("+job") + 5 : line.find(":")] - service = line[line.find(":") + 1 :].strip('"') + payload = line.split("+job=", 1)[1] + uid, separator, service = payload.partition(":") + if not separator: + continue + service = service.strip().strip('"') package_name = service.split("/")[0] elif line.find("-job") > 0: event = "end_job" - uid = line[line.find("-job") + 5 : line.find(":")] - service = line[line.find(":") + 1 :].strip('"') + payload = line.split("-job=", 1)[1] + uid, separator, service = payload.partition(":") + if not separator: + continue + service = service.strip().strip('"') package_name = service.split("/")[0] elif line.find("+running +wake_lock=") > 0: - uid = line[line.find("+running +wake_lock=") + 21 : line.find(":")] + payload = line.split("+running +wake_lock=", 1)[1] + uid, separator, _ = payload.partition(":") + if not separator: + continue event = "wake" service = ( line[line.find("*walarm*:") + 9 :].split(" ")[0].strip('"').strip() diff --git a/src/mvt/android/artifacts/dumpsys_dbinfo.py b/src/mvt/android/artifacts/dumpsys_dbinfo.py index a3a4eb1..050582d 100644 --- a/src/mvt/android/artifacts/dumpsys_dbinfo.py +++ b/src/mvt/android/artifacts/dumpsys_dbinfo.py @@ -29,10 +29,9 @@ class DumpsysDBInfoArtifact(AndroidArtifact): def parse(self, output: str) -> None: rxp = re.compile( - 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\=\"(.+?)\"" + 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=\"(.+?)\"" ) # pylint: disable=line-too-long pool = None @@ -56,29 +55,16 @@ class DumpsysDBInfoArtifact(AndroidArtifact): pool = None continue - matches = rxp.findall(line) - if not matches: - matches = rxp_no_pid.findall(line) - if not matches: - continue + match = rxp.match(line) + if not match: + continue - 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, - } - ) + 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) diff --git a/src/mvt/android/cli.py b/src/mvt/android/cli.py index 1123742..2830c98 100644 --- a/src/mvt/android/cli.py +++ b/src/mvt/android/cli.py @@ -4,6 +4,7 @@ # https://license.mvt.re/1.1/ import logging +from zipfile import BadZipFile import click @@ -212,7 +213,10 @@ def check_bugreport( log.info("Checking Android bug report at path: %s", bugreport_path) - cmd.run() + try: + cmd.run() + except BadZipFile as exc: + raise click.ClickException(f"Invalid bugreport archive: {exc}") from exc cmd.show_alerts_brief() cmd.show_support_message() diff --git a/src/mvt/android/cmd_check_androidqf.py b/src/mvt/android/cmd_check_androidqf.py index 41ad802..319b614 100644 --- a/src/mvt/android/cmd_check_androidqf.py +++ b/src/mvt/android/cmd_check_androidqf.py @@ -296,6 +296,7 @@ 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/") @@ -303,9 +304,15 @@ class CmdAndroidCheckAndroidQF(Command): if not relative or relative.endswith("/"): continue - 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: + 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: dst.write(src.read()) intrusion_logs_path = temp_dir diff --git a/src/mvt/android/modules/androidqf/__init__.py b/src/mvt/android/modules/androidqf/__init__.py index 1803949..1d8d619 100644 --- a/src/mvt/android/modules/androidqf/__init__.py +++ b/src/mvt/android/modules/androidqf/__init__.py @@ -5,6 +5,7 @@ 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 @@ -17,6 +18,7 @@ ANDROIDQF_MODULES = [ AQFGetProp, AQFSettings, AQFFiles, + AQFLogTimestamps, RootBinaries, Mounts, ] diff --git a/src/mvt/android/parsers/backup.py b/src/mvt/android/parsers/backup.py index 8c9bb5c..1f8da9b 100644 --- a/src/mvt/android/parsers/backup.py +++ b/src/mvt/android/parsers/backup.py @@ -131,20 +131,23 @@ def decrypt_backup_data(encrypted_backup, password, encryption_algo, format_vers if password is None: raise InvalidBackupPassword() - [ - user_salt, - checksum_salt, - pbkdf2_rounds, - user_iv, - master_key_blob, - encrypted_data, - ] = encrypted_backup.split(b"\n", 5) + try: + [ + 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")) + 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 # Derive decryption master key from password. master_key, master_iv = decrypt_master_key( @@ -174,10 +177,12 @@ def parse_backup_file(data, password=None): if not data.startswith(b"ANDROID BACKUP"): raise AndroidBackupParsingError("Invalid file header") - [_, version, is_compressed, encryption_algo, tar_data] = data.split(b"\n", 4) - - version = int(version) - is_compressed = int(is_compressed) + 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 if encryption_algo != b"none": tar_data = decrypt_backup_data( diff --git a/src/mvt/ios/modules/base.py b/src/mvt/ios/modules/base.py index 121bad5..a5c3354 100644 --- a/src/mvt/ios/modules/base.py +++ b/src/mvt/ios/modules/base.py @@ -9,6 +9,7 @@ import os import shutil import sqlite3 import subprocess +import tempfile from pathlib import Path from typing import Iterator, Optional, Union @@ -20,6 +21,20 @@ 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.""" @@ -44,6 +59,8 @@ 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 @@ -53,17 +70,12 @@ 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: - # 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) + 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): @@ -82,27 +94,54 @@ class IOSExtraction(MVTModule): raise DatabaseCorruptedError( "failed to recover without sqlite3 binary: please install sqlite3!" ) - 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) + 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)) ret = subprocess.call( - ["sqlite3", bak_path, f'.clone "{file_path}"'], + ["sqlite3", str(source_path), f'.clone "{recovered_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: - return sqlite3.connect(f"file:{file_path}?immutable=1", uri=True) + 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 def _get_backup_files_from_manifest( self, relative_path: Optional[str] = None, domain: Optional[str] = None @@ -166,7 +205,11 @@ 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 @@ -197,9 +240,9 @@ class IOSExtraction(MVTModule): :param backup_ids: Default value = None) """ - file_path: Optional[str] = None + file_path: Optional[str] = self.file_path # First we check if the was an explicit file path specified. - if not self.file_path: + if not 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 diff --git a/src/mvt/ios/modules/fs/analytics.py b/src/mvt/ios/modules/fs/analytics.py index 0a6e50c..2bf29e4 100644 --- a/src/mvt/ios/modules/fs/analytics.py +++ b/src/mvt/ios/modules/fs/analytics.py @@ -134,6 +134,8 @@ class Analytics(IOSExtraction): isodate = "" data = plistlib.loads(row[1]) data["isodate"] = isodate + else: + continue data["artifact"] = artifact diff --git a/src/mvt/ios/modules/net_base.py b/src/mvt/ios/modules/net_base.py index 99a851d..eb13fdf 100644 --- a/src/mvt/ios/modules/net_base.py +++ b/src/mvt/ios/modules/net_base.py @@ -42,7 +42,8 @@ class NetBase(IOSExtraction): ) def _extract_net_data(self): - conn = sqlite3.connect(self.file_path) + assert self.file_path is not None + conn = self._open_sqlite_db(self.file_path) conn.row_factory = sqlite3.Row cur = conn.cursor() try: @@ -237,9 +238,7 @@ 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( diff --git a/tests/android/test_artifact_dumpsys_battery_history.py b/tests/android/test_artifact_dumpsys_battery_history.py index 03d7d2e..dd0b120 100644 --- a/tests/android/test_artifact_dumpsys_battery_history.py +++ b/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" diff --git a/tests/android/test_artifact_dumpsys_dbinfo.py b/tests/android/test_artifact_dumpsys_dbinfo.py index 2becf65..691d43c 100644 --- a/tests/android/test_artifact_dumpsys_dbinfo.py +++ b/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", + } + ] diff --git a/tests/android/test_backup_parser.py b/tests/android/test_backup_parser.py index 6e5be72..c0f867d 100644 --- a/tests/android/test_backup_parser.py +++ b/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: diff --git a/tests/ios_backup/test_calendar.py b/tests/ios_backup/test_calendar.py index bf931f4..5f1035d 100644 --- a/tests/ios_backup/test_calendar.py +++ b/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()) diff --git a/tests/ios_backup/test_sqlite_handling.py b/tests/ios_backup/test_sqlite_handling.py new file mode 100644 index 0000000..f780640 --- /dev/null +++ b/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" diff --git a/tests/test_check_android_androidqf.py b/tests/test_check_android_androidqf.py index a2ed952..9b271be 100644 --- a/tests/test_check_android_androidqf.py +++ b/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 diff --git a/tests/test_check_android_bugreport.py b/tests/test_check_android_bugreport.py index b13c38d..bff4709 100644 --- a/tests/test_check_android_bugreport.py +++ b/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 diff --git a/tests/utils.py b/tests/utils.py index 9ae433d..d84dc2f 100644 --- a/tests/utils.py +++ b/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()