From 39a26d0f0be9d2696050a745b5becbc539e24d8a Mon Sep 17 00:00:00 2001 From: Janik Besendorf Date: Sun, 12 Apr 2026 10:51:56 +0200 Subject: [PATCH 01/16] Replace iOSbackup with iphone_backup_decrypt Replace the unmaintained iOSbackup dependency with iphone_backup_decrypt (MIT licensed, actively maintained). This fixes file corruption caused by iOSbackup truncating files to inaccurate sizes from backup metadata. The extract-key command and --key-file option are preserved via an MVTEncryptedBackup subclass that patches the keybag unlock to capture/reuse the derived PBKDF2 key. Closes #669 --- pyproject.toml | 3 +- src/mvt/ios/decrypt.py | 238 +++++++++++++++++++++++++++++++---------- 2 files changed, 182 insertions(+), 59 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index fedfff9a..e8ef917e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,7 +24,8 @@ dependencies = [ "simplejson==3.20.2", "packaging==26.0", "appdirs==1.4.4", - "iOSbackup==0.9.925", + "iphone_backup_decrypt==0.9.0", + "pycryptodome>=3.18", "adb-shell[usb]==0.4.4", "libusb1==3.3.1", "cryptography==46.0.6", diff --git a/src/mvt/ios/decrypt.py b/src/mvt/ios/decrypt.py index ffb2cb75..d615d051 100644 --- a/src/mvt/ios/decrypt.py +++ b/src/mvt/ios/decrypt.py @@ -6,17 +6,146 @@ import binascii import glob import logging -import multiprocessing import os import os.path +import plistlib import shutil import sqlite3 +import tempfile from typing import Optional -from iOSbackup import iOSbackup +from iphone_backup_decrypt import EncryptedBackup +from iphone_backup_decrypt import google_iphone_dataprotection log = logging.getLogger(__name__) +# Import pbkdf2_hmac from the same source iphone_backup_decrypt uses internally, +# so our key derivation is consistent with theirs. +try: + from fastpbkdf2 import pbkdf2_hmac +except ImportError: + import Crypto.Hash.SHA1 + import Crypto.Hash.SHA256 + import Crypto.Protocol.KDF + + _HASH_FNS = {"sha1": Crypto.Hash.SHA1, "sha256": Crypto.Hash.SHA256} + + def pbkdf2_hmac(hash_name, password, salt, iterations, dklen=None): + return Crypto.Protocol.KDF.PBKDF2( + password, salt, dklen, iterations, hmac_hash_module=_HASH_FNS[hash_name] + ) + + +class MVTEncryptedBackup(EncryptedBackup): + """Extends EncryptedBackup with derived key export/import. + + NOTE: This subclass relies on internal APIs of iphone_backup_decrypt + (specifically _read_and_unlock_keybag, _keybag, and the Keybag class + internals). Pinned to iphone_backup_decrypt==0.9.0. + """ + + def __init__(self, *, backup_directory, passphrase=None, derived_key=None): + if passphrase: + super().__init__(backup_directory=backup_directory, passphrase=passphrase) + self._derived_key = None # Will be set after keybag unlock + elif derived_key: + self._init_without_passphrase(backup_directory, derived_key) + else: + raise ValueError("Either passphrase or derived_key must be provided") + + def _init_without_passphrase(self, backup_directory, derived_key): + """Replicate parent __init__ state without requiring a passphrase.""" + self.decrypted = False + self._backup_directory = os.path.expandvars(backup_directory) + self._passphrase = None + self._manifest_plist_path = os.path.join( + self._backup_directory, "Manifest.plist" + ) + self._manifest_plist = None + self._manifest_db_path = os.path.join(self._backup_directory, "Manifest.db") + self._keybag = None + self._unlocked = False + self._temporary_folder = tempfile.mkdtemp() + self._temp_decrypted_manifest_db_path = os.path.join( + self._temporary_folder, "Manifest.db" + ) + self._temp_manifest_db_conn = None + self._derived_key = derived_key # 32 raw bytes + + def _read_and_unlock_keybag(self): + """Override to capture derived key on password unlock, or use + a pre-derived key to skip PBKDF2.""" + if self._unlocked: + return self._unlocked + + with open(self._manifest_plist_path, "rb") as infile: + self._manifest_plist = plistlib.load(infile) + self._keybag = google_iphone_dataprotection.Keybag( + self._manifest_plist["BackupKeyBag"] + ) + + if self._derived_key: + # Skip PBKDF2, unwrap class keys directly with pre-derived key + self._unlocked = _unlock_keybag_with_derived_key( + self._keybag, self._derived_key + ) + else: + # Normal path: full PBKDF2 derivation, capturing the intermediate key + self._unlocked, self._derived_key = _unlock_keybag_and_capture_key( + self._keybag, self._passphrase + ) + self._passphrase = None + + if not self._unlocked: + raise ValueError("Failed to decrypt keys: incorrect passphrase?") + return True + + def get_decryption_key(self): + """Return derived key as hex string (64 chars / 32 bytes).""" + if self._derived_key is None: + raise ValueError("No derived key available") + return self._derived_key.hex() + + +def _unlock_keybag_with_derived_key(keybag, passphrase_key): + """Unlock keybag class keys using a pre-derived passphrase_key, + skipping the expensive PBKDF2 rounds.""" + WRAP_PASSPHRASE = 2 + for classkey in keybag.classKeys.values(): + if b"WPKY" not in classkey: + continue + if classkey[b"WRAP"] & WRAP_PASSPHRASE: + k = google_iphone_dataprotection._AESUnwrap( + passphrase_key, classkey[b"WPKY"] + ) + if not k: + return False + classkey[b"KEY"] = k + return True + + +def _unlock_keybag_and_capture_key(keybag, passphrase): + """Run full PBKDF2 key derivation and AES unwrap, returning + (success, passphrase_key) so the derived key can be exported.""" + passphrase_round1 = pbkdf2_hmac( + "sha256", passphrase, keybag.attrs[b"DPSL"], keybag.attrs[b"DPIC"], 32 + ) + passphrase_key = pbkdf2_hmac( + "sha1", passphrase_round1, keybag.attrs[b"SALT"], keybag.attrs[b"ITER"], 32 + ) + WRAP_PASSPHRASE = 2 + for classkey in keybag.classKeys.values(): + if b"WPKY" not in classkey: + continue + if classkey[b"WRAP"] & WRAP_PASSPHRASE: + k = google_iphone_dataprotection._AESUnwrap( + passphrase_key, classkey[b"WPKY"] + ) + if not k: + return False, None + classkey[b"KEY"] = k + return True, passphrase_key + class DecryptBackup: """This class provides functions to decrypt an encrypted iTunes backup @@ -55,41 +184,27 @@ class DecryptBackup: log.critical("The backup does not seem encrypted!") return False - def _process_file( - self, relative_path: str, domain: str, item, file_id: str, item_folder: str - ) -> None: - self._backup.getFileDecryptedCopy( - manifestEntry=item, targetName=file_id, targetFolder=item_folder - ) - log.info( - "Decrypted file %s [%s] to %s/%s", - relative_path, - domain, - item_folder, - file_id, - ) - def process_backup(self) -> None: if not os.path.exists(self.dest_path): os.makedirs(self.dest_path) manifest_path = os.path.join(self.dest_path, "Manifest.db") - # We extract a decrypted Manifest.db. - self._backup.getManifestDB() - # We store it to the destination folder. - shutil.copy(self._backup.manifestDB, manifest_path) - - pool = multiprocessing.Pool(multiprocessing.cpu_count()) - - for item in self._backup.getBackupFilesList(): - try: - file_id = item["backupFile"] - relative_path = item["relativePath"] - domain = item["domain"] + # Extract a decrypted Manifest.db to the destination folder. + self._backup.save_manifest_file(output_filename=manifest_path) + # Iterate over all files in the backup and decrypt them, + # preserving the XX/file_id directory structure that downstream + # modules expect. + with self._backup.manifest_db_cursor() as cur: + cur.execute( + "SELECT fileID, domain, relativePath, file FROM Files WHERE flags=1" + ) + for file_id, domain, relative_path, file_bplist in cur: # This may be a partial backup. Skip files from the manifest # which do not exist locally. - source_file_path = os.path.join(self.backup_path, file_id[0:2], file_id) + source_file_path = os.path.join( + self.backup_path, file_id[:2], file_id + ) if not os.path.exists(source_file_path): log.debug( "Skipping file %s. File not found in encrypted backup directory.", @@ -97,24 +212,26 @@ class DecryptBackup: ) continue - item_folder = os.path.join(self.dest_path, file_id[0:2]) - if not os.path.exists(item_folder): - os.makedirs(item_folder) + item_folder = os.path.join(self.dest_path, file_id[:2]) + os.makedirs(item_folder, exist_ok=True) - # iOSBackup getFileDecryptedCopy() claims to read a "file" - # parameter but the code actually is reading the "manifest" key. - # Add manifest plist to both keys to handle this. - item["manifest"] = item["file"] - - pool.apply_async( - self._process_file, - args=(relative_path, domain, item, file_id, item_folder), - ) - except Exception as exc: - log.error("Failed to decrypt file %s: %s", relative_path, exc) - - pool.close() - pool.join() + try: + decrypted = self._backup._decrypt_inner_file( + file_id=file_id, file_bplist=file_bplist + ) + with open( + os.path.join(item_folder, file_id), "wb" + ) as handle: + handle.write(decrypted) + log.info( + "Decrypted file %s [%s] to %s/%s", + relative_path, + domain, + item_folder, + file_id, + ) + except Exception as exc: + log.error("Failed to decrypt file %s: %s", relative_path, exc) # Copying over the root plist files as well. for file_name in os.listdir(self.backup_path): @@ -155,20 +272,23 @@ class DecryptBackup: return try: - self._backup = iOSbackup( - udid=os.path.basename(self.backup_path), - cleartextpassword=password, - backuproot=os.path.dirname(self.backup_path), + self._backup = MVTEncryptedBackup( + backup_directory=self.backup_path, + passphrase=password, ) + # Eagerly trigger keybag unlock so wrong-password errors + # surface here rather than later during process_backup(). + self._backup.test_decryption() except Exception as exc: + self._backup = None if ( - isinstance(exc, KeyError) - and len(exc.args) > 0 - and exc.args[0] == b"KEY" + isinstance(exc, ValueError) + and "passphrase" in str(exc).lower() ): log.critical("Failed to decrypt backup. Password is probably wrong.") elif ( isinstance(exc, FileNotFoundError) + and hasattr(exc, "filename") and os.path.basename(exc.filename) == "Manifest.plist" ): log.critical( @@ -211,12 +331,14 @@ class DecryptBackup: try: key_bytes_raw = binascii.unhexlify(key_bytes) - self._backup = iOSbackup( - udid=os.path.basename(self.backup_path), - derivedkey=key_bytes_raw, - backuproot=os.path.dirname(self.backup_path), + self._backup = MVTEncryptedBackup( + backup_directory=self.backup_path, + derived_key=key_bytes_raw, ) + # Eagerly trigger keybag unlock so wrong-key errors surface here. + self._backup.test_decryption() except Exception as exc: + self._backup = None log.exception(exc) log.critical( "Failed to decrypt backup. Did you provide the correct key file?" @@ -227,7 +349,7 @@ class DecryptBackup: if not self._backup: return - self._decryption_key = self._backup.getDecryptionKey() + self._decryption_key = self._backup.get_decryption_key() log.info( 'Derived decryption key for backup at path %s is: "%s"', self.backup_path, From 545ac19158120eb13292d6df9ca4ef6f3ea94b3d Mon Sep 17 00:00:00 2001 From: Janik Besendorf Date: Mon, 17 Aug 2026 13:24:30 +0200 Subject: [PATCH 02/16] Restore concurrent backup decryption --- src/mvt/common/help.py | 1 + src/mvt/ios/cli.py | 18 +++- src/mvt/ios/decrypt.py | 147 +++++++++++++++++++++++-------- tests/ios_backup/test_decrypt.py | 68 +++++++++++++- 4 files changed, 195 insertions(+), 39 deletions(-) diff --git a/src/mvt/common/help.py b/src/mvt/common/help.py index 5101f939..74615382 100644 --- a/src/mvt/common/help.py +++ b/src/mvt/common/help.py @@ -28,6 +28,7 @@ HELP_MSG_DECRYPT_BACKUP = "Decrypt an encrypted iTunes backup" HELP_MSG_BACKUP_DESTINATION = ( "Path to the folder where the decrypted backup should be stored" ) +HELP_MSG_DECRYPT_JOBS = "Number of files to decrypt concurrently" HELP_MSG_IOS_BACKUP_PASSWORD = ( "Password to use to decrypt the backup (or, set the {MVT_IOS_BACKUP_PASSWORD} " "environment variable)" diff --git a/src/mvt/ios/cli.py b/src/mvt/ios/cli.py index c338fa54..0bd362a2 100644 --- a/src/mvt/ios/cli.py +++ b/src/mvt/ios/cli.py @@ -33,6 +33,7 @@ from mvt.common.help import ( HELP_MSG_VERSION, HELP_MSG_DECRYPT_BACKUP, HELP_MSG_BACKUP_DESTINATION, + HELP_MSG_DECRYPT_JOBS, HELP_MSG_IOS_BACKUP_PASSWORD, HELP_MSG_BACKUP_KEYFILE, HELP_MSG_HASHES, @@ -58,7 +59,11 @@ from mvt.common.password import prompt_password from .cmd_check_backup import CmdIOSCheckBackup from .cmd_check_fs import CmdIOSCheckFS from .cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose -from .decrypt import DecryptBackup +from .decrypt import ( + DEFAULT_DECRYPT_WORKERS, + MAX_DECRYPT_WORKERS, + DecryptBackup, +) from .modules.backup import BACKUP_MODULES from .modules.fs import FS_MODULES from .modules.mixed import MIXED_MODULES @@ -162,6 +167,13 @@ def completion(ctx, shell, install): "decrypt-backup", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_DECRYPT_BACKUP ) @click.option("--destination", "-d", required=True, help=HELP_MSG_BACKUP_DESTINATION) +@click.option( + "--jobs", + type=click.IntRange(1, MAX_DECRYPT_WORKERS), + default=DEFAULT_DECRYPT_WORKERS, + show_default=True, + help=HELP_MSG_DECRYPT_JOBS, +) @click.option( "--password", "-p", @@ -180,8 +192,8 @@ def completion(ctx, shell, install): @click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES) @click.argument("BACKUP_PATH", type=click.Path(exists=True)) @click.pass_context -def decrypt_backup(ctx, destination, password, key_file, hashes, backup_path): - backup = DecryptBackup(backup_path, destination) +def decrypt_backup(ctx, destination, jobs, password, key_file, hashes, backup_path): + backup = DecryptBackup(backup_path, destination, max_workers=jobs) if key_file: if MVT_IOS_BACKUP_PASSWORD in os.environ: diff --git a/src/mvt/ios/decrypt.py b/src/mvt/ios/decrypt.py index 2ed3efe8..faf3000e 100644 --- a/src/mvt/ios/decrypt.py +++ b/src/mvt/ios/decrypt.py @@ -12,6 +12,13 @@ import plistlib import shutil import sqlite3 import tempfile +from concurrent.futures import ( + ALL_COMPLETED, + FIRST_COMPLETED, + Future, + ThreadPoolExecutor, + wait, +) from pathlib import Path from typing import Optional @@ -21,6 +28,9 @@ from iphone_backup_decrypt.utils import FilePlist log = logging.getLogger(__name__) +DEFAULT_DECRYPT_WORKERS = 4 +MAX_DECRYPT_WORKERS = 32 + # Import pbkdf2_hmac from the same source iphone_backup_decrypt uses internally, # so our key derivation is consistent with theirs. try: @@ -178,19 +188,68 @@ class DecryptBackup: """ - def __init__(self, backup_path: str, dest_path: Optional[str] = None) -> None: + def __init__( + self, + backup_path: str, + dest_path: Optional[str] = None, + max_workers: int = DEFAULT_DECRYPT_WORKERS, + ) -> None: """Decrypts an encrypted iOS backup. :param backup_path: Path to the encrypted backup folder :param dest_path: Path to the folder where to store the decrypted backup """ self.backup_path = os.path.abspath(backup_path) self.dest_path = dest_path + if not 1 <= max_workers <= MAX_DECRYPT_WORKERS: + raise ValueError(f"max_workers must be between 1 and {MAX_DECRYPT_WORKERS}") + self.max_workers = max_workers self._backup: Optional[MVTEncryptedBackup] = None self._decryption_key: Optional[str] = None def can_process(self) -> bool: return self._backup is not None + def _process_file( + self, + *, + file_id: str, + file_bplist: bytes, + output_path: Path, + relative_path: str, + domain: str, + ) -> None: + assert self._backup is not None + self._backup.extract_file_by_id( + file_id=file_id, + file_bplist=file_bplist, + output_filename=str(output_path), + ) + log.info( + "Decrypted file %s [%s] to %s/%s", + relative_path, + domain, + output_path.parent, + file_id, + ) + + @staticmethod + def _wait_for_files( + pending: dict[Future[None], str], *, all_files: bool = False + ) -> None: + if not pending: + return + + done, _ = wait( + pending, + return_when=ALL_COMPLETED if all_files else FIRST_COMPLETED, + ) + for future in done: + relative_path = pending.pop(future) + try: + future.result() + except Exception as exc: + log.error("Failed to decrypt file %s: %s", relative_path, exc) + @staticmethod def is_encrypted(backup_path: str) -> bool: """Query Manifest.db file to see if it's encrypted or not. @@ -226,45 +285,63 @@ class DecryptBackup: # modules expect. backup_root = Path(self.backup_path).resolve() dest_root = Path(self.dest_path).resolve() - with self._backup.manifest_db_cursor() as cur: - cur.execute( - "SELECT fileID, domain, relativePath, file FROM Files WHERE flags=1" - ) - for file_id, domain, relative_path, file_bplist in cur: - # This may be a partial backup. Skip files from the manifest - # which do not exist locally. - source_file_path = backup_root / file_id[:2] / file_id - if not source_file_path.resolve().is_relative_to(backup_root): - log.warning("Skipping unsafe file_id: %r", file_id) - continue - if not os.path.exists(source_file_path): - log.debug( - "Skipping file %s. File not found in encrypted backup directory.", - source_file_path, - ) - continue + pending: dict[Future[None], str] = {} + with ThreadPoolExecutor(max_workers=self.max_workers) as executor: + with self._backup.manifest_db_cursor() as cur: + cur.execute( + "SELECT fileID, domain, relativePath, file FROM Files WHERE flags=1" + ) + for file_id, domain, relative_path, file_bplist in cur: + # This may be a partial backup. Skip files from the manifest + # which do not exist locally. + source_file_path = backup_root / file_id[:2] / file_id + if not source_file_path.resolve().is_relative_to(backup_root): + log.warning("Skipping unsafe file_id: %r", file_id) + continue + if not os.path.exists(source_file_path): + log.debug( + "Skipping file %s. File not found in encrypted " + "backup directory.", + source_file_path, + ) + continue - output_path = dest_root / file_id[:2] / file_id - if not output_path.resolve().is_relative_to(dest_root): - log.warning("Skipping unsafe file_id: %r", file_id) - continue - output_path.parent.mkdir(parents=True, exist_ok=True) + output_path = dest_root / file_id[:2] / file_id + if not output_path.resolve().is_relative_to(dest_root): + log.warning("Skipping unsafe file_id: %r", file_id) + continue + output_path.parent.mkdir(parents=True, exist_ok=True) - try: - self._backup.extract_file_by_id( + if self.max_workers == 1: + try: + self._process_file( + file_id=file_id, + file_bplist=file_bplist, + output_path=output_path, + relative_path=relative_path, + domain=domain, + ) + except Exception as exc: + log.error( + "Failed to decrypt file %s: %s", + relative_path, + exc, + ) + continue + + future = executor.submit( + self._process_file, file_id=file_id, file_bplist=file_bplist, - output_filename=str(output_path), + output_path=output_path, + relative_path=relative_path, + domain=domain, ) - log.info( - "Decrypted file %s [%s] to %s/%s", - relative_path, - domain, - output_path.parent, - file_id, - ) - except Exception as exc: - log.error("Failed to decrypt file %s: %s", relative_path, exc) + pending[future] = relative_path + if len(pending) >= self.max_workers: + self._wait_for_files(pending) + + self._wait_for_files(pending, all_files=True) # Copying over the root plist files as well. for file_name in os.listdir(self.backup_path): diff --git a/tests/ios_backup/test_decrypt.py b/tests/ios_backup/test_decrypt.py index 28eeb83b..25decce5 100644 --- a/tests/ios_backup/test_decrypt.py +++ b/tests/ios_backup/test_decrypt.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 logging +import threading from pathlib import Path from Crypto.Cipher import AES @@ -114,7 +116,9 @@ def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_pat Path(output_filename).write_bytes(b"decrypted") backup.extract_file_by_id.side_effect = extract_file_by_id - decryptor = DecryptBackup(str(backup_path), str(destination)) + decryptor = DecryptBackup( + str(backup_path), str(destination), max_workers=1 + ) decryptor._backup = backup decryptor.process_backup() @@ -123,3 +127,65 @@ def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_pat assert not (outside / symlink_file_id).exists() backup.extract_file_by_id.assert_called_once() assert backup.extract_file_by_id.call_args.kwargs["file_id"] == safe_file_id + + +def test_process_backup_decrypts_files_concurrently(mocker, tmp_path): + backup_path = tmp_path / "backup" + destination = tmp_path / "destination" + backup_path.mkdir() + + file_ids = ["ab" + "1" * 38, "cd" + "2" * 38] + for file_id in file_ids: + source_path = backup_path / file_id[:2] / file_id + source_path.parent.mkdir() + source_path.write_bytes(b"encrypted") + + cursor = mocker.MagicMock() + cursor.__iter__.return_value = iter( + (file_id, "Domain", file_id, b"plist") for file_id in file_ids + ) + cursor_context = mocker.MagicMock() + cursor_context.__enter__.return_value = cursor + + barrier = threading.Barrier(2) + backup = mocker.MagicMock() + backup.manifest_db_cursor.return_value = cursor_context + + def extract_file_by_id(*, file_id, output_filename, **kwargs): + barrier.wait(timeout=5) + Path(output_filename).write_bytes(file_id.encode()) + + backup.extract_file_by_id.side_effect = extract_file_by_id + decryptor = DecryptBackup(str(backup_path), str(destination), max_workers=2) + decryptor._backup = backup + + decryptor.process_backup() + + for file_id in file_ids: + assert (destination / file_id[:2] / file_id).read_bytes() == file_id.encode() + + +def test_process_backup_logs_worker_errors(mocker, tmp_path, caplog): + backup_path = tmp_path / "backup" + destination = tmp_path / "destination" + backup_path.mkdir() + file_id = "ef" + "3" * 38 + source_path = backup_path / file_id[:2] / file_id + source_path.parent.mkdir() + source_path.write_bytes(b"encrypted") + + cursor = mocker.MagicMock() + cursor.__iter__.return_value = iter([(file_id, "Domain", "failing-file", b"plist")]) + cursor_context = mocker.MagicMock() + cursor_context.__enter__.return_value = cursor + + backup = mocker.MagicMock() + backup.manifest_db_cursor.return_value = cursor_context + backup.extract_file_by_id.side_effect = ValueError("broken file") + decryptor = DecryptBackup(str(backup_path), str(destination)) + decryptor._backup = backup + + with caplog.at_level(logging.ERROR, logger="mvt.ios.decrypt"): + decryptor.process_backup() + + assert "Failed to decrypt file failing-file: broken file" in caplog.text From 531d63ab071dc039f0837b94626edc6816f2eafa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Tue, 1 Sep 2026 15:49:25 +0200 Subject: [PATCH 03/16] Release monthly from a scheduled workflow The package version now comes from the latest v* git tag, so a release is a tag and nothing else. A scheduled workflow tags main on the first of the month when something was merged since the last release, creates the GitHub release with generated notes, and publishes to PyPI and the container registry. Pushing a v* tag by hand goes through the same path. --- .github/workflows/publish-release-docker.yml | 2 + .github/workflows/release.yml | 44 ++++++++++++++++++++ Dockerfile | 1 + Makefile | 5 --- pyproject.toml | 7 ++-- src/mvt/common/version.py | 4 +- 6 files changed, 54 insertions(+), 9 deletions(-) create mode 100644 .github/workflows/release.yml diff --git a/.github/workflows/publish-release-docker.yml b/.github/workflows/publish-release-docker.yml index b3fde3ae..a90202d2 100644 --- a/.github/workflows/publish-release-docker.yml +++ b/.github/workflows/publish-release-docker.yml @@ -37,6 +37,8 @@ jobs: steps: - name: Checkout repository uses: actions/checkout@v7 + with: + fetch-depth: 0 # the package version comes from the git tags # Uses the `docker/login-action` action to log in to the Container registry registry using the account and password that will publish the packages. Once published, the packages are scoped to the account defined here. - name: Log in to the Container registry uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 00000000..adec25c3 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,44 @@ +name: Release + +on: + schedule: + - cron: "0 9 1 * *" # first of the month + workflow_dispatch: + push: + tags: ["v*"] + +jobs: + tag: + if: github.ref_type != 'tag' + runs-on: ubuntu-latest + permissions: + contents: write + actions: write + env: + GH_TOKEN: ${{ github.token }} + steps: + - uses: actions/checkout@v7 + with: + fetch-depth: 0 + - run: | + last=$(git describe --tags --abbrev=0 --match 'v*') + if [ "$(git rev-list "$last..HEAD" --count)" = 0 ]; then + echo "Nothing merged since $last"; exit 0 + fi + tag=v$(date -u +%Y.%-m.%-d) + gh release create "$tag" --target "$GITHUB_SHA" --generate-notes + # Events made with GITHUB_TOKEN don't start other workflows; run them by hand. + gh workflow run release.yml --ref "$tag" + gh workflow run publish-release-docker.yml --ref "$tag" + + publish: + if: github.ref_type == 'tag' + runs-on: ubuntu-latest + permissions: + id-token: write # PyPI trusted publishing + steps: + - uses: actions/checkout@v7 + with: + fetch-depth: 0 + - uses: astral-sh/setup-uv@v10.0.0 + - run: uv build && uv publish diff --git a/Dockerfile b/Dockerfile index 58bc2166..138edc7e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -122,6 +122,7 @@ RUN apt-get update \ binutils \ default-jre-headless \ file \ + git \ jq \ less \ libcurl4 \ diff --git a/Makefile b/Makefile index 528c6975..62140097 100644 --- a/Makefile +++ b/Makefile @@ -32,8 +32,3 @@ clean: dist: $(UV) build -upload: - $(UV) tool run twine upload dist/* - -test-upload: - $(UV) tool run twine upload --repository testpypi dist/* diff --git a/pyproject.toml b/pyproject.toml index fb90f8db..3124c465 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -72,7 +72,7 @@ docs = [ ] [build-system] -requires = ["setuptools>=61.0"] +requires = ["setuptools>=61.0", "setuptools-scm>=8"] build-backend = "setuptools.build_meta" [tool.coverage.run] @@ -121,5 +121,6 @@ where = ["src"] [tool.setuptools.package-data] mvt = ["ios/data/*.json"] -[tool.setuptools.dynamic] -version = { attr = "mvt.common.version.MVT_VERSION" } +[tool.setuptools_scm] +# The version is the latest v* tag; ignore the archive/* tags. +git_describe_command = "git describe --dirty --tags --long --match 'v*'" diff --git a/src/mvt/common/version.py b/src/mvt/common/version.py index c8e77161..31496047 100644 --- a/src/mvt/common/version.py +++ b/src/mvt/common/version.py @@ -3,4 +3,6 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ -MVT_VERSION = "2026.7.29" +from importlib.metadata import version + +MVT_VERSION = version("mvt") From 27cfe7f1e7d98919a6db3b221651e06d2d90b4ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Tue, 1 Sep 2026 19:31:23 +0200 Subject: [PATCH 04/16] Release weekly instead of monthly --- .github/workflows/release.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index adec25c3..3c20dc71 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -2,7 +2,7 @@ name: Release on: schedule: - - cron: "0 9 1 * *" # first of the month + - cron: "0 9 * * 1" # Mondays, 09:00 UTC workflow_dispatch: push: tags: ["v*"] From 0e231eefaf4b26dff35de02b5de2fe1cb91fd52c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Fri, 4 Sep 2026 17:44:51 +0200 Subject: [PATCH 05/16] Parse dumpsys settings as per-record results The `dumpsys settings` parser matched a single regex per line, which truncated every value that spans more than one line and, when `defaultSystemSet:` did not fall on the first line, left the trailing `default:` metadata inside the value. It also keyed results by setting name within a namespace, so a name recorded twice kept only the last row and a row without a `pkg:` field was dropped entirely. Replace it with a line loop that accumulates one record at a time and splits the `key:value` fields once the whole record has been read. Results become a list of records carrying the fields dumpsys prints: namespace, user, _id, name, value, pkg, default and defaultSystemSet, plus the per-setting change history. History timestamps are printed without a year, so they are resolved against the "ending at:" time of the section and serialized into the timeline. This shows which package changed a security-relevant setting, and when. The androidqf settings module shares this artifact, so it now emits the same record shape. --- src/mvt/android/artifacts/settings.py | 264 +++++++++++++++--- .../android/modules/androidqf/aqf_settings.py | 46 +-- src/mvt/android/modules/bugreport/settings.py | 3 +- tests/android/test_artifact_settings.py | 138 +++++++++ tests/android_androidqf/test_settings.py | 25 +- tests/android_bugreport/test_bugreport.py | 20 ++ .../android_data/bugreport/dumpstate.txt | 42 +++ 7 files changed, 448 insertions(+), 90 deletions(-) create mode 100644 tests/android/test_artifact_settings.py diff --git a/src/mvt/android/artifacts/settings.py b/src/mvt/android/artifacts/settings.py index e0cd5f0b..878165a5 100644 --- a/src/mvt/android/artifacts/settings.py +++ b/src/mvt/android/artifacts/settings.py @@ -4,6 +4,11 @@ # https://license.mvt.re/1.1/ import re +from datetime import datetime +from typing import Optional, Sequence + +from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult +from mvt.common.utils import convert_datetime_to_iso from .artifact import AndroidArtifact @@ -60,45 +65,232 @@ ANDROID_DANGEROUS_SETTINGS = [ }, ] +# dumpsys prints the fields of a setting record, and of a change history entry, +# always in this order and separated by a single space. +SETTING_FIELDS = ("_id", "name", "pkg", "value") +HISTORY_FIELDS = ("time", "mode", "oldValue", "newValue", "package") + +NAMESPACE_PATTERN = re.compile( + r"^(CONFIG|GLOBAL|SECURE|SYSTEM) SETTINGS \(user (\d+)\)$" +) +SECTION_END_PATTERN = re.compile(r"ending at: (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})") + class Settings(AndroidArtifact): - def parse(self, content: str) -> None: - self.results: dict[str, dict[str, str]] = {} - namespace: str | None = None - for line in content.splitlines(): - heading = re.match( - r"^(CONFIG|GLOBAL|SECURE|SYSTEM) SETTINGS \(user (\d+)\)$", - line.strip(), - ) - if heading: - namespace = f"{heading.group(1).lower()}:user_{heading.group(2)}" - self.results[namespace] = {} + """Parser for the `dumpsys settings` output. + + Every row of the settings provider becomes one result, keeping the fields + dumpsys prints alongside the value: the row id, the package which recorded + the setting, the default, and the change history. A setting name can appear + more than once within a namespace, so results are a list rather than a + mapping. + """ + + def serialize(self, result: ModuleAtomicResult) -> ModuleSerializedResult: + records = [] + for entry in result.get("history", []): + if not entry.get("timestamp"): continue - if namespace is None or not line.startswith("_id:"): - continue - setting = re.match( - r"^_id:\S+\s+name:(.*?)\s+pkg:.*?\s+value:(.*?)" - r"(?:\s+default:.*\s+defaultSystemSet:(?:true|false))?$", - line, + + records.append( + { + "timestamp": entry["timestamp"], + "module": self.__class__.__name__, + "event": "settings_change", + "data": ( + f"{result.get('namespace')} setting " + f'"{result.get("name")}" changed from ' + f'"{entry.get("oldValue")}" to "{entry.get("newValue")}" ' + f"by {entry.get('pkg')}" + ), + } ) - if setting: - self.results[namespace][setting.group(1)] = setting.group(2) + + return records def check_indicators(self) -> None: - for namespace, settings in self.results.items(): - for key, value in settings.items(): - for danger in ANDROID_DANGEROUS_SETTINGS: - # Check if one of the dangerous settings is using an unsafe - # value (different than the one specified). - if danger["key"] == key and danger["safe_value"] != value: - self.alertstore.medium( - f'Found suspicious "{namespace}" setting "{key} = {value}" ({danger["description"]})', - "", - { - "namespace": namespace, - "key": key, - "value": value, - "description": danger["description"], - }, - ) - break + for result in self.results: + name = result.get("name") + value = result.get("value") + for danger in ANDROID_DANGEROUS_SETTINGS: + # Check if one of the dangerous settings is using an unsafe + # value (different than the one specified). + if danger["key"] != name or danger["safe_value"] == value: + continue + + history = result.get("history") or [] + self.alertstore.medium( + f'Found suspicious "{result.get("namespace")}" setting ' + f'"{name} = {value}" ({danger["description"]})', + history[-1]["timestamp"] if history else "", + result, + ) + break + + def parse(self, content: str) -> None: + self.results: list[ModuleAtomicResult] = [] + section_end = self._parse_section_end(content) + namespace: Optional[str] = None + user: Optional[str] = None + record_lines: list[str] = [] + history_lines: list[str] = [] + in_history = False + + def flush() -> None: + nonlocal record_lines, history_lines, in_history + if record_lines: + self.results.append( + self._build_record( + namespace, user, record_lines, history_lines, section_end + ) + ) + record_lines = [] + history_lines = [] + in_history = False + + for line in content.splitlines(): + heading = NAMESPACE_PATTERN.match(line.strip()) + if heading: + flush() + namespace = heading.group(1).lower() + user = heading.group(2) + continue + + if line.startswith("--------- "): + # dumpsys closes every section with a duration trailer. + flush() + namespace = None + continue + + if namespace is None: + continue + + if line.startswith("_id:"): + flush() + record_lines = [line] + continue + + if not record_lines: + continue + + stripped = line.strip() + if stripped.startswith("History ("): + in_history = True + continue + + if in_history: + if stripped.startswith("time:"): + history_lines.append(stripped) + elif stripped and history_lines: + # A history entry can be wrapped over several lines. + history_lines[-1] += " " + stripped + continue + + # Anything else continues the value of the record being read. + record_lines.append(line) + + flush() + + @staticmethod + def _split_fields(text: str, keys: Sequence[str]) -> dict[str, str]: + """Split the `key:value` fields of one record. + + Values are free-form and may contain spaces and newlines, so a field + runs up to the start of the next key which is actually present. Keys + dumpsys did not print are skipped. + """ + fields: dict[str, str] = {} + key = keys[0] + if not text.startswith(f"{key}:"): + return fields + + remainder = text[len(key) + 1 :] + for next_key in keys[1:]: + value, separator, rest = remainder.partition(f" {next_key}:") + if separator: + fields[key] = value + key, remainder = next_key, rest + + fields[key] = remainder + return fields + + @staticmethod + def _parse_section_end(content: str) -> Optional[datetime]: + """Return the time the settings section was dumped, if reported.""" + match = SECTION_END_PATTERN.search(content) + if not match: + return None + + try: + return datetime.strptime(match.group(1), "%Y-%m-%d %H:%M:%S") + except ValueError: + return None + + @staticmethod + def _resolve_timestamp( + value: str, section_end: Optional[datetime] + ) -> Optional[str]: + """Add the missing year to a `MM-DD HH:MM:SS.mmm` history timestamp. + + dumpsys prints the change history without a year, so it is resolved + against the time the section was dumped: the most recent matching date + at or before that time. + """ + if section_end is None: + return None + + try: + partial = datetime.strptime(value, "%m-%d %H:%M:%S.%f") + timestamp = partial.replace(year=section_end.year) + if timestamp > section_end: + timestamp = partial.replace(year=section_end.year - 1) + except ValueError: + return None + + return convert_datetime_to_iso(timestamp) + + def _parse_history( + self, line: str, section_end: Optional[datetime] + ) -> ModuleAtomicResult: + fields = self._split_fields(line, HISTORY_FIELDS) + return { + "timestamp": self._resolve_timestamp(fields.get("time", ""), section_end), + "oldValue": fields.get("oldValue"), + "newValue": fields.get("newValue"), + "pkg": fields.get("package"), + } + + def _build_record( + self, + namespace: Optional[str], + user: Optional[str], + record_lines: list[str], + history_lines: list[str], + section_end: Optional[datetime], + ) -> ModuleAtomicResult: + text = "\n".join(record_lines).rstrip() + + # `default:` and `defaultSystemSet:` are printed after the value, and + # the default may itself be multi-line, so peel them off the end first. + default = None + default_system_set = None + head, separator, tail = text.rpartition(" defaultSystemSet:") + if separator: + default_system_set = tail.strip() + text = head + head, separator, tail = text.rpartition(" default:") + if separator: + default = tail + text = head + + record: ModuleAtomicResult = {"namespace": namespace, "user": user} + record.update(self._split_fields(text, SETTING_FIELDS)) + if default is not None: + record["default"] = default + if default_system_set is not None: + record["defaultSystemSet"] = default_system_set + + record["history"] = [ + self._parse_history(entry, section_end) for entry in history_lines + ] + return record diff --git a/src/mvt/android/modules/androidqf/aqf_settings.py b/src/mvt/android/modules/androidqf/aqf_settings.py index 8d5bb518..397e121f 100644 --- a/src/mvt/android/modules/androidqf/aqf_settings.py +++ b/src/mvt/android/modules/androidqf/aqf_settings.py @@ -3,11 +3,7 @@ # 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 -from typing import Optional - from mvt.android.artifacts.settings import Settings as SettingsArtifact -from mvt.common.module_types import ModuleResults from .base import AndroidQFModule @@ -15,43 +11,23 @@ from .base import AndroidQFModule class AQFSettings(SettingsArtifact, AndroidQFModule): """This module analyse setting files""" - def __init__( - self, - file_path: Optional[str] = None, - target_path: Optional[str] = None, - results_path: Optional[str] = None, - module_options: Optional[dict] = None, - log: logging.Logger = logging.getLogger(__name__), - results: Optional[ModuleResults] = None, - ) -> None: - super().__init__( - file_path=file_path, - target_path=target_path, - results_path=results_path, - module_options=module_options, - log=log, - results=results, - ) - self.results: dict = results if results is not None else {} - def run(self) -> None: for setting_file in self._get_files_by_pattern("*/settings_*.txt"): namespace = setting_file[setting_file.rfind("_") + 1 : -4] - self.results[namespace] = {} data = self._get_file_content(setting_file) for line in data.decode("utf-8").splitlines(): - line = line.strip() - try: - key, value = line.split("=", 1) - except ValueError: + name, separator, value = line.strip().partition("=") + if not separator: continue - try: - self.results[namespace][key] = value - except IndexError: - continue + self.results.append( + { + "namespace": namespace, + "user": None, + "name": name, + "value": value, + } + ) - self.log.info( - "Identified %d settings", sum([len(val) for val in self.results.values()]) - ) + self.log.info("Identified %d settings", len(self.results)) diff --git a/src/mvt/android/modules/bugreport/settings.py b/src/mvt/android/modules/bugreport/settings.py index 20180adf..6df77a65 100644 --- a/src/mvt/android/modules/bugreport/settings.py +++ b/src/mvt/android/modules/bugreport/settings.py @@ -18,5 +18,4 @@ class Settings(SettingsArtifact, BugReportModule): data.decode("utf-8", errors="replace"), "DUMP OF SERVICE settings:" ) self.parse(section) - count = sum(len(settings) for settings in self.results.values()) - self.log.info("Identified %d Android settings", count) + self.log.info("Identified %d Android settings", len(self.results)) diff --git a/tests/android/test_artifact_settings.py b/tests/android/test_artifact_settings.py new file mode 100644 index 00000000..d3f10521 --- /dev/null +++ b/tests/android/test_artifact_settings.py @@ -0,0 +1,138 @@ +# 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/ + +from mvt.android.artifacts.settings import Settings + +from ..utils import get_artifact + + +def parse_bugreport_settings() -> Settings: + settings = Settings() + with open(get_artifact("android_data/bugreport/dumpstate.txt")) as handle: + data = handle.read() + + settings.parse(settings.extract_dumpsys_section(data, "DUMP OF SERVICE settings:")) + return settings + + +def find(settings: Settings, name: str) -> list: + return [result for result in settings.results if result["name"] == name] + + +class TestSettingsArtifact: + def test_parsing(self): + settings = parse_bugreport_settings() + + assert len(settings.results) == 11 + assert {result["namespace"] for result in settings.results} == { + "config", + "global", + "secure", + } + assert settings.results[0] == { + "namespace": "config", + "user": "0", + "_id": "682", + "name": "namespace_one/blocked_components", + "pkg": "com.example.services", + "value": ( + "com.android.settings,com.android.vending,\n" + "com.example.dialer,\n" + "com.example.camera" + ), + "default": ( + "com.android.settings,\n" + " com.android.vending,\n" + " com.example.dialer" + ), + "defaultSystemSet": "false", + "history": [], + } + + def test_multiline_values_are_kept_whole(self): + settings = parse_bugreport_settings() + + assert find(settings, "namespace_one/allowed_packages")[0]["value"] == ( + "com.example.messaging,\ncom.example.chat" + ) + assert find(settings, "widget_instance_data")[0]["value"] == ( + '{\n "version": 1,\n "data": [\n {\n "number": 10000,\n' + ' "package_name": "com.example.widget"\n }\n ]\n}' + ) + + def test_trailing_default_is_not_part_of_the_value(self): + settings = parse_bugreport_settings() + + record = find(settings, "namespace_one/streaming_blocked_components")[0] + assert record["value"] == "com.example.dialer,com.example.camera" + assert record["default"] == "com.android.settings,\n com.android.vending" + + def test_repeated_names_are_kept_as_separate_records(self): + settings = parse_bugreport_settings() + + widgets = find(settings, "widget_instance_data") + assert [record["_id"] for record in widgets] == ["771", "41654"] + + accessibility = find(settings, "accessibility_enabled") + assert [(record["user"], record["value"]) for record in accessibility] == [ + ("0", "1"), + ("10", "0"), + ] + + def test_setting_without_recording_package(self): + settings = parse_bugreport_settings() + + record = find(settings, "hidden_api_blacklist_exemptions")[0] + assert "pkg" not in record + assert record["value"] == "{null}" + + def test_history_timestamps_resolved_against_section_end(self): + settings = parse_bugreport_settings() + + # The section was dumped on 2022-03-29, so an 11-02 entry belongs to + # the previous year and an 03-14 entry to the same year. + assert find(settings, "development_settings_enabled")[0]["history"] == [ + { + "timestamp": "2021-11-02 11:21:22.212000", + "oldValue": "null", + "newValue": "1", + "pkg": "com.android.settings", + }, + { + "timestamp": "2022-03-14 09:02:11.100000", + "oldValue": "1", + "newValue": "0", + "pkg": "com.example.updater", + }, + ] + + def test_history_without_a_section_end_has_no_timestamp(self): + settings = Settings() + settings.parse( + "SECURE SETTINGS (user 0)\n" + "_id:240 name:accessibility_enabled pkg:android value:1\n" + "\tHistory (accessibility_enabled)\n" + "\t\ttime:03-28 22:41:07.980 mode:update oldValue:0 newValue:1 " + "package:com.example.helper\n" + ) + + assert settings.results[0]["history"] == [ + { + "timestamp": None, + "oldValue": "0", + "newValue": "1", + "pkg": "com.example.helper", + } + ] + + def test_dangerous_setting_is_detected_with_the_changing_package(self): + settings = parse_bugreport_settings() + settings.check_indicators() + + assert len(settings.alertstore.alerts) == 1 + alert = settings.alertstore.alerts[0] + assert "accessibility_enabled = 1" in alert.message + assert alert.event_time == "2022-03-28 22:41:07.980000" + assert alert.event["history"][0]["pkg"] == "com.example.helper" diff --git a/tests/android_androidqf/test_settings.py b/tests/android_androidqf/test_settings.py index 6edfec81..3391565c 100644 --- a/tests/android_androidqf/test_settings.py +++ b/tests/android_androidqf/test_settings.py @@ -6,27 +6,12 @@ from pathlib import Path from mvt.android.modules.androidqf.aqf_settings import AQFSettings -from mvt.android.artifacts.settings import Settings from mvt.common.module import run_module from ..utils import get_android_androidqf, list_files class TestSettingsModule: - def test_bugreport_settings_format(self): - settings = Settings() - settings.parse( - "GLOBAL SETTINGS (user 0)\n" - "_id:1 name:adb_wifi_enabled pkg:android value:0 default:0 defaultSystemSet:true\n" - "SECURE SETTINGS (user 10)\n" - "_id:2 name:accessibility_enabled pkg:android value:1\n" - ) - - assert settings.results == { - "global:user_0": {"adb_wifi_enabled": "0"}, - "secure:user_10": {"accessibility_enabled": "1"}, - } - def test_parsing(self): data_path = get_android_androidqf() m = AQFSettings(target_path=data_path) @@ -34,7 +19,13 @@ class TestSettingsModule: parent_path = Path(data_path).absolute().parent.as_posix() m.from_dir(parent_path, files) run_module(m) - assert len(m.results) == 1 - assert "random" in m.results.keys() + assert len(m.results) == 9 + assert {result["namespace"] for result in m.results} == {"random"} + assert m.results[0] == { + "namespace": "random", + "user": None, + "name": "samsung_errorlog_agree", + "value": "0", + } assert len(m.alertstore.alerts) == 1 assert "samsung_errorlog_agree" in m.alertstore.alerts[0].message diff --git a/tests/android_bugreport/test_bugreport.py b/tests/android_bugreport/test_bugreport.py index 413bb180..e0a92f7a 100644 --- a/tests/android_bugreport/test_bugreport.py +++ b/tests/android_bugreport/test_bugreport.py @@ -10,6 +10,7 @@ from mvt.android.modules.bugreport.dumpsys_appops import DumpsysAppops from mvt.android.modules.bugreport.dumpsys_getprop import DumpsysGetProp from mvt.android.modules.bugreport.dumpsys_packages import DumpsysPackages from mvt.android.modules.bugreport.dumpsys_receivers import DumpsysReceivers +from mvt.android.modules.bugreport.settings import Settings from mvt.android.modules.bugreport.tombstones import Tombstones from mvt.common.module import run_module @@ -93,6 +94,25 @@ class TestBugreportAnalysis: assert alert.event == malicious_receiver assert alert.matched_indicator.value == "com.android.services" + def test_settings_module(self): + m = self.launch_bug_report_module(Settings) + assert len(m.results) == 11 + + assert len(m.alertstore.alerts) == 1 + assert "accessibility_enabled = 1" in m.alertstore.alerts[0].message + + assert len(m.timeline) == 3 + change = [ + entry + for entry in m.timeline + if entry["timestamp"] == "2022-03-28 22:41:07.980000" + ][0] + assert change["event"] == "settings_change" + assert change["data"] == ( + 'secure setting "accessibility_enabled" changed from "0" to "1" ' + "by com.example.helper" + ) + def test_tombstones_modules(self): m = self.launch_bug_report_module(Tombstones) assert len(m.results) == 2 diff --git a/tests/artifacts/android_data/bugreport/dumpstate.txt b/tests/artifacts/android_data/bugreport/dumpstate.txt index c888ed5c..34479e9f 100644 --- a/tests/artifacts/android_data/bugreport/dumpstate.txt +++ b/tests/artifacts/android_data/bugreport/dumpstate.txt @@ -264,5 +264,47 @@ ChangeId(143539591; name=SELINUX_LATEST_CHANGES; disabled) ChangeId(247079863; name=DISALLOW_INVALID_GROUP_REFERENCE; enableSinceTargetSdk=34) ChangeId(174227820; name=FORCE_DISABLE_HEVC_SUPPORT; disabled) ChangeId(168419799; name=DOWNSCALED; disabled; packageOverrides={com.google.android.apps.tachyon=false, org.torproject.torbrowser=false}; rawOverrides={org.torproject.torbrowser=false, org.article19.circulo.next=false}; overridable) +------------------------------------------------------------------------------- +DUMP OF SERVICE settings: +CONFIG SETTINGS (user 0) +_id:682 name:namespace_one/blocked_components pkg:com.example.services value:com.android.settings,com.android.vending, +com.example.dialer, +com.example.camera default:com.android.settings, + com.android.vending, + com.example.dialer defaultSystemSet:false +_id:684 name:namespace_one/streaming_blocked_components pkg:com.example.services value:com.example.dialer,com.example.camera default:com.android.settings, + com.android.vending defaultSystemSet:false +_id:680 name:namespace_one/allowed_packages pkg:com.example.services value:com.example.messaging, +com.example.chat +GLOBAL SETTINGS (user 0) +_id:2070 name:adb_wifi_enabled pkg:android value:0 default:0 defaultSystemSet:true +_id:778 name:hidden_api_blacklist_exemptions value:{null} +_id:9631 name:development_settings_enabled pkg:com.android.settings value:1 default:1 defaultSystemSet:true + History (development_settings_enabled) + time:11-02 11:21:22.212 mode:update oldValue:null newValue:1 package:com.android.settings + time:03-14 09:02:11.100 mode:update oldValue:1 newValue:0 package:com.example.updater +_id:771 name:widget_instance_data pkg:com.android.systemui value:{ + "version": 1, + "data": [ + { + "number": 10000, + "package_name": "com.example.widget" + } + ] +} defaultSystemSet:true +_id:41654 name:widget_instance_data pkg:com.android.systemui value:{ + "version": 3, + "data": [] +} defaultSystemSet:true +SECURE SETTINGS (user 0) +_id:907 name:lock_screen_show_notifications pkg:com.android.settings value:1 default:1 defaultSystemSet:true +_id:240 name:accessibility_enabled pkg:android value:1 default:0 defaultSystemSet:true + History (accessibility_enabled) + time:03-28 22:41:07.980 mode:update oldValue:0 newValue:1 package:com.example.helper + +SECURE SETTINGS (user 10) +_id:311 name:accessibility_enabled pkg:android value:0 default:0 defaultSystemSet:true + +--------- 0.019s was the duration of dumpsys settings, ending at: 2022-03-29 23:14:28 From 03012d2c2dc102e3ab20177ab521de4f3188d259 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Fri, 4 Sep 2026 20:25:07 +0200 Subject: [PATCH 06/16] Fix CI coverage comment and test each matrix Python version The coverage comment never posted: the per-file table with a link on every file and missing line range exceeds GitHub's 65536-character comment limit, so PRs got only a badge. Every matrix job also raced to post the same comment, and the action was unpinned at @main. Post from one job only, limit the table to files changed in the PR, drop per-line links, and pin the action. Write the full coverage table to the job summary as well, which also works for PRs from forks where the token is read-only. Set UV_PYTHON from the matrix. Without it, .python-version pins 3.10 and `uv run` rebuilt the venv with 3.10 after `uv sync --python X`, so all five matrix jobs were testing on Python 3.10. --- .github/workflows/tests.yml | 23 +++++++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index c98ab223..a2d37a28 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -5,6 +5,10 @@ on: pull_request: branches: [ main ] +permissions: + contents: read + pull-requests: write # coverage comment + jobs: build: name: Run Python Tests @@ -13,6 +17,10 @@ jobs: fail-fast: false matrix: python-version: ['3.10', '3.11', '3.12', '3.13', '3.14'] + env: + # Takes precedence over .python-version, which otherwise makes `uv run` + # rebuild the venv with 3.10 and test every matrix entry on 3.10. + UV_PYTHON: ${{ matrix.python-version }} steps: - uses: actions/checkout@v7 @@ -26,16 +34,23 @@ jobs: python-version: ${{ matrix.python-version }} - name: Install Python dependencies run: | - uv sync --locked --group dev --python ${{ matrix.python-version }} + uv sync --locked --group dev - name: Test with pytest run: | set -o pipefail make test-ci | tee pytest-coverage.txt + - name: Coverage job summary + run: uv run coverage report --format=markdown --show-missing --skip-covered >> "$GITHUB_STEP_SUMMARY" - name: Pytest coverage comment - continue-on-error: true # Workflows running on a fork can't post comments - uses: MishaKav/pytest-coverage-comment@main - if: github.event_name == 'pull_request' + # One comment per PR, not one per matrix entry. PRs from forks get a + # read-only token and can't post; the job summary above still works. + if: github.event_name == 'pull_request' && matrix.python-version == '3.13' + continue-on-error: true + uses: MishaKav/pytest-coverage-comment@v1.12.2 with: pytest-coverage-path: ./pytest-coverage.txt junitxml-path: ./pytest.xml + # The full table with per-line links exceeds GitHub's 65536-char comment limit. + report-only-changed-files: true + remove-links-to-lines: true From e5fced2a08d5c1a7ed483409ab6846b3ccadbcc8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Sat, 5 Sep 2026 15:56:55 +0200 Subject: [PATCH 07/16] Peel tag and restore flags off settings rows A dumpsys settings row can end in `tag:` and, on some vendor builds, in `isValuePreservedInRestore:` or a bare `notPreservedInRestore` token. The parser only peeled `default:` and `defaultSystemSet:` off the end of a record, so on a row without a default those tokens stayed inside the value, and on a row with one they landed in `defaultSystemSet`. A value of `0 tag:null` is not the safe value `0`, so a setting at its safe value was reported as dangerous; these are the false positives measured in #912. The metadata keys are ordered fields like the rest of the record, so `_split_fields` reads them now, which also replaces the separate handling of the default. The bare token has no `key:` shape and is printed last, so it is stripped first and recorded as `isValuePreservedInRestore: false`. The bugreport fixture gains the three row shapes from #912: a restore flag after `defaultSystemSet:`, and a `tag:` or a bare token directly after the value. Two of them sit on dangerous settings at their safe value, so the unchanged alert count of one is the false-positive check. --- src/mvt/android/artifacts/settings.py | 46 +++++++++---------- tests/android/test_artifact_settings.py | 20 +++++++- tests/android_bugreport/test_bugreport.py | 2 +- .../android_data/bugreport/dumpstate.txt | 5 +- 4 files changed, 46 insertions(+), 27 deletions(-) diff --git a/src/mvt/android/artifacts/settings.py b/src/mvt/android/artifacts/settings.py index 878165a5..627fafe3 100644 --- a/src/mvt/android/artifacts/settings.py +++ b/src/mvt/android/artifacts/settings.py @@ -66,8 +66,20 @@ ANDROID_DANGEROUS_SETTINGS = [ ] # dumpsys prints the fields of a setting record, and of a change history entry, -# always in this order and separated by a single space. -SETTING_FIELDS = ("_id", "name", "pkg", "value") +# always in this order and separated by a single space. After the value come +# `default:` and `defaultSystemSet:` when a default is recorded, then `tag:`; +# some vendor builds add whether the value survives a restore, either as +# `isValuePreservedInRestore:` or as a bare `notPreservedInRestore` token. +SETTING_FIELDS = ( + "_id", + "name", + "pkg", + "value", + "default", + "defaultSystemSet", + "tag", + "isValuePreservedInRestore", +) HISTORY_FIELDS = ("time", "mode", "oldValue", "newValue", "package") NAMESPACE_PATTERN = re.compile( @@ -81,9 +93,9 @@ class Settings(AndroidArtifact): Every row of the settings provider becomes one result, keeping the fields dumpsys prints alongside the value: the row id, the package which recorded - the setting, the default, and the change history. A setting name can appear - more than once within a namespace, so results are a list rather than a - mapping. + the setting, the default, the tag, and the change history. A setting name + can appear more than once within a namespace, so results are a list rather + than a mapping. """ def serialize(self, result: ModuleAtomicResult) -> ModuleSerializedResult: @@ -269,26 +281,14 @@ class Settings(AndroidArtifact): section_end: Optional[datetime], ) -> ModuleAtomicResult: text = "\n".join(record_lines).rstrip() - - # `default:` and `defaultSystemSet:` are printed after the value, and - # the default may itself be multi-line, so peel them off the end first. - default = None - default_system_set = None - head, separator, tail = text.rpartition(" defaultSystemSet:") - if separator: - default_system_set = tail.strip() - text = head - head, separator, tail = text.rpartition(" default:") - if separator: - default = tail - text = head + # The bare `notPreservedInRestore` token has no `key:` shape and is + # printed last, so peel it off before splitting the fields. + head = text.removesuffix(" notPreservedInRestore") record: ModuleAtomicResult = {"namespace": namespace, "user": user} - record.update(self._split_fields(text, SETTING_FIELDS)) - if default is not None: - record["default"] = default - if default_system_set is not None: - record["defaultSystemSet"] = default_system_set + record.update(self._split_fields(head, SETTING_FIELDS)) + if head != text: + record["isValuePreservedInRestore"] = "false" record["history"] = [ self._parse_history(entry, section_end) for entry in history_lines diff --git a/tests/android/test_artifact_settings.py b/tests/android/test_artifact_settings.py index d3f10521..c3fde0d2 100644 --- a/tests/android/test_artifact_settings.py +++ b/tests/android/test_artifact_settings.py @@ -25,7 +25,7 @@ class TestSettingsArtifact: def test_parsing(self): settings = parse_bugreport_settings() - assert len(settings.results) == 11 + assert len(settings.results) == 12 assert {result["namespace"] for result in settings.results} == { "config", "global", @@ -69,6 +69,24 @@ class TestSettingsArtifact: assert record["value"] == "com.example.dialer,com.example.camera" assert record["default"] == "com.android.settings,\n com.android.vending" + def test_trailing_metadata_is_not_part_of_the_value(self): + settings = parse_bugreport_settings() + + record = find(settings, "lock_screen_show_notifications")[0] + assert record["value"] == "1" + assert record["defaultSystemSet"] == "true" + assert record["isValuePreservedInRestore"] == "true" + + # Without a default, the tag or the restore token follows the value. + record = find(settings, "accessibility_enabled")[1] + assert record["value"] == "0" + assert record["tag"] == "null" + assert "default" not in record + + record = find(settings, "send_action_app_error")[0] + assert record["value"] == "1" + assert record["isValuePreservedInRestore"] == "false" + def test_repeated_names_are_kept_as_separate_records(self): settings = parse_bugreport_settings() diff --git a/tests/android_bugreport/test_bugreport.py b/tests/android_bugreport/test_bugreport.py index e0a92f7a..0709e5ad 100644 --- a/tests/android_bugreport/test_bugreport.py +++ b/tests/android_bugreport/test_bugreport.py @@ -96,7 +96,7 @@ class TestBugreportAnalysis: def test_settings_module(self): m = self.launch_bug_report_module(Settings) - assert len(m.results) == 11 + assert len(m.results) == 12 assert len(m.alertstore.alerts) == 1 assert "accessibility_enabled = 1" in m.alertstore.alerts[0].message diff --git a/tests/artifacts/android_data/bugreport/dumpstate.txt b/tests/artifacts/android_data/bugreport/dumpstate.txt index 34479e9f..56a5d864 100644 --- a/tests/artifacts/android_data/bugreport/dumpstate.txt +++ b/tests/artifacts/android_data/bugreport/dumpstate.txt @@ -280,6 +280,7 @@ com.example.chat GLOBAL SETTINGS (user 0) _id:2070 name:adb_wifi_enabled pkg:android value:0 default:0 defaultSystemSet:true _id:778 name:hidden_api_blacklist_exemptions value:{null} +_id:9640 name:send_action_app_error pkg:android value:1 notPreservedInRestore _id:9631 name:development_settings_enabled pkg:com.android.settings value:1 default:1 defaultSystemSet:true History (development_settings_enabled) time:11-02 11:21:22.212 mode:update oldValue:null newValue:1 package:com.android.settings @@ -299,12 +300,12 @@ _id:41654 name:widget_instance_data pkg:com.android.systemui value:{ } defaultSystemSet:true SECURE SETTINGS (user 0) -_id:907 name:lock_screen_show_notifications pkg:com.android.settings value:1 default:1 defaultSystemSet:true +_id:907 name:lock_screen_show_notifications pkg:com.android.settings value:1 default:1 defaultSystemSet:true isValuePreservedInRestore:true _id:240 name:accessibility_enabled pkg:android value:1 default:0 defaultSystemSet:true History (accessibility_enabled) time:03-28 22:41:07.980 mode:update oldValue:0 newValue:1 package:com.example.helper SECURE SETTINGS (user 10) -_id:311 name:accessibility_enabled pkg:android value:0 default:0 defaultSystemSet:true +_id:311 name:accessibility_enabled pkg:android value:0 tag:null --------- 0.019s was the duration of dumpsys settings, ending at: 2022-03-29 23:14:28 From 858496e60bba1d66016b243b00cb8c7c1a0c1119 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Sat, 5 Sep 2026 16:17:43 +0200 Subject: [PATCH 08/16] End a settings record at a blank line dumpsys prints a blank line after every namespace block and after a change history, and the generation registry and any vendor dumps follow the last block before the section trailer. A record ran until the next `_id:` line, heading or trailer, so the last row of the section absorbed those dumps into whichever field came last. A blank line now closes the record being read; lines that follow it without an `_id:` are skipped until the next row or heading. The fixture carries a generation registry after its last block, modelled on the AOSP dump, and the last row is pinned to exactly its own fields. --- src/mvt/android/artifacts/settings.py | 8 ++++++++ tests/android/test_artifact_settings.py | 14 ++++++++++++++ .../artifacts/android_data/bugreport/dumpstate.txt | 5 +++++ 3 files changed, 27 insertions(+) diff --git a/src/mvt/android/artifacts/settings.py b/src/mvt/android/artifacts/settings.py index 627fafe3..39257c06 100644 --- a/src/mvt/android/artifacts/settings.py +++ b/src/mvt/android/artifacts/settings.py @@ -177,6 +177,14 @@ class Settings(AndroidArtifact): if namespace is None: continue + if not line.strip(): + # dumpsys prints a blank line after every namespace block and + # after a change history, and other dumps such as the + # generation registry follow the last block, so a blank line + # closes the record being read. + flush() + continue + if line.startswith("_id:"): flush() record_lines = [line] diff --git a/tests/android/test_artifact_settings.py b/tests/android/test_artifact_settings.py index c3fde0d2..2192c12b 100644 --- a/tests/android/test_artifact_settings.py +++ b/tests/android/test_artifact_settings.py @@ -87,6 +87,20 @@ class TestSettingsArtifact: assert record["value"] == "1" assert record["isValuePreservedInRestore"] == "false" + def test_dumps_after_the_last_block_are_not_part_of_the_last_row(self): + settings = parse_bugreport_settings() + + assert settings.results[-1] == { + "namespace": "secure", + "user": "10", + "_id": "311", + "name": "accessibility_enabled", + "pkg": "android", + "value": "0", + "tag": "null", + "history": [], + } + def test_repeated_names_are_kept_as_separate_records(self): settings = parse_bugreport_settings() diff --git a/tests/artifacts/android_data/bugreport/dumpstate.txt b/tests/artifacts/android_data/bugreport/dumpstate.txt index 56a5d864..4dffc585 100644 --- a/tests/artifacts/android_data/bugreport/dumpstate.txt +++ b/tests/artifacts/android_data/bugreport/dumpstate.txt @@ -308,4 +308,9 @@ _id:240 name:accessibility_enabled pkg:android value:1 default:0 defaultSystemSe SECURE SETTINGS (user 10) _id:311 name:accessibility_enabled pkg:android value:0 tag:null +GENERATION REGISTRY +Maximum number of backing stores:8 +Number of backing stores:1 +_Backing store for type:SETTINGS_SECURE user:10 size:1024 cachedEntries:1 + --------- 0.019s was the duration of dumpsys settings, ending at: 2022-03-29 23:14:28 From d7148c03b4e73e9e18c482f08a8fe96c85705dca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Sat, 5 Sep 2026 16:30:55 +0200 Subject: [PATCH 09/16] Stop persisting environment variables to config.yaml MVTSettings.initialise() constructs the settings once with load_env=False and writes the result to config.yaml, so that values taken from MVT_* environment variables are never persisted, then constructs them again with the environment applied. Since #716 settings_customise_sources() has added env_settings unconditionally, making load_env dead code. Any MVT_* variable set for a single run, including MVT_IOS_BACKUP_PASSWORD, MVT_ANDROID_BACKUP_PASSWORD and MVT_VT_API_KEY, was written in plaintext to config.yaml and read back on every later run. Only add env_settings when load_env is true, and add a regression test. Fixes #915 --- src/mvt/common/config.py | 13 ++++++++----- tests/common/test_config.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 5 deletions(-) create mode 100644 tests/common/test_config.py diff --git a/src/mvt/common/config.py b/src/mvt/common/config.py index ce29fb33..24bf4072 100644 --- a/src/mvt/common/config.py +++ b/src/mvt/common/config.py @@ -59,13 +59,16 @@ class MVTSettings(BaseSettings): dotenv_settings: PydanticBaseSettingsSource, file_secret_settings: PydanticBaseSettingsSource, ) -> Tuple[PydanticBaseSettingsSource, ...]: - yaml_source = YamlConfigSettingsSource(settings_cls, MVT_CONFIG_PATH) sources: Tuple[PydanticBaseSettingsSource, ...] = ( - yaml_source, + YamlConfigSettingsSource(settings_cls, MVT_CONFIG_PATH), init_settings, ) - # Always load env variables by default - sources = (env_settings,) + sources + # Load env variables only when asked to. initialise() constructs the + # settings once without them so that what gets written back to + # config.yaml never includes values taken from the environment. + # init_settings() returns the keyword arguments passed to the constructor. + if init_settings().get("load_env", True): + sources = (env_settings,) + sources return sources def save_settings( @@ -92,7 +95,7 @@ class MVTSettings(BaseSettings): Afterwards we load the settings again, this time including the env variables. """ - # Set invalid env prefix to avoid loading env variables. + # Construct the settings without env variables so they are not persisted. settings = cls(load_env=False) settings.save_settings() diff --git a/tests/common/test_config.py b/tests/common/test_config.py new file mode 100644 index 00000000..f471a583 --- /dev/null +++ b/tests/common/test_config.py @@ -0,0 +1,30 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2026 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 os + +import yaml + +from mvt.common import config +from mvt.common.config import MVTSettings + + +def test_env_variables_are_not_persisted_to_config_file(tmp_path, monkeypatch): + config_path = tmp_path / "config.yaml" + monkeypatch.setattr(config, "MVT_CONFIG_FOLDER", str(tmp_path)) + monkeypatch.setattr(config, "MVT_CONFIG_PATH", str(config_path)) + monkeypatch.setenv("MVT_NETWORK_ACCESS_ALLOWED", "false") + monkeypatch.setenv("MVT_IOS_BACKUP_PASSWORD", "env-only-password") + + settings = MVTSettings.initialise() + + assert os.path.isfile(config_path) + saved = yaml.safe_load(config_path.read_text()) or {} + assert "NETWORK_ACCESS_ALLOWED" not in saved + assert "IOS_BACKUP_PASSWORD" not in saved + + # The environment must still apply to the settings in use. + assert settings.NETWORK_ACCESS_ALLOWED is False + assert settings.IOS_BACKUP_PASSWORD == "env-only-password" From 2eb40b85cfd3d3788bb4600c49906e136a6c7db5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Sat, 5 Sep 2026 23:45:14 +0200 Subject: [PATCH 10/16] Add a SysdiagnoseInfo module to check-sysdiagnose (#917) * Add a SysdiagnoseInfo module to check-sysdiagnose check-sysdiagnose had no module of its own: it prepared the archive for plugin modules and refused to run without one. SysdiagnoseInfo is the first built-in module. It writes sysdiagnose_info.json with details about the device and the archive: product type and model, iOS version and build, serial number, IMEI, MEID and UDID from remotectl_dumpstate.txt and the mobile activation request, the Apple account name and email from the App Store daemon database, and the archive's original file name and creation time from sysdiagnose.log. The build is checked against the known iOS versions the way BackupInfo does. The App Store database is copied out of the archive together with its -wal and -shm sidecars before it is opened, so rows still in the write-ahead log are read. With a built-in module the command's list is never empty, so the "no custom modules" error and its test go. The module joins IOS_CHECK_IOCS_MODULES like every other module that writes a results file. * Note that newer sysdiagnoses lack the App Store daemon database * Keep refusing check-sysdiagnose runs without a custom module * Warn instead of refusing when no forensic sysdiagnose module is loaded --- docs/ios/records.md | 8 + docs/ios/sysdiagnose.md | 11 +- src/mvt/ios/cli.py | 13 +- src/mvt/ios/cmd_check_sysdiagnose.py | 3 + src/mvt/ios/command_modules.py | 3 +- src/mvt/ios/modules/sysdiagnose/__init__.py | 5 + .../modules/sysdiagnose/sysdiagnose_info.py | 228 ++++++++++++++++++ tests/common/test_command_modules.py | 5 +- tests/ios_sysdiagnose/__init__.py | 4 + .../ios_sysdiagnose/test_sysdiagnose_info.py | 161 +++++++++++++ tests/test_check_ios_sysdiagnose.py | 15 +- tests/test_cmd_check_sysdiagnose.py | 17 +- 12 files changed, 451 insertions(+), 22 deletions(-) create mode 100644 src/mvt/ios/modules/sysdiagnose/sysdiagnose_info.py create mode 100644 tests/ios_sysdiagnose/__init__.py create mode 100644 tests/ios_sysdiagnose/test_sysdiagnose_info.py diff --git a/docs/ios/records.md b/docs/ios/records.md index b34183ad..856d0db9 100644 --- a/docs/ios/records.md +++ b/docs/ios/records.md @@ -435,3 +435,11 @@ This JSON file is created by mvt-ios' `WhatsappContacts` module. The module extr This database is often missing from incremental backups. When it cannot be found, the module logs a warning and produces no results, in which case the disappearing messages state of chats cannot be determined from the backup. + +--- + +## Records extracted by `check-sysdiagnose` + +### `sysdiagnose_info.json` + +This JSON file is created by mvt-ios' `SysdiagnoseInfo` module. The module extracts details about the device and the sysdiagnose itself: the product type and model, iOS version and build, serial number, IMEI, MEID and UDID from the remotectl dump state and the mobile activation request, the Apple account name and email from the App Store daemon database (no longer part of a sysdiagnose on newer iOS versions, still read from older archives), and the original file name and creation time of the archive from *sysdiagnose.log*. diff --git a/docs/ios/sysdiagnose.md b/docs/ios/sysdiagnose.md index d0d84fb3..0107ce6b 100644 --- a/docs/ios/sysdiagnose.md +++ b/docs/ios/sysdiagnose.md @@ -1,10 +1,13 @@ # Check an iOS Sysdiagnose -`mvt-ios check-sysdiagnose` prepares an iOS sysdiagnose archive for analysis by -custom MVT modules. MVT does not include built-in sysdiagnose modules. The -command runs the modules of the installed +`mvt-ios check-sysdiagnose` analyzes an iOS sysdiagnose archive. MVT's own +`SysdiagnoseInfo` module extracts details about the device and the archive +(see [`sysdiagnose_info.json`](records.md#sysdiagnose_infojson)); the checks +come from the modules of the installed [plugin packages](../development/index.md#installed-module-packages) which -declare support for it. Install at least one such package first. +declare support for the command. Without any such module the command still +records the device details, and warns that no forensic sysdiagnose modules +have been loaded so that the run cannot pass for a clean analysis. The command accepts either an extracted sysdiagnose directory or the original gzip-compressed tar archive. diff --git a/src/mvt/ios/cli.py b/src/mvt/ios/cli.py index 56c4a11d..4560010a 100644 --- a/src/mvt/ios/cli.py +++ b/src/mvt/ios/cli.py @@ -449,11 +449,14 @@ def check_sysdiagnose( custom_modules=custom_modules, ) - if not cmd._available_modules(): - raise click.ClickException( - "No custom modules support mvt-ios check-sysdiagnose. " - "Load a module that declares supported_commands = " - "((\"ios\", \"check-sysdiagnose\"),)." + # MVT's own module only records the device details; the checks come from + # custom modules, so a run without any must not look like a clean analysis. + if all(module in cmd.modules for module in cmd._available_modules()): + log.warning( + "No forensic sysdiagnose modules have been loaded: MVT's own " + "SysdiagnoseInfo module only records the device details. Install a " + "module package or load a module that declares supported_commands = " + '(("ios", "check-sysdiagnose"),) to check the sysdiagnose.' ) if list_modules: diff --git a/src/mvt/ios/cmd_check_sysdiagnose.py b/src/mvt/ios/cmd_check_sysdiagnose.py index 05d5b6b7..3450a03f 100644 --- a/src/mvt/ios/cmd_check_sysdiagnose.py +++ b/src/mvt/ios/cmd_check_sysdiagnose.py @@ -16,6 +16,8 @@ from mvt.common.command import Command from mvt.common.indicators import Indicators from mvt.common.module import MVTModule +from .modules.sysdiagnose import SYSDIAGNOSE_MODULES + log = logging.getLogger(__name__) @@ -52,6 +54,7 @@ class CmdIOSCheckSysdiagnose(Command): ) self.platform = "ios" self.name = "check-sysdiagnose" + self.modules = SYSDIAGNOSE_MODULES self.sysdiagnose_format: Optional[str] = None self.sysdiagnose_archive: Optional[tarfile.TarFile] = None self.sysdiagnose_files: list[str] = [] diff --git a/src/mvt/ios/command_modules.py b/src/mvt/ios/command_modules.py index 26fa1d61..ab12d607 100644 --- a/src/mvt/ios/command_modules.py +++ b/src/mvt/ios/command_modules.py @@ -16,7 +16,8 @@ from mvt.common.module import MVTModule from .modules.backup import BACKUP_MODULES from .modules.fs import FS_MODULES from .modules.mixed import MIXED_MODULES +from .modules.sysdiagnose import SYSDIAGNOSE_MODULES IOS_CHECK_IOCS_MODULES: list[type[MVTModule]] = ( - BACKUP_MODULES + FS_MODULES + MIXED_MODULES + BACKUP_MODULES + FS_MODULES + MIXED_MODULES + SYSDIAGNOSE_MODULES ) diff --git a/src/mvt/ios/modules/sysdiagnose/__init__.py b/src/mvt/ios/modules/sysdiagnose/__init__.py index 3963ca54..4f586ecc 100644 --- a/src/mvt/ios/modules/sysdiagnose/__init__.py +++ b/src/mvt/ios/modules/sysdiagnose/__init__.py @@ -3,4 +3,9 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +from mvt.common.module import MVTModule + from .base import SysdiagnoseExtraction +from .sysdiagnose_info import SysdiagnoseInfo + +SYSDIAGNOSE_MODULES: list[type[MVTModule]] = [SysdiagnoseInfo] diff --git a/src/mvt/ios/modules/sysdiagnose/sysdiagnose_info.py b/src/mvt/ios/modules/sysdiagnose/sysdiagnose_info.py new file mode 100644 index 00000000..c37208c2 --- /dev/null +++ b/src/mvt/ios/modules/sysdiagnose/sysdiagnose_info.py @@ -0,0 +1,228 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2026 The MVT Authors. +# Use of this software is governed by the MVT License 1.1 that can be found at +# https://license.mvt.re/1.1/ + +import json +import logging +import os +import plistlib +import re +import sqlite3 +import tempfile +from datetime import datetime +from pathlib import Path +from typing import Optional + +from mvt.common.module_types import ModuleResults +from mvt.common.utils import convert_datetime_to_iso +from mvt.ios.versions import ( + find_version_by_build, + get_device_desc_from_id, + is_ios_version_outdated, +) + +from .base import SysdiagnoseExtraction + +# The fields dumpsys prints in the remotectl dump state and the mobile +# activation request which are worth a log line of their own. +LOGGED_FIELDS = ( + "ProductName", + "ProductType", + "SerialNumber", + "OSVersion", + "RegionCode", + "IMEI", + "BuildVersion", +) + + +class SysdiagnoseInfo(SysdiagnoseExtraction): + """Extract details about the device and the sysdiagnose itself. + + The fields come from four files of the archive: the remotectl dump state + (product type, OS version, serial number, region and the rest of its + Properties block), the mobile activation request (UDID, IMEI, MEID and the + OS build), the App Store daemon database (the Apple account name and email) + and sysdiagnose.log (the archive's original file name and creation time). + + Newer iOS versions no longer include the App Store daemon database in a + sysdiagnose; it is still read for the analysis of older archives. + """ + + def __init__( + self, + file_path: Optional[str] = None, + target_path: Optional[str] = None, + results_path: Optional[str] = None, + module_options: Optional[dict] = None, + log: logging.Logger = logging.getLogger(__name__), + results: Optional[ModuleResults] = None, + ) -> None: + super().__init__( + file_path=file_path, + target_path=target_path, + results_path=results_path, + module_options=module_options, + log=log, + results=results, + ) + + self.results: dict = results if results is not None else {} + + def _copy_sqlite_db(self, file_path: str, directory: str) -> str: + """Copy a database and its WAL sidecars out of the archive. + + A database dumped mid-transaction keeps its latest rows in the -wal + file next to it, which SQLite only reads when both sit in the same + directory under the same name. + """ + available_files = self.tar_files if self.tar else self.files + for suffix in ("", "-wal", "-shm"): + if suffix and f"{file_path}{suffix}" not in available_files: + continue + copy_path = os.path.join(directory, f"{Path(file_path).name}{suffix}") + with open(copy_path, "wb") as handle: + handle.write(self._get_file_content(f"{file_path}{suffix}")) + + return os.path.join(directory, Path(file_path).name) + + def _process_appstored(self, file_path: str) -> None: + self.log.info("Found App Store daemon database at: %s", file_path) + with tempfile.TemporaryDirectory(prefix="mvt_sqlite_") as directory: + db_path = Path(self._copy_sqlite_db(file_path, directory)).resolve() + conn = sqlite3.connect(f"{db_path.as_uri()}?mode=ro", uri=True) + try: + self._read_appstored(conn) + finally: + conn.close() + + def _read_appstored(self, conn: sqlite3.Connection) -> None: + cur = conn.cursor() + # The account name sits in an opaque structure of every asset row. + try: + rows = cur.execute("SELECT sinfs_data FROM asset;").fetchall() + except sqlite3.DatabaseError as exc: + self.log.debug("Unable to read the asset table: %s", exc) + rows = [] + + for (sinfs_data,) in rows: + try: + sinf = plistlib.loads(sinfs_data)[0]["sinf"] + except (plistlib.InvalidFileException, IndexError, KeyError, TypeError): + continue + match = re.search(rb"name(.*?)\x00", sinf) + if match: + self.results["Account Name"] = match.group(1).decode( + "utf-8", errors="replace" + ) + break + + try: + row = cur.execute( + "SELECT store_account_name FROM job_software " + "WHERE store_account_name IS NOT NULL LIMIT 1;" + ).fetchone() + except sqlite3.DatabaseError as exc: + self.log.debug("Unable to read the job_software table: %s", exc) + return + + if row: + self.results["Email Address"] = row[0] + + def _process_activation_log(self, file_path: str) -> None: + self.log.info("Found mobile activation request at: %s", file_path) + content = self._get_file_content(file_path) + match = re.search(rb"BODY:\s+({.+?})\s", content, re.MULTILINE) + if not match: + return + + try: + body = json.loads(match.group(1)) + except json.JSONDecodeError as exc: + self.log.warning("Unable to parse the activation request body: %s", exc) + return + + self.results.update( + { + "SerialNumber": body.get("serial-number"), + "ProductType": body.get("productType"), + "ProductName": body.get("productName"), + "IMEI": body.get("imei"), + "ProductVersion": body.get("os-version"), + "UniqueIdentifier": body.get("udid"), + "MEID": body.get("meid"), + "BuildVersion": body.get("os-build"), + } + ) + + def _process_dumpstate(self, file_path: str) -> None: + self.log.info("Found remotectl dump state at: %s", file_path) + content = self._get_file_content(file_path).decode("utf-8", errors="replace") + in_properties = False + for line in content.splitlines(): + if not in_properties: + in_properties = line == "\tProperties: {" + continue + + if line == "\t}": + break + + key, separator, value = line.partition("=>") + if separator: + self.results[key.strip()] = value.strip() + + def _process_sysdiagnose_log(self, file_path: str) -> None: + self.log.info("Found sysdiagnose.log at: %s", file_path) + content = self._get_file_content(file_path).decode("utf-8", errors="replace") + match = re.search(r"sysdiagnose_\S+?\.tar\.gz", content) + if not match: + self.log.info("Could not find the original output path in sysdiagnose.log") + return + + file_name = os.path.basename(match.group(0)) + try: + created = datetime.strptime( + "_".join(file_name.split("_")[1:3]), "%Y.%m.%d_%H-%M-%S%z" + ) + except ValueError: + self.log.warning("Unexpected sysdiagnose file name: %s", file_name) + return + + self.results["OriginalFilename"] = file_name + self.results["CreatedTimestamp"] = convert_datetime_to_iso(created) + + def run(self) -> None: + for file_path in self._get_files_by_pattern( + "*/logs/appinstallation/appstored.sqlitedb" + ): + self._process_appstored(file_path) + + for file_path in self._get_files_by_pattern( + "*/logs/MobileActivation/collection_oob_request.txt" + ): + self._process_activation_log(file_path) + + for file_path in self._get_files_by_pattern("*/remotectl_dumpstate.txt"): + self._process_dumpstate(file_path) + + for file_path in self._get_files_by_pattern("*/sysdiagnose.log"): + self._process_sysdiagnose_log(file_path) + + # The activation request names the product "iPhone OS"; the model + # description is what an analyst wants to read. + product_name = get_device_desc_from_id(self.results.get("ProductType", "")) + if product_name: + self.results["ProductName"] = product_name + + for field in LOGGED_FIELDS: + if field not in self.results: + continue + value = self.results[field] + if field == "BuildVersion" and value: + self.log.info("%s: %s - %s", field, value, find_version_by_build(value)) + else: + self.log.info("%s: %s", field, value) + + if self.results.get("BuildVersion"): + is_ios_version_outdated(self.results["BuildVersion"], self.log) diff --git a/tests/common/test_command_modules.py b/tests/common/test_command_modules.py index 8f79ce9d..a79cda2c 100644 --- a/tests/common/test_command_modules.py +++ b/tests/common/test_command_modules.py @@ -12,13 +12,16 @@ from mvt.ios.command_modules import IOS_CHECK_IOCS_MODULES from mvt.ios.modules.backup import BACKUP_MODULES as IOS_BACKUP_MODULES from mvt.ios.modules.fs import FS_MODULES from mvt.ios.modules.mixed import MIXED_MODULES +from mvt.ios.modules.sysdiagnose import SYSDIAGNOSE_MODULES def test_the_check_iocs_lists_are_the_families_of_their_platform(): # The CLI reads these same lists, so nothing composing one elsewhere can # drift from what the command runs. This pins what the lists are composed # of. - assert IOS_CHECK_IOCS_MODULES == IOS_BACKUP_MODULES + FS_MODULES + MIXED_MODULES + assert IOS_CHECK_IOCS_MODULES == ( + IOS_BACKUP_MODULES + FS_MODULES + MIXED_MODULES + SYSDIAGNOSE_MODULES + ) assert ANDROID_CHECK_IOCS_MODULES == ( ANDROID_BACKUP_MODULES + BUGREPORT_MODULES diff --git a/tests/ios_sysdiagnose/__init__.py b/tests/ios_sysdiagnose/__init__.py new file mode 100644 index 00000000..4e7aeb6d --- /dev/null +++ b/tests/ios_sysdiagnose/__init__.py @@ -0,0 +1,4 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2026 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/ diff --git a/tests/ios_sysdiagnose/test_sysdiagnose_info.py b/tests/ios_sysdiagnose/test_sysdiagnose_info.py new file mode 100644 index 00000000..d5590897 --- /dev/null +++ b/tests/ios_sysdiagnose/test_sysdiagnose_info.py @@ -0,0 +1,161 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2026 The MVT Authors. +# Use of this software is governed by the MVT License 1.1 that can be found at +# https://license.mvt.re/1.1/ + +import json +import plistlib +import sqlite3 +import tarfile + +from mvt.common.module import run_module +from mvt.ios.cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose +from mvt.ios.modules.sysdiagnose.sysdiagnose_info import SysdiagnoseInfo +from mvt.ios.versions import get_device_desc_from_id + +# The name sysdiagnose gives its archive: the time it ran, then the OS and build. +ARCHIVE_NAME = "sysdiagnose_2024.01.02_03-04-05+0200_iPhone-OS_iPhone_21C62" + +DUMPSTATE = ( + "Found device: ...\n" + "\tProperties: {\n" + "\t\tProductType => iPhone12,1\n" + "\t\tOSVersion => 17.2\n" + "\t\tSerialNumber => C0FFEE000000\n" + "\t\tRegionCode => LL\n" + "\t}\n" + "\tServices: {\n" + "\t\tcom.apple.example => ignored\n" + "\t}\n" +) + +ACTIVATION_BODY = { + "serial-number": "C0FFEE000000", + "productType": "iPhone12,1", + "productName": "iPhone OS", + "imei": "000000000000000", + "os-version": "17.2", + "os-build": "21C62", + "udid": "00000000-0000000000000000", + "meid": "00000000000000", +} + + +def make_sysdiagnose(tmp_path, activation_body=None): + folder = tmp_path / ARCHIVE_NAME + folder.mkdir() + (folder / "sysdiagnose.log").write_text( + f"Output available at '/private/var/tmp/{ARCHIVE_NAME}.tar.gz'\n", + encoding="utf-8", + ) + (folder / "remotectl_dumpstate.txt").write_text(DUMPSTATE, encoding="utf-8") + + activation = folder / "logs" / "MobileActivation" + activation.mkdir(parents=True) + body = json.dumps( + activation_body if activation_body is not None else ACTIVATION_BODY + ) + (activation / "collection_oob_request.txt").write_text( + f"HEADERS: {{}}\nBODY: {body}\nEND\n", encoding="utf-8" + ) + + appinstallation = folder / "logs" / "appinstallation" + appinstallation.mkdir(parents=True) + conn = sqlite3.connect(appinstallation / "appstored.sqlitedb") + conn.execute("CREATE TABLE asset (sinfs_data BLOB)") + conn.execute( + "INSERT INTO asset VALUES (?)", + (plistlib.dumps([{"sinf": b"\x00\x10nameExample Person\x00\x00rest"}]),), + ) + conn.execute("CREATE TABLE job_software (store_account_name TEXT)") + conn.execute("INSERT INTO job_software VALUES (NULL)") + conn.execute("INSERT INTO job_software VALUES ('person@example.com')") + conn.commit() + conn.close() + return folder + + +def run_command(target, results_path=None): + command = CmdIOSCheckSysdiagnose(target_path=str(target), results_path=results_path) + command.run() + (module,) = [m for m in command.executed if isinstance(m, SysdiagnoseInfo)] + return module + + +def test_device_details_from_a_sysdiagnose_folder(tmp_path): + results_path = tmp_path / "results" + results_path.mkdir() + module = run_command(make_sysdiagnose(tmp_path), str(results_path)) + + assert module.results["SerialNumber"] == "C0FFEE000000" + assert module.results["ProductType"] == "iPhone12,1" + assert module.results["ProductName"] == get_device_desc_from_id("iPhone12,1") + assert module.results["ProductName"] != "iPhone OS" + assert module.results["OSVersion"] == "17.2" + assert module.results["BuildVersion"] == "21C62" + assert module.results["UniqueIdentifier"] == "00000000-0000000000000000" + assert module.results["RegionCode"] == "LL" + assert "com.apple.example" not in module.results + assert module.results["Account Name"] == "Example Person" + assert module.results["Email Address"] == "person@example.com" + assert module.results["OriginalFilename"] == f"{ARCHIVE_NAME}.tar.gz" + assert module.results["CreatedTimestamp"] == "2024-01-02 01:04:05.000000" + assert (results_path / "sysdiagnose_info.json").exists() + + +def test_device_details_from_a_sysdiagnose_archive(tmp_path): + folder = make_sysdiagnose(tmp_path) + archive_path = tmp_path / f"{ARCHIVE_NAME}.tar.gz" + with tarfile.open(archive_path, "w:gz") as archive: + archive.add(folder, arcname=ARCHIVE_NAME) + + module = run_command(archive_path) + + assert module.results["SerialNumber"] == "C0FFEE000000" + assert module.results["Account Name"] == "Example Person" + assert module.results["OriginalFilename"] == f"{ARCHIVE_NAME}.tar.gz" + + +def test_wal_sidecars_are_copied_beside_the_database(tmp_path): + folder = tmp_path / ARCHIVE_NAME + (folder / "logs").mkdir(parents=True) + (folder / "logs" / "db.sqlite").write_bytes(b"main") + (folder / "logs" / "db.sqlite-wal").write_bytes(b"wal") + module = SysdiagnoseInfo() + module.from_sysdiagnose_folder( + str(folder), + [f"{ARCHIVE_NAME}/logs/db.sqlite", f"{ARCHIVE_NAME}/logs/db.sqlite-wal"], + ) + copies = tmp_path / "copies" + copies.mkdir() + + db_path = module._copy_sqlite_db(f"{ARCHIVE_NAME}/logs/db.sqlite", str(copies)) + + assert db_path == str(copies / "db.sqlite") + assert (copies / "db.sqlite").read_bytes() == b"main" + assert (copies / "db.sqlite-wal").read_bytes() == b"wal" + assert not (copies / "db.sqlite-shm").exists() + + +def test_a_sysdiagnose_without_the_files_yields_nothing(tmp_path): + folder = tmp_path / ARCHIVE_NAME + folder.mkdir() + (folder / "other.txt").write_text("nothing here", encoding="utf-8") + + module = SysdiagnoseInfo() + module.from_sysdiagnose_folder(str(folder), [f"{ARCHIVE_NAME}/other.txt"]) + run_module(module) + + assert module.results == {} + + +def test_a_malformed_activation_request_is_skipped(tmp_path): + folder = make_sysdiagnose(tmp_path) + (folder / "logs" / "MobileActivation" / "collection_oob_request.txt").write_text( + "BODY: {not json}\n", encoding="utf-8" + ) + + module = run_command(folder) + + assert "IMEI" not in module.results + assert module.results["SerialNumber"] == "C0FFEE000000" diff --git a/tests/test_check_ios_sysdiagnose.py b/tests/test_check_ios_sysdiagnose.py index d73c0ffb..0cd4221a 100644 --- a/tests/test_check_ios_sysdiagnose.py +++ b/tests/test_check_ios_sysdiagnose.py @@ -1,3 +1,5 @@ +import logging + from click.testing import CliRunner from mvt.ios.cli import check_sysdiagnose @@ -50,8 +52,13 @@ def test_check_sysdiagnose_runs_explicitly_scoped_custom_module(tmp_path): assert (output_path / "custom_sysdiagnose_module.json").exists() -def test_check_sysdiagnose_requires_an_explicitly_scoped_module(tmp_path): - result = CliRunner().invoke(check_sysdiagnose, [str(_create_sysdiagnose_folder(tmp_path))]) +def test_check_sysdiagnose_warns_without_a_custom_module(tmp_path, caplog): + # The built-in SysdiagnoseInfo alone performs no check, so the run goes + # ahead but says so. + with caplog.at_level(logging.WARNING, logger="mvt"): + result = CliRunner().invoke( + check_sysdiagnose, [str(_create_sysdiagnose_folder(tmp_path))] + ) - assert result.exit_code != 0 - assert "No custom modules support mvt-ios check-sysdiagnose" in result.output + assert result.exit_code == 0 + assert "No forensic sysdiagnose modules have been loaded" in caplog.text diff --git a/tests/test_cmd_check_sysdiagnose.py b/tests/test_cmd_check_sysdiagnose.py index 67f126bd..020c8616 100644 --- a/tests/test_cmd_check_sysdiagnose.py +++ b/tests/test_cmd_check_sysdiagnose.py @@ -45,6 +45,11 @@ def _create_sysdiagnose_archive(tmp_path, folder): return archive_path +def _test_module(command): + (module,) = [m for m in command.executed if isinstance(m, SysdiagnoseTestModule)] + return module + + def _run_command(path): command = CmdIOSCheckSysdiagnose( target_path=str(path), custom_modules=[SysdiagnoseTestModule] @@ -56,10 +61,10 @@ def _run_command(path): def test_check_sysdiagnose_from_folder(tmp_path): command = _run_command(_create_sysdiagnose_folder(tmp_path)) - assert command.executed[0].results == [ + assert _test_module(command).results == [ {"content": "artifact", "timezone_offset": timedelta(hours=2).seconds} ] - assert command.executed[0].ips_files == [ + assert _test_module(command).ips_files == [ {"file_path": str(tmp_path / "sysdiagnose" / "report.ips"), "bug_type": 210} ] @@ -68,14 +73,12 @@ def test_check_sysdiagnose_from_archive_closes_archive(tmp_path): folder = _create_sysdiagnose_folder(tmp_path) command = _run_command(_create_sysdiagnose_archive(tmp_path, folder)) - assert command.executed[0].results == [ + assert _test_module(command).results == [ {"content": "artifact", "timezone_offset": timedelta(hours=2).seconds} ] - assert command.executed[0].ips_files == [ + assert _test_module(command).ips_files == [ { - "file_path": str( - Path(command.extracted_sysdiagnose_path) / "report.ips" - ), + "file_path": str(Path(command.extracted_sysdiagnose_path) / "report.ips"), "bug_type": 210, } ] From 92f780df5a58b79981b68850587e993f81dc2c10 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Mon, 7 Sep 2026 23:58:43 +0100 Subject: [PATCH 11/16] Report an unreadable sysdiagnose archive instead of aborting silently (#920) A sysdiagnose tarball whose download stopped halfway ends in an EOFError from gzip while check-sysdiagnose extracts it. Click turns EOFError into click.Abort, so the command printed nothing but "Aborted!", even with -v. The extraction now catches the read errors an archive can raise, names the file and the reason at critical level, says the file may be truncated or not a gzip tarball, and exits 1. A plain .tar given to the gzip reader gets the same message rather than a traceback. --- src/mvt/ios/cmd_check_sysdiagnose.py | 17 +++++++++++++++-- tests/test_check_ios_sysdiagnose.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/src/mvt/ios/cmd_check_sysdiagnose.py b/src/mvt/ios/cmd_check_sysdiagnose.py index 3450a03f..88a12db0 100644 --- a/src/mvt/ios/cmd_check_sysdiagnose.py +++ b/src/mvt/ios/cmd_check_sysdiagnose.py @@ -7,7 +7,9 @@ import json import logging import os import shutil +import sys import tarfile +import zlib from pathlib import Path, PurePosixPath from tempfile import TemporaryDirectory from typing import Any, Optional @@ -97,8 +99,19 @@ class CmdIOSCheckSysdiagnose(Command): self.log.info("Parsing sysdiagnose archive. This might take a while...") self.sysdiagnose_format = "tar" - self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz") - self._extract_sysdiagnose_archive() + try: + self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz") + self._extract_sysdiagnose_archive() + except (tarfile.ReadError, EOFError, zlib.error, OSError) as exc: + # A truncated archive ends in EOFError from gzip, which Click would + # otherwise report as a bare "Aborted!" with no reason. + self.log.critical( + "Unable to read the sysdiagnose archive %s: %s. " + "The file may be truncated or not a gzip-compressed tarball.", + self.target_path, + exc, + ) + sys.exit(1) def _extract_sysdiagnose_archive(self) -> None: archive = self.sysdiagnose_archive diff --git a/tests/test_check_ios_sysdiagnose.py b/tests/test_check_ios_sysdiagnose.py index 0cd4221a..c81e213a 100644 --- a/tests/test_check_ios_sysdiagnose.py +++ b/tests/test_check_ios_sysdiagnose.py @@ -1,4 +1,6 @@ import logging +import os +import tarfile from click.testing import CliRunner @@ -62,3 +64,29 @@ def test_check_sysdiagnose_warns_without_a_custom_module(tmp_path, caplog): assert result.exit_code == 0 assert "No forensic sysdiagnose modules have been loaded" in caplog.text + + +def _create_truncated_sysdiagnose_archive(tmp_path): + folder = tmp_path / "sysdiagnose_2026.01.01_00-00-00+0000_iPhone-OS_iPhone_23A000" + folder.mkdir() + (folder / "sysdiagnose.log").write_bytes(os.urandom(200_000)) + archive = tmp_path / "sysdiagnose.tar.gz" + with tarfile.open(archive, "w:gz") as tar: + tar.add(folder, arcname=folder.name) + data = archive.read_bytes() + archive.write_bytes(data[: len(data) // 2]) + return archive + + +def test_check_sysdiagnose_reports_a_truncated_archive(tmp_path, caplog): + # A download that stopped halfway ends in EOFError from gzip, which Click + # would otherwise turn into a bare "Aborted!" with no reason given. + archive = _create_truncated_sysdiagnose_archive(tmp_path) + + with caplog.at_level(logging.CRITICAL, logger="mvt"): + result = CliRunner().invoke(check_sysdiagnose, [str(archive)]) + + assert result.exit_code == 1 + assert "Unable to read the sysdiagnose archive" in caplog.text + assert "truncated" in caplog.text + assert "Aborted!" not in result.output From 60e652bafcd7a5c11de057b8da7a2a8828c394a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Tue, 8 Sep 2026 00:08:30 +0100 Subject: [PATCH 12/16] Read bugreport file timestamps in the device's timezone (#921) A bugreport zip stores each entry's time as the device's wall clock, and an unpacked bugreport's mtimes are whatever the extraction left. The bugreport modules read both as naive local datetimes, so a tombstone's file_timestamp moved with the analysing machine's timezone and sat three hours from the crash time the tombstone itself records for a device in Nairobi. check-bugreport now resolves the device's timezone once, from --timezone or from persist.sys.timezone in the dumpstate's SYSTEM PROPERTIES, and hands it to the modules as device_timezone the way check-androidqf does; a zone already known, as when androidqf drives the bugreport inside its own archive, is kept. Zip entry times are read in that zone and the mtimes of an unpacked bugreport as UTC instants, so convert_datetime_to_iso writes both in UTC. Without a zone the wall clock is kept naive and a warning says so, and an unpacked bugreport gets a warning that its file timestamps are the extraction's, not the device's. BugReportTimestamps uses the same reading instead of parsing the properties itself. --- src/mvt/android/cli.py | 17 ++++ src/mvt/android/cmd_check_bugreport.py | 54 ++++++++++- src/mvt/android/modules/bugreport/base.py | 32 +++++-- .../modules/bugreport/fs_timestamps.py | 48 ++++------ tests/test_check_android_bugreport.py | 92 +++++++++++++++++++ 5 files changed, 200 insertions(+), 43 deletions(-) diff --git a/src/mvt/android/cli.py b/src/mvt/android/cli.py index 7b78a4e4..e9f8d019 100644 --- a/src/mvt/android/cli.py +++ b/src/mvt/android/cli.py @@ -150,6 +150,16 @@ def check_adb(ctx): default=[], help=HELP_MSG_LOAD_MODULE, ) +@click.option( + "--timezone", + "-t", + default=None, + help=( + "IANA timezone name for the device, for example 'Europe/Paris'. " + "Bugreport file timestamps are the device's wall clock; by default the " + "zone is read from persist.sys.timezone in the bugreport itself." + ), +) @click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE_COMMAND) @click.argument("BUGREPORT_PATH", type=click.Path(exists=True)) @click.pass_context @@ -160,6 +170,7 @@ def check_bugreport( list_modules, module, load_module, + timezone, verbose, bugreport_path, ): @@ -167,12 +178,18 @@ def check_bugreport( set_verbose_logging(verbose or _get_verbose(ctx)) custom_modules = _load_custom_modules(load_module) + + module_options = {} + if timezone: + module_options["device_timezone"] = timezone + # Always generate hashes as bug reports are small. cmd = CmdAndroidCheckBugreport( target_path=bugreport_path, results_path=output, ioc_files=iocs, module_name=module, + module_options=module_options if module_options else None, hashes=True, disable_version_check=_get_disable_flags(ctx)[0], disable_indicator_check=_get_disable_flags(ctx)[1], diff --git a/src/mvt/android/cmd_check_bugreport.py b/src/mvt/android/cmd_check_bugreport.py index 1c03c6d2..946feb97 100644 --- a/src/mvt/android/cmd_check_bugreport.py +++ b/src/mvt/android/cmd_check_bugreport.py @@ -9,6 +9,7 @@ from pathlib import Path from typing import List, Optional from zipfile import ZipFile +from mvt.android.artifacts.getprop import GetProp from mvt.android.modules.bugreport.base import BugReportModule from mvt.common.command import Command from mvt.common.indicators import Indicators @@ -88,13 +89,56 @@ class CmdAndroidCheckBugreport(Command): self.__files.append(file_name) def init(self) -> None: - if not self.target_path: + if self.target_path: + if os.path.isfile(self.target_path): + self.from_zip(ZipFile(self.target_path)) + elif os.path.isdir(self.target_path): + self.from_dir(self.target_path) + self.log.warning( + "Analysing an unpacked bugreport: file timestamps come from " + "the extraction, not from the device. Analyse the original " + "zip to keep the device's file timestamps." + ) + if self.__format: + self._resolve_device_timezone() + + def _resolve_device_timezone(self) -> None: + """Name the device's timezone in module_options unless it is known already. + + A bugreport's SYSTEM PROPERTIES section carries persist.sys.timezone. + Zip entry times are the device's wall clock, and modules read them in + this zone; --timezone or check-androidqf's own reading takes precedence. + """ + if self.module_options.get("device_timezone"): + self.log.info("Device timezone: %s", self.module_options["device_timezone"]) return - if os.path.isfile(self.target_path): - self.from_zip(ZipFile(self.target_path)) - elif os.path.isdir(self.target_path): - self.from_dir(self.target_path) + probe = BugReportModule(log=self.log) + self.module_init(probe) + timezone = None + try: + dumpstate = probe._get_dumpstate_file() + except Exception as exc: + self.log.warning("Could not read the bugreport's dumpstate: %s", exc) + dumpstate = None + if dumpstate: + properties = GetProp() + properties.parse( + BugReportModule.extract_command_section( + dumpstate.decode("utf-8", errors="replace"), + "------ SYSTEM PROPERTIES", + ) + ) + timezone = properties.get_device_timezone() + if timezone: + self.log.info("Device timezone identified from the bugreport: %s", timezone) + self.module_options["device_timezone"] = timezone + else: + self.log.warning( + "persist.sys.timezone not found in the bugreport; file timestamps " + "are the device's wall clock without a timezone. Pass --timezone " + "to name it." + ) def module_init(self, module: BugReportModule) -> None: # type: ignore[override] if self.__format == "zip": diff --git a/src/mvt/android/modules/bugreport/base.py b/src/mvt/android/modules/bugreport/base.py index 01af8c96..25396f95 100644 --- a/src/mvt/android/modules/bugreport/base.py +++ b/src/mvt/android/modules/bugreport/base.py @@ -9,6 +9,7 @@ import os from pathlib import Path from typing import List, Optional from zipfile import ZipFile +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from mvt.common.module import ModuleResults, MVTModule @@ -125,12 +126,31 @@ class BugReportModule(MVTModule): lines.append(line) return "\n".join(lines) + def _device_timezone(self) -> Optional[datetime.tzinfo]: + """The device's timezone named in module_options, or None when unknown.""" + name = self.module_options.get("device_timezone") + if not name: + return None + try: + return ZoneInfo(name) + except ZoneInfoNotFoundError: + self.log.warning("Unknown device timezone %s", name) + return None + def _get_file_modification_time(self, file_path: str) -> datetime.datetime: + """When the file was last modified. + + A zip entry carries the device's wall clock, so it is returned in the + device's timezone when the bugreport names one and naive otherwise. + An unpacked bugreport's mtime is whatever the extraction left, an + instant returned in UTC. + """ if self.zip_archive: file_timetuple = self.zip_archive.getinfo(file_path).date_time - return datetime.datetime(*file_timetuple) - else: - if not self.extract_path: - raise ValueError("extract_path is not set") - file_stat = os.stat(os.path.join(self.extract_path, file_path)) - return datetime.datetime.fromtimestamp(file_stat.st_mtime) + return datetime.datetime(*file_timetuple, tzinfo=self._device_timezone()) + if not self.extract_path: + raise ValueError("extract_path is not set") + file_stat = os.stat(os.path.join(self.extract_path, file_path)) + return datetime.datetime.fromtimestamp( + file_stat.st_mtime, tz=datetime.timezone.utc + ) diff --git a/src/mvt/android/modules/bugreport/fs_timestamps.py b/src/mvt/android/modules/bugreport/fs_timestamps.py index 3c7e3248..800869a9 100644 --- a/src/mvt/android/modules/bugreport/fs_timestamps.py +++ b/src/mvt/android/modules/bugreport/fs_timestamps.py @@ -4,15 +4,12 @@ # https://license.mvt.re/1.1/ import logging -import datetime from typing import Optional -from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from mvt.common.utils import convert_datetime_to_iso from .base import BugReportModule from mvt.common.module_types import ModuleResults from mvt.android.artifacts.file_timestamps import FileTimestampsArtifact -from mvt.android.artifacts.getprop import GetProp class BugReportTimestamps(FileTimestampsArtifact, BugReportModule): @@ -41,41 +38,28 @@ class BugReportTimestamps(FileTimestampsArtifact, BugReportModule): def run(self) -> None: filesystem_files = self._get_files_by_pattern("FS/*") - timezone_name = None - dumpstate = self._get_dumpstate_file() - if dumpstate: - section = self.extract_command_section( - dumpstate.decode("utf-8", errors="replace"), - "------ SYSTEM PROPERTIES", - ) - properties = GetProp() - properties.parse(section) - timezone_name = properties.get_device_timezone() - timezone = None - if timezone_name: - try: - timezone = ZoneInfo(timezone_name) - except ZoneInfoNotFoundError: - self.log.warning("Unknown device timezone %s", timezone_name) - self.results = [] for file in filesystem_files: - # Only the modification time is available in the zip file metadata. - # The timezone is the local timezone of the machine the phone. + # A zip entry keeps the device's wall clock, read in the device's + # timezone when the command found one (see CmdAndroidCheckBugreport); + # an unpacked bugreport's mtime is read as a UTC instant. modification_time = self._get_file_modification_time(file) - utc_time = None - if timezone is not None: - utc_time = convert_datetime_to_iso( - modification_time.replace(tzinfo=timezone).astimezone( - datetime.timezone.utc - ) - ) self.results.append( { "path": file, - "modified_time": convert_datetime_to_iso(modification_time), - "modified_time_utc": utc_time, - "timezone": timezone_name, + "modified_time": convert_datetime_to_iso( + modification_time.replace(tzinfo=None) + ), + "modified_time_utc": ( + convert_datetime_to_iso(modification_time) + if modification_time.tzinfo + else None + ), + "timezone": ( + self.module_options.get("device_timezone") + if self.zip_archive + else "UTC" + ), "timestamp_source": ( "zip_metadata" if self.zip_archive else "filesystem_metadata" ), diff --git a/tests/test_check_android_bugreport.py b/tests/test_check_android_bugreport.py index bff47090..ba5b6dcc 100644 --- a/tests/test_check_android_bugreport.py +++ b/tests/test_check_android_bugreport.py @@ -3,11 +3,16 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +import datetime +import logging import os +import shutil +import zipfile from click.testing import CliRunner from mvt.android.cli import check_bugreport +from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport from .utils import get_artifact_folder @@ -28,3 +33,90 @@ class TestCheckBugreportCommand: assert result.exit_code == 1 assert "Invalid bugreport archive" in result.output assert "Traceback" not in result.output + + +PROPERTIES = ( + "------ SYSTEM PROPERTIES (getprop) ------\n" + "[persist.sys.timezone]: [Africa/Nairobi]\n" + "------ 0.01s was the duration of 'SYSTEM PROPERTIES' ------\n" +) +TOMBSTONE = "android_data/bugreport/FS/data/tombstones/tombstone_00" + + +def _bugreport_zip(tmp_path, dumpstate=PROPERTIES): + """A bugreport zip holding one tombstone written at 11:38:10 device time. + + An even second: zip entry times have a two-second resolution. + """ + path = tmp_path / "bugreport.zip" + with open(os.path.join(get_artifact_folder(), TOMBSTONE), "rb") as handle: + tombstone = handle.read() + with zipfile.ZipFile(path, "w") as archive: + archive.writestr("main_entry.txt", "dumpstate.txt") + archive.writestr("dumpstate.txt", dumpstate) + entry = zipfile.ZipInfo( + "FS/data/tombstones/tombstone_00", date_time=(2023, 3, 10, 11, 38, 10) + ) + archive.writestr(entry, tombstone) + return str(path) + + +def _tombstone_timestamp(target, **options): + cmd = CmdAndroidCheckBugreport( + target_path=target, + module_name="Tombstones", + disable_version_check=True, + disable_indicator_check=True, + **options, + ) + cmd.run() + return cmd, cmd.executed[0].results[0]["file_timestamp"] + + +class TestCheckBugreportTimezone: + def test_zip_entry_times_are_read_in_the_device_timezone(self, tmp_path): + cmd, file_timestamp = _tombstone_timestamp(_bugreport_zip(tmp_path)) + + assert cmd.module_options["device_timezone"] == "Africa/Nairobi" + # 11:38:10 in Nairobi is 08:38:10 UTC. + assert file_timestamp == "2023-03-10 08:38:10.000000" + + def test_timezone_option_wins_over_the_bugreport(self, tmp_path): + _, file_timestamp = _tombstone_timestamp( + _bugreport_zip(tmp_path), module_options={"device_timezone": "Europe/Paris"} + ) + + assert file_timestamp == "2023-03-10 10:38:10.000000" + + result = CliRunner().invoke( + check_bugreport, + ["-t", "Europe/Paris", "-m", "Tombstones", _bugreport_zip(tmp_path)], + ) + assert result.exit_code == 0, result.output + + def test_without_a_timezone_the_wall_clock_is_kept_and_a_warning_given( + self, tmp_path, caplog + ): + with caplog.at_level(logging.WARNING, logger="mvt"): + _, file_timestamp = _tombstone_timestamp(_bugreport_zip(tmp_path, "")) + + assert file_timestamp == "2023-03-10 11:38:10.000000" + assert "persist.sys.timezone not found" in caplog.text + + def test_unpacked_bugreport_warns_and_reads_mtimes_as_utc(self, tmp_path, caplog): + unpacked = tmp_path / "bugreport" + shutil.copytree( + os.path.join(get_artifact_folder(), "android_data/bugreport"), unpacked + ) + instant = datetime.datetime( + 2023, 3, 10, 8, 38, 11, tzinfo=datetime.timezone.utc + ).timestamp() + for name in ("tombstone_00", "tombstone_01"): + os.utime(unpacked / "FS" / "data" / "tombstones" / name, (instant, instant)) + + with caplog.at_level(logging.WARNING, logger="mvt"): + _, file_timestamp = _tombstone_timestamp(str(unpacked)) + + assert "unpacked bugreport" in caplog.text + # Whatever the zone of the machine running the analysis. + assert file_timestamp == "2023-03-10 08:38:11.000000" From 3623e430f4ecadf4634b9daafd05f5332228b9b2 Mon Sep 17 00:00:00 2001 From: Daniel Conor Sullivan Date: Mon, 7 Sep 2026 20:01:02 -0400 Subject: [PATCH 13/16] Defer indicator loading until first use (#919) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Defer indicator loading until first use * Load indicators once at the start of a run The lazy property loads indicators wherever they are first read, which in Command.run() is inside the module loop, after init(). For check-androidqf that means the whole acquisition is walked before a missing --iocs file is reported, and the STIX parsing lines land between the module list and the first module. Read the indicators once after the module list is settled, so that a bad --module name still loads nothing, and hand the same object to every module. check-iocs reads them once too, so that a wrong --iocs path is reported even when no stored result matches a module. * Drop two test assertions the change does not need Retrying after a failed indicator load is incidental to the property rather than a requirement, so nothing should pin it. The exact wording of the check-backup rejection belongs to that command's own tests; exit code 1 already shows the path was rejected before indicators were loaded. * Create the output folder and command.log when a run starts Command.__init__ created the --output folder and attached the command.log handler, so `--list-modules -o out` or a rejected target path left an empty folder behind for a run that never happened. Do both at the top of run(), before the module list is resolved so that its warnings still reach the log. The nested commands run by check-androidqf re-attach the handler at the start of their runs, as they did at construction. Lines the CLI logs between construction and run(), such as the target path being checked, no longer reach command.log; info.json records the target path. This is the remaining part of #888. * Cache the indicators with functools.cached_property A property with a setter and a backing attribute does by hand what cached_property does: compute on first read, store the result on the instance, and accept assignment, which is how nested commands receive their parent's indicators. A failed load is still not cached. * Announce the target from the command so that it reaches command.log The CLI logged which backup, filesystem or acquisition it was about to check just before run(), which is now before command.log exists, so the line only reached the console. Each command logs it from init() instead, which runs once the log is attached, and run() logs the module list after init() so that the target still comes first. Nested commands without a target path log nothing, as before. * Log the Android backup path from the typed local mypy cannot determine the type of target_path in this command, which the surrounding lines already silence, so read the announced path from the local that carries the type instead of adding another ignore. --------- Co-authored-by: bitmeta69 <206962233+bitmeta69@users.noreply.github.com> Co-authored-by: besendorf Co-authored-by: Donncha Ó Cearbhaill --- src/mvt/android/cli.py | 8 -- src/mvt/android/cmd_check_androidqf.py | 2 + src/mvt/android/cmd_check_backup.py | 1 + src/mvt/android/cmd_check_bugreport.py | 1 + src/mvt/android/cmd_check_intrusion_logs.py | 2 + src/mvt/common/cmd_check_iocs.py | 8 +- src/mvt/common/command.py | 34 ++++++--- src/mvt/ios/cli.py | 5 -- src/mvt/ios/cmd_check_backup.py | 3 + src/mvt/ios/cmd_check_fs.py | 3 + src/mvt/ios/cmd_check_sysdiagnose.py | 2 + tests/common/test_command.py | 85 +++++++++++++++++++++ tests/test_check_ios_backup.py | 8 ++ 13 files changed, 137 insertions(+), 25 deletions(-) diff --git a/src/mvt/android/cli.py b/src/mvt/android/cli.py index e9f8d019..4c159857 100644 --- a/src/mvt/android/cli.py +++ b/src/mvt/android/cli.py @@ -200,8 +200,6 @@ def check_bugreport( cmd.list_modules() return - log.info("Checking Android bug report at path: %s", bugreport_path) - try: cmd.run() except BadZipFile as exc: @@ -276,8 +274,6 @@ def check_backup( cmd.list_modules() return - log.info("Checking Android backup at path: %s", backup_path) - cmd.run() cmd.show_alerts_brief() cmd.show_support_message() @@ -359,8 +355,6 @@ def check_androidqf( cmd.list_modules() return - log.info("Checking AndroidQF acquisition at path: %s", androidqf_path) - cmd.run() cmd.show_alerts_brief() cmd.show_disable_adb_warning() @@ -441,8 +435,6 @@ def check_intrusion_logs( cmd.list_modules() return - log.info("Checking intrusion logs at path: %s", logs_path) - cmd.run() 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 99ac5a3f..a86e643e 100644 --- a/src/mvt/android/cmd_check_androidqf.py +++ b/src/mvt/android/cmd_check_androidqf.py @@ -82,6 +82,8 @@ class CmdAndroidCheckAndroidQF(Command): if not self.target_path: raise NoAndroidQFTargetPath + self.log.info("Checking AndroidQF acquisition at path: %s", self.target_path) + if os.path.isdir(self.target_path): self.__format = "dir" parent_path = Path(self.target_path).absolute().parent.as_posix() diff --git a/src/mvt/android/cmd_check_backup.py b/src/mvt/android/cmd_check_backup.py index 94be4ee2..066cd8aa 100644 --- a/src/mvt/android/cmd_check_backup.py +++ b/src/mvt/android/cmd_check_backup.py @@ -129,6 +129,7 @@ class CmdAndroidCheckBackup(Command): assert self.target_path is not None # type: ignore[has-type] # Use a different local variable name to avoid any scoping issues backup_path: str = self.target_path # type: ignore[has-type] + self.log.info("Checking Android backup at path: %s", backup_path) if os.path.isfile(backup_path): self.__type = "ab" diff --git a/src/mvt/android/cmd_check_bugreport.py b/src/mvt/android/cmd_check_bugreport.py index 946feb97..036160d8 100644 --- a/src/mvt/android/cmd_check_bugreport.py +++ b/src/mvt/android/cmd_check_bugreport.py @@ -90,6 +90,7 @@ class CmdAndroidCheckBugreport(Command): def init(self) -> None: if self.target_path: + self.log.info("Checking Android bug report at path: %s", self.target_path) if os.path.isfile(self.target_path): self.from_zip(ZipFile(self.target_path)) elif os.path.isdir(self.target_path): diff --git a/src/mvt/android/cmd_check_intrusion_logs.py b/src/mvt/android/cmd_check_intrusion_logs.py index 95f20896..a652c229 100644 --- a/src/mvt/android/cmd_check_intrusion_logs.py +++ b/src/mvt/android/cmd_check_intrusion_logs.py @@ -63,6 +63,8 @@ class CmdAndroidCheckIntrusionLogs(Command): if not self.target_path: raise ValueError("No target path specified") + self.log.info("Checking intrusion logs at path: %s", self.target_path) + if not os.path.isdir(self.target_path) and not ( os.path.isfile(self.target_path) and self.target_path.lower().endswith(".zip") diff --git a/src/mvt/common/cmd_check_iocs.py b/src/mvt/common/cmd_check_iocs.py index c2dbdcd9..7c7e71e5 100644 --- a/src/mvt/common/cmd_check_iocs.py +++ b/src/mvt/common/cmd_check_iocs.py @@ -56,6 +56,10 @@ class CmdCheckIOCS(Command): if entry not in all_modules: all_modules.append(entry) + # Read the indicators once, so that a missing indicators file is + # reported even when no stored result matches a module. + iocs = self.iocs + log.info("Checking stored results against provided indicators...") total_detections = 0 @@ -83,8 +87,8 @@ class CmdCheckIOCS(Command): log.warning("No result from this module, skipping it") continue - if self.iocs.total_ioc_count > 0: - m.indicators = self.iocs + if iocs.total_ioc_count > 0: + m.indicators = iocs m.indicators.log = m.log try: diff --git a/src/mvt/common/command.py b/src/mvt/common/command.py index 12429417..6848e8c0 100644 --- a/src/mvt/common/command.py +++ b/src/mvt/common/command.py @@ -8,6 +8,7 @@ import logging import os import sys from datetime import datetime +from functools import cached_property from heapq import heappop, heappush from typing import Any, Optional @@ -84,18 +85,18 @@ class Command: self.timeline: ModuleTimeline = [] self.url_results: list[URLResult] = [] - # Load IOCs - self._create_storage() - self._setup_logging() - if iocs is not None: self.iocs = iocs - else: - self.iocs = Indicators(self.log) - self.iocs.load_indicators_files(self.ioc_files) self.alertstore = AlertStore() + @cached_property + def iocs(self) -> Indicators: + """Load indicators on first use. Nested commands share their parent's.""" + iocs = Indicators(self.log) + iocs.load_indicators_files(self.ioc_files) + return iocs + def _create_storage(self) -> None: if self.results_path and not os.path.exists(self.results_path): try: @@ -710,17 +711,30 @@ class Command: return ordered def run(self) -> None: + # The output folder and its command.log exist for a run, so that + # listing modules or rejecting a target leaves nothing behind. + # Resolving the module list can warn, so the log comes first. + self._create_storage() + self._setup_logging() + ordered_modules = self._ordered_modules() if ordered_modules is None: return - self._log_loaded_modules(ordered_modules) + # Read the indicators once the run is certain to happen, before + # init() does any work on the target, so that a missing indicators + # file is reported first and every module gets the same object. + iocs = self.iocs + # Commands announce their target from init(), so it goes before the + # module list. try: self.init() except NotImplementedError: pass + self._log_loaded_modules(ordered_modules) + executed_by_type: dict[type[MVTModule], MVTModule] = {} for module in ordered_modules: @@ -740,8 +754,8 @@ class Command: for dependency, resolved in self._module_dependencies(module) } - if self.iocs.total_ioc_count: - m.indicators = self.iocs + if iocs.total_ioc_count: + m.indicators = iocs m.indicators.log = m.log if self.serial: diff --git a/src/mvt/ios/cli.py b/src/mvt/ios/cli.py index 4560010a..f9f58885 100644 --- a/src/mvt/ios/cli.py +++ b/src/mvt/ios/cli.py @@ -318,8 +318,6 @@ def check_backup( if not cmd.resolve_backup_path(): ctx.exit(1) - log.info("Checking iTunes backup located at: %s", cmd.target_path) - cmd.run() cmd.show_alerts_brief() cmd.show_support_message() @@ -386,8 +384,6 @@ def check_fs( cmd.list_modules() return - log.info("Checking iOS filesystem located at: %s", dump_path) - cmd.run() cmd.show_alerts_brief() cmd.show_support_message() @@ -463,7 +459,6 @@ def check_sysdiagnose( cmd.list_modules() return - log.info("Checking iOS sysdiagnose at path: %s", sysdiagnose_path) cmd.run() cmd.show_alerts_brief() cmd.show_support_message() diff --git a/src/mvt/ios/cmd_check_backup.py b/src/mvt/ios/cmd_check_backup.py index 1b90584b..42b13c56 100644 --- a/src/mvt/ios/cmd_check_backup.py +++ b/src/mvt/ios/cmd_check_backup.py @@ -59,6 +59,9 @@ class CmdIOSCheckBackup(Command): self.name = "check-backup" self.modules = BACKUP_MODULES + MIXED_MODULES + def init(self) -> None: + self.log.info("Checking iTunes backup located at: %s", self.target_path) + def resolve_backup_path(self) -> bool: target_path = getattr(self, "target_path", None) if not isinstance(target_path, str) or not target_path: diff --git a/src/mvt/ios/cmd_check_fs.py b/src/mvt/ios/cmd_check_fs.py index e76146ec..3daed974 100644 --- a/src/mvt/ios/cmd_check_fs.py +++ b/src/mvt/ios/cmd_check_fs.py @@ -51,5 +51,8 @@ class CmdIOSCheckFS(Command): self.name = "check-fs" self.modules = FS_MODULES + MIXED_MODULES + def init(self) -> None: + self.log.info("Checking iOS filesystem located at: %s", self.target_path) + def module_init(self, module): module.is_fs_dump = True diff --git a/src/mvt/ios/cmd_check_sysdiagnose.py b/src/mvt/ios/cmd_check_sysdiagnose.py index 88a12db0..c700e6bb 100644 --- a/src/mvt/ios/cmd_check_sysdiagnose.py +++ b/src/mvt/ios/cmd_check_sysdiagnose.py @@ -81,6 +81,8 @@ class CmdIOSCheckSysdiagnose(Command): if not self.target_path: raise ValueError("A sysdiagnose path is required") + self.log.info("Checking iOS sysdiagnose at path: %s", self.target_path) + if os.path.isdir(self.target_path): self.sysdiagnose_format = "dir" parent_path = Path(self.target_path).absolute().parent diff --git a/tests/common/test_command.py b/tests/common/test_command.py index 8e09f57c..357d3b7d 100644 --- a/tests/common/test_command.py +++ b/tests/common/test_command.py @@ -5,8 +5,13 @@ import json import logging +from unittest.mock import patch + +import pytest +from click.testing import CliRunner from mvt.common.command import Command +from mvt.common.indicators import Indicators from mvt.common.module import MVTModule @@ -197,6 +202,86 @@ class RecordingCommand(Command): class TestCommand: + def test_listing_modules_does_not_load_indicators(self): + with patch("mvt.common.command.Indicators.load_indicators_files") as load: + cmd = RecordingCommand() + cmd.list_modules() + load.assert_not_called() + + def test_output_folder_is_created_by_a_run_not_by_listing(self, tmp_path): + output_path = tmp_path / "out" + with patch("mvt.common.command.Indicators.load_indicators_files"): + cmd = RecordingCommand(results_path=str(output_path)) + cmd.list_modules() + assert not output_path.exists() + cmd.run() + assert (output_path / "command.log").is_file() + + def test_indicators_load_once_and_are_shared(self, indicator_file, monkeypatch): + from mvt.common.config import settings + + monkeypatch.setattr(settings, "STIX2", "") + monkeypatch.setattr(Indicators, "_load_downloaded_indicators", lambda self: None) + original = Indicators.load_indicators_files + with patch.object( + Indicators, "load_indicators_files", autospec=True, side_effect=original + ) as load: + cmd = RecordingCommand(ioc_files=[indicator_file]) + load.assert_not_called() + indicators = cmd.iocs + assert indicators.total_ioc_count == 9 + assert len(indicators.ioc_collections) == 1 + assert cmd.iocs is indicators + child = RecordingCommand(iocs=indicators) + assert child.iocs is indicators + load.assert_called_once_with(indicators, [indicator_file]) + + @pytest.mark.parametrize("assign", [False, True]) + def test_supplied_empty_indicators_are_not_loaded(self, assign): + indicators = Indicators(logging.getLogger(__name__)) + with patch.object(Indicators, "load_indicators_files") as load: + cmd = RecordingCommand(iocs=None if assign else indicators) + if assign: + cmd.iocs = indicators + assert cmd.iocs is indicators + assert cmd.iocs.total_ioc_count == 0 + load.assert_not_called() + + @pytest.mark.parametrize("list_modules", [False, True]) + def test_backup_cli_does_not_load_indicators_before_analysis( + self, tmp_path, list_modules + ): + from mvt.ios.cli import check_backup + + args = [str(tmp_path)] + if list_modules: + args.insert(0, "--list-modules") + with patch.object(Indicators, "load_indicators_files") as load: + result = CliRunner().invoke(check_backup, args) + assert result.exit_code == (0 if list_modules else 1) + load.assert_not_called() + + def test_run_checks_synthetic_indicators(self, indicator_file, monkeypatch): + from mvt.common.config import settings + + monkeypatch.setattr(settings, "STIX2", "") + monkeypatch.setattr(Indicators, "_load_downloaded_indicators", lambda self: None) + + class MatchingModule(RecordingModule): + def run(self): + self.results = ["https://example.org/test"] + + def check_indicators(self): + self.detected = [ + url for url in self.results if self.indicators.check_domain(url) + ] + + cmd = RecordingCommand(ioc_files=[indicator_file]) + cmd.modules = [MatchingModule] + cmd.run() + assert cmd.executed[0].detected == ["https://example.org/test"] + assert cmd.executed[0].indicators is cmd.iocs + def setup_method(self): RecordingModule.run_order = [] diff --git a/tests/test_check_ios_backup.py b/tests/test_check_ios_backup.py index eb10b131..3290e6d1 100644 --- a/tests/test_check_ios_backup.py +++ b/tests/test_check_ios_backup.py @@ -19,6 +19,14 @@ class TestCheckBackupCommand: result = runner.invoke(check_backup, [path]) assert result.exit_code == 0 + def test_check_logs_the_backup_path_to_the_command_log(self, tmp_path): + path = get_ios_backup_folder() + output_path = tmp_path / "out" + result = CliRunner().invoke(check_backup, ["--output", str(output_path), path]) + assert result.exit_code == 0 + command_log = (output_path / "command.log").read_text(encoding="utf-8") + assert f"Checking iTunes backup located at: {path}" in command_log + def test_check_finds_backup_in_subfolder(self, tmp_path, caplog): runner = CliRunner() backup_path = tmp_path / "MobileSync" / "Backup" / "device-id" From ffae240355a9245450ecbbc9f089c592b3423000 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Tue, 8 Sep 2026 01:03:05 +0100 Subject: [PATCH 14/16] Skip AppleDouble sidecars when listing a sysdiagnose (#922) Device-generated sysdiagnose archives carry a ._name entry beside every file that has extended attributes, an ACL or Finder info; one iOS 26 archive held 1234 of them among 3648 members, and the count grows with each release. bsdtar folds them back into the file on extraction and hides them from listings, but tarfile returns them as regular members, so check-sysdiagnose extracted them and handed them to every module. A module that globs for plists or logs then tries to parse AppleDouble headers and logs one warning per sidecar. Leave them out of the file list, both for archives and for folders extracted on a system that keeps them as files. --- src/mvt/ios/cmd_check_sysdiagnose.py | 7 +++++++ tests/test_cmd_check_sysdiagnose.py | 3 +++ 2 files changed, 10 insertions(+) diff --git a/src/mvt/ios/cmd_check_sysdiagnose.py b/src/mvt/ios/cmd_check_sysdiagnose.py index c700e6bb..26e4d49f 100644 --- a/src/mvt/ios/cmd_check_sysdiagnose.py +++ b/src/mvt/ios/cmd_check_sysdiagnose.py @@ -88,6 +88,8 @@ class CmdIOSCheckSysdiagnose(Command): parent_path = Path(self.target_path).absolute().parent for root, _, filenames in os.walk(self.target_path): for filename in filenames: + if filename.startswith("._"): + continue absolute_path = os.path.join(root, filename) file_path = os.path.relpath(absolute_path, parent_path) self.sysdiagnose_files.append(file_path) @@ -137,6 +139,11 @@ class CmdIOSCheckSysdiagnose(Command): if not member_path.parts: continue + # AppleDouble sidecars (._name) carry a file's extended attributes, + # not sysdiagnose content. Device archives hold hundreds of them; + # bsdtar hides them from listings, tarfile does not. + if member_path.name.startswith("._"): + continue archive_roots.add(member_path.parts[0]) if member.isdir(): diff --git a/tests/test_cmd_check_sysdiagnose.py b/tests/test_cmd_check_sysdiagnose.py index 020c8616..b00b648b 100644 --- a/tests/test_cmd_check_sysdiagnose.py +++ b/tests/test_cmd_check_sysdiagnose.py @@ -34,6 +34,7 @@ def _create_sysdiagnose_folder(tmp_path): "sysdiagnose_2024.01.02_03-04-05+0200.tar.gz", encoding="utf-8" ) (folder / "report.ips").write_text('{"bug_type": 210}\nbody', encoding="utf-8") + (folder / "._artifact.txt").write_bytes(b"\x00\x05\x16\x07AppleDouble") return folder @@ -67,6 +68,7 @@ def test_check_sysdiagnose_from_folder(tmp_path): assert _test_module(command).ips_files == [ {"file_path": str(tmp_path / "sysdiagnose" / "report.ips"), "bug_type": 210} ] + assert "sysdiagnose/._artifact.txt" not in command.sysdiagnose_files def test_check_sysdiagnose_from_archive_closes_archive(tmp_path): @@ -83,6 +85,7 @@ def test_check_sysdiagnose_from_archive_closes_archive(tmp_path): } ] assert command.sysdiagnose_archive is None + assert "sysdiagnose/._artifact.txt" not in command.sysdiagnose_files def test_archive_is_extracted_once_and_unsafe_members_are_skipped(tmp_path): From eceafdc785db4c77ae839db1b3885ba307caa2f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Tue, 8 Sep 2026 01:04:43 +0100 Subject: [PATCH 15/16] Keep the test suite out of the user's MVT folders (#923) The tests read the settings file of whoever runs them, and importing mvt.common.config writes it back. On a machine whose config.yaml sets NETWORK_ACCESS_ALLOWED to false, the plugin update and URL batch tests fail before their mocked requests are reached, and every Command run in the suite parses the indicators downloaded on that machine, which made the suite take minutes instead of seconds. MVT_CONFIG_FOLDER and MVT_DATA_FOLDER in the environment now relocate the settings file and the downloaded indicators with their update-check state. The test conftest points both at a throwaway folder before any mvt module is imported, and removes it at exit. Subprocesses started by the tests inherit the variables; the isolated interpreter helper already gives them a temporary home. On the machine that prompted this the suite goes from 6 failures in six minutes to none in seven seconds, and config.yaml is left alone. --- src/mvt/common/config.py | 4 +++- src/mvt/common/indicators.py | 12 +++++++----- tests/conftest.py | 27 ++++++++++++++++++++------- 3 files changed, 30 insertions(+), 13 deletions(-) diff --git a/src/mvt/common/config.py b/src/mvt/common/config.py index 24bf4072..fc4b2ea5 100644 --- a/src/mvt/common/config.py +++ b/src/mvt/common/config.py @@ -12,7 +12,9 @@ from pydantic_settings import ( YamlConfigSettingsSource, ) -MVT_CONFIG_FOLDER = user_config_dir("mvt") +# MVT_CONFIG_FOLDER in the environment relocates the settings file, so that +# a test run or a scripted install never touches the user's own. +MVT_CONFIG_FOLDER = os.environ.get("MVT_CONFIG_FOLDER") or user_config_dir("mvt") MVT_CONFIG_PATH = os.path.join(MVT_CONFIG_FOLDER, "config.yaml") diff --git a/src/mvt/common/indicators.py b/src/mvt/common/indicators.py index 94afadf1..3886e34f 100644 --- a/src/mvt/common/indicators.py +++ b/src/mvt/common/indicators.py @@ -18,7 +18,9 @@ from appdirs import user_data_dir from .config import settings from .url import URL -MVT_DATA_FOLDER = user_data_dir("mvt") +# MVT_DATA_FOLDER in the environment relocates the downloaded indicators and +# the update-check state kept next to them. +MVT_DATA_FOLDER = os.environ.get("MVT_DATA_FOLDER") or user_data_dir("mvt") MVT_INDICATORS_FOLDER = os.path.join(MVT_DATA_FOLDER, "indicators") logger = logging.getLogger(__name__) @@ -71,7 +73,9 @@ class Indicators: if os.path.isfile(path) and path.lower().endswith(".stix2"): self.parse_stix2(path) elif os.path.isdir(path): - for file in glob.glob(os.path.join(path, "**", "*.stix2"), recursive=True): + for file in glob.glob( + os.path.join(path, "**", "*.stix2"), recursive=True + ): self.parse_stix2(file) else: self.log.error( @@ -518,9 +522,7 @@ class Indicators: the original URL order. """ batches = [list(urls) if urls else [] for urls in url_batches] - unique_urls = list( - dict.fromkeys(url for urls in batches for url in urls) - ) + unique_urls = list(dict.fromkeys(url for urls in batches for url in urls)) if not unique_urls: return [None] * len(batches) diff --git a/tests/conftest.py b/tests/conftest.py index 06a890a9..40a084a8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,20 +3,26 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +import atexit import logging import os +import shutil +import tempfile import pytest -from mvt.common.cli_plugins import ( - MVT_ANDROID_CUSTOM_COMMANDS_ENV, - MVT_CUSTOM_COMMANDS_ENV, - MVT_IOS_CUSTOM_COMMANDS_ENV, -) -from mvt.common.indicators import Indicators - from .artifacts.generate_stix import generate_test_stix_file +# The suite must neither read nor write the developer's own MVT settings, +# downloaded indicators or update-check state, and mvt.common.config saves +# the settings file as soon as it is imported. Both folders are redirected +# before any mvt module is imported, which is why this file imports none at +# the top; the subprocesses the tests start inherit the variables. +MVT_TEST_HOME = tempfile.mkdtemp(prefix="mvt-tests-") +atexit.register(shutil.rmtree, MVT_TEST_HOME, ignore_errors=True) +os.environ["MVT_CONFIG_FOLDER"] = os.path.join(MVT_TEST_HOME, "config") +os.environ["MVT_DATA_FOLDER"] = os.path.join(MVT_TEST_HOME, "data") + @pytest.fixture(scope="session", autouse=True) def indicator_file(request, tmp_path_factory): @@ -47,6 +53,8 @@ def indicators_factory(indicator_file): android_property_names=[], files_sha256=[], ): + from mvt.common.indicators import Indicators + ind = Indicators(log=logging.getLogger()) ind.parse_stix2(indicator_file) @@ -77,6 +85,11 @@ def restore_cli_commands(monkeypatch): """ from mvt.android.cli import cli as android_cli from mvt.cli import cli as neutral_cli + from mvt.common.cli_plugins import ( + MVT_ANDROID_CUSTOM_COMMANDS_ENV, + MVT_CUSTOM_COMMANDS_ENV, + MVT_IOS_CUSTOM_COMMANDS_ENV, + ) from mvt.ios.cli import cli as ios_cli groups = (neutral_cli, ios_cli, android_cli) From b4b0a0647080bd0bbcddb7508e6427fd51ea6847 Mon Sep 17 00:00:00 2001 From: DonnchaC <3081375+DonnchaC@users.noreply.github.com> Date: Tue, 8 Sep 2026 18:16:19 +0000 Subject: [PATCH 16/16] Add new iOS versions and build numbers --- src/mvt/ios/data/ios_versions.json | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/mvt/ios/data/ios_versions.json b/src/mvt/ios/data/ios_versions.json index 9b4613bb..d8a93c91 100644 --- a/src/mvt/ios/data/ios_versions.json +++ b/src/mvt/ios/data/ios_versions.json @@ -1271,5 +1271,9 @@ { "version": "26.6.1", "build": "23G83" + }, + { + "version": "26.6.2", + "build": "23G90" } ] \ No newline at end of file