Fix ruff F821: use self.log instead of undefined log

When `check-backup` is run against an encrypted backup, Manifest.db
cannot be opened as a plain SQLite database. Previously this caused
a flood of confusing "file is not a database" errors across all modules.

Now the Manifest module detects the sqlite3.DatabaseError on its first
query and raises a new EncryptedBackupError. This exception propagates
out of run_module() and is caught in Command.run(), which logs a clear
critical message instructing the user to decrypt the backup first with
`mvt-ios decrypt-backup`, then stops the analysis immediately.

Fixes #769

pyproject.toml

@@ -24,8 +24,7 @@ dependencies = [
     "simplejson==3.20.2",
     "packaging==26.0",
     "appdirs==1.4.4",
-    "iphone_backup_decrypt==0.9.0",
-    "pycryptodome>=3.18",
+    "iOSbackup==0.9.925",
     "adb-shell[usb]==0.4.4",
     "libusb1==3.3.1",
     "cryptography==46.0.6",

src/mvt/common/command.py

@@ -11,7 +11,7 @@ from datetime import datetime
 from typing import Optional
 
 from mvt.common.indicators import Indicators
-from mvt.common.module import MVTModule, run_module, save_timeline
+from mvt.common.module import EncryptedBackupError, MVTModule, run_module, save_timeline
 from mvt.common.utils import (
     convert_datetime_to_iso,
     generate_hashes_from_path,
@@ -244,7 +244,14 @@ class Command:
             except NotImplementedError:
                 pass
 
-            run_module(m)
+            try:
+                run_module(m)
+            except EncryptedBackupError:
+                self.log.critical(
+                    "The backup appears to be encrypted. "
+                    "Please decrypt it first using `mvt-ios decrypt-backup`."
+                )
+                return
 
             self.executed.append(m)
 

src/mvt/common/module.py

@@ -21,6 +21,10 @@ class DatabaseCorruptedError(Exception):
     pass
 
 
+class EncryptedBackupError(Exception):
+    pass
+
+
 class InsufficientPrivileges(Exception):
     pass
 
@@ -169,6 +173,8 @@ def run_module(module: MVTModule) -> None:
 
     try:
         exec_or_profile("module.run()", globals(), locals())
+    except EncryptedBackupError:
+        raise
     except NotImplementedError:
         module.log.exception(
             "The run() procedure of module %s was not implemented yet!",

src/mvt/ios/decrypt.py

@@ -6,146 +6,17 @@
 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 iphone_backup_decrypt import EncryptedBackup
-from iphone_backup_decrypt import google_iphone_dataprotection
+from iOSbackup import iOSbackup
 
 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
@@ -184,27 +55,41 @@ 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")
-        # Extract a decrypted Manifest.db to the destination folder.
-        self._backup.save_manifest_file(output_filename=manifest_path)
+        # 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"]
 
-        # 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[:2], file_id
-                )
+                source_file_path = os.path.join(self.backup_path, file_id[0:2], file_id)
                 if not os.path.exists(source_file_path):
                     log.debug(
                         "Skipping file %s. File not found in encrypted backup directory.",
@@ -212,26 +97,24 @@ class DecryptBackup:
                     )
                     continue
 
-                item_folder = os.path.join(self.dest_path, file_id[:2])
-                os.makedirs(item_folder, exist_ok=True)
+                item_folder = os.path.join(self.dest_path, file_id[0:2])
+                if not os.path.exists(item_folder):
+                    os.makedirs(item_folder)
 
-                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)
+                # 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()
 
         # Copying over the root plist files as well.
         for file_name in os.listdir(self.backup_path):
@@ -272,23 +155,20 @@ class DecryptBackup:
             return
 
         try:
-            self._backup = MVTEncryptedBackup(
-                backup_directory=self.backup_path,
-                passphrase=password,
+            self._backup = iOSbackup(
+                udid=os.path.basename(self.backup_path),
+                cleartextpassword=password,
+                backuproot=os.path.dirname(self.backup_path),
             )
-            # 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, ValueError)
-                and "passphrase" in str(exc).lower()
+                isinstance(exc, KeyError)
+                and len(exc.args) > 0
+                and exc.args[0] == b"KEY"
             ):
                 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(
@@ -331,14 +211,12 @@ class DecryptBackup:
 
         try:
             key_bytes_raw = binascii.unhexlify(key_bytes)
-            self._backup = MVTEncryptedBackup(
-                backup_directory=self.backup_path,
-                derived_key=key_bytes_raw,
+            self._backup = iOSbackup(
+                udid=os.path.basename(self.backup_path),
+                derivedkey=key_bytes_raw,
+                backuproot=os.path.dirname(self.backup_path),
             )
-            # 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?"
@@ -349,7 +227,7 @@ class DecryptBackup:
         if not self._backup:
             return
 
-        self._decryption_key = self._backup.get_decryption_key()
+        self._decryption_key = self._backup.getDecryptionKey()
         log.info(
             'Derived decryption key for backup at path %s is: "%s"',
             self.backup_path,

src/mvt/ios/modules/backup/manifest.py

@@ -8,9 +8,10 @@ import io
 import logging
 import os
 import plistlib
+import sqlite3
 from typing import Optional
 
-from mvt.common.module import DatabaseNotFoundError
+from mvt.common.module import DatabaseNotFoundError, EncryptedBackupError
 from mvt.common.url import URL
 from mvt.common.utils import convert_datetime_to_iso, convert_unix_to_iso
 
@@ -127,7 +128,14 @@ class Manifest(IOSExtraction):
         conn = self._open_sqlite_db(manifest_db_path)
         cur = conn.cursor()
 
-        cur.execute("SELECT * FROM Files;")
+        try:
+            cur.execute("SELECT * FROM Files;")
+        except sqlite3.DatabaseError:
+            conn.close()
+            raise EncryptedBackupError(
+                "Manifest.db is not a valid SQLite database. "
+                "The backup may be encrypted."
+            )
         names = [description[0] for description in cur.description]
 
         for file_entry in cur: