mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-28 16:56:23 +01:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
215b460a2e | ||
|
|
e9f868a71a | ||
|
|
f9d179b263 |
No files matched your search
@@ -0,0 +1,16 @@
|
||||
# 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
@@ -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",
|
||||
|
||||
@@ -646,14 +646,19 @@ class Indicators:
|
||||
if not file_path:
|
||||
return None
|
||||
|
||||
ioc_match = self.check_file_name(os.path.basename(file_path))
|
||||
# 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])
|
||||
if ioc_match:
|
||||
return ioc_match
|
||||
|
||||
for ioc in self.get_iocs("file_paths"):
|
||||
# Strip any trailing slash from indicator paths to match
|
||||
# directories.
|
||||
if file_path.startswith(ioc.value.rstrip("/")):
|
||||
indicator_path = ioc.value.replace("\\", "/").rstrip("/")
|
||||
if file_path == indicator_path or file_path.startswith(
|
||||
indicator_path + "/"
|
||||
):
|
||||
return IndicatorMatch(
|
||||
ioc=ioc,
|
||||
message=f'Found a known suspicious file path "{file_path}" matching indicators form "{ioc.name}"',
|
||||
|
||||
@@ -8,6 +8,7 @@ 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
|
||||
|
||||
@@ -56,6 +57,8 @@ 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
|
||||
|
||||
|
||||
@@ -7,6 +7,7 @@ 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
|
||||
|
||||
@@ -48,6 +49,8 @@ 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
|
||||
|
||||
|
||||
@@ -0,0 +1,168 @@
|
||||
"""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
|
||||
+161
-50
@@ -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."""
|
||||
|
||||
@@ -18,7 +18,19 @@ 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
|
||||
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 ..base import IOSExtraction
|
||||
|
||||
@@ -94,14 +106,34 @@ 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
|
||||
|
||||
ioc_match = self.indicators.check_file_path("/" + result["relative_path"])
|
||||
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"]
|
||||
)
|
||||
if ioc_match:
|
||||
self.alertstore.high(
|
||||
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
||||
@@ -213,6 +245,21 @@ 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,7 +7,15 @@ import logging
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.utils import convert_unix_to_iso
|
||||
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.module_types import (
|
||||
ModuleAtomicResult,
|
||||
ModuleSerializedResult,
|
||||
@@ -49,14 +57,23 @@ class Filesystem(IOSExtraction):
|
||||
}
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
if not self.indicators:
|
||||
return
|
||||
|
||||
correlate_coruna_records(self.results, self.alertstore)
|
||||
for result in self.results:
|
||||
if "path" not in result:
|
||||
continue
|
||||
|
||||
ioc_match = self.indicators.check_file_path(result["path"])
|
||||
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"])
|
||||
if ioc_match:
|
||||
self.alertstore.high(
|
||||
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
||||
@@ -66,7 +83,9 @@ class Filesystem(IOSExtraction):
|
||||
if self.module_options.get("fast_mode", None):
|
||||
continue
|
||||
|
||||
ioc_match = self.indicators.check_file_path_process(result["path"])
|
||||
ioc_match = self.indicators.check_file_path_process(
|
||||
normalize_ios_path(result["path"])
|
||||
)
|
||||
if ioc_match:
|
||||
self.alertstore.high(
|
||||
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
||||
@@ -81,11 +100,12 @@ class Filesystem(IOSExtraction):
|
||||
for dir_name in dirs:
|
||||
try:
|
||||
dir_path = os.path.join(root, dir_name)
|
||||
result = {
|
||||
result: ModuleAtomicResult = {
|
||||
"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
|
||||
@@ -101,6 +121,14 @@ 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:
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
"""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("\\", "/")))
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,284 @@
|
||||
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"]
|
||||
@@ -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" },
|
||||
|
||||
Reference in new issue
Block a user