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
    
    print(f"[CREATE_VM] Starting VM creation: name={name}, memory={memory}, disk_size={disk_size}, cpus={cpus}")
    print(f"[CREATE_VM] Using ISO path: {iso_path}")
    
    # Create disk using sudo script for proper permissions
    try:
        # Get absolute path to the script (up one level from app directory)
        app_dir = os.path.dirname(os.path.abspath(__file__))
        project_root = os.path.dirname(app_dir)
        script_path = os.path.join(project_root, 'create_disk.sh')
        print(f"[CREATE_VM] Script path: {script_path}")
        result = subprocess.run(['sudo', script_path, disk_path, f'{disk_size}G'], capture_output=True, text=True, timeout=60)
        print(f"[CREATE_VM] Disk creation return code: {result.returncode}")
        if result.returncode != 0:
            error_msg = result.stderr if result.stderr else result.stdout
            print(f"[CREATE_VM] ERROR creating disk: {error_msg}")
            return {"error": f"Failed to create disk: {error_msg}"}
        print(f"[CREATE_VM] Disk created successfully: {disk_path}")
    except subprocess.TimeoutExpired:
        return {"error": "Disk creation timed out"}
    except Exception as e:
        print(f"[CREATE_VM] Exception creating disk: {str(e)}")
        return {"error": f"Error creating disk: {str(e)}"}
    
    # Verify disk exists
    if not os.path.exists(disk_path):
        print(f"[CREATE_VM] ERROR: Disk file not found after creation: {disk_path}")
        return {"error": f"Disk file not created at {disk_path}"}
    
    # Generate unique MAC
    mac = f"52:54:00:{random.randint(0, 255):02x}:{random.randint(0, 255):02x}:{random.randint(0, 255):02x}"
    print(f"[CREATE_VM] Generated MAC: {mac}")
    
    # 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>"""
    
    print(f"[CREATE_VM] XML template prepared for VM definition")
    
    try:
        print(f"[CREATE_VM] Defining VM with libvirt...")
        domain = conn.defineXML(xml_template)
        print(f"[CREATE_VM] VM defined successfully. Starting VM...")
        domain.create()
        print(f"[CREATE_VM] VM started successfully!")
        return {"message": f"VM {name} created with {cpus} CPUs, {memory} MiB memory, {disk_size}G disk, booting from {iso}"}
    except libvirt.libvirtError as e:
        print(f"[CREATE_VM] Libvirt error: {str(e)}")
        # Try to clean up the disk if VM definition failed
        try:
            os.remove(disk_path)
            print(f"[CREATE_VM] Cleaned up disk file after error")
        except:
            pass
        return {"error": f"Libvirt error: {str(e)}"}
    except Exception as e:
        print(f"[CREATE_VM] Unexpected error: {str(e)}")
        # Try to clean up the disk if VM definition failed
        try:
            os.remove(disk_path)
            print(f"[CREATE_VM] Cleaned up disk file after error")
        except:
            pass
        return {"error": f"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("/vms/{vm_name}/add-disk")
async def add_disk_to_vm(vm_name: str, disk_name: str, disk_size: int):
    images_path = os.environ.get('LIBVIRT_IMAGES_PATH', '/var/lib/libvirt/images')
    disk_path = f"{images_path}/{disk_name}.qcow2"
    
    print(f"[ADD_DISK] Starting disk addition: vm={vm_name}, disk_name={disk_name}, disk_size={disk_size}")
    
    try:
        # Get the VM first to check its state
        print(f"[ADD_DISK] Looking up VM: {vm_name}")
        domain = conn.lookupByName(vm_name)
        
        # Check if VM is running BEFORE creating disk
        print(f"[ADD_DISK] Checking VM state...")
        if domain.isActive():
            print(f"[ADD_DISK] ERROR: VM is running, must be shut off to add disk")
            return {"error": "Error: La máquina virtual debe estar PARADA (shut off) para agregar un disco"}
        print(f"[ADD_DISK] VM state OK (shut off)")
        
        # Check if disk already exists
        print(f"[ADD_DISK] Checking if disk exists: {disk_path}")
        if os.path.exists(disk_path):
            print(f"[ADD_DISK] ERROR: Disk already exists")
            return {"error": f"Error: El disco {disk_name}.qcow2 ya existe. Usa un nombre de disco diferente"}
        print(f"[ADD_DISK] Disk does not exist (OK)")
        
        # Create the disk using sudo script for proper permissions
        print(f"[ADD_DISK] Creating disk...")
        app_dir = os.path.dirname(os.path.abspath(__file__))
        project_root = os.path.dirname(app_dir)
        script_path = os.path.join(project_root, 'create_disk.sh')
        result = subprocess.run(['sudo', script_path, disk_path, f'{disk_size}G'], 
                              capture_output=True, text=True, timeout=60)
        if result.returncode != 0:
            error_msg = result.stderr if result.stderr else result.stdout
            print(f"[ADD_DISK] ERROR creating disk: {error_msg}")
            return {"error": f"Error al crear disco: {error_msg}"}
        print(f"[ADD_DISK] Disk created successfully")
        
        # Get the XML and add the disk
        print(f"[ADD_DISK] Getting VM XML...")
        xml_desc = domain.XMLDesc()
        root = ET.fromstring(xml_desc)
        
        # Find the devices section
        devices = root.find("devices")
        if devices is None:
            print(f"[ADD_DISK] ERROR: No devices section in XML")
            return {"error": "Error: No se encontró sección devices en la configuración de la VM"}
        
        # Find the next available disk target (vda, vdb, vdc, etc.)
        print(f"[ADD_DISK] Finding next available disk target...")
        existing_targets = set()
        for disk in root.findall(".//disk/target"):
            existing_targets.add(disk.get('dev'))
        print(f"[ADD_DISK] Existing targets: {existing_targets}")
        
        # Generate next available target
        target_letters = "abcdefghijklmnopqrstuvwxyz"
        next_target = None
        for i, letter in enumerate(target_letters):
            target = f"vd{letter}"
            if target not in existing_targets:
                next_target = target
                break
        
        if next_target is None:
            print(f"[ADD_DISK] ERROR: No available disk targets")
            return {"error": "Error: No hay dispositivos de disco disponibles"}
        print(f"[ADD_DISK] Next target: {next_target}")
        
        # Create the new disk element
        print(f"[ADD_DISK] Creating disk XML element...")
        new_disk = ET.Element("disk", {"type": "file", "device": "disk"})
        ET.SubElement(new_disk, "driver", {"name": "qemu", "type": "qcow2"})
        ET.SubElement(new_disk, "source", {"file": disk_path})
        ET.SubElement(new_disk, "target", {"dev": next_target, "bus": "virtio"})
        
        # Add the disk to devices
        devices.append(new_disk)
        
        # Convert back to string and redefine the VM
        print(f"[ADD_DISK] Redefining VM with new disk...")
        new_xml = ET.tostring(root, encoding='unicode')
        domain.undefine()
        conn.defineXML(new_xml)
        print(f"[ADD_DISK] VM redefined successfully")
        
        print(f"[ADD_DISK] SUCCESS: Disk added")
        return {"message": f"Disco {disk_name}.qcow2 ({disk_size}G) agregado a {vm_name} como {next_target}"}
    except libvirt.libvirtError as e:
        print(f"[ADD_DISK] Libvirt error: {str(e)}")
        return {"error": f"Error de libvirt: {str(e)}"}
    except Exception as e:
        print(f"[ADD_DISK] Unexpected error: {str(e)}")
        return {"error": f"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)}

@app.get("/vms/{vm_name}/config")
async def get_vm_config(vm_name: str):
    """Get VM configuration in XML format"""
    try:
        conn = libvirt.open(None)
        domain = conn.lookupByName(vm_name)
        xml_desc = domain.XMLDesc()
        conn.close()
        
        return {
            "name": vm_name,
            "xml": xml_desc
        }
    except libvirt.libvirtError as e:
        return {"error": f"Error de libvirt: {str(e)}"}

@app.post("/vms/{vm_name}/snapshot")
async def create_snapshot(vm_name: str, snapshot_name: str, description: str = ""):
    """Create a snapshot of a VM"""
    try:
        conn = libvirt.open(None)
        domain = conn.lookupByName(vm_name)
        
        # Create snapshot XML
        snapshot_xml = f"""<domainsnapshot>
            <name>{snapshot_name}</name>
            <description>{description}</description>
        </domainsnapshot>"""
        
        print(f"[SNAPSHOT] Creating snapshot '{snapshot_name}' for VM '{vm_name}'")
        domain.snapshotCreateXML(snapshot_xml, 0)
        conn.close()
        
        print(f"[SNAPSHOT] Snapshot '{snapshot_name}' created successfully")
        return {"message": f"Snapshot '{snapshot_name}' creado exitosamente"}
    except libvirt.libvirtError as e:
        error_msg = str(e)
        print(f"[SNAPSHOT] Error creating snapshot: {error_msg}")
        return {"error": f"Error al crear snapshot: {error_msg}"}
    except Exception as e:
        error_msg = str(e)
        print(f"[SNAPSHOT] Unexpected error: {error_msg}")
        return {"error": f"Error: {error_msg}"}

@app.get("/vms/{vm_name}/snapshots")
async def list_snapshots(vm_name: str):
    """List all snapshots of a VM"""
    try:
        conn = libvirt.open(None)
        domain = conn.lookupByName(vm_name)
        
        snapshots = []
        try:
            # Get all snapshots
            snapshot_list = domain.listAllSnapshots()
            
            for snapshot in snapshot_list:
                snap_name = snapshot.getName()
                
                # Get snapshot info using available methods
                try:
                    # Try to get current status
                    is_current = snapshot.isCurrent()
                    state = "current" if is_current else "available"
                except:
                    state = "available"
                
                # Try to get creation time from snapshot info
                created_time = 0
                try:
                    # getInfo returns: (type, hypervisor, creationTime, state_flags, dom_name, parent)
                    snap_info = snapshot.getInfo()
                    if snap_info and len(snap_info) >= 3:
                        created_time = snap_info[2]  # creationTime is at index 2
                except:
                    created_time = 0
                
                snapshots.append({
                    "name": snap_name,
                    "created": created_time,
                    "description": "",
                    "state": state
                })
        except libvirt.libvirtError as e:
            # No snapshots available
            print(f"[SNAPSHOT] No snapshots found: {str(e)}")
            pass
        
        conn.close()
        
        return {
            "vm_name": vm_name,
            "snapshots": snapshots
        }
    except libvirt.libvirtError as e:
        error_msg = str(e)
        print(f"[SNAPSHOT] Error listing snapshots: {error_msg}")
        return {"error": f"Error al listar snapshots: {error_msg}"}

@app.delete("/vms/{vm_name}/snapshots/{snapshot_name}")
async def delete_snapshot(vm_name: str, snapshot_name: str):
    """Delete a snapshot of a VM"""
    try:
        conn = libvirt.open(None)
        domain = conn.lookupByName(vm_name)
        
        print(f"[SNAPSHOT] Deleting snapshot '{snapshot_name}' from VM '{vm_name}'")
        snapshot = domain.snapshotLookupByName(snapshot_name)
        snapshot.delete(0)
        conn.close()
        
        print(f"[SNAPSHOT] Snapshot '{snapshot_name}' deleted successfully")
        return {"message": f"Snapshot '{snapshot_name}' eliminado exitosamente"}
    except libvirt.libvirtError as e:
        error_msg = str(e)
        print(f"[SNAPSHOT] Error deleting snapshot: {error_msg}")
        return {"error": f"Error al eliminar snapshot: {error_msg}"}
    except Exception as e:
        error_msg = str(e)
        print(f"[SNAPSHOT] Unexpected error: {error_msg}")