From 8a62d826cc2e354004d94a8306bab2830e762618 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 1 Aug 2026 13:50:50 +0300 Subject: [PATCH] =?UTF-8?q?=D0=A3=D0=B4=D0=B0=D0=BB=D0=B8=D1=82=D1=8C=20pr?= =?UTF-8?q?oxmox=5Ftools.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- proxmox_tools.py | 891 ----------------------------------------------- 1 file changed, 891 deletions(-) delete mode 100644 proxmox_tools.py diff --git a/proxmox_tools.py b/proxmox_tools.py deleted file mode 100644 index 62ab6fa..0000000 --- a/proxmox_tools.py +++ /dev/null @@ -1,891 +0,0 @@ -""" -title: Proxmox Tools (PVE) -author: (your name) -description: Инструменты для управления Proxmox VE: LXC (создание, старт, стоп, удаление), VM (клонирование из шаблона, создание из ISO, управление), мониторинг (ноды, хранилище, ресурсы, задачи). -version: 1.0.0 -""" - -import time -import requests -import urllib3 -from pydantic import BaseModel, Field -from typing import Optional - - -class Tools: - class Valves(BaseModel): - proxmox_host: str = Field( - default="https://192.168.31.2:8006", - description="Base URL of the Proxmox API (e.g. https://192.168.31.2:8006)", - ) - proxmox_user: str = Field( - default="root@pam", description="Proxmox API user (e.g. root@pam)" - ) - proxmox_token_name: str = Field( - default="openwebui-token", description="Proxmox API token name/id" - ) - proxmox_token_value: str = Field( - default="", - description="Proxmox API token secret (set this in the Valves UI, never commit it to code)", - ) - verify_ssl: bool = Field( - default=False, description="Verify TLS certificate of the Proxmox host" - ) - default_node: str = Field( - default="pve", description="Default Proxmox node name" - ) - default_storage: str = Field( - default="local", description="Default storage for LXC rootfs / templates" - ) - default_bridge: str = Field( - default="vmbr0", description="Default network bridge" - ) - default_template_vmid: int = Field( - default=1000, - description="Default VMID of the cloud-init template used for VM cloning", - ) - request_timeout: int = Field( - default=30, description="HTTP request timeout in seconds" - ) - - def __init__(self): - self.valves = self.Valves() - # Silence "InsecureRequestWarning" spam when verify_ssl is False - urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - - # ============================================================ - # INTERNAL HELPERS (not exposed to the LLM) - # ============================================================ - - def _base_url(self) -> str: - return self.valves.proxmox_host.rstrip("/") + "/api2/json" - - def _auth_header(self) -> dict: - return { - "Authorization": ( - f"PVEAPIToken={self.valves.proxmox_user}!" - f"{self.valves.proxmox_token_name}={self.valves.proxmox_token_value}" - ) - } - - def _api( - self, - method: str, - path: str, - params: Optional[dict] = None, - data: Optional[dict] = None, - ): - """Execute a Proxmox API request.""" - url = self._base_url() + path - resp = requests.request( - method=method, - url=url, - headers=self._auth_header(), - params=params, - json=data, - verify=self.valves.verify_ssl, - timeout=self.valves.request_timeout, - ) - resp.raise_for_status() - return resp.json().get("data", {}) - - @staticmethod - def _fmt(b) -> str: - """Format bytes to human-readable.""" - if b is None: - return "N/A" - b = float(b) - for u in ["B", "KB", "MB", "GB", "TB"]: - if b < 1024: - return f"{b:.1f} {u}" - b /= 1024 - return f"{b:.1f} PB" - - @classmethod - def _pct(cls, u, t) -> str: - """Format used/total as '1.2 GB / 4.0 GB (30.0%)'.""" - if u is None or t is None or t == 0: - return "N/A" - return f"{cls._fmt(u)} / {cls._fmt(t)} ({u/t*100:.1f}%)" - - @staticmethod - def _uptime(s) -> str: - """Format uptime seconds to '1d 2h 3m 4s'.""" - if s is None: - return "N/A" - d, r = divmod(int(s), 86400) - h, r = divmod(r, 3600) - m, s2 = divmod(r, 60) - parts = [] - if d: - parts.append(f"{d}d") - if h: - parts.append(f"{h}h") - if m: - parts.append(f"{m}m") - parts.append(f"{s2}s") - return " ".join(parts) - - def _wait_for_task(self, upid: str, node: str, timeout: int = 60): - """Wait for a Proxmox task to complete by UPID, on the given node.""" - if not upid or not isinstance(upid, str) or not upid.startswith("UPID:"): - return True, "no UPID to wait" - start = time.time() - while time.time() - start < timeout: - try: - tasks = self._api("GET", f"/nodes/{node}/tasks") - for t in tasks: - if t.get("upid") == upid: - status = t.get("status", "") - if status == "stopped": - return ( - (True, "OK") - if t.get("exitstatus") == "OK" - else (False, str(t)) - ) - break - except Exception: - pass - time.sleep(2) - return False, "timeout waiting for task" - - # ============================================================ - # TOOLS (exposed to the LLM) - # ============================================================ - - def connection(self) -> str: - """Check the connection to the Proxmox VE API and report its version.""" - try: - v = self._api("GET", "/version") - return ( - "OK Proxmox VE " - + v.get("version", "?") - + f" | node: {self.valves.default_node} | host: {self.valves.proxmox_host}" - ) - except Exception as e: - return "ERR " + str(e) - - def nodes(self) -> str: - """List all cluster nodes with their CPU/RAM usage and uptime.""" - try: - items = self._api("GET", "/nodes") - out = ["Nodes:"] - for n in items: - out.append(f"- {n.get('node', '?')} ({n.get('status', '?')})") - out.append( - f" CPU: {n.get('cpu', 0)*100:.1f}% | RAM: {self._pct(n.get('mem'), n.get('maxmem'))}" - ) - out.append(f" Uptime: {self._uptime(n.get('uptime'))}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def node_status(self, node: Optional[str] = None) -> str: - """Show detailed status of a Proxmox node: CPU model, load, RAM, swap, disk, uptime, kernel and PVE version. - - Args: - node: Node name. Defaults to the configured default node if omitted. - """ - node = node or self.valves.default_node - try: - s = self._api("GET", f"/nodes/{node}/status") - out = [f"Node {node}:"] - ci = s.get("cpuinfo", {}) - out.append(f"CPU: {ci.get('model', '?')} ({s.get('cpus', 0)} cores)") - out.append(f"Load: {s.get('cpu', 0)*100:.1f}%") - la = s.get("loadavg", []) - if la: - out.append(f"Loadavg: {' / '.join(f'{float(v):.2f}' for v in la)}") - m = s.get("memory", {}) - out.append(f"RAM: {self._pct(m.get('used'), m.get('total'))}") - sw = s.get("swap", {}) - if sw.get("total", 0): - out.append(f"SWAP: {self._pct(sw.get('used'), sw.get('total'))}") - r = s.get("rootfs", {}) - if r: - out.append(f"Disk: {self._pct(r.get('used'), r.get('total'))}") - out.append(f"Uptime: {self._uptime(s.get('uptime'))}") - out.append(f"Kernel: {s.get('kversion', '?')}") - if s.get("pveversion"): - out.append(f"PVE: {s['pveversion']}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def cluster(self) -> str: - """Show cluster status: quorum state and cluster members.""" - try: - items = self._api("GET", "/cluster/status") - out = ["Cluster:"] - has_cluster_info = False - for item in items: - t = item.get("type", "") - if t == "cluster": - has_cluster_info = True - out.append( - f"Cluster: {item.get('name', '?')} | Quorum: {'OK' if item.get('quorate') else 'FAIL'}" - ) - elif t == "node": - out.append( - f" {item.get('name', '?')} - {item.get('status', '?')} ({item.get('ip', '-')})" - ) - if not has_cluster_info: - node_count = sum(1 for i in items if i.get("type") == "node") - out.append("Cluster: single-node | Quorum: OK (N/A)") - out.append(f" Nodes: {node_count}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def resources(self, ftype: str = "") -> str: - """List all cluster resources (nodes, QEMU VMs, LXC containers, storage), optionally filtered by type. - - Args: - ftype: Optional filter: "node", "qemu", "lxc", or "storage". Empty string returns all types. - """ - try: - p = {"type": ftype} if ftype else {} - items = self._api("GET", "/cluster/resources", params=p) - out = ["Resources:"] - g = {} - for r in items: - g.setdefault(r.get("type", "?"), []).append(r) - labels = { - "node": "Nodes", - "qemu": "QEMU", - "lxc": "LXC", - "storage": "Storage", - } - for t, lst in g.items(): - out.append(f"{labels.get(t, t)}: {len(lst)}") - for r in lst[:25]: - n = r.get("name") or r.get("id") or "?" - st = r.get("status", "?") - line = f" {n} - {st}" - if t in ("qemu", "lxc"): - line += f" (ID:{r.get('vmid', '?')})" - if r.get("maxmem"): - line += f" RAM:{self._pct(r.get('mem'), r.get('maxmem'))}" - if r.get("cpu"): - line += f" CPU:{r.get('cpu')*100:.1f}%" - out.append(line) - out.append("") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def storage(self) -> str: - """List all configured storages with their content types and usage.""" - try: - items = self._api("GET", "/storage") - if not items: - return "No storage" - out = ["Storage:"] - for s in items: - name = s.get("storage", "?") - out.append(f"- {name} ({s.get('type', '?')})") - out.append(f" Content: {s.get('content', '?')}") - if s.get("total"): - out.append(f" Used: {self._pct(s.get('used'), s.get('total'))}") - out.append(f" Free: {self._fmt(s.get('avail'))}") - out.append("") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def tasks(self, limit: int = 10, node: Optional[str] = None) -> str: - """Show recent tasks on a node. - - Args: - limit: Maximum number of tasks to return. - node: Node name. Defaults to the configured default node if omitted. - """ - node = node or self.valves.default_node - try: - items = self._api("GET", f"/nodes/{node}/tasks", params={"limit": limit}) - if not items: - return f"No tasks on {node}" - out = [f"Tasks on {node}:"] - for t in items: - start = time.ctime(t.get("starttime", 0)) if t.get("starttime") else "?" - out.append( - f"- {t.get('type', '?')} | {t.get('status', '?')} | {t.get('user', '?')} | {start}" - ) - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def lxc_list(self, node: Optional[str] = None) -> str: - """List LXC containers on a node with status, CPU, RAM, disk and uptime. - - Args: - node: Node name. Defaults to the configured default node if omitted. - """ - node = node or self.valves.default_node - try: - items = self._api("GET", f"/nodes/{node}/lxc") - if not items: - return f"No LXC on {node}" - out = [f"LXC on {node}:"] - for c in items: - out.append( - f"- {c.get('name', '?')} (ID:{c.get('vmid', '?')}) - {c.get('status', '?')}" - ) - if c.get("cpu"): - out.append(f" CPU:{c['cpu']*100:.1f}%") - if c.get("maxmem"): - out.append(f" RAM:{self._pct(c.get('mem'), c.get('maxmem'))}") - if c.get("maxdisk"): - out.append(f" Disk:{self._pct(c.get('disk'), c.get('maxdisk'))}") - if c.get("uptime"): - out.append(f" Uptime:{self._uptime(c['uptime'])}") - out.append("") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def lxc_detail(self, vmid: int, node: Optional[str] = None) -> str: - """Show detailed info about a specific LXC container: OS, CPU, RAM, disk, network, usage and uptime. - - Args: - vmid: The numeric ID of the LXC container. - node: Node name. Defaults to the configured default node if omitted. - """ - node = node or self.valves.default_node - try: - s = self._api("GET", f"/nodes/{node}/lxc/{vmid}/status/current") - c = self._api("GET", f"/nodes/{node}/lxc/{vmid}/config") - name = s.get("name", c.get("hostname", "?")) - out = [f"LXC #{vmid}: {name} - {s.get('status', '?')}"] - out.append( - f"OS: {c.get('ostype', '?')} | CPU: {c.get('cores', '?')} | RAM: {c.get('memory', '?')}MB" - ) - out.append(f"Disk: {c.get('rootfs', '?')} | SWAP: {c.get('swap', '?')}MB") - out.append(f"Hostname: {c.get('hostname', '?')}") - nets = [f" {k}: {v}" for k, v in c.items() if k.startswith("net")] - if nets: - out.append("Network:\n" + "\n".join(nets)) - if s.get("maxmem"): - out.append(f"RAM use: {self._pct(s.get('mem'), s.get('maxmem'))}") - if s.get("cpu"): - out.append(f"CPU: {s['cpu']*100:.1f}%") - if s.get("uptime"): - out.append(f"Uptime: {self._uptime(s['uptime'])}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def lxc_action(self, vmid: int, action: str, node: Optional[str] = None) -> str: - """Perform a lifecycle action on an LXC container. - - Args: - vmid: The numeric ID of the LXC container. - action: One of "start", "stop", "shutdown", "restart", "suspend", "resume", "delete". - node: Node name. Defaults to the configured default node if omitted. - """ - node = node or self.valves.default_node - try: - acts = { - "start": "start", - "stop": "stop", - "shutdown": "shutdown", - "restart": "reboot", - "suspend": "suspend", - "resume": "resume", - "delete": "del", - } - if action not in acts: - return f"Unknown action: {action}. Allowed: {', '.join(acts.keys())}" - if action == "delete": - result = self._api("DELETE", f"/nodes/{node}/lxc/{vmid}") - else: - result = self._api( - "POST", f"/nodes/{node}/lxc/{vmid}/status/{acts[action]}" - ) - return f"OK LXC #{vmid} {action}\n{result}" - except Exception as e: - return "ERR " + str(e) - - def templates( - self, storage: Optional[str] = None, node: Optional[str] = None - ) -> str: - """List available LXC templates in a storage. - - Args: - storage: Storage name. Defaults to the configured default storage if omitted. - node: Node name. Defaults to the configured default node if omitted. - """ - node = node or self.valves.default_node - storage = storage or self.valves.default_storage - try: - content = self._api("GET", f"/nodes/{node}/storage/{storage}/content") - tmpl = [x for x in content if x.get("content") == "vztmpl"] - if not tmpl: - return f"No templates in {storage}" - out = [f"Templates in {storage}:"] - for t in tmpl: - out.append(f" {t.get('volid', '?')} - {self._fmt(t.get('size'))}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def lxc_create( - self, - vmid: int, - hostname: str, - ostemplate: str, - password: str = "", - sshkey: str = "", - storage: Optional[str] = None, - cores: int = 1, - memory: int = 512, - swap: int = 0, - disk: str = "8G", - bridge: Optional[str] = None, - ip: str = "dhcp", - netmask: int = 24, - gw: str = "", - dns: str = "", - domain: str = "", - node: Optional[str] = None, - unpriv: bool = True, - start_now: bool = True, - wait: bool = True, - ) -> str: - """Create a new LXC container from a template. - - Args: - vmid: New numeric ID for the container (must not already exist). - hostname: Hostname for the container. - ostemplate: Template volume id, e.g. "local:vztmpl/debian-12-standard_12.7-1_amd64.tar.zst". - password: Root password (required if sshkey is not provided). - sshkey: SSH public key (required if password is not provided). - storage: Storage for the rootfs. Defaults to the configured default storage if omitted. - cores: Number of CPU cores. - memory: RAM in MB. - swap: Swap in MB. - disk: Root disk size, e.g. "8G". - bridge: Network bridge. Defaults to the configured default bridge if omitted. - ip: IP address, or "dhcp". - netmask: Netmask prefix length, used only if ip is not "dhcp". - gw: Gateway IP, used only if ip is not "dhcp". - dns: Nameserver IP. - domain: Search domain. - node: Node name. Defaults to the configured default node if omitted. - unpriv: Whether to create an unprivileged container. - start_now: Whether to start the container right after creation. - wait: Whether to wait for the creation task to finish before returning. - """ - node = node or self.valves.default_node - storage = storage or self.valves.default_storage - bridge = bridge or self.valves.default_bridge - try: - if not hostname: - return "Error: hostname required" - if not ostemplate: - return "Error: ostemplate required" - if not password and not sshkey: - return "Error: password or sshkey required" - exist = [str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/lxc")] - exist += [ - str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/qemu") - ] - if str(vmid) in exist: - return f"Error: VMID {vmid} already exists" - params = { - "vmid": vmid, - "hostname": hostname, - "ostemplate": ostemplate, - "storage": storage, - "cores": cores, - "memory": memory, - "swap": swap, - "unprivileged": 1 if unpriv else 0, - } - params["rootfs"] = disk if ":" in disk else f"{storage}:{disk}" - net = f"name=eth0,bridge={bridge}" - if ip and ip.lower() != "dhcp": - net += f",ip={ip}/{netmask}" - if gw: - net += f",gw={gw}" - else: - net += ",ip=dhcp" - params["net0"] = net - if password: - params["password"] = password - if sshkey: - params["ssh-public-keys"] = sshkey - if dns: - params["nameserver"] = dns - if domain: - params["searchdomain"] = domain - if start_now: - params["start"] = 1 - result = self._api("POST", f"/nodes/{node}/lxc", data=params) - out = [ - f"OK LXC #{vmid} creating!", - f" Name: {hostname}", - f" Template: {ostemplate}", - f" CPU: {cores} cores, RAM: {memory}MB, Disk: {disk}", - ] - out.append( - f" Net: {bridge}, IP: {'DHCP' if ip == 'dhcp' else f'{ip}/{netmask}'}" - ) - upid = result if isinstance(result, str) else "" - if upid: - out.append(f"UPID: {upid}") - if wait and upid: - out.append("Waiting for creation to finish...") - ok, msg = self._wait_for_task(upid, node=node, timeout=120) - out.append( - " Creation done" - if ok - else f" WARN Creation may not have finished: {msg}" - ) - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def vm_list(self, node: Optional[str] = None) -> str: - """List QEMU VMs on a node with status, CPU, RAM and uptime. - - Args: - node: Node name. Defaults to the configured default node if omitted. - """ - node = node or self.valves.default_node - try: - items = self._api("GET", f"/nodes/{node}/qemu") - if not items: - return f"No VMs on {node}" - out = [f"VMs on {node}:"] - for v in items: - out.append( - f"- {v.get('name', '?')} (ID:{v.get('vmid', '?')}) - {v.get('status', '?')}" - ) - if v.get("cpu"): - out.append(f" CPU:{v['cpu']*100:.1f}%") - if v.get("maxmem"): - out.append(f" RAM:{self._pct(v.get('mem'), v.get('maxmem'))}") - if v.get("uptime"): - out.append(f" Uptime:{self._uptime(v['uptime'])}") - out.append("") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def vm_detail(self, vmid: int, node: Optional[str] = None) -> str: - """Show detailed status of a specific QEMU VM: RAM, CPU, disk and uptime. - - Args: - vmid: The numeric ID of the VM. - node: Node name. Defaults to the configured default node if omitted. - """ - node = node or self.valves.default_node - try: - s = self._api("GET", f"/nodes/{node}/qemu/{vmid}/status/current") - out = [ - f"VM #{vmid}: {s.get('name', '?')}", - f"Status: {s.get('status', '?')}", - ] - if s.get("maxmem"): - out.append(f"RAM: {self._pct(s.get('mem'), s.get('maxmem'))}") - if s.get("cpu"): - out.append(f"CPU: {s['cpu']*100:.1f}%") - if s.get("maxdisk") and s.get("maxdisk", 0) > 0: - out.append(f"Disk: {self._pct(s.get('disk'), s.get('maxdisk'))}") - if s.get("uptime"): - out.append(f"Uptime: {self._uptime(s['uptime'])}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def vm_action(self, vmid: int, action: str, node: Optional[str] = None) -> str: - """Perform a lifecycle action on a QEMU VM. - - Args: - vmid: The numeric ID of the VM. - action: One of "start", "stop", "shutdown", "restart", "suspend", "resume", "delete". - node: Node name. Defaults to the configured default node if omitted. - """ - node = node or self.valves.default_node - try: - acts = { - "start": "start", - "stop": "stop", - "shutdown": "shutdown", - "restart": "reboot", - "suspend": "suspend", - "resume": "resume", - "delete": "del", - } - if action not in acts: - return f"Unknown action: {action}. Allowed: {', '.join(acts.keys())}" - if action == "delete": - result = self._api("DELETE", f"/nodes/{node}/qemu/{vmid}") - else: - result = self._api( - "POST", f"/nodes/{node}/qemu/{vmid}/status/{acts[action]}" - ) - return f"OK VM #{vmid} {action}\n{result}" - except Exception as e: - return "ERR " + str(e) - - def vm_create_from_template( - self, - vmid: int, - name: str, - template_vmid: Optional[int] = None, - full: bool = True, - memory: Optional[int] = None, - cores: Optional[int] = None, - storage: Optional[str] = None, - target_node: Optional[str] = None, - pool: Optional[str] = None, - start_now: bool = True, - ciuser: str = "", - cipassword: str = "", - sshkeys: str = "", - node: Optional[str] = None, - ) -> str: - """Create a new VM by cloning it from an existing template VM (cloud-init). - - The clone step never sends a "start" flag to the API — the VM is started - via a separate call afterwards, once cloning and cloud-init settings are applied. - - Args: - vmid: New numeric ID for the cloned VM (must not already exist). - name: Name for the new VM. - template_vmid: VMID of the source template to clone from. Defaults to the configured default template if omitted. - full: True for a full clone, False for a linked clone. - memory: RAM in MB to set after cloning. Leave unset to keep the template's value. - cores: CPU cores to set after cloning. Leave unset to keep the template's value. - storage: Target storage for the clone. Leave unset to use the template's storage. - target_node: Target node if cloning to a different node than the source. - pool: Resource pool to assign the clone to. - start_now: Whether to start the VM right after cloning. - ciuser: Cloud-init user to set (e.g. "ubuntu"). - cipassword: Cloud-init password to set. - sshkeys: Cloud-init SSH public keys to set (newline-separated). - node: Node name where the template lives. Defaults to the configured default node if omitted. - """ - node = node or self.valves.default_node - template_vmid = template_vmid or self.valves.default_template_vmid - try: - if not vmid: - return "Error: vmid required" - if not name: - return "Error: name required" - exist_vm = [ - str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/qemu") - ] - exist_lxc = [ - str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/lxc") - ] - if str(vmid) in exist_vm + exist_lxc: - return f"Error: VMID {vmid} already exists" - # Step 1: clone (without "start" - startup is a separate call below) - params = {"newid": vmid, "name": name, "full": 1 if full else 0} - if storage: - params["storage"] = storage - if target_node: - params["target"] = target_node - if pool: - params["pool"] = pool - result = self._api( - "POST", f"/nodes/{node}/qemu/{template_vmid}/clone", data=params - ) - out = [ - f"OK VM #{vmid}: {name} cloning from template #{template_vmid}!", - f" Type: {'full' if full else 'linked'} clone", - ] - if storage: - out.append(f" Storage: {storage}") - upid = result if isinstance(result, str) else "" - if upid: - out.append(f"UPID: {upid}") - # Step 2: wait for clone to finish - out.append("Waiting for clone to finish...") - clone_node = target_node or node - ok, msg = self._wait_for_task(upid, node=clone_node, timeout=120) - out.append( - " Clone done" if ok else f" WARN Clone may not have finished: {msg}" - ) - # Step 3: apply settings (resources + cloud-init) - config_params = {} - if memory is not None: - config_params["memory"] = memory - if cores is not None: - config_params["cores"] = cores - if ciuser: - config_params["ciuser"] = ciuser - if cipassword: - config_params["cipassword"] = cipassword - if sshkeys: - config_params["sshkeys"] = sshkeys - if config_params: - out.append("Applying settings...") - try: - self._api( - "PUT", - f"/nodes/{clone_node}/qemu/{vmid}/config", - data=config_params, - ) - if memory is not None: - out.append(f" RAM -> {memory} MB") - if cores is not None: - out.append(f" CPU -> {cores} cores") - if ciuser: - out.append(f" Cloud-init user: {ciuser}") - if cipassword: - out.append(" Cloud-init password: (set)") - if sshkeys: - out.append(" Cloud-init SSH keys: (set)") - out.append(" Settings applied") - except Exception as e2: - out.append(f" WARN Could not apply settings: {e2}") - # Step 4: start VM (separate call) - if start_now: - out.append("Starting VM...") - try: - self._api("POST", f"/nodes/{clone_node}/qemu/{vmid}/status/start") - out.append(f" VM #{vmid} started!") - except Exception as e3: - out.append(f" WARN Could not start: {e3}") - else: - out.append(f" VM #{vmid} created (not started)") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - def vm_create( - self, - vmid: int, - name: str, - template_vmid: Optional[int] = None, - memory: int = 4096, - cores: int = 2, - disk_size: str = "32G", - disk_storage: str = "local-lvm", - iso: str = "", - bridge: Optional[str] = None, - ip: str = "dhcp", - netmask: int = 24, - gw: str = "", - ostype: str = "l26", - agent: int = 1, - ciuser: str = "", - cipassword: str = "", - sshkeys: str = "", - node: Optional[str] = None, - start_now: bool = True, - wait: bool = True, - ) -> str: - """Create a new VM. Two modes: clone from a template (default), or create fresh from an ISO. - - Mode 1 (default): clones the configured cloud-init template. Use ciuser/cipassword/sshkeys - to set up the guest user via cloud-init. - Mode 2 (ISO): set template_vmid=0 and provide iso, ip, gw as needed. - - Args: - vmid: New numeric ID for the VM (must not already exist). - name: Name for the new VM. - template_vmid: VMID of a template to clone from. Set to 0 to instead create from an ISO. Defaults to the configured default template if omitted. - memory: RAM in MB. - cores: CPU cores. - disk_size: Disk size, e.g. "32G". Only used in ISO mode. - disk_storage: Storage for the disk. - iso: ISO volume id to boot from, only used in ISO mode, e.g. "local:iso/debian-12.iso". - bridge: Network bridge. Defaults to the configured default bridge if omitted. - ip: IP address, or "dhcp". Only used in ISO mode. - netmask: Netmask prefix length. Only used in ISO mode. - gw: Gateway IP. Only used in ISO mode. - ostype: Guest OS type, e.g. "l26" for Linux 2.6+/3.x/4.x/5.x/6.x. - agent: Whether to enable the QEMU guest agent (1 or 0). Only used in ISO mode. - ciuser: Cloud-init user (clone mode only). - cipassword: Cloud-init password (clone mode only). - sshkeys: Cloud-init SSH public keys (clone mode only). - node: Node name. Defaults to the configured default node if omitted. - start_now: Whether to start the VM right after creation. - wait: Whether to wait for the creation task to finish before returning (ISO mode only; clone mode always waits). - """ - node = node or self.valves.default_node - bridge = bridge or self.valves.default_bridge - try: - # Mode 1: clone from template - if template_vmid is None or template_vmid: - return self.vm_create_from_template( - vmid=vmid, - name=name, - template_vmid=template_vmid, - full=True, - memory=memory, - cores=cores, - storage=disk_storage, - start_now=start_now, - ciuser=ciuser, - cipassword=cipassword, - sshkeys=sshkeys, - node=node, - ) - # Mode 2: create from scratch (ISO) - if not name or not vmid: - return "Error: name and vmid required" - exist_vm = [ - str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/qemu") - ] - exist_lxc = [ - str(x.get("vmid")) for x in self._api("GET", f"/nodes/{node}/lxc") - ] - if str(vmid) in exist_vm + exist_lxc: - return f"Error: VMID {vmid} already exists" - params = { - "vmid": vmid, - "name": name, - "memory": memory, - "cores": cores, - "sockets": 1, - "ostype": ostype, - "agent": str(agent), - } - params["virtio0"] = ( - disk_size if ":" in disk_size else f"{disk_storage}:{disk_size}" - ) - if iso: - params["ide2"] = f"{iso},media=cdrom" - params["boot"] = "order=ide2;virtio0" - net = f"name=eth0,bridge={bridge}" - if ip and ip.lower() != "dhcp": - net += f",ip={ip}/{netmask}" - if gw: - net += f",gw={gw}" - else: - net += ",ip=dhcp" - params["net0"] = net - if start_now: - params["start"] = 1 - result = self._api("POST", f"/nodes/{node}/qemu", data=params) - out = [ - f"OK VM #{vmid}: {name} creating!", - f" CPU: {cores} cores, RAM: {memory} MB", - f" Disk: {params['virtio0']}", - ] - if iso: - out.append(f" ISO: {iso}") - out.append( - f" Net: {bridge}, IP: {'DHCP' if ip == 'dhcp' else f'{ip}/{netmask}'}" - ) - out.append(f" QEMU Agent: {'on' if agent else 'off'}") - upid = result if isinstance(result, str) else "" - if upid: - out.append(f"UPID: {upid}") - if wait and upid: - out.append("Waiting for creation to finish...") - ok, msg = self._wait_for_task(upid, node=node, timeout=120) - out.append( - " Creation done" - if ok - else f" WARN Creation may not have finished: {msg}" - ) - return "\n".join(out) - except Exception as e: - return "ERR " + str(e)