QEMU Docker Like CLI

qemu

#!/usr/bin/env python3
"""qemu — docker-style CLI for managing QEMU VM command lines.

VM definitions are stored as INI files. System store: /etc/qemu-cli/vms
(used when writable, i.e. root); otherwise falls back to
~/.config/qemu-cli/vms. Both are merged for lookup/list, user store wins.

Usage:
  qemu vm create -n NAME --cmdline "qemu-system-x86_64 ..."
  qemu vm list
  qemu vm ps
  qemu vm inspect NAME
  qemu vm remove NAME
  qemu run [-d] NAME [extra qemu args...]
  qemu stop NAME
"""

import argparse
import configparser
import datetime
import os
import shlex
import signal
import subprocess
import sys
import time

SYSTEM_DIR = os.environ.get("QEMU_CLI_SYSTEM_DIR", "/etc/qemu-cli/vms")
USER_DIR = os.path.join(
    os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config")),
    "qemu-cli", "vms",
)
STATE_DIR = os.path.join(
    os.environ.get("XDG_STATE_HOME", os.path.expanduser("~/.local/state")),
    "qemu-cli", "run",
)


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


# ---------------------------------------------------------------- store

def stores():
    """Directories searched for VM definitions (priority order)."""
    return [USER_DIR, SYSTEM_DIR]


def write_store():
    """Preferred directory for writes: /etc if writable, else user config."""
    try:
        os.makedirs(SYSTEM_DIR, exist_ok=True)
        if os.access(SYSTEM_DIR, os.W_OK):
            return SYSTEM_DIR
    except PermissionError:
        pass
    os.makedirs(USER_DIR, exist_ok=True)
    return USER_DIR


def vm_path(name):
    for d in stores():
        p = os.path.join(d, f"{name}.ini")
        if os.path.isfile(p):
            return p
    return None


def load_vm(name):
    p = vm_path(name)
    if not p:
        die(f"no such vm: {name}")
    cfg = configparser.ConfigParser(interpolation=None)
    cfg.read(p)
    if "vm" not in cfg or "cmdline" not in cfg["vm"]:
        die(f"corrupt definition: {p}")
    return cfg["vm"], p


def all_vms():
    seen = {}
    for d in stores():
        if not os.path.isdir(d):
            continue
        for f in sorted(os.listdir(d)):
            if f.endswith(".ini"):
                name = f[:-4]
                seen.setdefault(name, os.path.join(d, f))
    return seen


# ---------------------------------------------------------------- pids

def pidfile(name):
    return os.path.join(STATE_DIR, f"{name}.pid")


def read_pid(name):
    try:
        with open(pidfile(name)) as fh:
            return int(fh.read().strip())
    except (OSError, ValueError):
        return None


def alive(pid):
    if pid is None:
        return False
    try:
        os.kill(pid, 0)
        return True
    except ProcessLookupError:
        return False
    except PermissionError:
        return True


def running_pid(name):
    pid = read_pid(name)
    if alive(pid):
        return pid
    # stale
    try:
        os.unlink(pidfile(name))
    except OSError:
        pass
    return None


# ---------------------------------------------------------------- commands

def cmd_create(args):
    name = args.name
    if not name or "/" in name:
        die("invalid vm name")
    if vm_path(name) and not args.force:
        die(f"vm '{name}' already exists (use -f to overwrite)")
    cmdline = args.cmdline.strip()
    if not cmdline:
        die("empty --cmdline")
    # sanity check it parses
    try:
        argv = shlex.split(cmdline)
    except ValueError as e:
        die(f"cannot parse cmdline: {e}")
    if not argv:
        die("empty cmdline")

    cfg = configparser.ConfigParser(interpolation=None)
    cfg["vm"] = {
        "name": name,
        "cmdline": cmdline,
        "workdir": os.getcwd(),
        "created": datetime.datetime.now().isoformat(timespec="seconds"),
    }
    dest = os.path.join(write_store(), f"{name}.ini")
    with open(dest, "w") as fh:
        cfg.write(fh)
    print(f"{name}  ->  {dest}")


def cmd_list(_args):
    vms = all_vms()
    if not vms:
        print("no vms defined")
        return
    print(f"{'NAME':<24}{'STATUS':<12}{'BINARY':<22}DEFINITION")
    for name, path in vms.items():
        cfg = configparser.ConfigParser(interpolation=None)
        cfg.read(path)
        cmdline = cfg.get("vm", "cmdline", fallback="")
        binary = shlex.split(cmdline)[0] if cmdline else "?"
        status = "running" if running_pid(name) else "-"
        print(f"{name:<24}{status:<12}{os.path.basename(binary):<22}{path}")


def cmd_ps(_args):
    rows = []
    for name in all_vms():
        pid = running_pid(name)
        if pid:
            try:
                started = os.stat(pidfile(name)).st_mtime
                up = int(time.time() - started)
                uptime = f"{up // 3600}h{(up % 3600) // 60:02d}m"
            except OSError:
                uptime = "?"
            rows.append((name, pid, uptime))
    if not rows:
        print("no vms running")
        return
    print(f"{'NAME':<24}{'PID':<10}UPTIME")
    for name, pid, uptime in rows:
        print(f"{name:<24}{pid:<10}{uptime}")


def cmd_inspect(args):
    vm, path = load_vm(args.name)
    print(f"path:    {path}")
    for k in ("name", "created", "workdir"):
        print(f"{k}:{' ' * (8 - len(k))}{vm.get(k, '-')}")
    pid = running_pid(args.name)
    print(f"status:  {'running (pid ' + str(pid) + ')' if pid else 'stopped'}")
    print(f"cmdline: {vm['cmdline']}")


def cmd_remove(args):
    p = vm_path(args.name)
    if not p:
        die(f"no such vm: {args.name}")
    if running_pid(args.name):
        die(f"vm '{args.name}' is running; stop it first")
    try:
        os.unlink(p)
    except PermissionError:
        die(f"permission denied removing {p} (try sudo)")
    print(f"removed {p}")


def cmd_run(args):
    vm, _ = load_vm(args.name)
    if running_pid(args.name):
        die(f"vm '{args.name}' is already running")

    argv = shlex.split(vm["cmdline"]) + list(args.extra or [])
    argv = [os.path.expanduser(a) for a in argv]
    workdir = vm.get("workdir", os.getcwd())
    if not os.path.isdir(workdir):
        workdir = os.getcwd()

    os.makedirs(STATE_DIR, exist_ok=True)

    try:
        proc = subprocess.Popen(
            argv,
            cwd=workdir,
            start_new_session=args.detach,
            stdout=subprocess.DEVNULL if args.detach else None,
            stderr=subprocess.DEVNULL if args.detach else None,
        )
    except FileNotFoundError:
        die(f"binary not found: {argv[0]}")

    with open(pidfile(args.name), "w") as fh:
        fh.write(str(proc.pid))

    if args.detach:
        print(f"{args.name} started (pid {proc.pid})")
        return

    try:
        rc = proc.wait()
    except KeyboardInterrupt:
        proc.terminate()
        rc = proc.wait()
    finally:
        try:
            os.unlink(pidfile(args.name))
        except OSError:
            pass
    sys.exit(rc)


def cmd_stop(args):
    pid = running_pid(args.name)
    if not pid:
        die(f"vm '{args.name}' is not running")
    os.kill(pid, signal.SIGTERM)
    for _ in range(int(args.timeout * 10)):
        if not alive(pid):
            break
        time.sleep(0.1)
    else:
        os.kill(pid, signal.SIGKILL)
        print(f"{args.name}: SIGKILL after {args.timeout}s")
    try:
        os.unlink(pidfile(args.name))
    except OSError:
        pass
    print(f"{args.name} stopped")


# ---------------------------------------------------------------- argparse

def main():
    p = argparse.ArgumentParser(prog="qemu", description=__doc__.splitlines()[0])
    sub = p.add_subparsers(dest="cmd", required=True)

    vm = sub.add_parser("vm", help="manage vm definitions")
    vmsub = vm.add_subparsers(dest="vmcmd", required=True)

    c = vmsub.add_parser("create", help="define a new vm")
    c.add_argument("-n", "--name", required=True)
    c.add_argument("--cmdline", required=True, help="full qemu command line")
    c.add_argument("-f", "--force", action="store_true", help="overwrite")
    c.set_defaults(func=cmd_create)

    vmsub.add_parser("list", help="list defined vms").set_defaults(func=cmd_list)
    vmsub.add_parser("ps", help="list running vms").set_defaults(func=cmd_ps)

    i = vmsub.add_parser("inspect", help="show vm details")
    i.add_argument("name")
    i.set_defaults(func=cmd_inspect)

    r = vmsub.add_parser("remove", aliases=["rm"], help="delete a vm definition")
    r.add_argument("name")
    r.set_defaults(func=cmd_remove)

    run = sub.add_parser("run", help="start a vm")
    run.add_argument("name")
    run.add_argument("-d", "--detach", action="store_true",
                     help="run in background")
    run.add_argument("extra", nargs=argparse.REMAINDER,
                     help="extra args appended to the stored cmdline")
    run.set_defaults(func=cmd_run)

    ps = sub.add_parser("ps", help="alias for 'vm ps'")
    ps.set_defaults(func=cmd_ps)

    stop = sub.add_parser("stop", help="stop a running vm (SIGTERM)")
    stop.add_argument("name")
    stop.add_argument("-t", "--timeout", type=float, default=10.0)
    stop.set_defaults(func=cmd_stop)

    args = p.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()