cloudhost/run-debian13.py

#!/usr/bin/env python3
"""Start the Debian 13 virtual machine (Python port of run-debian13.sh)."""

import os
import platform
import shutil
import subprocess
import sys
from pathlib import Path

script_dir = Path(__file__).resolve().parent
env = os.environ.get


def die(message, code=1):
    print(f"Error: {message}", file=sys.stderr)
    sys.exit(code)


def run(command):
    """Run a command, exiting with its status on failure (like `set -e`)."""
    result = subprocess.run(command)
    if result.returncode != 0:
        sys.exit(result.returncode)


def is_executable(path):
    return path.is_file() and os.access(path, os.X_OK)


if env("VM_IMAGE_HELPER"):
    image_downloader = Path(env("VM_IMAGE_HELPER"))
elif is_executable(script_dir / "vm-images" / "download-images.sh"):
    image_downloader = script_dir / "vm-images" / "download-images.sh"
else:
    image_downloader = script_dir / ".." / "vm-images" / "download-images.sh"

uses_shared_base_image = False
if env("VM_BASE_IMAGE"):
    base_image = Path(env("VM_BASE_IMAGE"))
else:
    if not is_executable(image_downloader):
        die(f"shared image helper is unavailable: {image_downloader}")
    result = subprocess.run(
        [str(image_downloader), "--print-path", "debian13"],
        stdout=subprocess.PIPE,
        text=True,
    )
    if result.returncode != 0:
        sys.exit(result.returncode)
    base_image = Path(result.stdout.rstrip("\n"))
    uses_shared_base_image = True

state_dir = Path(env("VM_STATE_DIR") or script_dir / ".vm" / "debian13")
disk_image = Path(env("VM_DISK_IMAGE") or state_dir / "disk.qcow2")
seed_image = Path(env("VM_SEED_IMAGE") or disk_image.parent / "seed.iso")
disk_size = env("VM_DISK_SIZE") or "20G"
memory = env("VM_MEMORY") or "8G"
cpus = env("VM_CPUS") or "8"
ssh_port = env("VM_SSH_PORT") or "2222"


def normalize_arch(arch):
    if arch in ("arm64", "aarch64"):
        return "arm64"
    if arch in ("x86_64", "amd64"):
        return "amd64"
    die(f"unsupported architecture: {arch}")


native_arch = normalize_arch(platform.machine())
host_arch = normalize_arch(env("VM_IMAGE_ARCH") or native_arch)
if host_arch == "arm64":
    qemu_binary = "qemu-system-aarch64"
    machine_type = "virt"
else:
    qemu_binary = "qemu-system-x86_64"
    machine_type = "q35"


def pick_firmware(override, *candidates):
    if override:
        return Path(override)
    for candidate in candidates:
        if Path(candidate).is_file():
            return Path(candidate)
    return Path(candidates[0])


if host_arch == "arm64":
    uefi_code = pick_firmware(
        env("VM_UEFI_CODE"),
        "/usr/share/AAVMF/AAVMF_CODE.fd",
        "/usr/share/edk2/aarch64/QEMU_EFI.fd",
        "/opt/homebrew/share/qemu/edk2-aarch64-code.fd",
        "/usr/local/share/qemu/edk2-aarch64-code.fd",
    )
    uefi_vars_template = pick_firmware(
        env("VM_UEFI_VARS"),
        "/usr/share/AAVMF/AAVMF_VARS.fd",
        "/usr/share/edk2/aarch64/vars-template-pflash.raw",
        "/opt/homebrew/share/qemu/edk2-arm-vars.fd",
        "/usr/local/share/qemu/edk2-arm-vars.fd",
    )
else:
    uefi_code = pick_firmware(
        env("VM_UEFI_CODE"),
        "/usr/share/OVMF/OVMF_CODE_4M.fd",
        "/usr/share/edk2/ovmf/OVMF_CODE.fd",
        "/opt/homebrew/share/qemu/edk2-x86_64-code.fd",
        "/usr/local/share/qemu/edk2-x86_64-code.fd",
    )
    uefi_vars_template = pick_firmware(
        env("VM_UEFI_VARS"),
        "/usr/share/OVMF/OVMF_VARS_4M.fd",
        "/usr/share/edk2/ovmf/OVMF_VARS.fd",
        "/opt/homebrew/share/qemu/edk2-i386-vars.fd",
        "/usr/local/share/qemu/edk2-i386-vars.fd",
    )
uefi_vars = state_dir / "OVMF_VARS_4M.fd"
pid_file = state_dir / "qemu.pid"
log_file = state_dir / "console.log"
daemonize = False
extra_args = []


def usage(stream=sys.stdout):
    stream.write(f"""\
Usage: {Path(sys.argv[0]).name} [--daemon] [-- QEMU_ARGUMENT ...]

Start the Debian 13 virtual machine. A writable overlay is created under
.vm/debian13/, so the downloaded base image is never modified.

Options:
  --daemon  Run in the background while keeping the graphical window
  -h, --help  Show this help

Environment variables:
  VM_MEMORY      Guest memory (default: 8G)
  VM_CPUS        Virtual CPU count (default: 8)
  VM_SSH_PORT    Host port forwarded to guest SSH (default: 2222)
  VM_BASE_IMAGE  Base qcow2 image path (default: resolved by shared image helper)
  VM_SEED_IMAGE  cloud-init seed ISO path (default: alongside the VM disk)
  VM_STATE_DIR   Runtime state directory (default: .vm/debian13)
  VM_DISK_IMAGE  Writable overlay path
  VM_DISK_SIZE   New overlay virtual capacity (default: 20G)
  VM_IMAGE_ARCH  Guest/host architecture: arm64 or amd64 (default: auto-detected from the host)
  VM_UEFI_CODE   UEFI read-only firmware path (auto-detected for the selected architecture)
  VM_UEFI_VARS   UEFI variable-store template path (auto-detected for the selected architecture)
  VM_DISPLAY     QEMU display backend (default: cocoa on macOS, gtk elsewhere)
""")


args = sys.argv[1:]
while args:
    arg = args.pop(0)
    if arg == "--daemon":
        daemonize = True
    elif arg in ("-h", "--help"):
        usage()
        sys.exit(0)
    elif arg == "--":
        extra_args.extend(args)
        break
    else:
        print(f"Error: unknown option: {arg}", file=sys.stderr)
        usage(sys.stderr)
        sys.exit(2)

for command_name in (qemu_binary, "qemu-img"):
    if shutil.which(command_name) is None:
        die(f"{command_name} is not installed.")

if not base_image.is_file():
    if uses_shared_base_image:
        run([str(image_downloader), "debian13"])
    else:
        die(f"Debian base image was not found: {base_image}")
if not base_image.is_file():
    die(f"downloader did not create: {base_image}")
for firmware_file in (uefi_code, uefi_vars_template):
    if not firmware_file.is_file():
        die(f"required UEFI firmware was not found: {firmware_file}")

if not seed_image.is_file():
    seed_builder = script_dir / "seediso" / "debian13" / "build-seed.sh"
    if not is_executable(seed_builder):
        die(f"seed image is missing and builder is unavailable: {seed_builder}")
    print(f"Seed image not found; building {seed_image}...", flush=True)
    run([str(seed_builder), str(seed_image)])

state_dir.mkdir(parents=True, exist_ok=True)

# QEMU requires the code and vars pflash images to be exactly the same size.
# Homebrew's edk2 firmware ships code/vars pairs at mismatched sizes (true for
# both the aarch64 and x86_64 pairs), so pad per-VM copies of whichever one is
# smaller up to match the other.
uefi_code_flash = state_dir / "OVMF_CODE_4M.fd"
if not uefi_code_flash.is_file() or not uefi_vars.is_file():
    flash_size = max(uefi_code.stat().st_size, uefi_vars_template.stat().st_size)
    for source, target in ((uefi_code, uefi_code_flash), (uefi_vars_template, uefi_vars)):
        if not target.is_file():
            shutil.copyfile(source, target)
            os.truncate(target, flash_size)

if pid_file.is_file():
    old_pid = pid_file.read_text().strip()
    if old_pid.isdigit():
        try:
            os.kill(int(old_pid), 0)
        except OSError:
            pass
        else:
            die(f"the virtual machine is already running (PID {old_pid}).")
    pid_file.unlink(missing_ok=True)

if not disk_image.is_file():
    run(["qemu-img", "create", "-q", "-f", "qcow2", "-F", "qcow2",
         "-b", str(base_image), str(disk_image), disk_size])
    print(f"Created writable disk overlay: {disk_image}")

is_macos = platform.system() == "Darwin"
display_backend = env("VM_DISPLAY") or ("cocoa" if is_macos else "gtk")

qemu_args = [
    "-name", "debian13",
    "-machine", machine_type,
    "-display", display_backend,
    "-m", memory,
    "-smp", cpus,
    "-drive", f"if=pflash,format=raw,readonly=on,file={uefi_code_flash}",
    "-drive", f"if=pflash,format=raw,file={uefi_vars}",
    "-drive", f"file={disk_image},if=none,id=disk0,format=qcow2",
    "-device", "virtio-blk-pci,drive=disk0",
    "-drive", f"file={seed_image},if=none,id=seed0,format=raw,readonly=on",
    "-device", "virtio-blk-pci,drive=seed0",
    "-nic", f"user,model=virtio-net-pci,hostfwd=tcp::{ssh_port}-:22",
]

if machine_type == "virt":
    # Unlike q35, the "virt" machine has no default GPU or USB input devices.
    qemu_args += [
        "-device", "virtio-gpu-pci",
        "-device", "qemu-xhci",
        "-device", "usb-kbd",
        "-device", "usb-tablet",
    ]

if is_macos and host_arch == native_arch:
    qemu_args += ["-accel", "hvf", "-cpu", "host"]
    accelerator = "HVF"
elif host_arch == native_arch and os.access("/dev/kvm", os.R_OK | os.W_OK):
    qemu_args += ["-enable-kvm", "-cpu", "host"]
    accelerator = "KVM"
else:
    qemu_args += ["-accel", "tcg", "-cpu", "max"]
    accelerator = "TCG"

print(f"Starting {host_arch} VM with {accelerator} acceleration; "
      f"SSH is forwarded to localhost:{ssh_port}.", flush=True)

if daemonize:
    log_file.write_text("")
    qemu_args += [
        "-serial", f"file:{log_file}",
        "-monitor", "none",
        "-daemonize",
        "-pidfile", str(pid_file),
    ]
    run([qemu_binary, *qemu_args, *extra_args])
    print(f"VM started (PID {pid_file.read_text().strip()}); console log: {log_file}")
else:
    os.execvp(qemu_binary, [qemu_binary, *qemu_args, *extra_args])