Install Ventoy on a USB drive from macOS - no VM, no macFUSE, no Linux required. Writes GPT + boot code directly via Python.

ventoy-macos-install.py

#!/usr/bin/env python3
"""
ventoy-macos-install.py - Install Ventoy on a USB drive from macOS.

This script installs Ventoy on a USB drive without requiring macFUSE,
Linux, or a VM. It writes the GPT partition table, boot code, and
EFI partition image directly to the raw block device.

Usage:
    sudo python3 ventoy-macos-install.py /dev/diskN [--exfat|--ntfs] [--ventoy-version VERSION]

Requirements:
    - macOS (tested on macOS 13+)
    - Python 3.8+
    - xz (install via: brew install xz)
    - Root privileges (sudo)
    - An external USB drive

What it does:
    1. Downloads the specified Ventoy release (default: latest)
    2. Decompresses boot images
    3. Creates a GPT partition table with:
       - Partition 1: exFAT (default) or NTFS, starting at sector 2048 (1MB)
       - Partition 2: FAT16, 32MB, at the end of the disk (Ventoy EFI)
    4. Writes Ventoy boot code (boot.img, core.img) to the GPT gap
    5. Writes the Ventoy EFI partition image (ventoy.disk.img)

License: MIT
"""

import argparse
import json
import os
import struct
import subprocess
import sys
import tempfile
import time
import urllib.request
import uuid
import zlib

SECTOR_SIZE = 512
VENTOY_SECTOR_NUM = 65536  # 32MB for EFI partition
VENTOY_GITHUB_REPO = "ventoy/Ventoy"

# GPT Partition Type GUID for "Microsoft Basic Data" (FAT, NTFS, exFAT, etc.)
# Ventoy uses this for both partitions instead of the EFI System Partition type.
GPT_BASIC_DATA_GUID = uuid.UUID("EBD0A0A2-B9E5-4433-87C0-68B6B72699C7")


def die(msg):
    print(f"\nERROR: {msg}", file=sys.stderr)
    sys.exit(1)


def run(cmd, check=True, capture=True, timeout=30):
    result = subprocess.run(cmd, capture_output=capture, text=True, timeout=timeout)
    if check and result.returncode != 0:
        stderr = result.stderr if capture else ""
        die(f"Command failed: {' '.join(cmd)}\n{stderr}")
    return result


def confirm(msg):
    answer = input(f"\n{msg} (y/N): ").strip().lower()
    if answer != "y":
        print("Aborted.")
        sys.exit(0)


# ── Ventoy download ──────────────────────────────────────────────────


def get_latest_version():
    url = f"https://api.github.com/repos/{VENTOY_GITHUB_REPO}/releases/latest"
    print(f"Fetching latest Ventoy version from GitHub...")
    req = urllib.request.Request(url, headers={"User-Agent": "ventoy-macos-install"})
    with urllib.request.urlopen(req, timeout=15) as resp:
        data = json.loads(resp.read())
    tag = data["tag_name"].lstrip("v")
    return tag


def download_ventoy(version, workdir):
    tarball = f"ventoy-{version}-linux.tar.gz"
    url = f"https://github.com/{VENTOY_GITHUB_REPO}/releases/download/v{version}/{tarball}"
    dest = os.path.join(workdir, tarball)

    if os.path.exists(dest):
        print(f"Using cached {tarball}")
        return dest

    print(f"Downloading {url} ...")
    urllib.request.urlretrieve(url, dest)
    print(f"Downloaded {tarball} ({os.path.getsize(dest) / (1024 * 1024):.1f} MB)")
    return dest


def extract_ventoy(tarball, workdir):
    print("Extracting Ventoy package...")
    run(["tar", "xzf", tarball, "-C", workdir])
    # Find the extracted directory
    for entry in os.listdir(workdir):
        if entry.startswith("ventoy-") and os.path.isdir(os.path.join(workdir, entry)):
            return os.path.join(workdir, entry)
    die("Could not find extracted Ventoy directory")


def decompress_images(ventoy_dir, workdir):
    boot_img = os.path.join(ventoy_dir, "boot", "boot.img")
    core_xz = os.path.join(ventoy_dir, "boot", "core.img.xz")
    disk_xz = os.path.join(ventoy_dir, "ventoy", "ventoy.disk.img.xz")

    for f in [boot_img, core_xz, disk_xz]:
        if not os.path.exists(f):
            die(f"Missing file: {f}")

    core_img = os.path.join(workdir, "core.img")
    disk_img = os.path.join(workdir, "ventoy.disk.img")

    if not os.path.exists(core_img):
        print("Decompressing core.img...")
        run(["xzcat", core_xz], check=False, capture=False)
        with open(core_img, "wb") as out:
            result = subprocess.run(["xzcat", core_xz], stdout=out, timeout=30)
            if result.returncode != 0:
                die(
                    "Failed to decompress core.img.xz (is xz installed? brew install xz)"
                )

    if not os.path.exists(disk_img):
        print("Decompressing ventoy.disk.img...")
        with open(disk_img, "wb") as out:
            result = subprocess.run(["xzcat", disk_xz], stdout=out, timeout=60)
            if result.returncode != 0:
                die("Failed to decompress ventoy.disk.img.xz")

    return boot_img, core_img, disk_img


# ── Disk operations ──────────────────────────────────────────────────


def get_disk_info(disk):
    """Get disk size in sectors and verify it's external."""
    result = run(["diskutil", "info", disk])
    output = result.stdout

    if "External" not in output and "Removable" not in output:
        die(
            f"{disk} does not appear to be an external/removable disk. Refusing to proceed."
        )

    for line in output.split("\n"):
        if "Disk Size" in line:
            # Extract byte count from format like: "256.1 GB (256060514304 Bytes)"
            try:
                bytes_str = line.split("(")[1].split("Bytes")[0].strip()
                total_bytes = int(bytes_str)
                return total_bytes // SECTOR_SIZE
            except (IndexError, ValueError):
                pass

    die(f"Could not determine size of {disk}")


def calculate_layout(disk_sectors):
    """Calculate Ventoy-compatible partition layout."""
    part1_start = 2048  # 1MB - Ventoy requirement
    part1_end = disk_sectors - VENTOY_SECTOR_NUM - 34

    part2_start = part1_end + 1
    mod = part2_start % 8
    if mod > 0:
        part1_end -= mod
        part2_start = part1_end + 1

    part2_end = part2_start + VENTOY_SECTOR_NUM - 1

    return {
        "part1_start": part1_start,
        "part1_end": part1_end,
        "part1_sectors": part1_end - part1_start + 1,
        "part2_start": part2_start,
        "part2_end": part2_end,
        "part2_sectors": part2_end - part2_start + 1,
    }


# ── GPT construction ─────────────────────────────────────────────────


def uuid_to_mixed_endian(u):
    b = u.bytes
    return b[3::-1] + b[5:3:-1] + b[7:5:-1] + b[8:16]


def make_gpt_entry(type_guid, unique_guid, start, end, attrs, name):
    t = uuid_to_mixed_endian(type_guid)
    u = uuid_to_mixed_endian(unique_guid)
    n = name.encode("utf-16-le")
    n += b"\x00" * (72 - len(n))
    return struct.pack("<16s16sQQQ72s", t, u, start, end, attrs, n)


def make_gpt_header(params):
    data = struct.pack(
        "<8sIIIIQQQQ16sQIII",
        b"EFI PART",
        0x00010000,
        92,
        0,  # CRC placeholder
        0,
        params["my_lba"],
        params["alt_lba"],
        params["first_usable"],
        params["last_usable"],
        uuid_to_mixed_endian(params["disk_guid"]),
        params["entry_start"],
        params["num_entries"],
        params["entry_size"],
        params["entries_crc"],
    )
    crc = zlib.crc32(data) & 0xFFFFFFFF
    data = data[:16] + struct.pack("<I", crc) + data[20:]
    return data + b"\x00" * (SECTOR_SIZE - len(data))


def build_gpt(disk_sectors, layout):
    """Build complete GPT structures."""
    disk_guid = uuid.uuid4()

    e1 = make_gpt_entry(
        GPT_BASIC_DATA_GUID,
        uuid.uuid4(),
        layout["part1_start"],
        layout["part1_end"],
        0,
        "Ventoy",
    )
    e2 = make_gpt_entry(
        GPT_BASIC_DATA_GUID,
        uuid.uuid4(),
        layout["part2_start"],
        layout["part2_end"],
        0,
        "VTOYEFI",
    )
    entries = e1 + e2 + b"\x00" * (128 * 128 - 256)
    entries_crc = zlib.crc32(entries) & 0xFFFFFFFF

    common = {
        "disk_guid": disk_guid,
        "first_usable": 2048,
        "last_usable": disk_sectors - 34,
        "num_entries": 128,
        "entry_size": 128,
        "entries_crc": entries_crc,
    }

    primary = make_gpt_header(
        {**common, "my_lba": 1, "alt_lba": disk_sectors - 1, "entry_start": 2}
    )
    backup = make_gpt_header(
        {
            **common,
            "my_lba": disk_sectors - 1,
            "alt_lba": 1,
            "entry_start": disk_sectors - 33,
        }
    )

    # Protective MBR
    mbr = bytearray(512)
    mbr[446:447] = b"\x00"
    mbr[448] = 0x02
    mbr[450] = 0xEE
    mbr[451] = mbr[452] = mbr[453] = 0xFF
    struct.pack_into("<I", mbr, 454, 1)
    struct.pack_into("<I", mbr, 458, min(disk_sectors - 1, 0xFFFFFFFF))
    mbr[510] = 0x55
    mbr[511] = 0xAA

    return bytes(mbr), primary, entries, backup


def patch_sector(fd, sector_lba, offset, data):
    """Read-modify-write a sector to patch sub-sector bytes."""
    os.lseek(fd, sector_lba * SECTOR_SIZE, os.SEEK_SET)
    sector = bytearray(os.read(fd, SECTOR_SIZE))
    sector[offset : offset + len(data)] = data
    os.lseek(fd, sector_lba * SECTOR_SIZE, os.SEEK_SET)
    os.write(fd, bytes(sector))


def write_to_disk(
    raw_device,
    disk,
    disk_sectors,
    layout,
    mbr,
    primary,
    entries,
    backup,
    boot_img_path,
    core_img_path,
    disk_img_path,
):
    """Write GPT and Ventoy boot code to the disk."""

    boot_img = open(boot_img_path, "rb").read()
    core_img = open(core_img_path, "rb").read()
    disk_img = open(disk_img_path, "rb").read()

    # Unmount
    print("Unmounting disk...")
    run(["diskutil", "unmountDisk", "force", disk], check=False)
    time.sleep(2)

    print("Writing to disk...")
    fd = os.open(raw_device, os.O_RDWR)
    try:
        # Zero first 1MB (protective MBR + GPT header + entries area)
        print("  Zeroing first 1MB...")
        os.lseek(fd, 0, os.SEEK_SET)
        os.write(fd, b"\x00" * (2048 * SECTOR_SIZE))

        # Zero backup GPT area
        print("  Zeroing backup GPT area...")
        os.lseek(fd, (disk_sectors - 33) * SECTOR_SIZE, os.SEEK_SET)
        os.write(fd, b"\x00" * (33 * SECTOR_SIZE))

        # Write protective MBR
        print("  Writing protective MBR...")
        os.lseek(fd, 0, os.SEEK_SET)
        os.write(fd, mbr)

        # Write primary GPT header (sector 1)
        print("  Writing primary GPT header...")
        os.lseek(fd, SECTOR_SIZE, os.SEEK_SET)
        os.write(fd, primary)

        # Write primary GPT entries (sectors 2-33)
        print("  Writing GPT entries...")
        os.lseek(fd, 2 * SECTOR_SIZE, os.SEEK_SET)
        os.write(fd, entries)

        # Write backup GPT entries + header
        print("  Writing backup GPT...")
        os.lseek(fd, (disk_sectors - 33) * SECTOR_SIZE, os.SEEK_SET)
        os.write(fd, entries)
        os.lseek(fd, (disk_sectors - 1) * SECTOR_SIZE, os.SEEK_SET)
        os.write(fd, backup)

        # Write Ventoy boot.img (446 bytes of BIOS boot code to MBR)
        print("  Writing Ventoy boot.img...")
        patch_sector(fd, 0, 0, boot_img[:446])

        # GPT marker at offset 92
        patch_sector(fd, 0, 92, b"\x22")

        # Write core.img to sectors 34-2047 (GPT gap area)
        print("  Writing core.img...")
        os.lseek(fd, 34 * SECTOR_SIZE, os.SEEK_SET)
        core = core_img[: 2014 * SECTOR_SIZE]
        if len(core) % SECTOR_SIZE:
            core += b"\x00" * (SECTOR_SIZE - len(core) % SECTOR_SIZE)
        os.write(fd, core)

        # Second GPT marker at offset 17908
        patch_sector(fd, 17908 // SECTOR_SIZE, 17908 % SECTOR_SIZE, b"\x23")

        # Write ventoy.disk.img to partition 2
        part2_start = layout["part2_start"]
        print(f"  Writing ventoy.disk.img at sector {part2_start}...")
        os.lseek(fd, part2_start * SECTOR_SIZE, os.SEEK_SET)
        os.write(fd, disk_img)

        # Disk UUID at offset 384
        patch_sector(fd, 0, 384, uuid.uuid4().bytes)

        # Disk signature at offset 440
        patch_sector(fd, 0, 440, os.urandom(4))

        os.fsync(fd)
        print("  All writes complete.")
    finally:
        os.close(fd)


def format_partition1(disk, fs_type="exfat"):
    """Format partition 1 with the specified filesystem."""
    part1 = f"{disk}s1"

    print(f"\nWaiting for macOS to detect partitions...")
    time.sleep(3)

    run(["diskutil", "unmountDisk", "force", disk], check=False)
    time.sleep(1)

    if fs_type == "exfat":
        print(f"Formatting {part1} as exFAT...")
        result = run(["newfs_exfat", "-v", "Ventoy", part1], check=False)
        if result.returncode != 0:
            print("  newfs_exfat failed, trying diskutil...")
            run(["diskutil", "eraseVolume", "ExFAT", "Ventoy", part1])
    else:
        # NTFS is not natively supported for formatting on macOS
        # but Ventoy data partition can remain unformatted and be formatted elsewhere
        print(f"Warning: macOS cannot natively format NTFS.")
        print(f"Formatting {part1} as exFAT instead (readable on Windows/Linux/macOS).")
        result = run(["newfs_exfat", "-v", "Ventoy", part1], check=False)
        if result.returncode != 0:
            run(["diskutil", "eraseVolume", "ExFAT", "Ventoy", part1])


def verify(disk):
    """Print final disk layout and verify partition 1 offset."""
    time.sleep(2)
    run(["diskutil", "mountDisk", disk], check=False)
    time.sleep(1)

    print("\n" + "=" * 50)
    print("Disk layout:")
    print("=" * 50)
    run(["diskutil", "list", disk], capture=False)

    result = run(["diskutil", "info", f"{disk}s1"])
    for line in result.stdout.split("\n"):
        if "Offset" in line:
            print(f"\n{line.strip()}")
            if "2048" in line:
                print("  -> Correct! Partition 1 starts at sector 2048 (1MB)")
            else:
                print("  -> WARNING: Partition 1 does not start at sector 2048!")


# ── Main ─────────────────────────────────────────────────────────────


def main():
    parser = argparse.ArgumentParser(
        description="Install Ventoy on a USB drive from macOS.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
    sudo python3 ventoy-macos-install.py /dev/disk4
    sudo python3 ventoy-macos-install.py /dev/disk4 --ventoy-version 1.1.10
    sudo python3 ventoy-macos-install.py /dev/disk4 --exfat
        """,
    )
    parser.add_argument("disk", help="Target disk device (e.g., /dev/disk4)")
    parser.add_argument(
        "--exfat",
        action="store_const",
        const="exfat",
        dest="fs_type",
        default="exfat",
        help="Format data partition as exFAT (default)",
    )
    parser.add_argument(
        "--ventoy-version",
        default=None,
        help="Ventoy version to install (default: latest)",
    )
    parser.add_argument(
        "--work-dir",
        default=None,
        help="Working directory for downloads (default: temp directory)",
    )

    args = parser.parse_args()

    disk = args.disk
    raw_device = disk.replace("/dev/disk", "/dev/rdisk")

    if os.geteuid() != 0:
        die(
            "This script must be run as root. Use: sudo python3 ventoy-macos-install.py ..."
        )

    if sys.platform != "darwin":
        die("This script is designed for macOS only.")

    # Check xz is available
    if subprocess.run(["which", "xzcat"], capture_output=True).returncode != 0:
        die("xz is required but not found. Install it with: brew install xz")

    # Validate disk
    if not disk.startswith("/dev/disk"):
        die(f"Invalid disk device: {disk}")

    if disk in ("/dev/disk0", "/dev/disk1"):
        die("Refusing to operate on disk0/disk1 (likely your system disk).")

    # Get disk info
    print(f"Target disk: {disk}")
    disk_sectors = get_disk_info(disk)
    disk_gb = (disk_sectors * SECTOR_SIZE) / (1024**3)
    print(f"Disk size: {disk_gb:.1f} GiB ({disk_sectors} sectors)")

    # Show current layout
    print(f"\nCurrent layout:")
    run(["diskutil", "list", disk], capture=False)

    # Calculate partition layout
    layout = calculate_layout(disk_sectors)
    p1_gb = (layout["part1_sectors"] * SECTOR_SIZE) / (1024**3)
    p2_mb = (layout["part2_sectors"] * SECTOR_SIZE) / (1024**2)

    print(f"\nPlanned Ventoy layout:")
    print(
        f"  Part 1 (Ventoy, exFAT):  {p1_gb:.1f} GiB  [sectors {layout['part1_start']}-{layout['part1_end']}]"
    )
    print(
        f"  Part 2 (VTOYEFI, FAT16): {p2_mb:.0f} MiB   [sectors {layout['part2_start']}-{layout['part2_end']}]"
    )

    confirm("ALL DATA ON THIS DISK WILL BE DESTROYED. Continue?")

    # Setup working directory
    if args.work_dir:
        workdir = args.work_dir
        os.makedirs(workdir, exist_ok=True)
    else:
        workdir = tempfile.mkdtemp(prefix="ventoy-macos-")

    print(f"\nWorking directory: {workdir}")

    # Get Ventoy version
    version = args.ventoy_version or get_latest_version()
    print(f"Ventoy version: {version}")

    # Download and extract
    tarball = download_ventoy(version, workdir)
    ventoy_dir = extract_ventoy(tarball, workdir)
    boot_img, core_img, disk_img = decompress_images(ventoy_dir, workdir)

    print(f"\nBoot images ready:")
    print(f"  boot.img:         {os.path.getsize(boot_img)} bytes")
    print(f"  core.img:         {os.path.getsize(core_img)} bytes")
    print(f"  ventoy.disk.img:  {os.path.getsize(disk_img)} bytes")

    # Build GPT
    print("\nBuilding GPT partition table...")
    mbr, primary, entries, backup = build_gpt(disk_sectors, layout)

    # Final confirmation
    confirm("Ready to write. This is your last chance to abort. Proceed?")

    # Write everything
    write_to_disk(
        raw_device,
        disk,
        disk_sectors,
        layout,
        mbr,
        primary,
        entries,
        backup,
        boot_img,
        core_img,
        disk_img,
    )

    # Format partition 1
    format_partition1(disk, args.fs_type)

    # Verify
    verify(disk)

    print(f"\n{'=' * 50}")
    print(f"Ventoy {version} installed successfully!")
    print(f"{'=' * 50}")
    print(f"\nYou can now copy ISO/WIM/IMG/VHD files to the 'Ventoy' partition.")
    print(f"Boot from the USB drive and Ventoy will list all bootable images.")


if __name__ == "__main__":
    main()