Compare commits

..
14 changed files with 957 additions and 166 deletions

No files matched your search

-1
View File
@@ -31,4 +31,3 @@ clean:
dist:
$(UV) build
+21
View File
@@ -62,6 +62,27 @@ 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
View File
@@ -24,7 +24,7 @@ dependencies = [
"simplejson==4.1.1",
"packaging==26.3",
"appdirs==1.4.4",
"iphone_backup_decrypt==0.11.2",
"iphone_backup_decrypt==0.10.0",
"pycryptodome>=3.20.0",
"cryptography==50.0.0",
"PyYAML>=6.0.2",
@@ -0,0 +1,162 @@
# 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
+54
View File
@@ -21,6 +21,7 @@ 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,
@@ -361,6 +362,59 @@ 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
# ==============================================================================
+47
View File
@@ -0,0 +1,47 @@
# 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,6 +11,7 @@ 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
@@ -23,4 +24,5 @@ ANDROIDQF_MODULES: list[type[MVTModule]] = [
AQFLogTimestamps,
RootBinaries,
Mounts,
BrowserHistory,
]
@@ -0,0 +1,115 @@
# 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)
)
+10
View File
@@ -0,0 +1,10 @@
# 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]
@@ -0,0 +1,115 @@
# 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)
)
+161 -50
View File
@@ -8,8 +8,10 @@ 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,
@@ -20,26 +22,100 @@ from concurrent.futures import (
from pathlib import Path
from typing import Optional
from iphone_backup_decrypt import (
BackupNotEncryptedError,
EncryptedBackup,
IncorrectPassphraseError,
NotABackupFolderError,
)
from iphone_backup_decrypt import EncryptedBackup
from iphone_backup_decrypt import google_iphone_dataprotection
from iphone_backup_decrypt.utils import FilePlist
from .decrypt_config import DEFAULT_DECRYPT_WORKERS, MAX_DECRYPT_WORKERS
log = logging.getLogger(__name__)
# Import pbkdf2_hmac from the same source iphone_backup_decrypt uses internally,
# so our key derivation is consistent with theirs.
try:
from fastpbkdf2 import pbkdf2_hmac
except ImportError:
import Crypto.Hash.SHA1
import Crypto.Hash.SHA256
import Crypto.Protocol.KDF
_HASH_FNS = {"sha1": Crypto.Hash.SHA1, "sha256": Crypto.Hash.SHA256}
def pbkdf2_hmac(hash_name, password, salt, iterations, dklen=None):
return Crypto.Protocol.KDF.PBKDF2(
password, salt, dklen, iterations, hmac_hash_module=_HASH_FNS[hash_name]
)
class MVTEncryptedBackup(EncryptedBackup):
"""Extract backup files by ID while MVT processes them concurrently."""
"""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
def get_decryption_key(self):
"""Return derived key as hex string (64 chars / 32 bytes)."""
if self.keybag is None or self.keybag.passphrase_key is None:
if self._derived_key is None:
raise ValueError("No derived key available")
return self.keybag.passphrase_key.hex()
return self._derived_key.hex()
def extract_file_by_id(self, *, file_id, file_bplist, output_filename):
"""Extract one manifest entry without loading the whole file into memory."""
@@ -53,9 +129,8 @@ class MVTEncryptedBackup(EncryptedBackup):
shutil.copy2(source_filename, output_filename)
return
inner_key = self.keybag.unwrap_key_for_class(
protection_class=file_plist.protection_class,
wrapped_file_key=file_plist.encryption_key,
inner_key = self._keybag.unwrapKeyForClass(
file_plist.protection_class, file_plist.encryption_key
)
self._decrypt_file_to_disk(
file_id=file_id,
@@ -64,6 +139,47 @@ 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.
@@ -260,36 +376,41 @@ 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
log.exception(exc)
log.critical("Failed to decrypt backup.")
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?"
)
def decrypt_with_key_file(self, key_file: str) -> None:
"""Decrypts an encrypted iOS backup using a key file.
@@ -303,6 +424,10 @@ 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()
@@ -317,30 +442,16 @@ class DecryptBackup:
key_bytes_raw = binascii.unhexlify(key_bytes)
self._backup = MVTEncryptedBackup(
backup_directory=self.backup_path,
passphrase_key=key_bytes_raw,
derived_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.")
log.critical(
"Failed to decrypt backup. Did you provide the correct key file?"
)
def get_key(self) -> None:
"""Retrieve and prints the encryption key."""
+261
View File
@@ -0,0 +1,261 @@
# 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
+4 -110
View File
@@ -3,15 +3,12 @@
# 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
@@ -27,89 +24,6 @@ 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
):
@@ -125,15 +39,13 @@ 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), passphrase_key=b"d" * 32
backup_directory=str(tmp_path), derived_key=b"d" * 32
)
mocker.patch.object(backup, "_read_and_unlock_keybag", return_value=True)
backup.keybag = mocker.Mock()
backup.keybag.unwrap_key_for_class.return_value = inner_key
backup._keybag = mocker.Mock()
backup._keybag.unwrapKeyForClass.return_value = inner_key
streaming_decrypt = mocker.spy(backup, "_decrypt_file_to_disk")
output_path = tmp_path / "output"
@@ -145,9 +57,6 @@ 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):
@@ -158,10 +67,8 @@ 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), passphrase_key=b"d" * 32
backup_directory=str(tmp_path), derived_key=b"d" * 32
)
mocker.patch.object(backup, "_read_and_unlock_keybag", return_value=True)
output_path = tmp_path / "output"
@@ -175,19 +82,6 @@ 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
Generated
+4 -4
View File
@@ -609,14 +609,14 @@ wheels = [
[[package]]
name = "iphone-backup-decrypt"
version = "0.11.2"
version = "0.10.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pycryptodome" },
]
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" }
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" }
wheels = [
{ 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" },
{ 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" },
]
[[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.11.2" },
{ name = "iphone-backup-decrypt", specifier = "==0.10.0" },
{ name = "nskeyedunarchiver", specifier = "==1.5.2" },
{ name = "packaging", specifier = "==26.3" },
{ name = "pyahocorasick", specifier = "==2.3.1" },