ゴーストAPNプロファイルをバックアップの編集と復元で修復する。

#!/usr/bin/env python3
"""Prepare an encrypted backup copy without one APN plist; restore separately.
 
For pymobiledevice3 11.19.1. No device access during prepare.
Experimental APN repair: preparation checks do NOT prove an iOS restore will
succeed or resolve the ghost APN. Never use APN-Diagnostic as the source.
"""
import argparse
from contextlib import closing
import getpass
import hashlib
from importlib.metadata import version
import json
import os
from pathlib import Path
import plistlib
import re
import runpy
import shutil
import sqlite3
import sys
import tempfile
 
UDID = os.environ.get('APN_TARGET_UDID', '').strip()
DOMAIN = 'WirelessDomain'
RELATIVE = 'Library/Preferences/com.apple.commcenter.cellular_profile.plist'
RECEIPT = 'APN_REPAIR_READY.json'
 
 
def require(condition, message):
    if not condition:
        raise RuntimeError(message)
 
 
def plist(path):
    return plistlib.loads(path.read_bytes())
 
 
def digest(path):
    with path.open('rb') as f:
        return hashlib.file_digest(f, 'sha256').hexdigest()
 
 
def readonly(path):
    return sqlite3.connect(path.resolve().as_uri() + '?mode=ro', uri=True)
 
 
def check_db(db):
    require(db.execute('PRAGMA quick_check').fetchall() == [('ok',)],
            'Manifest database integrity check failed.')
 
 
def payload_path(folder, file_id):
    require(isinstance(file_id, str) and re.fullmatch('[0-9a-fA-F]{40}', file_id),
            'Unexpected backup file ID.')
    nested = folder / file_id[:2] / file_id
    return nested if nested.is_file() else folder / file_id
 
 
def backend():
    require(version('pymobiledevice3') == '11.19.1',
            'This script was checked against pymobiledevice3 11.19.1. Stop on a different version.')
    from pymobiledevice3.services.mobilebackup2 import Mobilebackup2Service
    return Mobilebackup2Service
 
 
def edit_manifest(original, edited):
    """Copy the decrypted database, remove exactly one row, compare all rows."""
    with closing(readonly(original)) as db:
        check_db(db)
        matches = db.execute(
            'SELECT fileID, flags FROM Files WHERE domain=? AND relativePath=?',
            (DOMAIN, RELATIVE)).fetchall()
        require(len(matches) == 1 and matches[0][1] == 1,
                'Expected exactly one regular APN file entry.')
        file_id = matches[0][0]
        count = db.execute('SELECT COUNT(*) FROM Files').fetchone()[0]
        # Reject incomplete/filtered snapshots before creating a repair copy.
        regular_ids = [r[0] for r in db.execute('SELECT fileID FROM Files WHERE flags=1')]
        with closing(sqlite3.connect(edited)) as out:
            db.backup(out)
    with closing(sqlite3.connect(edited)) as db:
        with db:
            n = db.execute('DELETE FROM Files WHERE fileID=? AND domain=? AND relativePath=?',
                           (file_id, DOMAIN, RELATIVE)).rowcount
            require(n == 1, 'Deletion count is not one.')
        check_db(db)
        require(db.execute('SELECT COUNT(*) FROM Files').fetchone()[0] == count - 1,
                'Unexpected remaining row count.')
        db.execute('ATTACH DATABASE ? AS original', (str(original),))
        removed = db.execute('SELECT * FROM original.Files EXCEPT SELECT * FROM main.Files').fetchall()
        added = db.execute('SELECT * FROM main.Files EXCEPT SELECT * FROM original.Files').fetchall()
        require(len(removed) == 1 and not added,
                'Other file metadata changed unexpectedly.')
        db.execute('DETACH DATABASE original')
    return file_id, count, regular_ids
 
 
def prepare(source, output, password, service):
    source, output = source.resolve(), output.resolve()
    require(source.is_dir(), 'Source backup folder does not exist.')
    require(not output.exists(), 'Output folder already exists. Nothing was overwritten.')
    require(not source.is_relative_to(output) and not output.is_relative_to(source),
            'Source and output folders must be separate.')
    require(not any(name in str(source) for name in ('APN-Diagnostic', 'APN-Current')),
            'A diagnostic backup must never be restored.')
    require(plist(source / 'Manifest.plist').get('IsEncrypted') is True,
            'The source is not encrypted. Do not use the earlier APN-Repair copy.')
    require(plist(source / 'Status.plist').get('SnapshotState') == 'finished',
            'The source backup is not marked finished.')
    info = plist(source / 'Info.plist')
    identity = info.get('Target Identifier') or info.get('Unique Identifier')
    require(isinstance(identity, str) and identity.lower() == UDID.lower(),
            'The source backup is not from the expected new iPhone.')
    for suffix in ('-wal', '-journal'):
        p = source / ('Manifest.db' + suffix)
        require(not p.exists() or p.stat().st_size == 0, 'Active database journal found. Close Apple Devices.')
    original_hash = digest(source / 'Manifest.db')
    metadata_hash = digest(source / 'Manifest.plist')
    print('Source:', source, flush=True)
    print('Backup date (as stored):', info.get('Last Backup Date', 'unknown'), flush=True)
    print('Decrypting and checking the manifest...', flush=True)
    # Closed files in a TemporaryDirectory avoid Windows NamedTemporaryFile locks.
    with tempfile.TemporaryDirectory(prefix='apn-check-') as tmp:
        tmp = Path(tmp)
        original, edited = tmp / 'original.db', tmp / 'edited.db'
        key = service._decrypt_backup_manifest_db(source, password, original)
        file_id, count, regular_ids = edit_manifest(original, edited)
        print('Checking that all backed-up file payloads exist...', flush=True)
        for fid in regular_ids:
            require(payload_path(source, fid).is_file(),
                    'Source contains missing payloads; it may be incomplete or filtered. Stopped.')
        old_blob = payload_path(source, file_id)
        old_blob_hash = digest(old_blob)
        print('Calculating required disk space...', flush=True)
        total = sum(p.stat().st_size for p in source.rglob('*') if p.is_file())
        require(output.parent.is_dir(), 'Output parent folder does not exist.')
        free = shutil.disk_usage(output.parent).free
        print(f'Backup: {total / 2**30:.2f} GiB; free: {free / 2**30:.2f} GiB', flush=True)
        require(free >= total + 2 * 1024**3, 'Not enough space for a physical copy plus 2 GiB.')
        output.mkdir()
        (output / 'INCOMPLETE_DO_NOT_RESTORE.txt').write_text(
            'Preparation is incomplete. Do not restore this folder.\n', encoding='utf-8')
        dest = output / UDID
        print('Copying the COMPLETE encrypted backup. This can take several minutes...', flush=True)
        shutil.copytree(source, dest)
        require(digest(dest / 'Manifest.db') == original_hash, 'Manifest changed during copying.')
        require(digest(dest / 'Manifest.plist') == metadata_hash, 'Encryption metadata changed during copying.')
        for fid in regular_ids:
            a, b = payload_path(source, fid), payload_path(dest, fid)
            require(b.is_file() and b.stat().st_size == a.stat().st_size,
                    'A copied payload is missing or has the wrong size.')
        print('Re-encrypting the edited manifest...', flush=True)
        staged = tmp / 'Manifest.encrypted'
        service._encrypt_backup_manifest_db(edited, staged, key)
        # The copy alone is modified. Every other encrypted payload stays intact.
        shutil.copyfile(staged, dest / 'Manifest.db')
        verify = tmp / 'verify.db'
        service._decrypt_backup_manifest_db(dest, password, verify)
        require(digest(verify) == digest(edited), 'Encrypted manifest round-trip check failed.')
        with closing(readonly(verify)) as db:
            check_db(db)
            require(db.execute('SELECT COUNT(*) FROM Files WHERE domain=? AND relativePath=?',
                               (DOMAIN, RELATIVE)).fetchone()[0] == 0, 'APN entry is still present.')
        payload_path(dest, file_id).unlink()
        require(digest(source / 'Manifest.db') == original_hash and
                digest(source / 'Manifest.plist') == metadata_hash and
                digest(old_blob) == old_blob_hash, 'Source changed during preparation. Do not restore.')
        receipt = {
            'format': 1, 'udid': UDID, 'source': str(source), 'removed_file_id': file_id,
            'removed_domain': DOMAIN, 'removed_path': RELATIVE,
            'source_entries': count, 'remaining_entries': count - 1,
            'manifest_sha256': digest(dest / 'Manifest.db'),
            'manifest_plist_sha256': digest(dest / 'Manifest.plist')
        }
        (output / RECEIPT).write_text(json.dumps(receipt, indent=2), encoding='utf-8')
        (output / 'INCOMPLETE_DO_NOT_RESTORE.txt').unlink()
    print('\nPREPARED: encrypted APN-excluded FULL backup copy')
    print('Output:', output)
    print('Removed entries: 1; all other Files rows unchanged')
    print('Encryption round-trip: OK; source metadata and APN payload unchanged')
    print('No device restore was performed.')
 
 
def restore(output, service):
    output = output.resolve()
    require(not (output / 'INCOMPLETE_DO_NOT_RESTORE.txt').exists(), 'Preparation is incomplete.')
    receipt = json.loads((output / RECEIPT).read_text(encoding='utf-8'))
    require(receipt.get('format') == 1 and receipt.get('udid') == UDID and
            receipt.get('removed_domain') == DOMAIN and receipt.get('removed_path') == RELATIVE,
            'Unexpected preparation receipt.')
    dest = output / UDID
    require(digest(dest / 'Manifest.db') == receipt['manifest_sha256'] and
            digest(dest / 'Manifest.plist') == receipt['manifest_plist_sha256'],
            'The prepared backup has changed. Stopped.')
    require(plist(dest / 'Manifest.plist').get('IsEncrypted') is True, 'Backup must remain encrypted.')
    require(not payload_path(dest, receipt['removed_file_id']).exists(), 'Excluded APN payload is present.')
    print('This command RESTORES DATA AND SETTINGS to the connected new iPhone.')
    print('Use after recovery/erase and minimal setup. APN repair is not guaranteed.')
    print('Backup:', output)
    password = getpass.getpass('Encrypted backup password (hidden): ')
    require(bool(password), 'Empty password. Stopped.')
    with tempfile.TemporaryDirectory(prefix='apn-restore-check-') as tmp:
        db_path = Path(tmp) / 'verify.db'
        service._decrypt_backup_manifest_db(dest, password, db_path)
        with closing(readonly(db_path)) as db:
            check_db(db)
            require(db.execute('SELECT COUNT(*) FROM Files').fetchone()[0] == receipt['remaining_entries'],
                    'Unexpected manifest row count.')
            require(db.execute('SELECT COUNT(*) FROM Files WHERE domain=? AND relativePath=?',
                               (DOMAIN, RELATIVE)).fetchone()[0] == 0, 'APN entry is still present.')
    # Password is passed only in this Python process, not the OS command line.
    sys.argv = ['pymobiledevice3', 'backup2', 'restore', '--udid', UDID,
                '--system', '--no-remove', '--reboot', '--password', password, str(output)]
    try:
        runpy.run_module('pymobiledevice3', run_name='__main__')
    finally:
        sys.argv = [sys.argv[0]]
 
 
def main():
    user = Path(os.environ.get('USERPROFILE', str(Path.home())))
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest='command', required=True)
    prep = sub.add_parser('prepare', help='Copy and edit the backup on this PC; no device access')
    prep.add_argument('--source', type=Path, default=user / 'Apple/MobileSync/Backup' / UDID)
    prep.add_argument('--output', type=Path, default=user / 'APN-Repair-Encrypted')
    rest = sub.add_parser('restore', help='Restore the prepared copy to the specified iPhone')
    rest.add_argument('--output', type=Path, default=user / 'APN-Repair-Encrypted')
    args = parser.parse_args()
    require(re.fullmatch(r'(?:[0-9A-Fa-f]{8}-[0-9A-Fa-f]{16}|[0-9A-Fa-f]{40})', UDID) is not None,
            'Set APN_TARGET_UDID to the target iPhone UDID before running this script.')
    service = backend()
    if args.command == 'prepare':
        require((args.source / 'Manifest.plist').is_file(), 'Source backup not found.')
        require(plist(args.source / 'Manifest.plist').get('IsEncrypted') is True,
                'Source is not encrypted. Do not recreate or overwrite your preserved backup.')
        password = getpass.getpass('Encrypted backup password (hidden): ')
        require(bool(password), 'Empty password. Stopped.')
        prepare(args.source, args.output, password, service)
    else:
        restore(args.output, service)
 
 
if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('\nInterrupted. Do not restore a copy without PREPARED status.', file=sys.stderr)
        sys.exit(130)
    except Exception as exc:
        print(f'ERROR: {type(exc).__name__}: {exc}', file=sys.stderr)
        print('Stopped. Check the error above. A restore error does not by itself mean the prepared copy is incomplete.', file=sys.stderr)
        sys.exit(1)