diff --git a/README.md b/README.md index fab1d1f..75cb466 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,18 @@ A mechanism it could not read completely is `None` ("did not look"), a host with `[]`. The command is about 4 kB, so a driver sends it on an exec channel (`_run_virtual_ips_command(command, privileged=)`), not typed into a shell. +`OpenSshServerMixin` (`harden_ssh_server`, `get_ssh_server_algorithms`) takes algorithms +out of what an OpenSSH server offers -- the ones ssh-audit flags. It reads the lists sshd +really uses (`sshd -T`), takes out what was asked and writes the rest to +`/etc/ssh/sshd_config.d/00-netork-ssh-hardening.conf`; only where sshd_config has no such +`Include` is sshd_config itself rewritten, after a copy. It never removes the last algorithm +of a class, nor -- given the managing client's lists -- the last one that client speaks. +`sshd -t` must accept the result and `sshd -T` must show it; after the reload the caller's +`verify` opens a new session, and if that fails the still-open one puts the old state back. +A driver mixes it in and supplies `run_command`; a platform that regenerates sshd_config +itself (an appliance with its own database) must not, or opts out with +`harden_ssh_server = None`. + **Update readers raise when they cannot read.** `get_available_updates` returns an empty list only when nothing is pending; netOrk keeps "pending since" per package, and an empty list for "don't know" would reset it. diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index 54d2b82..a6e27d2 100644 --- a/napalm_device_types/__init__.py +++ b/napalm_device_types/__init__.py @@ -120,6 +120,13 @@ from napalm_device_types.virtual_ips import ( parse_virtual_ips, ) from napalm_device_types.residential_gateway import ResidentialGatewayDriver +from napalm_device_types.ssh_server import ( + SSH_ALGORITHM_CLASSES, + OpenSshServerMixin, + SshHardeningPlan, + SshHardeningResult, + plan_ssh_hardening, +) from napalm_device_types.storage import StorageDriver from napalm_device_types.switch import SwitchDriver @@ -178,6 +185,11 @@ __all__ = [ "UpdateMixin", "VIRTUAL_IPS_COMMAND", "VirtualIpsMixin", + "OpenSshServerMixin", + "SSH_ALGORITHM_CLASSES", + "SshHardeningPlan", + "SshHardeningResult", + "plan_ssh_hardening", "parse_virtual_ips", "add_lag_interfaces", "driver_supports_ping", diff --git a/napalm_device_types/ssh_server.py b/napalm_device_types/ssh_server.py new file mode 100644 index 0000000..c642a80 --- /dev/null +++ b/napalm_device_types/ssh_server.py @@ -0,0 +1,435 @@ +# -*- coding: utf-8 -*- +"""OpenSSH server hardening: taking algorithms out of what sshd offers. + +A scanner such as ssh-audit names algorithms a server should stop offering -- +``hmac-sha1``, the NIST curves. Taking them out is the same on every host that +runs OpenSSH, so the reading, the plan and the write live here once, and a +driver only mixes in :class:`OpenSshServerMixin` and supplies ``run_command``. + +**What the server offers is read, not assumed.** ``sshd -T`` prints the lists +sshd actually uses, after its defaults, ``sshd_config``, every ``Include`` and +first-match-wins. The new list is that one minus what was asked, so nothing the +host already restricted comes back, and nothing it allows is dropped by +accident. + +**A drop-in, not an edit.** Where ``sshd_config`` includes +``/etc/ssh/sshd_config.d/*.conf`` -- Debian and Ubuntu do, at the top -- the +lists go to :data:`DROPIN_PATH`, which sorts first and so wins. sshd_config +stays as the distribution shipped it, and removing the file undoes the change. +Only without such an ``Include`` is sshd_config itself rewritten, after a copy. + +**Nobody is locked out, by three checks.** + +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. +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. +3. After the reload, *verify* -- the caller's -- opens a new session. The + running one survives a reload, so if the new one cannot get in, this one + puts the old state back and reloads again. +""" + +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 + +if TYPE_CHECKING: # pragma: no cover - for type checkers only + from napalm_device_types.channel import CommandResult + +#: The algorithm classes: key -> (``sshd -T`` keyword, sshd_config directive). +SSH_ALGORITHM_CLASSES: Dict[str, Tuple[str, str]] = { + "kex": ("kexalgorithms", "KexAlgorithms"), + "hostkey": ("hostkeyalgorithms", "HostKeyAlgorithms"), + "cipher": ("ciphers", "Ciphers"), + "mac": ("macs", "MACs"), +} + +MAIN_CONFIG = "/etc/ssh/sshd_config" +DROPIN_DIR = "/etc/ssh/sshd_config.d" +#: ``00-`` so it sorts before every other drop-in: sshd keeps the first value it reads. +DROPIN_PATH = f"{DROPIN_DIR}/00-netork-ssh-hardening.conf" +#: What marks a line this module disabled in sshd_config. +DISABLED_PREFIX = "# netork: " + +_HEADER = ( + "# Written by netOrk: the SSH algorithms a security assessment flagged,\n" + "# taken out of what this server offers. Delete this file to undo it.\n" +) + +_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;" + " else for f in /run/sshd.pid /var/run/sshd.pid; do" + ' [ -r "$f" ] && kill -HUP "$(cat "$f")" && exit 0; done; exit 0; fi' +) + + +def _directive_key(line: str) -> Optional[str]: + """The algorithm class a global sshd_config line sets, if it sets one.""" + parts = line.split(None, 1) + if not parts: + return None + word = parts[0].lower() + for kind, (keyword, _directive) in SSH_ALGORITHM_CLASSES.items(): + if word == keyword: + return kind + return None + + +def parse_sshd_effective(text: str) -> Dict[str, List[str]]: + """The four algorithm lists from ``sshd -T`` output, in the server's order. + + :raises ValueError: if a list is missing -- not sshd -T output, or cut short. + """ + found: Dict[str, List[str]] = {} + for raw in text.splitlines(): + parts = raw.strip().split(None, 1) + if len(parts) != 2: + continue + kind = _directive_key(parts[0]) + if kind and kind not in found: + found[kind] = [a for a in parts[1].split(",") if a] + missing = [k for k in SSH_ALGORITHM_CLASSES if k not in found] + if missing: + raise ValueError(f"sshd -T printed no {', '.join(missing)} list") + return found + + +@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"`` + (the server does not offer it, so there is nothing to remove). + """ + + kind: str + algorithm: str + reason: str + + +@dataclass(frozen=True) +class SshHardeningPlan: + keep: Dict[str, List[str]] + remove: Dict[str, List[str]] + refused: List[Refusal] = field(default_factory=list) + not_offered: List[Refusal] = field(default_factory=list) + + @property + def changed_classes(self) -> List[str]: + return [k for k in SSH_ALGORITHM_CLASSES if self.remove.get(k)] + + +def plan_ssh_hardening( + effective: Mapping[str, Sequence[str]], + remove: Mapping[str, Iterable[str]], + *, + client: Optional[Mapping[str, 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". + + 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``. + + :raises ValueError: for a class that is not one of :data:`SSH_ALGORITHM_CLASSES`. + """ + unknown = [k for k in remove if k not in SSH_ALGORITHM_CLASSES] + if unknown: + raise ValueError(f"unknown SSH algorithm class: {', '.join(unknown)}") + keep: Dict[str, List[str]] = {} + removed: Dict[str, List[str]] = {} + refused: List[Refusal] = [] + not_offered: List[Refusal] = [] + 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] + 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) + if spared: + going.remove(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")) + keep[kind] = [a for a in offered if a not in going] + if going: + removed[kind] = going + return SshHardeningPlan(keep=keep, remove=removed, refused=refused, not_offered=not_offered) + + +def render_dropin(keep: Mapping[str, Sequence[str]], kinds: Iterable[str]) -> str: + """The drop-in file stating the full list of each class in *kinds*.""" + lines = [ + f"{SSH_ALGORITHM_CLASSES[k][1]} {','.join(keep[k])}\n" + for k in SSH_ALGORITHM_CLASSES + if k in set(kinds) + ] + return _HEADER + "".join(lines) + + +def dropin_classes(text: str) -> List[str]: + """The classes an existing drop-in sets.""" + return [k for k in (_directive_key(line.strip()) for line in text.splitlines()) if k] + + +def _is_match(line: str) -> bool: + parts = line.split(None, 1) + return bool(parts) and parts[0].lower() == "match" + + +def has_dropin_include(text: str) -> bool: + """Whether sshd_config's global section includes :data:`DROPIN_DIR`.""" + for raw in text.splitlines(): + line = raw.strip() + if _is_match(line): + return False + parts = line.split() + if len(parts) >= 2 and parts[0].lower() == "include": + if any(p.startswith(DROPIN_DIR + "/") for p in parts[1:]): + return True + return False + + +def _split_our_block(text: str) -> Tuple[List[str], str]: + """(the classes an earlier rewrite put above the file, the file without them).""" + if not text.startswith(_HEADER): + return [], text + lines = text[len(_HEADER):].splitlines(keepends=True) + ours = 0 + while ours < len(lines) and _directive_key(lines[ours].strip()): + ours += 1 + return dropin_classes("".join(lines[:ours])), "".join(lines[ours:]) + + +def render_main_config(text: str, keep: Mapping[str, Sequence[str]], kinds: Iterable[str]) -> str: + """sshd_config with the lists of *kinds* first and their old global lines disabled. + + First because sshd keeps the first value it reads. A ``Match`` block is left + alone -- it is not the global setting. A class an earlier rewrite set stays + set, and rewriting a file this already rewrote gives the same file. + """ + earlier, body = _split_our_block(text) + wanted = set(kinds) | set(earlier) + out: List[str] = [] + in_match = False + for raw in body.splitlines(): + if _is_match(raw.strip()): + in_match = True + if not in_match and _directive_key(raw.strip()) in wanted: + out.append(DISABLED_PREFIX + raw) + else: + out.append(raw) + return render_dropin(keep, wanted) + "\n".join(out) + ("\n" if body.endswith("\n") else "") + + +@dataclass +class SshHardeningResult: + """What :meth:`OpenSshServerMixin.harden_ssh_server` did. + + *applied* is True only when the new lists are live and a new session got + in. *rolled_back* says the host was changed and then put back. *backup* is + where sshd_config was copied to, when it was the file changed. + """ + + plan: SshHardeningPlan + applied: bool = False + rolled_back: bool = False + path: Optional[str] = None + backup: Optional[str] = None + message: str = "" + + +class OpenSshServerMixin: + """Adds :meth:`harden_ssh_server` to a driver that can run commands as root on an OpenSSH host. + + Mixed in by the drivers whose host runs a stock OpenSSH, not by a base class, + so ``hasattr(driver, "harden_ssh_server")`` stays a truthful answer. A driver + whose platform writes sshd_config itself (an appliance that regenerates it + from its own database) must not inherit this, or must opt out with + ``harden_ssh_server = None``: its file would be overwritten on the next + apply. + """ + + if TYPE_CHECKING: # pragma: no cover - declared for type checkers only + + def run_command( + self, + command: str, + *, + privileged: bool = False, + timeout: float = 60, + stdin: Optional[bytes] = None, + ) -> CommandResult: ... + + # ── the host ──────────────────────────────────────────────────────── + + def _ssh_run(self, command: str, stdin: Optional[bytes] = None) -> "CommandResult": + return self.run_command(command, privileged=True, timeout=30, stdin=stdin) + + def _ssh_read(self, path: str) -> Optional[str]: + result = self._ssh_run(f"cat -- {quote(path)}") + return result.stdout if result.exit_code == 0 else None + + def _ssh_write(self, path: str, content: str) -> None: + script = ( + f'tmp={quote(path + ".netork-tmp")} && cat > "$tmp" && chmod 644 "$tmp"' + f' && mv -f -- "$tmp" {quote(path)} || {{ s=$?; rm -f -- "$tmp"; exit $s; }}' + ) + result = self._ssh_run(script, stdin=content.encode()) + if result.exit_code != 0: + raise RuntimeError(f"could not write {path}: {(result.stderr or result.stdout).strip()}") + + def _ssh_remove(self, path: str) -> None: + self._ssh_run(f"rm -f -- {quote(path)}") + + def _reload_ssh_server(self) -> None: + """Make the running sshd read its configuration again; sessions stay up.""" + result = self._ssh_run(_RELOAD) + if result.exit_code != 0: + raise RuntimeError(f"could not reload sshd: {(result.stderr or result.stdout).strip()}") + + def get_ssh_server_algorithms(self) -> Dict[str, List[str]]: + """The algorithm lists sshd uses now (``sshd -T``), per class. + + :raises RuntimeError: if sshd -T fails -- no OpenSSH, or no root. + """ + 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) + + def _sshd_accepts(self) -> Optional[str]: + """None when ``sshd -t`` accepts the configuration, else what it said.""" + result = self._ssh_run("PATH=$PATH:/usr/sbin:/sbin sshd -t") + return None if result.exit_code == 0 else (result.stderr or result.stdout).strip() + + # ── the change ────────────────────────────────────────────────────── + + def harden_ssh_server( + self, + remove: Mapping[str, Iterable[str]], + *, + client: Optional[Mapping[str, Iterable[str]]] = None, + verify: Optional[Callable[[], bool]] = None, + log: Optional[Callable[[str], None]] = None, + ) -> SshHardeningResult: + """Stop the server offering the algorithms in *remove* (class -> names). + + *client* guards the managing client's algorithms (see + :func:`plan_ssh_hardening`); *verify* opens a new session after the + reload and returns whether it got in; *log* receives a line per step. + + Never raises after the host was touched: a failure from there on puts + the previous state back and says so in the result. + + :raises RuntimeError: if the current state cannot be read. + """ + 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) + for r in plan.refused: + why = "it is the last one" if r.reason == "last" else "netOrk could no longer connect" + say(f"Keeping {r.algorithm}: {why}") + kinds = plan.changed_classes + result = SshHardeningResult(plan=plan) + if not kinds: + result.message = "Nothing to change: the server offers none of the selected algorithms." + say(result.message) + return result + for kind in kinds: + say(f"Removing {', '.join(plan.remove[kind])} from {SSH_ALGORITHM_CLASSES[kind][1]}") + + main = self._ssh_read(MAIN_CONFIG) + if main is None: + raise RuntimeError(f"cannot read {MAIN_CONFIG}") + restore = self._write_change(plan, main, result, say) + return self._apply_or_restore(plan, kinds, result, restore, verify, say) + + def _write_change( + self, plan: SshHardeningPlan, main: str, result: SshHardeningResult, say: Callable[[str], None] + ) -> Callable[[], None]: + """Write the new lists; return what puts the previous state back.""" + if has_dropin_include(main): + previous = self._ssh_read(DROPIN_PATH) + kinds = set(plan.changed_classes) | set(dropin_classes(previous or "")) + self._ssh_run(f"mkdir -p -- {quote(DROPIN_DIR)}") + self._ssh_write(DROPIN_PATH, render_dropin(plan.keep, kinds)) + result.path = DROPIN_PATH + say(f"Wrote {DROPIN_PATH}") + + def restore() -> None: + if previous is None: + self._ssh_remove(DROPIN_PATH) + else: + self._ssh_write(DROPIN_PATH, previous) + + return restore + backup = f"{MAIN_CONFIG}.netork-{time.strftime('%Y%m%d-%H%M%S')}.bak" + self._ssh_write(backup, main) + 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})") + + def restore_main() -> None: + self._ssh_write(MAIN_CONFIG, main) + + return restore_main + + def _apply_or_restore( + self, + plan: SshHardeningPlan, + kinds: List[str], + result: SshHardeningResult, + restore: Callable[[], None], + verify: Optional[Callable[[], bool]], + say: Callable[[str], None], + ) -> SshHardeningResult: + def back(message: str, reload: bool) -> SshHardeningResult: + say(message) + say("Putting the previous configuration back") + restore() + if reload: + self._reload_ssh_server() + result.rolled_back = True + result.message = message + return result + + try: + say("Checking the configuration (sshd -t)") + complaint = self._sshd_accepts() + if complaint: + return back(f"sshd rejected the new configuration: {complaint}", reload=False) + now = self.get_ssh_server_algorithms() + ignored = [SSH_ALGORITHM_CLASSES[k][1] for k in kinds if now[k] != plan.keep[k]] + if ignored: + return back( + f"sshd does not use the new {', '.join(ignored)}: a setting read before" + f" {result.path} decides them", + reload=False, + ) + say("Reloading sshd") + self._reload_ssh_server() + if verify is not None: + say("Opening a new SSH session to check netOrk still gets in") + if not verify(): + return back("A new SSH session could not log in after the reload", reload=True) + except Exception as exc: # noqa: BLE001 - whatever went wrong, the host goes back + return back(f"The change failed: {exc}", reload=True) + result.applied = True + result.message = "The server no longer offers the selected algorithms." + say(result.message) + return result diff --git a/pyproject.toml b/pyproject.toml index f8d3716..36ed61a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-device-types" -version = "4.4.0" +version = "4.5.0" description = "Abstract device-type base classes for NAPALM drivers" readme = "README.md" requires-python = ">=3.10" diff --git a/tests/test_ssh_server.py b/tests/test_ssh_server.py new file mode 100644 index 0000000..fba29e2 --- /dev/null +++ b/tests/test_ssh_server.py @@ -0,0 +1,303 @@ +"""OpenSSH server hardening: removing algorithms a scanner flagged, without locking anyone out. + +The plan is pure and tested on its own. The mixin is tested against a fake host +that keeps files and answers ``sshd -T`` from them, so every path -- the +drop-in, the sshd_config fallback, a refused config, a reload the new session +cannot get through -- runs without a machine. +""" + +from __future__ import annotations + +import re +from typing import Dict, List, Optional + +import pytest + +from napalm_device_types.channel import CommandResult +from napalm_device_types.ssh_server import ( + DROPIN_PATH, + MAIN_CONFIG, + OpenSshServerMixin, + has_dropin_include, + parse_sshd_effective, + plan_ssh_hardening, + render_dropin, + render_main_config, +) + +SSHD_T = """\ +port 22 +kexalgorithms sntrup761x25519-sha512@openssh.com,curve25519-sha256,curve25519-sha256@libssh.org,ecdh-sha2-nistp256,ecdh-sha2-nistp384,diffie-hellman-group14-sha256 +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 +""" + +PARAMIKO = { + "kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "diffie-hellman-group14-sha256"], + "hostkey": ["ssh-ed25519", "ecdsa-sha2-nistp256", "rsa-sha2-512"], + "cipher": ["aes128-ctr", "aes256-gcm@openssh.com"], + "mac": ["hmac-sha2-256", "hmac-sha2-256-etm@openssh.com", "hmac-sha1"], +} + + +# ── parse ──────────────────────────────────────────────────────────────── + + +def test_parse_reads_the_four_lists_in_server_order(): + eff = parse_sshd_effective(SSHD_T) + assert eff["kex"][:2] == ["sntrup761x25519-sha512@openssh.com", "curve25519-sha256"] + assert eff["hostkey"] == ["ssh-ed25519", "ecdsa-sha2-nistp256", "rsa-sha2-512", "rsa-sha2-256"] + assert eff["cipher"] == ["chacha20-poly1305@openssh.com", "aes128-ctr", "aes256-gcm@openssh.com"] + assert eff["mac"][-1] == "hmac-sha2-256" + + +def test_parse_refuses_output_without_the_lists(): + with pytest.raises(ValueError): + parse_sshd_effective("port 22\n") + + +# ── plan ───────────────────────────────────────────────────────────────── + + +def test_plan_removes_what_was_asked_and_keeps_the_order(): + plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"mac": ["hmac-sha1", "umac-64-etm@openssh.com"]}) + assert plan.remove == {"mac": ["umac-64-etm@openssh.com", "hmac-sha1"]} + assert plan.keep["mac"] == ["hmac-sha2-256-etm@openssh.com", "hmac-sha2-256"] + assert plan.changed_classes == ["mac"] + assert plan.refused == [] + + +def test_plan_ignores_an_algorithm_the_server_no_longer_offers(): + plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"cipher": ["3des-cbc"]}) + assert plan.changed_classes == [] + assert [r.algorithm for r in plan.not_offered] == ["3des-cbc"] + + +def test_plan_never_removes_the_last_algorithm_of_a_class(): + plan = plan_ssh_hardening( + parse_sshd_effective(SSHD_T), + {"cipher": ["chacha20-poly1305@openssh.com", "aes128-ctr", "aes256-gcm@openssh.com"]}, + ) + # The first in the server's order stays, the rest go. + assert plan.keep["cipher"] == ["chacha20-poly1305@openssh.com"] + assert plan.remove["cipher"] == ["aes128-ctr", "aes256-gcm@openssh.com"] + assert [(r.kind, r.algorithm, r.reason) for r in plan.refused] == [ + ("cipher", "chacha20-poly1305@openssh.com", "last") + ] + + +def test_plan_keeps_one_algorithm_the_managing_client_still_speaks(): + """paramiko knows no sntrup or plain curve25519: removing everything it does + know would leave the server reachable, but not by netOrk.""" + plan = plan_ssh_hardening( + parse_sshd_effective(SSHD_T), + {"kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "ecdh-sha2-nistp384", + "diffie-hellman-group14-sha256"]}, + client=PARAMIKO, + ) + assert "curve25519-sha256@libssh.org" in plan.keep["kex"] + assert [(r.algorithm, r.reason) for r in plan.refused] == [("curve25519-sha256@libssh.org", "client")] + assert plan.remove["kex"] == ["ecdh-sha2-nistp256", "ecdh-sha2-nistp384", "diffie-hellman-group14-sha256"] + + +def test_plan_without_a_client_list_only_guards_the_last_algorithm(): + plan = plan_ssh_hardening( + parse_sshd_effective(SSHD_T), + {"kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "ecdh-sha2-nistp384", + "diffie-hellman-group14-sha256"]}, + ) + assert plan.refused == [] + assert plan.keep["kex"] == ["sntrup761x25519-sha512@openssh.com", "curve25519-sha256"] + + +def test_plan_rejects_an_unknown_class(): + with pytest.raises(ValueError): + plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"compression": ["zlib"]}) + + +# ── render ─────────────────────────────────────────────────────────────── + + +def test_dropin_states_full_lists_for_the_changed_classes_only(): + plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"mac": ["hmac-sha1"]}) + text = render_dropin(plan.keep, ["mac"]) + assert "MACs umac-64-etm@openssh.com,hmac-sha2-256-etm@openssh.com,hmac-sha2-256\n" in text + assert "KexAlgorithms" not in text + assert text.startswith("#") # says who wrote it + + +def test_include_is_found_only_in_the_global_section(): + assert has_dropin_include("Include /etc/ssh/sshd_config.d/*.conf\nPort 22\n") + assert has_dropin_include(" include /etc/ssh/sshd_config.d/*.conf\n") + assert not has_dropin_include("Port 22\n") + assert not has_dropin_include("Port 22\nMatch User x\n Include /etc/ssh/sshd_config.d/*.conf\n") + + +def test_main_config_gets_the_lists_first_and_the_old_lines_disabled(): + main = "# defaults\nPort 22\nMACs hmac-sha1,hmac-sha2-256\nMatch User backup\n MACs hmac-sha1\n" + text = render_main_config(main, {"mac": ["hmac-sha2-256"]}, ["mac"]) + lines = text.splitlines() + first_directive = next(line for line in lines if line and not line.startswith("#")) + assert first_directive == "MACs hmac-sha2-256" + assert "# netork: MACs hmac-sha1,hmac-sha2-256" in lines + # A Match block is left alone: it is not the global setting. + assert " MACs hmac-sha1" in lines + + +def test_main_config_rewrite_is_idempotent(): + main = "Port 22\nMACs hmac-sha1,hmac-sha2-256\n" + once = render_main_config(main, {"mac": ["hmac-sha2-256"]}, ["mac"]) + twice = render_main_config(once, {"mac": ["hmac-sha2-256"]}, ["mac"]) + assert once == twice + + +# ── the mixin, against a fake host ─────────────────────────────────────── + +_DIRECTIVE = {"kexalgorithms": "kex", "hostkeyalgorithms": "hostkey", "ciphers": "cipher", "macs": "mac"} + + +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: + self.files: Dict[str, str] = {MAIN_CONFIG: main} + self.defaults = parse_sshd_effective(SSHD_T) + self.ran: List[str] = [] + self.syntax_ok = True + self.reloads = 0 + + def _effective(self) -> Dict[str, List[str]]: + eff = {k: list(v) for k, v in self.defaults.items()} + seen = set() + lines: List[str] = [] + for line in self.files[MAIN_CONFIG].splitlines(): + if re.match(r"\s*include\s", line, re.I): + lines += self.files.get(DROPIN_PATH, "").splitlines() + elif re.match(r"\s*match\s", line, re.I): + break + else: + lines.append(line) + for line in lines: + parts = line.split() + if len(parts) == 2 and parts[0].lower() in _DIRECTIVE: + kind = _DIRECTIVE[parts[0].lower()] + if kind not in seen: + seen.add(kind) + eff[kind] = parts[1].split(",") + return eff + + def run_command(self, command: str, *, privileged: bool = False, timeout: float = 60, + stdin: Optional[bytes] = None) -> CommandResult: + self.ran.append(command) + assert privileged, command + 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()) + 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) + m = re.match(r"cat -- '?([^' ]+)'?$", command) + if m: + path = m.group(1) + if path in self.files: + return CommandResult(self.files[path], "", 0) + return CommandResult("", "No such file", 1) + m = re.search(r"mv -f -- \"\$tmp\" '?([^' ]+)'?", command) + if m: + self.files[m.group(1)] = (stdin or b"").decode() + return CommandResult("", "", 0) + m = re.match(r"rm -f -- '?([^' ]+)'?$", command) + if m: + self.files.pop(m.group(1), None) + return CommandResult("", "", 0) + if "reload" in command or "HUP" in command: + self.reloads += 1 + return CommandResult("", "", 0) + if command.startswith("mkdir"): + return CommandResult("", "", 0) + raise AssertionError(f"unexpected command: {command}") + + +def test_mixin_writes_the_dropin_checks_it_and_reloads(): + host = FakeHost() + lines: List[str] = [] + result = host.harden_ssh_server({"mac": ["hmac-sha1"]}, client=PARAMIKO, log=lines.append) + assert result.applied and not result.rolled_back + assert result.path == DROPIN_PATH + assert "hmac-sha1" not in host._effective()["mac"] + assert host.reloads == 1 + assert any("sshd -t" in c for c in host.ran) + assert lines # it says what it does + + +def test_mixin_keeps_what_an_earlier_fix_wrote(): + host = FakeHost() + host.harden_ssh_server({"hostkey": ["ecdsa-sha2-nistp256"]}) + host.harden_ssh_server({"mac": ["hmac-sha1"]}) + eff = host._effective() + assert "ecdsa-sha2-nistp256" not in eff["hostkey"] + assert "hmac-sha1" not in eff["mac"] + + +def test_mixin_falls_back_to_sshd_config_without_an_include(): + host = FakeHost(main="Port 22\nMACs hmac-sha1,hmac-sha2-256\n") + result = host.harden_ssh_server({"mac": ["hmac-sha1"]}) + assert result.path == MAIN_CONFIG + assert result.backup and host.files[result.backup] == "Port 22\nMACs hmac-sha1,hmac-sha2-256\n" + assert host._effective()["mac"] == ["hmac-sha2-256"] + + +def test_mixin_restores_when_sshd_rejects_the_config(): + host = FakeHost() + host.syntax_ok = False + result = host.harden_ssh_server({"mac": ["hmac-sha1"]}) + assert not result.applied and result.rolled_back + assert DROPIN_PATH not in host.files + assert host.reloads == 0 + assert "bad line 3" in result.message + + +def test_mixin_restores_when_the_effective_lists_do_not_change(): + """Another file read before ours sets the list: ours is valid and ignored.""" + host = FakeHost(main="MACs hmac-sha1,hmac-sha2-256\nInclude /etc/ssh/sshd_config.d/*.conf\n") + result = host.harden_ssh_server({"mac": ["hmac-sha1"]}) + assert not result.applied and result.rolled_back + assert DROPIN_PATH not in host.files + assert "MACs" in result.message + + +def test_mixin_rolls_back_when_a_new_session_cannot_get_in(): + host = FakeHost() + host.files[DROPIN_PATH] = "# netOrk\nHostKeyAlgorithms ssh-ed25519\n" + before = host.files[DROPIN_PATH] + result = host.harden_ssh_server({"mac": ["hmac-sha1"]}, verify=lambda: False) + assert not result.applied and result.rolled_back + assert host.files[DROPIN_PATH] == before + assert host.reloads == 2 # the change, and the way back + + +def test_mixin_does_nothing_when_nothing_would_change(): + host = FakeHost() + result = host.harden_ssh_server({"cipher": ["3des-cbc"]}) + assert not result.applied and not result.rolled_back + assert host.reloads == 0 + assert DROPIN_PATH not in host.files + + +def test_declared_on_no_base_class(): + """hasattr is the capability probe: only a driver that mixes it in answers True.""" + from napalm_device_types import DeviceTypeDriver, OSDriver + + assert not hasattr(DeviceTypeDriver, "harden_ssh_server") + assert not hasattr(OSDriver, "harden_ssh_server") + + +def test_main_config_second_fix_keeps_what_the_first_set(): + host = FakeHost(main="Port 22\n") + host.harden_ssh_server({"hostkey": ["ecdsa-sha2-nistp256"]}) + host.harden_ssh_server({"mac": ["hmac-sha1"]}) + eff = host._effective() + assert "ecdsa-sha2-nistp256" not in eff["hostkey"] + assert "hmac-sha1" not in eff["mac"]