tools/install_bundle.py

#!/usr/bin/env python3
"""Install a verified, already-downloaded snapshot; inference stays offline."""
import argparse
import hashlib
import json
import platform
import shutil
import tarfile
import tempfile
from pathlib import Path


def digest(path):
    with path.open('rb') as stream:
        return hashlib.file_digest(stream, 'sha256').hexdigest()


def verify(root):
    manifest = json.loads((root / 'manifest.json').read_text())
    for name, metadata in manifest['files'].items():
        path = root / name
        if Path(name).is_absolute() or '..' in Path(name).parts:
            raise ValueError('Unsafe manifest path')
        if not path.is_file() or path.stat().st_size != metadata['bytes'] or digest(path) != metadata['sha256']:
            raise ValueError(f'Integrity check failed: {name}')
    return manifest


def install(snapshot, destination):
    verify(snapshot)
    if platform.system() != 'Darwin' or platform.machine() not in ('arm64', 'aarch64'):
        raise ValueError('This release archive is qualified only for macOS ARM64; see PLATFORMS.md')
    version = platform.mac_ver()[0]
    if version and tuple(map(int, version.split('.')[:2])) < (14, 0):
        raise ValueError('The preserved dylib requires macOS 14.0 or newer')
    if destination.exists():
        raise ValueError('Destination already exists; select a new directory')
    destination.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(dir=destination.parent) as temporary:
        stage = Path(temporary) / 'bundle'
        stage.mkdir()
        with tarfile.open(snapshot / 'archives/datafog-pii-v0.1.0-macos-arm64.tar.gz') as archive:
            for member in archive.getmembers():
                if member.name.startswith('/') or '..' in Path(member.name).parts or not (member.isfile() or member.isdir()):
                    raise ValueError('Unsafe archive member')
            archive.extractall(stage, filter='data')
        (stage / 'datafog-pii').chmod(0o755)
        # Validate extracted content independently, including executable and notices.
        checksums = json.loads((stage / 'checksums.json').read_text())
        for name, expected in checksums.items():
            if digest(stage / name) != expected:
                raise ValueError(f'Extracted bundle checksum failed: {name}')
        stage.rename(destination)
    return destination


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--snapshot', required=True, type=Path)
    parser.add_argument('--destination', required=True, type=Path)
    args = parser.parse_args()
    print(install(args.snapshot.resolve(), args.destination.resolve()))