docs: what the drivers need from the SNMP profile, opt-outs, and how getters behave

The SNMP profile has to express fallback OIDs, tables folded by first, average or maximum, numbers inside text values, memory as used, free or percent, and the uptime source. TX errors counting drops becomes OPNsense's entry instead of every firewall's. A driver opts out of an inherited template with None, so callers ask callable(). USES_SSH, FULL_UPGRADE and platform stay as driver facts. Getters do not write, a failed read raises, a write reports its real result, and the application uses public driver methods only.
docs: a structure that keeps device-types vendor- and product-agnostic
2026-10-14 10:56:17 +02:00 · 2026-10-14 10:20:55 +02:00 · 2026-10-13 12:20:10 +02:00 · 2026-10-13 10:19:54 +02:00 · 2026-10-13 07:20:02 +02:00 · 2026-10-13 06:58:40 +02:00
47 changed files with 1822 additions and 128 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
@@ -110,6 +128,15 @@ 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.

+`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 +210,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 +270,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 +283,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 +416,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):
@@ -0,0 +1,241 @@
+# Structure
+
+Status: agreed plan. Nothing here is implemented yet; the migration at the end says in
+which order it will be.
+
+## Principle
+
+**napalm-device-types is vendor-, OS-, protocol- and product-agnostic.** It holds the
+contracts every driver implements and the generic logic that works unchanged for every
+driver of a role. It knows no shell, no paramiko, no package manager, no container CLI,
+no SNMP OID, no cloud-init and no application that uses the drivers.
+
+That rules out three kinds of code that are here today:
+
+- **How a kind of host is read.** Shell commands and their parsers for POSIX hosts move to
+  a new package, **napalm-unix**, which napalm-linux and napalm-bsd depend on.
+- **What an application does with a device.** Polling a device over SNMP, sweeping a range
+  with pings, scoring discovery evidence, choosing what a provisioned guest installs, and
+  deciding what counts as a problem belong to the application, not to the drivers.
+- **Product naming.** Files, markers and messages a driver leaves on a device carry the
+  driver's name or none, never the name of the application that called it.
+
+```
+napalm-device-types        roles, capabilities, templates, models, reconcilers  <- every driver
+        ^
+napalm-unix                POSIX host readers, SSH exec transport, privilege    <- napalm-linux, napalm-bsd
+        ^
+napalm-linux, napalm-bsd   connection, platform getters, platform steps
+
+application                SNMP polling, ping sweeps, discovery scoring, guest
+                           profiles, warnings, and the values every fix is given
+```
+
+## What stays in napalm-device-types
+
+| Kind | Rule | Modules |
+|---|---|---|
+| **Role** | Declares only, under `TYPE_CHECKING`. No runtime method. | `access_point`, `switch`, `firewall`, `hypervisor`, `os`, `storage`, `phone`, `media`, `residential_gateway`, `management_controller` |
+| **Capability** | Declares only. | `packages`, `services`, `updates`, `nat_vpn`, `mac_acl`, `config_lifecycle`, `interface_filter`, `host_reboot` |
+| **Template** | A concrete method that runs a fixed sequence of abstract steps. A driver implements the steps, or overrides the whole method when its device does it in one go. All values come from the caller. | the fix routines below |
+| **Contract types** | Interfaces and data shapes other packages implement or fill. | `CommandChannelMixin`, `CommandResult`, `ByteStream`; the declaration of `ContainerEngineMixin`; the reader mixins' public method declarations; the SNMP profile type; `models` |
+| **Reconciler** | Concrete and vendor-neutral, without I/O of its own. | `firewall_rules` and `dhcp` diff/apply, `_matching`, `lag` |
+| **Introspection** | Concrete, without I/O. | `roles`; `base` (`FingerprintRule`, `PortSpec`, the fingerprint attributes) |
+| **Driver facts** | Class attributes that describe the device or how it is reached, not what an application does with it. | `USES_SSH`, `FULL_UPGRADE`, `platform`, `SNMP_HEALTH_PROFILE` |
+
+### Fix routines are templates
+
+A change the application wants made on a device, for example "run an SNMP agent that
+answers us", is one template per purpose. The template fixes the order of the steps; the
+driver knows how each step is done on its device; the caller supplies every value.
+
+```python
+class SnmpAgentMixin:
+    def configure_snmp_agent(self, settings: SnmpAgentSettingsDict) -> FixResultDict:
+        steps = [self._install_snmp_agent, self._write_snmp_agent_config,
+                 self._enable_snmp_agent, self._verify_snmp_agent]
+        ...                                   # runs them in order, stops at the first failure
+
+    # declared only; napalm-unix implements them for net-snmp
+    def _install_snmp_agent(self) -> FixStepDict: ...
+    def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepDict: ...
+    def _enable_snmp_agent(self) -> FixStepDict: ...
+    def _verify_snmp_agent(self, settings: SnmpAgentSettingsDict) -> FixStepDict: ...
+```
+
+napalm-unix implements the four steps for net-snmp, for Linux and the BSDs alike. A REST
+driver such as OPNsense overrides `configure_snmp_agent` as a whole, because its device
+takes the whole configuration in one call. The community, the allowed sources, contact
+and location are the caller's.
+
+The same shape replaces every action in today's `run_device_action`:
+
+| Today | Template |
+|---|---|
+| `fix_snmp` | `configure_snmp_agent(settings)` |
+| `fix_apt_proxy` | `configure_package_proxy(url)` |
+| `fix_docker_permissions` | `grant_container_engine_access(user)` |
+| `install_lldpd`, `install_auc` | `ensure_package(name)` |
+
+The mixin is mixed in by the driver, or by the platform package, not by a role base. A
+driver that inherits a template it cannot run on its device, as a NAS driver built on the
+Linux driver inherits the package proxy, opts out by setting the method to `None`. So a
+caller asks `callable(getattr(driver, "configure_snmp_agent", None))`, never `hasattr`.
+
+The caller verifies the result from its own side where it can; the SNMP agent is asked
+by the application's SNMP client, not by a probe the driver hand-builds.
+
+### SNMP: drivers declare data, the application polls
+
+The drivers do not collect over SNMP. A driver whose device answers the standard UCD-MIB
+and IF-MIB declares nothing. A driver whose device publishes the same values under
+proprietary OIDs declares them as data, in a class attribute of a type defined here:
+
+```python
+class ProcurveDriver(SwitchDriver):
+    SNMP_HEALTH_PROFILE: ClassVar[SnmpHealthProfile] = {
+        "uptime": {"oids": ["1.3.6.1.4.1.11.2.14.11.1.2.4.1.5.1"], "format": "ticks"},
+        "cpu": {"oids": ["1.3.6.1.4.1.11.2.14.11.5.1.9.6.1.0",
+                         "1.3.6.1.4.1.11.2.3.7.8.6.1.0"]},
+        "memory": {"total": "...", "free": "...", "unit": "bytes"},
+    }
+```
+
+One OID per value is not enough for the drivers there are. The profile type has to say:
+
+- **Fallbacks:** an ordered list of OIDs; the first that answers wins (ProCurve's CPU).
+- **Tables:** an OID to walk and how to fold its rows: first, average or maximum (TP-Link
+  reports CPU and memory per stack unit).
+- **Text values:** a pattern that takes the number out of a string (Netgear reports CPU
+  as `"5 Secs ( 3.2%)"`).
+- **Memory:** total with used or free, or a percentage only, and the unit.
+- **Uptime:** which OID, standard ones included (`sysUpTime` where a device has no
+  `hrSystemUptime`), and its format.
+- **Interfaces:** a pattern of names to skip, and whether the TX-error counter counts
+  drops. Today the second is `True` on every firewall, because OPNsense counts drops
+  there; it becomes OPNsense's profile entry.
+
+Polling, the standard OIDs, the counter arithmetic (64-bit counters first, 32-bit as the
+fallback) and the defaults are the application's.
+
+### Warnings are facts
+
+A driver reports facts: the proxy a host is configured with, a DSL line's SNR, the
+firmware version it runs and the one it is offered. It does not compare them with a
+setting of the application or with a threshold, and it does not name a warning.
+
+The same goes for lookups a device does not make itself: what firmware a vendor
+currently publishes is the application's catalogue, not a getter.
+
+### Getters read, and say when they cannot
+
+- A getter changes nothing on the device. It does not start a daemon, rewrite its
+  configuration or install a package on the way to an answer.
+- A read that fails raises. An empty list means "there is none", never "could not ask";
+  a sync that deletes and re-adds on top of an empty answer would otherwise duplicate
+  everything.
+- A write reports failure from the device's answer or exit status, never `success: True`
+  by default.
+
+### The application uses public methods only
+
+The application never calls a driver's private methods (`_get`, `_post`,
+`_send_command`, `_action_*`). What it needs from a device becomes a declared method of
+the driver.
+
+## What moves to napalm-unix
+
+| Module | Knows |
+|---|---|
+| `host_status` | apt, dnf, dpkg, systemd timers, `df` |
+| `kernel` | `/proc`, `/lib/modules`, gzip, base64 |
+| `listening` | `ss`, `netstat`, `/proc/<pid>/cgroup` |
+| `systemd` | `systemctl` |
+| `virtual_ips` | vip-manager, keepalived, awk |
+| `system_identity` | SMBIOS via sysfs and `dmidecode` |
+| `package_updates` | apt and dnf output |
+| `ssh_server` | OpenSSH (`sshd -T`, `sshd -t`, `sshd_config`); which algorithms to drop stays the caller's |
+| `terminal` | ANSI escape codes in shell output |
+| `channel`: `run_on_transport`, `open_stream_on_transport`, `ParamikoExecStream` | paramiko exec channels |
+| `container_engine`: `ContainerEngineConnection`, engine probing | the Docker CLI, `system dial-stdio` |
+
+It also takes what napalm-linux and napalm-bsd each carry today:
+
+- **Privilege:** one `sudo -S` / `sudo -n` / root layer, including the rule that a
+  privileged command never gets stdin while a sudo password is set, and one way to write a
+  file as root (stage with `mktemp`, then `install`). Two copies of this have already
+  drifted apart.
+- **Parsers:** `/etc/passwd` with `/etc/group`, crontab lines, `ps` with `lstart`.
+- **Fix steps:** net-snmp, the package proxy, container engine group membership.
+
+The FreeBSD part of `host_status` (`freebsd-version -k` against `-r`) moves to napalm-bsd.
+
+In napalm-unix every reader follows one pattern:
+
+- **Shape:** `FOO_COMMAND`, `parse_foo(output)`, and `FooMixin.get_foo()`.
+- **Reaching the host:** the mixin uses `run_command(command, privileged=...)` from
+  `CommandChannelMixin`. The six per-reader `_run_*` hooks, with four different signatures,
+  go away.
+- **Framing:** one frame helper writes and parses the `BEGIN`/`END` markers for every reader.
+- **Errors:** one exception for "cannot read" and one for "not on this host". A read that
+  fails or is cut short raises; it never reads as "nothing there".
+- **Names on the host:** temporary files, markers and drop-ins carry `napalm` or the
+  driver's name. A drop-in written under an older name is still recognised and replaced.
+
+Layout: `napalm_unix/{posix,linux,freebsd,openssh,transport}/`.
+
+## What moves to the application
+
+| Today | Why it is not a driver's |
+|---|---|
+| `health_metrics`, `_ucd_metrics` | Polling over SNMP is something the application does to a device; the driver is not even connected while it runs. |
+| `ping_sweep` | Looping over addresses, capping the targets and reporting progress is a scan. A device with a faster batch keeps it as a capability of its own. |
+| `rules_max_weight` | Scoring is how one application ranks drivers. The rules themselves stay with the driver. |
+| `provisioning` (`QEMU_GUEST_AGENTS`, cloud-init network-config, decompressors) | Which agent a guest gets and how its network is configured is a guest profile. The hypervisor driver is handed the packages and commands. |
+| `TYPE_LABEL`, `REBOOT_SETTLE_SECONDS` | A display name and a polling delay. |
+| firmware catalogues (napalm-netgear's `netgear_plus`) | What a vendor publishes is not something the device reports. |
+| `run_device_action` | Replaced by the templates above. |
+
+`OSDriver.ROLE` stops defaulting to `"linux"`. Each OS driver states its own role.
+
+## Rules, enforced by tests in this package
+
+1. Roles and capabilities own no runtime method; a template runs only declared steps.
+2. No module imports `paramiko`, `socket`, `subprocess` or `asyncio`, and none defines a
+   `*_COMMAND`.
+3. No module, string or default names an application.
+4. Role methods return TypedDicts, not `Any`.
+
+Each rule is added with the step that makes it pass.
+
+## Migration
+
+This package cannot keep re-exports pointing at napalm-unix: the contract package would
+then depend on its implementation. So a moved name is first deprecated here and then
+removed, and nothing breaks in between.
+
+| Step | Change |
+|---|---|
+| 1 | The application moves to 4.9 and the drivers that need it; napalm-bsd stops passing stdin to a privileged command while a sudo password is set. |
+| 2 | This document; the application's architecture docs and this README follow it. |
+| 3 | `SnmpHealthProfile`. The application polls on its own; the four switch and phone drivers declare their OIDs as data. `HealthMetricsMixin` is deprecated. |
+| 4 | The fix templates. napalm-linux and napalm-bsd implement the steps; the application calls the templates with its values. `run_device_action` is deprecated. The other drivers follow, one issue each. |
+| 5 | Ping sweep, scoring, provisioning, `TYPE_LABEL` and `REBOOT_SETTLE_SECONDS` move to the application and are deprecated here. Drivers report facts instead of warnings. |
+| 6 | napalm-unix, with the moved code and its tests, the unified hooks and the frame helper. napalm-linux and napalm-bsd import from it; the moved names are deprecated here. |
+| 7 | 5.0 removes everything deprecated. It also splits `models.py` by area, types the remaining `Dict[str, Any]` role returns, declares what drivers already offer undeclared (`get_networks`, `search_packages`, `first_boot_pending`, the base-system update entry), and makes `install_package` declare the result drivers return. |
+| 8 | The README shrinks to the concepts; the reader details move with the readers. |
+
+## Proposals this replaces
+
+Several open issues proposed adding shared driver code here. Under this structure:
+
+| Issue | Goes to |
+|---|---|
+| #8 netmiko session | not here: a CLI-driver helper package, or left in the drivers |
+| #9 config mode for CLI switches | not here: a CLI-driver helper package, or left in the drivers |
+| #10 LLDP rows | the projection of rows into NAPALM's two shapes stays here; the `lldpctl` parser goes to napalm-unix |
+| #11 CLI text helpers | not here; the natural port sort, which `lag` already has, may become public |
+| #12 ping | the reply builder stays here; the BusyBox and iputils parser goes to napalm-unix |
+| #13 Linux userland | napalm-unix; the snmpd part becomes the SNMP template's steps |
+| #14 `run_device_action` | replaced by the fix templates |
+| #22 hooks onto `run_command` | step 6 |
@@ -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,
@@ -92,7 +93,17 @@ 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,
+)
 from napalm_device_types.nat_vpn import NatVpnMixin
 from napalm_device_types.packages import PackageManagementMixin
 from napalm_device_types.phone import PhoneDriver
@@ -114,6 +125,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,
@@ -145,6 +162,7 @@ __all__ = [
    "DeviceTypeDriver",
    "DhcpServerMixin",
    "FingerprintRule",
+    "rules_max_weight",
    "FirewallDriver",
    "FirewallRuleMixin",
    "HealthMetricsMixin",
@@ -154,6 +172,14 @@ __all__ = [
    "HypervisorDriver",
    "InterfaceFilterMixin",
    "MacAclMixin",
+    "ManagementControllerDriver",
+    "POWER_ACTIONS",
+    "PowerAction",
+    "PowerStateDict",
+    "HardwareInventoryDict",
+    "EventLogEntryDict",
+    "FirmwareDict",
+    "ManagementProtocolsDict",
    "MediaDriver",
    "NatVpnMixin",
    "OSDriver",
@@ -185,6 +211,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,7 +29,7 @@ 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
@@ -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:
@@ -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
@@ -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


@@ -1035,3 +1035,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.

@@ -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,14 +348,17 @@ 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:
+    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 (mode 644, owned by root) and removes it.
+        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"',
@@ -304,7 +371,10 @@ class OpenSshServerMixin:
            raise RuntimeError(f"could not stage {path}: {(staged.stderr or staged.stdout).strip()}")
        part = path + ".netork-tmp"
        result = self._ssh_run(
-            f"install -m 644 -- {quote(tmp)} {quote(part)} && mv -f -- {quote(part)} {quote(path)};"
+            # 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"
        )
        if result.exit_code != 0:
@@ -324,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."""
@@ -357,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)
@@ -396,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})")
@@ -418,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.1"
+version = "4.9.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 == []
@@ -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"
@@ -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
@@ -204,16 +281,24 @@ class FakeHost(OpenSshServerMixin):
            self.files[path] = (stdin or b"").decode()
            return CommandResult(path, "", 0)
        assert privileged, command
-        m = re.search(r"install -m 644 -- '?([^' ]+)'? '?([^' ]+)'?", 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)
@@ -226,6 +311,7 @@ class FakeHost(OpenSshServerMixin):
        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
@@ -335,3 +421,60 @@ def test_restoring_an_earlier_dropin_leaks_nothing_either():
    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	c7ea15dbb4	docs: what the drivers need from the SNMP profile, opt-outs, and how getters behave 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 26s Details CI / test (3.13) (push) Successful in 25s Details CI / test (3.10) (pull_request) Successful in 25s 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 25s Details The SNMP profile has to express fallback OIDs, tables folded by first, average or maximum, numbers inside text values, memory as used, free or percent, and the uptime source. TX errors counting drops becomes OPNsense's entry instead of every firewall's. A driver opts out of an inherited template with None, so callers ask callable(). USES_SSH, FULL_UPGRADE and platform stay as driver facts. Getters do not write, a failed read raises, a write reports its real result, and the application uses public driver methods only.	2026-10-14 10:56:17 +02:00
christianmanivong	01fe680456	docs: a structure that keeps device-types vendor- and product-agnostic CI / test (3.10) (push) Successful in 28s Details CI / test (3.11) (push) Successful in 26s 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 25s Details CI / test (3.11) (pull_request) Successful in 24s Details CI / test (3.12) (pull_request) Successful in 25s Details CI / test (3.13) (pull_request) Successful in 24s Details Roles and capabilities declare, templates run declared steps, reconcilers stay free of I/O. POSIX host readers, the SSH exec transport and the privilege layer move to a new package, napalm-unix. SNMP polling, ping sweeps, discovery scoring, guest profiles, display labels and warnings belong to the application; drivers declare SNMP OIDs as data and report facts. Fixes become templates whose values come from the caller. Says in which order this happens and where the open proposals #8-#14 and #22 go instead.	2026-10-14 10:20:55 +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