cloudhost/run-windows11.py

#!/usr/bin/env python3
"""Start the Windows 11 virtual machine."""

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 capture(command):
    """Return a command's stdout without the trailing newline, like $(...)."""
    result = subprocess.run(command, stdout=subprocess.PIPE, text=True)
    if result.returncode != 0:
        sys.exit(result.returncode)
    return result.stdout.rstrip("\n")


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


state_dir = Path(env("WIN11_VM_STATE_DIR") or script_dir / ".vm" / "windows11")
disk_size = env("WIN11_VM_DISK_SIZE") or "64G"
memory = env("WIN11_VM_MEMORY") or "8G"
cpus = env("WIN11_VM_CPUS") or "8"
rdp_port = env("WIN11_VM_RDP_PORT") or "3389"
ssh_port = env("WIN11_VM_SSH_PORT") or "2222"


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])


uefi_code = pick_firmware(
    env("WIN11_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("WIN11_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",
)
daemonize = False
install_from_iso = False
extra_args = []


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

If the system disk (DIR/disk.qcow2, see --state-dir) is missing, create it from
the Windows 11 QCOW2 base image.
Use --install-from-iso to perform an automatic ISO installation instead.
If it exists, start Windows normally without attaching installation media.

Options:
  --daemon            Run QEMU in the background
  --install-from-iso  Install from ISO when the system disk is missing
  --state-dir DIR     Directory for the VM disk, seed ISO and runtime files
                      (default: .vm/windows11; overrides WIN11_VM_STATE_DIR)
  -h, --help          Show this help

Environment variables:
  WIN11_VM_BASE_IMAGE   Windows 11 base QCOW2 image (default: beside shared images)
  WIN11_VM_INSTALL_ISO  Windows 11 installation ISO used with --install-from-iso
  WIN11_VM_TOOLS_ISO    VirtIO Windows tools ISO (default: resolved by shared image helper)
  WIN11_VM_MEMORY       Guest memory (default: 8G)
  WIN11_VM_CPUS         Virtual CPU count (default: 8)
  WIN11_VM_RDP_PORT     Host RDP port (default: 3389)
  WIN11_VM_SSH_PORT     Host SSH port (default: 2222)
  WIN11_VM_STATE_DIR    Runtime directory (default: .vm/windows11)
  WIN11_VM_DISK_IMAGE   System disk path
  WIN11_VM_DISK_SIZE    New system disk size (default: 64G)
  WIN11_VM_SEED_ISO     Answer-file ISO path (default: alongside the VM disk)
  WIN11_VM_UEFI_CODE    OVMF read-only firmware path (auto-detected if unset)
  WIN11_VM_UEFI_VARS    OVMF variable-store template path (auto-detected if unset)
  WIN11_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 == "--install-from-iso":
        install_from_iso = True
    elif arg == "--state-dir":
        if not args:
            die(f"{arg} requires a directory argument.", 2)
        state_dir = Path(args.pop(0))
    elif arg.startswith("--state-dir="):
        state_dir = Path(arg.split("=", 1)[1])
    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)

disk_image = Path(env("WIN11_VM_DISK_IMAGE") or state_dir / "disk.qcow2")
seed_iso = Path(env("WIN11_VM_SEED_ISO") or disk_image.parent / "seed.iso")
uefi_vars = state_dir / "OVMF_VARS_4M.fd"
pid_file = state_dir / "qemu.pid"
log_file = state_dir / "console.log"
monitor_socket = state_dir / "qemu-monitor.sock"
guest_agent_socket = state_dir / "qemu-guest-agent.sock"

installing = False
installing_from_iso = False
if not disk_image.is_file():
    installing = True
    installing_from_iso = install_from_iso
    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"
    if not is_executable(image_downloader):
        die(f"shared image helper is unavailable: {image_downloader}")
    if install_from_iso:
        install_iso = Path(env("WIN11_VM_INSTALL_ISO")
                           or capture([str(image_downloader), "--print-path", "windows11", "--iso"]))
        uses_shared_tools_iso = False
        if env("WIN11_VM_TOOLS_ISO"):
            tools_iso = Path(env("WIN11_VM_TOOLS_ISO"))
        else:
            tools_iso = Path(capture([str(image_downloader), "--print-path", "virtio-win"]))
            uses_shared_tools_iso = True
    else:
        base_image = Path(env("WIN11_VM_BASE_IMAGE")
                          or capture([str(image_downloader), "--print-path", "windows11"]))

if shutil.which("qemu-system-x86_64") is None:
    die("qemu-system-x86_64 is not installed.")
if installing:
    if shutil.which("qemu-img") is None:
        die("qemu-img is not installed.")
    if installing_from_iso:
        if not install_iso.is_file():
            die(f"Windows 11 ISO was not found: {install_iso}")
    elif not base_image.is_file():
        die(f"Windows 11 base QCOW2 image was not found: {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}")

state_dir.mkdir(parents=True, exist_ok=True)
if installing_from_iso:
    if not tools_iso.is_file():
        if uses_shared_tools_iso:
            run([str(image_downloader), "virtio-win"])
        else:
            die(f"VirtIO Windows tools ISO was not found: {tools_iso}")
    if not tools_iso.is_file():
        die(f"downloader did not create: {tools_iso}")

    seed_sources = [script_dir / "seediso" / "windows11" / name
                    for name in ("Autounattend.xml", "Configure-Windows.ps1")]

    def newer_than_seed(source):
        # Mirrors bash `-nt`: true if source exists and is newer than seed_iso.
        return source.exists() and source.stat().st_mtime > seed_iso.stat().st_mtime

    if not seed_iso.is_file() or any(newer_than_seed(s) for s in seed_sources):
        run([str(script_dir / "seediso" / "windows11" / "build-seed.sh"), str(seed_iso)])

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"Windows 11 VM is already running (PID {old_pid}).")
    pid_file.unlink(missing_ok=True)
monitor_socket.unlink(missing_ok=True)
guest_agent_socket.unlink(missing_ok=True)
if not uefi_vars.is_file():
    shutil.copyfile(uefi_vars_template, uefi_vars)
if installing:
    if installing_from_iso:
        run(["qemu-img", "create", "-q", "-f", "qcow2", str(disk_image), disk_size])
    else:
        run(["qemu-img", "create", "-q", "-f", "qcow2", "-F", "qcow2",
             "-b", str(base_image), str(disk_image), disk_size])
    print(f"Created Windows 11 system disk: {disk_image}")

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

qemu_args = [
    "-name", "windows11", "-machine", "q35", "-m", memory, "-smp", cpus,
    "-display", display_backend,
    # "-display", "none", "-spice", "port=5900,addr=127.0.0.1,disable-ticketing=on,agent-mouse=off", "-device", "virtio-vga",
    "-drive", f"if=pflash,format=raw,readonly=on,file={uefi_code}",
    "-drive", f"if=pflash,format=raw,file={uefi_vars}",
    "-drive", f"file={disk_image},if=none,id=system_disk,format=qcow2",
    "-device", "ide-hd,drive=system_disk,bus=ide.0,bootindex=1",
    "-nic", f"user,model=e1000e,hostfwd=tcp::{rdp_port}-:3389,hostfwd=tcp::{ssh_port}-:22",
    "-device", "virtio-serial-pci,id=virtio_serial",
    "-chardev", "qemu-vdagent,id=vdagent,name=vdagent,clipboard=on,mouse=off",
    "-device", "virtserialport,chardev=vdagent,name=com.redhat.spice.0",
    # Register the USB tablet after vdagent so QEMU selects it as the active pointer.
    "-device", "qemu-xhci,id=usb_controller", "-device", "usb-tablet,bus=usb_controller.0",
    "-chardev", f"socket,path={guest_agent_socket},server=on,wait=off,id=qga",
    "-device", "virtserialport,chardev=qga,name=org.qemu.guest_agent.0",
    "-monitor", f"unix:{monitor_socket},server=on,wait=off",
]
if installing_from_iso:
    qemu_args += [
        "-drive", f"file={install_iso},if=none,id=install_media,media=cdrom,readonly=on",
        "-device", "ide-cd,drive=install_media,bus=ide.1,bootindex=2",
        "-drive", f"file={seed_iso},if=none,id=answer_media,media=cdrom,readonly=on",
        "-device", "ide-cd,drive=answer_media,bus=ide.2,bootindex=3",
        "-drive", f"file={tools_iso},if=none,id=tools_media,media=cdrom,readonly=on",
        "-device", "ide-cd,drive=tools_media,bus=ide.3,bootindex=4",
    ]
if 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"
if installing_from_iso:
    print(f"Starting Windows 11 automatic installation with {accelerator} acceleration.")
elif installing:
    print(f"Starting Windows 11 from the QCOW2 base image with {accelerator} acceleration.")
else:
    print(f"Starting Windows 11 normally with {accelerator} acceleration.")
print(f"Account: guofang / password; RDP: localhost:{rdp_port}; SSH: localhost:{ssh_port}",
      flush=True)

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