# Copyright 2025 The NetOrk Project Authors. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """System-level NAPALM getters for Proxmox VE nodes (environment, NTP, SNMP, users, packages, services, updates, disk SMART).""" from __future__ import annotations import base64 import ipaddress import logging import re from typing import Any from napalm_device_types import APT_UPGRADABLE_COMMAND, parse_apt_upgradable from napalm_proxmox import utils logger = logging.getLogger(__name__) _JsonDict = dict[str, Any] #: The subscription API's status (lower-cased) as a napalm-device-types #: ``LicenseStatus``; "new" and "suspended" have none of their own. _SUBSCRIPTION_STATUS = { "notfound": "none", "active": "active", "invalid": "invalid", "expired": "expired", } _PACKAGE_NAME = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_.+:~-]*$") #: What may stand in an snmpd.conf value: no whitespace, no quote, no comment. _SNMP_WORD = re.compile(r'^[^\s"#]+$') def _net(value: str) -> str: return str(ipaddress.ip_network(value.strip(), strict=False)) def render_snmpd_conf(settings: dict) -> str: """net-snmp's configuration for *settings*: read-only v2c, per allowed source.""" community = settings["community"] if not _SNMP_WORD.match(community): raise ValueError("the community must be one word without quotes or #") listen = settings.get("listen") lines = [f"agentAddress udp:{listen + ':' if listen else ''}161"] sources = [_net(s) for s in settings.get("allowed_sources", [])] lines += [f"rocommunity {community} {s}" for s in sources] or [f"rocommunity {community}"] for key, directive in (("location", "sysLocation"), ("contact", "sysContact")): value = settings.get(key, "") if value: if "\n" in value: raise ValueError(f"{key} must be one line") lines.append(f"{directive} {value}") return "\n".join(lines) + "\n" class ProxmoxSystemMixin: """Mixin providing system-level NAPALM methods.""" # ------------------------------------------------------------------ # # get_environment # ------------------------------------------------------------------ # def get_environment(self) -> _JsonDict: """Return environment status (CPU, memory, temperature).""" status = self._get_node_status() env: _JsonDict = { "fans": {}, "temperature": {}, "power": {}, "cpu": {}, "memory": {"available_ram": 0, "used_ram": 0}, } # CPU cpu_usage = status.get("cpu", 0.0) env["cpu"]["0"] = {"%usage": round(float(cpu_usage) * 100, 2)} # Memory (Proxmox reports in bytes) mem = status.get("memory", {}) total = int(mem.get("total", 0) or 0) used = int(mem.get("used", 0) or 0) env["memory"]["available_ram"] = total env["memory"]["used_ram"] = used # Temperature (from node sensors if available) try: sensors = self._node_api().hardware.sensors.get() or [] except Exception as exc: logger.debug("Failed to fetch hardware sensors: %s", exc) sensors = [] for sensor in sensors: name = sensor.get("name", "unknown") value = sensor.get("value", None) if value is not None: try: temp_c = float(value) env["temperature"][name] = { "temperature": temp_c, "is_alert": temp_c >= 80.0, "is_critical": temp_c >= 95.0, } except (TypeError, ValueError): pass return env # ------------------------------------------------------------------ # # get_ntp_servers / get_ntp_stats # ------------------------------------------------------------------ # def get_ntp_servers(self) -> dict[str, _JsonDict]: """Return configured NTP servers.""" ntp = self._get_node_ntp() servers: dict[str, _JsonDict] = {} # Proxmox reports a comma-separated or space-separated server list raw = ntp.get("server", "") or ntp.get("servers", "") for srv in re.split(r"[\s,]+", raw): srv = srv.strip() if srv: servers[srv] = {} return servers def get_ntp_stats(self) -> list[_JsonDict]: """Return NTP synchronisation statistics from chronyc/ntpq output.""" raw = self._exec_ssh_command( "chronyc -n tracking 2>/dev/null || ntpq -pn 2>/dev/null || true" ) stats: list[_JsonDict] = [] for line in raw.splitlines(): line = line.strip() # ntpq -pn format: *remote refid st t when poll reach delay offset jitter m = re.match( r"^([\*\+\-\s])([\d.]+)\s+([\d.]+)\s+(\d+)\s+\S+\s+(\S+)\s+(\d+)\s+(\d+)\s+([\d.]+)\s+([-\d.]+)\s+([\d.]+)", line, ) if m: synced = m.group(1).strip() == "*" stats.append( { "remote": m.group(2), "referenceid": m.group(3), "synchronized": synced, "stratum": int(m.group(4)), "type": "", "when": m.group(5), "hostpoll": int(m.group(6)), "reachability": int(m.group(7)), "delay": float(m.group(8)), "offset": float(m.group(9)), "jitter": float(m.group(10)), } ) return stats # ------------------------------------------------------------------ # # get_snmp_information # ------------------------------------------------------------------ # def get_snmp_information(self) -> _JsonDict: """Return SNMP information. Proxmox does not expose SNMP configuration via the REST API. We read /etc/snmp/snmpd.conf via exec if available. """ raw = self._exec_ssh_command( "cat /etc/snmp/snmpd.conf 2>/dev/null || true" ) communities: dict[str, _JsonDict] = {} location = "" contact = "" for line in raw.splitlines(): line = line.strip() if line.startswith("#") or not line: continue # rocommunity [source] m = re.match(r"^(ro|rw)community\s+(\S+)", line) if m: mode = "ro" if m.group(1) == "ro" else "rw" community = m.group(2) communities[community] = {"acl": "N/A", "mode": mode} m_loc = re.match(r"^sysLocation\s+(.+)", line) if m_loc: location = m_loc.group(1).strip() m_con = re.match(r"^sysContact\s+(.+)", line) if m_con: contact = m_con.group(1).strip() return { "chassis_id": self._node_name, "community": communities, "contact": contact, "location": location, } # ------------------------------------------------------------------ # # get_users # ------------------------------------------------------------------ # def get_users(self) -> dict[str, _JsonDict]: """Return users configured on the Proxmox node. Reads from both the Proxmox access/users API and local /etc/passwd. """ result: dict[str, _JsonDict] = {} try: pve_users = self._api.access.users.get() or [] # type: ignore[union-attr] except Exception as exc: logger.debug("Failed to fetch Proxmox users: %s", exc) pve_users = [] for user in pve_users: uid = user.get("userid", "") if not uid: continue # Proxmox roles: Administrator -> 15, otherwise 1 groups = user.get("groups", "") or "" level = 1 try: roles = self._api.access.users(uid).get() or {} # type: ignore[union-attr] if "Administrator" in str(roles): level = 15 except Exception as exc: logger.debug("Failed to fetch roles for user %s: %s", uid, exc) result[uid] = { "level": level, "password": "", "sshkeys": [], } # Merge local OS users from /etc/passwd raw = self._exec_ssh_command("getent passwd 2>/dev/null || cat /etc/passwd") for line in raw.splitlines(): parts = line.split(":") if len(parts) < 7: continue uname, _, uid_str, *_ = parts try: uid_int = int(uid_str) except ValueError: continue if uname not in result and uid_int < 1000 or uid_int == 0: result[uname] = { "level": 15 if uid_int == 0 else 0, "password": "", "sshkeys": [], } return result # ------------------------------------------------------------------ # # Kernel facts (KernelFactsMixin supplies get_kernel_facts) # ------------------------------------------------------------------ # def _run_kernel_facts_command(self, command: str) -> str: """The transport for ``KernelFactsMixin.get_kernel_facts``: the exec path.""" return self._exec_ssh_command(command) def _run_service_packages_command(self, command: str) -> str: """The transport for ``ServicePackagesMixin.get_service_packages``: the exec path.""" return str(self._exec_ssh_command(command)) # ------------------------------------------------------------------ # # Listening sockets (ListeningSocketsMixin supplies get_listening_sockets) # ------------------------------------------------------------------ # def _run_system_identity_command(self, command: str, *, privileged: bool) -> str: """The transport for ``SystemIdentityMixin.get_system_identity``: the exec path, which runs as root either way.""" return str(self._exec_ssh_command(command)) def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str: """The transport for ``ListeningSocketsMixin.get_listening_sockets``: the exec path, which runs as root either way.""" return str(self._exec_ssh_command(command)) # ------------------------------------------------------------------ # # Packages (Debian APT) # ------------------------------------------------------------------ # def get_packages(self) -> list[_JsonDict]: """Return installed Debian packages with available-update info. Installed list comes from ``dpkg-query`` via SSH (the Proxmox API ``/apt/installed`` endpoint is not implemented on PVE 8.x). Available updates come from the Proxmox API ``/apt/update``. """ # Available updates from Proxmox API (keyed by package name) upgradable: dict[str, str] = {} try: for upd in self._api.nodes(self._node_name).apt.update.get(): pkg = upd.get("Package", "") if pkg: upgradable[pkg] = upd.get("Version", "") except Exception as exc: logger.debug("Failed to fetch available APT updates: %s", exc) # Installed packages via SSH dpkg-query raw = self._exec_ssh_command( "dpkg-query -W -f=" "'${Package}\\t${Version}\\t${db:Status-Status}\\t${Installed-Size}" "\\t${source:Package}\\t${source:Version}\\n'" " 2>/dev/null" ) result: list[_JsonDict] = [] for line in raw.splitlines(): parts = line.strip().split("\t") if len(parts) < 2: continue name = parts[0] version = parts[1] if len(parts) > 1 else "" status = parts[2] if len(parts) > 2 else "installed" size_kb = parts[3] if len(parts) > 3 else "0" # Debian source package (differs from the binary for split packages, # e.g. openssh-server → openssh). Needed for accurate OSV matching. source_package = parts[4] if len(parts) > 4 and parts[4] else name # And its version, which is a different number from this package's. # # OSV states Debian ranges in *source* versions, so a consumer given # the source package and only the binary version compares two # unrelated numbers: libldb2 is 2:2.11.0+samba4.22.11+dfsg-… while # its source, samba, is 2:4.22.11+dfsg-…. Reporting the source # without its version is worse than reporting neither, because it # looks usable. Every package on all three Proxmox nodes was in that # state — the format string never asked for the field. # # dpkg leaves it empty when it equals `Version`, and so does an # older dpkg that does not know the field at all. source_version = parts[5] if len(parts) > 5 and parts[5] else version if not name or status != "installed": continue size_bytes = int(size_kb) * 1024 if size_kb.isdigit() else 0 result.append({ "name": name, "version": version, "installed": True, "description": "", "size": size_bytes, "source": "pve", "source_package": source_package, "source_version": source_version, "upgrade_version": upgradable.get(name, ""), }) return result # ------------------------------------------------------------------ # # Licenses (napalm-device-types' LicenseMixin) # ------------------------------------------------------------------ # def get_licenses(self) -> list[_JsonDict]: """The node's subscription, as the API reports it. A fact for the caller: whether running without one is a problem is not the driver's to say. An unreadable subscription reports nothing. """ sub = self._get_node_subscription() raw = str(sub.get("status", "")) if not raw: return [] lic: _JsonDict = { "name": "pve-subscription", "status": _SUBSCRIPTION_STATUS.get(raw.lower(), "unknown"), "detail": sub.get("message") or raw, } if sub.get("productname"): lic["edition"] = sub["productname"] if sub.get("nextduedate"): lic["expires"] = sub["nextduedate"] return [lic] # ------------------------------------------------------------------ # # Services (systemd, through napalm-device-types' SystemdServicesMixin) # ------------------------------------------------------------------ # def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str: """The transport for ``SystemdServicesMixin``: the exec path, as root. *privileged* needs nothing more on a node the driver reaches as root, and the exec path keeps its own timeout. """ return str(self._exec_ssh_command(command)) # ------------------------------------------------------------------ # # Available updates # ------------------------------------------------------------------ # def get_available_updates(self) -> list[_JsonDict]: """Return the node's upgradable packages, with origin and security status. ``apt list --upgradable`` over the exec path names each candidate's suite (``trixie-security``); the APT API names only an Origin ("Debian", "Proxmox") and is the fallback, with the security status unknown. :raises Exception: when neither answers -- never an empty list for "could not read". """ try: updates = parse_apt_upgradable(self._exec_ssh_command(APT_UPGRADABLE_COMMAND) or "") except ValueError as exc: logger.debug("apt list over the exec path failed, using the API: %s", exc) updates = self._updates_from_api() return sorted(updates, key=lambda u: u["name"]) def _updates_from_api(self) -> list[_JsonDict]: return [ { "name": upd["Package"], "current_version": upd.get("OldVersion", ""), "new_version": upd.get("Version", ""), "origin": upd.get("Origin"), "security": None, } for upd in self._api.nodes(self._node_name).apt.update.get() # type: ignore[union-attr] if upd.get("Package") ] def refresh_available_updates(self) -> _JsonDict: """Resynchronise the node's package index (``POST nodes/{n}/apt/update``).""" try: task = self._api.nodes(self._node_name).apt.update.post() # type: ignore[union-attr] except Exception as exc: return {"success": False, "output": str(exc)} return {"success": True, "output": f"Package index refresh started ({task})"} def _run_host_status_command(self, command: str) -> str: """The transport for ``HostStatusMixin.get_host_status``: the exec path.""" return str(self._exec_ssh_command(command)) def apply_updates(self, packages: list[str]) -> _JsonDict: """Upgrade the given packages via ``apt-get install`` over SSH.""" for pkg in packages: if not re.match(r'^[a-zA-Z0-9_\-\+\.]+$', pkg): raise ValueError(f"Invalid package name: {pkg!r}") pkg_args = " ".join(packages) output = self._exec_ssh_command( f"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y {pkg_args} 2>&1 || true" ) return {"success": True, "output": output} # ------------------------------------------------------------------ # # SNMP / Health # ------------------------------------------------------------------ # def get_snmp_config(self): """Return SNMP agent config if snmpd is installed and running on the node. Uses _exec_ssh_command (Proxmox API exec or SSH) to inspect the node. Returns a SNMPConfigDict or None. """ try: from napalm_device_types.models import SNMPConfigDict except ImportError: return None running = ( self._exec_ssh_command("systemctl is-active snmpd 2>/dev/null || true").strip() == "active" ) if not running: return None community = "public" port = 161 try: conf = self._exec_ssh_command( "grep -E '^[[:space:]]*(ro|rw)?community' /etc/snmp/snmpd.conf 2>/dev/null | head -5" ) for line in conf.splitlines(): parts = line.split() if not parts: continue kw = parts[0].lower() if kw in ("rocommunity", "rwcommunity") and len(parts) >= 2: community = parts[1] break elif kw == "com2sec" and len(parts) >= 4: community = parts[3] break except Exception as exc: logger.debug("Failed to parse snmpd.conf: %s", exc) return SNMPConfigDict(running=True, community=community, port=port, version="2c") # ------------------------------------------------------------------ # Fixes (napalm-device-types SnmpAgentMixin, EnsurePackageMixin): the # caller's values. The node's shell runs as root. # ------------------------------------------------------------------ def _rc(self, command: str) -> tuple[str, int]: """Run *command* and return its output and exit status (-1 when unknown).""" out = self._exec_ssh_command(f"{{ {command}; }} 2>&1; echo __PVE_RC=$?") text, _, tail = out.rpartition("__PVE_RC=") try: return text.strip(), int(tail.strip()) except ValueError: return out.strip(), -1 def _step(self, command: str, done: str) -> dict: output, status = self._rc(command) return {"ok": status == 0, "output": done if status == 0 else output[-300:]} def install_package(self, name: str, version: str = "") -> None: """apt-get install *name* (``PackageManagementMixin``); raises when apt fails.""" if not _PACKAGE_NAME.match(name) or (version and not _PACKAGE_NAME.match(version)): raise ValueError(f"not a package name: {name!r}") target = f"{name}={version}" if version else name output, status = self._rc(f"DEBIAN_FRONTEND=noninteractive apt-get install -y {target}") if status != 0: raise RuntimeError(f"apt-get install {target} failed: {output[-300:]}") def get_management_peer(self) -> str | None: """The address the node sees the SSH connection come from (``SSH_CONNECTION``).""" out = self._exec_ssh_command('printf "%s\\n" "$SSH_CONNECTION"').split() try: return str(ipaddress.ip_address(out[0])) if out else None except ValueError: return None def _install_snmp_agent(self) -> dict: return self._step( "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp", "net-snmp installed" ) def _write_snmp_agent_config(self, settings: dict) -> dict: encoded = base64.b64encode(render_snmpd_conf(settings).encode()).decode() return self._step( f"umask 077 && t=$(mktemp) && echo {encoded} | base64 -d > \"$t\" && " f"install -m 644 -o root -g root \"$t\" /etc/snmp/snmpd.conf; " "s=$?; rm -f \"$t\"; exit $s", "wrote /etc/snmp/snmpd.conf", ) def _allow_snmp_sources(self, settings: dict) -> dict: sources = [_net(s) for s in settings.get("allowed_sources", [])] if not sources: return {"ok": True, "output": "no sources given; firewall unchanged"} if self._rc("command -v ufw")[1] != 0: return {"ok": True, "output": "no ufw; pve-firewall rules are the operator's"} rules = " && ".join(f"ufw allow from {s} to any port 161 proto udp" for s in sources) return self._step(rules, f"udp/161 open for {', '.join(sources)}") def _start_snmp_agent(self) -> dict: return self._step( "systemctl enable snmpd && systemctl restart snmpd && systemctl is-active snmpd", "snmpd enabled and running", ) def set_hostname(self, new_hostname: str) -> None: """Set the system hostname on the Proxmox node. Performs all steps required for a clean rename on a Debian/Proxmox host: 1. ``/etc/hostname`` — short hostname only. 2. ``/etc/hosts`` — old hostname replaced by new (short or FQDN). 3. ``/etc/mailname`` — FQDN, if the file exists. 4. Postfix ``myhostname`` — updated via ``postconf -e`` if Postfix is installed. 5. Runtime hostname — applied immediately via ``hostname(1)`` so SSH and the Proxmox API see the new name without a reboot. 6. ``/etc/pve/nodes/`` → ``/etc/pve/nodes/`` rename so VM/CT configs are found after the reboot under the new node name. 7. Proxmox node TLS certificate — regenerated via ``pvenode cert create --overwrite``. **Important**: the Proxmox *node name* (shown in the web UI and stored in ``/etc/pve/nodes//``) only changes after a full reboot. Until then the web UI still shows the old node name, but the OS-level hostname and the SSL certificate already reflect the new value. Accepts a bare hostname (``pve-see``) or an FQDN (``pve-see.see.local``). """ import base64 as _b64 if "." in new_hostname: short_hostname, domain = new_hostname.split(".", 1) fqdn = new_hostname else: short_hostname = new_hostname domain = None fqdn = new_hostname # Derive old short hostname for substitution in /etc/hosts old_short = self._exec_ssh_command( "cat /etc/hostname 2>/dev/null || hostname -s 2>/dev/null" ).strip().split(".")[0] # ── 1. /etc/hostname ───────────────────────────────────────────── hostname_b64 = _b64.b64encode(f"{short_hostname}\n".encode()).decode() self._exec_ssh_command(f"echo {hostname_b64} | base64 -d > /etc/hostname") # ── 2. /etc/hosts ──────────────────────────────────────────────── # Read the file, substitute in Python (safe for all characters), # write back via base64 to avoid shell-escaping issues. if old_short and old_short != short_hostname: hosts_raw = self._exec_ssh_command("cat /etc/hosts 2>/dev/null") # Replace whole-word occurrences of the old short hostname. # Both "hostname" and "hostname.domain" forms are covered because # the new value is the full fqdn when a domain was given. import re as _re new_replace = fqdn if domain else short_hostname hosts_new = _re.sub( r"\b" + _re.escape(old_short) + r"\b", new_replace, hosts_raw, ) hosts_b64 = _b64.b64encode(hosts_new.encode()).decode() self._exec_ssh_command( f"echo {hosts_b64} | base64 -d > /tmp/_netork_hosts" f" && mv /tmp/_netork_hosts /etc/hosts" f" && chmod 644 /etc/hosts" ) # ── 3. /etc/mailname ───────────────────────────────────────────── mailname_b64 = _b64.b64encode(f"{fqdn}\n".encode()).decode() self._exec_ssh_command( f"[ -f /etc/mailname ]" f" && echo {mailname_b64} | base64 -d > /etc/mailname" f" || true" ) # ── 4. Postfix myhostname ───────────────────────────────────────── self._exec_ssh_command( f"command -v postconf >/dev/null 2>&1" f" && postconf -e 'myhostname = {fqdn}'" f" || true" ) # ── 5. Apply hostname at runtime ───────────────────────────────── self._exec_ssh_command(f"hostname '{short_hostname}'") # ── 6. Migrate /etc/pve/nodes/ → /etc/pve/nodes/ ────── # Proxmox stores VM and CT configs under /etc/pve/nodes//. # This rename MUST happen before the reboot while pve-cluster is # running and pmxcfs is mounted — pmxcfs supports the rename live. # Without this step the node boots under the new hostname, finds an # empty /etc/pve/nodes// and all VMs appear missing. if old_short and old_short != short_hostname: pve_old = f"/etc/pve/nodes/{old_short}" pve_new = f"/etc/pve/nodes/{short_hostname}" migrate_out = self._exec_ssh_command( f"if [ -d '{pve_old}' ] && [ ! -e '{pve_new}' ]; then" f" mv '{pve_old}' '{pve_new}'" f" && echo 'migrated {pve_old} -> {pve_new}';" f"elif [ -d '{pve_new}' ]; then" f" echo 'target already exists, skipping migration';" f"else" f" echo 'source {pve_old} not found, skipping migration';" f"fi" ) logger.info("Proxmox node dir migration: %s", migrate_out) # ── 7. Regenerate Proxmox node TLS certificate ─────────────────── # Use pvenode cert create --overwrite rather than pvecm updatecerts -f. # pvecm updatecerts -f restarts pve-cluster which briefly unmounts the # cluster filesystem (/etc/pve) and can crash running VMs. # pvenode cert create --overwrite only touches the node's own cert and # restarts pveproxy (safe while VMs are running). cert_out = self._exec_ssh_command( "pvenode cert create --overwrite 1 2>&1" " || { systemctl restart pveproxy 2>&1; echo 'restarted pveproxy'; }" " || true" ) logger.info("Proxmox cert regeneration: %s", cert_out[:200]) logger.info( "Proxmox set_hostname: %s → %s (fqdn: %s). " "Reboot required for node name to update in web UI / cluster.", old_short, short_hostname, fqdn, ) # ------------------------------------------------------------------ # # _send_command — generic SSH command helper used by netOrk core # ------------------------------------------------------------------ # def _send_command(self, command: str) -> str: """Execute *command* on the node and return its output. Delegates to _exec_ssh_command so that netOrk's generic helpers (e.g. the reboot action) work without knowing the driver internals. """ return self._exec_ssh_command(command) def get_disk_smart(self) -> dict: """Return SMART health data for all disks on this Proxmox node. Combines the disk list (model, size, wearout) with per-disk SMART data (health, temperature, percentage used). Returns a dict keyed by device path, e.g. {"/dev/nvme0n1": {...}}. """ import re as _re result: dict = {} try: disks = self._node_api().disks.list.get() or [] except Exception as exc: logger.debug("Failed to list disks: %s", exc) return result for disk in disks: dev = disk.get("devpath") or disk.get("dev") if not dev: continue entry: dict = { "model": disk.get("model", ""), "serial": disk.get("serial", ""), "type": disk.get("type", ""), "size": disk.get("size", 0), "health": disk.get("health", "unknown").lower(), "wearout": disk.get("wearout"), # NVMe wear indicator 0-100 "temperature": None, "percentage_used": None, "available_spare": None, "reallocated_sectors": None, "power_on_hours": None, } try: smart = self._node_api().disks.smart.get(disk=dev) or {} # health from SMART endpoint may be more accurate if smart.get("health"): entry["health"] = smart["health"].lower() text = smart.get("text", "") # Parse temperature m = _re.search(r"Temperature[^:]*:\s*(\d+)\s*Celsius", text) if m: entry["temperature"] = int(m.group(1)) # NVMe-specific m = _re.search(r"Percentage Used:\s*(\d+)%", text) if m: entry["percentage_used"] = int(m.group(1)) m = _re.search(r"Available Spare:\s*(\d+)%", text) if m: entry["available_spare"] = int(m.group(1)) m = _re.search(r"Power On Hours:\s*([\d,]+)", text) if m: entry["power_on_hours"] = int(m.group(1).replace(",", "")) # HDD-specific SMART attributes for attr in smart.get("attributes", []): name = attr.get("name", "").lower() raw = attr.get("raw", "") try: raw_int = int(str(raw).split()[0]) except (ValueError, TypeError): raw_int = None if "temperature" in name and raw_int is not None: entry["temperature"] = raw_int elif "reallocated" in name and "sector" in name and raw_int is not None: entry["reallocated_sectors"] = raw_int elif "power_on" in name and raw_int is not None: entry["power_on_hours"] = raw_int except Exception as exc: logger.debug("Failed to fetch SMART data for %s: %s", dev, exc) result[dev] = entry return result def get_system_config(self) -> dict[str, Any]: """Return system-level configuration for this Proxmox node. Collected from ``/cluster/status`` and the node config API: * ``cluster_name`` (str | None) — Proxmox cluster name; ``None`` when the node is standalone (not part of a cluster). * ``cluster_nodes`` (list[str]) — hostnames of all online cluster nodes. * ``hostname`` (str) — this node's name as reported by the cluster. * ``timezone`` (str | None) — e.g. ``"Europe/Berlin"``. * ``ntp_servers`` (list[str]). * ``ssh_port`` (int) — always 22 for Proxmox nodes. * ``ssh_password_auth`` (bool) — ``False`` as a safe default. """ cfg: dict[str, Any] = { "cluster_name": None, "cluster_nodes": [], "hostname": self._node_name, "timezone": None, "ntp_servers": [], "ssh_port": 22, "ssh_password_auth": False, } # Cluster name and node list from /cluster/status try: status = self._api.cluster.status.get() or [] for entry in status: if entry.get("type") == "cluster": cfg["cluster_name"] = entry.get("name") elif entry.get("type") == "node" and entry.get("online"): cfg["cluster_nodes"].append(entry.get("name", "")) except Exception as exc: logger.debug("Failed to fetch cluster status for system config: %s", exc) # Timezone from node time API try: time_cfg = self._node_api().time.get() or {} cfg["timezone"] = time_cfg.get("timezone") or None except Exception as exc: logger.debug("Failed to fetch node time config: %s", exc) return cfg