napalm-redfish/napalm_redfish/bmc.py

"""NAPALM driver for a BMC that takes the best of Redfish and IPMI, chosen by itself.

``redfish`` and ``ipmi`` make the operator choose, and the right choice changes with the
BMC's license: a Supermicro X10 refuses Redfish without one, answers most of it with
SFT-OOB-LIC, and all of it with DCMS. This driver asks Redfish first and IPMI where Redfish
does not answer:

* A BMC that refuses Redfish altogether (no license) is read over IPMI entirely.
* A getter that fails over Redfish (an error, a resource the BMC withholds) is asked over
  IPMI, where IPMI has it. IPMI is opened only then.
* Power control goes over Redfish when it answers, else over IPMI. A refusal by the BMC is
  an answer, not a failure, and is not retried over IPMI.
* The BMC itself is reset over Redfish only (``Manager.Reset``); IPMI's cold reset is not
  offered, as in the ``ipmi`` driver.

Both protocols use the same account. ``optional_args``: everything ``RedfishDriver``
takes, plus ``ipmi_port`` (623).
"""

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_redfish.driver import BMC_HTTP_FINGERPRINT, BMC_PORT_SPECS, RedfishDriver
from napalm_redfish.ipmi import IpmiDriver
from napalm_redfish.transport import RedfishError

LOG = logging.getLogger(__name__)

_JsonDict = dict[str, Any]
_IPMI_PORT = 623


class BmcDriver(HealthMetricsMixin, ManagementControllerDriver):
    """One BMC over Redfish and IPMI: Redfish first, IPMI where Redfish does not answer."""

    platform = "bmc"
    DRIVER_NAME = "bmc"
    default_port = 443
    #: Discovery suggests this driver for a BMC it finds, not ``redfish``.
    PORT_SPECS = BMC_PORT_SPECS
    HTTP_FINGERPRINT = BMC_HTTP_FINGERPRINT
    #: A BMC restart takes a few minutes before Redfish answers again.
    REBOOT_SETTLE_SECONDS = RedfishDriver.REBOOT_SETTLE_SECONDS

    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 = dict(optional_args or {})
        ipmi_port = int(self.optional_args.pop("ipmi_port", None) or _IPMI_PORT)
        self.ipmi_port = ipmi_port
        self._redfish = RedfishDriver(hostname, username, password, timeout, self.optional_args)
        self._ipmi = IpmiDriver(hostname, username, password, timeout, {"port": ipmi_port})
        self._redfish_ok = False
        self._ipmi_open = False

    # --- connection ----------------------------------------------------------------------

    def open(self) -> None:
        """Connect over Redfish; a BMC that refuses it or does not answer it is opened over
        IPMI instead. Fails only when neither answers."""
        try:
            self._redfish.open()
        except ConnectionException as exc:
            LOG.info("%s: no Redfish (%s); reading it over IPMI", self.hostname, exc)
            self._redfish_ok = False
        else:
            self._redfish_ok = self._redfish._license_refused is None
        if not self._redfish_ok:
            self._ipmi_conn()

    def close(self) -> None:
        self._redfish.close()
        if self._ipmi_open:
            self._ipmi.close()
            self._ipmi_open = False

    def is_alive(self) -> _JsonDict:
        if self._redfish_ok:
            return self._redfish.is_alive()
        return self._ipmi.is_alive()

    def _ipmi_conn(self) -> IpmiDriver:
        if not self._ipmi_open:
            self._ipmi.open()
            self._ipmi_open = True
        return self._ipmi

    def _call(self, name: str, *args: Any) -> Any:
        """``name`` over Redfish; over IPMI when Redfish is out or fails and IPMI has it."""
        has_ipmi = callable(getattr(IpmiDriver, name, None))
        if self._redfish_ok:
            try:
                return getattr(self._redfish, name)(*args)
            except (RedfishError, ConnectionException) as exc:
                if not has_ipmi:
                    raise
                LOG.info("%s: %s failed over Redfish (%s); asking IPMI", self.hostname, name,
                         exc)
        if not has_ipmi:
            raise NotImplementedError(f"{self.hostname}: {name} needs Redfish")
        return getattr(self._ipmi_conn(), name)(*args)

    # --- NAPALM getters ------------------------------------------------------------------

    def get_facts(self) -> _JsonDict:
        return self._call("get_facts")

    def get_interfaces(self) -> _JsonDict:
        return self._call("get_interfaces")

    def get_interfaces_ip(self) -> _JsonDict:
        return self._call("get_interfaces_ip")

    def get_environment(self) -> _JsonDict:
        return self._call("get_environment")

    def get_users(self) -> _JsonDict:
        return self._call("get_users")

    def get_ntp_servers(self) -> _JsonDict:
        return self._call("get_ntp_servers")

    def get_snmp_information(self) -> _JsonDict:
        return self._call("get_snmp_information")

    def get_device_warnings(self) -> list[_JsonDict]:
        """Redfish's signals; for a BMC that refuses Redfish, with ``ipmi_fallback`` set, as
        this driver reads it over IPMI meanwhile."""
        warnings = self._redfish.get_device_warnings()
        if self._redfish_ok:
            return warnings
        return [{**w, "meta": {**w.get("meta", {}), "ipmi_fallback": True}} for w in warnings]

    # --- role methods --------------------------------------------------------------------

    def get_power_state(self) -> _JsonDict:
        return self._call("get_power_state")

    def set_power_state(self, action: str) -> None:
        self._call("set_power_state", action)

    def reboot_host(self) -> None:
        """Restart the BMC (``Manager.Reset``), never the server. Needs Redfish."""
        if not self._redfish_ok:
            raise RuntimeError(f"{self.hostname}: resetting the BMC needs Redfish")
        self._redfish.reboot_host()

    def get_hardware_inventory(self) -> _JsonDict:
        return self._call("get_hardware_inventory")

    def get_event_log(self, limit: int = 500) -> list[_JsonDict]:
        return self._call("get_event_log", limit)

    def get_firmware_inventory(self) -> list[_JsonDict]:
        return self._call("get_firmware_inventory")

    def get_management_protocols(self) -> _JsonDict:
        """Over Redfish, what the BMC reports. Over IPMI only one thing is known: IPMI over
        LAN is on, since this driver is talking it."""
        if self._redfish_ok:
            return self._redfish.get_management_protocols()
        return {"ipmi": {"enabled": True, "port": self.ipmi_port}}