From 39a26d0f0be9d2696050a745b5becbc539e24d8a Mon Sep 17 00:00:00 2001 From: Janik Besendorf Date: Sun, 12 Apr 2026 10:51:56 +0200 Subject: [PATCH 1/2] 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 | 240 +++++++++++++++++++++++++++++++---------- 2 files changed, 183 insertions(+), 60 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index fedfff9ae..e8ef917ea 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 ffb2cb757..d615d051b 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 @@ def is_encrypted(backup_path: str) -> bool: 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 @@ def process_backup(self) -> None: ) continue - item_folder = os.path.join(self.dest_path, file_id[0:2]) - if not os.path.exists(item_folder): - os.makedirs(item_folder) - - # 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"] + item_folder = os.path.join(self.dest_path, file_id[:2]) + os.makedirs(item_folder, exist_ok=True) - 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 @@ def decrypt_with_password(self, password: str) -> None: 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 @@ def decrypt_with_key_file(self, key_file: str) -> None: 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 @@ def get_key(self) -> None: 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 2/2] Restore concurrent backup decryption --- src/mvt/common/help.py | 1 + src/mvt/ios/cli.py | 18 +++- src/mvt/ios/decrypt.py | 151 +++++++++++++++++++++++-------- tests/ios_backup/test_decrypt.py | 68 +++++++++++++- 4 files changed, 197 insertions(+), 41 deletions(-) diff --git a/src/mvt/common/help.py b/src/mvt/common/help.py index 5101f939d..746153822 100644 --- a/src/mvt/common/help.py +++ b/src/mvt/common/help.py @@ -28,6 +28,7 @@ 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 c338fa549..0bd362a23 100644 --- a/src/mvt/ios/cli.py +++ b/src/mvt/ios/cli.py @@ -33,6 +33,7 @@ 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 .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 2ed3efe82..faf3000e6 100644 --- a/src/mvt/ios/decrypt.py +++ b/src/mvt/ios/decrypt.py @@ -12,6 +12,13 @@ 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 @@ 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 @@ def process_backup(self) -> None: # 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 - - 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( + 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) + + 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 28eeb83bf..25decce51 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 extract_file_by_id(*, output_filename, **kwargs): 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 extract_file_by_id(*, output_filename, **kwargs): 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