virtmanager/app/main.py

from fastapi import FastAPI, UploadFile, File
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}/info")
async def get_vm_info(vm_name: str):
    try:
        domain = conn.lookupByName(vm_name)
        xml_desc = domain.XMLDesc()
        root = ET.fromstring(xml_desc)
        
        # Get basic info
        memory = root.find("memory").text if root.find("memory") is not None else "Unknown"
        cpus = root.find("vcpu").text if root.find("vcpu") is not None else "Unknown"
        
        # Get VNC port
        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)
        
        # Get disks
        disks = []
        for disk in root.findall(".//disk"):
            device = disk.get('device')
            driver = disk.find("driver")
            source = disk.find("source")
            target = disk.find("target")
            disks.append({
                "type": device,
                "driver": driver.get('type') if driver is not None else "unknown",
                "source": source.get('file') if source is not None else "unknown",
                "target": target.get('dev') if target is not None else "unknown"
            })
        
        # Get network interfaces
        interfaces = []
        for iface in root.findall(".//interface"):
            mac = iface.find("mac")
            source = iface.find("source")
            model = iface.find("model")
            interfaces.append({
                "mac": mac.get('address') if mac is not None else "unknown",
                "network": source.get('network') if source is not None else "unknown",
                "model": model.get('type') if model is not None else "unknown"
            })
        
        return {
            "id": domain.ID(),
            "name": domain.name(),
            "state": get_state_string(domain.state()[0]),
            "memory": int(memory),
            "cpus": int(cpus),
            "vnc_port": vnc_port,
            "disks": disks,
            "interfaces": interfaces
        }
    except Exception as e:
        return {"error": str(e)}, 500

@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, cpus: int = 1, iso: str = ""):
    if not iso:
        return {"error": "ISO path is required"}
    images_path = os.environ.get('LIBVIRT_IMAGES_PATH', '/var/lib/libvirt/images')
    disk_path = f"{images_path}/{name}.qcow2"
    iso_path = iso  # iso ya es la ruta completa desde el select
    
    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'>{cpus}</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 {cpus} CPUs, {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.get("/isos")
async def list_isos():
    images_path = os.environ.get('LIBVIRT_IMAGES_PATH', '/var/lib/libvirt/images')
    iso_dir = f"{images_path}/isos"
    if os.path.exists(iso_dir):
        isos = [{"name": f, "path": os.path.join(iso_dir, f)} for f in os.listdir(iso_dir) if f.endswith('.iso')]
        return {"isos": isos}
    return {"isos": []}

@app.post("/upload-iso")
async def upload_iso(file: UploadFile = File(...)):
    images_path = os.environ.get('LIBVIRT_IMAGES_PATH', '/var/lib/libvirt/images')
    iso_dir = f"{images_path}/isos"
    os.makedirs(iso_dir, exist_ok=True)
    file_path = os.path.join(iso_dir, file.filename)
    with open(file_path, "wb") as f:
        while chunk := await file.read(1024 * 1024):  # Read in 1MB chunks
            f.write(chunk)
    return {"message": f"ISO {file.filename} uploaded successfully", "path": file_path}

@app.delete("/isos/{iso_name}")
async def delete_iso(iso_name: str):
    images_path = os.environ.get('LIBVIRT_IMAGES_PATH', '/var/lib/libvirt/images')
    iso_dir = f"{images_path}/isos"
    file_path = os.path.join(iso_dir, iso_name)
    if os.path.exists(file_path) and iso_name.endswith('.iso'):
        os.remove(file_path)
        return {"message": f"ISO {iso_name} deleted successfully"}
    else:
        return {"error": "ISO not found or invalid"}

@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)}