NAPALM/napalm-redfish
1
0 Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Add RedfishDriver on ManagementControllerDriver

Browse Source 
The redfish driver implements get_facts, get_interfaces, get_interfaces_ip, get_environment, get_users, get_ntp_servers and get_snmp_information for the BMC, and the role methods get_power_state, set_power_state, get_hardware_inventory, get_event_log, get_firmware_inventory and get_management_protocols for the managed server. reboot_host resets the BMC through Manager.Reset, never the server.

Registered as the napalm.drivers entry point redfish; depends on napalm and napalm-device-types >= 4.6.
This commit is contained in:
Christian Manivong committed 2026-10-13 05:02:58 +00:00
1 parent 7c366955bb
commit 42940d7e90
5 files changed
+1066 -6
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

napalm_redfish/driver.py
+542
View File
@@ -0,0 +1,542 @@
"""NAPALM driver for server BMCs over DMTF Redfish.
The device is the BMC. NAPALM's getters describe the BMC (its ports, users, services);
the ``ManagementControllerDriver`` methods describe and control the server it manages.
``reboot_host`` restarts the BMC (``Manager.Reset``), never the server.
Optional args
-------------
port : int
HTTPS port (default 443).
verify_ssl : bool or str
``True`` (default) verifies the certificate against the CA bundle of ``requests``,
a path verifies against that CA file, ``False`` turns verification off (self-signed
BMC certificates are common).
auth : str
``auto`` (default: session, basic auth if the service cannot create sessions),
``session`` or ``basic``.
retries : int
Extra attempts after a connection error (default 1).
The managed server is the first member of ``Systems``; its BMC is the first manager in
its ``Links.ManagedBy`` and its chassis the first in its ``Links.Chassis`` (the first
member of ``Managers``/``Chassis`` when the links are missing).
"""
from __future__ import annotations
import logging
from datetime import datetime, timezone
from typing import Any
from napalm.base.exceptions import ConnectionException
from napalm_device_types import ManagementControllerDriver
from napalm_device_types.base import FingerprintRule, PortSpec
from napalm_device_types.models import POWER_ACTIONS
from napalm_redfish.parsers import (
build_environment,
environment_sources,
parse_account,
parse_action_info,
parse_drive,
parse_ethernet_interface,
parse_fan,
parse_firmware,
parse_log_entry,
parse_log_service,
parse_manager,
parse_memory,
parse_network_protocol,
parse_power,
parse_power_state,
parse_power_subsystem,
parse_power_supply,
parse_processor,
parse_reset_action,
parse_sensor,
parse_service_root,
parse_simple_storage,
parse_storage,
parse_system,
parse_thermal,
parse_thermal_metrics,
parse_thermal_subsystem,
to_napalm_users,
)
from napalm_redfish.parsers.common import link, link_list, odata_id
from napalm_redfish.transport import RedfishError, RedfishNotFound, RedfishTransport
LOG = logging.getLogger(__name__)
_JsonDict = dict[str, Any]
#: ``get_management_protocols`` key -> ManagerNetworkProtocol property.
_PROTOCOLS = {
"ipmi": "IPMI",
"ssh": "SSH",
"http": "HTTP",
"https": "HTTPS",
"snmp": "SNMP",
"kvm": "KVMIP",
"virtual_media": "VirtualMedia",
}
_SEVERITIES = ("OK", "Warning", "Critical")
def _timestamp(created: str | None) -> float | None:
"""Unix time of a Redfish ``Created`` value; a time without offset counts as UTC."""
if not created:
return None
try:
when = datetime.fromisoformat(created.replace("Z", "+00:00"))
except ValueError:
return None
if when.tzinfo is None:
when = when.replace(tzinfo=timezone.utc)
return when.timestamp()
def _status(item: _JsonDict) -> str:
"""Role ``status``: ``Absent`` for an empty slot, else the Redfish health."""
if not item.get("present", True):
return "Absent"
return item.get("health") or ""
class RedfishDriver(ManagementControllerDriver):
"""NAPALM driver for Redfish BMCs (Supermicro, iDRAC, iLO, XCC and others)."""
platform = "redfish"
DRIVER_NAME = "redfish"
#: Every Redfish service answers ``GET /redfish/v1/`` without credentials.
PORT_SPECS = [PortSpec("https", 443, ("/redfish/v1/",))]
HTTP_FINGERPRINT = [FingerprintRule("redfishversion", 10.0),
FingerprintRule("#serviceroot.", 5.0)]
#: A BMC restart takes a few minutes before Redfish answers again.
REBOOT_SETTLE_SECONDS = 180
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.transport = RedfishTransport(
hostname,
username,
password,
port=self.optional_args.get("port"),
verify=self.optional_args.get("verify_ssl", True),
auth=self.optional_args.get("auth", "auto"),
timeout=timeout,
retries=self.optional_args.get("retries", 1),
)
self._cache: _JsonDict = {}
# --- connection ----------------------------------------------------------------------
def open(self) -> None:
try:
self.transport.open()
except RedfishError as exc:
raise ConnectionException(f"Cannot connect to Redfish at {self.hostname}: {exc}") \
from exc
def close(self) -> None:
self.transport.close()
self._cache = {}
def is_alive(self) -> _JsonDict:
try:
self.transport.get("/redfish/v1/")
except Exception:
return {"is_alive": False}
return {"is_alive": True}
# --- discovery (cached per connection) -------------------------------------------------
def _cached(self, key: str, load):
if key not in self._cache:
self._cache[key] = load()
return self._cache[key]
def _root(self) -> _JsonDict:
return self._cached("root", lambda: parse_service_root(self.transport.root))
def _first(self, collection: str) -> _JsonDict:
path = self._root()["links"].get(collection)
if path is None:
raise RedfishError(f"the service root links no {collection}")
members = self.transport.members(path, limit=1)
if not members:
raise RedfishError(f"{path} has no members")
return members[0]
def _system_raw(self) -> _JsonDict:
return self._cached("system", lambda: self._first("Systems"))
def _system_path(self) -> str:
return odata_id(self._system_raw()) or ""
def _linked_or_first(self, key: str, links_key: str, collection: str) -> _JsonDict:
def load() -> _JsonDict:
linked = link_list(self._system_raw(), "Links", links_key)
if linked:
return self.transport.get(linked[0])
return self._first(collection)
return self._cached(key, load)
def _manager_raw(self) -> _JsonDict:
return self._linked_or_first("manager", "ManagedBy", "Managers")
def _chassis_raw(self) -> _JsonDict:
return self._linked_or_first("chassis", "Chassis", "Chassis")
def _members_of(self, resource: _JsonDict, key: str) -> list[_JsonDict]:
"""Members of the collection ``resource[key]`` links; [] if not linked."""
path = link(resource, key)
return self.transport.members(path) if path else []
def _service(self, name: str) -> _JsonDict | None:
path = self._root()["links"].get(name)
return self.transport.get_optional(path) if path else None
def _network_protocol(self) -> _JsonDict:
raw = self.transport.follow(self._manager_raw(), "NetworkProtocol")
return parse_network_protocol(raw or {})
def _manager_interfaces(self) -> list[_JsonDict]:
return [parse_ethernet_interface(raw)
for raw in self._members_of(self._manager_raw(), "EthernetInterfaces")]
# --- NAPALM getters ----------------------------------------------------------------
def get_facts(self) -> _JsonDict:
system = parse_system(self._system_raw())
manager = parse_manager(self._manager_raw())
interfaces = self._manager_interfaces()
protocol = self._network_protocol()
hostname = (protocol["hostname"]
or next((i["hostname"] for i in interfaces if i["hostname"]), None)
or self.hostname)
fqdn = (protocol["fqdn"]
or next((i["fqdn"] for i in interfaces if i["fqdn"]), None)
or hostname)
return {
"uptime": -1.0, # Redfish has no uptime
"vendor": system["manufacturer"] or "",
"model": system["model"] or "",
"hostname": hostname,
"fqdn": fqdn,
"os_version": manager["firmware_version"] or "",
"serial_number": system["serial_number"] or "",
"interface_list": sorted(i["name"] for i in interfaces if i["name"]),
}
def get_interfaces(self) -> _JsonDict:
return {i["name"]: i["interface"] for i in self._manager_interfaces() if i["name"]}
def get_interfaces_ip(self) -> _JsonDict:
result = {}
for iface in self._manager_interfaces():
families = {f: iface[f] for f in ("ipv4", "ipv6") if iface[f]}
if iface["name"] and families:
result[iface["name"]] = families
return result
def get_environment(self) -> _JsonDict:
"""Fans, temperatures and power supplies of the chassis; ``cpu`` and ``memory`` are
empty, since a BMC does not know the host OS's load."""
sources = environment_sources(self._chassis_raw())
fans: _JsonDict = {}
temperature: _JsonDict = {}
power: _JsonDict = {}
thermal = sources["thermal"]
if thermal and thermal["schema"] == "Thermal":
legacy = parse_thermal(self.transport.get(thermal["path"]))
fans, temperature = legacy["fans"], legacy["temperature"]
elif thermal:
subsystem = parse_thermal_subsystem(self.transport.get(thermal["path"]))
parsed_fans = [parse_fan(raw) for raw in self.transport.members(
subsystem["fans_path"])] if subsystem["fans_path"] else []
if sources["sensors"]:
sensors = [parse_sensor(raw)
for raw in self.transport.members(sources["sensors"])]
temperature = build_environment(sensors=sensors)["temperature"]
elif subsystem["thermal_metrics_path"]:
metrics = self.transport.get_optional(subsystem["thermal_metrics_path"])
temperature = parse_thermal_metrics(metrics or {})["temperature"]
fans = build_environment(fans=parsed_fans)["fans"]
supply = sources["power"]
if supply and supply["schema"] == "Power":
power = parse_power(self.transport.get(supply["path"]))["power"]
elif supply:
psus = [parse_power_supply(raw, self.transport.follow(raw, "Metrics"))
for raw in self._power_supplies_raw(supply["path"])]
power = build_environment(power_supplies=psus)["power"]
return {"fans": fans, "temperature": temperature, "power": power,
"cpu": {}, "memory": {}}
def _power_supplies_raw(self, subsystem_path: str) -> list[_JsonDict]:
subsystem = parse_power_subsystem(self.transport.get(subsystem_path))
path = subsystem["power_supplies_path"]
return self.transport.members(path) if path else []
def get_users(self) -> _JsonDict:
service = self._service("AccountService")
if service is None:
return {}
return to_napalm_users(parse_account(raw) for raw in self._members_of(service,
"Accounts"))
def get_ntp_servers(self) -> _JsonDict:
return {server: {} for server in self._network_protocol()["ntp"]["servers"]}
def get_snmp_information(self) -> _JsonDict:
"""SNMP state from ManagerNetworkProtocol. Redfish hides community strings, so the
``community`` keys are the communities' names; contact and location are not in
Redfish."""
snmp = self._network_protocol()["snmp"]
communities = {}
if snmp["enabled"]:
for community in snmp["communities"]:
if community["name"]:
mode = "rw" if community["access"] == "Full" else "ro"
communities[community["name"]] = {"acl": "", "mode": mode}
return {"chassis_id": "", "community": communities, "contact": "", "location": ""}
# --- role: power -------------------------------------------------------------------
def _accepted_actions(self, system: _JsonDict) -> list[str]:
"""The PowerAction values the BMC accepts for ``system`` in its current state."""
reset = parse_reset_action(system)
if reset["target"] is None:
return []
allowed = reset["allowed"]
if allowed is None and reset["action_info"]:
info = self.transport.get_optional(reset["action_info"])
allowed = parse_action_info(info) if info else None
if allowed is None:
# Not advertised: assume the standard values; the BMC rejects what it lacks.
allowed = list(POWER_ACTIONS)
actions = [a for a in POWER_ACTIONS if a in allowed]
state = parse_power_state(system)
if state == "Off":
return [a for a in actions if a == "On"]
if state == "On":
return [a for a in actions if a != "On"]
return actions
def get_power_state(self) -> _JsonDict:
system = self.transport.get(self._system_path())
return {"state": parse_power_state(system) or "",
"actions": self._accepted_actions(system)}
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)}")
system = self.transport.get(self._system_path())
target = parse_reset_action(system)["target"]
if target is None:
raise RuntimeError(f"{self.hostname}: the system offers no reset action")
accepted = self._accepted_actions(system)
if action not in accepted:
raise RuntimeError(
f"{self.hostname}: {action} not accepted while the server is "
f"{parse_power_state(system) or 'in an unknown state'} "
f"(accepted: {', '.join(accepted) or 'none'})")
try:
self.transport.post(target, {"ResetType": action})
except RedfishError as exc:
raise RuntimeError(f"{self.hostname}: {action} refused: {exc}") from exc
def reboot_host(self) -> None:
"""Restart the BMC (``Manager.Reset``, graceful when offered). The server keeps
running; the Redfish session ends with the restart."""
manager = self.transport.get(odata_id(self._manager_raw()) or "")
reset = parse_reset_action(manager, "#Manager.Reset")
if reset["target"] is None:
raise RuntimeError(f"{self.hostname}: the BMC offers no reset action")
body: _JsonDict = {}
if reset["allowed"] is not None:
reset_type = next((t for t in ("GracefulRestart", "ForceRestart")
if t in reset["allowed"]), None)
if reset_type is None:
raise RuntimeError(f"{self.hostname}: BMC reset types "
f"{reset['allowed']} include no restart")
body["ResetType"] = reset_type
try:
self.transport.post(reset["target"], body)
except RedfishError as exc:
raise RuntimeError(f"{self.hostname}: BMC reset refused: {exc}") from exc
# --- role: inventory -----------------------------------------------------------------
def get_hardware_inventory(self) -> _JsonDict:
raw = self._system_raw()
system = parse_system(raw)
return {
"manufacturer": system["manufacturer"] or "",
"model": system["model"] or "",
"serial_number": system["serial_number"] or "",
"system_uuid": (system["uuid"] or "").lower(),
"cpus": self._cpus(raw),
"memory": self._dimms(raw),
"drives": self._drives(raw),
"power_supplies": self._psus(),
"nics": self._nics(raw),
}
def _cpus(self, system: _JsonDict) -> list[_JsonDict]:
cpus = []
for raw in self._members_of(system, "Processors"):
cpu = parse_processor(raw)
if cpu["processor_type"] not in (None, "CPU"):
continue # GPUs, FPGAs, accelerators
cpus.append({"socket": cpu["socket"] or cpu["id"] or "", "model": cpu["model"] or "",
"cores": cpu["cores"], "threads": cpu["threads"],
"status": _status(cpu)})
return cpus
def _dimms(self, system: _JsonDict) -> list[_JsonDict]:
dimms = []
for raw in self._members_of(system, "Memory"):
dimm = parse_memory(raw)
size = dimm["capacity_mib"]
dimms.append({
"slot": dimm["location"] or dimm["id"] or "",
"size_bytes": size * 1024 * 1024 if size else None,
"type": dimm["memory_device_type"] or "",
"speed_mts": dimm["speed_mhz"], # Redfish: MHz or MT/s "as appropriate"
"part_number": dimm["part_number"] or "",
"serial_number": dimm["serial_number"] or "",
"status": _status(dimm),
})
return dimms
def _drives(self, system: _JsonDict) -> list[_JsonDict]:
parsed: list[_JsonDict] = []
if link(system, "Storage"):
seen: set[str] = set()
for storage in self._members_of(system, "Storage"):
for path in parse_storage(storage)["drive_paths"]:
if path not in seen:
seen.add(path)
drive = self.transport.get_optional(path)
if drive is not None:
parsed.append(parse_drive(drive))
else:
for storage in self._members_of(system, "SimpleStorage"):
parsed.extend(parse_simple_storage(storage)["drives"])
return [{
"slot": d["name"] or d["id"] or "",
"model": d["model"] or "",
"serial_number": d["serial_number"] or "",
"size_bytes": d["capacity_bytes"],
"media": d["media_type"] or "",
"protocol": d["protocol"] or "",
"status": _status(d),
} for d in parsed]
def _psus(self) -> list[_JsonDict]:
source = environment_sources(self._chassis_raw())["power"]
if source is None:
return []
if source["schema"] == "Power":
items = self.transport.get(source["path"]).get("PowerSupplies") or []
else:
items = self._power_supplies_raw(source["path"])
psus = []
for raw in items:
if not isinstance(raw, dict):
continue
psu = parse_power_supply(raw)
psus.append({
"slot": psu["name"],
"model": psu["model"] or "",
"serial_number": psu["serial_number"] or "",
"capacity_watts": int(psu["capacity"]) if psu["capacity"] >= 0 else None,
"status": _status(psu),
})
return psus
def _nics(self, system: _JsonDict) -> list[_JsonDict]:
nics = []
for raw in self._members_of(system, "EthernetInterfaces"):
nic = parse_ethernet_interface(raw)
speed = nic["interface"]["speed"]
link_up = {"LinkUp": True, "LinkDown": False, "NoLink": False}.get(
nic["link_status"] or "")
nics.append({
"name": nic["name"] or "",
"mac_address": nic["interface"]["mac_address"].lower(),
"speed_mbps": int(speed) if speed >= 0 else None,
"link_up": link_up,
})
return nics
# --- role: event log, firmware, protocols --------------------------------------------
def get_event_log(self, limit: int = 500) -> list[_JsonDict]:
"""Entries of every log service of the system and of the BMC, newest first.
Where the SEL lives varies by vendor, so all of them are read; entries without a
time sort last.
"""
entries: list[_JsonDict] = []
for owner in (self._system_raw(), self._manager_raw()):
try:
services = self._members_of(owner, "LogServices")
except RedfishNotFound:
continue
for raw_service in services:
service = parse_log_service(raw_service)
if not service["entries_path"]:
continue
try:
raw_entries = self.transport.members(service["entries_path"])
except RedfishNotFound:
LOG.warning("%s: log %s has no entries collection", self.hostname,
service["id"])
continue
for raw in raw_entries:
entry = parse_log_entry(raw)
severity = entry["severity"]
entries.append({
"id": odata_id(raw) or f"{service['id']}/{entry['id']}",
"timestamp": _timestamp(entry["created"]),
"severity": severity if severity in _SEVERITIES else "OK",
"message": entry["message"] or "",
})
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]:
service = self._service("UpdateService")
if service is None:
return []
items = [parse_firmware(raw) for raw in self._members_of(service, "FirmwareInventory")]
return [{"name": fw["name"] or fw["id"] or "", "version": fw["version"] or "",
"updateable": fw["updateable"]} for fw in items]
def get_management_protocols(self) -> _JsonDict:
"""What the BMC listens on. A protocol the BMC does not report counts as off."""
protocols = self._network_protocol()["protocols"]
result = {}
for key, name in _PROTOCOLS.items():
entry = protocols.get(name) or {}
result[key] = {"enabled": bool(entry.get("enabled")), "port": entry.get("port")}
return result