"""Read-only NAPALM driver for BMCs over IPMI 2.0, for boards that refuse or lack Redfish. IPMI over LAN gives a password hash to anyone who names a valid user (CVE-2013-4786), so this is the fallback, not the main path. It only reads: facts and inventory from the FRU, sensors, the power state and the system event log. It neither switches the server's power nor resets the BMC; those go through Redfish. """ from __future__ import annotations from typing import Any from napalm.base.exceptions import ConnectionException from napalm_device_types import HealthMetricsMixin, ManagementControllerDriver from pyghmi.ipmi.command import Command from pyghmi.ipmi.private.session import Session _JsonDict = dict[str, Any] #: pyghmi's health bits (pyghmi.constants.Health). _WARNING, _CRITICAL, _FAILED = 1, 2, 4 #: The only interface a BMC reports over IPMI: its own LAN channel. _BMC = "bmc" def _severity(health: int) -> str: if health & (_CRITICAL | _FAILED): return "Critical" if health & _WARNING: return "Warning" return "OK" def _supply_present(reading: Any) -> bool: return "Present" in (getattr(reading, "states", None) or []) class IpmiDriver(HealthMetricsMixin, ManagementControllerDriver): """NAPALM driver for BMCs over IPMI 2.0 (lanplus), read-only. ``HealthMetricsMixin``: the BMC's own load, memory and interfaces over SNMP (UCD-MIB), where the BMC runs net-snmp. The managed server's sensors come from ``get_environment``. """ platform = "ipmi" DRIVER_NAME = "ipmi" #: IPMI is UDP; discovery probes no UDP, so this driver has no fingerprint. default_port = 623 def __init__( self, hostname: str, username: str, password: str, timeout: int = 60, optional_args: _JsonDict | None = None, ) -> None: self.hostname = hostname self.username = username self.password = password self.timeout = timeout self.optional_args: _JsonDict = optional_args or {} self.port = int(self.optional_args.get("port") or self.default_port) self._ipmi: Any = None # --- connection ---------------------------------------------------------------------- def open(self) -> None: self._forget_session() try: self._ipmi = Command( bmc=self.hostname, userid=self.username, password=self.password, port=self.port, ) except Exception as exc: raise ConnectionException(f"Cannot connect to IPMI at {self.hostname}: {exc}") \ from exc def close(self) -> None: ipmi, self._ipmi = self._ipmi, None if ipmi is not None: try: ipmi.ipmi_session.logout() except Exception: # a session left open times out on the BMC pass self._forget_session() def _forget_session(self) -> None: """Drop this BMC's session from pyghmi's cache of sessions being set up. pyghmi files a new session there under ``(bmc, userid, password, port, kg)`` with the credentials as str, and only ever deletes it under the credentials as bytes, so the entry stays. The next ``Command`` for the same BMC in this worker process then got the old, logged-out session back and failed with ``'<' not supported between instances of 'float' and 'NoneType'``. """ key = (self.hostname, self.username, self.password, self.port, None) Session.initting_sessions.pop(key, None) def is_alive(self) -> _JsonDict: return {"is_alive": self._ipmi is not None} def _cmd(self) -> Any: if self._ipmi is None: raise ConnectionException(f"{self.hostname}: not connected, call open() first") return self._ipmi # --- reads --------------------------------------------------------------------------- def _system(self) -> _JsonDict: """The "System" FRU record: manufacturer, product, serial number, UUID.""" for name, record in self._cmd().get_inventory(): if name == "System": return record or {} return {} def _bmc_version(self) -> str: for name, info in self._cmd().get_firmware(): if name == "BMC Version": return str(info.get("version") or "") return "" def _sensors(self) -> list[Any]: return list(self._cmd().get_sensor_data()) # --- NAPALM getters ------------------------------------------------------------------ def get_facts(self) -> _JsonDict: system = self._system() return { "uptime": -1.0, # IPMI has no uptime "vendor": system.get("Manufacturer") or "", "model": system.get("Product name") or system.get("Model") or "", "hostname": self.hostname, "fqdn": self.hostname, "os_version": self._bmc_version(), "serial_number": system.get("Serial Number") or "", "interface_list": [_BMC], } def get_interfaces(self) -> _JsonDict: net = self._cmd().get_net_configuration() source = net.get("ipv4_configuration") or "" return { _BMC: { "is_up": True, "is_enabled": True, "description": f"BMC LAN ({source})" if source else "BMC LAN", "last_flapped": -1.0, "speed": -1.0, "mtu": -1, "mac_address": (net.get("mac_address") or "").lower(), } } def get_interfaces_ip(self) -> _JsonDict: address = self._cmd().get_net_configuration().get("ipv4_address") or "" if "/" not in address: return {} ip, prefix = address.split("/", 1) return {_BMC: {"ipv4": {ip: {"prefix_length": int(prefix)}}}} def get_environment(self) -> _JsonDict: """Fans, temperatures and present power supplies from the sensor records; ``cpu`` and ``memory`` are empty, since a BMC does not know the host OS's load.""" fans: _JsonDict = {} temperature: _JsonDict = {} power: _JsonDict = {} for reading in self._sensors(): health = int(getattr(reading, "health", 0) or 0) value = getattr(reading, "value", None) if reading.type == "Temperature" and value is not None: temperature[reading.name] = { "temperature": float(value), "is_alert": bool(health), "is_critical": bool(health & (_CRITICAL | _FAILED)), } elif reading.type == "Fan" and value is not None: fans[reading.name] = {"status": value > 0 and not health} elif reading.type == "Power Supply" and _supply_present(reading): power[reading.name] = {"status": not health, "capacity": -1.0, "output": -1.0} return {"fans": fans, "temperature": temperature, "power": power, "cpu": {}, "memory": {}} # --- role methods -------------------------------------------------------------------- def get_power_state(self) -> _JsonDict: state = str(self._cmd().get_power().get("powerstate") or "") # Read-only: no action is offered. return {"state": state.capitalize(), "actions": []} def get_hardware_inventory(self) -> _JsonDict: """What the FRU and the sensor records tell; IPMI lists no CPUs, DIMMs or drives.""" system = self._system() supplies = [ { "slot": reading.name.removesuffix(" Status"), "model": "", "serial_number": "", "capacity_watts": None, "status": _severity(int(reading.health or 0)) if _supply_present(reading) else "Absent", } for reading in self._sensors() if reading.type == "Power Supply" ] return { "manufacturer": system.get("Manufacturer") or "", "model": system.get("Product name") or system.get("Model") or "", "serial_number": system.get("Serial Number") or "", "system_uuid": str(system.get("UUID") or "").lower(), "cpus": [], "memory": [], "drives": [], "power_supplies": supplies, "nics": [], } def get_event_log(self, limit: int = 500) -> list[_JsonDict]: """The system event log, newest first.""" entries = [] for event in self._cmd().get_event_log(): component = event.get("component") or "" text = event.get("event") or "" timecode = event.get("timecode") entries.append({ "id": str(event.get("record_id", "")), "timestamp": float(timecode) if timecode else None, "severity": _severity(int(event.get("severity") or 0)), "message": f"{component}: {text}" if component and text else component or text, }) entries.sort(key=lambda e: (e["timestamp"] is not None, e["timestamp"] or 0.0), reverse=True) return entries[:limit] def get_firmware_inventory(self) -> list[_JsonDict]: version = self._bmc_version() return [{"name": "BMC", "version": version, "updateable": None}] if version else []