netbox-proxmox-power-button/proxmox_power_button/proxmox.py

# 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