mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-30 01:36:26 +01:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ec0e30544f | ||
|
|
3efcfaf993 |
No files matched your search
@@ -62,27 +62,6 @@ AndroidQF will prompt the user to download, decrypt and collect device intrusion
|
||||
|
||||
For cases where intrusion logs were collected outside of an AndroidQF acquisition, the standalone `mvt-android check-intrusion-logs` command can analyse them directly. See [Check Android Intrusion Logs](intrusion_logs.md) for details, and the [feature announcment from Amnesty International's Security Lab](https://securitylab.amnesty.org/latest/2026/05/android-intrusion-logging-as-a-new-source-of-data-for-consensual-forensic-analysis/) for background on the data source.
|
||||
|
||||
## Browser history
|
||||
|
||||
AndroidQF can optionally collect browser `History` databases when a device
|
||||
already has working root access. When its acquisition contains the
|
||||
`browser_history/manifest.json` file, `mvt-android check-androidqf` parses every
|
||||
listed database, including any collected SQLite WAL and SHM sidecars. Browser
|
||||
visits are added to the module results, timeline, URL output, and indicator
|
||||
checks.
|
||||
|
||||
MVT can also inspect the supported database locations in a full Android
|
||||
filesystem dump, or inspect one explicitly supplied Chromium `History` file:
|
||||
|
||||
```bash
|
||||
mvt-android check-fs --output /path/to/results /path/to/filesystem-dump
|
||||
mvt-android check-fs --output /path/to/results /path/to/History
|
||||
```
|
||||
|
||||
The built-in locations cover Chrome, Brave, Microsoft Edge, and Samsung
|
||||
Internet. Other Chromium-compatible databases can be supplied explicitly as a
|
||||
file without assigning an unverified browser identity.
|
||||
|
||||
## Android Debug Bridge analysis removed
|
||||
|
||||
The ability to analyze Android devices directly over ADB has been removed from MVT. Direct extraction of data from ADB was error-prone and frequently resulted in inconsistent data collection between ADB and AndroidQF acquisitions. Use AndroidQF for device acquisition and `mvt-android check-androidqf` for analysis.
|
||||
|
||||
+1
-1
@@ -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",
|
||||
|
||||
@@ -1,162 +0,0 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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 sqlite3
|
||||
import shutil
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
|
||||
from mvt.common.utils import convert_chrometime_to_datetime, convert_datetime_to_iso
|
||||
|
||||
from .artifact import AndroidArtifact
|
||||
|
||||
|
||||
class BrowserHistoryArtifact(AndroidArtifact):
|
||||
"""Shared Chromium History database parsing and result handling."""
|
||||
|
||||
def _parse_browser_history(
|
||||
self,
|
||||
connection: sqlite3.Connection,
|
||||
*,
|
||||
browser: str,
|
||||
package: str,
|
||||
profile: str,
|
||||
source_path: str,
|
||||
) -> None:
|
||||
cursor = connection.cursor()
|
||||
try:
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT
|
||||
urls.id,
|
||||
urls.url,
|
||||
urls.title,
|
||||
urls.visit_count,
|
||||
urls.typed_count,
|
||||
visits.id,
|
||||
visits.visit_time,
|
||||
visits.from_visit,
|
||||
visits.transition
|
||||
FROM urls
|
||||
JOIN visits ON visits.url = urls.id
|
||||
ORDER BY visits.visit_time;
|
||||
"""
|
||||
)
|
||||
for row in cursor:
|
||||
timestamp = int(row[6])
|
||||
self.results.append(
|
||||
{
|
||||
"id": row[0],
|
||||
"url": row[1],
|
||||
"title": row[2],
|
||||
"visit_count": row[3],
|
||||
"typed_count": row[4],
|
||||
"visit_id": row[5],
|
||||
"timestamp": timestamp,
|
||||
"isodate": convert_datetime_to_iso(
|
||||
convert_chrometime_to_datetime(timestamp)
|
||||
),
|
||||
"redirect_source": row[7],
|
||||
"transition": row[8],
|
||||
"browser": browser,
|
||||
"package": package,
|
||||
"profile": profile,
|
||||
"source_path": source_path,
|
||||
}
|
||||
)
|
||||
finally:
|
||||
cursor.close()
|
||||
|
||||
def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
|
||||
return {
|
||||
"timestamp": record["isodate"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "browser_history",
|
||||
"data": (
|
||||
f"{record['browser']} visit to {record['url']} "
|
||||
f"(visit ID: {record['visit_id']}, profile: {record['profile']})"
|
||||
),
|
||||
}
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
if not self.indicators:
|
||||
return
|
||||
|
||||
for result, match in zip(
|
||||
self.results,
|
||||
self.indicators.check_url_batches(
|
||||
[[result["url"]] for result in self.results]
|
||||
),
|
||||
):
|
||||
if match:
|
||||
self.alertstore.critical(
|
||||
match.message, "", result, matched_indicator=match.ioc
|
||||
)
|
||||
|
||||
def collect_url_results(self) -> None:
|
||||
for result in self.results:
|
||||
self.add_url_result(result["url"], result.get("isodate"), "browser_history")
|
||||
|
||||
|
||||
class TemporarySQLiteConnection(sqlite3.Connection):
|
||||
temporary_directory: tempfile.TemporaryDirectory | None = None
|
||||
|
||||
def close(self) -> None:
|
||||
try:
|
||||
super().close()
|
||||
finally:
|
||||
if self.temporary_directory:
|
||||
self.temporary_directory.cleanup()
|
||||
self.temporary_directory = None
|
||||
|
||||
|
||||
def open_browser_history_database(database_path: Path) -> sqlite3.Connection:
|
||||
"""Open a staged History database without modifying forensic evidence."""
|
||||
database_uri = database_path.resolve().as_uri()
|
||||
if not Path(f"{database_path}-wal").is_file():
|
||||
return sqlite3.connect(f"{database_uri}?mode=ro&immutable=1", uri=True)
|
||||
|
||||
temporary_directory = tempfile.TemporaryDirectory(prefix="mvt_sqlite_")
|
||||
temporary_path = Path(temporary_directory.name) / database_path.name
|
||||
shutil.copy2(database_path, temporary_path)
|
||||
for suffix in ("-wal", "-shm"):
|
||||
sidecar = Path(f"{database_path}{suffix}")
|
||||
if sidecar.is_file():
|
||||
shutil.copy2(sidecar, Path(f"{temporary_path}{suffix}"))
|
||||
|
||||
try:
|
||||
connection = sqlite3.connect(
|
||||
f"{temporary_path.resolve().as_uri()}?mode=ro",
|
||||
uri=True,
|
||||
factory=TemporarySQLiteConnection,
|
||||
)
|
||||
except Exception:
|
||||
temporary_directory.cleanup()
|
||||
raise
|
||||
connection.temporary_directory = temporary_directory
|
||||
return connection
|
||||
|
||||
|
||||
def validate_manifest_database(database: Any) -> dict[str, Any]:
|
||||
if not isinstance(database, dict):
|
||||
raise ValueError("database entry is not an object")
|
||||
|
||||
required = ("browser", "package", "profile", "device_path", "archive_path")
|
||||
for field in required:
|
||||
if not isinstance(database.get(field), str) or not database[field]:
|
||||
raise ValueError(f"database entry has invalid {field}")
|
||||
|
||||
archive_path = database["archive_path"]
|
||||
path = Path(archive_path)
|
||||
if (
|
||||
"\\" in archive_path
|
||||
or path.is_absolute()
|
||||
or ".." in path.parts
|
||||
or path.parts[:1] != ("browser_history",)
|
||||
):
|
||||
raise ValueError(f"unsafe browser history archive path: {archive_path}")
|
||||
return database
|
||||
@@ -21,7 +21,6 @@ from mvt.common.help import (
|
||||
HELP_MSG_CHECK_ANDROID_BACKUP,
|
||||
HELP_MSG_CHECK_ANDROIDQF,
|
||||
HELP_MSG_CHECK_BUGREPORT,
|
||||
HELP_MSG_CHECK_FS,
|
||||
HELP_MSG_CHECK_IOCS,
|
||||
HELP_MSG_CHECK_INTRUSION_LOGS,
|
||||
HELP_MSG_DELAY_CHECKS,
|
||||
@@ -362,59 +361,6 @@ def check_androidqf(
|
||||
cmd.show_support_message()
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: check-fs
|
||||
# ==============================================================================
|
||||
@cli.command("check-fs", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_CHECK_FS)
|
||||
@click.option(
|
||||
"--iocs",
|
||||
"-i",
|
||||
type=click.Path(exists=True),
|
||||
multiple=True,
|
||||
default=[],
|
||||
help=HELP_MSG_IOC,
|
||||
)
|
||||
@click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
|
||||
@click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
|
||||
@click.option("--module", "-m", help=HELP_MSG_MODULE)
|
||||
@click.option(
|
||||
"--load-module",
|
||||
type=click.Path(exists=True),
|
||||
multiple=True,
|
||||
default=[],
|
||||
help=HELP_MSG_LOAD_MODULE,
|
||||
)
|
||||
@click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
|
||||
@click.argument("DUMP_PATH", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def check_fs(
|
||||
ctx, iocs, output, list_modules, module, load_module, hashes, verbose, dump_path
|
||||
):
|
||||
set_verbose_logging(verbose)
|
||||
custom_modules = _load_custom_modules(load_module)
|
||||
from .cmd_check_fs import CmdAndroidCheckFS
|
||||
|
||||
cmd = CmdAndroidCheckFS(
|
||||
target_path=dump_path,
|
||||
results_path=output,
|
||||
ioc_files=iocs,
|
||||
module_name=module,
|
||||
hashes=hashes,
|
||||
disable_version_check=_get_disable_flags(ctx)[0],
|
||||
disable_indicator_check=_get_disable_flags(ctx)[1],
|
||||
custom_modules=custom_modules,
|
||||
)
|
||||
if list_modules:
|
||||
cmd.list_modules()
|
||||
return
|
||||
|
||||
log.info("Checking Android filesystem located at: %s", dump_path)
|
||||
cmd.run()
|
||||
cmd.show_alerts_brief()
|
||||
cmd.show_support_message()
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: check-intrusion-logs
|
||||
# ==============================================================================
|
||||
|
||||
@@ -1,47 +0,0 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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 logging
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.command import Command
|
||||
from mvt.common.indicators import Indicators
|
||||
from mvt.common.module import MVTModule
|
||||
|
||||
from .modules.fs import FS_MODULES
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CmdAndroidCheckFS(Command):
|
||||
def __init__(
|
||||
self,
|
||||
target_path: Optional[str] = None,
|
||||
results_path: Optional[str] = None,
|
||||
ioc_files: Optional[list] = None,
|
||||
iocs: Optional[Indicators] = None,
|
||||
module_name: Optional[str] = None,
|
||||
module_options: Optional[dict] = None,
|
||||
hashes: bool = False,
|
||||
disable_version_check: bool = False,
|
||||
disable_indicator_check: bool = False,
|
||||
custom_modules: Optional[list[type[MVTModule]]] = None,
|
||||
) -> None:
|
||||
super().__init__(
|
||||
target_path=target_path,
|
||||
results_path=results_path,
|
||||
ioc_files=ioc_files,
|
||||
iocs=iocs,
|
||||
module_name=module_name,
|
||||
module_options=module_options,
|
||||
hashes=hashes,
|
||||
log=log,
|
||||
disable_version_check=disable_version_check,
|
||||
disable_indicator_check=disable_indicator_check,
|
||||
custom_modules=custom_modules,
|
||||
)
|
||||
self.platform = "android"
|
||||
self.name = "check-fs"
|
||||
self.modules = FS_MODULES
|
||||
@@ -11,7 +11,6 @@ from .aqf_log_timestamps import AQFLogTimestamps
|
||||
from .aqf_packages import AQFPackages
|
||||
from .aqf_processes import AQFProcesses
|
||||
from .aqf_settings import AQFSettings
|
||||
from .browser_history import BrowserHistory
|
||||
from .mounts import Mounts
|
||||
from .root_binaries import RootBinaries
|
||||
|
||||
@@ -24,5 +23,4 @@ ANDROIDQF_MODULES: list[type[MVTModule]] = [
|
||||
AQFLogTimestamps,
|
||||
RootBinaries,
|
||||
Mounts,
|
||||
BrowserHistory,
|
||||
]
|
||||
@@ -1,115 +0,0 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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 sqlite3
|
||||
import tempfile
|
||||
from pathlib import Path, PurePosixPath
|
||||
|
||||
from mvt.android.artifacts.browser_history import (
|
||||
BrowserHistoryArtifact,
|
||||
open_browser_history_database,
|
||||
validate_manifest_database,
|
||||
)
|
||||
|
||||
from .base import AndroidQFModule
|
||||
|
||||
|
||||
class BrowserHistory(BrowserHistoryArtifact, AndroidQFModule):
|
||||
"""Extract browser visits collected by AndroidQF."""
|
||||
|
||||
supported_commands = (("android", "check-androidqf"),)
|
||||
|
||||
def _find_manifest(self) -> str | None:
|
||||
manifests = [
|
||||
file_path
|
||||
for file_path in self.files
|
||||
if file_path.replace("\\", "/").endswith("browser_history/manifest.json")
|
||||
]
|
||||
if not manifests:
|
||||
return None
|
||||
if len(manifests) > 1:
|
||||
self.log.warning(
|
||||
"Found multiple browser history manifests; using %s", manifests[0]
|
||||
)
|
||||
return manifests[0]
|
||||
|
||||
def _stage_database(
|
||||
self, archive_path: str, prefix: str, temporary_path: Path
|
||||
) -> Path:
|
||||
available_files = {
|
||||
str(PurePosixPath(file_path.replace("\\", "/"))): file_path
|
||||
for file_path in self.files
|
||||
}
|
||||
normalized_path = str(PurePosixPath(prefix, archive_path))
|
||||
source_path = available_files.get(normalized_path)
|
||||
if not source_path:
|
||||
raise FileNotFoundError(archive_path)
|
||||
|
||||
staged_path = temporary_path / "History"
|
||||
staged_path.write_bytes(self._get_file_content(source_path))
|
||||
for suffix in ("-wal", "-shm"):
|
||||
sidecar = available_files.get(normalized_path + suffix)
|
||||
if sidecar:
|
||||
Path(f"{staged_path}{suffix}").write_bytes(
|
||||
self._get_file_content(sidecar)
|
||||
)
|
||||
return staged_path
|
||||
|
||||
def run(self) -> None:
|
||||
manifest_path = self._find_manifest()
|
||||
if not manifest_path:
|
||||
self.log.info("No AndroidQF browser history manifest found")
|
||||
return
|
||||
|
||||
try:
|
||||
manifest = json.loads(self._get_file_content(manifest_path))
|
||||
except (json.JSONDecodeError, OSError, TypeError, UnicodeDecodeError) as exc:
|
||||
self.log.error("Unable to read browser history manifest: %s", exc)
|
||||
return
|
||||
|
||||
if not isinstance(manifest, dict) or manifest.get("schema_version") != 1:
|
||||
self.log.error("Unsupported AndroidQF browser history manifest")
|
||||
return
|
||||
databases = manifest.get("databases", [])
|
||||
if not isinstance(databases, list):
|
||||
self.log.error("Invalid AndroidQF browser history database list")
|
||||
return
|
||||
|
||||
normalized_manifest = manifest_path.replace("\\", "/")
|
||||
marker = "browser_history/manifest.json"
|
||||
prefix = normalized_manifest[: -len(marker)].rstrip("/")
|
||||
|
||||
for raw_database in databases:
|
||||
try:
|
||||
database = validate_manifest_database(raw_database)
|
||||
with tempfile.TemporaryDirectory(prefix="mvt_browser_history_") as temp:
|
||||
staged_path = self._stage_database(
|
||||
database["archive_path"], prefix, Path(temp)
|
||||
)
|
||||
connection = open_browser_history_database(staged_path)
|
||||
try:
|
||||
self._parse_browser_history(
|
||||
connection,
|
||||
browser=database["browser"],
|
||||
package=database["package"],
|
||||
profile=database["profile"],
|
||||
source_path=database["device_path"],
|
||||
)
|
||||
finally:
|
||||
connection.close()
|
||||
except (
|
||||
FileNotFoundError,
|
||||
OSError,
|
||||
OverflowError,
|
||||
sqlite3.Error,
|
||||
TypeError,
|
||||
ValueError,
|
||||
) as exc:
|
||||
self.log.error("Unable to parse browser history database: %s", exc)
|
||||
|
||||
self.log.info(
|
||||
"Extracted a total of %d browser history items", len(self.results)
|
||||
)
|
||||
@@ -1,10 +0,0 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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/
|
||||
|
||||
from mvt.common.module import MVTModule
|
||||
|
||||
from .browser_history import BrowserHistory
|
||||
|
||||
FS_MODULES: list[type[MVTModule]] = [BrowserHistory]
|
||||
@@ -1,115 +0,0 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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 logging
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from mvt.android.artifacts.browser_history import (
|
||||
BrowserHistoryArtifact,
|
||||
open_browser_history_database,
|
||||
)
|
||||
from mvt.common.module import MVTModule
|
||||
from mvt.common.module_types import ModuleResults
|
||||
|
||||
|
||||
# These locations are deliberately limited to paths backed by public parser
|
||||
# fixtures or the historical MVT implementation.
|
||||
BROWSER_HISTORY_PATHS = {
|
||||
"data/data/com.android.chrome/app_chrome/Default/History": (
|
||||
"Chrome",
|
||||
"com.android.chrome",
|
||||
"Default",
|
||||
),
|
||||
"data/data/com.brave.browser/app_chrome/Default/History": (
|
||||
"Brave",
|
||||
"com.brave.browser",
|
||||
"Default",
|
||||
),
|
||||
"data/data/com.microsoft.emmx/app_chrome/Default/History": (
|
||||
"Microsoft Edge",
|
||||
"com.microsoft.emmx",
|
||||
"Default",
|
||||
),
|
||||
"data/data/com.sec.android.app.sbrowser/app_sbrowser/Default/History": (
|
||||
"Samsung Internet",
|
||||
"com.sec.android.app.sbrowser",
|
||||
"Default",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
class BrowserHistory(BrowserHistoryArtifact, MVTModule):
|
||||
"""Extract supported Chromium History databases from a filesystem dump."""
|
||||
|
||||
supported_commands = (("android", "check-fs"),)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
file_path: Optional[str] = None,
|
||||
target_path: Optional[str] = None,
|
||||
results_path: Optional[str] = None,
|
||||
module_options: Optional[dict] = None,
|
||||
log: logging.Logger = logging.getLogger(__name__),
|
||||
results: Optional[ModuleResults] = None,
|
||||
) -> None:
|
||||
super().__init__(
|
||||
file_path=file_path,
|
||||
target_path=target_path,
|
||||
results_path=results_path,
|
||||
module_options=module_options,
|
||||
log=log,
|
||||
results=results,
|
||||
)
|
||||
|
||||
def _database_paths(self) -> list[tuple[Path, str, str, str]]:
|
||||
if not self.target_path:
|
||||
return []
|
||||
target = Path(self.target_path)
|
||||
if target.is_file():
|
||||
normalized_target = target.resolve().as_posix()
|
||||
for relative_path, identity in BROWSER_HISTORY_PATHS.items():
|
||||
if normalized_target.endswith(f"/{relative_path}"):
|
||||
return [(target, *identity)]
|
||||
return [(target, "Chromium", "unknown", "unknown")]
|
||||
|
||||
databases = []
|
||||
for relative_path, identity in BROWSER_HISTORY_PATHS.items():
|
||||
database_path = target / relative_path
|
||||
if database_path.is_file():
|
||||
databases.append((database_path, *identity))
|
||||
return databases
|
||||
|
||||
def run(self) -> None:
|
||||
for database_path, browser, package, profile in self._database_paths():
|
||||
try:
|
||||
connection = open_browser_history_database(database_path)
|
||||
try:
|
||||
self._parse_browser_history(
|
||||
connection,
|
||||
browser=browser,
|
||||
package=package,
|
||||
profile=profile,
|
||||
source_path=str(database_path),
|
||||
)
|
||||
finally:
|
||||
connection.close()
|
||||
except (
|
||||
OSError,
|
||||
OverflowError,
|
||||
sqlite3.Error,
|
||||
TypeError,
|
||||
ValueError,
|
||||
) as exc:
|
||||
self.log.error(
|
||||
"Unable to parse browser history database %s: %s",
|
||||
database_path,
|
||||
exc,
|
||||
)
|
||||
|
||||
self.log.info(
|
||||
"Extracted a total of %d browser history items", len(self.results)
|
||||
)
|
||||
+50
-161
@@ -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."""
|
||||
|
||||
@@ -1,261 +0,0 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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 sqlite3
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from mvt.android.modules.androidqf.browser_history import (
|
||||
BrowserHistory as AndroidQFBrowserHistory,
|
||||
)
|
||||
from mvt.android.modules.fs.browser_history import BrowserHistory as FSBrowserHistory
|
||||
from mvt.android.cmd_check_fs import CmdAndroidCheckFS
|
||||
|
||||
CHROME_TIME = 13_348_540_800_000_000
|
||||
URL = "https://example.org/path"
|
||||
|
||||
|
||||
def create_history_database(path: Path, *, url: str = URL) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with sqlite3.connect(path) as connection:
|
||||
connection.executescript(
|
||||
"""
|
||||
CREATE TABLE urls (
|
||||
id INTEGER PRIMARY KEY,
|
||||
url TEXT,
|
||||
title TEXT,
|
||||
visit_count INTEGER,
|
||||
typed_count INTEGER
|
||||
);
|
||||
CREATE TABLE visits (
|
||||
id INTEGER PRIMARY KEY,
|
||||
url INTEGER,
|
||||
visit_time INTEGER,
|
||||
from_visit INTEGER,
|
||||
transition INTEGER
|
||||
);
|
||||
"""
|
||||
)
|
||||
connection.execute("INSERT INTO urls VALUES (1, ?, 'Example', 1, 0)", (url,))
|
||||
connection.execute(
|
||||
"INSERT INTO visits VALUES (7, 1, ?, 0, 805306368)",
|
||||
(CHROME_TIME,),
|
||||
)
|
||||
|
||||
|
||||
def manifest(database_paths: list[tuple[str, str]]) -> dict:
|
||||
return {
|
||||
"schema_version": 1,
|
||||
"status": "collected",
|
||||
"databases": [
|
||||
{
|
||||
"browser": "Chrome",
|
||||
"package": package,
|
||||
"profile": "Default",
|
||||
"device_path": f"/data/data/{package}/app_chrome/Default/History",
|
||||
"archive_path": archive_path,
|
||||
"sidecars": [],
|
||||
}
|
||||
for package, archive_path in database_paths
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_zip", [False, True])
|
||||
def test_androidqf_browser_history_directory_and_zip(tmp_path, use_zip):
|
||||
source_database = tmp_path / "source" / "History"
|
||||
create_history_database(source_database)
|
||||
archive_path = "browser_history/com.android.chrome/Default/History"
|
||||
manifest_data = manifest([("com.android.chrome", archive_path)])
|
||||
|
||||
module = AndroidQFBrowserHistory()
|
||||
if use_zip:
|
||||
acquisition_path = tmp_path / "acquisition.zip"
|
||||
with zipfile.ZipFile(acquisition_path, "w") as archive:
|
||||
archive.write(source_database, archive_path)
|
||||
archive.writestr("browser_history/manifest.json", json.dumps(manifest_data))
|
||||
with zipfile.ZipFile(acquisition_path) as archive:
|
||||
module.from_zip(archive, archive.namelist())
|
||||
module.run()
|
||||
else:
|
||||
acquisition = tmp_path / "acquisition"
|
||||
database_path = acquisition / archive_path
|
||||
database_path.parent.mkdir(parents=True)
|
||||
database_path.write_bytes(source_database.read_bytes())
|
||||
manifest_path = acquisition / "browser_history" / "manifest.json"
|
||||
manifest_path.write_text(json.dumps(manifest_data))
|
||||
files = [
|
||||
path.relative_to(tmp_path).as_posix()
|
||||
for path in acquisition.rglob("*")
|
||||
if path.is_file()
|
||||
]
|
||||
module.from_dir(str(tmp_path), files)
|
||||
module.run()
|
||||
|
||||
assert len(module.results) == 1
|
||||
assert module.results[0]["url"] == URL
|
||||
assert module.results[0]["browser"] == "Chrome"
|
||||
assert module.results[0]["source_path"].endswith("/History")
|
||||
module.collect_url_results()
|
||||
module.to_timeline()
|
||||
assert module.url_results[0]["url"] == URL
|
||||
assert module.timeline[0]["event"] == "browser_history"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("prefix", ["", "./", "acquisition/", "./acquisition/"])
|
||||
@pytest.mark.parametrize("separator", ["/", "\\"])
|
||||
def test_androidqf_zip_paths_preserve_wal_visits(tmp_path, prefix, separator):
|
||||
database = tmp_path / "History"
|
||||
create_history_database(database)
|
||||
connection = sqlite3.connect(database)
|
||||
connection.execute("PRAGMA journal_mode=WAL")
|
||||
connection.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
||||
connection.execute(
|
||||
"INSERT INTO urls VALUES (2, ?, 'WAL', 1, 0)", (URL + "/wal",)
|
||||
)
|
||||
connection.execute(
|
||||
"INSERT INTO visits VALUES (8, 2, ?, 0, 805306368)", (CHROME_TIME,)
|
||||
)
|
||||
connection.commit()
|
||||
archive_path = "browser_history/com.android.chrome/Default/History"
|
||||
acquisition = tmp_path / "acquisition.zip"
|
||||
try:
|
||||
with zipfile.ZipFile(acquisition, "w") as archive:
|
||||
for suffix in ("", "-wal", "-shm"):
|
||||
archive.writestr(
|
||||
(prefix + archive_path + suffix).replace("/", separator),
|
||||
Path(f"{database}{suffix}").read_bytes(),
|
||||
)
|
||||
archive.writestr(
|
||||
(prefix + "browser_history/manifest.json").replace("/", separator),
|
||||
json.dumps(manifest([("com.android.chrome", archive_path)])),
|
||||
)
|
||||
module = AndroidQFBrowserHistory()
|
||||
with zipfile.ZipFile(acquisition) as archive:
|
||||
module.from_zip(archive, archive.namelist())
|
||||
module.run()
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
assert {result["url"] for result in module.results} == {URL, URL + "/wal"}
|
||||
|
||||
|
||||
def test_androidqf_browser_history_isolates_malformed_database(tmp_path, caplog):
|
||||
acquisition = tmp_path / "acquisition"
|
||||
good_archive_path = "browser_history/com.android.chrome/Default/History"
|
||||
bad_archive_path = "browser_history/com.brave.browser/Default/History"
|
||||
create_history_database(acquisition / good_archive_path)
|
||||
bad_path = acquisition / bad_archive_path
|
||||
bad_path.parent.mkdir(parents=True)
|
||||
bad_path.write_bytes(b"not sqlite")
|
||||
manifest_path = acquisition / "browser_history" / "manifest.json"
|
||||
manifest_path.write_text(
|
||||
json.dumps(
|
||||
manifest(
|
||||
[
|
||||
("com.brave.browser", bad_archive_path),
|
||||
("com.android.chrome", good_archive_path),
|
||||
]
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
module = AndroidQFBrowserHistory()
|
||||
files = [
|
||||
path.relative_to(tmp_path).as_posix()
|
||||
for path in acquisition.rglob("*")
|
||||
if path.is_file()
|
||||
]
|
||||
module.from_dir(str(tmp_path), files)
|
||||
module.run()
|
||||
|
||||
assert [result["url"] for result in module.results] == [URL]
|
||||
assert "Unable to parse browser history database" in caplog.text
|
||||
|
||||
|
||||
def test_androidqf_browser_history_rejects_unsafe_manifest_path(tmp_path, caplog):
|
||||
acquisition = tmp_path / "acquisition"
|
||||
manifest_path = acquisition / "browser_history" / "manifest.json"
|
||||
manifest_path.parent.mkdir(parents=True)
|
||||
manifest_path.write_text(
|
||||
json.dumps(manifest([("com.android.chrome", "browser_history/../secret")]))
|
||||
)
|
||||
module = AndroidQFBrowserHistory()
|
||||
module.from_dir(str(tmp_path), [manifest_path.relative_to(tmp_path).as_posix()])
|
||||
|
||||
module.run()
|
||||
|
||||
assert module.results == []
|
||||
assert "unsafe browser history archive path" in caplog.text
|
||||
|
||||
|
||||
def test_filesystem_browser_history_reads_wal_only_visit(tmp_path):
|
||||
database_path = tmp_path / "History"
|
||||
connection = sqlite3.connect(database_path)
|
||||
connection.execute("PRAGMA journal_mode=WAL")
|
||||
connection.executescript(
|
||||
"""
|
||||
CREATE TABLE urls (
|
||||
id INTEGER PRIMARY KEY,
|
||||
url TEXT,
|
||||
title TEXT,
|
||||
visit_count INTEGER,
|
||||
typed_count INTEGER
|
||||
);
|
||||
CREATE TABLE visits (
|
||||
id INTEGER PRIMARY KEY,
|
||||
url INTEGER,
|
||||
visit_time INTEGER,
|
||||
from_visit INTEGER,
|
||||
transition INTEGER
|
||||
);
|
||||
"""
|
||||
)
|
||||
connection.commit()
|
||||
connection.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
||||
connection.execute("INSERT INTO urls VALUES (1, ?, 'WAL', 1, 0)", (URL,))
|
||||
connection.execute(
|
||||
"INSERT INTO visits VALUES (8, 1, ?, 0, 805306368)", (CHROME_TIME,)
|
||||
)
|
||||
connection.commit()
|
||||
try:
|
||||
assert Path(f"{database_path}-wal").stat().st_size > 0
|
||||
module = FSBrowserHistory(target_path=str(database_path))
|
||||
module.run()
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
assert [result["url"] for result in module.results] == [URL]
|
||||
assert module.results[0]["browser"] == "Chromium"
|
||||
|
||||
|
||||
def test_android_check_fs_finds_all_supported_browser_paths(tmp_path):
|
||||
for index, relative_path in enumerate(
|
||||
(
|
||||
"data/data/com.android.chrome/app_chrome/Default/History",
|
||||
"data/data/com.brave.browser/app_chrome/Default/History",
|
||||
"data/data/com.microsoft.emmx/app_chrome/Default/History",
|
||||
"data/data/com.sec.android.app.sbrowser/app_sbrowser/Default/History",
|
||||
)
|
||||
):
|
||||
create_history_database(
|
||||
tmp_path / relative_path, url=f"https://example.org/{index}"
|
||||
)
|
||||
|
||||
command = CmdAndroidCheckFS(target_path=str(tmp_path), module_name="BrowserHistory")
|
||||
command.run()
|
||||
|
||||
assert len(command.executed) == 1
|
||||
assert {result["package"] for result in command.executed[0].results} == {
|
||||
"com.android.chrome",
|
||||
"com.brave.browser",
|
||||
"com.microsoft.emmx",
|
||||
"com.sec.android.app.sbrowser",
|
||||
}
|
||||
assert len(command.url_results) == 4
|
||||
assert len(command.timeline) == 4
|
||||
@@ -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
|
||||
|
||||
@@ -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" },
|
||||
|
||||
Reference in new issue
Block a user