Check for outdated VirtIO drivers on Windows VMs in OpenShift Virtualization

virtio-check.py

#!/usr/bin/env python3
"""Check for outdated VirtIO drivers on Windows VMs in OpenShift Virtualization.

Compares installed driver versions (queried from the QEMU guest agent running
inside each VM) against a known-good baseline version and reports which drivers
are outdated.

How it works:

  1. For each target VM, the script finds the corresponding virt-launcher pod
     using the `kubevirt.io/domain=<vm-name>` label.

  2. It runs `virsh qemu-agent-command` inside that pod to invoke the guest
     agent's `guest-get-devices` command, which returns every installed device
     driver including name, version, and PCI hardware IDs.

  3. It filters to VirtIO devices only (PCI vendor 0x1AF4) and compares each
     driver's version against the provided baseline.

The baseline version is the driver version shipped in the virtio-win container
disk for your OCP Virtualization release. To determine it:

  CID=$(podman create registry.redhat.io/container-native-virtualization/virtio-win-rhel9:v4.22 true)
  podman cp $CID:/disk /tmp/virtio-disk
  podman rm $CID
  isoinfo -i /tmp/virtio-disk/virtio-win.iso -R -x "/viostor/2k22/amd64/viostor.inf" | grep DriverVer

Prerequisites:
  - oc (or kubectl) with access to the cluster
  - python3 (3.6+)
  - QEMU guest agent installed and running in the target Windows VM(s)

Examples:
  python3 virtio-check.py win2022 --baseline-version 100.103.104.29700
  python3 virtio-check.py all -n all --baseline-version 100.103.104.29700
  python3 virtio-check.py all -n all --baseline-version 100.103.104.29700 --json
"""

import argparse
import json
import subprocess
import sys

# PCI vendor ID for all VirtIO devices (Red Hat, Inc.)
VIRTIO_VENDOR_ID = 0x1AF4


# ---------------------------------------------------------------------------
# Cluster interaction
# ---------------------------------------------------------------------------

def oc(*args):
    """Run an oc command and return its stdout, or empty string on failure."""
    result = subprocess.run(["oc", *args], capture_output=True, text=True, check=False)
    return result.stdout.strip()


def find_virt_launcher_pod(vm_name, namespace):
    """Find the virt-launcher pod for a given VM by its label.

    KubeVirt labels every virt-launcher pod with kubevirt.io/domain=<vm-name>.
    Returns the pod name (e.g. "pod/virt-launcher-win2022-abc123") or None.
    """
    pod = oc("get", "pod", "-l", f"kubevirt.io/domain={vm_name}",
             "-n", namespace, "-o", "name")
    return pod or None


def list_running_vmis(namespace):
    """List all running VirtualMachineInstances as (namespace, name) tuples.

    If namespace is "all", searches across all namespaces.
    """
    ns_flag = ["--all-namespaces"] if namespace == "all" else ["-n", namespace]
    raw = oc("get", "vmi", *ns_flag,
             "-o", "jsonpath={range .items[*]}{.metadata.namespace}/{.metadata.name} {end}")
    return [v.split("/") for v in raw.split() if "/" in v]


# ---------------------------------------------------------------------------
# Guest agent interaction
# ---------------------------------------------------------------------------

def query_installed_drivers(pod, namespace):
    """Query the QEMU guest agent for installed VirtIO device drivers.

    Runs `guest-get-devices` via the guest agent, which returns every device
    driver installed in Windows. We filter to VirtIO devices (vendor 0x1AF4)
    and return a list of dicts with name, version, and device_id.

    Returns None if the guest agent is not responding.
    """
    raw = oc("exec", "-n", namespace, pod, "--",
             "virsh", "qemu-agent-command", "1",
             '{"execute": "guest-get-devices"}')
    if not raw:
        return None

    for line in raw.splitlines():
        if line.startswith("{"):
            try:
                data = json.loads(line)
            except json.JSONDecodeError:
                continue
            break
    else:
        return None

    drivers = []
    for dev in data.get("return", []):
        dev_id = dev.get("id", {})
        if dev_id.get("vendor-id") != VIRTIO_VENDOR_ID:
            continue
        drivers.append({
            "name": dev.get("driver-name", ""),
            "version": dev.get("driver-version", ""),
            "device_id": f"{dev_id.get('device-id', 0):04X}",
        })
    return drivers if drivers else None


# ---------------------------------------------------------------------------
# Version comparison
# ---------------------------------------------------------------------------

def compare_versions(installed, baseline):
    """Compare two dotted version strings (e.g. "100.100.104.26600").

    Returns "OUTDATED", "OK", or "NEWER".
    """
    try:
        inst = tuple(int(x) for x in installed.split("."))
        base = tuple(int(x) for x in baseline.split("."))
    except ValueError:
        return "UNKNOWN"
    if inst < base:
        return "OUTDATED"
    if inst == base:
        return "OK"
    return "NEWER"


# ---------------------------------------------------------------------------
# Per-VM check
# ---------------------------------------------------------------------------

def check_vm(vm_name, namespace, baseline_version):
    """Check a single VM's VirtIO drivers against the baseline version.

    Returns a result dict with either an "error" key (string) or a "drivers"
    key (list of per-driver results).
    """
    pod = find_virt_launcher_pod(vm_name, namespace)
    if not pod:
        return {"vm": vm_name, "namespace": namespace,
                "error": "no running pod found"}

    drivers = query_installed_drivers(pod, namespace)
    if drivers is None:
        return {"vm": vm_name, "namespace": namespace,
                "error": "guest agent not responding (not installed or not a Windows VM)"}

    results = []
    for drv in drivers:
        results.append({
            "driver": drv["name"],
            "installed": drv["version"],
            "available": baseline_version,
            "status": compare_versions(drv["version"], baseline_version),
            "device_id": drv["device_id"],
        })

    return {"vm": vm_name, "namespace": namespace, "drivers": results}


# ---------------------------------------------------------------------------
# Output formatting
# ---------------------------------------------------------------------------

def print_vm_result(vm_result):
    """Print a single VM's results as a human-readable table."""
    if "error" in vm_result:
        print(f"\nVM: {vm_result['vm']} ({vm_result['namespace']})")
        print(f"  Skipped: {vm_result['error']}")
        return

    results = vm_result["drivers"]
    print(f"\nVM: {vm_result['vm']} ({vm_result['namespace']})")
    print(f"  {'Driver':<45} {'Installed':<25} {'Available':<25} Status")
    print(f"  {'-' * 115}")

    for r in results:
        color = {"OUTDATED": "\033[91m", "OK": "\033[92m"}.get(r["status"], "")
        reset = "\033[0m" if color else ""
        print(f"  {r['driver']:<45} {r['installed']:<25} {r['available']:<25} {color}{r['status']}{reset}")

    outdated = [r for r in results if r["status"] == "OUTDATED"]
    ok = [r for r in results if r["status"] == "OK"]
    if outdated:
        print(f"\033[91m  {len(outdated)} outdated driver(s). Update recommended.\033[0m")
    elif ok:
        print(f"\033[92m  All {len(ok)} matched drivers are up to date.\033[0m")


def print_summary(all_results):
    """Print a one-line summary when checking multiple VMs."""
    if len(all_results) <= 1:
        return
    checked = len([r for r in all_results if "drivers" in r])
    outdated = sum(1 for r in all_results if "drivers" in r
                   and any(d["status"] == "OUTDATED" for d in r["drivers"]))
    skipped = len([r for r in all_results if "error" in r])
    print(f"\n--- Summary: {checked} VM(s) checked, {outdated} with outdated drivers, {skipped} skipped ---")


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("vm",
        help="VM name, or 'all' to scan all running VMs")
    parser.add_argument("-n", "--namespace", default="default",
        help="Namespace (default: 'default', use 'all' for cluster-wide)")
    parser.add_argument("--baseline-version", required=True,
        help="Expected driver version to compare against (e.g. 100.103.104.29700)")
    parser.add_argument("--json", action="store_true", dest="json_output",
        help="Output results as JSON")
    args = parser.parse_args()

    # Build the list of VMs to check
    if args.vm == "all":
        vms = list_running_vmis(args.namespace)
        if not vms:
            print("No running VMs found.", file=sys.stderr)
            sys.exit(1)
        print(f"Found {len(vms)} running VM(s)", file=sys.stderr)
    else:
        vms = [(args.namespace, args.vm)]

    # Check each VM
    all_results = []
    for i, (namespace, vm_name) in enumerate(vms):
        print(f"  [{i+1}/{len(vms)}] Checking {vm_name} ({namespace})...", file=sys.stderr)
        all_results.append(check_vm(vm_name, namespace, args.baseline_version))

    # Output results
    if args.json_output:
        print(json.dumps(all_results if len(all_results) > 1 else all_results[0], indent=2))
    else:
        for vm_result in all_results:
            print_vm_result(vm_result)
        print_summary(all_results)


if __name__ == "__main__":
    main()