napalm-redfish/napalm_redfish/parsers/manager.py

"""Manager (the BMC itself), its EthernetInterfaces and its NetworkProtocol."""

from __future__ import annotations

import ipaddress
from collections.abc import Mapping
from typing import Any

from .common import boolean, integer, link_list, links, number, status, text


def parse_manager(manager: Mapping[str, Any]) -> dict[str, Any]:
    """``Managers/{id}``: model, firmware and links of the BMC."""
    state, health = status(manager)
    return {
        "id": text(manager.get("Id")),
        "name": text(manager.get("Name")),
        "manager_type": text(manager.get("ManagerType")),
        "model": text(manager.get("Model")),
        "firmware_version": text(manager.get("FirmwareVersion")),
        "uuid": text(manager.get("UUID")),
        "power_state": text(manager.get("PowerState")),
        "date_time": text(manager.get("DateTime")),
        "state": state,
        "health": health,
        "links": links(manager),
        "manager_for_servers": link_list(manager, "Links", "ManagerForServers"),
        "manager_for_chassis": link_list(manager, "Links", "ManagerForChassis"),
    }


def _mac(value: Any) -> str:
    mac = text(value)
    return mac.upper().replace("-", ":") if mac else ""


def _ipv4(entries: Any) -> dict[str, dict[str, int]]:
    found = {}
    for entry in entries if isinstance(entries, list) else []:
        if not isinstance(entry, Mapping):
            continue
        try:
            addr = ipaddress.IPv4Address(text(entry.get("Address")) or "")
            mask = text(entry.get("SubnetMask"))
            prefix = ipaddress.IPv4Network(f"0.0.0.0/{mask}").prefixlen if mask else 32
        except ValueError:
            continue
        if addr.is_unspecified:
            continue
        found[str(addr)] = {"prefix_length": prefix}
    return found


def _ipv6(entries: Any) -> dict[str, dict[str, int]]:
    found = {}
    for entry in entries if isinstance(entries, list) else []:
        if not isinstance(entry, Mapping):
            continue
        try:
            addr = ipaddress.IPv6Address(text(entry.get("Address")) or "")
        except ValueError:
            continue
        if addr.is_unspecified:
            continue
        prefix = integer(entry.get("PrefixLength"))
        found[str(addr)] = {"prefix_length": prefix if prefix is not None else 128}
    return found


def parse_ethernet_interface(iface: Mapping[str, Any]) -> dict[str, Any]:
    """One EthernetInterface, as NAPALM ``get_interfaces``/``get_interfaces_ip`` parts.

    ``interface`` is the ``get_interfaces`` value, ``ipv4``/``ipv6`` the ``get_interfaces_ip``
    one. Without ``LinkStatus`` (older schemas) an enabled interface counts as up. Unset
    addresses (``0.0.0.0``, ``::``) are left out.
    """
    state, _ = status(iface)
    enabled = boolean(iface.get("InterfaceEnabled"))
    if enabled is None:
        enabled = state == "Enabled"
    link_status = text(iface.get("LinkStatus"))
    is_up = enabled and (link_status == "LinkUp" if link_status else state == "Enabled")
    speed = number(iface.get("SpeedMbps"))
    vlan = iface.get("VLAN")
    vlan_id = None
    if isinstance(vlan, Mapping) and vlan.get("VLANEnable"):
        vlan_id = integer(vlan.get("VLANId"))
    return {
        "name": text(iface.get("Id")),
        "interface": {
            "is_up": bool(is_up),
            "is_enabled": bool(enabled),
            "description": text(iface.get("Description")) or "",
            "last_flapped": -1.0,
            "speed": speed if speed is not None else -1.0,
            "mtu": integer(iface.get("MTUSize")) or 0,
            "mac_address": _mac(iface.get("MACAddress")),
        },
        "ipv4": _ipv4(iface.get("IPv4Addresses")),
        "ipv6": _ipv6(iface.get("IPv6Addresses")),
        "hostname": text(iface.get("HostName")),
        "fqdn": text(iface.get("FQDN")),
        "link_status": link_status,
        "vlan": vlan_id,
        "permanent_mac_address": _mac(iface.get("PermanentMACAddress")) or None,
    }


def parse_network_protocol(proto: Mapping[str, Any]) -> dict[str, Any]:
    """``Managers/{id}/NetworkProtocol``: which services the BMC runs, NTP and SNMP.

    ``protocols`` holds every property with a ``ProtocolEnabled`` flag or a ``Port``, so
    services the schema adds later show up without code changes. NTP servers and the SNMP details
    exist only in newer schema versions; missing values are ``None``.
    """
    protocols = {}
    for key, value in proto.items():
        if isinstance(value, Mapping) and ("ProtocolEnabled" in value or "Port" in value):
            protocols[key] = {
                "enabled": boolean(value.get("ProtocolEnabled")),
                "port": integer(value.get("Port")),
            }

    ntp = proto.get("NTP") if isinstance(proto.get("NTP"), Mapping) else {}
    servers: list[str] = []
    for server in ntp.get("NTPServers") or []:
        server = text(server)
        if server and server not in servers:
            servers.append(server)

    snmp = proto.get("SNMP") if isinstance(proto.get("SNMP"), Mapping) else {}
    communities = []
    for community in snmp.get("CommunityStrings") or []:
        if isinstance(community, Mapping):
            # The community string itself is never kept, even if a BMC returns it.
            communities.append(
                {"name": text(community.get("Name")), "access": text(community.get("AccessMode"))}
            )

    return {
        "hostname": text(proto.get("HostName")),
        "fqdn": text(proto.get("FQDN")),
        "protocols": protocols,
        "ntp": {"enabled": boolean(ntp.get("ProtocolEnabled")), "servers": servers},
        "snmp": {
            "enabled": boolean(snmp.get("ProtocolEnabled")),
            "port": integer(snmp.get("Port")),
            "v1": boolean(snmp.get("EnableSNMPv1")),
            "v2c": boolean(snmp.get("EnableSNMPv2c")),
            "v3": boolean(snmp.get("EnableSNMPv3")),
            "communities": communities,
        },
    }