Merge branch 'main' into fix/update-ios-backup-decryption

Makefile

@@ -31,3 +31,4 @@ clean:
 
 dist:
 	$(UV) build
+

docs/development/index.md

@@ -389,9 +389,3 @@ MVT_PROFILE=1 dev/mvt-ios check-backup test_backup
 ```
 
 Open an issue or PR if you are encountering significant performance issues when analyzing a device with MVT.
-
-## Output schemas
-
-Custom modules should declare an `output_model` using a Pydantic `RootModel`.
-This validates saved results and lets consumers obtain the module's JSON Schema.
-See [Output schemas](../output-schemas.md#custom-modules) for a complete example.

docs/output-schemas.md

@@ -1,77 +0,0 @@
-# Output schemas
-
-MVT validates its JSON output with versioned Pydantic models. The schema version
-used for a run is recorded as `output_schema_version` in `info.json`.
-
-MVT preserves the established on-disk format: most module files contain an array
-of result objects, while modules whose results are naturally grouped contain an
-object keyed by source or namespace. Detection files and `alerts.json` contain
-arrays of alert objects. Timestamps remain strings because their timezone and
-precision depend on the source artifact.
-
-## Exporting JSON Schema
-
-Both platform commands can print a versioned JSON Schema bundle:
-
-```bash
-mvt-ios schemas
-mvt-android schemas
-```
-
-Use `--output` to write one Draft 2020-12 JSON Schema file for each output:
-
-```bash
-mvt-ios schemas --output ./mvt-ios-schemas
-mvt-android schemas --output ./mvt-android-schemas
-```
-
-## Python API
-
-Models and schema discovery functions are available from `mvt.schemas`:
-
-```python
-from mvt.schemas import get_output_model
-
-SafariHistoryOutput = get_output_model("safari_history", platform="ios")
-validated = SafariHistoryOutput.model_validate(records)
-json_schema = SafariHistoryOutput.model_json_schema()
-```
-
-Common outputs have dedicated field-level models. Built-in module outputs have a
-declared root shape, and modules with an established dedicated record model expose
-its complete field schema.
-
-## Custom modules
-
-Custom modules can publish a precise contract by assigning a Pydantic root model
-to `output_model`:
-
-```python
-from pydantic import BaseModel, RootModel
-
-from mvt.common.module import MVTModule
-
-
-class ExampleRecord(BaseModel):
-    message: str
-    timestamp: str | None = None
-
-
-class ExampleOutput(RootModel[list[ExampleRecord]]):
-    pass
-
-
-class ExampleModule(MVTModule):
-    output_model = ExampleOutput
-```
-
-For compatibility, a custom module without `output_model` can still write an
-array or object containing JSON values. MVT logs a warning when it uses this
-generic contract. A future major release may require custom modules to declare
-their output models.
-
-## Compatibility policy
-
-Within one output schema major version, required fields are not removed and field
-types are not narrowed. New optional fields may be added. A breaking output
-change requires a new schema major version and a migration note.

mkdocs.yml

@@ -32,7 +32,6 @@ nav:
     - Introduction: "introduction.md"
     - Installation: "install.md"
     - Command Completion: "command_completion.md"
-    - Output Schemas: "output-schemas.md"
     - Using Docker: "docker.md"
     - MVT for iOS:
         - iOS Forensic Methodology: "ios/methodology.md"

pyproject.toml

@@ -24,7 +24,7 @@ dependencies = [
     "simplejson==4.1.1",
     "packaging==26.3",
     "appdirs==1.4.4",
-    "iphone_backup_decrypt==0.10.0",
+    "iphone_backup_decrypt==0.11.2",
     "pycryptodome>=3.20.0",
     "cryptography==50.0.0",
     "PyYAML>=6.0.2",

src/mvt/android/artifacts/tombstone_crashes.py

@@ -4,7 +4,7 @@
 #   https://license.mvt.re/1.1/
 
 import datetime
-from typing import ClassVar, List, Optional
+from typing import List, Optional
 
 import pydantic
 import betterproto2
@@ -13,7 +13,6 @@ from dateutil import parser
 from mvt.android.parsers.proto.tombstone import Tombstone
 from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
 from mvt.common.utils import convert_datetime_to_iso
-from mvt.schemas import OutputModel
 
 from .artifact import AndroidArtifact
 
@@ -74,10 +73,6 @@ class TombstoneCrashResult(pydantic.BaseModel):
     extra: Optional[str] = None
 
 
-class TombstoneCrashOutput(pydantic.RootModel[List[TombstoneCrashResult]]):
-    """Complete JSON document written by tombstone crash modules."""
-
-
 class TombstoneCrashArtifact(AndroidArtifact):
     """
     Parser for Android tombstone crash files.
@@ -85,8 +80,6 @@ class TombstoneCrashArtifact(AndroidArtifact):
     This parser can parse both text and protobuf tombstone crash files.
     """
 
-    output_model: ClassVar[OutputModel] = TombstoneCrashOutput
-
     def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
         return {
             "timestamp": record["timestamp"],

src/mvt/android/cli.py

@@ -98,13 +98,12 @@ def cli(ctx, disable_update_check, disable_indicator_update_check, verbose):
     ctx.obj["verbose"] = verbose
     set_verbose_logging(verbose)
 
-    if ctx.invoked_subcommand not in ("completion", "schemas"):
-        from mvt.common.logo import logo
+    from mvt.common.logo import logo
 
-        logo(
-            disable_version_check=disable_update_check,
-            disable_indicator_check=disable_indicator_update_check,
-        )
+    logo(
+        disable_version_check=disable_update_check,
+        disable_indicator_check=disable_indicator_update_check,
+    )
 
 
 # ==============================================================================
@@ -116,21 +115,6 @@ def version():
 
 
 # ==============================================================================
-# Command: schemas
-# ==============================================================================
-@cli.command("schemas", help="Print or export the JSON Schemas for MVT outputs.")
-@click.option(
-    "--output",
-    "-o",
-    type=click.Path(file_okay=False, dir_okay=True),
-    help="Write one JSON Schema file per output to this directory.",
-)
-def schemas(output):
-    from mvt.common.cmd_schemas import emit_schemas
-
-    emit_schemas("android", output)
-
-
 # Command: check-adb (removed)
 # ==============================================================================
 @cli.command(

src/mvt/android/modules/androidqf/aqf_settings.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 from mvt.android.artifacts.settings import Settings as SettingsArtifact
-from mvt.schemas import RecordListOutput
 
 from .base import AndroidQFModule
 
@@ -12,8 +11,6 @@ from .base import AndroidQFModule
 class AQFSettings(SettingsArtifact, AndroidQFModule):
     """This module analyse setting files"""
 
-    output_model = RecordListOutput
-
     def run(self) -> None:
         for setting_file in self._get_files_by_pattern("*/settings_*.txt"):
             namespace = setting_file[setting_file.rfind("_") + 1 : -4]

src/mvt/android/modules/androidqf/base.py

@@ -7,18 +7,15 @@ import fnmatch
 import logging
 import os
 import zipfile
-from typing import ClassVar, List, Optional
+from typing import List, Optional
 
 from mvt.common.module import MVTModule
 from mvt.common.module_types import ModuleResults
-from mvt.schemas import OutputModel, RecordListOutput
 
 
 class AndroidQFModule(MVTModule):
     """This class provides a base for all Android Data analysis modules."""
 
-    output_model: ClassVar[OutputModel] = RecordListOutput
-
     def __init__(
         self,
         file_path: Optional[str] = None,

src/mvt/android/modules/backup/base.py

@@ -7,17 +7,14 @@ import fnmatch
 import logging
 import os
 from tarfile import TarFile
-from typing import ClassVar, List, Optional
+from typing import List, Optional
 
 from mvt.common.module import ModuleResults, MVTModule
-from mvt.schemas import OutputModel, RecordListOutput
 
 
 class BackupModule(MVTModule):
     """This class provides a base for all backup extractios modules"""
 
-    output_model: ClassVar[OutputModel] = RecordListOutput
-
     def __init__(
         self,
         file_path: Optional[str] = None,

src/mvt/android/modules/bugreport/base.py

@@ -8,12 +8,11 @@ import logging
 import os
 import re
 from pathlib import Path
-from typing import ClassVar, List, Optional
+from typing import List, Optional
 from zipfile import ZipFile
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from mvt.common.module import ModuleResults, MVTModule
-from mvt.schemas import OutputModel, RecordListOutput
 
 # `------ 0.101s was the duration of 'SOME SECTION' ------`, printed when that
 # section finishes and not necessarily between two sections.
@@ -28,8 +27,6 @@ DUMPSTATE_ENTRY_POINTS = ("main_entry.txt", "dumpState_*.log", "*/dumpsys.txt")
 class BugReportModule(MVTModule):
     """This class provides a base for all Android Bug Report modules."""
 
-    output_model: ClassVar[OutputModel] = RecordListOutput
-
     def __init__(
         self,
         file_path: Optional[str] = None,

src/mvt/android/modules/bugreport/dumpsys_receivers.py

@@ -8,7 +8,6 @@ from typing import Optional
 
 from mvt.android.artifacts.dumpsys_receivers import DumpsysReceiversArtifact
 from mvt.common.module_types import ModuleResults
-from mvt.schemas import MappingOutput
 
 from .base import BugReportModule
 
@@ -16,8 +15,6 @@ from .base import BugReportModule
 class DumpsysReceivers(DumpsysReceiversArtifact, BugReportModule):
     """This module extracts details on receivers for risky activities."""
 
-    output_model = MappingOutput
-
     def __init__(
         self,
         file_path: Optional[str] = None,

src/mvt/android/modules/intrusion_logs/base.py

@@ -9,7 +9,7 @@ import json
 import logging
 import zipfile
 from pathlib import Path
-from typing import ClassVar, Optional, Union
+from typing import Optional, Union
 
 try:
     import zoneinfo
@@ -18,7 +18,6 @@ except ImportError:
 
 from mvt.common.module import MVTModule
 from mvt.common.utils import convert_datetime_to_iso, convert_unix_to_iso
-from mvt.schemas import OutputModel, RecordListOutput
 
 
 class IntrusionLogsModule(MVTModule):
@@ -42,8 +41,6 @@ class IntrusionLogsModule(MVTModule):
     original file-loading code path.
     """
 
-    output_model: ClassVar[OutputModel] = RecordListOutput
-
     def __init__(
         self,
         file_path: Optional[str] = None,

src/mvt/common/cmd_schemas.py

@@ -1,22 +0,0 @@
-# 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/
-
-"""Shared implementation for the platform schema-export commands."""
-
-import json
-from typing import Optional
-
-import click
-
-from mvt.schemas import export_json_schemas, schema_bundle
-
-
-def emit_schemas(platform: str, output: Optional[str]) -> None:
-    if output:
-        paths = export_json_schemas(output, platform)
-        click.echo(f"Exported {len(paths)} {platform} output schemas to {output}")
-        return
-
-    click.echo(json.dumps(schema_bundle(platform), indent=2))

src/mvt/common/command.py

@@ -34,13 +34,6 @@ from .utils import (
     get_sha256_from_file_path,
 )
 from .version import MVT_VERSION
-from mvt.schemas import (
-    OUTPUT_SCHEMA_VERSION,
-    AlertResults,
-    RunInfo,
-    URLResults,
-    validate_output,
-)
 
 
 class Command:
@@ -164,8 +157,6 @@ class Command:
         if not alerts:
             return
 
-        alerts = validate_output(AlertResults, alerts)
-
         alerts_path = os.path.join(self.results_path, "alerts.json")
         with open(alerts_path, "w+", encoding="utf-8") as handle:
             json.dump(alerts, handle, indent=4, cls=CustomJSONEncoder)
@@ -174,10 +165,9 @@ class Command:
         if not self.results_path or not self.url_results:
             return
 
-        urls = validate_output(URLResults, self.url_results)
         urls_path = os.path.join(self.results_path, "urls.json")
         with open(urls_path, "w", encoding="utf-8") as handle:
-            json.dump(urls, handle, indent=4, cls=CustomJSONEncoder)
+            json.dump(self.url_results, handle, indent=4, cls=CustomJSONEncoder)
 
     def _store_alerts_timeline(self) -> None:
         if not self.results_path:
@@ -200,7 +190,6 @@ class Command:
             "date": convert_datetime_to_iso(datetime.now()),
             "ioc_files": [],
             "hashes": [],
-            "output_schema_version": OUTPUT_SCHEMA_VERSION,
         }
 
         for coll in self.iocs.ioc_collections:
@@ -213,7 +202,6 @@ class Command:
 
         info["hashes"] = self.hash_values
 
-        info = validate_output(RunInfo, info)
         info_path = os.path.join(self.results_path, "info.json")
         with open(info_path, "w+", encoding="utf-8") as handle:
             json.dump(info, handle, indent=4)

src/mvt/common/module.py

@@ -9,12 +9,7 @@ import logging
 import os
 import re
 from dataclasses import asdict, is_dataclass
-from typing import Any, ClassVar, Dict, Optional, Sequence
-
-from pydantic import BaseModel, ValidationError
-
-from mvt.schemas.models import AlertResults, GenericModuleOutput, TimelineResults
-from mvt.schemas.serialization import validate_output
+from typing import Any, Dict, Optional, Sequence
 
 from .alerts import AlertStore
 from .indicators import Indicators
@@ -51,7 +46,6 @@ class MVTModule:
     slug: Optional[str] = None
     dependencies: Sequence[type["MVTModule"]] = ()
     supported_commands: Sequence[tuple[str, str]] = ()
-    output_model: ClassVar[type[BaseModel]] = GenericModuleOutput
     # A custom module can name a module class it supersedes, usually a
     # built-in one. When both are available to a command, the named module is
     # dropped from the run and this module takes its place, including in the
@@ -106,7 +100,7 @@ class MVTModule:
     @classmethod
     def from_json(cls, json_path: str, log: logging.Logger):
         with open(json_path, "r", encoding="utf-8") as handle:
-            results = validate_output(cls.output_model, json.load(handle))
+            results = json.load(handle)
             if log:
                 log.info('Loaded %d results from "%s"', len(results), json_path)
 
@@ -147,35 +141,38 @@ class MVTModule:
         name = self.get_slug()
 
         if self.results:
+            converted_results: Any
+            if isinstance(self.results, dict):
+                converted_results = self.results
+            else:
+                converted_results = [
+                    asdict(result)
+                    if is_dataclass(result) and not isinstance(result, type)
+                    else result
+                    for result in self.results
+                ]
             results_file_name = f"{name}.json"
             results_json_path = os.path.join(self.results_path, results_file_name)
-            if self.output_model is GenericModuleOutput:
-                self.log.warning(
-                    "Module %s does not declare an output_model; using the "
-                    "generic compatibility schema",
-                    self.__class__.__name__,
-                )
-            try:
-                converted_results = validate_output(self.output_model, self.results)
-            except (TypeError, ValueError, ValidationError) as exc:
-                self.log.error(
-                    "Output from module %s does not match schema %s: %s",
-                    self.__class__.__name__,
-                    self.output_model.__name__,
-                    exc,
-                )
-            else:
-                with open(results_json_path, "w", encoding="utf-8") as handle:
+            with open(results_json_path, "w", encoding="utf-8") as handle:
+                try:
                     json.dump(
                         converted_results, handle, indent=4, cls=CustomJSONEncoder
                     )
+                except Exception as exc:
+                    self.log.error(
+                        "Unable to store results of module %s to file %s: %s",
+                        self.__class__.__name__,
+                        results_file_name,
+                        exc,
+                    )
 
         if self.alertstore.alerts:
             detected_file_name = f"{name}_detected.json"
             detected_json_path = os.path.join(self.results_path, detected_file_name)
-            alerts = validate_output(AlertResults, self.alertstore.as_json())
             with open(detected_json_path, "w", encoding="utf-8") as handle:
-                json.dump(alerts, handle, indent=4, cls=CustomJSONEncoder)
+                json.dump(
+                    self.alertstore.as_json(), handle, indent=4, cls=CustomJSONEncoder
+                )
 
     def serialize(self, result: ModuleAtomicResult) -> ModuleSerializedResult:
         raise NotImplementedError
@@ -306,7 +303,6 @@ def save_timeline(timeline: list, timeline_path: str, is_utc: bool = True) -> No
     :param timeline_path: Path to the csv file to store the timeline to
 
     """
-    timeline = validate_output(TimelineResults, timeline)
     with open(timeline_path, "w", encoding="utf-8") as handle:
         csvoutput = csv.writer(
             handle, delimiter=",", quotechar='"', quoting=csv.QUOTE_ALL, escapechar="\\"

src/mvt/ios/cli.py

@@ -110,13 +110,12 @@ def cli(ctx, disable_update_check, disable_indicator_update_check, verbose):
     ctx.obj["verbose"] = verbose
     set_verbose_logging(verbose)
 
-    if ctx.invoked_subcommand not in ("completion", "schemas"):
-        from mvt.common.logo import logo
+    from mvt.common.logo import logo
 
-        logo(
-            disable_version_check=disable_update_check,
-            disable_indicator_check=disable_indicator_update_check,
-        )
+    logo(
+        disable_version_check=disable_update_check,
+        disable_indicator_check=disable_indicator_update_check,
+    )
 
 
 # ==============================================================================
@@ -128,21 +127,6 @@ def version():
 
 
 # ==============================================================================
-# Command: schemas
-# ==============================================================================
-@cli.command("schemas", help="Print or export the JSON Schemas for MVT outputs.")
-@click.option(
-    "--output",
-    "-o",
-    type=click.Path(file_okay=False, dir_okay=True),
-    help="Write one JSON Schema file per output to this directory.",
-)
-def schemas(output):
-    from mvt.common.cmd_schemas import emit_schemas
-
-    emit_schemas("ios", output)
-
-
 # Command: decrypt-backup
 # ==============================================================================
 @cli.command(

src/mvt/ios/decrypt.py

@@ -8,10 +8,8 @@ import glob
 import logging
 import os
 import os.path
-import plistlib
 import shutil
 import sqlite3
-import tempfile
 from concurrent.futures import (
     ALL_COMPLETED,
     FIRST_COMPLETED,
@@ -22,100 +20,26 @@ from concurrent.futures import (
 from pathlib import Path
 from typing import Optional
 
-from iphone_backup_decrypt import EncryptedBackup
-from iphone_backup_decrypt import google_iphone_dataprotection
+from iphone_backup_decrypt import (
+    BackupNotEncryptedError,
+    EncryptedBackup,
+    IncorrectPassphraseError,
+    NotABackupFolderError,
+)
 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.10.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
+    """Extract backup files by ID while MVT processes them concurrently."""
 
     def get_decryption_key(self):
         """Return derived key as hex string (64 chars / 32 bytes)."""
-        if self._derived_key is None:
+        if self.keybag is None or self.keybag.passphrase_key is None:
             raise ValueError("No derived key available")
-        return self._derived_key.hex()
+        return self.keybag.passphrase_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."""
@@ -129,8 +53,9 @@ class MVTEncryptedBackup(EncryptedBackup):
             shutil.copy2(source_filename, output_filename)
             return
 
-        inner_key = self._keybag.unwrapKeyForClass(
-            file_plist.protection_class, file_plist.encryption_key
+        inner_key = self.keybag.unwrap_key_for_class(
+            protection_class=file_plist.protection_class,
+            wrapped_file_key=file_plist.encryption_key,
         )
         self._decrypt_file_to_disk(
             file_id=file_id,
@@ -139,47 +64,6 @@ class MVTEncryptedBackup(EncryptedBackup):
             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
     using either a password or a key file.
@@ -376,41 +260,36 @@ class DecryptBackup:
                 )
                 return
 
-        # Before proceeding, we check whether the backup is indeed encrypted.
-        if not self.is_encrypted(self.backup_path):
-            return
-
         try:
             self._backup = MVTEncryptedBackup(
                 backup_directory=self.backup_path,
                 passphrase=password,
             )
+            # Manifest.db is a more reliable indication of encryption than
+            # the IsEncrypted field in Manifest.plist.
+            if not self.is_encrypted(self.backup_path):
+                self._backup = None
+                return
             # Eagerly trigger keybag unlock so wrong-password errors
             # surface here rather than later during process_backup().
             self._backup.test_decryption()
+        except IncorrectPassphraseError:
+            self._backup = None
+            log.critical("Failed to decrypt backup. Password is probably wrong.")
+        except (NotABackupFolderError, FileNotFoundError):
+            self._backup = None
+            log.critical(
+                "Failed to find a valid backup at %s. "
+                "Did you point to the right backup path?",
+                self.backup_path,
+            )
+        except BackupNotEncryptedError:
+            self._backup = None
+            log.critical("The backup does not seem encrypted!")
         except Exception as exc:
             self._backup = None
-            if (
-                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(
-                    "Failed to find a valid backup at %s. "
-                    "Did you point to the right backup path?",
-                    self.backup_path,
-                )
-            else:
-                log.exception(exc)
-                log.critical(
-                    "Failed to decrypt backup. Did you provide the correct password? "
-                    "Did you point to the right backup path?"
-                )
+            log.exception(exc)
+            log.critical("Failed to decrypt backup.")
 
     def decrypt_with_key_file(self, key_file: str) -> None:
         """Decrypts an encrypted iOS backup using a key file.
@@ -424,10 +303,6 @@ class DecryptBackup:
             key_file,
         )
 
-        # Before proceeding, we check whether the backup is indeed encrypted.
-        if not self.is_encrypted(self.backup_path):
-            return
-
         with open(key_file, "rb") as handle:
             key_bytes = handle.read()
 
@@ -442,16 +317,30 @@ class DecryptBackup:
             key_bytes_raw = binascii.unhexlify(key_bytes)
             self._backup = MVTEncryptedBackup(
                 backup_directory=self.backup_path,
-                derived_key=key_bytes_raw,
+                passphrase_key=key_bytes_raw,
             )
+            if not self.is_encrypted(self.backup_path):
+                self._backup = None
+                return
             # Eagerly trigger keybag unlock so wrong-key errors surface here.
             self._backup.test_decryption()
+        except (binascii.Error, IncorrectPassphraseError):
+            self._backup = None
+            log.critical("Failed to decrypt backup. Key file is probably wrong.")
+        except (NotABackupFolderError, FileNotFoundError):
+            self._backup = None
+            log.critical(
+                "Failed to find a valid backup at %s. "
+                "Did you point to the right backup path?",
+                self.backup_path,
+            )
+        except BackupNotEncryptedError:
+            self._backup = None
+            log.critical("The backup does not seem encrypted!")
         except Exception as exc:
             self._backup = None
             log.exception(exc)
-            log.critical(
-                "Failed to decrypt backup. Did you provide the correct key file?"
-            )
+            log.critical("Failed to decrypt backup.")
 
     def get_key(self) -> None:
         """Retrieve and prints the encryption key."""

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

@@ -10,7 +10,6 @@ from typing import Optional
 
 from mvt.common.module import DatabaseNotFoundError
 from mvt.common.module_types import ModuleResults
-from mvt.schemas import MappingOutput
 from mvt.ios.versions import get_device_desc_from_id, is_ios_version_outdated
 
 from ..base import IOSExtraction
@@ -19,8 +18,6 @@ from ..base import IOSExtraction
 class BackupInfo(IOSExtraction):
     """This module extracts information about the device and the backup."""
 
-    output_model = MappingOutput
-
     def __init__(
         self,
         file_path: Optional[str] = None,

src/mvt/ios/modules/base.py

@@ -11,7 +11,7 @@ import sqlite3
 import subprocess
 import tempfile
 from pathlib import Path
-from typing import ClassVar, Iterator, Optional, Union
+from typing import Iterator, Optional, Union
 
 from mvt.common.module import (
     DatabaseCorruptedError,
@@ -19,7 +19,6 @@ from mvt.common.module import (
     ModuleResults,
     MVTModule,
 )
-from mvt.schemas import OutputModel, RecordListOutput
 
 
 class TemporarySQLiteConnection(sqlite3.Connection):
@@ -40,8 +39,6 @@ class IOSExtraction(MVTModule):
     """This class provides a base for all iOS filesystem/backup extraction
     modules."""
 
-    output_model: ClassVar[OutputModel] = RecordListOutput
-
     def __init__(
         self,
         file_path: Optional[str] = None,

src/mvt/ios/modules/fs/cache_files.py

@@ -13,14 +13,11 @@ from mvt.common.module_types import (
     ModuleResults,
     ModuleSerializedResult,
 )
-from mvt.schemas import MappingOutput
 
 from ..base import IOSExtraction
 
 
 class CacheFiles(IOSExtraction):
-    output_model = MappingOutput
-
     def __init__(
         self,
         file_path: Optional[str] = None,

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

@@ -10,7 +10,6 @@ from typing import Optional
 
 from mvt.common.module_types import ModuleResults
 from mvt.common.utils import convert_datetime_to_iso
-from mvt.schemas import MappingOutput
 
 from ..base import IOSExtraction
 
@@ -33,8 +32,6 @@ class WebkitSessionResourceLog(IOSExtraction):
 
     """
 
-    output_model = MappingOutput
-
     def __init__(
         self,
         file_path: Optional[str] = None,

src/mvt/schemas/__init__.py

@@ -1,49 +0,0 @@
-"""Stable public API for MVT output models and JSON Schemas."""
-
-from .models import (
-    OUTPUT_SCHEMA_VERSION,
-    AlertResult,
-    AlertResults,
-    FileHash,
-    GenericModuleOutput,
-    MappingOutput,
-    ModuleRecord,
-    OutputModel,
-    RecordListOutput,
-    RunInfo,
-    TimelineEvent,
-    TimelineResults,
-    URLResult,
-    URLResults,
-)
-from .registry import (
-    export_json_schemas,
-    get_output_model,
-    get_output_models,
-    schema_bundle,
-)
-from .serialization import json_compatible, validate_output, write_output
-
-__all__ = [
-    "OUTPUT_SCHEMA_VERSION",
-    "AlertResult",
-    "AlertResults",
-    "FileHash",
-    "GenericModuleOutput",
-    "MappingOutput",
-    "ModuleRecord",
-    "OutputModel",
-    "RecordListOutput",
-    "RunInfo",
-    "TimelineEvent",
-    "TimelineResults",
-    "URLResult",
-    "URLResults",
-    "export_json_schemas",
-    "get_output_model",
-    "get_output_models",
-    "json_compatible",
-    "schema_bundle",
-    "validate_output",
-    "write_output",
-]

src/mvt/schemas/models.py

@@ -1,101 +0,0 @@
-# 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/
-
-"""Public Pydantic models for files written by MVT.
-
-Module records intentionally remain extensible: each extraction module has its own
-record fields, while the root models make the top-level output shape predictable.
-More specific modules can replace :class:`ModuleRecord` with a dedicated model.
-"""
-
-from typing import Literal, Optional
-
-from pydantic import BaseModel, ConfigDict, JsonValue, RootModel
-
-OUTPUT_SCHEMA_VERSION: Literal["1.0"] = "1.0"
-OutputModel = type[BaseModel]
-
-
-class MVTOutputModel(BaseModel):
-    """Base for stable output objects with no undocumented fields."""
-
-    model_config = ConfigDict(extra="forbid", strict=True)
-
-
-class ModuleRecord(BaseModel):
-    """A JSON object produced by an extraction module.
-
-    Module-specific keys are retained and represented in JSON Schema through
-    ``additionalProperties``. Dedicated module models can provide stronger field
-    contracts without changing the serialization machinery.
-    """
-
-    model_config = ConfigDict(extra="allow", strict=True)
-
-
-class RecordListOutput(RootModel[list[ModuleRecord]]):
-    """The standard output shape for built-in extraction modules."""
-
-
-class MappingOutput(RootModel[dict[str, JsonValue]]):
-    """Output shape for modules grouped by a dynamic string key."""
-
-
-class GenericModuleOutput(RootModel[list[JsonValue] | dict[str, JsonValue]]):
-    """Compatibility output shape for third-party modules without a model."""
-
-
-class FileHash(MVTOutputModel):
-    file_path: str
-    sha256: str
-
-
-class RunInfo(MVTOutputModel):
-    target_path: Optional[str]
-    mvt_version: str
-    date: str
-    ioc_files: list[str]
-    hashes: list[FileHash]
-    output_schema_version: Literal["1.0"]
-
-
-class URLResult(MVTOutputModel):
-    url: str
-    expanded_url: Optional[str]
-    timestamp: Optional[str]
-    source: str
-
-
-class URLResults(RootModel[list[URLResult]]):
-    pass
-
-
-AlertLevelName = Literal["INFORMATIONAL", "LOW", "MEDIUM", "HIGH", "CRITICAL"]
-
-
-class AlertResult(MVTOutputModel):
-    level: AlertLevelName
-    module: str
-    message: str
-    event_time: str
-    event: dict[str, JsonValue]
-    matched_indicator: Optional[JsonValue] = None
-
-
-class AlertResults(RootModel[list[AlertResult]]):
-    pass
-
-
-class TimelineEvent(MVTOutputModel):
-    """Public contract for the records used to create ``timeline.csv``."""
-
-    timestamp: Optional[str]
-    module: str
-    event: str
-    data: str
-
-
-class TimelineResults(RootModel[list[TimelineEvent]]):
-    pass

src/mvt/schemas/registry.py

@@ -1,133 +0,0 @@
-# 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/
-
-"""Discovery and JSON Schema export for MVT output models."""
-
-import json
-import os
-from pathlib import Path
-from typing import TYPE_CHECKING, Any, Iterable
-
-from pydantic import BaseModel
-
-from .models import (
-    OUTPUT_SCHEMA_VERSION,
-    AlertResults,
-    RunInfo,
-    TimelineResults,
-    URLResults,
-)
-
-if TYPE_CHECKING:
-    from mvt.common.module import MVTModule
-
-COMMON_OUTPUT_MODELS: dict[str, type[BaseModel]] = {
-    "alerts": AlertResults,
-    "info": RunInfo,
-    "timeline": TimelineResults,
-    "urls": URLResults,
-}
-
-JSON_SCHEMA_DIALECT = "https://json-schema.org/draft/2020-12/schema"
-
-
-def _module_classes(platform: str) -> Iterable[type["MVTModule"]]:
-    if platform == "ios":
-        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
-
-        return IOS_BACKUP_MODULES + FS_MODULES + MIXED_MODULES
-    if platform == "android":
-        from mvt.android.modules.androidqf import ANDROIDQF_MODULES
-        from mvt.android.modules.backup import BACKUP_MODULES as ANDROID_BACKUP_MODULES
-        from mvt.android.modules.bugreport import BUGREPORT_MODULES
-        from mvt.android.modules.intrusion_logs import INTRUSION_LOGS_MODULES
-
-        return (
-            ANDROID_BACKUP_MODULES
-            + BUGREPORT_MODULES
-            + ANDROIDQF_MODULES
-            + INTRUSION_LOGS_MODULES
-        )
-    raise ValueError(f"Unsupported MVT platform: {platform}")
-
-
-def get_output_models(platform: str) -> dict[str, type[BaseModel]]:
-    """Return common and module output models available for a platform."""
-
-    models = dict(COMMON_OUTPUT_MODELS)
-    for module in _module_classes(platform):
-        models[module.get_slug()] = module.output_model
-    return dict(sorted(models.items()))
-
-
-def get_output_model(name: str, platform: str | None = None) -> type[BaseModel]:
-    """Look up an output model by file stem/module slug.
-
-    ``platform`` is optional for common outputs. It is required when an output
-    slug exists on both platforms and resolves to different models.
-    """
-
-    normalized = name.removesuffix(".json")
-    if normalized.endswith("_detected"):
-        return AlertResults
-    if normalized in COMMON_OUTPUT_MODELS:
-        return COMMON_OUTPUT_MODELS[normalized]
-    if platform:
-        try:
-            return get_output_models(platform)[normalized]
-        except KeyError as exc:
-            raise KeyError(f"Unknown {platform} output schema: {name}") from exc
-
-    matches = {
-        models[normalized]
-        for candidate in ("ios", "android")
-        if normalized in (models := get_output_models(candidate))
-    }
-    if len(matches) == 1:
-        return matches.pop()
-    if len(matches) > 1:
-        raise KeyError(f"Output schema {name!r} is ambiguous; specify a platform")
-    raise KeyError(f"Unknown MVT output schema: {name}")
-
-
-def schema_bundle(platform: str) -> dict[str, Any]:
-    """Build a versioned bundle of Draft 2020-12 JSON Schemas."""
-
-    return {
-        "output_schema_version": OUTPUT_SCHEMA_VERSION,
-        "platform": platform,
-        "schemas": {
-            name: _output_json_schema(name, model, platform)
-            for name, model in get_output_models(platform).items()
-        },
-    }
-
-
-def _output_json_schema(
-    name: str, model: type[BaseModel], platform: str
-) -> dict[str, Any]:
-    schema = model.model_json_schema()
-    return {
-        "$schema": JSON_SCHEMA_DIALECT,
-        "$id": f"urn:mvt:output-schema:{OUTPUT_SCHEMA_VERSION}:{platform}:{name}",
-        **schema,
-    }
-
-
-def export_json_schemas(destination: str | os.PathLike[str], platform: str) -> list[Path]:
-    """Write one JSON Schema per output and return the paths created."""
-
-    destination_path = Path(destination)
-    destination_path.mkdir(parents=True, exist_ok=True)
-    written = []
-    for name, model in get_output_models(platform).items():
-        output_path = destination_path / f"{name}.schema.json"
-        with output_path.open("w", encoding="utf-8") as handle:
-            json.dump(_output_json_schema(name, model, platform), handle, indent=2)
-            handle.write("\n")
-        written.append(output_path)
-    return written

src/mvt/schemas/serialization.py

@@ -1,35 +0,0 @@
-# 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/
-
-"""Validation and serialization helpers shared by all MVT output writers."""
-
-import json
-from pathlib import Path
-from typing import Any
-
-from pydantic import BaseModel
-
-from mvt.common.utils import CustomJSONEncoder
-
-
-def json_compatible(value: Any) -> Any:
-    """Apply MVT's legacy conversions before validating JSON-native values."""
-
-    return json.loads(json.dumps(value, cls=CustomJSONEncoder))
-
-
-def validate_output(model: type[BaseModel], value: Any) -> Any:
-    """Validate an output document and return JSON-serializable Python values."""
-
-    compatible = json_compatible(value)
-    return model.model_validate(compatible).model_dump(mode="json")
-
-
-def write_output(path: str | Path, model: type[BaseModel], value: Any) -> None:
-    """Validate and write an MVT JSON document."""
-
-    validated = validate_output(model, value)
-    with open(path, "w", encoding="utf-8") as handle:
-        json.dump(validated, handle, indent=4)

tests/common/test_schemas.py

@@ -1,131 +0,0 @@
-# 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/
-
-import json
-import logging
-
-import pytest
-from click.testing import CliRunner
-from pydantic import BaseModel, ConfigDict, RootModel, ValidationError
-
-from mvt.android.cli import cli as android_cli
-from mvt.common.module import MVTModule
-from mvt.ios.cli import cli as ios_cli
-from mvt.schemas import (
-    OUTPUT_SCHEMA_VERSION,
-    GenericModuleOutput,
-    MappingOutput,
-    RecordListOutput,
-    export_json_schemas,
-    get_output_model,
-    get_output_models,
-    schema_bundle,
-    validate_output,
-)
-
-
-class CountRecord(BaseModel):
-    model_config = ConfigDict(strict=True)
-    count: int
-
-
-class CountOutput(RootModel[list[CountRecord]]):
-    pass
-
-
-class CountModule(MVTModule):
-    output_model = CountOutput
-
-
-def test_common_output_model_validates_without_changing_root_shape():
-    assert validate_output(RecordListOutput, [{"name": "example", "size": 2}]) == [
-        {"name": "example", "size": 2}
-    ]
-    assert validate_output(MappingOutput, {"global": {"enabled": "1"}}) == {
-        "global": {"enabled": "1"}
-    }
-
-
-def test_module_model_is_used_when_loading_existing_output(tmp_path):
-    path = tmp_path / "count.json"
-    path.write_text('[{"count": 4}]', encoding="utf-8")
-
-    module = CountModule.from_json(str(path), logging.getLogger(__name__))
-
-    assert module.results == [{"count": 4}]
-
-
-def test_invalid_module_output_is_not_written(tmp_path, caplog):
-    module = CountModule(results_path=str(tmp_path), results=[{"count": "four"}])
-
-    with caplog.at_level(logging.ERROR):
-        module.save_to_json()
-
-    assert not (tmp_path / "count_module.json").exists()
-    assert "does not match schema CountOutput" in caplog.text
-
-
-def test_invalid_existing_output_raises_validation_error(tmp_path):
-    path = tmp_path / "count.json"
-    path.write_text('[{"count": "four"}]', encoding="utf-8")
-
-    with pytest.raises(ValidationError):
-        CountModule.from_json(str(path), logging.getLogger(__name__))
-
-
-@pytest.mark.parametrize("platform", ["ios", "android"])
-def test_every_builtin_module_has_a_non_generic_output_model(platform):
-    module_models = {
-        name: model
-        for name, model in get_output_models(platform).items()
-        if name not in {"alerts", "info", "timeline", "urls"}
-    }
-
-    assert module_models
-    assert GenericModuleOutput not in module_models.values()
-
-
-def test_known_mapping_and_specific_models_are_registered():
-    assert get_output_model("backup_info.json", "ios") is MappingOutput
-    assert get_output_model("dumpsys_receivers", "android") is MappingOutput
-    assert get_output_model("tombstones", "android").__name__ == "TombstoneCrashOutput"
-    assert get_output_model("sms_detected.json", "ios").__name__ == "AlertResults"
-
-
-def test_schema_bundle_has_stable_versioned_shape():
-    bundle = schema_bundle("android")
-
-    assert bundle["output_schema_version"] == OUTPUT_SCHEMA_VERSION
-    assert bundle["platform"] == "android"
-    assert bundle["schemas"]["urls"]["$schema"].endswith("2020-12/schema")
-    assert bundle["schemas"]["urls"]["type"] == "array"
-    assert bundle["schemas"]["info"]["additionalProperties"] is False
-    assert bundle["schemas"]["dumpsys_receivers"]["type"] == "object"
-
-
-def test_export_json_schemas_writes_one_file_per_registered_output(tmp_path):
-    paths = export_json_schemas(tmp_path, "android")
-
-    assert len(paths) == len(get_output_models("android"))
-    assert json.loads((tmp_path / "info.schema.json").read_text())["title"] == (
-        "RunInfo"
-    )
-
-
-@pytest.mark.parametrize(
-    ("cli", "platform"), [(ios_cli, "ios"), (android_cli, "android")]
-)
-def test_schema_cli_prints_machine_readable_bundle(cli, platform):
-    result = CliRunner().invoke(cli, ["schemas"])
-
-    assert result.exit_code == 0
-    assert json.loads(result.output)["platform"] == platform
-
-
-def test_schema_cli_exports_schema_files(tmp_path):
-    result = CliRunner().invoke(android_cli, ["schemas", "--output", str(tmp_path)])
-
-    assert result.exit_code == 0
-    assert (tmp_path / "urls.schema.json").exists()

tests/ios_backup/test_decrypt.py

@@ -3,12 +3,15 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
+import hashlib
 import logging
+import plistlib
 import threading
 from pathlib import Path
 
 import pytest
 from Crypto.Cipher import AES
+from iphone_backup_decrypt import IncorrectPassphraseError
 
 from mvt.ios.decrypt import DecryptBackup, MVTEncryptedBackup
 
@@ -24,6 +27,89 @@ def _encrypted_file(backup_path, file_id, key, plaintext):
     source_path.write_bytes(encrypted)
 
 
+def _keybag_tlv(tag, value):
+    return tag + len(value).to_bytes(4, "big") + value
+
+
+def test_derived_key_can_unlock_keybag(mocker, tmp_path, caplog):
+    password = b"backup password"
+    salt = b"salt" * 4
+    dpsl = b"dpsl" * 4
+    round1 = hashlib.pbkdf2_hmac("sha256", password, dpsl, 1, 32)
+    derived_key = hashlib.pbkdf2_hmac("sha1", round1, salt, 1, 32)
+    wrapped_class_key = AES.new(derived_key, AES.MODE_KW).seal(b"k" * 32)
+    keybag_data = b"".join(
+        [
+            _keybag_tlv(b"TYPE", (1).to_bytes(4, "big")),
+            _keybag_tlv(b"DPSL", dpsl),
+            _keybag_tlv(b"DPIC", (1).to_bytes(4, "big")),
+            _keybag_tlv(b"SALT", salt),
+            _keybag_tlv(b"ITER", (1).to_bytes(4, "big")),
+            _keybag_tlv(b"UUID", b"g" * 16),
+            _keybag_tlv(b"WRAP", (2).to_bytes(4, "big")),
+            _keybag_tlv(b"UUID", b"c" * 16),
+            _keybag_tlv(b"CLAS", (1).to_bytes(4, "big")),
+            _keybag_tlv(b"WRAP", (2).to_bytes(4, "big")),
+            _keybag_tlv(b"WPKY", wrapped_class_key),
+        ]
+    )
+    (tmp_path / "Manifest.plist").write_bytes(
+        plistlib.dumps({"IsEncrypted": True, "BackupKeyBag": keybag_data})
+    )
+    (tmp_path / "Manifest.db").touch()
+
+    with pytest.raises(IncorrectPassphraseError):
+        MVTEncryptedBackup(
+            backup_directory=str(tmp_path), passphrase="wrong"
+        )._read_and_unlock_keybag()
+
+    password_backup = MVTEncryptedBackup(
+        backup_directory=str(tmp_path), passphrase=password
+    )
+    password_backup._read_and_unlock_keybag()
+    assert password_backup.get_decryption_key() == derived_key.hex()
+
+    key_backup = MVTEncryptedBackup(
+        backup_directory=str(tmp_path), passphrase_key=derived_key
+    )
+    key_backup._read_and_unlock_keybag()
+    assert key_backup.get_decryption_key() == derived_key.hex()
+
+    # Exercise MVT's password and key-file entry points with the native unlock.
+    mocker.patch.object(DecryptBackup, "is_encrypted", return_value=True)
+    mocker.patch.object(
+        MVTEncryptedBackup,
+        "test_decryption",
+        MVTEncryptedBackup._read_and_unlock_keybag,
+    )
+    password_decryptor = DecryptBackup(str(tmp_path))
+    password_decryptor.decrypt_with_password(password)
+    password_decryptor.get_key()
+    assert password_decryptor._decryption_key == derived_key.hex()
+
+    key_file = tmp_path / "key.txt"
+    key_file.write_text(derived_key.hex())
+    key_decryptor = DecryptBackup(str(tmp_path))
+    key_decryptor.decrypt_with_key_file(str(key_file))
+    assert key_decryptor.can_process()
+    assert key_decryptor._backup.get_decryption_key() == derived_key.hex()
+
+    with caplog.at_level(logging.CRITICAL, logger="mvt.ios.decrypt"):
+        wrong_password = DecryptBackup(str(tmp_path))
+        wrong_password.decrypt_with_password("wrong")
+    assert not wrong_password.can_process()
+    assert "Password is probably wrong" in caplog.text
+
+
+def test_invalid_backup_folder_uses_specific_error(tmp_path, caplog):
+    decryptor = DecryptBackup(str(tmp_path))
+    with caplog.at_level(logging.CRITICAL, logger="mvt.ios.decrypt"):
+        decryptor.decrypt_with_password("password")
+
+    assert not decryptor.can_process()
+    assert "Failed to find a valid backup" in caplog.text
+
+
 def test_extract_file_by_id_preserves_bytes_with_wrong_manifest_size(
     mocker, tmp_path
 ):
@@ -39,13 +125,15 @@ def test_extract_file_by_id_preserves_bytes_with_wrong_manifest_size(
         mtime=None,
     )
     mocker.patch("mvt.ios.decrypt.FilePlist", return_value=file_plist)
+    (tmp_path / "Manifest.plist").touch()
+    (tmp_path / "Manifest.db").touch()
 
     backup = MVTEncryptedBackup(
-        backup_directory=str(tmp_path), derived_key=b"d" * 32
+        backup_directory=str(tmp_path), passphrase_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
+    backup.keybag = mocker.Mock()
+    backup.keybag.unwrap_key_for_class.return_value = inner_key
     streaming_decrypt = mocker.spy(backup, "_decrypt_file_to_disk")
     output_path = tmp_path / "output"
 
@@ -57,6 +145,9 @@ def test_extract_file_by_id_preserves_bytes_with_wrong_manifest_size(
 
     assert output_path.read_bytes() == plaintext
     streaming_decrypt.assert_called_once()
+    backup.keybag.unwrap_key_for_class.assert_called_once_with(
+        protection_class=1, wrapped_file_key=b"wrapped-key"
+    )
 
 
 def test_extract_file_by_id_copies_unencrypted_files(mocker, tmp_path):
@@ -67,8 +158,10 @@ def test_extract_file_by_id_copies_unencrypted_files(mocker, tmp_path):
 
     file_plist = mocker.Mock(encryption_key=None)
     mocker.patch("mvt.ios.decrypt.FilePlist", return_value=file_plist)
+    (tmp_path / "Manifest.plist").touch()
+    (tmp_path / "Manifest.db").touch()
     backup = MVTEncryptedBackup(
-        backup_directory=str(tmp_path), derived_key=b"d" * 32
+        backup_directory=str(tmp_path), passphrase_key=b"d" * 32
     )
     mocker.patch.object(backup, "_read_and_unlock_keybag", return_value=True)
     output_path = tmp_path / "output"
@@ -82,6 +175,19 @@ def test_extract_file_by_id_copies_unencrypted_files(mocker, tmp_path):
     assert output_path.read_bytes() == b"plain content"
 
 
+def test_get_decryption_key_uses_unlocked_keybag(tmp_path):
+    (tmp_path / "Manifest.plist").touch()
+    (tmp_path / "Manifest.db").touch()
+    backup = MVTEncryptedBackup(
+        backup_directory=str(tmp_path), passphrase_key=b"d" * 32
+    )
+    with pytest.raises(ValueError, match="No derived key available"):
+        backup.get_decryption_key()
+
+    backup.keybag = type("Keybag", (), {"passphrase_key": b"d" * 32})()
+    assert backup.get_decryption_key() == "64" * 32
+
+
 @pytest.mark.parametrize("with_symlink", [False, True], ids=["file-id", "symlink"])
 def test_process_backup_rejects_unsafe_file_ids_and_destinations(
     mocker, tmp_path, with_symlink

uv.lock

@@ -609,14 +609,14 @@ wheels = [
 
 [[package]]
 name = "iphone-backup-decrypt"
-version = "0.10.0"
+version = "0.11.2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "pycryptodome" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/01/b8/a2ec2b10d792403d36bb6661fea6357179c31ece1ed3285c442e636d6335/iphone_backup_decrypt-0.10.0.tar.gz", hash = "sha256:5f2799e58f908cce28d7929dd82231abc8acd57c25b37bce6a2d6a82e018d343", size = 20000, upload-time = "2026-09-17T21:46:44Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/ee/f4/a2c3e8bfd937346d87753dd2e7e51efff14744ec268ebafe8e7be0269b9e/iphone_backup_decrypt-0.11.2.tar.gz", hash = "sha256:5f631927419e670e461bc0b9e299fae396445a3c84e0afe747a3464ccbb23ed1", size = 23923, upload-time = "2026-09-24T21:02:17.219Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/89/10/349b2a720e012921d1457e1d1c2b921b16300a5f3570c4c64d0d3787539f/iphone_backup_decrypt-0.10.0-py3-none-any.whl", hash = "sha256:1ac4345a07b104a3eacab20932b8621cd1e0164e0f03d913e66eed6ea5ee173e", size = 17400, upload-time = "2026-09-17T21:46:42Z" },
+    { url = "https://files.pythonhosted.org/packages/7b/33/72414c56bc507b0c104605e16fab2c86465b55567e1aaead8371a4c5884d/iphone_backup_decrypt-0.11.2-py3-none-any.whl", hash = "sha256:f18281997a94313c174aa3839d42db292d198e879fa570799c151ac36ac9c9d7", size = 21779, upload-time = "2026-09-24T21:02:15.878Z" },
 ]
 
 [[package]]
@@ -1038,7 +1038,7 @@ requires-dist = [
     { name = "betterproto2", specifier = "==0.10.0" },
     { name = "click", specifier = "==8.4.2" },
     { name = "cryptography", specifier = "==50.0.0" },
-    { name = "iphone-backup-decrypt", specifier = "==0.10.0" },
+    { name = "iphone-backup-decrypt", specifier = "==0.11.2" },
     { name = "nskeyedunarchiver", specifier = "==1.5.2" },
     { name = "packaging", specifier = "==26.3" },
     { name = "pyahocorasick", specifier = "==2.3.1" },