feat(perception): integrate calibrated operator pipeline
Add calibrated K1 projection, recorded and near-live perception qualification, unified Rerun operator layers, bounded replay admission, audited viewer controls, worker experiments, and lab evidence.
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,240 @@
|
||||
"""Export only LixelGO preferences from an owner-authorized USB iPhone.
|
||||
|
||||
MobileBackup2 asks the host how much free space is available before it starts.
|
||||
The device estimates that requirement as if the transfer were an ordinary full
|
||||
backup, even when this client discards every payload outside one exact app
|
||||
preferences domain. This lab tool reports a synthetic capacity only inside
|
||||
this filtered process. Non-matching payload bytes are consumed without being
|
||||
written by pymobiledevice3's existing ``preserve_file`` boundary.
|
||||
|
||||
The completed raw backup exists only inside a private temporary directory. The
|
||||
tool copies the one reviewed preferences plist to an ignored mode-0600 session
|
||||
and removes the temporary backup before returning. It never prints plist
|
||||
values, the device identifier, or any credential.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import plistlib
|
||||
import secrets
|
||||
import shutil
|
||||
import sqlite3
|
||||
import tempfile
|
||||
from contextlib import closing
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[4]
|
||||
SESSION_ROOT = REPO_ROOT / "sessions" / "iphone-k1-observation"
|
||||
BUNDLE_ID = "com.XGrids.LixelGo"
|
||||
APP_DOMAIN = f"AppDomain-{BUNDLE_ID}"
|
||||
PREFERENCES_PATH = f"Library/Preferences/{BUNDLE_ID}.plist"
|
||||
PREFERENCES_FILE_ID = hashlib.sha1(
|
||||
f"{APP_DOMAIN}-{PREFERENCES_PATH}".encode(),
|
||||
usedforsecurity=False,
|
||||
).hexdigest()
|
||||
SYNTHETIC_AVAILABLE_BYTES = 2 * 1024**4
|
||||
MAX_EXPORTED_PREFERENCES_BYTES = 16 * 1024 * 1024
|
||||
|
||||
|
||||
class TargetedProfileBackupError(RuntimeError):
|
||||
"""The bounded LixelGO preference export could not be completed safely."""
|
||||
|
||||
|
||||
def _is_preferences_backup_file(backup_file: Any) -> bool:
|
||||
"""Select only the exact LixelGO preferences payload or manifest row."""
|
||||
|
||||
# During transfer the hashed destination is exposed as ``file_name`` while
|
||||
# some protocol versions also expose it as ``device_name``. Manifest
|
||||
# pruning supplies the logical domain/path instead.
|
||||
for attribute in ("file_name", "device_name"):
|
||||
candidate = getattr(backup_file, attribute, None)
|
||||
if isinstance(candidate, str) and Path(candidate).name == PREFERENCES_FILE_ID:
|
||||
return True
|
||||
return (
|
||||
getattr(backup_file, "domain", None) == APP_DOMAIN
|
||||
and getattr(backup_file, "relative_path", None) == PREFERENCES_PATH
|
||||
)
|
||||
|
||||
|
||||
def _utc_now() -> str:
|
||||
return datetime.now(UTC).isoformat()
|
||||
|
||||
|
||||
def _single_device_directory(backup_root: Path) -> Path:
|
||||
candidates = [
|
||||
path
|
||||
for path in backup_root.iterdir()
|
||||
if path.is_dir() and not path.is_symlink()
|
||||
]
|
||||
if len(candidates) != 1:
|
||||
raise TargetedProfileBackupError("filtered backup did not create one device directory")
|
||||
return candidates[0]
|
||||
|
||||
|
||||
def _manifest_file_id(device_directory: Path) -> str:
|
||||
manifest_path = device_directory / "Manifest.db"
|
||||
if not manifest_path.is_file() or manifest_path.is_symlink():
|
||||
raise TargetedProfileBackupError("filtered backup has no readable Manifest.db")
|
||||
with closing(sqlite3.connect(f"file:{manifest_path}?mode=ro", uri=True)) as connection:
|
||||
rows = connection.execute(
|
||||
"SELECT fileID FROM Files WHERE domain = ? AND relativePath = ?",
|
||||
(APP_DOMAIN, PREFERENCES_PATH),
|
||||
).fetchall()
|
||||
if len(rows) != 1 or not isinstance(rows[0][0], str):
|
||||
raise TargetedProfileBackupError("LixelGO preferences were not present in the backup")
|
||||
file_id = rows[0][0]
|
||||
if len(file_id) != 40 or any(character not in "0123456789abcdef" for character in file_id):
|
||||
raise TargetedProfileBackupError("LixelGO preferences have an invalid backup file id")
|
||||
if file_id != PREFERENCES_FILE_ID:
|
||||
raise TargetedProfileBackupError(
|
||||
"LixelGO preferences backup file id does not match the exact requested path"
|
||||
)
|
||||
return file_id
|
||||
|
||||
|
||||
def _read_preferences(device_directory: Path) -> bytes:
|
||||
manifest = plistlib.loads((device_directory / "Manifest.plist").read_bytes())
|
||||
if manifest.get("IsEncrypted") is True:
|
||||
raise TargetedProfileBackupError(
|
||||
"encrypted filtered backups are not supported by this probe"
|
||||
)
|
||||
file_id = _manifest_file_id(device_directory)
|
||||
source = device_directory / file_id[:2] / file_id
|
||||
if source.is_symlink() or not source.is_file():
|
||||
raise TargetedProfileBackupError("LixelGO preferences payload is missing")
|
||||
size = source.stat().st_size
|
||||
if not 1 <= size <= MAX_EXPORTED_PREFERENCES_BYTES:
|
||||
raise TargetedProfileBackupError(
|
||||
"LixelGO preferences payload size is outside the safe bound"
|
||||
)
|
||||
payload = source.read_bytes()
|
||||
parsed = plistlib.loads(payload)
|
||||
if not isinstance(parsed, dict):
|
||||
raise TargetedProfileBackupError("LixelGO preferences root is not a dictionary")
|
||||
return payload
|
||||
|
||||
|
||||
async def _filtered_backup(backup_root: Path) -> None:
|
||||
from pymobiledevice3.lockdown import create_using_usbmux
|
||||
from pymobiledevice3.services.device_link import DeviceLink
|
||||
from pymobiledevice3.services.mobilebackup2 import Mobilebackup2Service
|
||||
|
||||
original_free_disk_handler = DeviceLink.get_free_disk_space
|
||||
|
||||
async def report_filtered_capacity(self: Any, message: Any) -> None:
|
||||
del message
|
||||
actual = shutil.disk_usage(self.root_path).free
|
||||
await self.status_response(
|
||||
0,
|
||||
status_dict=max(actual, SYNTHETIC_AVAILABLE_BYTES),
|
||||
)
|
||||
|
||||
lockdown = await create_using_usbmux(connection_type="USB")
|
||||
try:
|
||||
with patch.object(
|
||||
DeviceLink,
|
||||
"get_free_disk_space",
|
||||
report_filtered_capacity,
|
||||
):
|
||||
async with Mobilebackup2Service(lockdown) as service:
|
||||
if await service.get_will_encrypt():
|
||||
raise TargetedProfileBackupError(
|
||||
"device backup encryption is enabled; no filtered export was attempted"
|
||||
)
|
||||
await service.backup(
|
||||
full=True,
|
||||
backup_directory=backup_root,
|
||||
filter_callback=_is_preferences_backup_file,
|
||||
)
|
||||
finally:
|
||||
DeviceLink.get_free_disk_space = original_free_disk_handler
|
||||
await lockdown.close()
|
||||
|
||||
|
||||
def _write_session(preferences: bytes) -> Path:
|
||||
session_id = (
|
||||
datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
|
||||
+ f"_lixelgo-profile-export_{secrets.token_hex(2)}"
|
||||
)
|
||||
session_dir = SESSION_ROOT / session_id
|
||||
private_dir = session_dir / "private"
|
||||
private_dir.mkdir(parents=True, mode=0o700)
|
||||
preferences_path = private_dir / "lixelgo-preferences.plist"
|
||||
preferences_path.write_bytes(preferences)
|
||||
preferences_path.chmod(0o600)
|
||||
manifest = {
|
||||
"schema": "missioncore.xgrids-k1/lixelgo-profile-export/v1alpha1",
|
||||
"session_id": session_id,
|
||||
"status": "completed",
|
||||
"completed_at_utc": _utc_now(),
|
||||
"bundle_id": BUNDLE_ID,
|
||||
"source": "owner-authorized-usb-mobilebackup2-filtered",
|
||||
"preferences_present": True,
|
||||
"preferences_size_bytes": len(preferences),
|
||||
"device_identity_recorded": False,
|
||||
"preference_values_recorded_in_manifest": False,
|
||||
"credential_values_recorded_in_manifest": False,
|
||||
"temporary_backup_retained": False,
|
||||
}
|
||||
manifest_path = session_dir / "manifest.redacted.json"
|
||||
manifest_path.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
|
||||
manifest_path.chmod(0o600)
|
||||
return session_dir
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Export only LixelGO preferences from one trusted USB iPhone"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--acknowledge-filtered-capacity",
|
||||
action="store_true",
|
||||
help=(
|
||||
"confirm that the operator approved synthetic free-space reporting while "
|
||||
"all non-LixelGO preference payloads are discarded"
|
||||
),
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
from preflight import pymobiledevice3_binary, usb_device_count
|
||||
|
||||
args = parse_args()
|
||||
if not args.acknowledge_filtered_capacity:
|
||||
raise SystemExit("explicit --acknowledge-filtered-capacity is required")
|
||||
os.umask(0o077)
|
||||
if usb_device_count(pymobiledevice3_binary()) != 1:
|
||||
raise SystemExit("exactly one trusted USB iPhone must be connected")
|
||||
|
||||
try:
|
||||
with tempfile.TemporaryDirectory(prefix="mission-core-lixelgo-") as temporary:
|
||||
backup_root = Path(temporary) / "backup"
|
||||
backup_root.mkdir(mode=0o700)
|
||||
asyncio.run(_filtered_backup(backup_root))
|
||||
preferences = _read_preferences(_single_device_directory(backup_root))
|
||||
session_dir = _write_session(preferences)
|
||||
except (
|
||||
OSError,
|
||||
sqlite3.Error,
|
||||
plistlib.InvalidFileException,
|
||||
TargetedProfileBackupError,
|
||||
) as exc:
|
||||
raise SystemExit(f"filtered LixelGO profile export failed: {exc}") from exc
|
||||
|
||||
print(f"session: {session_dir}")
|
||||
print("status: completed")
|
||||
print("credential values printed: no")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user