feat: Intel AMT driver over WS-Management (0.1.0)

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.
This commit is contained in:
christianmanivong committed 2026-10-18 14:55:32 +02:00
commit 0052ba7884
61 files changed
+4303

No files matched your search

+31
View File
@@ -0,0 +1,31 @@
"""Pure functions from WS-Management instances (dicts) to the shapes the driver returns."""
from napalm_amt.parsers.inventory import parse_cpus, parse_drives, parse_memory
from napalm_amt.parsers.logs import parse_event_record
from napalm_amt.parsers.network import parse_ethernet_ports
from napalm_amt.parsers.power import POWER_REQUESTS, accepted_actions, parse_power_state
from napalm_amt.parsers.protocols import parse_management_protocols
from napalm_amt.parsers.system import (
parse_firmware,
parse_platform_guid,
parse_software_identity,
parse_system_identity,
uuid_from_base64,
)
__all__ = [
"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",
"uuid_from_base64",
]
+56
View File
@@ -0,0 +1,56 @@
"""Helpers shared by the parsers: WS-Management values are strings, lists or missing."""
from __future__ import annotations
from typing import Any
#: Values firmware puts where it has nothing to say (SMBIOS defaults), compared lowercased.
PLACEHOLDERS = frozenset(
{
"",
"0",
"none",
"default string",
"not specified",
"to be filled by o.e.m.",
"system product name",
"system serial number",
"123456789",
}
)
def as_list(value: Any) -> list[Any]:
"""A repeated element as a list; a single one as a list of one; none as empty."""
if value is None or value == "":
return []
return value if isinstance(value, list) else [value]
def first(value: Any) -> Any:
"""The first of a value that may be repeated."""
values = as_list(value)
return values[0] if values else None
def to_int(value: Any) -> int | None:
try:
return int(str(first(value)).strip())
except (TypeError, ValueError):
return None
def is_true(value: Any) -> bool:
return str(first(value)).strip().lower() in ("true", "1")
def text(value: Any) -> str:
"""A string value, stripped; a missing or nested one as empty."""
value = first(value)
return value.strip() if isinstance(value, str) else ""
def meaningful(value: Any) -> str:
"""``text(value)``, or empty for a firmware placeholder."""
result = text(value)
return "" if result.lower() in PLACEHOLDERS else result
+90
View File
@@ -0,0 +1,90 @@
"""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
+173
View File
@@ -0,0 +1,173 @@
"""AMT's hardware event log: the base64 records of ``AMT_MessageLog.GetRecords``.
Each record is a 21-byte platform event (ASF/PET): a little-endian Unix timestamp, then
device address, sensor type, event type, offset, source type, severity, sensor number,
entity, entity instance and eight bytes of event data. The texts follow Intel's decoder in
go-wsman-messages, which knows the events AMT itself writes: failed logins, BIOS progress
and errors, watchdog and boot failures.
"""
from __future__ import annotations
import base64
import binascii
from typing import Any
_RECORD_LENGTH = 21
#: Entity codes (IPMI entity IDs) -> name.
_ENTITIES = {
0: "Unspecified",
1: "Other",
2: "Unknown",
3: "Processor",
4: "Disk",
5: "Peripheral",
6: "System management module",
7: "System board",
8: "Memory module",
9: "Processor module",
10: "Power supply",
11: "Add in card",
12: "Front panel board",
13: "Back panel board",
14: "Power system board",
15: "Drive backplane",
16: "System internal expansion board",
17: "Other system board",
18: "Processor board",
19: "Power unit",
20: "Power module",
21: "Power management board",
22: "Chassis back panel board",
23: "System chassis",
24: "Sub chassis",
25: "Other chassis board",
26: "Disk drive bay",
27: "Peripheral bay",
28: "Device bay",
29: "Fan cooling",
30: "Cooling unit",
31: "Cable interconnect",
32: "Memory device",
33: "System management software",
34: "BIOS",
35: "Intel(r) ME",
36: "System bus",
37: "Group",
38: "Intel(r) ME",
39: "External environment",
40: "Battery",
41: "Processing blade",
42: "Connectivity switch",
43: "Processor/memory module",
44: "I/O module",
45: "Processor I/O module",
46: "Management controller firmware",
47: "IPMI channel",
48: "PCI bus",
49: "PCI express bus",
50: "SCSI bus",
51: "SATA/SAS bus",
52: "Processor front side bus",
}
_FIRMWARE_ERRORS = {
0: "Unspecified.",
1: "No system memory is physically installed in the system.",
2: "No usable system memory, all installed memory has experienced an unrecoverable failure.",
3: "Unrecoverable hard-disk/ATAPI/IDE device failure.",
4: "Unrecoverable system-board failure.",
5: "Unrecoverable diskette subsystem failure.",
6: "Unrecoverable hard-disk controller failure.",
7: "Unrecoverable PS/2 or USB keyboard failure.",
8: "Removable boot media not found.",
9: "Unrecoverable video controller failure.",
10: "No video device detected.",
11: "Firmware (BIOS) ROM corruption detected.",
12: "CPU voltage mismatch (processors that share same supply have mismatched voltage "
"requirements)",
13: "CPU speed matching failure",
}
_FIRMWARE_PROGRESS = {
0: "Unspecified.",
1: "Memory initialization.",
2: "Starting hard-disk initialization and test",
3: "Secondary processor(s) initialization",
4: "User authentication",
5: "User-initiated system setup",
6: "USB resource configuration",
7: "PCI resource configuration",
8: "Option ROM initialization",
9: "Video initialization",
10: "Cache initialization",
11: "SM Bus initialization",
12: "Keyboard controller initialization",
13: "Embedded controller/management controller initialization",
14: "Docking station attachment",
15: "Enabling docking station",
16: "Docking station ejection",
17: "Disabling docking station",
18: "Calling operating system wake-up vector",
19: "Starting operating system boot process",
20: "Baseboard or motherboard initialization",
21: "reserved",
22: "Floppy initialization",
23: "Keyboard test",
24: "Pointing device test",
25: "Primary processor initialization",
}
_WATCHDOG_STATES = {1: "Not Started", 2: "Stopped", 4: "Running", 8: "Expired", 16: "Suspended"}
_FIXED = {
30: "No bootable media",
32: "Operating system lockup or power interrupt",
35: "System boot failure",
37: "System firmware started (at least one CPU is properly executing).",
}
def _severity(code: int) -> str:
"""PET severity: 16 critical, 32 non-recoverable; 8 non-critical; the rest is fine."""
if code >= 16:
return "Critical"
if code >= 8:
return "Warning"
return "OK"
def _description(sensor_type: int, offset: int, data: bytes) -> str:
if sensor_type == 6:
return (
f"Authentication failed {data[1] + (data[2] << 8)} times. "
"The system may be under attack."
)
if sensor_type == 15:
if data[0] == 235:
return "Invalid Data"
table = _FIRMWARE_ERRORS if offset == 0 else _FIRMWARE_PROGRESS
return table.get(data[1], f"code {data[1]}")
if sensor_type == 18:
if data[0] == 170:
agent = f"{data[4]:x}{data[3]:x}{data[2]:x}{data[1]:x}-{data[6]:x}{data[5]:x}"
state = _WATCHDOG_STATES.get(data[7], str(data[7]))
return f"Agent watchdog {agent}-... changed to {state}"
return "Unknown event data field"
return _FIXED.get(sensor_type, f"sensor type {sensor_type}")
def parse_event_record(record: str, entry_id: str) -> dict[str, Any] | None:
"""One base64 record as a role event log entry; None for a record that is not one."""
try:
raw = base64.b64decode(record, validate=True)
except (binascii.Error, ValueError):
return None
if len(raw) < _RECORD_LENGTH:
return None
timestamp = int.from_bytes(raw[0:4], "little")
sensor_type, offset, severity, entity = raw[5], raw[7], raw[9], raw[11]
entity_name = _ENTITIES.get(entity, f"Entity {entity}")
return {
"id": entry_id,
"timestamp": float(timestamp) if timestamp else None,
"severity": _severity(severity),
"message": f"{entity_name}: {_description(sensor_type, offset, raw[13:21])}",
}
+57
View File
@@ -0,0 +1,57 @@
"""AMT's own network ports (``AMT_EthernetPortSettings``).
AMT shares the machine's network controller: the wired port has the host's MAC address
and, in DHCP mode, the host's IP address too. A port without a MAC address is a wireless
port on a machine without a WLAN card.
"""
from __future__ import annotations
import ipaddress
from typing import Any
from napalm_amt.parsers.common import is_true, text
_Instance = dict[str, Any]
_WIRELESS = "3" # PhysicalConnectionType: 0 integrated LAN, 1 discrete LAN, 2 dock, 3 WLAN
def _mac(value: str) -> str:
mac = value.replace("-", ":").lower()
return "" if not mac.replace(":", "").strip("0") else mac
def _prefix(mask: str) -> int | None:
if not mask:
return None
try:
return ipaddress.IPv4Network(f"0.0.0.0/{mask}").prefixlen
except ValueError:
return None
def parse_ethernet_ports(ports: list[_Instance]) -> list[dict[str, Any]]:
"""The ports in use: name (``wired``/``wireless``), MAC, link, DHCP, whether AMT shares
the host's IP address, and its own address where AMT reports one."""
parsed: list[dict[str, Any]] = []
for port in ports:
mac = _mac(text(port.get("MACAddress")))
if not mac:
continue
name = "wireless" if text(port.get("PhysicalConnectionType")) == _WIRELESS else "wired"
if any(p["name"] == name for p in parsed):
name = f"{name}{len(parsed)}"
dhcp = is_true(port.get("DHCPEnabled"))
ip = text(port.get("IPAddress"))
parsed.append(
{
"name": name,
"mac_address": mac,
"link_up": is_true(port.get("LinkIsUp")),
"dhcp": dhcp,
"shared_ip": is_true(port.get("SharedDynamicIP" if dhcp else "SharedStaticIp")),
"ip": ip,
"prefix_length": _prefix(text(port.get("SubnetMask"))) if ip else None,
}
)
return parsed
+55
View File
@@ -0,0 +1,55 @@
"""The machine's power state and the power actions AMT accepts.
``CIM_AssociatedPowerManagementService.PowerState`` uses the CIM power state codes, and
``AvailableRequestedPowerStates`` lists the codes AMT accepts now. AMT lists the graceful
ones (12, 14) only where the host can honour them.
"""
from __future__ import annotations
from typing import Any
from napalm_device_types.models import POWER_ACTIONS
from napalm_amt.parsers.common import as_list, to_int
_Instance = dict[str, Any]
#: Role power action -> the CIM power state AMT is asked for. 8 is "Power Off - Hard" and
#: 10 "Master Bus Reset" (a hard reset); Intel verified both, and 2 and 5, on AMT.
POWER_REQUESTS: dict[str, int] = {
"On": 2,
"GracefulShutdown": 12,
"ForceOff": 8,
"GracefulRestart": 14,
"ForceRestart": 10,
"PowerCycle": 5,
}
_ON = {2}
#: Off, but also asleep (3, 4) and hibernating (7): the OS is not running, and On wakes it.
_OFF = {3, 4, 6, 7, 8, 12, 13}
def parse_power_state(service: _Instance) -> dict[str, Any]:
"""``{"state": "On" | "Off" | "", "available": [CIM power state codes]}``."""
code = to_int(service.get("PowerState"))
state = "On" if code in _ON else "Off" if code in _OFF else ""
available = [
n
for n in (to_int(v) for v in as_list(service.get("AvailableRequestedPowerStates")))
if n is not None
]
return {"state": state, "available": available}
def accepted_actions(state: str, available: list[int]) -> list[str]:
"""The role actions that make sense in ``state`` and that AMT offers.
A machine that is off accepts On (AMT may leave 2 out of its list then). A machine that
is on accepts every other action whose state AMT lists.
"""
if state == "Off":
return ["On"] if not available or POWER_REQUESTS["On"] in available else []
if state == "On":
return [a for a in POWER_ACTIONS if a != "On" and POWER_REQUESTS[a] in available]
return []
+61
View File
@@ -0,0 +1,61 @@
"""What AMT listens on, in the role's ``get_management_protocols`` shape.
AMT has no IPMI, SSH or SNMP. It answers WS-Management and its web UI on 16992 (plain
HTTP, Digest auth) until TLS is set up, then on 16993; with TLS, plain connections from the
network are refused unless ``AcceptNonSecureConnections`` says otherwise. The KVM console,
serial-over-LAN and IDE/USB redirection (virtual media) go over the redirection port 16994,
or 16995 with TLS; KVM can also listen on the standard VNC port 5900.
"""
from __future__ import annotations
from typing import Any
from napalm_amt.parsers.common import is_true, text, to_int
from napalm_amt.wsman import PORT, TLS_PORT
_Instance = dict[str, Any]
_REDIRECTION_PORT = 16994
_REDIRECTION_TLS_PORT = 16995
_VNC_PORT = 5900
#: CIM_KVMRedirectionSAP.EnabledState: 2 enabled, 6 enabled but offline (no session yet).
_KVM_ENABLED = {2, 6}
#: AMT_RedirectionService.EnabledState: 32768 none, 32769 IDE-R, 32770 SOL, 32771 both.
_IDER_ENABLED = {32769, 32771}
def _network_tls(tls_settings: list[_Instance]) -> _Instance:
"""The TLS setting for the network ("802.3"); the LMS one only covers local access."""
return next((t for t in tls_settings if "802.3" in text(t.get("InstanceID"))), {})
def parse_management_protocols(
tls_settings: list[_Instance],
redirection: _Instance,
kvm_settings: _Instance,
kvm_saps: list[_Instance],
) -> dict[str, dict[str, Any]]:
network_tls = _network_tls(tls_settings)
tls = is_true(network_tls.get("Enabled"))
plain = not tls or is_true(network_tls.get("AcceptNonSecureConnections"))
redirection_port = _REDIRECTION_TLS_PORT if tls else _REDIRECTION_PORT
kvm_enabled = text(kvm_settings.get("EnabledByMEBx")).lower() != "false" and any(
to_int(sap.get("EnabledState")) in _KVM_ENABLED for sap in kvm_saps
)
vnc = is_true(kvm_settings.get("Is5900PortEnabled"))
ider = (
to_int(redirection.get("EnabledState")) in _IDER_ENABLED
and text(redirection.get("ListenerEnabled")).lower() != "false"
)
return {
"ipmi": {"enabled": False, "port": 623},
"ssh": {"enabled": False, "port": 22},
"http": {"enabled": plain, "port": PORT},
"https": {"enabled": tls, "port": TLS_PORT},
"snmp": {"enabled": False, "port": 161},
"kvm": {"enabled": kvm_enabled, "port": _VNC_PORT if vnc else redirection_port},
"virtual_media": {"enabled": ider, "port": redirection_port},
}
+92
View File
@@ -0,0 +1,92 @@
"""The managed machine's identity and AMT's firmware.
``CIM_Chassis`` carries the system's SMBIOS identity (manufacturer, product, serial), and
``CIM_Card`` the baseboard's. ``CIM_ComputerSystemPackage.PlatformGUID`` is the SMBIOS
system UUID, the same that the operating system on the machine reads.
"""
from __future__ import annotations
import base64
import binascii
import re
from typing import Any
from napalm_amt.parsers.common import meaningful, text
_Instance = dict[str, Any]
_HEX32 = re.compile(r"^[0-9a-fA-F]{32}$")
def _uuid_from_hex(raw: str) -> str:
"""The SMBIOS UUID from its 16 bytes as stored: the first three groups little-endian,
as dmidecode and Intel's console decode them. Empty for all-zero or all-FF bytes, which
SMBIOS defines as "not present" and "not set"."""
if not _HEX32.match(raw or ""):
return ""
raw = raw.lower()
if raw in ("0" * 32, "f" * 32):
return ""
b = bytes.fromhex(raw)
swapped = b[3::-1] + b[5:3:-1] + b[7:5:-1] + b[8:]
h = swapped.hex()
return f"{h[:8]}-{h[8:12]}-{h[12:16]}-{h[16:20]}-{h[20:]}"
def parse_platform_guid(guid: str) -> str:
"""``CIM_ComputerSystemPackage.PlatformGUID`` (32 hex digits) as the SMBIOS UUID."""
return _uuid_from_hex(text(guid))
def uuid_from_base64(value: str) -> str:
"""``AMT_SetupAndConfigurationService.GetUuid``'s base64 UUID as the SMBIOS UUID."""
try:
raw = base64.b64decode(text(value), validate=True)
except (binascii.Error, ValueError):
return ""
return _uuid_from_hex(raw.hex())
def parse_software_identity(identities: list[_Instance]) -> dict[str, str]:
"""AMT's version with its build number (``12.0.67.1579``, as AMT's ``Server`` header
and Intel's advisories write it) and the ME firmware (``Flash``) version."""
versions = {text(i.get("InstanceID")): text(i.get("VersionString")) for i in identities}
amt = versions.get("AMT", "")
build = versions.get("Build Number", "")
return {
"amt": f"{amt}.{build}" if amt and build else amt,
"flash": versions.get("Flash", ""),
}
def parse_system_identity(chassis: list[_Instance], cards: list[_Instance]) -> dict[str, str]:
"""Manufacturer, model and serial of the machine, from the chassis and, where it holds
only placeholders, the baseboard; plus the baseboard's own model and serial."""
system = chassis[0] if chassis else {}
board = cards[0] if cards else {}
def pick(key: str) -> str:
return meaningful(system.get(key)) or meaningful(board.get(key))
return {
"manufacturer": pick("Manufacturer"),
"model": pick("Model"),
"serial_number": pick("SerialNumber"),
"board_model": meaningful(board.get("Model")),
"board_serial": meaningful(board.get("SerialNumber")),
}
def parse_firmware(identity: dict[str, str], bios: list[_Instance]) -> list[dict[str, Any]]:
"""AMT and the BIOS. Neither is updated through AMT, so ``updateable`` is unknown."""
primary = next(
(b for b in bios if text(b.get("PrimaryBIOS")).lower() == "true"), bios[0] if bios else {}
)
firmware = []
if identity.get("amt"):
firmware.append({"name": "Intel AMT", "version": identity["amt"], "updateable": None})
if text(primary.get("Version")):
firmware.append(
{"name": "BIOS", "version": text(primary.get("Version")), "updateable": None}
)
return firmware