virtmanager/app/main.py

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
import os
import libvirt
import subprocess
import random
import xml.etree.ElementTree as ET
import socket
from pydantic import BaseModel

app = FastAPI(title="VirtManager API", description="API for managing virtual machines")

app.mount("/static", StaticFiles(directory="static"), name="static")

ws_processes = {}

def get_free_port():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(('', 0))
        return s.getsockname()[1]

def get_state_string(state_code):
    states = {
        0: "no state",
        1: "running",
        2: "blocked",
        3: "paused",
        4: "shutdown",
        5: "shut off",
        6: "crashed",
        7: "pm suspended"
    }
    return states.get(state_code, "unknown")

class BootOrder(BaseModel):
    order: list[str]

# Connect to libvirt
conn = libvirt.open('qemu:///system')

@app.get("/")
async def root():
    return FileResponse("static/index.html")

@app.get("/vms")
async def list_vms():
    domains = conn.listAllDomains()
    vms = []
    for domain in domains:
        xml_desc = domain.XMLDesc()
        root = ET.fromstring(xml_desc)
        graphics = root.find(".//graphics[@type='vnc']")
        vnc_port = None
        if graphics is not None:
            port_str = graphics.get('port')
            if port_str and port_str != '-1':
                vnc_port = int(port_str)
        vms.append({
            "id": domain.ID(),
            "name": domain.name(),
            "state": get_state_string(domain.state()[0]),
            "vnc_port": vnc_port
        })
    return {"vms": vms}

@app.get("/vms/{vm_name}/start")
async def start_vm(vm_name: str):
    try:
        domain = conn.lookupByName(vm_name)
        domain.create()
        return {"message": f"VM {vm_name} started"}
    except Exception as e:
        return {"error": str(e)}

@app.get("/vms/{vm_name}/stop")
async def stop_vm(vm_name: str):
    try:
        domain = conn.lookupByName(vm_name)
        domain.destroy()
        return {"message": f"VM {vm_name} stopped"}
    except Exception as e:
        return {"error": str(e)}

@app.post("/vms")
async def create_vm(name: str, memory: int, disk_size: int, iso: str):
    images_path = os.environ.get('LIBVIRT_IMAGES_PATH', '/var/lib/libvirt/images')
    disk_path = f"{images_path}/{name}.qcow2"
    iso_path = f"{images_path}/{iso}"
    
    print(f"Using ISO path: {iso_path}")
    
    # Create disk
    try:
        result = subprocess.run(['qemu-img', 'create', '-f', 'qcow2', disk_path, f'{disk_size}G'], capture_output=True, text=True)
        if result.returncode != 0:
            return {"error": f"Failed to create disk: {result.stderr}"}
    except Exception as e:
        return {"error": f"Error creating disk: {str(e)}"}
    
    # Generate unique MAC
    mac = f"52:54:00:{random.randint(0, 255):02x}:{random.randint(0, 255):02x}:{random.randint(0, 255):02x}"
    
    # Basic XML template
    xml_template = f"""<domain type='kvm'>
  <name>{name}</name>
  <memory unit='MiB'>{memory}</memory>
  <currentMemory unit='MiB'>{memory}</currentMemory>
  <vcpu placement='static'>1</vcpu>
  <os>
    <type arch='x86_64' machine='pc-i440fx-6.2'>hvm</type>
    <boot dev='cdrom'/>
  </os>
  <features>
    <acpi/>
    <apic/>
  </features>
  <cpu mode='host-model' check='partial'/>
  <clock offset='utc'>
    <timer name='rtc' tickpolicy='catchup'/>
    <timer name='pit' tickpolicy='delay'/>
    <timer name='hpet' present='no'/>
  </clock>
  <on_poweroff>destroy</on_poweroff>
  <on_reboot>restart</on_reboot>
  <on_crash>destroy</on_crash>
  <pm>
    <suspend-to-mem enabled='no'/>
    <suspend-to-disk enabled='no'/>
  </pm>
  <devices>
    <emulator>/usr/bin/kvm</emulator>
    <disk type='file' device='disk'>
      <driver name='qemu' type='qcow2'/>
      <source file='{disk_path}'/>
      <target dev='vda' bus='virtio'/>
    </disk>
    <disk type='file' device='cdrom'>
      <driver name='qemu' type='raw'/>
      <source file='{iso_path}'/>
      <target dev='hda' bus='ide'/>
      <readonly/>
    </disk>
    <interface type='network'>
      <mac address='{mac}'/>
      <source network='default'/>
      <model type='virtio'/>
    </interface>
    <console type='pty'>
      <target type='serial' port='0'/>
    </console>
    <input type='tablet' bus='usb'>
      <address type='usb' bus='0' port='1'/>
    </input>
    <input type='mouse' bus='ps2'/>
    <input type='keyboard' bus='ps2'/>
    <graphics type='vnc' port='-1' autoport='yes'/>
    <video>
      <model type='cirrus' vram='16384' heads='1' primary='yes'/>
    </video>
    <memballoon model='virtio'>
      <address type='pci' domain='0x0000' bus='0x00' slot='0x04' function='0x0'/>
    </memballoon>
  </devices>
</domain>"""
    
    try:
        domain = conn.defineXML(xml_template)
        domain.create()
        return {"message": f"VM {name} created with {memory} MiB memory, {disk_size}G disk, booting from {iso}"}
    except Exception as e:
        return {"error": str(e)}

@app.get("/console/{vm_name}")
async def get_console(vm_name: str):
    try:
        domain = conn.lookupByName(vm_name)
        xml_desc = domain.XMLDesc()
        root = ET.fromstring(xml_desc)
        graphics = root.find(".//graphics[@type='vnc']")
        if graphics is None:
            return {"error": "No VNC graphics found"}
        vnc_port = graphics.get('port')
        if not vnc_port or vnc_port == '-1':
            return {"error": "VNC port not assigned"}
        vnc_port = int(vnc_port)
        
        if vm_name not in ws_processes:
            ws_port = get_free_port()
            proc = subprocess.Popen(['websockify', str(ws_port), f'localhost:{vnc_port}'])
            ws_processes[vm_name] = (proc, ws_port)
        
        _, ws_port = ws_processes[vm_name]
        return {"url": f"/static/novnc/vnc.html?host={socket.gethostname()}&port={ws_port}"}
    except Exception as e:
        return {"error": str(e)}

@app.delete("/vms/{vm_name}")
async def delete_vm(vm_name: str):
    try:
        domain = conn.lookupByName(vm_name)
        if domain.isActive():
            domain.destroy()
        domain.undefine()
        # Kill websockify process if exists
        if vm_name in ws_processes:
            proc, _ = ws_processes[vm_name]
            proc.terminate()
            proc.wait()
            del ws_processes[vm_name]
        return {"message": f"VM {vm_name} deleted"}
    except Exception as e:
        return {"error": str(e)}

@app.put("/vms/{vm_name}/boot-order")
async def set_boot_order(vm_name: str, boot_order: BootOrder):
    try:
        domain = conn.lookupByName(vm_name)
        if domain.isActive():
            return {"error": "VM must be shut off to change boot order"}
        xml_desc = domain.XMLDesc()
        root = ET.fromstring(xml_desc)
        os_section = root.find("os")
        # Remove existing boot elements
        for boot in os_section.findall("boot"):
            os_section.remove(boot)
        # Add new boot elements
        for dev in boot_order.order:
            ET.SubElement(os_section, "boot", {"dev": dev})
        new_xml = ET.tostring(root, encoding='unicode')
        # Undefine and redefine
        domain.undefine()
        new_domain = conn.defineXML(new_xml)
        return {"message": f"Boot order for {vm_name} updated to {boot_order.order}"}
    except Exception as e:
        return {"error": str(e)}

@app.post("/disks")
async def create_disk(name: str, size: int):
    images_path = os.environ.get('LIBVIRT_IMAGES_PATH', '/var/lib/libvirt/images')
    disk_path = f"{images_path}/{name}.qcow2"
    try:
        result = subprocess.run(['qemu-img', 'create', '-f', 'qcow2', disk_path, f'{size}G'], capture_output=True, text=True)
        if result.returncode == 0:
            return {"message": f"Disk {name}.qcow2 created with size {size}G"}
        else:
            return {"error": result.stderr}
    except Exception as e:
        return {"error": str(e)}