feat: OpenSshServerMixin takes flagged algorithms out of what sshd offers

ssh-audit names algorithms an SSH server should stop offering. Taking them
out is the same on every OpenSSH host, so it lives here once: a driver
mixes in OpenSshServerMixin and supplies run_command.

harden_ssh_server reads the lists sshd really uses (sshd -T), removes the
requested algorithms and writes the rest to
/etc/ssh/sshd_config.d/00-netork-ssh-hardening.conf -- or, where
sshd_config has no Include of that directory, rewrites sshd_config after
a copy. It never removes the last algorithm of a list, 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
previous state back and reloads again.

Bumps to 4.5.0. NetOrk/netork#1027

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"]