#!/usr/bin/env python3 """Start the Windows 11 virtual machine.""" import os import platform import shutil import socket import string 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 port_in_use(port): """True if a host TCP port cannot be bound, as QEMU does for a hostfwd rule.""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: probe.bind(("", int(port))) except OSError: return True return False def is_executable(path): return path.is_file() and os.access(path, os.X_OK) def find_image_downloader(): 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}") return image_downloader state_dir = Path(env("WIN11_VM_STATE_DIR") or script_dir / ".vm" / "windows11") disk_size = env("WIN11_VM_DISK_SIZE") or "64G" rdp_port = env("WIN11_VM_RDP_PORT") or "3389" ssh_port = env("WIN11_VM_SSH_PORT") or "2222" qemu_config_template = Path( env("WIN11_VM_QEMU_CONFIG") or script_dir / "qemu" / "windows11.ini") qemu_arch_config = script_dir / "qemu" / "windows11-amd64.ini" 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 download_only = False extra_args = [] def usage(stream=sys.stdout): stream.write(f"""\ Usage: {Path(sys.argv[0]).name} [--daemon] [--download-only] [--install-from-iso] [--state-dir DIR] [--rdp-port PORT] [--ssh-port PORT] [-- QEMU_ARGUMENT ...] If the system disk (DIR/disk.qcow2, see --state-dir) is missing, create it from the Windows 11 QCOW2 base image. The VM hardware (accelerator, memory, CPUs, drives, devices) is defined in qemu/windows11.ini, which is rendered to qemu.ini in the state directory and loaded with QEMU's -readconfig, followed by qemu/windows11-amd64.ini. Later QEMU arguments override them, e.g. -- -m 4G -smp 4. 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 --download-only Download the VirtIO tools ISO if it is missing, then exit (the Windows ISO and base QCOW2 are supplied manually) --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) --rdp-port PORT Host port forwarded to guest RDP (default: 3389; overrides WIN11_VM_RDP_PORT) --ssh-port PORT Host port forwarded to guest SSH (default: 2222; overrides WIN11_VM_SSH_PORT) -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_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_QEMU_CONFIG QEMU config template passed to -readconfig (default: qemu/windows11.ini) 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 == "--download-only": download_only = 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 ("--rdp-port", "--ssh-port") or arg.startswith(("--rdp-port=", "--ssh-port=")): name, sep, value = arg.partition("=") if not sep: if not args: die(f"{name} requires a port argument.", 2) value = args.pop(0) if not (value.isascii() and value.isdigit() and 1 <= int(value) <= 65535): die(f"{name} must be a port number from 1 to 65535: {value}", 2) if name == "--rdp-port": rdp_port = value else: ssh_port = value 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) if download_only: # The VirtIO tools ISO is the only medium the launcher downloads itself. if env("WIN11_VM_TOOLS_ISO"): tools_iso = Path(env("WIN11_VM_TOOLS_ISO")) if not tools_iso.is_file(): die(f"VirtIO Windows tools ISO was not found: {tools_iso}") else: image_downloader = find_image_downloader() tools_iso = Path(capture([str(image_downloader), "--print-path", "virtio-win"])) if not tools_iso.is_file(): run([str(image_downloader), "virtio-win"]) if not tools_iso.is_file(): die(f"downloader did not create: {tools_iso}") print(f"VirtIO Windows tools ISO is ready: {tools_iso}") sys.exit(0) 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" qemu_config = state_dir / "qemu.ini" installing = False installing_from_iso = False if not disk_image.is_file(): installing = True installing_from_iso = install_from_iso image_downloader = find_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") # Most of the VM definition lives in the QEMU config template; fill in the # file paths and hand the rendered file to QEMU with -readconfig. config_values = { "uefi_code": uefi_code, "uefi_vars": uefi_vars, "disk_image": disk_image, "guest_agent_socket": guest_agent_socket, "monitor_socket": monitor_socket, } for key, value in config_values.items(): # QEMU's config file parser has no escape syntax for quoted values. if any(char in str(value) for char in '"\n'): die(f"{key} must not contain double quotes or newlines: {value}") for config_file in (qemu_config_template, qemu_arch_config): if not config_file.is_file(): die(f"QEMU config file was not found: {config_file}") try: qemu_config_text = string.Template(qemu_config_template.read_text()).substitute( {key: str(value) for key, value in config_values.items()}) except (KeyError, ValueError) as error: die(f"invalid placeholder in {qemu_config_template}: {error}") qemu_config.write_text(qemu_config_text) # QEMU refuses to start when a forwarded host port is taken, so warn and leave # that forward out instead. forwards = [] forwarded_ports = [] for label, host_port, guest_port in (("RDP", rdp_port, 3389), ("SSH", ssh_port, 22)): in_use = port_in_use(host_port) forwarded_ports.append((label, host_port, not in_use)) if in_use: print(f"Warning: host port {host_port} is already in use; " f"{label} will not be forwarded to the guest.", file=sys.stderr) else: forwards.append(f"hostfwd=tcp::{host_port}-:{guest_port}") qemu_args = [ "-readconfig", str(qemu_config), "-readconfig", str(qemu_arch_config), # "max" is the host CPU under KVM/HVF and the fullest model under TCG. "-cpu", "max", # A config file section cannot hold two hostfwd entries, so the network # that the NIC in the config file uses is defined here. -nic none stops # QEMU from adding its default NIC as well. "-netdev", ",".join(["user,id=net0", *forwards]), "-nic", "none", "-display", display_backend, # "-display", "none", "-spice", "port=5900,addr=127.0.0.1,disable-ticketing=on,agent-mouse=off", "-device", "virtio-vga", ] 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 installing_from_iso: print("Starting Windows 11 automatic installation.") elif installing: print("Starting Windows 11 from the QCOW2 base image.") else: print("Starting Windows 11 normally.") print("Account: guofang / password; " + "; ".join( f"{label}: localhost:{port}" for label, port, forwarded in forwarded_ports if forwarded), 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])