force-shutdown-api.py

#!/bin/python3
import qubesadmin
import subprocess
import sys
import time

q = qubesadmin.Qubes()


def main():
    if len(sys.argv) == 1:
        print("Syntax: force-shutdown.py vm-name")
        exit(1)

    vm = sys.argv[1]
    if not vm in q.domains:
        print("VM '{}' does not exist.".format(vm))
        exit(1)

    if q.domains[vm].is_paused():
        print(
            "VM '{}' is paused. Please unpause it before shutting it down.".format(vm)
        )
        exit(1)

    force_shutdown(vm)


def force_shutdown(vm):
    print("Attempting to shut down VM '{}'.".format(vm))
    try:
        subprocess.check_output(
            ["qvm-run", "-u", "root", vm, "poweroff"], stderr=subprocess.STDOUT
        )
    except subprocess.CalledProcessError as e:
        # Exit codes 1 and 143 may occur with successful shutdown; log others
        if e.returncode != 1 and e.returncode != 143:
            print(
                "poweroff returned exit code {} and the following output:\n".format(
                    e.returncode
                )
            )
            print(e.output.decode("utf-8"))
    wait_shutdown(vm)


def wait_shutdown(vm, timeout=60.0, interval=0.2):
    start_time = time.time()
    stop_time = start_time + timeout
    print("Polling VM '{}' until it is shut down.".format(vm), end="", flush=True)
    while time.time() < stop_time:
        elapsed = time.time() - start_time
        if q.domains[vm].is_running():
            print(".", end="", flush=True)
        else:
            print(
                "\nVM '{}' sucessfully shut down.\nElapsed time after qvm-run exited: {:.2f} seconds".format(
                    vm, elapsed
                )
            )
            return
        time.sleep(interval)
    print(
        "\nVM '{}' did not shut down in the provided timeout of {} seconds.".format(
            vm, timeout
        )
    )


if __name__ == "__main__":
    main()