138 lines
4.7 KiB
Python
138 lines
4.7 KiB
Python
"""Proxmox REST API client for querying infrastructure state."""
|
|
|
|
import requests
|
|
import urllib3
|
|
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
PROXMOX_HOSTS = {
|
|
"pve-hetzner": "100.88.230.59",
|
|
"pve1": "100.122.56.60",
|
|
"pve3": "100.109.101.12",
|
|
}
|
|
|
|
|
|
class ProxmoxClient:
|
|
def __init__(self, host_ip: str, user: str = "root@pam",
|
|
password: str = "", token_name: str = "", token_value: str = ""):
|
|
self.base_url = f"https://{host_ip}:8006/api2/json"
|
|
self.user = user
|
|
self.password = password
|
|
self.token_name = token_name
|
|
self.token_value = token_value
|
|
self._ticket = None
|
|
self._csrf = None
|
|
|
|
def _auth_header(self) -> dict:
|
|
if self.token_name and self.token_value:
|
|
return {"Authorization": f"PVEAPIToken={self.user}!{self.token_name}={self.token_value}"}
|
|
if self._ticket:
|
|
return {}
|
|
try:
|
|
r = requests.post(
|
|
f"{self.base_url}/access/ticket",
|
|
data={"username": self.user, "password": self.password},
|
|
verify=False, timeout=10,
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()["data"]
|
|
self._ticket = data["ticket"]
|
|
self._csrf = data["CSRFPreventionToken"]
|
|
except requests.RequestException as e:
|
|
raise ConnectionError(f"Proxmox auth failed for {self.base_url}: {e}")
|
|
return {}
|
|
|
|
def _get(self, path: str) -> dict:
|
|
headers = self._auth_header()
|
|
cookies = {}
|
|
if self._ticket:
|
|
cookies["PVEAuthCookie"] = self._ticket
|
|
headers["CSRFPreventionToken"] = self._csrf
|
|
r = requests.get(
|
|
f"{self.base_url}{path}",
|
|
cookies=cookies, headers=headers,
|
|
verify=False, timeout=10,
|
|
)
|
|
r.raise_for_status()
|
|
return r.json().get("data", {})
|
|
|
|
def get_node_status(self) -> dict:
|
|
nodes = self._get("/nodes")
|
|
if isinstance(nodes, list):
|
|
return nodes[0] if nodes else {}
|
|
return nodes
|
|
|
|
def get_containers(self) -> list[dict]:
|
|
nodes = self._get("/nodes")
|
|
if not isinstance(nodes, list):
|
|
return []
|
|
node_name = nodes[0]["node"]
|
|
return self._get(f"/nodes/{node_name}/lxc")
|
|
|
|
def get_container_status(self, vmid: int) -> dict:
|
|
nodes = self._get("/nodes")
|
|
if not isinstance(nodes, list):
|
|
return {"error": "no nodes"}
|
|
node_name = nodes[0]["node"]
|
|
return self._get(f"/nodes/{node_name}/lxc/{vmid}/status/current")
|
|
|
|
|
|
def get_all_containers(passwords: dict = None, tokens: dict = None) -> list[dict]:
|
|
"""Query all Proxmox hosts and return combined container list."""
|
|
if passwords is None:
|
|
passwords = {}
|
|
if tokens is None:
|
|
tokens = {}
|
|
|
|
all_cts = []
|
|
for host_name, host_ip in PROXMOX_HOSTS.items():
|
|
token = tokens.get(host_name, {})
|
|
pw = passwords.get(host_name, passwords.get("default", ""))
|
|
try:
|
|
client = ProxmoxClient(
|
|
host_ip, password=pw,
|
|
token_name=token.get("name", ""),
|
|
token_value=token.get("value", ""),
|
|
)
|
|
containers = client.get_containers()
|
|
for ct in containers:
|
|
ct["_host"] = host_name
|
|
ct["_host_ip"] = host_ip
|
|
all_cts.extend(containers)
|
|
except Exception as e:
|
|
all_cts.append({
|
|
"_host": host_name,
|
|
"_host_ip": host_ip,
|
|
"error": str(e),
|
|
})
|
|
return all_cts
|
|
|
|
|
|
def format_containers(containers: list[dict]) -> str:
|
|
"""Format container list for human/LLM consumption."""
|
|
if not containers:
|
|
return "No containers found."
|
|
|
|
lines = []
|
|
current_host = None
|
|
for ct in sorted(containers, key=lambda c: (c.get("_host", ""), c.get("vmid", 0))):
|
|
host = ct.get("_host", "unknown")
|
|
if host != current_host:
|
|
current_host = host
|
|
lines.append(f"\n## {host}")
|
|
lines.append("| CT | Name | Status | CPU | RAM (MB) |")
|
|
lines.append("|---|---|---|---|---|")
|
|
|
|
if "error" in ct:
|
|
lines.append(f"| — | ERROR | {ct['error'][:60]} | — | — |")
|
|
continue
|
|
|
|
vmid = ct.get("vmid", "?")
|
|
name = ct.get("name", "?")
|
|
status = ct.get("status", "?")
|
|
cpus = ct.get("cpus", "?")
|
|
mem_mb = ct.get("mem", 0) // (1024 * 1024) if ct.get("mem") else 0
|
|
maxmem_mb = ct.get("maxmem", 0) // (1024 * 1024) if ct.get("maxmem") else 0
|
|
lines.append(f"| {vmid} | {name} | {status} | {cpus} | {mem_mb}/{maxmem_mb} |")
|
|
|
|
return "\n".join(lines)
|