An `amt` driver (napalm_amt.AmtDriver), a ManagementControllerDriver for the management controller of Intel vPro machines, with its own lean WS-Management client (requests + ElementTree, HTTP Digest, ports 16992/16993). It reads the machine's identity, hardware with the SMBIOS system UUID, firmware (AMT with build, BIOS), power state and the hardware event log, reports what AMT listens on (plain HTTP or TLS, KVM incl. port 5900, IDE-R), and switches the machine's power. AMT cannot be reset over WS-Management, so there is no reboot_host. Discovery recognises AMT by its ports and its Server header. Tested against Intel's published AMT 12 recordings (go-wsman-messages, Apache-2.0, see NOTICE). `python -m napalm_amt.record` records a real AMT read-only; the first test devices are two Lenovo ThinkCentre M910x.
91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
"""CPUs, DIMMs and drives of the managed machine, as AMT reads them from SMBIOS.
|
|
|
|
AMT knows less than a server BMC: no core counts, no drive models, no health per DIMM.
|
|
What it does not know is ``None`` or empty, never guessed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from napalm_amt.parsers.common import text, to_int
|
|
|
|
_Instance = dict[str, Any]
|
|
|
|
#: CIM_Processor.CPUStatus -> role status.
|
|
_CPU_STATUS = {1: "OK", 4: "OK", 2: "Warning", 3: "Critical"}
|
|
#: CIM_PhysicalMemory.MemoryType, which AMT fills with the SMBIOS memory type.
|
|
_MEMORY_TYPES = {
|
|
18: "DDR",
|
|
19: "DDR2",
|
|
24: "DDR3",
|
|
26: "DDR4",
|
|
27: "LPDDR",
|
|
28: "LPDDR2",
|
|
29: "LPDDR3",
|
|
30: "LPDDR4",
|
|
34: "DDR5",
|
|
35: "LPDDR5",
|
|
}
|
|
#: CIM OperationalStatus -> role status.
|
|
_OPERATIONAL = {2: "OK", 3: "Warning", 4: "Warning", 5: "Warning", 6: "Critical", 7: "Critical"}
|
|
|
|
|
|
def parse_cpus(processors: list[_Instance], chips: list[_Instance]) -> list[dict[str, Any]]:
|
|
"""One entry per ``CIM_Processor``; the model is the ``CIM_Chip`` with its tag."""
|
|
models = {text(c.get("Tag")): text(c.get("Version")) for c in chips}
|
|
return [
|
|
{
|
|
"socket": text(p.get("DeviceID")),
|
|
"model": models.get(text(p.get("DeviceID")), ""),
|
|
"cores": None,
|
|
"threads": None,
|
|
"status": _CPU_STATUS.get(to_int(p.get("CPUStatus")) or 0, ""),
|
|
}
|
|
for p in processors
|
|
]
|
|
|
|
|
|
def _positive(value: Any) -> int | None:
|
|
number = to_int(value)
|
|
return number if number and number > 0 else None
|
|
|
|
|
|
def parse_memory(modules: list[_Instance]) -> list[dict[str, Any]]:
|
|
"""One entry per ``CIM_PhysicalMemory``. AMT reports no health per module."""
|
|
return [
|
|
{
|
|
"slot": text(m.get("BankLabel")) or text(m.get("Tag")),
|
|
"size_bytes": _positive(m.get("Capacity")),
|
|
"type": _MEMORY_TYPES.get(to_int(m.get("MemoryType")) or 0, ""),
|
|
"speed_mts": (
|
|
_positive(m.get("ConfiguredMemoryClockSpeed"))
|
|
or _positive(m.get("MaxMemorySpeed"))
|
|
or _positive(m.get("Speed"))
|
|
),
|
|
"part_number": text(m.get("PartNumber")),
|
|
"serial_number": text(m.get("SerialNumber")),
|
|
"status": "",
|
|
}
|
|
for m in modules
|
|
]
|
|
|
|
|
|
def parse_drives(devices: list[_Instance]) -> list[dict[str, Any]]:
|
|
"""One entry per ``CIM_MediaAccessDevice``; ``MaxMediaSize`` is in kilobytes of 1000."""
|
|
drives = []
|
|
for d in devices:
|
|
size = _positive(d.get("MaxMediaSize"))
|
|
drives.append(
|
|
{
|
|
"slot": text(d.get("DeviceID")),
|
|
"model": "",
|
|
"serial_number": "",
|
|
"size_bytes": size * 1000 if size else None,
|
|
"media": "",
|
|
"protocol": "",
|
|
"status": _OPERATIONAL.get(to_int(d.get("OperationalStatus")) or 0, ""),
|
|
}
|
|
)
|
|
return drives
|