Fix betterproto2 migration: update generated proto code and callers

The dependency switch from betterproto to betterproto2 was incomplete.
This updates all affected files to use the betterproto2 API:

- tombstone.py: rewrite generated code to use betterproto2.field() with
  explicit TYPE_* constants, repeated/optional/group flags, and map_meta()
  for map fields
- tombstone_crashes.py: update import and fix to_dict() call to use
  keyword-only casing= argument required by betterproto2
- pyproject.toml: replace betterproto[compiler] dev dep with betterproto2-compiler
- Makefile: update protoc plugin flag to --python_betterproto2_out

.github/workflows/add-issue-to-project.yml

@@ -11,7 +11,7 @@ jobs:
     name: Add issue to project
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/add-to-project@v0.5.0
+      - uses: actions/add-to-project@v1
         with:
           # You can target a project in a different organization
           # to the issue

README.md

@@ -4,9 +4,6 @@
 
 # Mobile Verification Toolkit
 
-> [!IMPORTANT]  
-> Soon we will merge the v3 pull request which will result in breaking changes. If you rely on mvt output in other script make sure to the the branch before we merge. More details: https://github.com/mvt-project/mvt/issues/757
-
 [![](https://img.shields.io/pypi/v/mvt)](https://pypi.org/project/mvt/)
 [![Documentation Status](https://readthedocs.org/projects/mvt/badge/?version=latest)](https://docs.mvt.re/en/latest/?badge=latest)
 [![CI](https://github.com/mvt-project/mvt/actions/workflows/tests.yml/badge.svg)](https://github.com/mvt-project/mvt/actions/workflows/tests.yml)

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/indicators.py

@@ -100,17 +100,6 @@ class Indicators:
         key, value = indicator.get("pattern", "").strip("[]").split("=")
         key = key.strip()
 
-        # Normalize hash algorithm keys so that both the STIX2-spec-compliant
-        # form (e.g. file:hashes.'SHA-256', which requires quotes around
-        # algorithm names that contain hyphens) and the non-standard lowercase
-        # form (e.g. file:hashes.sha256) are accepted.  Strip single quotes and
-        # hyphens from the algorithm name only, then lowercase it.
-        for sep in ("hashes.", "cert."):
-            if sep in key:
-                prefix, _, algo = key.partition(sep)
-                key = prefix + sep + algo.replace("'", "").replace("-", "").lower()
-                break
-
         if key == "domain-name:value":
             # We force domain names to lower case.
             self._add_indicator(

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/mixed/sms.py

@@ -123,11 +123,6 @@ class SMS(IOSExtraction):
                 """
                 )
                 items = list(cur)
-            elif "no such table" in str(exc):
-                self.log.info(
-                    "No SMS tables found in the database, skipping: %s", exc
-                )
-                return
             else:
                 raise exc
         names = [description[0] for description in cur.description]

src/mvt/ios/modules/mixed/sms_attachments.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-import sqlite3
 from base64 import b64encode
 from typing import Optional, Union
 
@@ -80,29 +79,21 @@ class SMSAttachments(IOSExtraction):
 
         conn = self._open_sqlite_db(self.file_path)
         cur = conn.cursor()
-        try:
-            cur.execute(
-                """
-                SELECT
-                    attachment.ROWID as "attachment_id",
-                    attachment.*,
-                    message.service as "service",
-                    handle.id as "phone_number"
-                FROM attachment
-                LEFT JOIN message_attachment_join ON
-                    message_attachment_join.attachment_id = attachment.ROWID
-                LEFT JOIN message ON
-                    message.ROWID = message_attachment_join.message_id
-                LEFT JOIN handle ON handle.ROWID = message.handle_id;
+        cur.execute(
             """
-            )
-        except sqlite3.OperationalError as exc:
-            self.log.info(
-                "No SMS attachment tables found in the database, skipping: %s", exc
-            )
-            cur.close()
-            conn.close()
-            return
+            SELECT
+                attachment.ROWID as "attachment_id",
+                attachment.*,
+                message.service as "service",
+                handle.id as "phone_number"
+            FROM attachment
+            LEFT JOIN message_attachment_join ON
+                message_attachment_join.attachment_id = attachment.ROWID
+            LEFT JOIN message ON
+                message.ROWID = message_attachment_join.message_id
+            LEFT JOIN handle ON handle.ROWID = message.handle_id;
+        """
+        )
         names = [description[0] for description in cur.description]
 
         for item in cur:

tests/artifacts/generate_stix.py

@@ -82,7 +82,7 @@ def generate_test_stix_file(file_path):
     for h in sha256:
         i = Indicator(
             indicator_types=["malicious-activity"],
-            pattern="[file:hashes.'SHA-256'='{}']".format(h),
+            pattern="[file:hashes.sha256='{}']".format(h),
             pattern_type="stix",
         )
         res.append(i)
@@ -91,7 +91,7 @@ def generate_test_stix_file(file_path):
     for h in sha1:
         i = Indicator(
             indicator_types=["malicious-activity"],
-            pattern="[file:hashes.'SHA-1'='{}']".format(h),
+            pattern="[file:hashes.sha1='{}']".format(h),
             pattern_type="stix",
         )
         res.append(i)

tests/common/test_indicators.py

@@ -94,78 +94,6 @@ class TestIndicators:
         )
         assert ind.check_file_hash("da0611a300a9ce9aa7a09d1212f203fca5856794")
 
-    def test_parse_stix2_hash_key_variants(self, tmp_path):
-        """STIX2 spec requires single-quoted algorithm names that contain hyphens,
-        e.g. file:hashes.'SHA-256'. Verify MVT accepts both spec-compliant and
-        non-standard lowercase spellings for MD5, SHA-1 and SHA-256."""
-        import json
-
-        sha256_hash = "570cd76bf49cf52e0cb347a68bdcf0590b2eaece134e1b1eba7e8d66261bdbe6"
-        sha1_hash = "da0611a300a9ce9aa7a09d1212f203fca5856794"
-        md5_hash = "d41d8cd98f00b204e9800998ecf8427e"
-
-        variants = [
-            # (pattern_key, expected_bucket)
-            ("file:hashes.'SHA-256'", "files_sha256"),
-            ("file:hashes.SHA-256", "files_sha256"),
-            ("file:hashes.SHA256", "files_sha256"),
-            ("file:hashes.sha256", "files_sha256"),
-            ("file:hashes.'SHA-1'", "files_sha1"),
-            ("file:hashes.SHA-1", "files_sha1"),
-            ("file:hashes.SHA1", "files_sha1"),
-            ("file:hashes.sha1", "files_sha1"),
-            ("file:hashes.MD5", "files_md5"),
-            ("file:hashes.'MD5'", "files_md5"),
-            ("file:hashes.md5", "files_md5"),
-        ]
-
-        hash_for = {
-            "files_sha256": sha256_hash,
-            "files_sha1": sha1_hash,
-            "files_md5": md5_hash,
-        }
-
-        for pattern_key, bucket in variants:
-            h = hash_for[bucket]
-            stix = {
-                "type": "bundle",
-                "id": "bundle--test",
-                "objects": [
-                    {
-                        "type": "malware",
-                        "id": "malware--test",
-                        "name": "TestMalware",
-                        "is_family": False,
-                    },
-                    {
-                        "type": "indicator",
-                        "id": "indicator--test",
-                        "indicator_types": ["malicious-activity"],
-                        "pattern": f"[{pattern_key}='{h}']",
-                        "pattern_type": "stix",
-                        "valid_from": "2024-01-01T00:00:00Z",
-                    },
-                    {
-                        "type": "relationship",
-                        "id": "relationship--test",
-                        "relationship_type": "indicates",
-                        "source_ref": "indicator--test",
-                        "target_ref": "malware--test",
-                    },
-                ],
-            }
-            stix_file = tmp_path / "test.stix2"
-            stix_file.write_text(json.dumps(stix))
-
-            ind = Indicators(log=logging)
-            ind.load_indicators_files([str(stix_file)], load_default=False)
-            assert len(ind.ioc_collections[0][bucket]) == 1, (
-                f"Pattern key '{pattern_key}' was not parsed into '{bucket}'"
-            )
-            assert ind.check_file_hash(h) is not None, (
-                f"check_file_hash failed for pattern key '{pattern_key}'"
-            )
-
     def test_check_android_property(self, indicator_file):
         ind = Indicators(log=logging)
         ind.load_indicators_files([indicator_file], load_default=False)