diff --git a/README.md b/README.md index 19982fb..57b27a2 100644 --- a/README.md +++ b/README.md @@ -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): diff --git a/napalm_device_types/_matching.py b/napalm_device_types/_matching.py new file mode 100644 index 0000000..8628053 --- /dev/null +++ b/napalm_device_types/_matching.py @@ -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) diff --git a/napalm_device_types/_ucd_metrics.py b/napalm_device_types/_ucd_metrics.py index 67d03e2..7148f9d 100644 --- a/napalm_device_types/_ucd_metrics.py +++ b/napalm_device_types/_ucd_metrics.py @@ -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 diff --git a/napalm_device_types/dhcp.py b/napalm_device_types/dhcp.py index e9ad5bc..0992c63 100644 --- a/napalm_device_types/dhcp.py +++ b/napalm_device_types/dhcp.py @@ -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 diff --git a/napalm_device_types/firewall_rules.py b/napalm_device_types/firewall_rules.py index 98cabcc..8858e8f 100644 --- a/napalm_device_types/firewall_rules.py +++ b/napalm_device_types/firewall_rules.py @@ -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 diff --git a/napalm_device_types/host_status.py b/napalm_device_types/host_status.py index ebf5a2b..ea257fa 100644 --- a/napalm_device_types/host_status.py +++ b/napalm_device_types/host_status.py @@ -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]: diff --git a/napalm_device_types/os.py b/napalm_device_types/os.py index c4bbae6..41785d0 100644 --- a/napalm_device_types/os.py +++ b/napalm_device_types/os.py @@ -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. diff --git a/napalm_device_types/package_updates.py b/napalm_device_types/package_updates.py index 7ca4697..55e3542 100644 --- a/napalm_device_types/package_updates.py +++ b/napalm_device_types/package_updates.py @@ -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])) diff --git a/napalm_device_types/roles.py b/napalm_device_types/roles.py index 3300245..dcf8815 100644 --- a/napalm_device_types/roles.py +++ b/napalm_device_types/roles.py @@ -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]: diff --git a/napalm_device_types/ssh_server.py b/napalm_device_types/ssh_server.py index 9b18673..16857d7 100644 --- a/napalm_device_types/ssh_server.py +++ b/napalm_device_types/ssh_server.py @@ -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})") diff --git a/napalm_device_types/virtual_ips.py b/napalm_device_types/virtual_ips.py index bef7b20..6733684 100644 --- a/napalm_device_types/virtual_ips.py +++ b/napalm_device_types/virtual_ips.py @@ -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 `` / ``interface`` / ``virtual_router_id`` / ``vip [dev ]`` #: / ``end``, and ``unreadable``/``too-deep``/``missing ``. 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; " diff --git a/pyproject.toml b/pyproject.toml index d484761..9bea142 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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] diff --git a/tests/test_dhcp_diff_apply.py b/tests/test_dhcp_diff_apply.py index c2c05cc..2d567ca 100644 --- a/tests/test_dhcp_diff_apply.py +++ b/tests/test_dhcp_diff_apply.py @@ -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) diff --git a/tests/test_dhcp_subnet_diff_apply.py b/tests/test_dhcp_subnet_diff_apply.py index 21048ef..f478c42 100644 --- a/tests/test_dhcp_subnet_diff_apply.py +++ b/tests/test_dhcp_subnet_diff_apply.py @@ -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) diff --git a/tests/test_firewall_diff_apply.py b/tests/test_firewall_diff_apply.py index 0e0d08f..471972c 100644 --- a/tests/test_firewall_diff_apply.py +++ b/tests/test_firewall_diff_apply.py @@ -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 == [] diff --git a/tests/test_host_status.py b/tests/test_host_status.py index e29e66d..1b3341c 100644 --- a/tests/test_host_status.py +++ b/tests/test_host_status.py @@ -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 diff --git a/tests/test_package_updates.py b/tests/test_package_updates.py index b4c575d..63d1342 100644 --- a/tests/test_package_updates.py +++ b/tests/test_package_updates.py @@ -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"), diff --git a/tests/test_role_contracts.py b/tests/test_role_contracts.py index 98a32ff..a8b9f50 100644 --- a/tests/test_role_contracts.py +++ b/tests/test_role_contracts.py @@ -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 diff --git a/tests/test_ssh_server.py b/tests/test_ssh_server.py index d91a974..462cea9 100644 --- a/tests/test_ssh_server.py +++ b/tests/test_ssh_server.py @@ -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" diff --git a/tests/test_ucd_metrics.py b/tests/test_ucd_metrics.py new file mode 100644 index 0000000..eec7614 --- /dev/null +++ b/tests/test_ucd_metrics.py @@ -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) diff --git a/tests/test_virtual_ips.py b/tests/test_virtual_ips.py index a200f74..8ec3cac 100644 --- a/tests/test_virtual_ips.py +++ b/tests/test_virtual_ips.py @@ -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):