napalm-amt/napalm_amt/driver.py

"""NAPALM driver for Intel Active Management Technology (AMT) over WS-Management.

AMT is the management controller built into Intel vPro client platforms (business desktops,
mini PCs, notebooks). Like a server BMC it works while the operating system is down or the
machine is off; unlike one it shares the machine's network controller, has no sensors to
speak of, and cannot be restarted over the network.

The device is AMT. NAPALM's getters describe AMT (its network port, its version); the
``ManagementControllerDriver`` methods describe and control the machine it manages. There
is no ``reboot_host``: WS-Management offers no way to restart AMT itself.

Optional args
-------------
port : int
    16992 (default, plain HTTP) or 16993 (TLS, once TLS is set up in AMT).
tls : bool
    Overrides the scheme the port implies.
verify_ssl : bool or str
    With TLS: ``True`` (default) verifies AMT's certificate against the CA bundle of
    ``requests``, a path verifies against that CA file, ``False`` turns verification off.
retries : int
    Extra attempts after a connection error (default 1).
"""

from __future__ import annotations

import logging
from collections.abc import Callable
from typing import Any, TypeVar

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_amt.parsers import (
    POWER_REQUESTS,
    accepted_actions,
    parse_cpus,
    parse_drives,
    parse_ethernet_ports,
    parse_event_record,
    parse_firmware,
    parse_management_protocols,
    parse_memory,
    parse_platform_guid,
    parse_power_state,
    parse_software_identity,
    parse_system_identity,
)
from napalm_amt.parsers.common import is_true, text
from napalm_amt.wsman import PORT, WsmanAuthError, WsmanError, WsmanFault, WsmanTransport

LOG = logging.getLogger(__name__)

_JsonDict = dict[str, Any]
_T = TypeVar("_T")

#: Records per ``AMT_MessageLog.GetRecords`` call; 390 is what Intel's tools ask for.
_RECORDS_PER_CALL = 390
#: GetRecords calls per log; AMT keeps a few hundred records, this stops a runaway loop.
_MAX_RECORD_CALLS = 50
#: ``PositionToFirstRecord`` ReturnValue for an empty log.
_NO_RECORD = "2"
#: ``RequestPowerStateChange`` ReturnValues that mean accepted: done, or job started.
_POWER_ACCEPTED = {"0", "4096"}
#: The machine AMT manages, as ``RequestPowerStateChange`` names it (from Intel's tools).
_MANAGED_SYSTEM = (
    "<h:ManagedElement>"
    '<Address xmlns="http://schemas.xmlsoap.org/ws/2004/08/addressing">'
    "http://schemas.xmlsoap.org/ws/2004/08/addressing</Address>"
    '<ReferenceParameters xmlns="http://schemas.xmlsoap.org/ws/2004/08/addressing">'
    '<ResourceURI xmlns="http://schemas.dmtf.org/wbem/wsman/1/wsman.xsd">'
    "http://schemas.dmtf.org/wbem/wscim/1/cim-schema/2/CIM_ComputerSystem</ResourceURI>"
    '<SelectorSet xmlns="http://schemas.dmtf.org/wbem/wsman/1/wsman.xsd">'
    '<Selector Name="CreationClassName">CIM_ComputerSystem</Selector>'
    '<Selector Name="Name">ManagedSystem</Selector>'
    "</SelectorSet></ReferenceParameters></h:ManagedElement>"
)


class AmtDriver(ManagementControllerDriver):
    """NAPALM driver for Intel AMT: reads the machine, and switches its power."""

    platform = "amt"
    DRIVER_NAME = "amt"
    default_port = PORT
    #: AMT listens on 16992 without TLS and on 16993 with it; one of them answers.
    PORT_SPECS = [
        PortSpec("http", 16992, ("/",), weight=5.0),
        PortSpec("https", 16993, ("/",), weight=5.0),
    ]
    #: AMT names itself in the ``Server`` header of every answer, and its web UI in the
    #: title ("Intel(R) Active Management Technology 11.8.50.3399").
    HTTP_FINGERPRINT = [FingerprintRule("active management technology", 10.0)]
    #: AMT runs no SNMP agent, so netOrk does not read it over SNMP.
    SNMP_CAPABLE = False

    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 = WsmanTransport(
            hostname,
            username,
            password,
            port=self.optional_args.get("port"),
            tls=self.optional_args.get("tls"),
            verify=self.optional_args.get("verify_ssl", True),
            timeout=timeout,
            retries=self.optional_args.get("retries", 1),
        )
        self._cache: _JsonDict = {}
        self._open = False

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

    def open(self) -> None:
        """Connect: read AMT's general settings, which every AMT has."""
        self._cache = {}
        try:
            self._cache["general"] = self.transport.get("AMT_GeneralSettings")
        except WsmanAuthError as exc:
            raise ConnectionException(f"{self.hostname}: AMT refused the credentials") from exc
        except WsmanError as exc:
            raise ConnectionException(f"Cannot connect to AMT at {self.hostname}: {exc}") from exc
        self._open = True

    def close(self) -> None:
        self.transport.close()
        self._cache = {}
        self._open = False

    def is_alive(self) -> _JsonDict:
        if not self._open:
            return {"is_alive": False}
        try:
            self.transport.get("AMT_GeneralSettings")
        except WsmanError:
            return {"is_alive": False}
        return {"is_alive": True}

    # --- reads (cached per connection) ---------------------------------------------------

    def _connected(self) -> WsmanTransport:
        if not self._open:
            raise ConnectionException(f"{self.hostname}: not connected, call open() first")
        return self.transport

    def _cached(self, key: str, load: Callable[[], _T]) -> _T:
        self._connected()
        if key not in self._cache:
            self._cache[key] = load()
        return self._cache[key]  # type: ignore[no-any-return]

    def _get(self, resource: str) -> _JsonDict:
        return self._cached(resource, lambda: self.transport.get(resource))

    def _all(self, resource: str) -> list[_JsonDict]:
        """Every instance of ``resource``; none where this AMT does not know the class
        (older versions know fewer)."""

        def load() -> list[_JsonDict]:
            try:
                return self.transport.enumerate(resource)
            except WsmanFault as exc:
                LOG.info("%s: no %s (%s)", self.hostname, resource, exc)
                return []

        return self._cached(resource, load)

    def _general(self) -> _JsonDict:
        return self._cached("general", lambda: self.transport.get("AMT_GeneralSettings"))

    def _identity(self) -> dict[str, str]:
        return parse_system_identity(self._all("CIM_Chassis"), self._all("CIM_Card"))

    def _software(self) -> dict[str, str]:
        return parse_software_identity(self._all("CIM_SoftwareIdentity"))

    def _ports(self) -> list[_JsonDict]:
        return parse_ethernet_ports(self._all("AMT_EthernetPortSettings"))

    def _power(self) -> _JsonDict:
        """The power state, read fresh: it is what changes between two calls."""
        services = self._connected().enumerate("CIM_AssociatedPowerManagementService")
        return parse_power_state(services[0] if services else {})

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

    def get_facts(self) -> _JsonDict:
        general = self._general()
        identity = self._identity()
        name = text(general.get("HostName"))
        domain = text(general.get("DomainName"))
        return {
            "uptime": -1.0,  # AMT keeps no uptime of the machine or of itself
            "vendor": identity["manufacturer"],
            "model": identity["model"],
            "hostname": name or self.hostname,
            "fqdn": f"{name}.{domain}" if name and domain else name or self.hostname,
            "os_version": self._software()["amt"],
            "serial_number": identity["serial_number"],
            "interface_list": [p["name"] for p in self._ports()],
        }

    def get_interfaces(self) -> _JsonDict:
        enabled = is_true(self._general().get("NetworkInterfaceEnabled", "true"))
        interfaces = {}
        for port in self._ports():
            address = "DHCP" if port["dhcp"] else "static"
            shared = ", shares the host's IP address" if port["shared_ip"] else ""
            interfaces[port["name"]] = {
                "is_up": port["link_up"],
                "is_enabled": enabled,
                "description": f"Intel AMT, {port['name']}, {address}{shared}",
                "last_flapped": -1.0,
                "speed": -1.0,
                "mtu": -1,
                "mac_address": port["mac_address"],
            }
        return interfaces

    def get_interfaces_ip(self) -> _JsonDict:
        return {
            p["name"]: {"ipv4": {p["ip"]: {"prefix_length": p["prefix_length"]}}}
            for p in self._ports()
            if p["ip"] and p["prefix_length"] is not None
        }

    def get_environment(self) -> _JsonDict:
        """Empty: AMT on a client platform reports no temperatures, fans or power supplies,
        and like any management controller it does not know the OS's load."""
        self._connected()
        return {"fans": {}, "temperature": {}, "power": {}, "cpu": {}, "memory": {}}

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

    def get_power_state(self) -> _JsonDict:
        power = self._power()
        return {
            "state": power["state"],
            "actions": accepted_actions(power["state"], power["available"]),
        }

    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)}")
        power = self._power()
        accepted = accepted_actions(power["state"], power["available"])
        if action not in accepted:
            raise RuntimeError(
                f"{self.hostname}: {action} not accepted while the machine is "
                f"{power['state'] or 'in an unknown state'} "
                f"(accepted: {', '.join(accepted) or 'none'})"
            )
        try:
            output = self.transport.invoke(
                "CIM_PowerManagementService",
                "RequestPowerStateChange",
                [("PowerState", POWER_REQUESTS[action])],
                raw=_MANAGED_SYSTEM,
                check=False,
            )
        except WsmanError as exc:
            raise RuntimeError(f"{self.hostname}: {action} refused: {exc}") from exc
        code = text(output.get("ReturnValue"))
        if code not in _POWER_ACCEPTED:
            raise RuntimeError(f"{self.hostname}: {action} refused by AMT (ReturnValue {code})")

    def get_hardware_inventory(self) -> _JsonDict:
        identity = self._identity()
        packages = self._all("CIM_ComputerSystemPackage")
        return {
            "manufacturer": identity["manufacturer"],
            "model": identity["model"],
            "serial_number": identity["serial_number"],
            "system_uuid": parse_platform_guid(
                packages[0].get("PlatformGUID", "") if packages else ""
            ),
            "cpus": parse_cpus(self._all("CIM_Processor"), self._all("CIM_Chip")),
            "memory": parse_memory(self._all("CIM_PhysicalMemory")),
            "drives": parse_drives(self._all("CIM_MediaAccessDevice")),
            "power_supplies": [],
            "nics": [
                {
                    "name": p["name"],
                    "mac_address": p["mac_address"],
                    "speed_mbps": None,
                    "link_up": p["link_up"],
                }
                for p in self._ports()
            ],
        }

    def get_event_log(self, limit: int = 500) -> list[_JsonDict]:
        """AMT's hardware event log, newest first. ``id`` is the record's position in the
        log, oldest first: AMT numbers its records no other way."""
        start = self._connected().invoke("AMT_MessageLog", "PositionToFirstRecord", check=False)
        if text(start.get("ReturnValue")) == _NO_RECORD:
            return []
        if text(start.get("ReturnValue")) != "0":
            raise WsmanError(
                f"AMT_MessageLog.PositionToFirstRecord failed: ReturnValue "
                f"{text(start.get('ReturnValue'))}"
            )
        identifier = text(start.get("IterationIdentifier"))
        records: list[str] = []
        for _ in range(_MAX_RECORD_CALLS):
            page = self.transport.invoke(
                "AMT_MessageLog",
                "GetRecords",
                [("IterationIdentifier", identifier), ("MaxReadRecords", _RECORDS_PER_CALL)],
            )
            batch = page.get("RecordArray") or []
            records.extend(batch if isinstance(batch, list) else [batch])
            identifier = text(page.get("IterationIdentifier"))
            if is_true(page.get("NoMoreRecords")) or not batch:
                break
        entries = [
            e for e in (parse_event_record(r, str(n)) for n, r in enumerate(records, start=1)) if e
        ]
        entries.sort(key=lambda e: e["timestamp"] or 0.0, reverse=True)
        return entries[:limit]

    def get_firmware_inventory(self) -> list[_JsonDict]:
        return parse_firmware(self._software(), self._all("CIM_BIOSElement"))

    def get_management_protocols(self) -> _JsonDict:
        def optional_get(resource: str) -> _JsonDict:
            try:
                return self._get(resource)
            except WsmanFault:
                return {}

        return parse_management_protocols(
            tls_settings=self._all("AMT_TLSSettingData"),
            redirection=optional_get("AMT_RedirectionService"),
            kvm_settings=optional_get("IPS_KVMRedirectionSettingData"),
            kvm_saps=self._all("CIM_KVMRedirectionSAP"),
        )