Compare commits

..
13 changed files with 176 additions and 782 deletions

No files matched your search

-16
View File
@@ -1,16 +0,0 @@
# Coruna and DarkSword artifacts
The Manifest and Filesystem modules flag two scoped Coruna path patterns without requiring an IOC file:
- `com.apple.<16 hexadecimal characters>.plist` in the mobile user's `Library/Preferences` directory (the `HomeDomain` equivalent in a backup).
- `Documents/.devcache` at the root of an application data container (`AppDomain-<bundle ID>` in a backup).
Path-only matches produce medium-severity alerts. When file contents are available, the modules read at most 1 MiB and check the structures published in [iVerify's September 2026 report](https://www.iverify.com/blog/proliferation-of-coruna-and-darksword). A photo-tracking plist must contain a DCIM path and a positive finite processing timestamp. A device cache must identify `exp` as `plasma` and contain nonempty `ecid`, `gci`, and `unique-id` strings. Matching structures produce high-severity alerts. Matching device identifiers in distinct application containers produce an additional correlation alert. Results retain a fingerprint for correlation instead of raw device identifiers or the tracked photo path.
File content checks accept binary/XML plists and JSON. Missing, malformed or oversized content does not erase a path finding. A path finding alone is not confirmation of compromise. The checks do not inspect arbitrary plists based solely on field names, or flag `.devcache` outside the documented application-container location.
Backup domains with known fixed roots are mapped to device paths for IOC matching. Application container UUIDs are not invented from bundle IDs. iOS path comparisons handle `/var` versus `/private/var`, `/tmp` versus `/private/tmp`, and host path separators. File path indicators match the exact path or descendants at a directory boundary, not similarly prefixed sibling names.
For payload hashes, pass `--hashes` to `check-fs` or `check-backup` (or enable `HASH_FILES` in the configuration). The Filesystem and Manifest modules then store SHA-256 values and compare them with loaded hash IOCs; the saved module results can also be rechecked later. This reads available file contents and increases processing time. Acquisition integrity hashes in `info.json` remain available as before. Reads for artifact inspection and these module hashes are confined to regular files inside the acquisition; symlinks are not followed.
Decrypt encrypted backups before analysis. Sysdiagnose and filesystem metadata may expose paths without the corresponding file bytes, and absence from a partial acquisition does not establish absence from the device. Changes to payload bytes defeat exact hash matching; source-level sample names and legitimate injected process names are not standalone malware indicators.
+1 -1
View File
@@ -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",
+2 -7
View File
@@ -646,19 +646,14 @@ class Indicators:
if not file_path:
return None
# Device paths use forward slashes even when analysed on Windows.
file_path = file_path.replace("\\", "/")
ioc_match = self.check_file_name(file_path.rsplit("/", 1)[-1])
ioc_match = self.check_file_name(os.path.basename(file_path))
if ioc_match:
return ioc_match
for ioc in self.get_iocs("file_paths"):
# Strip any trailing slash from indicator paths to match
# directories.
indicator_path = ioc.value.replace("\\", "/").rstrip("/")
if file_path == indicator_path or file_path.startswith(
indicator_path + "/"
):
if file_path.startswith(ioc.value.rstrip("/")):
return IndicatorMatch(
ioc=ioc,
message=f'Found a known suspicious file path "{file_path}" matching indicators form "{ioc.name}"',
-3
View File
@@ -8,7 +8,6 @@ import os
from typing import Optional
from mvt.common.command import Command
from mvt.common.config import settings
from mvt.common.indicators import Indicators
from mvt.common.module import MVTModule
@@ -57,8 +56,6 @@ class CmdIOSCheckBackup(Command):
)
self.platform = "ios"
if hashes or settings.HASH_FILES:
self.module_options["check_file_hashes"] = True
self.name = "check-backup"
self.modules = BACKUP_MODULES + MIXED_MODULES
-3
View File
@@ -7,7 +7,6 @@ import logging
from typing import Optional
from mvt.common.command import Command
from mvt.common.config import settings
from mvt.common.indicators import Indicators
from mvt.common.module import MVTModule
@@ -49,8 +48,6 @@ class CmdIOSCheckFS(Command):
)
self.platform = "ios"
if hashes or settings.HASH_FILES:
self.module_options["check_file_hashes"] = True
self.name = "check-fs"
self.modules = FS_MODULES + MIXED_MODULES
-168
View File
@@ -1,168 +0,0 @@
"""Coruna artifact checks shared by backups, filesystems and metadata listings.
Structures documented by iVerify on 2026-09-15:
https://www.iverify.com/blog/proliferation-of-coruna-and-darksword
"""
import hashlib
import json
import math
import plistlib
import re
import stat
from collections import defaultdict
from pathlib import Path
from typing import Optional
from xml.parsers.expat import ExpatError
from mvt.common.alerts import AlertStore
from .paths import backup_device_path, normalize_ios_path
_PHOTO_STATE = re.compile(
r"/private/var/mobile/Library/Preferences/com\.apple\.[0-9a-fA-F]{16}\.plist"
)
_DEVICE_CACHE = re.compile(
r"/private/var/mobile/Containers/Data/Application/"
r"(?P<container>[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12})"
r"/Documents/\.devcache"
)
MAX_ARTIFACT_SIZE = 1024 * 1024
def coruna_path_artifact(path: str, domain: Optional[str] = None) -> Optional[dict]:
"""Return a scoped candidate, never a verdict based on a basename alone."""
if domain is not None:
relative = path.replace("\\", "/")
if relative == "Documents/.devcache" and domain.startswith("AppDomain-"):
return {"kind": "device-cache", "container": domain}
path = backup_device_path(domain, relative) or ""
if not path:
return None
path = normalize_ios_path(path)
if _PHOTO_STATE.fullmatch(path):
return {"kind": "photo-state"}
match = _DEVICE_CACHE.fullmatch(path)
if match:
return {"kind": "device-cache", "container": match["container"].lower()}
return None
def inspect_coruna_artifact(file_path: str, artifact: dict) -> dict:
"""Enrich a candidate without exporting photo paths or device identifiers."""
result = dict(artifact)
try:
with open(file_path, "rb") as handle:
raw = handle.read(MAX_ARTIFACT_SIZE + 1)
if len(raw) > MAX_ARTIFACT_SIZE:
return {**result, "content_status": "too-large"}
try:
data = plistlib.loads(raw)
except (ValueError, plistlib.InvalidFileException, ExpatError):
data = json.loads(raw)
except (OSError, ValueError, UnicodeError, OverflowError):
return {**result, "content_status": "unavailable"}
matched = False
if isinstance(data, dict):
if artifact["kind"] == "photo-state":
photo = data.get("LastProcessedFilePath")
timestamp = data.get("LastProcessedTimestamp")
matched = (
isinstance(photo, str)
and normalize_ios_path(photo).startswith(
"/private/var/mobile/Media/DCIM/"
)
and isinstance(timestamp, (int, float))
and not isinstance(timestamp, bool)
and (isinstance(timestamp, int) or math.isfinite(timestamp))
and timestamp > 0
)
elif artifact["kind"] == "device-cache":
keys = ("ecid", "gci", "unique-id")
matched = data.get("exp") == "plasma" and all(
isinstance(data.get(key), str) and data[key].strip() for key in keys
)
if matched:
identifiers = {key: data[key] for key in keys}
result["identifier_sha256"] = hashlib.sha256(
json.dumps(identifiers, sort_keys=True).encode()
).hexdigest()
result["content_status"] = "matched" if matched else "unmatched"
return result
def enrich_coruna_record(record: dict, file_path: Optional[str]) -> None:
artifact = coruna_path_artifact(
record.get("relative_path", record.get("path", "")), record.get("domain")
)
if artifact:
record["coruna_artifact"] = (
inspect_coruna_artifact(file_path, artifact) if file_path else artifact
)
def alert_coruna_record(record: dict, alertstore: AlertStore) -> None:
if record.get("is_directory") or record.get("flags") == 2:
return
mode = record.get("mode")
if isinstance(mode, int) and not stat.S_ISREG(mode):
return
artifact = record.get("coruna_artifact")
if not artifact:
artifact = coruna_path_artifact(
record.get("relative_path", record.get("path", "")), record.get("domain")
)
if not artifact:
return
path = record.get("relative_path", record.get("path", ""))
timestamp = record.get("modified", record.get("isodate", ""))
if artifact.get("content_status") == "matched":
alertstore.high(
f"Coruna {artifact['kind']} artifact with matching content structure: {path}",
timestamp,
record,
)
else:
alertstore.medium(
f"Possible Coruna {artifact['kind']} artifact at {path}; "
"path match without corroborating content",
timestamp,
record,
)
def correlate_coruna_records(records: list, alertstore: AlertStore) -> None:
groups: dict[str, dict[str, dict]] = defaultdict(dict)
for record in records:
artifact = record.get("coruna_artifact", {})
if (
artifact.get("kind") != "device-cache"
or artifact.get("content_status") != "matched"
or not artifact.get("container")
):
continue
fingerprint = artifact.get("identifier_sha256")
if fingerprint:
groups[fingerprint][artifact["container"]] = record
for fingerprint, containers in groups.items():
if len(containers) < 2:
continue
alertstore.high(
f"Matching Coruna device identifiers in {len(containers)} application containers",
"",
{"identifier_sha256": fingerprint, "containers": sorted(containers)},
)
def contained_regular_file(root: str, file_path: str) -> bool:
"""Do not read symlinks or paths outside the supplied acquisition."""
path = Path(file_path)
try:
return (
not path.is_symlink()
and path.is_file()
and path.resolve().is_relative_to(Path(root).resolve())
)
except (OSError, RuntimeError, ValueError):
return False
+50 -161
View File
@@ -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."""
+2 -49
View File
@@ -18,19 +18,7 @@ from mvt.common.module_types import (
ModuleSerializedResult,
)
from mvt.common.url import URL
from mvt.common.utils import (
convert_datetime_to_iso,
convert_unix_to_iso,
get_sha256_from_file_path,
)
from mvt.ios.coruna import (
alert_coruna_record,
contained_regular_file,
correlate_coruna_records,
coruna_path_artifact,
enrich_coruna_record,
)
from mvt.ios.paths import backup_device_path, check_ios_path
from mvt.common.utils import convert_datetime_to_iso, convert_unix_to_iso
from ..base import IOSExtraction
@@ -106,34 +94,14 @@ class Manifest(IOSExtraction):
return records
def check_indicators(self) -> None:
correlate_coruna_records(self.results, self.alertstore)
for result in self.results:
if not result.get("relative_path"):
continue
alert_coruna_record(result, self.alertstore)
if not self.indicators:
continue
hash_match = self.indicators.check_file_hash(result.get("sha256", ""))
if hash_match:
self.alertstore.critical(
hash_match.message,
result.get("modified", ""),
result,
matched_indicator=hash_match.ioc,
)
device_path = backup_device_path(
result.get("domain", ""), result["relative_path"]
)
ioc_match = (
check_ios_path(self.indicators, device_path) if device_path else None
)
if not ioc_match:
ioc_match = self.indicators.check_file_path(
"/" + result["relative_path"]
)
ioc_match = self.indicators.check_file_path("/" + result["relative_path"])
if ioc_match:
self.alertstore.high(
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
@@ -245,21 +213,6 @@ class Manifest(IOSExtraction):
file_data["relativePath"],
)
inspect_content = coruna_path_artifact(
cleaned_metadata["relative_path"], cleaned_metadata["domain"]
)
hash_content = self.module_options.get("check_file_hashes", False)
if inspect_content or hash_content:
source_path = self._get_backup_file_from_id(cleaned_metadata["file_id"])
if source_path and contained_regular_file(
self.target_path, source_path
):
if inspect_content:
enrich_coruna_record(cleaned_metadata, source_path)
if hash_content:
cleaned_metadata["sha256"] = get_sha256_from_file_path(
source_path
)
self.results.append(cleaned_metadata)
cur.close()
+7 -35
View File
@@ -7,15 +7,7 @@ import logging
import os
from typing import Optional
from mvt.common.utils import convert_unix_to_iso, get_sha256_from_file_path
from mvt.ios.coruna import (
alert_coruna_record,
contained_regular_file,
correlate_coruna_records,
coruna_path_artifact,
enrich_coruna_record,
)
from mvt.ios.paths import check_ios_path, normalize_ios_path
from mvt.common.utils import convert_unix_to_iso
from mvt.common.module_types import (
ModuleAtomicResult,
ModuleSerializedResult,
@@ -57,23 +49,14 @@ class Filesystem(IOSExtraction):
}
def check_indicators(self) -> None:
correlate_coruna_records(self.results, self.alertstore)
if not self.indicators:
return
for result in self.results:
if "path" not in result:
continue
alert_coruna_record(result, self.alertstore)
if not self.indicators:
continue
hash_match = self.indicators.check_file_hash(result.get("sha256", ""))
if hash_match:
self.alertstore.critical(
hash_match.message,
result.get("modified", ""),
result,
matched_indicator=hash_match.ioc,
)
ioc_match = check_ios_path(self.indicators, result["path"])
ioc_match = self.indicators.check_file_path(result["path"])
if ioc_match:
self.alertstore.high(
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
@@ -83,9 +66,7 @@ class Filesystem(IOSExtraction):
if self.module_options.get("fast_mode", None):
continue
ioc_match = self.indicators.check_file_path_process(
normalize_ios_path(result["path"])
)
ioc_match = self.indicators.check_file_path_process(result["path"])
if ioc_match:
self.alertstore.high(
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
@@ -100,12 +81,11 @@ class Filesystem(IOSExtraction):
for dir_name in dirs:
try:
dir_path = os.path.join(root, dir_name)
result: ModuleAtomicResult = {
result = {
"path": os.path.relpath(dir_path, self.target_path).replace(
os.sep, "/"
),
"modified": convert_unix_to_iso(os.stat(dir_path).st_mtime),
"is_directory": True,
}
except Exception:
continue
@@ -121,14 +101,6 @@ class Filesystem(IOSExtraction):
),
"modified": convert_unix_to_iso(os.stat(file_path).st_mtime),
}
inspect_content = coruna_path_artifact(result["path"])
hash_content = self.module_options.get("check_file_hashes", False)
if (inspect_content or hash_content) and contained_regular_file(
self.target_path, file_path
):
enrich_coruna_record(result, file_path)
if hash_content:
result["sha256"] = get_sha256_from_file_path(file_path)
except Exception:
continue
else:
-47
View File
@@ -1,47 +0,0 @@
"""Compare device paths independently of the analysis host and backup layout."""
import posixpath
from typing import Optional
from mvt.common.indicators import IndicatorMatch, Indicators
def normalize_ios_path(path: str) -> str:
path = posixpath.normpath("/" + path.replace("\\", "/").lstrip("/"))
if path == "/var" or path.startswith("/var/"):
return "/private" + path
if path == "/tmp" or path.startswith("/tmp/"):
return "/private" + path
return path
def ios_path_variants(path: str) -> tuple[str, ...]:
path = normalize_ios_path(path)
if path.startswith(("/private/var/", "/private/tmp/")):
return path, path[len("/private") :]
return (path,)
def backup_device_path(domain: str, relative_path: str) -> Optional[str]:
"""Map only domains with a known fixed root; app UUIDs are not in Manifest.db."""
roots = {
"HomeDomain": "/private/var/mobile",
"CameraRollDomain": "/private/var/mobile",
"MediaDomain": "/private/var/mobile/Media",
"RootDomain": "/private/var/root",
"WirelessDomain": "/private/var/wireless",
}
root = roots.get(domain)
relative_path = relative_path.replace("\\", "/")
if not root or relative_path.startswith("/") or ".." in relative_path.split("/"):
return None
return normalize_ios_path(root + "/" + relative_path)
def check_ios_path(indicators: Indicators, path: str) -> Optional[IndicatorMatch]:
for candidate in ios_path_variants(path):
match = indicators.check_file_path(candidate)
if match:
return match
# Filesystem records and existing indicators may both use relative paths.
return indicators.check_file_path(posixpath.normpath(path.replace("\\", "/")))
+110 -4
View File
@@ -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
-284
View File
@@ -1,284 +0,0 @@
import hashlib
import json
import plistlib
import sqlite3
import pytest
from mvt.common.alerts import AlertLevel, AlertStore
from mvt.common.indicators import Indicators
from mvt.ios.coruna import (
MAX_ARTIFACT_SIZE,
alert_coruna_record,
contained_regular_file,
correlate_coruna_records,
coruna_path_artifact,
inspect_coruna_artifact,
)
from mvt.ios.modules.backup.manifest import Manifest
from mvt.ios.modules.fs.filesystem import Filesystem
from mvt.ios.paths import backup_device_path, check_ios_path
PHOTO = "/private/var/mobile/Library/Preferences/com.apple.0123456789abcdef.plist"
CONTAINER = "/private/var/mobile/Containers/Data/Application/12345678-1234-1234-1234-123456789abc"
PHOTO_DATA = {
"LastProcessedFilePath": "/var/mobile/Media/DCIM/100APPLE/IMG_0001.PNG",
"LastProcessedTimestamp": 1762537021.5,
}
CACHE_DATA = {
"exp": "plasma",
"ecid": "test-ecid",
"gci": "test-gci",
"unique-id": "test-id",
}
def indicators(**values):
result = Indicators()
result.ioc_collections.append(
{"name": "test", "stix2_file_name": "test.stix2", **values}
)
return result
@pytest.mark.parametrize(
"path", [PHOTO, PHOTO[1:], PHOTO.replace("/private", ""), PHOTO.replace("/", "\\")]
)
def test_photo_state_aliases(path):
assert coruna_path_artifact(path) == {"kind": "photo-state"}
@pytest.mark.parametrize(
"path",
[
PHOTO + ".bak",
PHOTO.replace("abcdef", "abcdeg"),
PHOTO.replace("abcdef", "abcde"),
PHOTO.replace("/mobile/", "/root/"),
"/tmp/" + PHOTO.rsplit("/", 1)[-1],
CONTAINER + "/Library/.devcache",
"/Documents/.devcache",
CONTAINER.replace("/Data/", "/Bundle/") + "/Documents/.devcache",
],
)
def test_artifact_paths_are_scoped(path):
assert coruna_path_artifact(path) is None
def test_backup_domain_scoping():
relative = "Library/Preferences/" + PHOTO.rsplit("/", 1)[-1]
assert coruna_path_artifact(relative, "HomeDomain")
assert coruna_path_artifact(relative, "AppDomain-com.example.app") is None
assert coruna_path_artifact("Documents/.devcache", "AppDomain-com.example.app")
assert coruna_path_artifact("Documents/.devcache", "HomeDomain") is None
assert (
coruna_path_artifact("Documents/.devcache.bak", "AppDomain-com.example.app")
is None
)
assert (
backup_device_path("MediaDomain", "PostLogs.txt")
== "/private/var/mobile/Media/PostLogs.txt"
)
assert (
backup_device_path("CameraRollDomain", "Media/DCIM/image.png")
== "/private/var/mobile/Media/DCIM/image.png"
)
assert backup_device_path("HomeDomain", "../root/file") is None
assert backup_device_path("AppDomain-com.example.app", "Documents/file") is None
@pytest.mark.parametrize(
"serializer",
[
plistlib.dumps,
lambda d: plistlib.dumps(d, fmt=plistlib.FMT_BINARY),
lambda d: json.dumps(d).encode(),
],
)
@pytest.mark.parametrize(
"kind,data", [("photo-state", PHOTO_DATA), ("device-cache", CACHE_DATA)]
)
def test_content_formats(tmp_path, serializer, kind, data):
path = tmp_path / "artifact"
path.write_bytes(serializer(data))
result = inspect_coruna_artifact(str(path), {"kind": kind})
assert result["content_status"] == "matched"
assert "test-ecid" not in json.dumps(result)
assert "IMG_0001.PNG" not in json.dumps(result)
@pytest.mark.parametrize(
"data",
[
{},
[],
{**PHOTO_DATA, "LastProcessedFilePath": "/tmp/photo"},
{**PHOTO_DATA, "LastProcessedTimestamp": True},
{**PHOTO_DATA, "LastProcessedTimestamp": float("inf")},
{"lastProcessedTimestamp": 123},
],
)
def test_photo_content_requires_full_structure(tmp_path, data):
path = tmp_path / "artifact"
path.write_text(json.dumps(data))
assert (
inspect_coruna_artifact(str(path), {"kind": "photo-state"})["content_status"]
== "unmatched"
)
@pytest.mark.parametrize(
"raw",
[
b"not a plist",
b"<?xml version='1.0'?><plist><broken",
b"x" * (MAX_ARTIFACT_SIZE + 1),
],
ids=["invalid", "malformed-xml", "oversized"],
)
def test_malformed_or_oversized_content(tmp_path, raw):
path = tmp_path / "artifact"
path.write_bytes(raw)
assert (
inspect_coruna_artifact(str(path), {"kind": "device-cache"})["content_status"]
!= "matched"
)
@pytest.mark.parametrize(
"data",
[{**CACHE_DATA, "exp": "other"}, {"exp": "plasma"}, {**CACHE_DATA, "ecid": ""}],
)
def test_device_cache_content_requires_identifiers(tmp_path, data):
path = tmp_path / "artifact"
path.write_text(json.dumps(data))
assert (
inspect_coruna_artifact(str(path), {"kind": "device-cache"})["content_status"]
== "unmatched"
)
def test_correlation_requires_matching_identifiers_in_distinct_containers(tmp_path):
path = tmp_path / "cache"
path.write_text(json.dumps(CACHE_DATA))
a = inspect_coruna_artifact(str(path), {"kind": "device-cache", "container": "one"})
store = AlertStore()
correlate_coruna_records([{"coruna_artifact": a}, {"coruna_artifact": a}], store)
assert not store.alerts
correlate_coruna_records(
[{"coruna_artifact": a}, {"coruna_artifact": {**a, "container": "two"}}], store
)
assert len(store.alerts) == 1
assert store.alerts[0].event["containers"] == ["one", "two"]
assert store.alerts[0].level == AlertLevel.HIGH
def test_path_only_alert_is_medium():
store = AlertStore()
alert_coruna_record({"path": PHOTO, "isodate": "2026-09-15"}, store)
assert store.alerts[0].level == AlertLevel.MEDIUM
assert store.alerts[0].event_time == "2026-09-15"
@pytest.mark.parametrize(
"ioc,path,expected",
[
("/private/var/tmp/.ghost_upper", "/var/tmp/.ghost_upper/manifest.json", True),
("/var/tmp/ioslab_ctl", "private\\var\\tmp\\ioslab_ctl", True),
("/var/tmp/ioslab_ctl", "/var/tmp/ioslab_ctl2", False),
("/private/var/tmp/.ghost_upper/", "/var/tmp/.ghost_upper-copy/file", False),
("matched/directory", "matched/directory/file", True),
("matched/directory", "matched\\directory\\file", True),
("matched/directory", "matched/directory-copy/file", False),
("matched/directory", "matched/directory/../other/file", False),
],
)
def test_ios_path_iocs_respect_aliases_and_boundaries(ioc, path, expected):
assert bool(check_ios_path(indicators(file_paths=[ioc]), path)) is expected
def test_filesystem_reads_candidate_and_hashes_when_requested(tmp_path):
path = tmp_path / PHOTO.lstrip("/")
path.parent.mkdir(parents=True)
path.write_bytes(plistlib.dumps(PHOTO_DATA))
digest = hashlib.sha256(path.read_bytes()).hexdigest()
module = Filesystem(
target_path=str(tmp_path), module_options={"check_file_hashes": True}
)
module.indicators = indicators(files_sha256=[digest])
module.run()
module.check_indicators()
assert sorted(a.level.value for a in module.alertstore.alerts) == [30, 40]
record = next(r for r in module.results if "sha256" in r)
replay = Filesystem(results=[record])
replay.indicators = module.indicators
replay.check_indicators()
assert sorted(a.level.value for a in replay.alertstore.alerts) == [30, 40]
def test_filesystem_does_not_hash_by_default(tmp_path):
(tmp_path / "file").write_bytes(b"example")
module = Filesystem(target_path=str(tmp_path))
module.run()
assert all("sha256" not in r for r in module.results)
def test_backup_reads_coruna_candidate_and_maps_path_ioc(tmp_path):
file_id = "a" * 40
path = tmp_path / "aa" / file_id
path.parent.mkdir()
path.write_bytes(plistlib.dumps(PHOTO_DATA))
with sqlite3.connect(tmp_path / "Manifest.db") as db:
db.execute(
"CREATE TABLE Files(fileID TEXT, domain TEXT, relativePath TEXT, flags INTEGER, file BLOB)"
)
db.execute(
"INSERT INTO Files VALUES(?,?,?,?,?)",
(
file_id,
"HomeDomain",
"Library/Preferences/" + PHOTO.rsplit("/", 1)[-1],
1,
None,
),
)
db.execute(
"INSERT INTO Files VALUES(?,?,?,?,?)",
("b" * 40, "MediaDomain", "PostLogs.txt", 1, None),
)
db.close()
module = Manifest(
target_path=str(tmp_path), module_options={"check_file_hashes": True}
)
module.indicators = indicators(
file_paths=["/var/mobile/Media/PostLogs.txt"],
files_sha256=[hashlib.sha256(path.read_bytes()).hexdigest()],
)
module.run()
module.check_indicators()
assert sorted(a.level.value for a in module.alertstore.alerts) == [30, 30, 40]
def test_acquisition_boundary(tmp_path):
root = tmp_path / "root"
root.mkdir()
outside = tmp_path / "outside"
outside.write_bytes(b"not evidence")
assert not contained_regular_file(str(root), str(outside))
@pytest.mark.parametrize(
"metadata", [{"flags": 2}, {"is_directory": True}, {"mode": 16877}]
)
def test_directories_are_not_coruna_file_artifacts(metadata):
store = AlertStore()
alert_coruna_record({"path": PHOTO, **metadata}, store)
assert not store.alerts
def test_hash_option_reaches_ios_modules():
from mvt.ios.cmd_check_backup import CmdIOSCheckBackup
from mvt.ios.cmd_check_fs import CmdIOSCheckFS
for command in (CmdIOSCheckBackup, CmdIOSCheckFS):
assert command(hashes=True).module_options["check_file_hashes"]
Generated
+4 -4
View File
@@ -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" },