added proxmox power button plugin 0.1

proxmox_power_button/proxmox.py

@@ -0,0 +1,151 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+# Copyright (C) 2026 Bryan Joshua Pedini
+"""
+Thin Proxmox VE client.
+
+Connection details come from the VM's cluster custom fields:
+  * cluster.cf["endpoint"] -> "host", "host:port" or "https://host:port"
+  * cluster.cf["token"]    -> "user@realm!tokenid=secret"
+and the target VM is identified by vm.cf["vmid"].
+"""
+import re
+import time
+from urllib.parse import urlparse
+
+from django.conf import settings
+
+try:
+    from proxmoxer import ProxmoxAPI
+except ImportError:  # pragma: no cover - dependency missing
+    ProxmoxAPI = None
+
+
+class ProxmoxError(Exception):
+    """
+    Any failure resolving config or talking to Proxmox.
+    """
+
+
+def _plugin_settings():
+    return settings.PLUGINS_CONFIG.get("proxmox_power_button", {})
+
+
+def _parse_endpoint(endpoint):
+    endpoint = (endpoint or "").strip()
+    if not endpoint:
+        raise ProxmoxError("cluster endpoint is empty")
+    if "://" in endpoint:
+        parsed = urlparse(endpoint)
+        return parsed.hostname, parsed.port or 8006
+    if ":" in endpoint:
+        host, _, port = endpoint.rpartition(":")
+        return host, int(port)
+    return endpoint, 8006
+
+
+def _parse_token(token):
+    match = re.match(r"^(?P<user>[^!]+)!(?P<name>[^=]+)=(?P<value>.+)$", (token or "").strip())
+    if not match:
+        raise ProxmoxError("token must be in the form 'user@realm!tokenid=secret'")
+    return match.group("user"), match.group("name"), match.group("value")
+
+
+def _command_for(action, cfg):
+    if action == "start":
+        return "start"
+    if action == "stop":
+        return cfg.get("stop_mode", "shutdown")
+    if action == "reboot":
+        return cfg.get("reboot_mode", "reboot")
+    raise ProxmoxError(f"unknown action '{action}'")
+
+
+def _find_by_vmid(prox, vmid):
+    target = int(vmid)
+    visible = []
+    for resource in prox.cluster.resources.get(type="vm"):
+        rid = resource.get("vmid")
+        if rid is None:
+            continue
+        visible.append(int(rid))
+        if int(rid) == target:
+            return resource  # has 'node' and 'type' (qemu|lxc)
+
+    if not visible:
+        # cluster/resources came back empty: almost always the API token has
+        # "Privilege Separation" enabled (default) and no permission granted,
+        # so it can see nothing.
+        raise ProxmoxError(
+            f"VMID {vmid} not found: this API token can see no VMs. Check the "
+            f"token's Privilege Separation / permissions (needs at least "
+            f"VM.Audit and VM.PowerMgmt on the VM, or Sys.Audit on /)."
+        )
+    raise ProxmoxError(
+        f"VMID {vmid} not found among {len(visible)} VM(s) visible to this token "
+        f"(visible VMIDs: {sorted(visible)[:30]}). Wrong cluster endpoint, or the "
+        f"token lacks permission on this VM."
+    )
+
+
+def power_action(vm, action):
+    """
+    Perform start/stop/reboot on the Proxmox VM backing this NetBox VM.
+
+    Raises ProxmoxError on any problem. For action == "start", returns the
+    verified power state ("running" / "stopped" / None if unconfirmed) after a
+    short delay; for stop/reboot returns None.
+    """
+    if ProxmoxAPI is None:
+        raise ProxmoxError("the 'proxmoxer' package is not installed")
+
+    cfg = _plugin_settings()
+
+    cluster = vm.cluster
+    if cluster is None:
+        raise ProxmoxError("VM is not assigned to a cluster")
+
+    endpoint = (cluster.custom_field_data or {}).get("endpoint")
+    token = (cluster.custom_field_data or {}).get("token")
+    vmid = (vm.custom_field_data or {}).get("vmid")
+
+    if not endpoint or not token:
+        raise ProxmoxError(
+            f"cluster '{cluster}' is missing the Proxmox 'endpoint'/'token' custom fields"
+        )
+    if vmid is None:
+        raise ProxmoxError("VM has no VMID set")
+
+    host, port = _parse_endpoint(endpoint)
+    user, token_name, token_value = _parse_token(token)
+    command = _command_for(action, cfg)
+
+    try:
+        prox = ProxmoxAPI(
+            host,
+            port=port,
+            user=user,
+            token_name=token_name,
+            token_value=token_value,
+            verify_ssl=cfg.get("verify_ssl", False),
+        )
+        resource = _find_by_vmid(prox, vmid)
+        node = resource["node"]
+        vtype = resource["type"]  # 'qemu' or 'lxc'
+        status_obj = getattr(prox.nodes(node), vtype)(vmid).status
+        getattr(status_obj, command).post()
+    except ProxmoxError:
+        raise
+    except Exception as exc:  # proxmoxer / network / auth errors
+        raise ProxmoxError(str(exc)) from exc
+
+    # For power-on only, do a quick confirmation call so NetBox status reflects
+    # reality (start is fast). Reboot/shutdown take too long to confirm inline,
+    # so they are not verified here.
+    if action == "start":
+        time.sleep(3)
+        try:
+            current = getattr(prox.nodes(node), vtype)(vmid).status.current.get()
+            return current.get("status")  # 'running' | 'stopped'
+        except Exception:  # verification is best-effort; command already sent
+            return None
+    return None