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
1 parent e509445572
commit 155b5bbbe4
@@ -74,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
@@ -378,8 +386,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,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)
@@ -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)

@@ -52,6 +58,9 @@ def build_if_metrics(
    *,
    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
@@ -61,8 +70,8 @@ def build_if_metrics(
            continue
        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
@@ -92,7 +105,9 @@ async def collect_ucd_metrics(
        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,6 +128,9 @@ 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),
        ),
    )

@@ -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
@@ -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}
@@ -227,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]:
@@ -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]:
@@ -49,17 +54,17 @@ def _joined_lines(output: str) -> List[str]:
    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]))
@@ -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]:
@@ -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})")
@@ -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.6.0"
+version = "4.7.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -41,6 +41,7 @@ classifiers = [
 ]
 dependencies = [
    "napalm>=4.0",
+    "typing-extensions>=4.0",
 ]

 [project.urls]
@@ -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)
@@ -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"
@@ -162,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

@@ -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,
@@ -139,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"),
@@ -141,6 +141,19 @@ 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
@@ -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
@@ -349,3 +435,46 @@ def test_mixin_reports_a_failed_way_back_instead_of_raising():

    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,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):