feat: declare get_management_peer for choosing an SNMP agent's sources

The caller has to name the sources an SNMP agent answers, and cannot reliably tell its own address as the device sees it (NAT, relays). SnmpAgentMixin now declares get_management_peer(): the address the device sees the management connection come from, or None. Optional: a driver that cannot tell does not implement it.
feat: fix routines as templates with the caller's values; deprecate run_device_action (4.12.0)
2026-10-16 15:18:45 +02:00 · 2026-10-16 12:06:19 +02:00 · 2026-10-16 12:00:06 +02:00 · 2026-10-16 09:19:00 +00:00 · 2026-10-16 11:11:16 +02:00 · 2026-10-16 09:10:36 +00:00
50 changed files with 2204 additions and 143 deletions
@@ -12,7 +12,7 @@ jobs:
    strategy:
      fail-fast: false
      matrix:
-        python-version: ["3.10", "3.11", "3.12"]
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
    steps:
      - name: Checkout
        uses: actions/checkout@v4
@@ -28,6 +28,10 @@ jobs:
          python -m pip install --upgrade pip
          python -m pip install -e ".[dev]"

+      - name: Type-check
+        run: |
+          python -m mypy napalm_device_types
+
      - name: Run unit tests
        run: |
          python -m pytest -q --tb=short
@@ -36,6 +36,16 @@ class QnapQtsDriver(StorageDriver, HypervisorDriver, LinuxDriver):
 `roles_of(cls)`, `role_keys_of(cls)` and `primary_role_of(cls)` read that back. Nothing
 restates the ranking: there is no precedence table and no attribute to override.

+A server's baseboard management controller (Supermicro BMC, Dell iDRAC, HPE iLO,
+Lenovo XCC) is a role of its own, `ManagementControllerDriver` (`management_controller`),
+reached over Redfish or IPMI (`USES_SSH = False`). It is not the server it manages:
+`reboot_host` restarts the BMC (Redfish `Manager.Reset`), while `set_power_state`
+(`On`, `GracefulShutdown`, `ForceOff`, `GracefulRestart`, `ForceRestart`, `PowerCycle`)
+acts on the server. `get_hardware_inventory` carries the server's serial number, system
+UUID and NIC MACs -- what ties the BMC to the host it manages. Its sensors come through
+NAPALM's `get_environment`, with `cpu` and `memory` empty, since a BMC does not know the
+host OS's load; it has no `HealthMetricsMixin` (SNMP) and no command channel.
+
 ### Role bases declare; they never implement

 **A role base must not contain a single runtime method — not even a
@@ -64,6 +74,14 @@ Two things follow, and both are improvements:
 - **A method that was never implemented raises `AttributeError`, not
  `NotImplementedError`.** Ask before calling.

+The rule covers this package's methods, not NAPALM's. `NetworkDriver` still carries
+a runtime `NotImplementedError` stub for each of its getters (`get_interfaces`,
+`get_environment`, `get_users`, …), and every role base inherits it. So a NAPALM getter
+must come from the driver class itself or from a mixin listed **before** the first role
+base; a mixin listed after it loses the MRO to NAPALM's stub. For the same reason
+`hasattr` says nothing about NAPALM's getters, and the OS role's account listing is
+`get_os_users`, not `get_users` -- NAPALM's `get_users` has a different shape.
+
 ## Function classes: what a device *can do*

 Behaviour shared across roles lives in a function class, exactly once, and a role base
@@ -73,6 +91,32 @@ is a thin bundle over them — `PackageManagementMixin`, `HealthMetricsMixin`,
 `HostRebootMixin` (`reboot_host`) is mixed into `DeviceTypeDriver` itself, since any
 device may be restartable; like the others it only declares.

+**Fixes are templates** (`napalm_device_types.fixes`, since 4.12). A change the
+application wants made on a device runs as fixed steps the driver implements, with values
+the caller supplies: `SnmpAgentMixin.configure_snmp_agent(settings)` (install, configure,
+firewall, start), `EnsurePackageMixin.ensure_package(name)`,
+`PackageProxyMixin.configure_package_proxy(url)` and
+`ContainerEngineMixin.grant_container_engine_access(user)`. A driver whose device takes the
+change in one call overrides the template method. A driver mixes in the fixes its device
+can take; none is on a role base. They replace `run_device_action`, which is deprecated.
+
+**SNMP is the application's** (`docs/STRUCTURE.md`). `HealthMetricsMixin` is deprecated
+since 4.11 and goes in 5.0. A driver does not collect over SNMP; a driver whose device
+publishes its health values under OIDs of its own declares them as data:
+
+```python
+class MySwitchDriver(SwitchDriver):
+    SNMP_HEALTH_PROFILE: ClassVar[SnmpHealthProfile] = {
+        "uptime": {"oids": ["1.3.6.1.2.1.1.3.0"]},
+        "cpu": {"oids": ["<vendor cpu oid>", "<older firmware's>"]},
+        "memory": {"total": {"oids": ["<oid>"]}, "free": {"oids": ["<oid>"]}, "unit": "KiB"},
+    }
+```
+
+A value is `oids` (the first that answers wins) or a `table` folded by `first`, `average`
+or `max`, with an optional `pattern` for an agent that answers with text. What the profile
+does not name, the application reads from the standard UCD-MIB and IF-MIB.
+
 `KernelFactsMixin` (`get_kernel_facts`) is the exception that is mixed in by a driver
 rather than by a role base: what a Linux kernel has built and loaded is read the same way
 everywhere, so the command and its parse are concrete here and a driver supplies only
@@ -110,6 +154,22 @@ A mechanism it could not read completely is `None` ("did not look"), a host with
 `[]`. The command is about 4 kB, so a driver sends it on an exec channel
 (`_run_virtual_ips_command(command, privileged=)`), not typed into a shell.

+A firewall's CARP addresses are a third mechanism, `carp`: a firewall driver reads them from
+its own API and reports them under the reading's `carp` key (`vip_manager` and `keepalived`
+then None). What a CARP state means is protocol knowledge and stays in that driver. The
+contract only carries the result: unlike the other two, a CARP entry says who holds the
+address (`holds`), because a firewall's interface list need not show it. `holds` None
+leaves the holder to the interfaces.
+
+`SystemIdentityMixin` (`get_system_identity`) reads a host's SMBIOS identity -- system UUID,
+product serial and board serial from `/sys/class/dmi/id` -- which is what ties a server to
+its BMC (`ManagementControllerDriver.get_hardware_inventory` reports the same values from
+the other side). Vendor placeholders (`0123456789` on Supermicro boards, `To be filled by
+O.E.M.`, `Default string`, all-zero or the common `03000200-…-000700080009` UUID) read as
+empty: matching on them would join unrelated machines. `is_placeholder_identity` is the
+same filter for a BMC driver's side. `product_uuid` and `product_serial` are root's, so a
+driver supplies `_run_system_identity_command(command, privileged=)`.
+
 `OpenSshServerMixin` (`harden_ssh_server`, `get_ssh_server_algorithms`) takes algorithms
 out of what an OpenSSH server offers -- the ones ssh-audit flags. It reads the lists sshd
 really uses (`sshd -T`), takes out what was asked and writes the rest to
@@ -183,14 +243,15 @@ Concretely: matching/comparison/reconciliation algorithms, orchestration flows,
 generic data shapes belong here. Only the actual device communication belongs in
 `vendor/napalm-<name>`.

-**Worked example — firewall rule diff/apply** (`FirewallDriver`):
+**Worked example — firewall rule diff/apply** (`FirewallRuleMixin`, in `FirewallDriver`):

 ```python
-class FirewallDriver(DeviceTypeDriver):
-    # Abstract — every driver implements its own device communication.
-    def get_firewall_rules(self) -> List[FirewallRuleDict]: raise NotImplementedError
-    def apply_firewall_rule(self, rule: FirewallRuleDict, *, uuid: Optional[str] = None) -> Dict[str, Any]: raise NotImplementedError
-    def commit_firewall_rules(self) -> Dict[str, Any]: raise NotImplementedError
+class FirewallRuleMixin:            # mixed into FirewallDriver
+    # Declared only — every driver implements its own device communication.
+    if TYPE_CHECKING:
+        def get_firewall_rules(self) -> List[FirewallRuleDict]: ...
+        def apply_firewall_rule(self, rule: FirewallRuleDict, *, uuid: Optional[str] = None) -> Dict[str, Any]: ...
+        def commit_firewall_rules(self) -> Dict[str, Any]: ...

    # Concrete — the matching/comparison/orchestration algorithm is identical
    # for every firewall vendor, so it lives here once.
@@ -242,7 +303,7 @@ detection vs. presentation there.
 pip install napalm-device-types
 ```

-Requires Python ≥ 3.9 and NAPALM ≥ 4.0.
+Requires Python ≥ 3.10 (tested up to 3.13) and NAPALM ≥ 4.0.

 ## Available base classes

@@ -255,12 +316,37 @@ Requires Python ≥ 3.9 and NAPALM ≥ 4.0.
 | `OSDriver` | General-purpose operating systems | Linux, BSD, macOS |
 | `StorageDriver` | Storage appliances & NAS/SAN | TrueNAS, Synology DSM, QNAP QTS |
 | `ResidentialGatewayDriver` | Router + firewall + AP in one box | OpenWrt, FritzBox |
+| `PhoneDriver` | IP telephony endpoints | Yealink |
+| `MediaDriver` | Speakers, streamers, media renderers | Sonos, Chromecast |
+| `ManagementControllerDriver` | Server baseboard management controllers | Supermicro BMC, Dell iDRAC, HPE iLO, Lenovo XCC |

-Mixins mixed into the classes above rather than used on their own:
-`ConfigLifecycleMixin` (config load/compare/commit/rollback), `PingSweepMixin`
-(subnet sweeps), and `DhcpServerMixin` (static DHCP reservations — mixed into
-`FirewallDriver` and `ResidentialGatewayDriver`, since both commonly run the
-DHCP server for their networks).
+Mixins that come with the classes above: `PingSweepMixin` (subnet sweeps, in
+every driver through `DeviceTypeDriver`) and `DhcpServerMixin` (static DHCP
+reservations — mixed into `FirewallDriver` and `ResidentialGatewayDriver`,
+since both commonly run the DHCP server for their networks).
+`ConfigLifecycleMixin` (config load/compare/commit/rollback) is mixed in by a
+driver itself.
+
+### Fingerprints
+
+`DeviceTypeDriver` carries what discovery matches a device against: `PORT_SPECS`,
+`SNMP_OBJECT_ID_PREFIX`, `SNMP_FINGERPRINT`, `SSH_FINGERPRINT`, `HTTP_FINGERPRINT` and
+`OUI_PREFIXES`. A `FingerprintRule` is a lowercase substring with a weight; `mandatory`
+disqualifies the driver when it is missing, `negative` subtracts.
+
+Discovery compares drivers by score divided by the most a driver could score. Rules that
+can never match together, such as one vendor's landing page or another's, share a
+`group`: only the strongest rule of a group counts towards that maximum
+(`rules_max_weight`). Without it, a driver that recognises three vendors' pages would
+score a third as well as one that recognises a single page:
+
+```python
+HTTP_FINGERPRINT = [
+    FingerprintRule("aten international", 6.0, group="landing_page"),  # Supermicro
+    FingerprintRule("idrac", 6.0, group="landing_page"),
+    FingerprintRule("integrated lights-out", 6.0, group="landing_page"),  # HPE iLO
+]
+```

 ## Usage

@@ -363,8 +449,8 @@ class LinuxDriver(OSDriver):
        # return List[ServiceDict]
        ...

-    def get_users(self):
-        # return List[UserDict]
+    def get_os_users(self):
+        # return List[UserDict] -- not NAPALM's get_users, whose shape differs
        ...

    def get_processes(self):
@@ -29,6 +29,7 @@ Role bases -- what a device *is*:
 * :class:`~napalm_device_types.storage.StorageDriver`
 * :class:`~napalm_device_types.phone.PhoneDriver`
 * :class:`~napalm_device_types.media.MediaDriver`
+* :class:`~napalm_device_types.management_controller.ManagementControllerDriver`
 * :class:`~napalm_device_types.residential_gateway.ResidentialGatewayDriver`

 Function classes -- what a device *can do*. Shared behaviour lives here once
@@ -59,7 +60,7 @@ Introspection -- :func:`~napalm_device_types.roles.roles_of`,
 :func:`~napalm_device_types.roles.primary_role_of`.
 """

-from napalm_device_types.base import DeviceTypeDriver, FingerprintRule, PortSpec
+from napalm_device_types.base import DeviceTypeDriver, FingerprintRule, PortSpec, rules_max_weight
 from napalm_device_types.access_point import AccessPointDriver
 from napalm_device_types.channel import (
    ByteStream,
@@ -80,6 +81,12 @@ from napalm_device_types.firewall import FirewallDriver
 from napalm_device_types.hypervisor import HypervisorDriver
 from napalm_device_types.os import OSDriver
 from napalm_device_types.firewall_rules import FirewallRuleMixin
+from napalm_device_types.fixes import (
+    EnsurePackageMixin,
+    PackageProxyMixin,
+    SnmpAgentMixin,
+    run_fix_steps,
+)
 from napalm_device_types.health_metrics import HealthMetricsMixin
 from napalm_device_types.host_reboot import HostRebootMixin
 from napalm_device_types.interface_filter import InterfaceFilterMixin
@@ -92,7 +99,25 @@ from napalm_device_types.listening import (
    parse_listening_sockets,
 )
 from napalm_device_types.mac_acl import MacAclMixin
+from napalm_device_types.management_controller import ManagementControllerDriver
 from napalm_device_types.media import MediaDriver
+from napalm_device_types.models import (
+    POWER_ACTIONS,
+    EventLogEntryDict,
+    FirmwareDict,
+    HardwareInventoryDict,
+    ManagementProtocolsDict,
+    PowerAction,
+    PowerStateDict,
+    FixResultDict,
+    FixStepDict,
+    FixStepOutcomeDict,
+    SnmpAgentSettingsDict,
+    SnmpHealthProfile,
+    SnmpInterfacesSpec,
+    SnmpMemorySpec,
+    SnmpValueSpec,
+)
 from napalm_device_types.nat_vpn import NatVpnMixin
 from napalm_device_types.packages import PackageManagementMixin
 from napalm_device_types.phone import PhoneDriver
@@ -114,6 +139,12 @@ from napalm_device_types.systemd import (
    parse_systemd_services,
 )
 from napalm_device_types.updates import UpdateMixin
+from napalm_device_types.system_identity import (
+    SYSTEM_IDENTITY_COMMAND,
+    SystemIdentityMixin,
+    is_placeholder_identity,
+    parse_system_identity,
+)
 from napalm_device_types.virtual_ips import (
    VIRTUAL_IPS_COMMAND,
    VirtualIpsMixin,
@@ -131,6 +162,18 @@ from napalm_device_types.storage import StorageDriver
 from napalm_device_types.switch import SwitchDriver

 __all__ = [
+    "EnsurePackageMixin",
+    "FixResultDict",
+    "FixStepDict",
+    "FixStepOutcomeDict",
+    "PackageProxyMixin",
+    "SnmpAgentMixin",
+    "SnmpAgentSettingsDict",
+    "run_fix_steps",
+    "SnmpHealthProfile",
+    "SnmpInterfacesSpec",
+    "SnmpMemorySpec",
+    "SnmpValueSpec",
    "AccessPointDriver",
    "ByteStream",
    "CommandChannelMixin",
@@ -145,6 +188,7 @@ __all__ = [
    "DeviceTypeDriver",
    "DhcpServerMixin",
    "FingerprintRule",
+    "rules_max_weight",
    "FirewallDriver",
    "FirewallRuleMixin",
    "HealthMetricsMixin",
@@ -154,6 +198,14 @@ __all__ = [
    "HypervisorDriver",
    "InterfaceFilterMixin",
    "MacAclMixin",
+    "ManagementControllerDriver",
+    "POWER_ACTIONS",
+    "PowerAction",
+    "PowerStateDict",
+    "HardwareInventoryDict",
+    "EventLogEntryDict",
+    "FirmwareDict",
+    "ManagementProtocolsDict",
    "MediaDriver",
    "NatVpnMixin",
    "OSDriver",
@@ -185,6 +237,10 @@ __all__ = [
    "UpdateMixin",
    "VIRTUAL_IPS_COMMAND",
    "VirtualIpsMixin",
+    "SystemIdentityMixin",
+    "SYSTEM_IDENTITY_COMMAND",
+    "is_placeholder_identity",
+    "parse_system_identity",
    "OpenSshServerMixin",
    "SSH_ALGORITHM_CLASSES",
    "SshHardeningPlan",
@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+"""Keying live and desired entries for a diff, refusing what cannot be matched.
+
+A diff matches a desired entry to a live one by a key (a rule's description, a
+reservation's MAC, a subnet's CIDR). An empty key or one held twice cannot
+name a single entry, and guessing would overwrite one the caller never meant.
+"""
+
+from __future__ import annotations
+
+from typing import Callable, Dict, Iterable, List, Set, Tuple, TypeVar
+
+T = TypeVar("T")
+
+
+def index_live(entries: Iterable[T], key: Callable[[T], str]) -> Tuple[Dict[str, T], Set[str]]:
+    """Live entries by key, and the keys more than one of them holds.
+
+    Entries without a key are left out: they are not the caller's to match.
+    """
+    by_key: Dict[str, T] = {}
+    shared: Set[str] = set()
+    for entry in entries:
+        k = key(entry)
+        if not k:
+            continue
+        if k in by_key:
+            shared.add(k)
+        by_key[k] = entry
+    for k in shared:
+        del by_key[k]
+    return by_key, shared
+
+
+def check_desired(entries: List[T], key: Callable[[T], str], shared: Set[str], what: str) -> None:
+    """``ValueError`` for a desired entry with no key, a key given twice, or a
+    key that names more than one live entry."""
+    seen: Set[str] = set()
+    for entry in entries:
+        k = key(entry)
+        if not k:
+            raise ValueError(f"a desired entry has no {what}")
+        if k in seen:
+            raise ValueError(f"{what} {k!r} is given twice")
+        if k in shared:
+            raise ValueError(f"{what} {k!r} names more than one entry on the device")
+        seen.add(k)
@@ -3,7 +3,7 @@ from __future__ import annotations

 import asyncio
 import re
-from typing import Awaitable, Callable, Dict, Optional
+from typing import Any, Awaitable, Callable, Dict, Optional

 # ── UCD-MIB OIDs ──────────────────────────────────────────────────────────────
 OID_SYS_UPTIME  = "1.3.6.1.2.1.25.1.1.0"      # hrSystemUptime (centiseconds)
@@ -25,6 +25,12 @@ OID_IF_IN_OCT  = "1.3.6.1.2.1.2.2.1.10"
 OID_IF_OUT_OCT = "1.3.6.1.2.1.2.2.1.16"
 OID_IF_IN_ERR  = "1.3.6.1.2.1.2.2.1.14"
 OID_IF_OUT_ERR = "1.3.6.1.2.1.2.2.1.20"
+# ifXTable: 64-bit octets and Mbps. The 32-bit counters wrap every ~34 s at
+# 1 Gbit/s and ifSpeed stops at 4294 Mbps; an agent without ifXTable (SNMPv1)
+# falls back to them.
+OID_IF_HC_IN_OCT  = "1.3.6.1.2.1.31.1.1.1.6"
+OID_IF_HC_OUT_OCT = "1.3.6.1.2.1.31.1.1.1.10"
+OID_IF_HIGH_SPEED = "1.3.6.1.2.1.31.1.1.1.15"

 IF_SKIP_DEFAULT = re.compile(r'^(lo|sit\d|tun\d|docker|veth|br-|virbr)', re.IGNORECASE)

@@ -42,7 +48,7 @@ def ticks_to_seconds(raw: str | None) -> int | None:


 def build_if_metrics(
-    metrics: dict,
+    metrics: Dict[str, Any],
    descr: Dict[str, str],
    speed: Dict[str, str],
    in_oct: Dict[str, str],
@@ -50,19 +56,22 @@ def build_if_metrics(
    in_err: Dict[str, str],
    out_err: Dict[str, str],
    *,
-    if_skip: re.Pattern | None = None,
+    if_skip: re.Pattern[str] | None = None,
    tx_err_is_drop: bool = False,
+    hc_in_oct: Optional[Dict[str, str]] = None,
+    hc_out_oct: Optional[Dict[str, str]] = None,
+    high_speed: Optional[Dict[str, str]] = None,
 ) -> None:
    if if_skip is None:
        if_skip = IF_SKIP_DEFAULT
-    interfaces: dict = {}
+    interfaces: Dict[str, Any] = {}
    for idx, name in descr.items():
        if if_skip.match(name):
            continue
-        iface: dict = {"name": name}
+        iface: Dict[str, Any] = {"name": name}
        try:
-            iface["rx_bytes"]  = int(in_oct.get(idx, 0)  or 0)
-            iface["tx_bytes"]  = int(out_oct.get(idx, 0) or 0)
+            iface["rx_bytes"]  = int((hc_in_oct or {}).get(idx) or in_oct.get(idx) or 0)
+            iface["tx_bytes"]  = int((hc_out_oct or {}).get(idx) or out_oct.get(idx) or 0)
            iface["rx_errors"] = int(in_err.get(idx, 0)  or 0)
            tx_val = int(out_err.get(idx, 0) or 0)
            if tx_err_is_drop:
@@ -70,8 +79,12 @@ def build_if_metrics(
                iface["tx_drops"]  = tx_val
            else:
                iface["tx_errors"] = tx_val
+            high = (high_speed or {}).get(idx)
            spd = speed.get(idx)
-            iface["speed_mbps"] = int(spd) // 1_000_000 if spd and int(spd) > 0 else 0
+            if high and int(high) > 0:
+                iface["speed_mbps"] = int(high)
+            else:
+                iface["speed_mbps"] = int(spd) // 1_000_000 if spd and int(spd) > 0 else 0
        except (ValueError, TypeError):
            pass
        interfaces[name] = iface
@@ -84,15 +97,17 @@ async def collect_ucd_metrics(
    snmp_walk: SnmpWalkFn,
    *,
    tx_err_is_drop: bool = False,
-    if_skip: re.Pattern | None = None,
-) -> dict:
+    if_skip: re.Pattern[str] | None = None,
+) -> Dict[str, Any]:
    """Collect UCD-MIB system metrics + IF-MIB interface counters in parallel."""
    (
        uptime_raw, cpu_idle_raw,
        mem_total_raw, mem_free_raw, mem_buf_raw, mem_cache_raw,
        swap_total_raw, swap_avail_raw,
        load1_raw, load5_raw, load15_raw,
-    ), (descr, speed, in_oct, out_oct, in_err, out_err) = await asyncio.gather(
+    ), (
+        descr, speed, in_oct, out_oct, in_err, out_err, hc_in_oct, hc_out_oct, high_speed,
+    ) = await asyncio.gather(
        asyncio.gather(
            snmp_get(OID_SYS_UPTIME),
            snmp_get(OID_CPU_IDLE),
@@ -113,10 +128,13 @@ async def collect_ucd_metrics(
            snmp_walk(OID_IF_OUT_OCT),
            snmp_walk(OID_IF_IN_ERR),
            snmp_walk(OID_IF_OUT_ERR),
+            snmp_walk(OID_IF_HC_IN_OCT),
+            snmp_walk(OID_IF_HC_OUT_OCT),
+            snmp_walk(OID_IF_HIGH_SPEED),
        ),
    )

-    metrics: dict = {}
+    metrics: Dict[str, Any] = {}

    secs = ticks_to_seconds(uptime_raw)
    if secs is not None:
@@ -161,5 +179,6 @@ async def collect_ucd_metrics(
                pass

    build_if_metrics(metrics, descr, speed, in_oct, out_oct, in_err, out_err,
-                     if_skip=if_skip, tx_err_is_drop=tx_err_is_drop)
+                     if_skip=if_skip, tx_err_is_drop=tx_err_is_drop,
+                     hc_in_oct=hc_in_oct, hc_out_oct=hc_out_oct, high_speed=high_speed)
    return metrics
@@ -8,6 +8,7 @@ live connection.

 from __future__ import annotations

+from collections.abc import Iterable
 from typing import NamedTuple

 from napalm.base import NetworkDriver
@@ -24,12 +25,36 @@ class FingerprintRule(NamedTuple):
    weight    — evidence added to the driver's score on match
    mandatory — pattern absent → driver immediately disqualified
    negative  — pattern present → weight subtracted instead of added
+    group     — rules sharing a group are alternatives that never match
+                together (one vendor's landing page *or* another's); only the
+                strongest of them counts towards the most a driver can score
    """

    pattern: str
    weight: float = 1.0
    mandatory: bool = False
    negative: bool = False
+    group: str | None = None
+
+
+def rules_max_weight(rules: Iterable[FingerprintRule]) -> float:
+    """The most evidence *rules* can add together.
+
+    Every positive rule counts, except that a group of alternatives counts once,
+    with its strongest rule. Discovery divides a driver's score by its maximum to
+    compare drivers; summing alternatives would make a driver that recognises
+    three vendors' pages lose to one that recognises none.
+    """
+    total = 0.0
+    group_best: dict[str, float] = {}
+    for rule in rules:
+        if rule.negative or rule.weight <= 0:
+            continue
+        if rule.group is None:
+            total += rule.weight
+        else:
+            group_best[rule.group] = max(group_best.get(rule.group, 0.0), rule.weight)
+    return total + sum(group_best.values())


 class PortSpec(NamedTuple):
@@ -49,7 +74,7 @@ class PortSpec(NamedTuple):
    mandatory: bool = False


-class DeviceTypeDriver(PingSweepMixin, HostRebootMixin, CommandChannelMixin, NetworkDriver):
+class DeviceTypeDriver(PingSweepMixin, HostRebootMixin, CommandChannelMixin, NetworkDriver):  # type: ignore[misc]
    """Common base for all netOrk device-type drivers.

    Sits between napalm.base.NetworkDriver and the type-specific abstract
@@ -28,7 +28,7 @@ from __future__ import annotations

 import socket
 import time
-from typing import Any, NamedTuple, Optional, Protocol, TYPE_CHECKING
+from typing import Any, NamedTuple, Optional, Protocol, Tuple, TYPE_CHECKING, cast

 #: How much of a stream's stderr is kept. Enough to tell "permission denied"
 #: from "no such command"; a stream that writes megabytes there keeps its tail.
@@ -111,7 +111,7 @@ class ParamikoExecStream:
    def read(self, max_bytes: int, timeout: Optional[float] = None) -> bytes:
        self._channel.settimeout(timeout)
        try:
-            return self._channel.recv(max_bytes)
+            return cast(bytes, self._channel.recv(max_bytes))
        except socket.timeout:
            raise TimeoutError(f"no data within {timeout}s") from None
        finally:
@@ -130,7 +130,7 @@ class ParamikoExecStream:
    def exit_status(self) -> Optional[int]:
        if not self._channel.exit_status_ready():
            return None
-        return self._channel.recv_exit_status()
+        return cast(int, self._channel.recv_exit_status())

    @property
    def stderr(self) -> str:
@@ -138,7 +138,7 @@ class ParamikoExecStream:
        return self._stderr.decode("utf-8", "replace")


-def _collect(channel: Any, command: str, timeout: float) -> tuple:
+def _collect(channel: Any, command: str, timeout: float) -> Tuple[bytes, bytes]:
    out, err = b"", b""
    deadline = time.monotonic() + timeout
    while True:
@@ -147,6 +147,14 @@ def _collect(channel: Any, command: str, timeout: float) -> tuple:
        elif channel.recv_stderr_ready():
            err += channel.recv_stderr(_READ)
        elif channel.exit_status_ready():
+            # sshd may send the exit status before the last output: read both to EOF.
+            try:
+                while chunk := channel.recv(_READ):
+                    out += chunk
+                while chunk := channel.recv_stderr(_READ):
+                    err += chunk
+            except socket.timeout:
+                raise TimeoutError(f"{command!r} did not finish within {timeout}s") from None
            return out, err
        elif time.monotonic() > deadline:
            raise TimeoutError(f"{command!r} did not finish within {timeout}s")
@@ -29,10 +29,10 @@ stays truthful.
 from __future__ import annotations

 import shlex
-from typing import Any, Dict, List, Optional, Sequence, TYPE_CHECKING
+from typing import Any, Dict, List, Optional, Sequence, TYPE_CHECKING, cast

 from napalm_device_types.channel import ByteStream, CommandResult
-from napalm_device_types.models import ContainerEngineDict
+from napalm_device_types.models import ContainerEngineDict, FixResultDict

 #: Engines this package knows how to reach, and the API each speaks.
 ENGINES: Dict[str, str] = {"docker": "docker-engine"}
@@ -62,7 +62,7 @@ class ContainerEngineConnection:

    def open_api(self) -> ByteStream:
        """A stream to the engine's API: HTTP/1.1 over ``system dial-stdio``."""
-        return self._driver.open_stream(self._command(["system", "dial-stdio"]))
+        return cast(ByteStream, self._driver.open_stream(self._command(["system", "dial-stdio"])))

    def run_cli(
        self,
@@ -77,9 +77,9 @@ class ContainerEngineConnection:
        *privileged* runs it as root, the way the driver gains root for any
        command: for a call the engine refused to the login user.
        """
-        return self._driver.run_command(
+        return cast(CommandResult, self._driver.run_command(
            self._command(args), privileged=privileged, timeout=timeout, stdin=stdin
-        )
+        ))

    def stream_cli(
        self, args: Sequence[str], *, merge_stderr: bool = False, privileged: bool = False
@@ -90,9 +90,9 @@ class ContainerEngineConnection:
        progress there (``compose up``); *privileged* as for :meth:`run_cli`.
        """
        command = self._command(args)
-        return self._driver.open_stream(
+        return cast(ByteStream, self._driver.open_stream(
            f"{command} 2>&1" if merge_stderr else command, privileged=privileged
-        )
+        ))


 class ContainerEngineMixin:
@@ -111,6 +111,12 @@ class ContainerEngineMixin:

        def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream: ...

+        def grant_container_engine_access(self, user: str) -> FixResultDict:
+            """Let *user* reach the container engine without privilege (the
+            ``docker`` group on most hosts). A fix template: see
+            :mod:`napalm_device_types.fixes`. Declared only; a driver implements it."""
+            ...
+
    def _container_engine_binary(self, engine: str) -> str:
        """The host command that is *engine*'s CLI. Override where it is not on PATH."""
        return engine
@@ -26,6 +26,7 @@ from __future__ import annotations
 import re
 from typing import Any, Dict, Iterator, List, Optional, TYPE_CHECKING

+from napalm_device_types._matching import check_desired, index_live
 from napalm_device_types.models import (
    DhcpReservationDiffDict,
    DhcpReservationDict,
@@ -238,20 +239,26 @@ class DhcpServerMixin:
        those apart from ones simply no longer wanted. Callers wanting
        delete/cleanup semantics must implement that themselves, deliberately.

+        Live reservations without a MAC are never matched. A desired one
+        without a MAC, a MAC given twice, or one more than one live
+        reservation holds raise ``ValueError``.
+
        :param desired: The complete desired reservation set.
        :returns: ``{"add": [...], "update": [{"uuid", "reservation",
            "changed_fields"}, ...]}``.
+        :raises ValueError: When a desired reservation cannot name a single live one.
        """
-        live_by_mac: Dict[str, DhcpReservationDict] = {
-            normalize_mac(reservation.get("mac")): reservation
-            for reservation in self.get_dhcp_reservations()
-        }
+        def key(reservation: DhcpReservationDict) -> str:
+            return normalize_mac(reservation.get("mac"))
+
+        live_by_mac, shared = index_live(self.get_dhcp_reservations(), key)
+        check_desired(desired, key, shared, "MAC")

        add: List[DhcpReservationDict] = []
        update: List[DhcpReservationUpdateDict] = []

        for desired_reservation in desired:
-            live = live_by_mac.get(normalize_mac(desired_reservation.get("mac")))
+            live = live_by_mac.get(key(desired_reservation))
            if live is None:
                add.append(desired_reservation)
                continue
@@ -333,19 +340,24 @@ class DhcpServerMixin:
        subnet takes DHCP down for a whole VLAN, and this method cannot tell
        "no longer wanted" from "was never netOrk's to describe".

+        An empty or repeated subnet raises ``ValueError``, as for reservations.
+
        :param desired: The complete desired subnet set.
        :returns: ``{"add": [...], "update": [{"uuid", "subnet",
            "changed_fields"}, ...]}``.
+        :raises ValueError: When a desired subnet cannot name a single live one.
        """
-        live_by_cidr: Dict[str, DhcpSubnetDict] = {
-            normalize_cidr(live.get("subnet")): live for live in self.get_dhcp_subnets()
-        }
+        def key(subnet: DhcpSubnetDict) -> str:
+            return normalize_cidr(subnet.get("subnet"))
+
+        live_by_cidr, shared = index_live(self.get_dhcp_subnets(), key)
+        check_desired(desired, key, shared, "subnet")

        add: List[DhcpSubnetDict] = []
        update: List[DhcpSubnetUpdateDict] = []

        for desired_subnet in desired:
-            live = live_by_cidr.get(normalize_cidr(desired_subnet.get("subnet")))
+            live = live_by_cidr.get(key(desired_subnet))
            if live is None:
                add.append(desired_subnet)
                continue
@@ -16,6 +16,7 @@ from __future__ import annotations

 from typing import Any, Dict, Iterator, List, Optional, TYPE_CHECKING

+from napalm_device_types._matching import check_desired, index_live
 from napalm_device_types.models import (
    FirewallRuleDict,
    FirewallRuleDiffDict,
@@ -102,19 +103,27 @@ class FirewallRuleMixin:
        from ones simply no longer wanted. Callers wanting delete/cleanup
        semantics must implement that themselves, deliberately.

+        Live rules without a description are never matched. A desired rule
+        without one, two desired rules with the same one, or one whose
+        description more than one live rule holds raise ``ValueError``: any
+        guess would overwrite a rule the caller did not mean.
+
        :param desired: The complete desired rule set.
        :returns: ``{"add": [...], "update": [{"uuid", "rule",
            "changed_fields"}, ...]}``.
+        :raises ValueError: When a desired rule cannot name a single live one.
        """
-        live_by_description: Dict[str, FirewallRuleDict] = {
-            rule["description"]: rule for rule in self.get_firewall_rules()
-        }
+        def key(rule: FirewallRuleDict) -> str:
+            return (rule.get("description") or "").strip()
+
+        live_by_description, shared = index_live(self.get_firewall_rules(), key)
+        check_desired(desired, key, shared, "description")

        add: List[FirewallRuleDict] = []
        update: List[FirewallRuleUpdateDict] = []

        for desired_rule in desired:
-            live_rule = live_by_description.get(desired_rule["description"])
+            live_rule = live_by_description.get(key(desired_rule))
            if live_rule is None:
                add.append(desired_rule)
                continue
@@ -0,0 +1,151 @@
+# -*- coding: utf-8 -*-
+"""Fix routines: changes an application wants made on a device, as templates.
+
+A template fixes the order of the steps; the driver knows how each step is done on
+its device; the caller supplies every value -- the community, the allowed sources,
+the proxy URL, the user. A driver whose device takes the whole change in one call
+(a REST API) overrides the template method instead of its steps.
+
+None of these is mixed into a role base. A driver mixes in the fixes its device can
+take, so ``callable(getattr(driver, "configure_snmp_agent", None))`` stays a truthful
+answer; a driver that inherits one it cannot run sets it to ``None``.
+
+They replace ``run_device_action`` (docs/STRUCTURE.md):
+
+==========================  ===================================================
+``fix_snmp``                :meth:`SnmpAgentMixin.configure_snmp_agent`
+``fix_apt_proxy``           :meth:`PackageProxyMixin.configure_package_proxy`
+``fix_docker_permissions``  ``ContainerEngineMixin.grant_container_engine_access``
+``install_lldpd`` and co.   :meth:`EnsurePackageMixin.ensure_package`
+==========================  ===================================================
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable, List, Optional, Sequence, Tuple, TYPE_CHECKING
+
+from napalm_device_types.models import (
+    FixResultDict,
+    FixStepDict,
+    FixStepOutcomeDict,
+    PackageDict,
+    SnmpAgentSettingsDict,
+)
+
+#: One step of a fix: its name and the call that does it.
+FixStep = Tuple[str, Callable[[], FixStepOutcomeDict]]
+
+
+def run_fix_steps(steps: Sequence[FixStep]) -> FixResultDict:
+    """Run *steps* in order and stop at the first that fails.
+
+    A step that raises has failed, with the exception's message as its output.
+    """
+    done: List[FixStepDict] = []
+    for name, step in steps:
+        try:
+            outcome = step()
+            ok, output = bool(outcome["ok"]), str(outcome.get("output", ""))
+        except Exception as exc:  # noqa: BLE001 - any failure of a step ends the fix
+            ok, output = False, str(exc)
+        done.append({"step": name, "ok": ok, "output": output})
+        if not ok:
+            break
+    return {
+        "success": bool(done) and all(s["ok"] for s in done) and len(done) == len(steps),
+        "steps": done,
+        "output": "\n".join(f"[{s['step']}] {s['output']}".rstrip() for s in done),
+    }
+
+
+class SnmpAgentMixin:
+    """Run an SNMP agent on the device that answers the caller.
+
+    The steps: install the agent, write its configuration, let the allowed sources
+    through the host's firewall, start it. Whether it answers is the caller's to ask,
+    with its own SNMP client.
+    """
+
+    def configure_snmp_agent(self, settings: SnmpAgentSettingsDict) -> FixResultDict:
+        """Install, configure, open and start the agent with *settings*."""
+        return run_fix_steps(
+            [
+                ("install", self._install_snmp_agent),
+                ("configure", lambda: self._write_snmp_agent_config(settings)),
+                ("firewall", lambda: self._allow_snmp_sources(settings)),
+                ("start", self._start_snmp_agent),
+            ]
+        )
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def get_management_peer(self) -> Optional[str]:
+            """The address the device sees this management connection come from
+            (for SSH, the client side of ``SSH_CONNECTION``), or None when it cannot
+            tell. A fact for the caller choosing ``allowed_sources``; optional --
+            a driver that cannot tell does not implement it."""
+            ...
+
+        # The steps a driver implements:
+
+        def _install_snmp_agent(self) -> FixStepOutcomeDict: ...
+
+        def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict: ...
+
+        def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
+            """Let ``settings["allowed_sources"]`` reach the agent; ``ok`` with nothing
+            to do where the host has no firewall of its own."""
+            ...
+
+        def _start_snmp_agent(self) -> FixStepOutcomeDict:
+            """(Re)start the agent so it reads the new configuration, and enable it at boot."""
+            ...
+
+
+class EnsurePackageMixin:
+    """Make sure a package is installed, installing it only where it is missing.
+
+    Works over :class:`~napalm_device_types.packages.PackageManagementMixin`'s
+    ``get_packages`` and ``install_package``, which the driver implements.
+    """
+
+    def ensure_package(self, name: str) -> FixResultDict:
+        """Install *name* unless it is installed already."""
+
+        def install() -> FixStepOutcomeDict:
+            if any(p.get("name") == name and p.get("installed") for p in self.get_packages()):
+                return {"ok": True, "output": f"{name} is already installed"}
+            result: Any = self.install_package(name)  # type: ignore[func-returns-value]
+            # The contract says None (and raise on failure); some drivers still
+            # return {"success", "output"}.
+            if isinstance(result, dict):
+                return {"ok": bool(result.get("success")), "output": str(result.get("output", ""))}
+            return {"ok": True, "output": f"installed {name}"}
+
+        return run_fix_steps([("install", install)])
+
+    if TYPE_CHECKING:  # pragma: no cover
+
+        def get_packages(self) -> List[PackageDict]: ...
+
+        def install_package(self, name: str, version: str = "") -> None: ...
+
+
+class PackageProxyMixin:
+    """Point the device's package manager at a caching proxy."""
+
+    if TYPE_CHECKING:  # pragma: no cover
+
+        def configure_package_proxy(self, url: str) -> FixResultDict:
+            """Have the package manager fetch through *url*; an empty *url* removes it."""
+            ...
+
+
+__all__ = [
+    "EnsurePackageMixin",
+    "FixStep",
+    "PackageProxyMixin",
+    "SnmpAgentMixin",
+    "run_fix_steps",
+]
+
@@ -1,6 +1,11 @@
 # -*- coding: utf-8 -*-
 """SNMP health collection, shared by every device type that speaks UCD-MIB.

+**Deprecated** (docs/STRUCTURE.md): drivers do not collect over SNMP. The
+application polls; a driver whose device publishes values under its own OIDs
+declares them as ``SNMP_HEALTH_PROFILE`` (:class:`~napalm_device_types.models.SnmpHealthProfile`).
+This mixin goes in 5.0.
+
 The collection itself is generic: CPU, memory, uptime, load and per-interface
 counters come from the same standard OIDs on any UCD-MIB/IF-MIB capable device.
 Only two details vary by device type, and both are class attributes rather than
@@ -15,6 +20,7 @@ implement ``get_health_metrics`` themselves.
 from __future__ import annotations

 import re
+import warnings
 from typing import Any, Awaitable, Callable, ClassVar, Dict, cast

 from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
@@ -39,7 +45,17 @@ class HealthMetricsMixin:
        snmp_get: Callable[..., Awaitable[Any]],
        snmp_walk: Callable[..., Awaitable[Dict[str, Any]]],
    ) -> HealthMetricsDict:
-        """Collect SNMP health metrics (CPU, memory, uptime, load, interfaces)."""
+        """Collect SNMP health metrics (CPU, memory, uptime, load, interfaces).
+
+        Deprecated: the application polls; see the module docstring.
+        """
+        warnings.warn(
+            "HealthMetricsMixin.get_health_metrics is deprecated and goes in 5.0: "
+            "the application polls over SNMP; a driver declares only its OIDs, "
+            "as SNMP_HEALTH_PROFILE",
+            DeprecationWarning,
+            stacklevel=2,
+        )
        # collect_ucd_metrics is untyped shared plumbing; the shape it returns is
        # the contract HealthMetricsDict describes.
        return cast(
@@ -23,8 +23,12 @@ It is ``None`` when none of these could be read, for example in a container
 without a ``/lib/modules`` of its own.

 **Auto updates** is apt's ``APT::Periodic::Unattended-Upgrade`` (set, not "0",
-and ``apt-daily-upgrade.timer`` not disabled) or an enabled dnf-automatic timer.
-It is ``None`` on a host with neither apt nor dnf-automatic. When they last ran
+``apt-daily-upgrade.timer`` not disabled, and unattended-upgrades installed: a
+package removed but not purged leaves its "1" behind) or an enabled
+``dnf-automatic-install.timer``. ``dnf-automatic.timer`` installs only with
+``apply_updates = yes`` in ``automatic.conf`` (dnf5's in ``dnf5-plugins/``
+wins); by default it only downloads. It is ``None`` on a host with neither apt
+nor dnf-automatic. When they last ran
 successfully is ``/var/lib/apt/periodic/upgrade-stamp``, which
 ``apt.systemd.daily`` touches only after unattended-upgrade succeeded.
 ``update-success-stamp`` would say nothing: every ``apt-get update`` touches it,
@@ -68,7 +72,11 @@ _BEGIN = "HSTAT_BEGIN"
 _END = "HSTAT_END"
 _REBOOT_FILE = "/var/run/reboot-required"
 _APT_TIMER = "apt-daily-upgrade.timer"
-_DNF_TIMERS = ("dnf-automatic.timer", "dnf-automatic-install.timer")
+_DNF_TIMER = "dnf-automatic.timer"
+_DNF_INSTALL_TIMER = "dnf-automatic-install.timer"
+_DNF_TIMERS = (_DNF_TIMER, _DNF_INSTALL_TIMER)
+#: dnf4's, then dnf5's: the last ``apply_updates`` printed is the one that holds.
+_DNF_AUTOMATIC_CONF = "/etc/dnf/automatic.conf /etc/dnf/dnf5-plugins/automatic.conf"
 _UPGRADE_STAMP = "/var/lib/apt/periodic/upgrade-stamp"
 _DPKG_JOURNAL = "/var/lib/dpkg/updates"
 #: Where an upgrade writes: the kernel's modules, its initrd, the package files.
@@ -100,6 +108,10 @@ HOST_STATUS_COMMAND = (
    "echo '[freebsd-running]'; freebsd-version -r; fi; "
    "if command -v apt-config >/dev/null 2>&1; then echo '[apt-config]'; "
    "apt-config dump 2>/dev/null | grep '^APT::Periodic::Unattended-Upgrade '; "
+    # Named through a variable: spelt out, the program name would match BUSY_CHECK
+    # in this very command line.
+    "uu=unattended-upgrade; { command -v \"$uu\" || [ -x \"/usr/bin/$uu\" ]; } >/dev/null 2>&1 "
+    "&& echo '[unattended-upgrade]'; "
    f"echo '[upgrade-stamp]'; stat -c %Y {_UPGRADE_STAMP} 2>/dev/null; fi; "
    "if command -v dpkg-query >/dev/null 2>&1; then "
    f"{BUSY_CHECK}"
@@ -109,10 +121,15 @@ HOST_STATUS_COMMAND = (
    f"echo '[filesystems]'; df -Pk {_FILESYSTEMS}; "
    f"echo '[timers]'; for u in {_APT_TIMER} {' '.join(_DNF_TIMERS)}; do "
    'printf \'%s %s\\n\' "$u" "$(systemctl is-enabled "$u" 2>/dev/null)"; done; '
+    "echo '[dnf-automatic]'; grep -h '^[[:space:]]*apply_updates[[:space:]]*=' "
+    f"{_DNF_AUTOMATIC_CONF} 2>/dev/null; "
    "printf '%s%s\\n' HSTAT_ END; } 2>/dev/null | cat"
 )

 _PERIODIC = re.compile(r'^APT::Periodic::Unattended-Upgrade\s+"([^"]*)"')
+_APPLY_UPDATES = re.compile(r"^apply_updates\s*=\s*(\S*)")
+#: What libdnf takes for a true boolean.
+_TRUE = frozenset({"1", "yes", "true", "on"})
 _OFF_STATES = frozenset({"disabled", "masked"})
 _INTERVAL = re.compile(r"(\d+)([smhd]?)")
 _UNIT_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400, "": 86400}
@@ -157,18 +174,31 @@ def kernel_reboot_pending(running: str, installed: List[str]) -> Optional[str]:
    running one; otherwise None.

    The flavour is what follows the last ``-`` (``generic``, ``amd64``, ``pve``,
-    ``v8``); a kernel of another flavour is never a reason to reboot.
+    ``v8``), or on an RPM kernel (``5.14.0-427.13.1.el9_4.x86_64``) the arch
+    after the release; a kernel of another flavour is never a reason to reboot.
    """
-    flavour = running.rsplit("-", 1)[-1]
-    same = [k for k in installed if k.rsplit("-", 1)[-1] == flavour]
+    version, flavour = _kernel_flavour(running)
+    same = [k for k in installed if _kernel_flavour(k)[1] == flavour]
    if not same:
        return None
-    newest = max(same, key=lambda k: _version_key(k.rsplit("-", 1)[0]))
-    if _version_key(newest.rsplit("-", 1)[0]) > _version_key(running.rsplit("-", 1)[0]):
+    newest = max(same, key=lambda k: _version_key(_kernel_flavour(k)[0]))
+    if _version_key(_kernel_flavour(newest)[0]) > _version_key(version):
        return newest
    return None


+def _kernel_flavour(release: str) -> Tuple[str, str]:
+    """``6.8.0-87-generic`` -> ``("6.8.0-87", "generic")``,
+    ``5.14.0-427.13.1.el9_4.x86_64`` -> ``("5.14.0-427.13.1.el9_4", "x86_64")``."""
+    base, sep, tail = release.rpartition("-")
+    if not sep:
+        return release, ""
+    if tail[:1].isdigit() and "." in tail:
+        rel, _, arch = tail.rpartition(".")
+        return f"{base}-{rel}", arch
+    return base, tail
+
+
 def _freebsd_reboot(sections: Dict[str, List[str]]) -> Optional[Tuple[bool, Optional[str]]]:
    """FreeBSD's answer from freebsd-version, or None on any other system."""
    installed = (sections.get("freebsd-kernel") or [""])[0]
@@ -214,14 +244,25 @@ def _periodic(sections: Dict[str, List[str]]) -> Optional[str]:
    return match.group(1) if match else None


+def _dnf_applies(sections: Dict[str, List[str]]) -> bool:
+    """Whether dnf-automatic.timer installs what it downloads; dnf's default is no."""
+    lines = sections.get("dnf-automatic") or []
+    values = [m.group(1) for m in map(_APPLY_UPDATES.match, lines) if m]
+    return bool(values) and values[-1].lower() in _TRUE
+
+
 def _auto_updates(sections: Dict[str, List[str]]) -> Optional[bool]:
    timers = _timer_states(sections)
-    if any(timers.get(t) == "enabled" for t in _DNF_TIMERS):
+    if timers.get(_DNF_INSTALL_TIMER) == "enabled":
+        return True
+    downloads_only = timers.get(_DNF_TIMER) == "enabled"
+    if downloads_only and _dnf_applies(sections):
        return True
    if "apt-config" not in sections:
-        return None
+        return False if downloads_only else None
    switched_on = _periodic(sections) not in (None, "", "0")
-    return switched_on and timers.get(_APT_TIMER) not in _OFF_STATES
+    installed = "unattended-upgrade" in sections
+    return switched_on and installed and timers.get(_APT_TIMER) not in _OFF_STATES


 def _interval(sections: Dict[str, List[str]]) -> Optional[int]:
@@ -70,9 +70,12 @@ def _report(output: str) -> str:
        raise ValueError("no kernel facts in the output") from None
    try:
        packed = base64.b64decode("".join(lines[start + 1 : end]), validate=True)
-        return gzip.decompress(packed).decode()
+        report = gzip.decompress(packed).decode()
    except (binascii.Error, OSError, EOFError, zlib.error, UnicodeDecodeError) as exc:
        raise ValueError(f"the kernel facts could not be decoded: {exc}") from exc
+    if "[release]" not in report.splitlines():
+        raise ValueError("the kernel facts are empty: gzip or base64 is missing on the host")
+    return report


 def _sections(report: str) -> Dict[str, List[str]]:
@@ -46,7 +46,7 @@ def add_lag_interfaces(
            "is_enabled": any(interfaces[m].get("is_enabled") for m in members),
            "description": f"LAG ({', '.join(members)})",
            "last_flapped": -1.0,
-            "speed": sum(float(interfaces[m].get("speed") or 0) for m in members),
+            "speed": sum(max(float(interfaces[m].get("speed") or 0), 0.0) for m in members),
            "mtu": -1,
            "mac_address": "",
            "lag_members": members,
@@ -110,13 +110,15 @@ def _sections(lines: List[str]) -> Dict[str, List[str]]:


 def _split_local(local: str) -> Optional[Tuple[str, Optional[str], int]]:
-    """``[fe80::1%eth0]:546`` -> ``("fe80::1", "eth0", 546)``; None if no port."""
+    """``[fe80::1]%eth0:546`` (or ``[fe80::1%eth0]:546``) -> ``("fe80::1", "eth0", 546)``;
+    None if no port."""
    host, sep, port = local.rpartition(":")
    if not sep or not port.isdigit():
        return None
-    if host.startswith("[") and host.endswith("]"):
-        host = host[1:-1]
    address, _, zone = host.partition("%")
+    zone = zone.rstrip("]")
+    if address.startswith("["):
+        address = address[1:].rstrip("]")
    return address or "*", zone or None, int(port)


@@ -24,7 +24,7 @@ class MacAclMixin:

            Keys are SSID names.  Each value contains:

-            * ssid (string) - SSID name (repeated for convenience)
+            * name (string) - SSID name (repeated for convenience)
            * policy (string) - ACL mode:

              * ``"allow"`` – whitelist: only listed MACs may associate
@@ -0,0 +1,152 @@
+# -*- coding: utf-8 -*-
+"""
+Abstract base class for server baseboard management controllers (BMCs).
+
+Usage::
+
+    from napalm_device_types import ManagementControllerDriver
+
+    class IdracDriver(ManagementControllerDriver):
+        ...
+"""
+
+from typing import TYPE_CHECKING, List
+
+from napalm_device_types.base import DeviceTypeDriver
+
+if TYPE_CHECKING:  # pragma: no cover - for type checkers only
+    from napalm_device_types.models import (
+        EventLogEntryDict,
+        FirmwareDict,
+        HardwareInventoryDict,
+        ManagementProtocolsDict,
+        PowerAction,
+        PowerStateDict,
+    )
+
+
+class ManagementControllerDriver(DeviceTypeDriver):
+    """
+    Abstract intermediate driver for the management controller of a server
+    (e.g. Supermicro BMC, Dell iDRAC, HPE iLO, Lenovo XCC), reached over
+    Redfish or IPMI rather than SSH.
+
+    A BMC is not the server it manages. ``reboot_host`` restarts the BMC
+    itself (Redfish ``Manager.Reset``), never the managed server; the server is
+    switched on, off and restarted through :meth:`set_power_state`.
+
+    The BMC's sensors (temperatures, fans, PSU power) come through NAPALM's
+    ``get_environment``, with ``cpu`` and ``memory`` empty: a BMC does not know
+    the load of the operating system on its server. There is no
+    ``HealthMetricsMixin`` here -- that is SNMP's -- and
+    ``CommandChannelMixin`` stays unimplemented: a BMC offers no shell to run a
+    command in.
+    """
+
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "management_controller"
+    TYPE_LABEL: str = "Management Controller"
+    USES_SSH: bool = False
+
+    if TYPE_CHECKING:
+
+        def get_power_state(self) -> PowerStateDict:
+            """
+            Returns the managed server's power state.
+
+            * state (string) - ``"On"``, ``"Off"``, ``"PoweringOn"`` or ``"PoweringOff"``
+            * actions (list) - the :data:`~napalm_device_types.models.PowerAction`
+              values the BMC accepts right now
+
+            Example::
+
+                {"state": "On", "actions": ["GracefulShutdown", "ForceOff", "ForceRestart"]}
+            """
+            ...
+
+        def set_power_state(self, action: PowerAction) -> None:
+            """
+            Switches, shuts down or restarts the managed server.
+
+            Returns once the BMC has accepted the request; the server takes its
+            time. ``GracefulShutdown`` and ``GracefulRestart`` ask the operating
+            system (ACPI) and may be ignored by it.
+
+            :param action: One of ``On``, ``GracefulShutdown``, ``ForceOff``,
+                ``GracefulRestart``, ``ForceRestart`` or ``PowerCycle``.
+            :raises ValueError: If *action* is not one of them.
+            :raises RuntimeError: If the BMC refuses it, e.g. ``On`` for a
+                server that is already on.
+            """
+            ...
+
+        def get_hardware_inventory(self) -> HardwareInventoryDict:
+            """
+            Returns the managed server's hardware: CPUs, DIMMs, drives, power
+            supplies and NICs, with the server's serial number, its SMBIOS
+            system UUID and the MAC address of each NIC -- what ties the BMC to
+            the operating system running on the server.
+
+            Example::
+
+                {
+                    "manufacturer": "Dell Inc.",
+                    "model": "PowerEdge R650",
+                    "serial_number": "7XK2Q93",
+                    "system_uuid": "4c4c4544-0058-4b10-8032-b7c04f513933",
+                    "cpus": [{"socket": "CPU.Socket.1", "model": "Intel(R) Xeon(R) Silver 4314",
+                              "cores": 16, "threads": 32, "status": "OK"}],
+                    "memory": [...],
+                    "drives": [...],
+                    "power_supplies": [...],
+                    "nics": [{"name": "NIC.Integrated.1-1", "mac_address": "b0:7b:25:de:1a:40",
+                              "speed_mbps": 10000, "link_up": True}],
+                }
+            """
+            ...
+
+        def get_event_log(self, limit: int = 500) -> List[EventLogEntryDict]:
+            """
+            Returns the BMC's system event log (SEL, Lifecycle Log, IML),
+            newest first.
+
+            :param limit: At most this many entries.
+            """
+            ...
+
+        def get_firmware_inventory(self) -> List[FirmwareDict]:
+            """
+            Returns every firmware component the BMC reports: its own, the
+            BIOS, storage controllers, NICs, power supplies.
+
+            Example::
+
+                [
+                    {"name": "Integrated Dell Remote Access Controller",
+                     "version": "7.00.00.172", "updateable": True},
+                    {"name": "BIOS", "version": "1.13.2", "updateable": True},
+                ]
+            """
+            ...
+
+        def get_management_protocols(self) -> ManagementProtocolsDict:
+            """
+            Returns what the BMC itself listens on: IPMI, SSH, HTTP, HTTPS,
+            SNMP, the KVM console and virtual media, each with whether it is
+            enabled and its port.
+
+            Example::
+
+                {
+                    "ipmi": {"enabled": False, "port": 623},
+                    "ssh": {"enabled": True, "port": 22},
+                    "http": {"enabled": False, "port": 80},
+                    "https": {"enabled": True, "port": 443},
+                    "snmp": {"enabled": False, "port": 161},
+                    "kvm": {"enabled": True, "port": 5900},
+                    "virtual_media": {"enabled": True, "port": None},
+                }
+            """
+            ...
@@ -5,7 +5,7 @@ These supplement the types defined in napalm.base.models and are used by the
 abstract device-type driver classes in this package.
 """

-from typing import Dict, List, Optional
+from typing import Dict, List, Literal, Optional, Tuple, get_args
 from typing_extensions import NotRequired, TypedDict


@@ -404,29 +404,42 @@ class VirtualIpDict(TypedDict):
    """An address the host's HA configuration lets float between machines.

    Declared, not observed: whether the host holds the address right now is in
-    its interface addresses, not here. Only these fields leave the host -- never
-    a password, an etcd endpoint or a notify script.
+    its interface addresses, not here -- unless the mechanism says so itself.
+    CARP does (``holds``): a firewall's interface list need not show the
+    address at all, but CARP's state names the MASTER. Only these fields leave
+    the host -- never a password, an etcd endpoint or a notify script.
    """

-    mechanism: str  # "vip-manager" | "keepalived"
-    instance: str  # the vrrp_instance name, or the vip-manager unit without ".service"
+    mechanism: str  # "vip-manager" | "keepalived" | "carp"
+    #: the vrrp_instance name, the vip-manager unit without ".service", or the
+    #: CARP address's interface description
+    instance: str
    address: str  # normalised: "10.7.224.10", "fd00::10"
    prefix_length: Optional[int]  # None where the configuration gives none
    interface: Optional[str]
-    group_key: Optional[str]  # vip-manager's trigger-key, or "vrid:<n>" for keepalived
-    running: Optional[bool]  # the unit is active
+    #: vip-manager's trigger-key, "vrid:<n>" for keepalived, "vhid:<n>" for CARP
+    group_key: Optional[str]
+    running: Optional[bool]  # the unit is active; for CARP: in MASTER or BACKUP
+    #: The host holds the address now, by the mechanism's own state (CARP's
+    #: MASTER). Absent or None: the interface addresses tell.
+    holds: NotRequired[Optional[bool]]


 class VirtualIpsDict(TypedDict):
-    """What ``VirtualIpsMixin.get_virtual_ips`` read, per mechanism.
+    """What a driver's ``get_virtual_ips`` read, per mechanism.

    A list is what was looked at -- ``[]`` a host without that mechanism. None is
    "did not look": a configuration it could not read, an include it could not
-    follow, a host without systemd. A consumer keeps what it knew for a None.
+    follow, a host without systemd, a mechanism the driver does not read (a
+    firewall's reading has ``vip_manager`` and ``keepalived`` None). A consumer
+    keeps what it knew for a None, and for a missing ``carp``.
    """

    vip_manager: Optional[List[VirtualIpDict]]
    keepalived: Optional[List[VirtualIpDict]]
+    #: A firewall's CARP addresses (not read by ``VirtualIpsMixin``, which reads Linux
+    #: hosts): every address bound to a VHID, the CARP address and IP aliases on it.
+    carp: NotRequired[Optional[List[VirtualIpDict]]]


 class PortForwardDict(TypedDict):
@@ -938,6 +951,86 @@ class HealthMetricsDict(TypedDict):
    interfaces: NotRequired[Dict[str, HealthMetricsIfaceDict]]


+class FixStepOutcomeDict(TypedDict):
+    """What one step of a fix routine reports (see ``napalm_device_types.fixes``)."""
+
+    ok: bool
+    output: str
+
+
+class FixStepDict(TypedDict):
+    """One step of a fix routine, as ``FixResultDict`` lists it."""
+
+    step: str
+    ok: bool
+    output: str
+
+
+class FixResultDict(TypedDict):
+    """The result of a fix routine: every step that ran, up to the first failure."""
+
+    success: bool
+    steps: List[FixStepDict]
+    output: str  # the steps' output, one "[step] output" line each
+
+
+class SnmpAgentSettingsDict(TypedDict):
+    """What the caller wants an SNMP agent to be (``SnmpAgentMixin``)."""
+
+    community: str  # the read-only v2c community
+    allowed_sources: List[str]  # addresses or CIDRs let through to the agent; [] changes no firewall
+    contact: NotRequired[str]  # sysContact
+    location: NotRequired[str]  # sysLocation
+    listen: NotRequired[str]  # the address the agent binds to; absent: every address
+
+
+class SnmpValueSpec(TypedDict, total=False):
+    """Where an SNMP agent publishes one value, inside a ``SnmpHealthProfile``.
+
+    Either ``oids``, scalars asked in order (the first non-empty answer wins), or
+    ``table``, a column walked and folded to one value by ``fold``. ``pattern`` is a
+    regular expression whose first group is the number, for an agent that answers
+    with text.
+    """
+
+    oids: List[str]
+    table: str
+    fold: Literal["first", "average", "max"]
+    pattern: str
+
+
+class SnmpMemorySpec(TypedDict, total=False):
+    """Memory: ``total`` with ``used`` or ``free``, or ``percent`` alone."""
+
+    total: SnmpValueSpec
+    used: SnmpValueSpec
+    free: SnmpValueSpec
+    percent: SnmpValueSpec
+    unit: Literal["KiB", "bytes"]  # of total, used and free; KiB, as UCD-MIB counts, by default
+
+
+class SnmpInterfacesSpec(TypedDict, total=False):
+    """How the IF-MIB counters of a device are to be read."""
+
+    skip: str  # a pattern of interface names whose counters are noise, not signal
+    tx_errors_are_drops: bool  # the device counts dropped packets as TX errors (OPNsense)
+
+
+class SnmpHealthProfile(TypedDict, total=False):
+    """Where a device's SNMP agent publishes its health values.
+
+    A driver declares it as ``SNMP_HEALTH_PROFILE`` when its device does not answer
+    the standard UCD-MIB and IF-MIB; the application polls and computes. What a
+    profile does not name is read the standard way. Uptime is in TimeTicks, CPU in
+    percent busy.
+    """
+
+    uptime: SnmpValueSpec
+    cpu: SnmpValueSpec
+    memory: SnmpMemorySpec
+    interfaces: SnmpInterfacesSpec
+
+
 class NICConfigDict(TypedDict):
    """Network interface configuration for VM provisioning."""

@@ -1035,3 +1128,129 @@ class PingSweepResultDict(TypedDict):
    scanned: int                       # destinations actually probed
    alive_count: int                   # entries with alive=True
    truncated: bool                    # True if targets were dropped at the sweep cap
+
+
+# ---------------------------------------------------------------------------
+# Management controller  (BMC: Supermicro, iDRAC, iLO, XCC)
+# ---------------------------------------------------------------------------
+
+#: What ``set_power_state`` does to the managed server; Redfish's ResetType names.
+PowerAction = Literal[
+    "On",
+    "GracefulShutdown",
+    "ForceOff",
+    "GracefulRestart",
+    "ForceRestart",
+    "PowerCycle",
+]
+POWER_ACTIONS: Tuple[str, ...] = get_args(PowerAction)
+
+
+class PowerStateDict(TypedDict):
+    """The managed server's power, as its BMC reports it."""
+
+    state: str                    # "On", "Off", "PoweringOn" or "PoweringOff"
+    actions: List[str]            # the PowerAction values this BMC accepts now
+
+
+class BmcCpuDict(TypedDict):
+    socket: str                   # "CPU1", "Proc 1"
+    model: str                    # "Intel(R) Xeon(R) Silver 4314 CPU @ 2.40GHz"
+    cores: Optional[int]
+    threads: Optional[int]
+    status: str                   # "OK", "Warning", "Critical", "Absent"
+
+
+class BmcDimmDict(TypedDict):
+    slot: str                     # "DIMMA1", "PROC 1 DIMM 3"
+    size_bytes: Optional[int]
+    type: str                     # "DDR4", "DDR5"; empty when unknown
+    speed_mts: Optional[int]      # MT/s
+    part_number: str
+    serial_number: str
+    status: str
+
+
+class BmcDriveDict(TypedDict):
+    slot: str
+    model: str
+    serial_number: str
+    size_bytes: Optional[int]
+    media: str                    # "HDD", "SSD"; empty when unknown
+    protocol: str                 # "SATA", "SAS", "NVMe"; empty when unknown
+    status: str
+
+
+class BmcPsuDict(TypedDict):
+    slot: str
+    model: str
+    serial_number: str
+    capacity_watts: Optional[int]
+    status: str
+
+
+class BmcNicDict(TypedDict):
+    """A network port of the managed server, not of the BMC."""
+
+    name: str                     # "NIC.Integrated.1-1", "Port 1"
+    mac_address: str              # lowercase, colon-separated; empty when unknown
+    speed_mbps: Optional[int]
+    link_up: Optional[bool]
+
+
+class HardwareInventoryDict(TypedDict):
+    """The managed server's hardware, as its BMC sees it."""
+
+    manufacturer: str
+    model: str
+    serial_number: str            # the server's, not the BMC's
+    system_uuid: str              # SMBIOS system UUID, lowercase
+    cpus: List[BmcCpuDict]
+    memory: List[BmcDimmDict]
+    drives: List[BmcDriveDict]
+    power_supplies: List[BmcPsuDict]
+    nics: List[BmcNicDict]
+
+
+class EventLogEntryDict(TypedDict):
+    """One entry of the BMC's system event log (SEL / Lifecycle / IML)."""
+
+    id: str
+    timestamp: Optional[float]    # Unix time; None when the BMC gave none
+    severity: str                 # "OK", "Warning" or "Critical"
+    message: str
+
+
+class FirmwareDict(TypedDict):
+    """One firmware component: the BMC's own, BIOS, a controller, a NIC."""
+
+    name: str                     # "BMC", "BIOS", "PERC H755 Front"
+    version: str
+    updateable: Optional[bool]
+
+
+class ProtocolStateDict(TypedDict):
+    enabled: bool
+    port: Optional[int]           # None for a protocol without a port of its own
+
+
+class ManagementProtocolsDict(TypedDict):
+    """What the BMC itself listens on."""
+
+    ipmi: ProtocolStateDict
+    ssh: ProtocolStateDict
+    http: ProtocolStateDict
+    https: ProtocolStateDict
+    snmp: ProtocolStateDict
+    kvm: ProtocolStateDict
+    virtual_media: ProtocolStateDict
+
+
+class SystemIdentityDict(TypedDict):
+    """A host's SMBIOS identity, what ties it to its BMC (:mod:`napalm_device_types.system_identity`).
+
+    Each value is empty when unknown or a vendor placeholder."""
+
+    system_uuid: str              # lowercase
+    product_serial: str
+    board_serial: str
@@ -36,7 +36,8 @@ class OSDriver(UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthM

    #: Stable key netOrk exposes as ``device_class``. The order in which a
    #: driver lists its role bases is the ranking; see
-    #: :func:`napalm_device_types.roles.primary_role_of`.
+    #: :func:`napalm_device_types.roles.primary_role_of`. A driver for another
+    #: OS sets its own (``ROLE = "windows"``), which replaces this one.
    ROLE: str = "linux"
    TYPE_LABEL: str = "OS"

@@ -62,7 +63,7 @@ class OSDriver(UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthM
        # Users
        # ------------------------------------------------------------------

-        def get_users(self) -> List[UserDict]:
+        def get_os_users(self) -> List[UserDict]:
            """
            Returns a list of local OS user accounts.

@@ -200,6 +201,11 @@ class OSDriver(UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthM
            """
            Executes a named administrative action on the device.

+            **Deprecated** (docs/STRUCTURE.md): the actions become fix templates
+            in :mod:`napalm_device_types.fixes`, whose values the caller supplies --
+            ``configure_snmp_agent``, ``configure_package_proxy``,
+            ``grant_container_engine_access``, ``ensure_package``. Goes in 5.0.
+
            This method is an extensibility point for driver-specific one-off
            operations that do not fit any other NAPALM API method.  Each driver
            documents the action names it supports.
@@ -28,13 +28,18 @@ from napalm_device_types.terminal import strip_terminal_codes
 #: so it is apt's, not cat's.
 APT_UPGRADABLE_COMMAND = "{ LC_ALL=C apt list --upgradable 2>/dev/null; echo __APT_RC=$?; } | cat"

-#: Read-only. Lists the packages that a pending security advisory covers.
-DNF_SECURITY_COMMAND = "LC_ALL=C dnf updateinfo list --security --quiet 2>/dev/null"
+#: Read-only. Lists the packages that a pending security advisory covers. A
+#: failed dnf prints nothing, the same as no advisories, so its exit status is
+#: printed inside the group, as apt's is.
+DNF_SECURITY_COMMAND = (
+    "{ LC_ALL=C dnf updateinfo list --security --quiet 2>/dev/null; echo __DNF_RC=$?; } | cat"
+)

 # openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable from: 3.0.13-0ubuntu3.5]
 _APT_LINE = re.compile(r"^(\S+)/(\S+)\s+(\S+)\s+\S+\s+\[upgradable from:\s+(\S+)\]")
 _SECURITY_SUITE = "-security"
 _APT_STATUS = re.compile(r"^__APT_RC=(\d+)\s*$", re.MULTILINE)
+_DNF_STATUS = re.compile(r"^__DNF_RC=(\d+)\s*$", re.MULTILINE)


 def _joined_lines(output: str) -> List[str]:
@@ -43,23 +48,23 @@ def _joined_lines(output: str) -> List[str]:
    lines: List[str] = []
    for line in output.splitlines():
        if line.startswith(" ") and lines:
-            lines[-1] += line.strip()
+            lines[-1] += line[1:].rstrip()
        else:
            lines.append(line)
    return lines


-def _apt_listing(output: str) -> str:
-    """The listing without its exit status, or ``ValueError`` when apt failed or
-    the output was cut short -- "could not read" must never look like "nothing
-    pending"."""
+def _listing(output: str, status: re.Pattern[str], what: str) -> str:
+    """The listing without its exit status, or ``ValueError`` when the command
+    failed or the output was cut short -- "could not read" must never look like
+    "nothing pending"."""
    text = strip_terminal_codes(output)
-    statuses = _APT_STATUS.findall(text)
+    statuses = status.findall(text)
    if not statuses:
-        raise ValueError("apt list --upgradable reported no exit status; the output was cut short")
+        raise ValueError(f"{what} reported no exit status; the output was cut short")
    if statuses[-1] != "0":
-        raise ValueError(f"apt list --upgradable failed with exit status {statuses[-1]}")
-    return _APT_STATUS.sub("", text)
+        raise ValueError(f"{what} failed with exit status {statuses[-1]}")
+    return status.sub("", text)


 def parse_apt_upgradable(output: str) -> List[UpdateDict]:
@@ -71,7 +76,7 @@ def parse_apt_upgradable(output: str) -> List[UpdateDict]:
    :raises ValueError: when apt failed or its exit status never arrived.
    """
    by_name: Dict[str, UpdateDict] = {}
-    for line in _joined_lines(_apt_listing(output)):
+    for line in _joined_lines(_listing(output, _APT_STATUS, "apt list --upgradable")):
        match = _APT_LINE.match(line)
        if not match:
            continue
@@ -102,9 +107,12 @@ def parse_dnf_security(output: str) -> Set[str]:
    """The names of the packages a pending security advisory covers.

    Parses :data:`DNF_SECURITY_COMMAND`'s ``ADVISORY  SEVERITY/Sec.  NEVRA`` lines.
+
+    :raises ValueError: when dnf failed or its exit status never arrived -- every
+        update would otherwise read as not security.
    """
    names: Set[str] = set()
-    for line in output.splitlines():
+    for line in _listing(output, _DNF_STATUS, "dnf updateinfo list --security").splitlines():
        parts = line.split()
        if len(parts) >= 3 and parts[1].endswith("/Sec."):
            names.add(nevra_name(parts[-1]))
@@ -68,7 +68,6 @@ class PackageManagementMixin:
            :param name: Package name as known to the package manager.
            :param version: Exact version to install.  An empty string (default)
                installs the latest available version.
-            :raises NotImplementedError: If the driver does not support package management.
            :raises ValueError: If the package name is unknown or the version does
                not exist in any configured feed.
            :raises RuntimeError: If the installation fails on the device side
@@ -88,7 +87,6 @@ class PackageManagementMixin:
            The method blocks until the removal is complete.

            :param name: Package name to remove.
-            :raises NotImplementedError: If the driver does not support package management.
            :raises ValueError: If the package is not currently installed.
            :raises RuntimeError: If the removal fails on the device side
                (e.g. other packages depend on it).
@@ -105,7 +103,6 @@ class PackageManagementMixin:
            dictionary.  The structure is package-specific.

            :param name: Package name.
-            :raises NotImplementedError: If the driver does not support package management.
            :raises ValueError: If the package is not installed.

            Example::
@@ -140,7 +137,6 @@ class PackageManagementMixin:

            :param name: Package name.
            :param config: New configuration as a nested dictionary.
-            :raises NotImplementedError: If the driver does not support package management.
            :raises ValueError: If the package is not installed or the configuration
                contains invalid values.
            :raises RuntimeError: If the device rejects the configuration.
@@ -150,6 +150,8 @@ class PingSweepMixin:
            return {"ip": destination, "alive": False, "rtt_ms": None, "error": error}

        success = reply.get("success") or {}
+        if not isinstance(success, dict):
+            return {"ip": destination, "alive": False, "rtt_ms": None, "error": "malformed ping reply"}
        probes_sent = _as_int(success.get("probes_sent"))
        packet_loss = _as_int(success.get("packet_loss"), default=probes_sent)
        alive = bool(success.get("results")) or probes_sent > packet_loss
@@ -40,9 +40,13 @@ def roles_of(driver_cls: type) -> List[Type[Any]]:
    """Every role base in ``driver_cls``'s MRO, most significant first.

    MRO order is inheritance order, so the list reflects exactly what the driver
-    author wrote in the class definition.
+    author wrote in the class definition. A class that sets ``ROLE`` itself
+    replaces the role it inherits rather than adding one: a Windows driver on
+    :class:`~napalm_device_types.os.OSDriver` sets ``ROLE = "windows"`` and is
+    not also ``"linux"``.
    """
-    return [cls for cls in driver_cls.__mro__ if _is_role_base(cls)]
+    roles = [cls for cls in driver_cls.__mro__ if _is_role_base(cls)]
+    return [cls for cls in roles if not any(o is not cls and issubclass(o, cls) for o in roles)]


 def role_keys_of(driver_cls: type) -> List[str]:
@@ -2,7 +2,7 @@
 """Init-system services: listing them and starting/stopping them.

 Deliberately *not* the same thing as a NAS's exported shares, which live on
-``StorageServiceCapability`` as ``get_storage_services``. Those two used to
+``StorageDriver`` as ``get_storage_services``. Those two used to
 share the name ``get_services`` with incompatible return types, which is why
 a QNAP -- storage and OS at once -- could not satisfy both.

@@ -22,7 +22,9 @@ Only without such an ``Include`` is sshd_config itself rewritten, after a copy.

 1. The plan never removes the last algorithm of a class, and -- given the
   lists of the client that manages the host -- never the last one that client
-   speaks. Each one kept back is reported, with why.
+   speaks. Each one kept back is reported, with why. For host key algorithms
+   only those the host has a key for count (``ssh-keygen -l`` on each
+   ``HostKey``); when the keys cannot be read, the host key list is not touched.
 2. ``sshd -t`` must accept the result, and ``sshd -T`` must show the new lists.
   A file read before ours can set the same directive; ours is then valid and
   ignored, and that is reported rather than called a success.
@@ -36,7 +38,7 @@ from __future__ import annotations
 import time
 from dataclasses import dataclass, field
 from shlex import quote
-from typing import Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple, TYPE_CHECKING
+from typing import Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple, TYPE_CHECKING

 if TYPE_CHECKING:  # pragma: no cover - for type checkers only
    from napalm_device_types.channel import CommandResult
@@ -61,6 +63,17 @@ _HEADER = (
    "# taken out of what this server offers. Delete this file to undo it.\n"
 )

+#: Host key type, as ``ssh-keygen -l`` prints it -> the algorithms that key serves.
+#: Certificate algorithms are left out: they also need a HostCertificate.
+_KEY_TYPE_ALGORITHMS: Dict[str, Tuple[str, ...]] = {
+    "ED25519": ("ssh-ed25519",),
+    "RSA": ("rsa-sha2-512", "rsa-sha2-256", "ssh-rsa"),
+    "DSA": ("ssh-dss",),
+    "ED25519-SK": ("sk-ssh-ed25519@openssh.com",),
+    "ECDSA-SK": ("sk-ecdsa-sha2-nistp256@openssh.com",),
+}
+_ECDSA_CURVES = ("256", "384", "521")
+
 _RELOAD = (
    "if systemctl is-active --quiet ssh 2>/dev/null; then systemctl reload ssh;"
    " elif systemctl is-active --quiet sshd 2>/dev/null; then systemctl reload sshd;"
@@ -100,12 +113,46 @@ def parse_sshd_effective(text: str) -> Dict[str, List[str]]:
    return found


+def parse_sshd_hostkeys(text: str) -> List[str]:
+    """The ``hostkey`` paths from ``sshd -T`` output, in order."""
+    paths: List[str] = []
+    for raw in text.splitlines():
+        parts = raw.strip().split(None, 1)
+        if len(parts) == 2 and parts[0].lower() == "hostkey":
+            paths.append(parts[1].strip())
+    return paths
+
+
+def host_key_algorithms(fingerprints: str) -> Optional[Set[str]]:
+    """The host key algorithms the keys in ``ssh-keygen -l`` output can serve.
+
+    None when a line is not understood or there is no key: then nothing is known.
+    """
+    usable: Set[str] = set()
+    for raw in fingerprints.splitlines():
+        parts = raw.split()
+        if not parts:
+            continue
+        kind = parts[-1].strip("()")
+        if len(parts) < 3 or not parts[-1].startswith("("):
+            return None
+        if kind == "ECDSA" and parts[0] in _ECDSA_CURVES:
+            usable.add(f"ecdsa-sha2-nistp{parts[0]}")
+        elif kind in _KEY_TYPE_ALGORITHMS:
+            usable.update(_KEY_TYPE_ALGORITHMS[kind])
+        else:
+            return None
+    return usable or None
+
+
@dataclass(frozen=True)
 class Refusal:
    """One algorithm that was asked to go and stays, with why.

-    *reason* is ``"last"`` (the class would be empty), ``"client"`` (the managing
-    client would share none of the class with the server) or ``"not_offered"``
+    *reason* is ``"last"`` (the class would be empty -- for host keys: have no
+    algorithm the host has a key for), ``"client"`` (the managing client would
+    share none of the class with the server), ``"host_keys_unknown"`` (the host
+    keys could not be read, so the host key list stays) or ``"not_offered"``
    (the server does not offer it, so there is nothing to remove).
    """

@@ -131,12 +178,18 @@ def plan_ssh_hardening(
    remove: Mapping[str, Iterable[str]],
    *,
    client: Optional[Mapping[str, Iterable[str]]] = None,
+    host_keys: Optional[Iterable[str]] = None,
 ) -> SshHardeningPlan:
    """What the lists become when *remove* is taken out of *effective*.

    *client* is what the client managing this host speaks, per class. A class it
    has no list for is not guarded beyond "never empty".

+    *host_keys* are the host key algorithms the host has a key for (see
+    :func:`host_key_algorithms`). Only those count as a host key algorithm that
+    stays: sshd offers the others but cannot use them. Without it nothing is
+    known, and no host key algorithm is removed.
+
    When a removal would leave a class empty, or without an algorithm *client*
    speaks, the first of the asked-for algorithms in the server's order that
    fixes it stays, and is reported in ``refused``.
@@ -150,21 +203,32 @@ def plan_ssh_hardening(
    removed: Dict[str, List[str]] = {}
    refused: List[Refusal] = []
    not_offered: List[Refusal] = []
+    usable_keys = set(host_keys) if host_keys is not None else None
    for kind in SSH_ALGORITHM_CLASSES:
        offered = list(effective.get(kind, []))
        asked = set(remove.get(kind, []))
        not_offered += [Refusal(kind, a, "not_offered") for a in sorted(asked - set(offered))]
        going = [a for a in offered if a in asked]
-        staying = [a for a in offered if a not in asked]
+        if kind == "hostkey" and usable_keys is None:
+            refused += [Refusal(kind, a, "host_keys_unknown") for a in going]
+            going = []
+        # A host key algorithm without a key is offered, but nobody gets in with it.
+        usable = usable_keys if kind == "hostkey" and usable_keys is not None else set(offered)
+        staying = [a for a in offered if a not in going and a in usable]
        speaks = set(client[kind]) if client and kind in client else None
        if going and speaks is not None and not speaks & set(staying):
-            spared = next((a for a in going if a in speaks), None)
+            spared = next((a for a in going if a in speaks and a in usable), None)
            if spared:
                going.remove(spared)
+                staying.append(spared)
                refused.append(Refusal(kind, spared, "client"))
-        if going and not [a for a in offered if a not in going]:
-            spared = going.pop(0)
-            refused.append(Refusal(kind, spared, "last"))
+        if going and not staying:
+            spared = next((a for a in going if a in usable), None)
+            if spared is None and not [a for a in offered if a not in going]:
+                spared = going[0]
+            if spared is not None:
+                going.remove(spared)
+                refused.append(Refusal(kind, spared, "last"))
        keep[kind] = [a for a in offered if a not in going]
        if going:
            removed[kind] = going
@@ -284,12 +348,35 @@ class OpenSshServerMixin:
        result = self._ssh_run(f"cat -- {quote(path)}")
        return result.stdout if result.exit_code == 0 else None

-    def _ssh_write(self, path: str, content: str) -> None:
-        script = (
-            f'tmp={quote(path + ".netork-tmp")} && cat > "$tmp" && chmod 644 "$tmp"'
-            f' && mv -f -- "$tmp" {quote(path)} || {{ s=$?; rm -f -- "$tmp"; exit $s; }}'
+    def _ssh_write(self, path: str, content: str, mode_of: Optional[str] = None) -> None:
+        """Replace *path* with *content*, as root, without root reading stdin.
+
+        A driver gains root with ``sudo -S`` and puts the password in front of
+        stdin. A sudo that does not prompt -- NOPASSWD, a cached timestamp --
+        leaves it there, and a ``cat`` under sudo writes it into the file. So the
+        content goes, without privilege, to a file only the login user can read;
+        root then installs that file, owned by root, and removes it.
+
+        The mode is that of *mode_of* (default: *path* itself), so a 0600
+        sshd_config stays 0600 and so does its copy; 644 for a new file.
+        """
+        staged = self.run_command(
+            'umask 077 && t=$(mktemp /tmp/netork-ssh.XXXXXX) && cat > "$t" && printf %s "$t"',
+            privileged=False,
+            timeout=30,
+            stdin=content.encode(),
+        )
+        tmp = staged.stdout.strip()
+        if staged.exit_code != 0 or not tmp.startswith("/tmp/netork-ssh."):
+            raise RuntimeError(f"could not stage {path}: {(staged.stderr or staged.stdout).strip()}")
+        part = path + ".netork-tmp"
+        result = self._ssh_run(
+            # GNU stat, then BSD's; 644 for a file that does not exist yet.
+            f"r={quote(mode_of or path)}; m=$(stat -c %a -- \"$r\" 2>/dev/null"
+            f' || stat -f %Lp -- "$r" 2>/dev/null) || m=644;'
+            f' install -m "$m" -- {quote(tmp)} {quote(part)} && mv -f -- {quote(part)} {quote(path)};'
+            f" s=$?; rm -f -- {quote(tmp)} {quote(part)}; exit $s"
        )
-        result = self._ssh_run(script, stdin=content.encode())
        if result.exit_code != 0:
            raise RuntimeError(f"could not write {path}: {(result.stderr or result.stdout).strip()}")

@@ -307,10 +394,29 @@ class OpenSshServerMixin:

        :raises RuntimeError: if sshd -T fails -- no OpenSSH, or no root.
        """
+        return parse_sshd_effective(self._sshd_effective_text())
+
+    def _sshd_effective_text(self) -> str:
        result = self._ssh_run("PATH=$PATH:/usr/sbin:/sbin sshd -T")
        if result.exit_code != 0:
            raise RuntimeError(f"sshd -T failed: {(result.stderr or result.stdout).strip()}")
-        return parse_sshd_effective(result.stdout)
+        return result.stdout
+
+    def _ssh_host_key_algorithms(self, sshd_t: str) -> Optional[Set[str]]:
+        """The host key algorithms the host has a key for; None when that cannot be read.
+
+        A ``HostKey`` whose file does not exist is skipped, as sshd skips it.
+        """
+        paths = parse_sshd_hostkeys(sshd_t)
+        if not paths:
+            return None
+        result = self._ssh_run(
+            f"for k in {' '.join(quote(p) for p in paths)}; do"
+            ' [ -e "$k" ] || continue; ssh-keygen -l -f "$k" || exit 1; done'
+        )
+        if result.exit_code != 0:
+            return None
+        return host_key_algorithms(result.stdout)

    def _sshd_accepts(self) -> Optional[str]:
        """None when ``sshd -t`` accepts the configuration, else what it said."""
@@ -340,9 +446,20 @@ class OpenSshServerMixin:
        """
        say = log or (lambda _line: None)
        say("Reading the algorithms sshd offers (sshd -T)")
-        plan = plan_ssh_hardening(self.get_ssh_server_algorithms(), remove, client=client)
+        sshd_t = self._sshd_effective_text()
+        host_keys: Optional[Set[str]] = None
+        if remove.get("hostkey"):
+            say("Reading which host keys the server has (ssh-keygen -l)")
+            host_keys = self._ssh_host_key_algorithms(sshd_t)
+        plan = plan_ssh_hardening(parse_sshd_effective(sshd_t), remove, client=client, host_keys=host_keys)
+        whys = {
+            "last": "it is the last one",
+            "client": "netOrk could no longer connect",
+            "host_keys_unknown": "the host keys could not be read",
+        }
        for r in plan.refused:
-            why = "it is the last one" if r.reason == "last" else "netOrk could no longer connect"
+            last_key = r.reason == "last" and r.kind == "hostkey"
+            why = "it is the last one the host has a key for" if last_key else whys[r.reason]
            say(f"Keeping {r.algorithm}: {why}")
        kinds = plan.changed_classes
        result = SshHardeningResult(plan=plan)
@@ -379,7 +496,7 @@ class OpenSshServerMixin:

            return restore
        backup = f"{MAIN_CONFIG}.netork-{time.strftime('%Y%m%d-%H%M%S')}.bak"
-        self._ssh_write(backup, main)
+        self._ssh_write(backup, main, mode_of=MAIN_CONFIG)
        self._ssh_write(MAIN_CONFIG, render_main_config(main, plan.keep, plan.changed_classes))
        result.path, result.backup = MAIN_CONFIG, backup
        say(f"{MAIN_CONFIG} has no Include of {DROPIN_DIR}; rewrote it (copy at {backup})")
@@ -401,9 +518,14 @@ class OpenSshServerMixin:
        def back(message: str, reload: bool) -> SshHardeningResult:
            say(message)
            say("Putting the previous configuration back")
-            restore()
-            if reload:
-                self._reload_ssh_server()
+            try:
+                restore()
+                if reload:
+                    self._reload_ssh_server()
+            except Exception as exc:  # noqa: BLE001 - report it, the caller must look
+                result.message = f"{message}; putting the previous configuration back failed: {exc}"
+                say(result.message)
+                return result
            result.rolled_back = True
            result.message = message
            return result
@@ -0,0 +1,115 @@
+# -*- coding: utf-8 -*-
+"""The SMBIOS identity of a host: system UUID, product serial, board serial.
+
+What ties a server to its BMC. The BMC reports the server's serial number and
+system UUID; the host reads the same values from its firmware tables. Both sides
+read them the same way on every Linux host, so the command, its parse and the
+placeholder filter live here once, and a driver only carries the command across.
+
+**Placeholders are no identity.** Board vendors leave defaults in the DMI tables
+that are the same on thousands of machines: a Supermicro board's product serial
+is ``0123456789``, others read ``To be filled by O.E.M.`` or ``Default string``,
+and a common UUID on cheap boards is ``03000200-0400-0500-0006-000700080009``.
+Matching on those would join unrelated machines, so they read as empty.
+
+**Root.** ``product_uuid`` and ``product_serial`` are readable by root only; without
+it they come back empty, which is "not known", not a mismatch.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING
+
+from napalm_device_types.models import SystemIdentityDict
+from napalm_device_types.terminal import strip_terminal_codes
+
+_BEGIN = "IDBEGIN"
+_END = "IDEND"
+_DMI = "/sys/class/dmi/id"
+
+#: Reads the three files in one round trip, framed so a banner or prompt around the
+#: output cannot shift a field.
+SYSTEM_IDENTITY_COMMAND = (
+    f"printf '{_BEGIN}\\n%s\\n%s\\n%s\\n{_END}\\n' "
+    f"\"$(cat {_DMI}/product_uuid 2>/dev/null)\" "
+    f"\"$(cat {_DMI}/product_serial 2>/dev/null)\" "
+    f"\"$(cat {_DMI}/board_serial 2>/dev/null)\""
+)
+
+#: Lowercased defaults board vendors leave in place.
+_PLACEHOLDERS = frozenset({
+    "", "none", "n/a", "na", "not specified", "not applicable", "not available",
+    "to be filled by o.e.m.", "to be filled by oem", "default string", "default",
+    "system serial number", "base board serial number", "chassis serial number",
+    "serial number", "0123456789", "1234567890", "123456789", "o.e.m.", "oem",
+    "03000200-0400-0500-0006-000700080009",
+})
+#: A value of one repeated character ("0", "00000000", "FFFF-...") or of zeros and Fs.
+_REPEATED = re.compile(r"^(.)\1*$")
+
+
+def is_placeholder_identity(value: str | None) -> bool:
+    """Whether *value* is a vendor default rather than this machine's identity."""
+    text = (value or "").strip().lower()
+    if text in _PLACEHOLDERS:
+        return True
+    compact = text.replace("-", "").replace(" ", "")
+    return bool(_REPEATED.match(compact))
+
+
+def _clean(value: str) -> str:
+    value = value.strip()
+    return "" if is_placeholder_identity(value) else value
+
+
+def parse_system_identity(output: str) -> SystemIdentityDict:
+    """The identity from :data:`SYSTEM_IDENTITY_COMMAND`'s output, placeholders empty.
+
+    :raises ValueError: when the output carries no intact frame.
+    """
+    lines = strip_terminal_codes(output).splitlines()
+    try:
+        start = lines.index(_BEGIN) + 1
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("system identity: no intact IDBEGIN/IDEND frame") from None
+    fields = (lines[start:end] + ["", "", ""])[:3]
+    uuid, product_serial, board_serial = (_clean(f) for f in fields)
+    return {
+        "system_uuid": uuid.lower(),
+        "product_serial": product_serial,
+        "board_serial": board_serial,
+    }
+
+
+class SystemIdentityMixin:
+    """Adds :meth:`get_system_identity` to a driver that can run a command on a Linux host.
+
+    The template form (README, "Function classes"): command and parse are the same
+    everywhere and concrete here; a driver supplies only
+    :meth:`_run_system_identity_command` -- how a command reaches its host, and how it
+    gains root there. Mixed in by the drivers that can, so
+    ``hasattr(driver, "get_system_identity")`` stays a truthful answer.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_system_identity_command(self, command: str, *, privileged: bool) -> str:
+            """Run *command* on the host and return what it printed; as root when
+            *privileged* (``product_uuid`` and ``product_serial`` are root's)."""
+            ...
+
+    def get_system_identity(self) -> SystemIdentityDict:
+        """
+        Returns the host's SMBIOS identity, each value empty when unknown or a vendor
+        placeholder:
+
+        * system_uuid (str) - lowercase
+        * product_serial (str)
+        * board_serial (str)
+
+        :raises ValueError: if the output carried no intact reading.
+        """
+        output = self._run_system_identity_command(SYSTEM_IDENTITY_COMMAND, privileged=True)
+        return parse_system_identity(output)
@@ -34,7 +34,8 @@ host.
  handed to a root shell.

 **None is "did not look".** A file that exists but cannot be read, an include
-that cannot be followed, a host without systemd: that mechanism is None, and the
+that cannot be followed, a host without systemd (no ``systemctl``, or systemd not
+running as PID 1, as in a container): that mechanism is None, and the
 caller keeps what it knew. ``[]`` is a host that has none.

 **Root, and without it.** The configuration may be root's alone. The whole
@@ -60,6 +61,8 @@ from napalm_device_types.terminal import strip_terminal_codes
 _BEGIN = "VIPS_BEGIN"
 _END = "VIPS_END"
 _NO_SYSTEMD = "no-systemd"
+#: Exists only while systemd runs as PID 1; a container may carry systemctl without.
+_SYSTEMD_RUN = "/run/systemd/system"
 #: A record that says a mechanism was not fully read.
 _BLIND = frozenset({"unreadable", "too-deep"})
 _KEEPALIVED_CONF = "/etc/keepalived/keepalived.conf"
@@ -72,7 +75,9 @@ _CFG_KEYS = "(ip|netmask|mask|interface|iface|trigger-key|key)"
 #: ``instance <name>`` / ``interface`` / ``virtual_router_id`` / ``vip <addr> [dev <if>]``
 #: / ``end``, and ``unreadable``/``too-deep``/``missing <path>``. No ``>`` but in
 #: ``2>/dev/null``: comparisons are written the other way round. ``\047`` is a
-#: single quote, so the program itself needs none.
+#: single quote, so the program itself needs none. A statement ends with its line
+#: unless the next one opens with ``{``: keepalived takes ``vrrp_instance VI_2``
+#: and its brace on two lines.
 _AWK = " ".join(
    (
        r'function q(p) { return "\047" p "\047" }',
@@ -102,10 +107,11 @@ _AWK = " ".join(
        r' if (4 < d || (f in seen)) { print "too-deep " f; return }'
        r" if (!ok(f)) return; seen[f] = 1;"
        r' while (0 < (r = (getline l < f))) { sub(/[#!].*/, "", l); gsub(/[{}]/, " & ", l);'
-        r' n = split(l, t, " "); nw = 0; for (k = 1; k <= n; k++) {'
+        r' n = split(l, t, " "); if (n && nw && t[1] != "{") { stmt(w, nw, f, d); nw = 0 }'
+        r' for (k = 1; k <= n; k++) {'
        r' if (t[k] == "{") { push(nw ? w[1] : "", 1 < nw ? w[2] : ""); nw = 0 }'
-        r' else if (t[k] == "}") { stmt(w, nw, f, d); nw = 0; pop() } else w[++nw] = t[k] }'
-        r" stmt(w, nw, f, d) }"
+        r' else if (t[k] == "}") { stmt(w, nw, f, d); nw = 0; pop() } else w[++nw] = t[k] } }'
+        r" stmt(w, nw, f, d);"
        r' if (r < 0) print "unreadable " f; close(f); delete seen[f] }',
        r"function inc(p, f, d,   c, g, dir, fs, n, i) {"
        r' if (p !~ /^\//) { dir = f; sub(/[^\/]*$/, "", dir); p = dir p }'
@@ -138,7 +144,7 @@ _AWK = " ".join(
 VIRTUAL_IPS_COMMAND = (
    "PATH=$PATH:/usr/sbin:/sbin; "
    "printf '%s%s\\n' VIPS_ BEGIN; "
-    "if command -v systemctl >/dev/null 2>&1; then echo '[systemd]'; "
+    f"if command -v systemctl >/dev/null 2>&1 && [ -d {_SYSTEMD_RUN} ]; then echo '[systemd]'; "
    "for u in $(systemctl list-units --all --plain --no-legend --type=service "
    "vip-manager.service 'vip-manager@*' keepalived.service 2>/dev/null | awk '{print $1}'); do "
    "case $u in keepalived*) m=keepalived;; *) m=vip-manager;; esac; "
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

 [project]
 name = "napalm-device-types"
-version = "4.5.0"
+version = "4.12.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -34,12 +34,14 @@ classifiers = [
    "Programming Language :: Python :: 3.10",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
    "Topic :: System :: Networking",
    "Topic :: System :: Systems Administration",
    "Typing :: Typed",
 ]
 dependencies = [
    "napalm>=4.0",
+    "typing-extensions>=4.0",
 ]

 [project.urls]
@@ -59,6 +61,9 @@ dev = [
 where = ["."]
 include = ["napalm_device_types*"]

+[tool.setuptools.package-data]
+napalm_device_types = ["py.typed"]
+
 [tool.mypy]
 python_version = "3.10"
 strict = true
@@ -167,3 +167,20 @@ def test_close_closes_the_channel():
    ParamikoExecStream(ch).close()

    assert ch.closed
+
+
+def test_run_keeps_output_that_arrives_after_the_exit_status():
+    """sshd may send exit-status before the last stdout bytes."""
+
+    class EarlyExit(FakeChannel):
+        def exit_status_ready(self):
+            return True
+
+        def recv_ready(self):
+            return False
+
+    channel = EarlyExit(stdout=[b"head ", b"tail"], stderr=[b"warn"])
+
+    result = run_on_transport(FakeTransport(channel), "cmd")
+
+    assert result == CommandResult("head tail", "warn", 0)
@@ -206,3 +206,25 @@ class TestApplyReservationSet:
        assert driver.commits == 0
        assert driver.applied == []
        assert lines == ["[commit] no changes"]
+
+
+class TestKeysThatCannotNameOneReservation:
+    def test_a_live_reservation_without_mac_is_never_matched(self):
+        driver = _FakeDhcpServer([_reservation(uuid="b", mac="")])
+
+        diff = driver.diff_dhcp_reservations([_reservation()])
+
+        assert diff == {"add": [_reservation()], "update": []}
+
+    @pytest.mark.parametrize(
+        "live, desired",
+        [
+            ([], [_reservation(mac="")]),
+            ([], [_reservation(mac="AA-BB-CC-DD-EE-02"), _reservation(mac="aabb.ccdd.ee02")]),
+            ([_reservation(uuid="a"), _reservation(uuid="b")], [_reservation()]),
+        ],
+        ids=["empty desired", "desired twice", "two live share it"],
+    )
+    def test_raises(self, live, desired):
+        with pytest.raises(ValueError):
+            _FakeDhcpServer(live).diff_dhcp_reservations(desired)
@@ -226,3 +226,13 @@ def test_device_specific_methods_are_absent_until_implemented(method: str) -> No
 def test_calling_a_missing_device_method_fails_loudly() -> None:
    with pytest.raises(AttributeError):
        _BareDriver().get_dhcp_subnets()
+
+
+@pytest.mark.parametrize(
+    "desired",
+    [[_subnet(subnet="")], [_subnet(), _subnet(subnet=" 10.10.20.0/24 ")]],
+    ids=["empty", "twice"],
+)
+def test_a_subnet_that_cannot_name_one_raises(desired):
+    with pytest.raises(ValueError):
+        _FakeDhcpServer([]).diff_dhcp_subnets(desired)
@@ -8,6 +8,7 @@ from napalm_device_types import (
    DeviceTypeDriver,
    FingerprintRule,
    FirewallDriver,
+    rules_max_weight,
    HypervisorDriver,
    OSDriver,
    PortSpec,
@@ -90,6 +91,53 @@ def test_fingerprint_rule_negative():
    assert r.negative is True


+def test_fingerprint_rule_has_no_group_by_default():
+    assert FingerprintRule("fritz!box").group is None
+
+
+def test_fingerprint_rule_group():
+    r = FingerprintRule("idrac", weight=6.0, group="landing_page")
+    assert r.group == "landing_page"
+
+
+# ── rules_max_weight: the most a set of rules can add ─────────────────────────
+
+def test_max_weight_sums_rules_that_can_match_together():
+    rules = [FingerprintRule("proxmox virtual environment", 9.0), FingerprintRule("pve", 2.0)]
+    assert rules_max_weight(rules) == 11.0
+
+
+def test_max_weight_counts_one_rule_per_group():
+    # Alternatives: a page is ATEN's or iDRAC's or iLO's, never all three. Summing
+    # them made a driver that knows three BMC families lose to one that knows none.
+    rules = [
+        FingerprintRule("aten international", 6.0, group="landing_page"),
+        FingerprintRule("idrac", 6.0, group="landing_page"),
+        FingerprintRule("integrated lights-out", 4.0, group="landing_page"),
+    ]
+    assert rules_max_weight(rules) == 6.0
+
+
+def test_max_weight_mixes_groups_and_single_rules():
+    rules = [
+        FingerprintRule("redfishversion", 10.0, group="redfish_answer"),
+        FingerprintRule("@message.extendedinfo", 8.0, group="redfish_answer"),
+        FingerprintRule("#serviceroot.", 5.0),
+        FingerprintRule("aten international", 6.0, group="landing_page"),
+        FingerprintRule("idrac", 6.0, group="landing_page"),
+    ]
+    assert rules_max_weight(rules) == 21.0
+
+
+def test_max_weight_ignores_negative_rules():
+    rules = [FingerprintRule("vcenter", 9.0), FingerprintRule("vmware esxi", 9.0, negative=True)]
+    assert rules_max_weight(rules) == 9.0
+
+
+def test_max_weight_of_no_rules_is_zero():
+    assert rules_max_weight([]) == 0.0
+
+
 def test_fingerprint_rule_is_immutable():
    r = FingerprintRule("test", weight=3.0)
    with pytest.raises(AttributeError):
@@ -169,3 +169,30 @@ class TestApplyFirewallRuleset:

        assert driver.applied == []
        assert driver.committed is True
+
+
+class TestKeysThatCannotNameOneRule:
+    """An empty or shared description must never overwrite a rule by guess."""
+
+    def test_a_live_rule_without_description_is_never_matched(self):
+        driver = _FakeFirewall([_rule(uuid="auto1", description="", action="block")])
+
+        diff = driver.diff_firewall_rules([_rule()])
+
+        assert diff == {"add": [_rule()], "update": []}
+
+    @pytest.mark.parametrize(
+        "live, desired",
+        [
+            ([], [_rule(description="")]),
+            ([], [_rule(), _rule(action="block")]),
+            ([_rule(uuid="u1"), _rule(uuid="u2")], [_rule()]),
+        ],
+        ids=["empty desired", "desired twice", "two live rules share it"],
+    )
+    def test_raises(self, live, desired):
+        driver = _FakeFirewall(live)
+
+        with pytest.raises(ValueError):
+            driver.diff_firewall_rules(desired)
+        assert driver.applied == []
@@ -0,0 +1,174 @@
+"""Fix routines are templates (docs/STRUCTURE.md, NetOrk/netork#1105).
+
+The template fixes the order of the steps, the driver knows how each step is done
+on its device, the caller supplies every value. None of them is mixed into a role
+base, so ``callable(getattr(driver, name, None))`` stays a truthful answer.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_device_types import (
+    AccessPointDriver,
+    EnsurePackageMixin,
+    FirewallDriver,
+    HypervisorDriver,
+    ManagementControllerDriver,
+    MediaDriver,
+    OSDriver,
+    PhoneDriver,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+    PackageProxyMixin,
+    SnmpAgentMixin,
+    run_fix_steps,
+)
+
+ROLE_BASES = [
+    AccessPointDriver, FirewallDriver, HypervisorDriver, ManagementControllerDriver, MediaDriver,
+    OSDriver, PhoneDriver, ResidentialGatewayDriver, StorageDriver, SwitchDriver,
+]  # fmt: skip
+
+
+def _ok(output: str = ""):
+    return lambda: {"ok": True, "output": output}
+
+
+def _fail(output: str):
+    return lambda: {"ok": False, "output": output}
+
+
+class TestRunFixSteps:
+    def test_every_step_runs_in_order_and_is_reported(self):
+        result = run_fix_steps([("install", _ok("installed")), ("start", _ok("started"))])
+
+        assert result["success"] is True
+        assert result["steps"] == [
+            {"step": "install", "ok": True, "output": "installed"},
+            {"step": "start", "ok": True, "output": "started"},
+        ]
+        assert result["output"] == "[install] installed\n[start] started"
+
+    def test_the_first_failure_stops_the_rest(self):
+        ran = []
+
+        def later():
+            ran.append("later")
+            return {"ok": True, "output": ""}
+
+        result = run_fix_steps([("install", _fail("no package")), ("start", later)])
+
+        assert result["success"] is False
+        assert [s["step"] for s in result["steps"]] == ["install"]
+        assert ran == []
+
+    def test_a_step_that_raises_is_a_failed_step(self):
+        def broken():
+            raise RuntimeError("connection lost")
+
+        result = run_fix_steps([("install", broken)])
+
+        assert result["success"] is False
+        assert result["steps"] == [{"step": "install", "ok": False, "output": "connection lost"}]
+
+
+class _Agent(SnmpAgentMixin):
+    def __init__(self, fail_at: str = ""):
+        self.calls: list = []
+        self.fail_at = fail_at
+
+    def _step(self, name, *args):
+        self.calls.append((name, *args))
+        return {"ok": name != self.fail_at, "output": name}
+
+    def _install_snmp_agent(self):
+        return self._step("install")
+
+    def _write_snmp_agent_config(self, settings):
+        return self._step("configure", settings)
+
+    def _allow_snmp_sources(self, settings):
+        return self._step("firewall", settings)
+
+    def _start_snmp_agent(self):
+        return self._step("start")
+
+
+class TestSnmpAgent:
+    SETTINGS = {"community": "c0mm", "allowed_sources": ["10.0.0.5"], "contact": "ops", "location": "rack 1"}
+
+    def test_install_configure_open_and_start_with_the_caller_s_values(self):
+        agent = _Agent()
+
+        result = agent.configure_snmp_agent(self.SETTINGS)
+
+        assert result["success"] is True
+        assert agent.calls == [
+            ("install",),
+            ("configure", self.SETTINGS),
+            ("firewall", self.SETTINGS),
+            ("start",),
+        ]
+
+    def test_a_failed_install_configures_nothing(self):
+        agent = _Agent(fail_at="install")
+
+        assert agent.configure_snmp_agent(self.SETTINGS)["success"] is False
+        assert agent.calls == [("install",)]
+
+
+class _Packages(EnsurePackageMixin):
+    def __init__(self, installed):
+        self.installed = installed
+        self.asked: list = []
+
+    def get_packages(self):
+        return [{"name": n, "installed": True} for n in self.installed] + [
+            {"name": "lldpd", "installed": False}
+        ]
+
+    def install_package(self, name, version=""):
+        self.asked.append(name)
+
+
+class TestEnsurePackage:
+    def test_installs_what_is_missing(self):
+        driver = _Packages(installed=["curl"])
+
+        result = driver.ensure_package("lldpd")
+
+        assert result["success"] is True
+        assert driver.asked == ["lldpd"]
+
+    def test_leaves_alone_what_is_there(self):
+        driver = _Packages(installed=["lldpd"])
+
+        result = driver.ensure_package("lldpd")
+
+        assert result["success"] is True
+        assert driver.asked == []
+        assert "already installed" in result["output"]
+
+    def test_an_install_that_reports_failure_fails(self):
+        driver = _Packages(installed=[])
+        driver.install_package = lambda name, version="": {"success": False, "output": "E: no candidate"}
+
+        result = driver.ensure_package("lldpd")
+
+        assert result["success"] is False
+        assert "no candidate" in result["output"]
+
+
+@pytest.mark.parametrize("base", ROLE_BASES, ids=lambda b: b.__name__)
+@pytest.mark.parametrize(
+    "method",
+    ["configure_snmp_agent", "ensure_package", "configure_package_proxy", "grant_container_engine_access"],
+)
+def test_no_role_base_offers_a_fix(base, method):
+    assert getattr(base, method, None) is None
+
+
+def test_the_package_proxy_is_declared_only():
+    assert not hasattr(PackageProxyMixin, "configure_package_proxy")
@@ -11,6 +11,7 @@ The fixtures are the real states of six hosts on netOrk's test server.

 from __future__ import annotations

+import os
 import re
 import subprocess

@@ -35,7 +36,9 @@ def _wire(
    periodic=None,
    stamp=None,
    timer="enabled",
+    unattended=True,
    dnf_timers=("not-found", "not-found"),
+    dnf_conf=(),
    dpkg=None,
    journal=(),
    busy=False,
@@ -47,7 +50,10 @@ def _wire(
        lines += ["[needs-restarting]", str(needs_restarting)]
    lines += ["[kernel]", running, "[modules]", *modules]
    if periodic is not None:
-        lines += ["[apt-config]", *periodic, "[upgrade-stamp]", *([str(stamp)] if stamp else [])]
+        lines += ["[apt-config]", *periodic]
+        if unattended:
+            lines.append("[unattended-upgrade]")
+        lines += ["[upgrade-stamp]", *([str(stamp)] if stamp else [])]
    if dpkg is not None:
        if busy:
            lines.append("[dpkg-busy]")
@@ -57,6 +63,8 @@ def _wire(
        f"apt-daily-upgrade.timer {timer}",
        f"dnf-automatic.timer {dnf_timers[0]}",
        f"dnf-automatic-install.timer {dnf_timers[1]}",
+        "[dnf-automatic]",
+        *dnf_conf,
        "HSTAT_END",
    ]
    return "\n".join(lines) + "\n"
@@ -131,6 +139,13 @@ class TestKernelRebootPending:
            ("7.0.14-19-pve", ["7.0.14-19-pve", "7.0.2-6-pve"], None),
            # Arch: the running kernel's modules were replaced by the upgrade.
            ("6.10.5-arch1-1", ["6.10.9-arch1-1"], "6.10.9-arch1-1"),
+            # RHEL/Fedora: the release changes with every update, the arch does not.
+            (
+                "5.14.0-427.13.1.el9_4.x86_64",
+                ["5.14.0-427.13.1.el9_4.x86_64", "5.14.0-503.11.1.el9_5.x86_64"],
+                "5.14.0-503.11.1.el9_5.x86_64",
+            ),
+            ("6.9.7-200.fc40.x86_64", ["6.9.7-200.fc40.x86_64"], None),
        ],
    )
    def test_the_newer_kernel_of_the_running_flavour(self, running, installed, newer):
@@ -155,11 +170,57 @@ class TestAutoUpdates:

        assert status["auto_updates"] is False

-    def test_dnf_automatic(self):
-        status = parse_host_status(_wire(dnf_timers=("enabled", "not-found")))
+    def test_removed_but_not_purged_does_not_patch_itself(self):
+        """``apt remove unattended-upgrades`` leaves its conffile, and its "1", behind."""
+        status = parse_host_status(_wire(periodic=UNATTENDED, unattended=False))
+
+        assert status["auto_updates"] is False
+
+    def test_dnf_automatic_install_timer_always_installs(self):
+        status = parse_host_status(_wire(dnf_timers=("not-found", "enabled")))

        assert status["auto_updates"] is True

+    def test_dnf_automatic_with_apply_updates(self):
+        conf = ["apply_updates = yes"]
+        status = parse_host_status(_wire(dnf_timers=("enabled", "not-found"), dnf_conf=conf))
+
+        assert status["auto_updates"] is True
+
+    @pytest.mark.parametrize(
+        "conf", [[], ["apply_updates = no"], ["apply_updates=yes", "apply_updates = False"]]
+    )
+    def test_dnf_automatic_by_default_only_downloads(self, conf):
+        """dnf-automatic.timer installs only with apply_updates; the default is no.
+        dnf5's file comes last and wins."""
+        status = parse_host_status(_wire(dnf_timers=("enabled", "not-found"), dnf_conf=conf))
+
+        assert status["auto_updates"] is False
+
+    def test_the_command_reads_apply_updates_of_dnf4_then_dnf5(self):
+        conf = "/etc/dnf/automatic.conf /etc/dnf/dnf5-plugins/automatic.conf"
+        assert conf in HOST_STATUS_COMMAND
+
+    @pytest.mark.parametrize("installed", [True, False])
+    def test_the_command_asks_whether_unattended_upgrade_is_installed(self, tmp_path, installed):
+        """Run for real, with a stand-in apt-config that still says "1"."""
+        if os.path.exists("/usr/bin/unattended-upgrade"):
+            pytest.skip("this host has unattended-upgrade installed")
+        apt_config = tmp_path / "apt-config"
+        apt_config.write_text("#!/bin/sh\necho 'APT::Periodic::Unattended-Upgrade \"1\";'\n")
+        stubs = [apt_config]
+        if installed:
+            stubs.append(tmp_path / "unattended-upgrade")
+            stubs[-1].write_text("#!/bin/sh\nexit 0\n")
+        for stub in stubs:
+            stub.chmod(0o755)
+        env = {**os.environ, "PATH": f"{tmp_path}:{os.environ.get('PATH', '/usr/bin:/bin')}"}
+        out = subprocess.run(
+            ["sh", "-c", HOST_STATUS_COMMAND], capture_output=True, text=True, env=env
+        ).stdout
+
+        assert ("[unattended-upgrade]" in out) is installed
+
    def test_neither_apt_nor_dnf_is_unknown(self):
        assert parse_host_status(_wire())["auto_updates"] is None

@@ -85,6 +85,14 @@ class TestParsing:
        with pytest.raises(ValueError):
            parse_kernel_facts("sh: gzip: not found\n")

+    def test_an_empty_payload_raises(self):
+        """gzip or base64 missing on the host: the frame is there, the report is not."""
+        with pytest.raises(ValueError):
+            parse_kernel_facts("KFACTS_BEGIN\nKFACTS_END\n")
+        empty = base64.b64encode(gzip.compress(b"")).decode()
+        with pytest.raises(ValueError):
+            parse_kernel_facts(f"KFACTS_BEGIN\n{empty}\nKFACTS_END\n")
+
    def test_a_damaged_payload_raises(self):
        with pytest.raises(ValueError):
            parse_kernel_facts("KFACTS_BEGIN\nnot base64 at all!\nKFACTS_END\n")
@@ -74,3 +74,12 @@ def test_does_not_modify_its_input():
    add_lag_interfaces(ifaces)

    assert list(ifaces) == ["3"]
+
+
+def test_an_unknown_member_speed_does_not_make_the_lag_negative():
+    ifaces = {
+        "3": _port(speed=-1.0, trunk_group="Trk3"),
+        "4": _port(speed=1000.0, trunk_group="Trk3"),
+    }
+
+    assert add_lag_interfaces(ifaces)["Trk3"]["speed"] == 1000.0
@@ -84,6 +84,7 @@ class TestParsing:
            ("0.0.0.0%ens18:68", "0.0.0.0", "ens18"),
            ("[::ffff:127.0.0.1]:8125", "::ffff:127.0.0.1", None),
            ("[fe80::1%eth0]:546", "fe80::1", "eth0"),
+            ("[fe80::1]%eth0:546", "fe80::1", "eth0"),  # what iproute2 ss prints
            (":::22", "::", None),  # iproute2 4.9 prints IPv6 without brackets
        ],
    )
@@ -11,10 +11,14 @@ Fixture lines are from real hosts (Ubuntu 24.04, Debian 13 / OMV, Proxmox VE 9).

 from __future__ import annotations

+import os
+import subprocess
+
 import pytest

 from napalm_device_types.package_updates import (
    APT_UPGRADABLE_COMMAND,
+    DNF_SECURITY_COMMAND,
    nevra_name,
    parse_apt_upgradable,
    parse_dnf_security,
@@ -81,6 +85,16 @@ class TestAptUpgradable:
            "3.0.13-0ubuntu3.5"
        )

+    def test_a_line_wrapped_at_a_blank_keeps_the_blank(self):
+        wrapped = (
+            "openssl/noble-updates 3.0.13-0ubuntu3.6\n"
+            "  amd64 [upgradable from: 3.0.13-0ubuntu3.5]\n__APT_RC=0\n"
+        )
+
+        assert _by_name(parse_apt_upgradable(wrapped))["openssl"]["current_version"] == (
+            "3.0.13-0ubuntu3.5"
+        )
+
    def test_one_package_for_several_architectures_is_one_entry(self):
        multiarch = (
            "libc6/noble-updates 2.39-0ubuntu8.5 amd64 [upgradable from: 2.39-0ubuntu8.4]\n"
@@ -129,13 +143,38 @@ class TestDnfSecurity:
 FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64
 FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-1:3.1.4-2.fc40.x86_64
 RLSA-2024:1234         Moderate/Sec.  kernel-core-5.14.0-427.13.1.el9_4.x86_64
+__DNF_RC=0
 """

    def test_the_names_of_packages_with_a_security_advisory(self):
        assert parse_dnf_security(self.ADVISORIES) == {"openssl-libs", "openssl", "kernel-core"}

    def test_nothing_is_an_empty_set(self):
-        assert parse_dnf_security("") == set()
+        assert parse_dnf_security("__DNF_RC=0\n") == set()
+
+    def test_a_failed_dnf_raises_instead_of_reading_as_no_advisories(self):
+        """dnf that could not reach its repositories prints nothing to stdout;
+        every pending update would have become not security."""
+        with pytest.raises(ValueError, match="1"):
+            parse_dnf_security("__DNF_RC=1\n")
+
+    def test_output_cut_short_raises(self):
+        with pytest.raises(ValueError, match="no exit status"):
+            parse_dnf_security(self.ADVISORIES.replace("__DNF_RC=0\n", ""))
+
+    def test_the_command_reports_dnfs_exit_status_not_cats(self, tmp_path):
+        """A stand-in dnf that fails quietly, as one without its repositories does."""
+        dnf = tmp_path / "dnf"
+        dnf.write_text("#!/bin/sh\nexit 1\n")
+        dnf.chmod(0o755)
+        env = {"PATH": f"{tmp_path}:{os.environ.get('PATH', '/usr/bin:/bin')}"}
+        out = subprocess.run(
+            ["sh", "-c", DNF_SECURITY_COMMAND], capture_output=True, text=True, env=env
+        ).stdout
+
+        assert out == "__DNF_RC=1\n"
+        with pytest.raises(ValueError, match="exit status 1"):
+            parse_dnf_security(out)

    @pytest.mark.parametrize(
        ("nevra", "name"),
@@ -0,0 +1,9 @@
+"""The package says it is typed (PEP 561), so it must ship the marker."""
+
+from pathlib import Path
+
+import napalm_device_types
+
+
+def test_the_package_ships_py_typed():
+    assert (Path(napalm_device_types.__file__).parent / "py.typed").is_file()
@@ -219,3 +219,10 @@ def test_ping_sweep_stops_when_stop_requested():

    assert result["scanned"] == 1
    assert len(driver.calls) == 1
+
+
+@pytest.mark.parametrize("success", [True, "yes", ["x"]])
+def test_a_malformed_success_is_no_answer_not_an_abort(success):
+    entry = PingSweepMixin._parse_ping_reply("192.0.2.1", {"success": success})
+
+    assert entry == {"ip": "192.0.2.1", "alive": False, "rtt_ms": None, "error": "malformed ping reply"}
@@ -22,6 +22,7 @@ from napalm_device_types import (
    DeviceTypeDriver,
    FirewallDriver,
    HypervisorDriver,
+    ManagementControllerDriver,
    MediaDriver,
    OSDriver,
    PhoneDriver,
@@ -35,6 +36,7 @@ ROLE_BASES = [
    AccessPointDriver,
    FirewallDriver,
    HypervisorDriver,
+    ManagementControllerDriver,
    MediaDriver,
    OSDriver,
    PhoneDriver,
@@ -84,6 +86,12 @@ class TestDeclaredMethodsDoNotExistAtRuntime:
            (ResidentialGatewayDriver, "get_wan_status"),
            (PhoneDriver, "get_sip_accounts"),
            (MediaDriver, "get_playback_state"),
+            (ManagementControllerDriver, "get_power_state"),
+            (ManagementControllerDriver, "set_power_state"),
+            (ManagementControllerDriver, "get_hardware_inventory"),
+            (ManagementControllerDriver, "get_event_log"),
+            (ManagementControllerDriver, "get_firmware_inventory"),
+            (ManagementControllerDriver, "get_management_protocols"),
        ],
    )
    def test_absent_until_a_driver_implements_it(self, base, method):
@@ -133,9 +141,76 @@ class TestRoleIntrospection:
        assert primary_role_of(Nas) == "storage"
        assert primary_role_of(Host) == "linux"

+    def test_a_driver_that_sets_its_own_role_replaces_the_inherited_one(self):
+        """A Windows host is an OS driver too, but not a "linux" one."""
+
+        class Windows(OSDriver):
+            ROLE = "windows"
+
+        class Nas(StorageDriver, Windows):
+            pass
+
+        assert primary_role_of(Windows) == "windows"
+        assert [r.__name__ for r in roles_of(Windows)] == ["Windows"]
+        assert [r.__name__ for r in roles_of(Nas)] == ["StorageDriver", "Windows"]
+
    def test_driver_without_a_role_has_none(self):
        class Bare(DeviceTypeDriver):
            pass

        assert roles_of(Bare) == []
        assert primary_role_of(Bare) is None
+
+
+class TestManagementController:
+    """A BMC is reached over Redfish or IPMI, and it is not the server it manages."""
+
+    def test_role_key_and_transport(self):
+        assert ManagementControllerDriver.ROLE == "management_controller"
+        assert ManagementControllerDriver.USES_SSH is False
+
+    def test_power_actions_are_redfish_reset_types(self):
+        from napalm_device_types import POWER_ACTIONS
+
+        assert POWER_ACTIONS == (
+            "On",
+            "GracefulShutdown",
+            "ForceOff",
+            "GracefulRestart",
+            "ForceRestart",
+            "PowerCycle",
+        )
+
+    def test_carries_neither_snmp_health_metrics_nor_a_command_channel(self):
+        """Health metrics are SNMP; the BMC's sensors come through get_environment."""
+        from napalm_device_types import HealthMetricsMixin
+
+        assert not issubclass(ManagementControllerDriver, HealthMetricsMixin)
+        assert not hasattr(ManagementControllerDriver, "get_health_metrics")
+        assert not hasattr(ManagementControllerDriver, "run_command")
+        assert not hasattr(ManagementControllerDriver, "open_stream")
+
+    def test_docstring_says_reboot_host_restarts_the_bmc_not_the_server(self):
+        doc = ManagementControllerDriver.__doc__ or ""
+        assert "reboot_host" in doc and "Manager.Reset" in doc
+        assert "set_power_state" in doc
+        assert "get_environment" in doc
+
+    def test_its_return_types_are_exported(self):
+        import napalm_device_types as package
+
+        for name in (
+            "PowerAction",
+            "PowerStateDict",
+            "HardwareInventoryDict",
+            "EventLogEntryDict",
+            "FirmwareDict",
+            "ManagementProtocolsDict",
+        ):
+            assert name in package.__all__ and hasattr(package, name), name
+
+    def test_a_server_driver_can_rank_it_first(self):
+        class Idrac(ManagementControllerDriver):
+            pass
+
+        assert primary_role_of(Idrac) == "management_controller"
@@ -0,0 +1,53 @@
+"""SNMP health: drivers declare data, the application polls (docs/STRUCTURE.md).
+
+``SnmpHealthProfile`` is the shape a driver states its device's OIDs in.
+``HealthMetricsMixin``, which collected in the base package, is deprecated.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import ClassVar, get_type_hints
+
+import pytest
+
+from napalm_device_types import (
+    HealthMetricsMixin,
+    SnmpHealthProfile,
+    SnmpInterfacesSpec,
+    SnmpMemorySpec,
+    SnmpValueSpec,
+)
+
+
+def test_a_profile_says_where_each_value_lives():
+    assert set(get_type_hints(SnmpHealthProfile)) == {"uptime", "cpu", "memory", "interfaces"}
+    assert set(get_type_hints(SnmpValueSpec)) == {"oids", "table", "fold", "pattern"}
+    assert set(get_type_hints(SnmpMemorySpec)) == {"total", "used", "free", "percent", "unit"}
+    assert set(get_type_hints(SnmpInterfacesSpec)) == {"skip", "tx_errors_are_drops"}
+
+
+def test_every_key_is_optional():
+    for spec in (SnmpHealthProfile, SnmpValueSpec, SnmpMemorySpec, SnmpInterfacesSpec):
+        assert spec.__required_keys__ == frozenset(), spec.__name__
+
+
+def test_a_driver_declares_its_profile_as_a_class_attribute():
+    class Driver:
+        SNMP_HEALTH_PROFILE: ClassVar[SnmpHealthProfile] = {
+            "cpu": {"oids": ["1.3.6.1.4.1.9.1", "1.3.6.1.4.1.9.2"]},
+            "memory": {"percent": {"table": "1.3.6.1.4.1.9.3", "fold": "first"}},
+        }
+
+    assert Driver.SNMP_HEALTH_PROFILE["cpu"]["oids"][0] == "1.3.6.1.4.1.9.1"
+
+
+def test_the_collector_in_the_base_package_warns_that_it_is_deprecated():
+    async def get(oid):
+        return None
+
+    async def walk(oid):
+        return {}
+
+    with pytest.warns(DeprecationWarning, match="SNMP_HEALTH_PROFILE"):
+        asyncio.run(HealthMetricsMixin.get_health_metrics(get, walk))
@@ -19,7 +19,9 @@ from napalm_device_types.ssh_server import (
    MAIN_CONFIG,
    OpenSshServerMixin,
    has_dropin_include,
+    host_key_algorithms,
    parse_sshd_effective,
+    parse_sshd_hostkeys,
    plan_ssh_hardening,
    render_dropin,
    render_main_config,
@@ -32,8 +34,26 @@ hostkeyalgorithms ssh-ed25519,ecdsa-sha2-nistp256,rsa-sha2-512,rsa-sha2-256
 ciphers chacha20-poly1305@openssh.com,aes128-ctr,aes256-gcm@openssh.com
 macs umac-64-etm@openssh.com,hmac-sha2-256-etm@openssh.com,hmac-sha1,hmac-sha2-256
 passwordauthentication no
+hostkey /etc/ssh/ssh_host_ecdsa_key
+hostkey /etc/ssh/ssh_host_ed25519_key
+hostkey /etc/ssh/ssh_host_rsa_key
 """

+#: What ``ssh-keygen -l`` prints per host key type.
+KEYGEN = {
+    "/etc/ssh/ssh_host_ecdsa_key": "256 SHA256:aaaa root@host (ECDSA)",
+    "/etc/ssh/ssh_host_ed25519_key": "256 SHA256:bbbb root@host (ED25519)",
+    "/etc/ssh/ssh_host_rsa_key": "3072 SHA256:cccc root@host (RSA)",
+}
+
+#: The lock-out case: sshd offers ECDSA, sk- and certificate algorithms, but the
+#: host has only an ed25519 and an RSA key.
+SSHD_T_FEW_KEYS = SSHD_T.replace(
+    "hostkeyalgorithms ssh-ed25519,ecdsa-sha2-nistp256,rsa-sha2-512,rsa-sha2-256",
+    "hostkeyalgorithms ssh-ed25519-cert-v01@openssh.com,ecdsa-sha2-nistp256-cert-v01@openssh.com,"
+    "ssh-ed25519,ecdsa-sha2-nistp256,sk-ssh-ed25519@openssh.com,rsa-sha2-512,rsa-sha2-256",
+).replace("hostkey /etc/ssh/ssh_host_ecdsa_key\n", "")
+
 PARAMIKO = {
    "kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "diffie-hellman-group14-sha256"],
    "hostkey": ["ssh-ed25519", "ecdsa-sha2-nistp256", "rsa-sha2-512"],
@@ -112,6 +132,58 @@ def test_plan_without_a_client_list_only_guards_the_last_algorithm():
    assert plan.keep["kex"] == ["sntrup761x25519-sha512@openssh.com", "curve25519-sha256"]


+def test_plan_counts_only_host_key_algorithms_the_host_has_a_key_for():
+    """Without ed25519 and RSA, only ECDSA, sk- and cert algorithms stay: offered,
+    but the host has no key for them, so nobody would get in."""
+    eff = parse_sshd_effective(SSHD_T_FEW_KEYS)
+    plan = plan_ssh_hardening(
+        eff,
+        {"hostkey": ["ssh-ed25519", "rsa-sha2-256", "rsa-sha2-512"]},
+        host_keys={"ssh-ed25519", "rsa-sha2-512", "rsa-sha2-256", "ssh-rsa"},
+    )
+    assert "ssh-ed25519" in plan.keep["hostkey"]
+    assert [(r.algorithm, r.reason) for r in plan.refused] == [("ssh-ed25519", "last")]
+    assert plan.remove["hostkey"] == ["rsa-sha2-512", "rsa-sha2-256"]
+
+
+def test_plan_client_guard_counts_only_usable_host_keys():
+    eff = parse_sshd_effective(SSHD_T_FEW_KEYS)
+    plan = plan_ssh_hardening(
+        eff,
+        {"hostkey": ["ssh-ed25519", "rsa-sha2-256", "rsa-sha2-512"]},
+        client=PARAMIKO,
+        host_keys={"ssh-ed25519", "rsa-sha2-512", "rsa-sha2-256", "ssh-rsa"},
+    )
+    # paramiko speaks ecdsa-sha2-nistp256, but the host has no such key.
+    assert [(r.algorithm, r.reason) for r in plan.refused] == [("ssh-ed25519", "client")]
+
+
+def test_plan_removes_a_host_key_algorithm_when_a_usable_one_stays():
+    plan = plan_ssh_hardening(
+        parse_sshd_effective(SSHD_T_FEW_KEYS),
+        {"hostkey": ["rsa-sha2-256"]},
+        host_keys={"ssh-ed25519", "rsa-sha2-512", "rsa-sha2-256", "ssh-rsa"},
+    )
+    assert plan.remove == {"hostkey": ["rsa-sha2-256"]}
+    assert plan.refused == []
+
+
+def test_plan_leaves_host_keys_alone_when_the_keys_are_unknown():
+    plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"hostkey": ["ecdsa-sha2-nistp256"]})
+    assert plan.changed_classes == []
+    assert [(r.algorithm, r.reason) for r in plan.refused] == [("ecdsa-sha2-nistp256", "host_keys_unknown")]
+
+
+def test_host_key_algorithms_from_ssh_keygen():
+    assert host_key_algorithms("\n".join(KEYGEN.values())) == {
+        "ecdsa-sha2-nistp256", "ssh-ed25519", "rsa-sha2-512", "rsa-sha2-256", "ssh-rsa",
+    }
+    assert host_key_algorithms("521 SHA256:x no comment (ECDSA)") == {"ecdsa-sha2-nistp521"}
+    assert host_key_algorithms("") is None
+    assert host_key_algorithms("garbage") is None
+    assert parse_sshd_hostkeys(SSHD_T)[0] == "/etc/ssh/ssh_host_ecdsa_key"
+
+
 def test_plan_rejects_an_unknown_class():
    with pytest.raises(ValueError):
        plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"compression": ["zlib"]})
@@ -161,9 +233,14 @@ _DIRECTIVE = {"kexalgorithms": "kex", "hostkeyalgorithms": "hostkey", "ciphers":
 class FakeHost(OpenSshServerMixin):
    """A host with files, an sshd that reads them first-match-wins, and a log of what ran."""

-    def __init__(self, main: str = "Include /etc/ssh/sshd_config.d/*.conf\nPort 22\n") -> None:
+    def __init__(self, main: str = "Include /etc/ssh/sshd_config.d/*.conf\nPort 22\n",
+                 sshd_t: str = SSHD_T) -> None:
        self.files: Dict[str, str] = {MAIN_CONFIG: main}
-        self.defaults = parse_sshd_effective(SSHD_T)
+        self.modes: Dict[str, str] = {MAIN_CONFIG: "644"}
+        self.sshd_t = sshd_t
+        self.defaults = parse_sshd_effective(sshd_t)
+        self.keys = {p: KEYGEN[p] for p in parse_sshd_hostkeys(sshd_t)}
+        self.keygen_ok = True
        self.ran: List[str] = []
        self.syntax_ok = True
        self.reloads = 0
@@ -188,13 +265,40 @@ class FakeHost(OpenSshServerMixin):
                    eff[kind] = parts[1].split(",")
        return eff

+    #: What a real ``sudo -S`` leaves on stdin when it does not prompt (NOPASSWD,
+    #: or a cached timestamp): the password the driver put in front of the data.
+    SUDO_PASSWORD = "hunter2"
+
    def run_command(self, command: str, *, privileged: bool = False, timeout: float = 60,
                    stdin: Optional[bytes] = None) -> CommandResult:
        self.ran.append(command)
+        if privileged and stdin is not None:
+            stdin = f"{self.SUDO_PASSWORD}\n".encode() + stdin
+        m = re.search(r"mktemp", command)
+        if m:
+            assert not privileged, "the content must reach a file without sudo in the way"
+            path = f"/tmp/netork-ssh.{len(self.files)}"
+            self.files[path] = (stdin or b"").decode()
+            return CommandResult(path, "", 0)
        assert privileged, command
+        m = re.search(r'install -m "\$m" -- \'?([^\' ]+)\'? \'?([^\' ]+)\'?', command)
+        if m:
+            src, tmp = m.group(1), m.group(2)
+            target = re.search(r"mv -f -- '?[^' ]+'? '?([^' ;]+)'?", command).group(1)
+            ref = re.search(r"r='?([^' ;]+)'?; m=\$\(stat -c %a", command).group(1)
+            self.modes[target] = self.modes.get(ref, "644") if ref in self.files else "644"
+            self.files[target] = self.files.pop(src)
+            assert tmp.endswith(".netork-tmp")
+            return CommandResult("", "", 0)
+        if "ssh-keygen -l" in command:
+            if not self.keygen_ok:
+                return CommandResult("", "Permission denied", 1)
+            paths = re.search(r"for k in (.*?); do", command).group(1).replace("'", "").split()
+            return CommandResult("".join(self.keys[p] + "\n" for p in paths if p in self.keys), "", 0)
        if command.startswith("sshd -T") or "sshd -T" in command:
            eff = self._effective()
            out = "".join(f"{d} {','.join(eff[k])}\n" for d, k in _DIRECTIVE.items())
+            out += "".join(line + "\n" for line in self.sshd_t.splitlines() if line.startswith("hostkey "))
            return CommandResult(out, "", 0)
        if "sshd -t" in command:
            return CommandResult("", "" if self.syntax_ok else "bad line 3", 0 if self.syntax_ok else 255)
@@ -204,13 +308,10 @@ class FakeHost(OpenSshServerMixin):
            if path in self.files:
                return CommandResult(self.files[path], "", 0)
            return CommandResult("", "No such file", 1)
-        m = re.search(r"mv -f -- \"\$tmp\" '?([^' ]+)'?", command)
-        if m:
-            self.files[m.group(1)] = (stdin or b"").decode()
-            return CommandResult("", "", 0)
        m = re.match(r"rm -f -- '?([^' ]+)'?$", command)
        if m:
            self.files.pop(m.group(1), None)
+            self.modes.pop(m.group(1), None)
            return CommandResult("", "", 0)
        if "reload" in command or "HUP" in command:
            self.reloads += 1
@@ -301,3 +402,79 @@ def test_main_config_second_fix_keeps_what_the_first_set():
    eff = host._effective()
    assert "ecdsa-sha2-nistp256" not in eff["hostkey"]
    assert "hmac-sha1" not in eff["mac"]
+
+
+def test_the_sudo_password_never_reaches_a_file():
+    """napalm-linux puts the sudo password in front of stdin for ``sudo -S``. A
+    sudo that does not prompt leaves it there, and a ``cat`` under sudo wrote it
+    into the drop-in (NetOrk/netork#1027, first run on a real host). The content
+    now reaches a file of the login user first; root only installs that file."""
+    host = FakeHost()
+    host.harden_ssh_server({"mac": ["hmac-sha1"]})
+    assert host.SUDO_PASSWORD not in host.files[DROPIN_PATH]
+    assert host.files[DROPIN_PATH].startswith("# Written by netOrk")
+    assert not [p for p in host.files if p.startswith("/tmp/")]
+
+
+def test_restoring_an_earlier_dropin_leaks_nothing_either():
+    host = FakeHost()
+    host.files[DROPIN_PATH] = "# netOrk\nHostKeyAlgorithms ssh-ed25519\n"
+    host.harden_ssh_server({"mac": ["hmac-sha1"]}, verify=lambda: False)
+    assert host.files[DROPIN_PATH] == "# netOrk\nHostKeyAlgorithms ssh-ed25519\n"
+
+
+def test_mixin_reports_a_failed_way_back_instead_of_raising():
+    class ReloadFails(FakeHost):
+        def run_command(self, command, **kwargs):
+            if "reload" in command or "HUP" in command:
+                return CommandResult("", "Failed to reload ssh.service", 1)
+            return super().run_command(command, **kwargs)
+
+    host = ReloadFails()
+    result = host.harden_ssh_server({"mac": ["hmac-sha1"]})
+
+    assert not result.applied and not result.rolled_back
+    assert "putting the previous configuration back failed" in result.message
+
+
+def test_mixin_refuses_to_lock_out_a_host_with_only_ed25519_and_rsa_keys():
+    host = FakeHost(sshd_t=SSHD_T_FEW_KEYS)
+    lines: List[str] = []
+    result = host.harden_ssh_server(
+        {"hostkey": ["ssh-ed25519", "rsa-sha2-256", "rsa-sha2-512"]}, log=lines.append
+    )
+    assert result.applied
+    assert "ssh-ed25519" in host._effective()["hostkey"]
+    assert any("Keeping ssh-ed25519" in line for line in lines)
+
+
+def test_mixin_removes_a_host_key_algorithm_when_a_usable_one_stays():
+    host = FakeHost(sshd_t=SSHD_T_FEW_KEYS)
+    result = host.harden_ssh_server({"hostkey": ["rsa-sha2-256", "rsa-sha2-512"]})
+    assert result.applied and result.plan.refused == []
+    eff = host._effective()["hostkey"]
+    assert "rsa-sha2-256" not in eff and "ssh-ed25519" in eff
+
+
+def test_mixin_leaves_host_keys_alone_when_it_cannot_read_them():
+    host = FakeHost()
+    host.keygen_ok = False
+    result = host.harden_ssh_server({"hostkey": ["ecdsa-sha2-nistp256"]})
+    assert not result.applied and host.reloads == 0
+    assert DROPIN_PATH not in host.files
+    assert [r.reason for r in result.plan.refused] == ["host_keys_unknown"]
+
+
+def test_a_0600_sshd_config_stays_0600_and_so_does_its_copy():
+    host = FakeHost(main="Port 22\nMACs hmac-sha1,hmac-sha2-256\n")
+    host.modes[MAIN_CONFIG] = "600"
+    result = host.harden_ssh_server({"mac": ["hmac-sha1"]})
+    assert result.applied
+    assert host.modes[MAIN_CONFIG] == "600"
+    assert host.modes[result.backup] == "600"
+
+
+def test_a_new_dropin_gets_644():
+    host = FakeHost()
+    host.harden_ssh_server({"mac": ["hmac-sha1"]})
+    assert host.modes[DROPIN_PATH] == "644"
@@ -0,0 +1,102 @@
+"""SystemIdentityMixin: the SMBIOS identity of a host, for matching it to its BMC."""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_device_types import (
+    SYSTEM_IDENTITY_COMMAND,
+    SystemIdentityMixin,
+    is_placeholder_identity,
+    parse_system_identity,
+)
+
+UUID = "31303043-534D-C40C-7A72-6F1300000000"
+
+
+def _out(uuid: str, product_serial: str, board_serial: str) -> str:
+    return f"IDBEGIN\n{uuid}\n{product_serial}\n{board_serial}\nIDEND\n"
+
+
+# ── parsing ───────────────────────────────────────────────────────────────────
+
+
+def test_reads_uuid_and_both_serials():
+    assert parse_system_identity(_out(UUID, "S123", "ZM155S026022")) == {
+        "system_uuid": "31303043-534d-c40c-7a72-6f1300000000",
+        "product_serial": "S123",
+        "board_serial": "ZM155S026022",
+    }
+
+
+def test_a_supermicro_placeholder_serial_is_dropped():
+    # pve-dual, a Supermicro X10DRL-i: the product serial is the board vendor's default.
+    identity = parse_system_identity(_out(UUID, "0123456789", "ZM155S026022"))
+    assert identity["product_serial"] == ""
+    assert identity["board_serial"] == "ZM155S026022"
+
+
+@pytest.mark.parametrize(
+    "value",
+    ["", "None", "N/A", "Not Specified", "To be filled by O.E.M.", "Default string",
+     "System Serial Number", "0123456789", "1234567890", "0", "00000000", "Not Applicable"],
+)
+def test_placeholders(value):
+    assert is_placeholder_identity(value)
+
+
+@pytest.mark.parametrize(
+    "uuid",
+    ["00000000-0000-0000-0000-000000000000", "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF",
+     # A UUID many cheap boards ship, the same on every one of them.
+     "03000200-0400-0500-0006-000700080009"],
+)
+def test_placeholder_uuids(uuid):
+    assert parse_system_identity(_out(uuid, "", ""))["system_uuid"] == ""
+
+
+def test_real_values_are_not_placeholders():
+    assert not is_placeholder_identity("ZM155S026022")
+    assert not is_placeholder_identity(UUID)
+
+
+def test_unreadable_files_read_as_empty():
+    # product_serial and product_uuid are root's alone; without root they come back empty.
+    assert parse_system_identity(_out("", "", "BS1")) == {
+        "system_uuid": "", "product_serial": "", "board_serial": "BS1"}
+
+
+def test_output_without_its_frame_is_not_a_reading():
+    with pytest.raises(ValueError):
+        parse_system_identity("sudo: a password is required\n")
+
+
+def test_terminal_noise_around_the_frame_is_ignored():
+    assert parse_system_identity("\x1b[?2004h" + _out(UUID, "S1", "B1"))["board_serial"] == "B1"
+
+
+# ── the mixin ─────────────────────────────────────────────────────────────────
+
+
+class _Host(SystemIdentityMixin):
+    def __init__(self, output: str):
+        self.output = output
+        self.calls: list = []
+
+    def _run_system_identity_command(self, command: str, *, privileged: bool) -> str:
+        self.calls.append((command, privileged))
+        return self.output
+
+
+def test_the_driver_only_carries_the_command_across():
+    host = _Host(_out(UUID, "S1", "B1"))
+
+    assert host.get_system_identity()["board_serial"] == "B1"
+    ((command, privileged),) = host.calls
+    assert command == SYSTEM_IDENTITY_COMMAND
+    assert privileged is True
+
+
+def test_the_command_reads_the_three_dmi_files():
+    for name in ("product_uuid", "product_serial", "board_serial"):
+        assert f"/sys/class/dmi/id/{name}" in SYSTEM_IDENTITY_COMMAND
@@ -5,6 +5,7 @@ from napalm_device_types import (
    AccessPointDriver,
    FirewallDriver,
    HypervisorDriver,
+    ManagementControllerDriver,
    OSDriver,
    ResidentialGatewayDriver,
    StorageDriver,
@@ -16,6 +17,7 @@ BASE_CLASSES = [
    (AccessPointDriver, "Access Point"),
    (FirewallDriver, "Firewall"),
    (HypervisorDriver, "Hypervisor"),
+    (ManagementControllerDriver, "Management Controller"),
    (OSDriver, "OS"),
    (ResidentialGatewayDriver, "Gateway"),
    (StorageDriver, "Storage"),
@@ -0,0 +1,53 @@
+"""SNMP interface counters: 64-bit where the agent has ifXTable, 32-bit where not."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Dict, Optional
+
+from napalm_device_types._ucd_metrics import (
+    OID_IF_DESCR,
+    OID_IF_HC_IN_OCT,
+    OID_IF_HC_OUT_OCT,
+    OID_IF_HIGH_SPEED,
+    OID_IF_IN_OCT,
+    OID_IF_OUT_OCT,
+    OID_IF_SPEED,
+    collect_ucd_metrics,
+)
+
+LEGACY = {
+    OID_IF_DESCR: {"1": "eth0"},
+    OID_IF_SPEED: {"1": "4294967295"},
+    OID_IF_IN_OCT: {"1": "100"},
+    OID_IF_OUT_OCT: {"1": "200"},
+}
+
+
+def _collect(tables: Dict[str, Dict[str, str]]) -> dict:
+    async def get(oid: str) -> Optional[str]:
+        return None
+
+    async def walk(oid: str) -> Dict[str, str]:
+        return tables.get(oid, {})
+
+    return asyncio.run(collect_ucd_metrics(get, walk))
+
+
+def test_the_64_bit_counters_and_high_speed_win():
+    tables = {
+        **LEGACY,
+        OID_IF_HC_IN_OCT: {"1": "8589934592"},
+        OID_IF_HC_OUT_OCT: {"1": "17179869184"},
+        OID_IF_HIGH_SPEED: {"1": "10000"},
+    }
+
+    eth0 = _collect(tables)["interfaces"]["eth0"]
+
+    assert (eth0["rx_bytes"], eth0["tx_bytes"], eth0["speed_mbps"]) == (8589934592, 17179869184, 10000)
+
+
+def test_an_agent_without_ifxtable_falls_back_to_the_32_bit_counters():
+    eth0 = _collect(LEGACY)["interfaces"]["eth0"]
+
+    assert (eth0["rx_bytes"], eth0["tx_bytes"], eth0["speed_mbps"]) == (100, 200, 4294)
@@ -365,6 +365,9 @@ class _Host:
        self.bin.mkdir()
        self.state = root / "systemd"
        self.state.mkdir()
+        #: Stands in for /run/systemd/system, there while systemd is PID 1.
+        self.booted = root / "run-systemd-system"
+        self.booted.mkdir()
        (self.state / "units").write_text("")
        stub = self.bin / "systemctl"
        stub.write_text(
@@ -392,11 +395,15 @@ class _Host:
        path.write_text(content)
        return path

-    def run(self, *, systemctl: bool = True) -> str:
+    def run(self, *, systemctl: bool = True, booted: bool = True) -> str:
        if not systemctl:
            (self.bin / "systemctl").unlink()
+        if not booted:
+            self.booted.rmdir()
+        assert VIRTUAL_IPS_COMMAND.count("/run/systemd/system") == 1
+        command = VIRTUAL_IPS_COMMAND.replace("/run/systemd/system", str(self.booted))
        return subprocess.run(
-            ["/bin/sh", "-c", VIRTUAL_IPS_COMMAND],
+            ["/bin/sh", "-c", command],
            capture_output=True,
            text=True,
            env={"PATH": str(self.bin)},
@@ -552,6 +559,46 @@ class TestTheCommandOnAHost:
            "keepalived": None,
        }

+    def test_systemctl_without_systemd_running_is_not_looked_at(self, host):
+        """A container that carries systemctl but runs something else as PID 1:
+        systemctl lists no units there, which is not "none declared"."""
+        config = host.file(
+            "etc/keepalived/keepalived.conf", "vrrp_instance a { virtual_ipaddress { 192.0.2.4 } }"
+        )
+        host.unit("keepalived.service", "ActiveState=active", _exec_start("/k", "-f", str(config)))
+
+        assert parse_virtual_ips(host.run(booted=False)) == {
+            "vip_manager": None,
+            "keepalived": None,
+        }
+
+    def test_braces_on_their_own_line(self, host):
+        """keepalived takes ``vrrp_instance VI_2`` and its ``{`` on separate lines,
+        the same for every block; the secrets inside stay on the host."""
+        config = host.file(
+            "etc/keepalived/keepalived.conf",
+            "vrrp_instance VI_1 {\n  interface eth0\n"
+            "  virtual_ipaddress {\n    192.0.2.50\n  }\n}\n"
+            "vrrp_instance VI_2\n{\n  interface eth1\n  virtual_router_id 53\n"
+            f"  authentication\n  {{\n    auth_type PASS\n    auth_pass {SECRET}\n  }}\n"
+            "  virtual_ipaddress\n\n  {\n    192.0.2.60/24\n  }\n"
+            f"  notify_master /etc/keepalived/master.sh {SECRET}\n}}\n",
+        )
+        host.unit(
+            "keepalived.service",
+            "ActiveState=active",
+            _exec_start("/usr/sbin/keepalived", "-f", str(config)),
+        )
+
+        out = host.run()
+
+        assert SECRET not in out
+        assert [(e["instance"], e["address"], e["prefix_length"], e["interface"], e["group_key"])
+                for e in parse_virtual_ips(out)["keepalived"]] == [
+            ("VI_1", "192.0.2.50", 32, "eth0", None),
+            ("VI_2", "192.0.2.60", 24, "eth1", "vrid:53"),
+        ]
+

 class TestTheCommand:
    def test_the_frame_is_not_in_the_command_itself(self):
Author	SHA1	Message	Date
christianmanivong	d60b3d3742	feat: declare get_management_peer for choosing an SNMP agent's sources CI / test (3.10) (push) Successful in 26s Details CI / test (3.11) (push) Successful in 25s Details CI / test (3.12) (push) Successful in 26s Details CI / test (3.13) (push) Successful in 29s Details CI / test (3.10) (pull_request) Successful in 29s Details CI / test (3.11) (pull_request) Successful in 25s Details CI / test (3.12) (pull_request) Successful in 27s Details CI / test (3.13) (pull_request) Successful in 26s Details The caller has to name the sources an SNMP agent answers, and cannot reliably tell its own address as the device sees it (NAT, relays). SnmpAgentMixin now declares get_management_peer(): the address the device sees the management connection come from, or None. Optional: a driver that cannot tell does not implement it.	2026-10-16 15:18:45 +02:00
christianmanivong	95222377f0	feat: fix routines as templates with the caller's values; deprecate run_device_action (4.12.0) CI / test (3.10) (push) Successful in 29s Details CI / test (3.11) (push) Successful in 26s Details CI / test (3.12) (push) Successful in 30s Details CI / test (3.13) (push) Successful in 30s Details CI / test (3.10) (pull_request) Successful in 35s Details CI / test (3.11) (pull_request) Successful in 33s Details CI / test (3.12) (pull_request) Successful in 35s Details CI / test (3.13) (pull_request) Successful in 34s Details run_device_action was a string dispatch drivers filled with an application's remediations, values hard-coded inside (community "public", the caller's address, a /24 firewall rule). napalm_device_types.fixes declares them as templates instead. A template runs declared steps in order and stops at the first failure (run_fix_steps); the driver implements the steps, or overrides the whole method where its device takes the change in one call; every value comes from the caller. - SnmpAgentMixin.configure_snmp_agent(settings): install, configure, firewall, start. The caller verifies with its own SNMP client. - EnsurePackageMixin.ensure_package(name): install only what is missing. - PackageProxyMixin.configure_package_proxy(url): declared. - ContainerEngineMixin.grant_container_engine_access(user): declared. None is mixed into a role base, so a capability check stays truthful. Closes #33	2026-10-16 12:06:19 +02:00
christianmanivong	64cc20ba00	feat: SnmpHealthProfile, the OIDs a driver declares; deprecate HealthMetricsMixin (4.11.0) CI / test (3.10) (push) Successful in 35s Details CI / test (3.11) (push) Successful in 28s Details CI / test (3.12) (push) Successful in 29s Details CI / test (3.13) (push) Successful in 27s Details CI / test (3.10) (pull_request) Successful in 31s Details CI / test (3.11) (pull_request) Successful in 31s Details CI / test (3.12) (pull_request) Successful in 35s Details CI / test (3.13) (pull_request) Successful in 35s Details Drivers no longer collect over SNMP (docs/STRUCTURE.md). A driver whose device publishes its health values under OIDs of its own declares them as SNMP_HEALTH_PROFILE: per value either scalars asked in order or a table folded by first, average or max, optionally with a pattern for text answers; memory as total with used or free, or a percentage, in KiB or bytes; and how to read the interfaces. The application polls and computes. HealthMetricsMixin.get_health_metrics now warns that it is deprecated; it goes in 5.0. Closes #32	2026-10-16 12:00:06 +02:00
christianmanivong	2739f10ecd	Merge pull request 'fix: keep CARP state knowledge out of the base package (4.10.1)' (#39 ) from fix/carp-helper-to-driver into main CI / test (3.10) (push) Successful in 26s Details CI / test (3.11) (push) Successful in 24s Details CI / test (3.12) (push) Successful in 25s Details CI / test (3.13) (push) Successful in 25s Details	2026-10-16 09:19:00 +00:00
christianmanivong	0330f17a67	fix: keep CARP state knowledge out of the base package (4.10.1) CI / test (3.10) (push) Successful in 25s Details CI / test (3.11) (push) Successful in 24s Details CI / test (3.12) (push) Successful in 26s Details CI / test (3.13) (push) Successful in 25s Details CI / test (3.10) (pull_request) Successful in 26s Details CI / test (3.11) (pull_request) Successful in 24s Details CI / test (3.12) (pull_request) Successful in 26s Details CI / test (3.13) (pull_request) Successful in 26s Details carp_virtual_ip() and its state table are CARP protocol knowledge; under the agreed structure (PR #37, docs/STRUCTURE.md) the base package carries only contracts, data shapes and I/O-free reconcilers. They move to the driver that speaks CARP (napalm-opnsense). The contract stays: VirtualIpsDict.carp and VirtualIpDict.holds. Nothing used the helper yet; it was released in 4.10.0 an hour ago. netOrk #1121.	2026-10-16 11:11:16 +02:00
christianmanivong	d3034bc8ea	Merge pull request 'feat: CARP addresses as virtual IPs, with their holder in the reading (4.10.0)' (#38 ) from feat/carp-virtual-ips into main CI / test (3.10) (push) Successful in 25s Details CI / test (3.11) (push) Successful in 24s Details CI / test (3.12) (push) Successful in 26s Details CI / test (3.13) (push) Successful in 26s Details	2026-10-16 09:10:36 +00:00
christianmanivong	4f3b88db0e	feat: CARP addresses as virtual IPs, with their holder in the reading (4.10.0) CI / test (3.10) (push) Successful in 27s Details CI / test (3.11) (push) Successful in 24s Details CI / test (3.12) (push) Successful in 27s Details CI / test (3.13) (push) Successful in 25s Details CI / test (3.10) (pull_request) Successful in 26s Details CI / test (3.11) (pull_request) Successful in 24s Details CI / test (3.12) (pull_request) Successful in 26s Details CI / test (3.13) (pull_request) Successful in 26s Details A firewall pair's CARP addresses are a third virtual IP mechanism. Unlike vip-manager and keepalived, a CARP reading says who holds the address: a firewall's interface list need not show it, but CARP's state names the MASTER. - VirtualIpsDict gains carp (NotRequired), VirtualIpDict gains holds (NotRequired; None leaves the holder to the interfaces). - carp_virtual_ip() shapes one address bound to a VHID: MASTER holds, BACKUP stands by, INIT and DISABLED neither run nor hold, an unknown state says nothing. The same for OPNsense and pfSense; a driver supplies the fields. netOrk #1121.	2026-10-16 11:06:43 +02:00
christianmanivong	fde6d717a8	feat: SystemIdentityMixin reads a host's SMBIOS identity (4.9.0) CI / test (3.10) (push) Successful in 27s Details CI / test (3.11) (push) Successful in 25s Details CI / test (3.12) (push) Successful in 27s Details CI / test (3.13) (push) Successful in 27s Details What ties a server to its BMC: the BMC reports the server's serial number and system UUID, the host reads the same values from its DMI tables. Both drivers that read DMI (napalm-linux, napalm-proxmox) had a copy of the shell for it, and neither read the board serial. SystemIdentityMixin.get_system_identity() reads product_uuid, product_serial and board_serial in one framed command and returns them with vendor placeholders empty: a Supermicro board's product serial is 0123456789 on every one of them, others read "To be filled by O.E.M." or "Default string", and many cheap boards share the UUID 03000200-0400-0500-0006-000700080009. is_placeholder_identity() is the same filter for a BMC driver's side. A driver supplies only _run_system_identity_command(command, privileged=); the files are root's.	2026-10-13 12:20:10 +02:00
christianmanivong	a1da0337e2	feat: alternative fingerprint rules count once (4.8.0) CI / test (3.10) (push) Successful in 28s Details CI / test (3.11) (push) Successful in 25s Details CI / test (3.12) (push) Successful in 27s Details CI / test (3.13) (push) Successful in 26s Details Discovery compares drivers by score divided by the most a driver could score, and that maximum summed every positive rule. Rules that never match together, such as one vendor's landing page or another's, made a driver that knows several vendors lose to one that knows none: napalm-redfish's Supermicro landing page scored 6 of 41 and lost to a bare "linux" in the same BMC's SNMP sysDescr, 5 of 18. FingerprintRule gets an optional group. rules_max_weight() sums the positive rules and counts each group once, with its strongest rule. Existing rules have no group and count as before.	2026-10-13 10:19:54 +02:00
christianmanivong	7afa590c0c	fix: ship py.typed so type checkers read the package's types CI / test (3.10) (push) Successful in 27s Details CI / test (3.11) (push) Successful in 26s Details CI / test (3.12) (push) Successful in 26s Details CI / test (3.13) (push) Successful in 1m16s Details The package declares "Typing :: Typed" but shipped no PEP 561 marker, so mypy in driver repositories treated it as untyped. Add the marker, include it as package data, and release 4.7.1.	2026-10-13 07:20:02 +02:00
christianmanivong	155b5bbbe4	feat!: settle the open review questions; release 4.7.0 ## feat!: settle the open review questions on roles, diffs, SSH and SNMP - roles: a driver that sets ROLE itself replaces the role it inherits, so a Windows driver on OSDriver is "windows", not also "linux". - os: OSDriver's account listing is get_os_users; NAPALM's get_users has another shape and a runtime stub, so hasattr could not tell. - README: NAPALM's own getters must come from the driver or a mixin listed before the first role base; its stubs win the MRO otherwise. - firewall/DHCP diffs: an empty key, a key given twice, or one more than one live entry holds raises ValueError instead of overwriting by guess; live entries without a key are never matched. - ssh_server: only host key algorithms the host has a key for count as staying; when the keys cannot be read, the host key list is not touched. sshd_config and its copy keep their file mode (GNU and BSD stat), 644 only for a new file. - SNMP: 64-bit ifXTable octets and ifHighSpeed, with the 32-bit counters as fallback for agents without ifXTable. - pyproject: declare typing-extensions, which models.py imports. BREAKING CHANGE: OSDriver.get_users is now get_os_users; diffs raise ValueError on keys that cannot name one entry; plan_ssh_hardening removes no host key algorithm without host_keys. ## fix: tell "could not read" and "does not install" apart in update and VIP readers - package_updates: the dnf security reader records dnf's exit status and raises when it is missing or non-zero, as the apt reader does; a failed dnf no longer reads as "no security advisories". - host_status: an enabled dnf-automatic.timer counts as auto updates only with apply_updates = yes (dnf5's automatic.conf wins over dnf4's); dnf-automatic-install.timer always does. Unattended-upgrades counts only while the program is installed, not from a conffile left behind. - virtual_ips: a host with systemctl but without systemd as PID 1 reads as "did not look" (None), not as "has none". keepalived blocks whose brace is on the next line are read. BREAKING CHANGE: parse_dnf_security expects DNF_SECURITY_COMMAND's __DNF_RC marker and raises ValueError without it.	2026-10-13 06:58:40 +02:00
christianmanivong	e509445572	feat: add ManagementControllerDriver for server BMCs A role base for baseboard management controllers (Supermicro, Dell iDRAC, HPE iLO, Lenovo XCC), reached over Redfish or IPMI, not SSH. It declares, under TYPE_CHECKING only: get_power_state, set_power_state (Redfish ResetType actions), get_hardware_inventory (CPUs, DIMMs, drives, PSUs, NICs, and the server's serial, system UUID and NIC MACs), get_event_log, get_firmware_inventory and get_management_protocols. reboot_host restarts the BMC itself (Manager.Reset), never the managed server, which goes through set_power_state. No HealthMetricsMixin and no command channel; sensors come through NAPALM's get_environment.	2026-10-13 06:47:19 +02:00
christianmanivong	dd8977750d	chore: Python 3.13, mypy --strict in CI, eight bug fixes, docs Squashes three changes: ## chore: support Python 3.13 and make mypy --strict pass in CI Add 3.13 to the CI matrix and classifiers, run mypy in CI, fix the 14 strict errors it reported, and correct the README's minimum Python. ## fix: eight bugs found in a full review, each with a regression test - channel: run_on_transport lost output sshd sent after the exit status; it now reads stdout and stderr to EOF before returning. - ssh_server: a failed way back (restore or reload) raised out of harden_ssh_server after the host was touched; it is now reported in the result, with rolled_back False. - host_status: a newer RPM kernel (RHEL, Fedora) was never seen, because its release, not its arch, followed the last "-". - package_updates: an apt line wrapped at a blank lost the blank, and with it the package. - kernel: an empty payload (gzip or base64 missing) read as a report. - listening: "[fe80::1]%eth0:546", as ss prints it, kept its brackets. - ping_sweep: a reply whose "success" was not a dict aborted the sweep. - lag: a member of unknown speed (-1) lowered the LAG's speed. ## docs: bring the README and docstrings in line with the code The firewall worked example showed the runtime stubs the README forbids; the class table missed PhoneDriver and MediaDriver; ConfigLifecycleMixin is mixed in by drivers, not by a role base; services, mac_acl and packages named a class, a key and an exception that do not exist.	2026-10-13 01:22:22 +02:00
christianmanivong	e58b65a3db	Merge pull request 'fix: never let root read the SSH configuration from stdin' (#30 ) from fix/ssh-write-without-sudo-stdin into main CI / test (3.10) (push) Successful in 25s Details CI / test (3.11) (push) Successful in 25s Details CI / test (3.12) (push) Successful in 25s Details	2026-10-11 21:50:08 +00:00
christianmanivong	7ec427c743	fix: never let root read the SSH configuration from stdin CI / test (3.10) (push) Successful in 23s Details CI / test (3.11) (push) Successful in 22s Details CI / test (3.12) (push) Successful in 24s Details CI / test (3.10) (pull_request) Successful in 23s Details CI / test (3.11) (pull_request) Successful in 25s Details CI / test (3.12) (pull_request) Successful in 24s Details A driver gains root with sudo -S and puts the password in front of stdin. A sudo that does not prompt -- NOPASSWD, a cached timestamp -- leaves it there, and the cat under sudo wrote it into the drop-in as its first line. Seen on the first real run (NetOrk/netork#1027): sshd -t rejected the file and the rollback removed it, but the password was in a 0644 file for a moment and in the error the caller showed. The content now goes without privilege to a mktemp file only the login user can read; root installs that file (mode 644, root-owned) and removes it. The fake host in the tests prepends the password to every privileged stdin, as a non-prompting sudo does. Bumps to 4.5.1. NetOrk/netork#1027, NetOrk/netork#1029	2026-10-11 23:48:51 +02:00
christianmanivong	d43e15e8db	Merge pull request 'feat: OpenSshServerMixin takes flagged algorithms out of what sshd offers' (#29 ) from feat/ssh-algorithm-fix into main CI / test (3.10) (push) Successful in 23s Details CI / test (3.11) (push) Successful in 22s Details CI / test (3.12) (push) Successful in 23s Details	2026-10-11 21:14:06 +00:00