#!/usr/bin/env python3 """ proxmox_tools_pve1.py - Proxmox API Tools (PVE1) Description: Full toolset for Proxmox VE management (second server): - LXC (create, start, stop, delete) - VM (clone from template, create from ISO, manage) - Monitoring (nodes, storage, resources, tasks) Usage: Import and call functions directly. import proxmox_tools_pve1 as pve1 print(pve1.pve1_connection()) """ import json import re import time import requests # ============================================================ # CONFIG - PVE1 (192.168.31.4) # ============================================================ PROXMOX_HOST = "https://192.168.31.4:8006" PROXMOX_USER = "root@pam" PROXMOX_TOKEN_NAME = "open-webui" PROXMOX_TOKEN_VALUE = "be44caad-e658-4929-9b24-c12b21495760" VERIFY_SSL = False DEFAULT_NODE = "pve1" DEFAULT_STORAGE = "local" DEFAULT_BRIDGE = "vmbr0" _base_url = PROXMOX_HOST.rstrip("/") + "/api2/json" _auth_header = { "Authorization": f"PVEAPIToken={PROXMOX_USER}!{PROXMOX_TOKEN_NAME}={PROXMOX_TOKEN_VALUE}" } def _api(method, path, params=None, data=None): """Execute a Proxmox API request.""" url = _base_url + path resp = requests.request( method=method, url=url, headers=_auth_header, params=params, json=data, verify=VERIFY_SSL, timeout=30, ) resp.raise_for_status() return resp.json().get("data", {}) def _fmt(b): """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" def _pct(u, t): """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"{_fmt(u)} / {_fmt(t)} ({u/t*100:.1f}%)" def _uptime(s): """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(upid, timeout=60): """Wait for a Proxmox task to complete by UPID.""" if not upid or not isinstance(upid, str) or not upid.startswith("UPID:"): return True, "no UPID to wait" node = DEFAULT_NODE start = time.time() while time.time() - start < timeout: try: tasks = _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" # ============================================================ # FUNCTIONS # ============================================================ def pve1_connection(): """Check Proxmox connection.""" try: v = _api("GET", "/version") return "OK Proxmox VE " + v.get("version", "?") + " | node: pve1 | IP: 192.168.31.4" except Exception as e: return "ERR " + str(e) def pve1_nodes(): """Show all cluster nodes.""" try: items = _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: {_pct(n.get('mem'), n.get('maxmem'))}") out.append(f" Uptime: {_uptime(n.get('uptime'))}") return "\n".join(out) except Exception as e: return "ERR " + str(e) def pve1_node_status(node=DEFAULT_NODE): """Show detailed node status.""" try: s = _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: {_pct(m.get('used'), m.get('total'))}") sw = s.get("swap", {}) if sw.get("total", 0): out.append(f"SWAP: {_pct(sw.get('used'), sw.get('total'))}") r = s.get("rootfs", {}) if r: out.append(f"Disk: {_pct(r.get('used'), r.get('total'))}") out.append(f"Uptime: {_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 pve1_cluster(): """Show cluster status (quorum, members).""" try: items = _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 pve1_resources(ftype=""): """Show all cluster resources.""" try: p = {"type": ftype} if ftype else {} items = _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:{_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 pve1_storage(): """Show all storages.""" try: items = _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: {_pct(s.get('used'), s.get('total'))}") out.append(f" Free: {_fmt(s.get('avail'))}") out.append("") return "\n".join(out) except Exception as e: return "ERR " + str(e) def pve1_tasks(limit=10, node=DEFAULT_NODE): """Show recent tasks on node.""" try: items = _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 pve1_lxc_list(node=DEFAULT_NODE): """Show LXC container list.""" try: items = _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:{_pct(c.get('mem'), c.get('maxmem'))}") if c.get("maxdisk"): out.append(f" Disk:{_pct(c.get('disk'), c.get('maxdisk'))}") if c.get("uptime"): out.append(f" Uptime:{_uptime(c['uptime'])}") out.append("") return "\n".join(out) except Exception as e: return "ERR " + str(e) def pve1_lxc_detail(vmid, node=DEFAULT_NODE): """Show detailed LXC info.""" try: s = _api("GET", f"/nodes/{node}/lxc/{vmid}/status/current") c = _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: {_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: {_uptime(s['uptime'])}") return "\n".join(out) except Exception as e: return "ERR " + str(e) def pve1_lxc_action(vmid, action, node=DEFAULT_NODE): """Manage LXC: start, stop, shutdown, restart, suspend, resume, delete.""" 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 = _api("DELETE", f"/nodes/{node}/lxc/{vmid}") else: result = _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 pve1_templates(storage="local", node=DEFAULT_NODE): """Show available LXC templates in storage.""" try: content = _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', '?')} - {_fmt(t.get('size'))}") return "\n".join(out) except Exception as e: return "ERR " + str(e) def pve1_lxc_create(vmid, hostname, ostemplate, password="", sshkey="", storage=DEFAULT_STORAGE, cores=1, memory=512, swap=0, disk="8G", bridge=DEFAULT_BRIDGE, ip="dhcp", netmask=24, gw="", dns="", domain="", node=DEFAULT_NODE, unpriv=True, start_now=True): """Create a new LXC container.""" try: if not hostname: return "Error: hostname required" if not password and not sshkey: return "Error: password or sshkey required" exist = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] exist += [str(x.get("vmid")) for x in _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 = _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}'}") if isinstance(result, str) and result: out.append(f"UPID: {result}") return "\n".join(out) except Exception as e: return "ERR " + str(e) def pve1_vm_list(node=DEFAULT_NODE): """Show all VM (QEMU) list.""" try: items = _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:{_pct(v.get('mem'), v.get('maxmem'))}") if v.get("uptime"): out.append(f" Uptime:{_uptime(v['uptime'])}") out.append("") return "\n".join(out) except Exception as e: return "ERR " + str(e) def pve1_vm_detail(vmid, node=DEFAULT_NODE): """Show detailed VM info.""" try: s = _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: {_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: {_pct(s.get('disk'), s.get('maxdisk'))}") if s.get("uptime"): out.append(f"Uptime: {_uptime(s['uptime'])}") return "\n".join(out) except Exception as e: return "ERR " + str(e) def pve1_vm_action(vmid, action, node=DEFAULT_NODE): """Manage VM: start, stop, shutdown, restart, suspend, resume, delete.""" 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 = _api("DELETE", f"/nodes/{node}/qemu/{vmid}") else: result = _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 pve1_vm_create_from_template(vmid, name, template_vmid=1000, full=True, memory=None, cores=None, storage=None, target_node=None, pool=None, start_now=True, ciuser="", cipassword="", sshkeys="", node=DEFAULT_NODE): """Create a new VM by cloning from template. Default: VM #1000 (cloud-init).""" try: if not vmid: return "Error: vmid required" if not name: return "Error: name required" exist_vm = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] exist_lxc = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] if str(vmid) in exist_vm + exist_lxc: return f"Error: VMID {vmid} already exists" 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 = _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}") out.append("Waiting for clone to finish...") ok, msg = _wait_for_task(upid, timeout=120) out.append(f" {'Clone done' if ok else f'WARN Clone may not have finished: {msg}'}") 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: _api("PUT", f"/nodes/{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}") if start_now: out.append("Starting VM...") try: _api("POST", f"/nodes/{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 pve1_vm_create(vmid, name, template_vmid=1000, memory=4096, cores=2, disk_size="32G", disk_storage="local-lvm", iso="", bridge=DEFAULT_BRIDGE, ip="dhcp", netmask=24, gw="", ostype="l26", agent=1, ciuser="", cipassword="", sshkeys="", node=DEFAULT_NODE, start_now=True): """Universal VM creation. Mode 1: clone from template. Mode 2: from ISO (template_vmid=None).""" try: if template_vmid: return pve1_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) if not name or not vmid: return "Error: name and vmid required" exist_vm = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] exist_lxc = [str(x.get("vmid")) for x in _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 = _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'}") if isinstance(result, str) and result: out.append(f"UPID: {result}") return "\n".join(out) except Exception as e: return "ERR " + str(e) # ============================================================ # TEST # ============================================================ if __name__ == "__main__": print(pve1_connection()) print() print(pve1_vm_list()) print() print(pve1_storage())