mirror of
https://github.com/mvt-project/mvt.git
synced 2026-09-03 00:21:07 +02:00
Merge pull request #774 from mvt-project/backup_extraction
Replace iOSbackup with iphone_backup_decrypt
This commit is contained in:
+2
-1
@@ -24,7 +24,8 @@ dependencies = [
|
||||
"simplejson==4.1.1",
|
||||
"packaging==26.3",
|
||||
"appdirs==1.4.4",
|
||||
"iOSbackup==0.9.925",
|
||||
"iphone_backup_decrypt==0.9.0",
|
||||
"pycryptodome>=3.20.0",
|
||||
"adb-shell[usb]==0.4.4",
|
||||
"libusb1==3.4.0",
|
||||
"cryptography==50.0.0",
|
||||
|
||||
@@ -39,6 +39,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)"
|
||||
|
||||
+14
-2
@@ -24,6 +24,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,
|
||||
@@ -45,6 +46,10 @@ from mvt.common.help import (
|
||||
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
||||
)
|
||||
from mvt.common.password import prompt_password
|
||||
from .decrypt_config import (
|
||||
DEFAULT_DECRYPT_WORKERS,
|
||||
MAX_DECRYPT_WORKERS,
|
||||
)
|
||||
|
||||
# The commands import what they run only when they are invoked. This module is
|
||||
# imported at every start of mvt-ios, including by shell completion on every
|
||||
@@ -128,6 +133,13 @@ def version():
|
||||
"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",
|
||||
@@ -146,10 +158,10 @@ def version():
|
||||
@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):
|
||||
def decrypt_backup(ctx, destination, jobs, password, key_file, hashes, backup_path):
|
||||
from .decrypt import DecryptBackup
|
||||
|
||||
backup = DecryptBackup(backup_path, destination)
|
||||
backup = DecryptBackup(backup_path, destination, max_workers=jobs)
|
||||
|
||||
if key_file:
|
||||
if MVT_IOS_BACKUP_PASSWORD in os.environ:
|
||||
|
||||
+294
-76
@@ -6,18 +6,179 @@
|
||||
import binascii
|
||||
import glob
|
||||
import logging
|
||||
import multiprocessing
|
||||
import os
|
||||
import os.path
|
||||
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
|
||||
|
||||
from iOSbackup import iOSbackup
|
||||
from iphone_backup_decrypt import EncryptedBackup
|
||||
from iphone_backup_decrypt import google_iphone_dataprotection
|
||||
from iphone_backup_decrypt.utils import FilePlist
|
||||
|
||||
from .decrypt_config import DEFAULT_DECRYPT_WORKERS, MAX_DECRYPT_WORKERS
|
||||
|
||||
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 extract_file_by_id(self, *, file_id, file_bplist, output_filename):
|
||||
"""Extract one manifest entry without loading the whole file into memory."""
|
||||
self._read_and_unlock_keybag()
|
||||
file_plist = FilePlist(file_bplist)
|
||||
|
||||
if file_plist.encryption_key is None:
|
||||
source_filename = os.path.join(
|
||||
self._backup_directory, file_id[:2], file_id
|
||||
)
|
||||
shutil.copy2(source_filename, output_filename)
|
||||
return
|
||||
|
||||
inner_key = self._keybag.unwrapKeyForClass(
|
||||
file_plist.protection_class, file_plist.encryption_key
|
||||
)
|
||||
self._decrypt_file_to_disk(
|
||||
file_id=file_id,
|
||||
key=inner_key,
|
||||
file_plist=file_plist,
|
||||
output_filepath=output_filename,
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
@@ -26,19 +187,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
|
||||
self._backup = None
|
||||
self._decryption_key = None
|
||||
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.
|
||||
@@ -58,21 +268,6 @@ class DecryptBackup:
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def _process_file(
|
||||
self, relative_path: str, domain: str, item, file_id: str, item_folder: str
|
||||
) -> None:
|
||||
assert self._backup is not 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:
|
||||
assert self._backup is not None
|
||||
assert self.dest_path is not None
|
||||
@@ -81,53 +276,71 @@ class DecryptBackup:
|
||||
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)
|
||||
# Extract a decrypted Manifest.db to the destination folder.
|
||||
self._backup.save_manifest_file(output_filename=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"]
|
||||
|
||||
# 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)
|
||||
if not Path(source_file_path).resolve().is_relative_to(Path(self.backup_path).resolve()):
|
||||
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
|
||||
|
||||
item_folder = os.path.join(self.dest_path, file_id[0:2]) # type: ignore[arg-type]
|
||||
if not Path(os.path.join(item_folder, file_id)).resolve().is_relative_to(Path(self.dest_path).resolve()):
|
||||
log.warning("Skipping unsafe file_id: %r", file_id)
|
||||
continue
|
||||
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"]
|
||||
|
||||
pool.apply_async(
|
||||
self._process_file,
|
||||
args=(relative_path, domain, item, file_id, item_folder),
|
||||
# Iterate over all files in the backup and decrypt them,
|
||||
# preserving the XX/file_id directory structure that downstream
|
||||
# modules expect.
|
||||
backup_root = Path(self.backup_path).resolve()
|
||||
dest_root = Path(self.dest_path).resolve()
|
||||
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"
|
||||
)
|
||||
except Exception as exc:
|
||||
log.error("Failed to decrypt file %s: %s", relative_path, exc)
|
||||
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
|
||||
|
||||
pool.close()
|
||||
pool.join()
|
||||
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_path=output_path,
|
||||
relative_path=relative_path,
|
||||
domain=domain,
|
||||
)
|
||||
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):
|
||||
@@ -168,20 +381,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(
|
||||
@@ -224,12 +440,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?"
|
||||
@@ -240,7 +458,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,
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 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/
|
||||
|
||||
"""Configuration shared by the iOS CLI and backup decryption implementation."""
|
||||
|
||||
DEFAULT_DECRYPT_WORKERS = 4
|
||||
MAX_DECRYPT_WORKERS = 32
|
||||
@@ -0,0 +1,191 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-2023 The MVT Authors.
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import logging
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
from Crypto.Cipher import AES
|
||||
|
||||
from mvt.ios.decrypt import DecryptBackup, MVTEncryptedBackup
|
||||
|
||||
|
||||
def _encrypted_file(backup_path, file_id, key, plaintext):
|
||||
padding_length = AES.block_size - (len(plaintext) % AES.block_size)
|
||||
padded = plaintext + bytes([padding_length]) * padding_length
|
||||
encrypted = AES.new(key, AES.MODE_CBC, iv=b"\x00" * AES.block_size).encrypt(
|
||||
padded
|
||||
)
|
||||
source_path = backup_path / file_id[:2] / file_id
|
||||
source_path.parent.mkdir(parents=True)
|
||||
source_path.write_bytes(encrypted)
|
||||
|
||||
|
||||
def test_extract_file_by_id_preserves_bytes_with_wrong_manifest_size(
|
||||
mocker, tmp_path
|
||||
):
|
||||
file_id = "ab" + "1" * 38
|
||||
plaintext = b"complete decrypted content"
|
||||
inner_key = b"k" * 32
|
||||
_encrypted_file(tmp_path, file_id, inner_key, plaintext)
|
||||
|
||||
file_plist = mocker.Mock(
|
||||
encryption_key=b"wrapped-key",
|
||||
protection_class=1,
|
||||
filesize=1,
|
||||
mtime=None,
|
||||
)
|
||||
mocker.patch("mvt.ios.decrypt.FilePlist", return_value=file_plist)
|
||||
|
||||
backup = MVTEncryptedBackup(
|
||||
backup_directory=str(tmp_path), derived_key=b"d" * 32
|
||||
)
|
||||
mocker.patch.object(backup, "_read_and_unlock_keybag", return_value=True)
|
||||
backup._keybag = mocker.Mock()
|
||||
backup._keybag.unwrapKeyForClass.return_value = inner_key
|
||||
streaming_decrypt = mocker.spy(backup, "_decrypt_file_to_disk")
|
||||
output_path = tmp_path / "output"
|
||||
|
||||
backup.extract_file_by_id(
|
||||
file_id=file_id,
|
||||
file_bplist=b"plist",
|
||||
output_filename=str(output_path),
|
||||
)
|
||||
|
||||
assert output_path.read_bytes() == plaintext
|
||||
streaming_decrypt.assert_called_once()
|
||||
|
||||
|
||||
def test_extract_file_by_id_copies_unencrypted_files(mocker, tmp_path):
|
||||
file_id = "cd" + "2" * 38
|
||||
source_path = tmp_path / file_id[:2] / file_id
|
||||
source_path.parent.mkdir(parents=True)
|
||||
source_path.write_bytes(b"plain content")
|
||||
|
||||
file_plist = mocker.Mock(encryption_key=None)
|
||||
mocker.patch("mvt.ios.decrypt.FilePlist", return_value=file_plist)
|
||||
backup = MVTEncryptedBackup(
|
||||
backup_directory=str(tmp_path), derived_key=b"d" * 32
|
||||
)
|
||||
mocker.patch.object(backup, "_read_and_unlock_keybag", return_value=True)
|
||||
output_path = tmp_path / "output"
|
||||
|
||||
backup.extract_file_by_id(
|
||||
file_id=file_id,
|
||||
file_bplist=b"plist",
|
||||
output_filename=str(output_path),
|
||||
)
|
||||
|
||||
assert output_path.read_bytes() == b"plain content"
|
||||
|
||||
|
||||
def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_path):
|
||||
backup_path = tmp_path / "backup"
|
||||
destination = tmp_path / "destination"
|
||||
outside = tmp_path / "outside"
|
||||
backup_path.mkdir()
|
||||
destination.mkdir()
|
||||
outside.mkdir()
|
||||
|
||||
safe_file_id = "ef" + "3" * 38
|
||||
unsafe_file_id = "../../outside-file"
|
||||
symlink_file_id = "ab" + "4" * 38
|
||||
for file_id in (safe_file_id, symlink_file_id):
|
||||
source_path = backup_path / file_id[:2] / file_id
|
||||
source_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
source_path.write_bytes(b"encrypted")
|
||||
(destination / "ab").symlink_to(outside, target_is_directory=True)
|
||||
|
||||
cursor = mocker.MagicMock()
|
||||
cursor.__iter__.return_value = iter(
|
||||
[
|
||||
(safe_file_id, "Domain", "safe", b"plist"),
|
||||
(unsafe_file_id, "Domain", "unsafe", b"plist"),
|
||||
(symlink_file_id, "Domain", "symlink", b"plist"),
|
||||
]
|
||||
)
|
||||
cursor_context = mocker.MagicMock()
|
||||
cursor_context.__enter__.return_value = cursor
|
||||
|
||||
backup = mocker.MagicMock()
|
||||
backup.manifest_db_cursor.return_value = cursor_context
|
||||
|
||||
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), max_workers=1
|
||||
)
|
||||
decryptor._backup = backup
|
||||
|
||||
decryptor.process_backup()
|
||||
|
||||
assert (destination / safe_file_id[:2] / safe_file_id).read_bytes() == b"decrypted"
|
||||
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
|
||||
@@ -624,16 +624,15 @@ wheels = [
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "iosbackup"
|
||||
version = "0.9.925"
|
||||
name = "iphone-backup-decrypt"
|
||||
version = "0.9.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "nskeyedunarchiver" },
|
||||
{ name = "pycryptodome" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/db/b8/4cd52322deceb942b9e18b127d45d112c2f7a3ec7821ab528659d4f04275/iOSbackup-0.9.925.tar.gz", hash = "sha256:33545a9249e5b3faaadf1ee782fe6bdfcdb70fae0defba1acee336a65f93d1ca", size = 25228, upload-time = "2022-10-16T01:02:46.474Z" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/6f/e7/bcdacdec21d628122ba240e7f742ab2175149e58672be63af55ff37a0f28/iphone_backup_decrypt-0.9.0.tar.gz", hash = "sha256:13b18fef3c8e3af627914f8c1a429bbc5555dfb0505239ba49efe99984cc0c96", size = 16125, upload-time = "2024-09-18T15:50:12.179Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/e0/4e/8da5fdc2df642080c79f128b9412a6f33768a610a58d59fbdb7b4d019364/iOSbackup-0.9.925-py3-none-any.whl", hash = "sha256:348edab2b82499d55c17c4bfe02fc6ae7915a2cc8da2728893ed1e9ae61bcef3", size = 18326, upload-time = "2022-10-16T01:02:44.164Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/b8/94/64a31be93f72e0a254bde68e4cf7d24aef37a0a985754a197fa1b028a665/iphone_backup_decrypt-0.9.0-py3-none-any.whl", hash = "sha256:55b5adfafac757f58aa6444b83a4cc2c20cdd699c6ff1d2f4b549936a5dad92c", size = 15767, upload-time = "2024-09-18T15:50:10.537Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1022,11 +1021,12 @@ dependencies = [
|
||||
{ name = "betterproto2" },
|
||||
{ name = "click" },
|
||||
{ name = "cryptography" },
|
||||
{ name = "iosbackup" },
|
||||
{ name = "iphone-backup-decrypt" },
|
||||
{ name = "libusb1" },
|
||||
{ name = "nskeyedunarchiver" },
|
||||
{ name = "packaging" },
|
||||
{ name = "pyahocorasick" },
|
||||
{ name = "pycryptodome" },
|
||||
{ name = "pydantic" },
|
||||
{ name = "pydantic-settings" },
|
||||
{ name = "python-dateutil" },
|
||||
@@ -1068,11 +1068,12 @@ requires-dist = [
|
||||
{ name = "betterproto2", specifier = "==0.10.0" },
|
||||
{ name = "click", specifier = "==8.4.2" },
|
||||
{ name = "cryptography", specifier = "==50.0.0" },
|
||||
{ name = "iosbackup", specifier = "==0.9.925" },
|
||||
{ name = "iphone-backup-decrypt", specifier = "==0.9.0" },
|
||||
{ name = "libusb1", specifier = "==3.4.0" },
|
||||
{ name = "nskeyedunarchiver", specifier = "==1.5.2" },
|
||||
{ name = "packaging", specifier = "==26.3" },
|
||||
{ name = "pyahocorasick", specifier = "==2.3.1" },
|
||||
{ name = "pycryptodome", specifier = ">=3.20.0" },
|
||||
{ name = "pydantic", specifier = "==2.13.4" },
|
||||
{ name = "pydantic-settings", specifier = "==2.15.0" },
|
||||
{ name = "python-dateutil", specifier = "==2.9.0.post0" },
|
||||
|
||||
Reference in New Issue
Block a user