Christian Manivong
01b2a2ae58
- ipmi: a failed logout on close is logged at debug level instead of passed over (B110); the session still times out on the BMC. - ipmi: the pyghmi imports are this driver's protocol, the fallback for BMCs without Redfish, documented as insecure in the module and in docs/security.md (nosec B415). - parsers.accounts: the empty password is NAPALM's get_users field; Redfish never returns one (nosec B105).
284 lines
11 KiB
Python
284 lines
11 KiB
Python
"""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 reads facts and inventory from the FRU,
|
|
sensors, the power state and the system event log, and switches the server's power
|
|
(chassis control). It does not reset the BMC; that goes through Redfish.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from napalm.base.exceptions import ConnectionException
|
|
from napalm_device_types import HealthMetricsMixin, ManagementControllerDriver
|
|
from napalm_device_types.models import POWER_ACTIONS
|
|
|
|
# IPMI is this driver's protocol: the fallback for BMCs without Redfish (see above).
|
|
from pyghmi.exceptions import IpmiException # nosec B415
|
|
from pyghmi.ipmi.command import Command # nosec B415
|
|
from pyghmi.ipmi.private.session import Session # nosec B415
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
_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"
|
|
#: Role power actions -> pyghmi's chassis control. IPMI has no graceful restart, and pyghmi
|
|
#: offers no power cycle; neither is offered.
|
|
_CHASSIS_CONTROL = {
|
|
"On": "on",
|
|
"GracefulShutdown": "shutdown",
|
|
"ForceOff": "off",
|
|
"ForceRestart": "reset",
|
|
}
|
|
|
|
|
|
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): reads, and the server's power.
|
|
|
|
``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
|
|
LOG.debug("%s: ignored", self.hostname, exc_info=True)
|
|
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 _power_state(self) -> str:
|
|
return str(self._cmd().get_power().get("powerstate") or "").capitalize()
|
|
|
|
@staticmethod
|
|
def _accepted_actions(state: str) -> list[str]:
|
|
"""The power actions that make sense in ``state``; IPMI advertises none itself."""
|
|
if state == "Off":
|
|
return ["On"]
|
|
if state == "On":
|
|
return [a for a in POWER_ACTIONS if a in _CHASSIS_CONTROL and a != "On"]
|
|
return []
|
|
|
|
def get_power_state(self) -> _JsonDict:
|
|
state = self._power_state()
|
|
return {"state": state, "actions": self._accepted_actions(state)}
|
|
|
|
def set_power_state(self, action: str) -> None:
|
|
if action not in POWER_ACTIONS:
|
|
raise ValueError(f"{action!r} is not one of {', '.join(POWER_ACTIONS)}")
|
|
state = self._power_state()
|
|
accepted = self._accepted_actions(state)
|
|
if action not in accepted:
|
|
raise RuntimeError(
|
|
f"{self.hostname}: {action} not accepted while the server is "
|
|
f"{state or 'in an unknown state'} (accepted: {', '.join(accepted) or 'none'})"
|
|
)
|
|
try:
|
|
self._cmd().set_power(_CHASSIS_CONTROL[action], wait=False)
|
|
except IpmiException as exc:
|
|
raise RuntimeError(f"{self.hostname}: {action} refused: {exc}") from exc
|
|
|
|
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 []
|