ssh_util: Handle sshd_config.d folder

Write sshd config to /etc/ssh/sshd_config.d/50-cloud-init.conf
if the sshd_config sources sshd_config.d

LP: #1968873

cloudinit/ssh_util.py

@@ -544,11 +544,28 @@ def parse_ssh_config_map(fname):
     return ret
 
 
+def _includes_dconf(fname: str) -> bool:
+    if not os.path.isfile(fname):
+        return False
+    with open(fname, "r") as f:
+        for line in f:
+            if line.startswith(f"Include {fname}.d/*.conf"):
+                return True
+    return False
+
+
 def update_ssh_config(updates, fname=DEF_SSHD_CFG):
     """Read fname, and update if changes are necessary.
 
     @param updates: dictionary of desired values {Option: value}
     @return: boolean indicating if an update was done."""
+    if _includes_dconf(fname):
+        if not os.path.isdir(f"{fname}.d"):
+            util.ensure_dir(f"{fname}.d", mode=0o755)
+        fname = os.path.join(f"{fname}.d", "50-cloud-init.conf")
+        if not os.path.isfile(fname):
+            # Ensure root read-only:
+            util.ensure_file(fname, 0o600)
     lines = parse_ssh_config(fname)
     changed = update_ssh_config_lines(lines=lines, updates=updates)
     if changed:

packages/debian/cloud-init.postrm

@@ -0,0 +1,11 @@
+#!/bin/bash
+
+set -e
+
+cleanup_sshd_config() {
+  rm -f "/etc/ssh/sshd_config.d/50-cloud-init.conf"
+}
+
+if [ "$1" = "purge" ]; then
+  cleanup_sshd_config
+fi

tests/integration_tests/modules/test_set_password.py

@@ -11,6 +11,7 @@
 import pytest
 import yaml
 
+from tests.integration_tests.clouds import ImageSpecification
 from tests.integration_tests.decorators import retry
 from tests.integration_tests.util import get_console_log
 
@@ -179,12 +180,22 @@ def test_shadow_expected_users(self, class_client):
             if "name" in user_dict:
                 assert f'{user_dict["name"]}:' in shadow
 
-    def test_sshd_config(self, class_client):
-        """Test that SSH password auth is enabled."""
-        sshd_config = class_client.read_from_file("/etc/ssh/sshd_config")
+    def test_sshd_config_file(self, class_client):
+        """Test that SSH config is written in the correct file."""
+        if ImageSpecification.from_os_image().release in {"bionic"}:
+            sshd_file_target = "/etc/ssh/sshd_config"
+        else:
+            sshd_file_target = "/etc/ssh/sshd_config.d/50-cloud-init.conf"
+        assert class_client.execute(f"ls {sshd_file_target}").ok
+        sshd_config = class_client.read_from_file(sshd_file_target)
         # We look for the exact line match, to avoid a commented line matching
         assert "PasswordAuthentication yes" in sshd_config.splitlines()
 
+    def test_sshd_config(self, class_client):
+        """Test that SSH password auth is enabled."""
+        sshd_config = class_client.execute("sshd -T").stdout
+        assert "passwordauthentication yes" in sshd_config
+
 
 @pytest.mark.user_data(LIST_USER_DATA)
 class TestPasswordList(Mixin):

tests/integration_tests/modules/test_ssh_keys_provided.py

@@ -9,6 +9,8 @@
 
 import pytest
 
+from tests.integration_tests.clouds import ImageSpecification
+
 USER_DATA = """\
 #cloud-config
 disable_root: false
@@ -109,10 +111,6 @@ class TestSshKeysProvided:
                 "AAAAHHNzaC1yc2EtY2VydC12MDFAb3BlbnNzaC5jb20AAAAgMpg"
                 "BP4Phn3L8I7Vqh7lmHKcOfIokEvSEbHDw83Y3JloAAAAD",
             ),
-            (
-                "/etc/ssh/sshd_config",
-                "HostCertificate /etc/ssh/ssh_host_rsa_key-cert.pub",
-            ),
             (
                 "/etc/ssh/ssh_host_ecdsa_key.pub",
                 "AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAAB"
@@ -138,3 +136,14 @@ class TestSshKeysProvided:
     def test_ssh_provided_keys(self, config_path, expected_out, class_client):
         out = class_client.read_from_file(config_path).strip()
         assert expected_out in out
+
+    @pytest.mark.parametrize(
+        "expected_out", ("HostCertificate /etc/ssh/ssh_host_rsa_key-cert.pub")
+    )
+    def test_sshd_config(self, expected_out, class_client):
+        if ImageSpecification.from_os_image().release in {"bionic"}:
+            sshd_config_path = "/etc/ssh/sshd_config"
+        else:
+            sshd_config_path = "/etc/ssh/sshd_config.d/50-cloud-init.conf"
+        sshd_config = class_client.read_from_file(sshd_config_path).strip()
+        assert expected_out in sshd_config

tests/unittests/config/test_cc_ssh.py

@@ -283,12 +283,30 @@ def test_handle_publish_hostkeys(
             expected_calls == cloud.datasource.publish_host_keys.call_args_list
         )
 
+    @pytest.mark.parametrize("with_sshd_dconf", [False, True])
+    @mock.patch(MODPATH + "util.ensure_dir")
     @mock.patch(MODPATH + "ug_util.normalize_users_groups")
     @mock.patch(MODPATH + "util.write_file")
-    def test_handle_ssh_keys_in_cfg(self, m_write_file, m_nug, m_setup_keys):
+    def test_handle_ssh_keys_in_cfg(
+        self,
+        m_write_file,
+        m_nug,
+        m_ensure_dir,
+        m_setup_keys,
+        with_sshd_dconf,
+        mocker,
+    ):
         """Test handle with ssh keys and certificate."""
         # Populate a config dictionary to pass to handle() as well
         # as the expected file-writing calls.
+        mocker.patch(
+            MODPATH + "ssh_util._includes_dconf", return_value=with_sshd_dconf
+        )
+        if with_sshd_dconf:
+            sshd_conf_fname = "/etc/ssh/sshd_config.d/50-cloud-init.conf"
+        else:
+            sshd_conf_fname = "/etc/ssh/sshd_config"
+
         cfg = {"ssh_keys": {}}
 
         expected_calls = []
@@ -324,7 +342,7 @@ def test_handle_ssh_keys_in_cfg(self, m_write_file, m_nug, m_setup_keys):
                         384,
                     ),
                     mock.call(
-                        "/etc/ssh/sshd_config",
+                        sshd_conf_fname,
                         "HostCertificate /etc/ssh/ssh_host_{}_key-cert.pub"
                         "\n".format(key_type),
                         preserve_mode=True,
@@ -343,6 +361,14 @@ def test_handle_ssh_keys_in_cfg(self, m_write_file, m_nug, m_setup_keys):
         for call_ in expected_calls:
             assert call_ in m_write_file.call_args_list
 
+        if with_sshd_dconf:
+            assert (
+                mock.call("/etc/ssh/sshd_config.d", mode=0o755)
+                in m_ensure_dir.call_args_list
+            )
+        else:
+            assert [] == m_ensure_dir.call_args_list
+
     @pytest.mark.parametrize(
         "key_type,reason",
         [

tests/unittests/test_ssh_util.py

@@ -1,6 +1,7 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 
 import os
+import stat
 from functools import partial
 from typing import NamedTuple
 from unittest import mock
@@ -579,6 +580,47 @@ def test_not_modified(self, tmpdir):
         assert self.cfgdata == util.load_file(mycfg)
         m_write_file.assert_not_called()
 
+    def test_without_include(self, tmpdir):
+        mycfg = tmpdir.join("sshd_config")
+        cfg = "X Y"
+        util.write_file(mycfg, cfg)
+        assert ssh_util.update_ssh_config({"key": "value"}, mycfg)
+        assert "X Y\nkey value\n" == util.load_file(mycfg)
+        expected_conf_file = f"{mycfg}.d/50-cloud-init.conf"
+        assert not os.path.isfile(expected_conf_file)
+
+    @pytest.mark.parametrize(
+        "cfg",
+        ["Include {mycfg}.d/*.conf", "Include {mycfg}.d/*.conf # comment"],
+    )
+    def test_with_include(self, cfg, tmpdir):
+        mycfg = tmpdir.join("sshd_config")
+        util.write_file(mycfg, cfg.format(mycfg=mycfg))
+        assert ssh_util.update_ssh_config({"key": "value"}, mycfg)
+        expected_conf_file = f"{mycfg}.d/50-cloud-init.conf"
+        assert os.path.isfile(expected_conf_file)
+        assert 0o600 == stat.S_IMODE(os.stat(expected_conf_file).st_mode)
+        assert "key value\n" == util.load_file(expected_conf_file)
+
+    def test_with_commented_include(self, tmpdir):
+        mycfg = tmpdir.join("sshd_config")
+        cfg = f"# Include {mycfg}.d/*.conf"
+        util.write_file(mycfg, cfg)
+        assert ssh_util.update_ssh_config({"key": "value"}, mycfg)
+        assert f"{cfg}\nkey value\n" == util.load_file(mycfg)
+        expected_conf_file = f"{mycfg}.d/50-cloud-init.conf"
+        assert not os.path.isfile(expected_conf_file)
+
+    def test_with_other_include(self, tmpdir):
+        mycfg = tmpdir.join("sshd_config")
+        cfg = f"Include other_{mycfg}.d/*.conf"
+        util.write_file(mycfg, cfg)
+        assert ssh_util.update_ssh_config({"key": "value"}, mycfg)
+        assert f"{cfg}\nkey value\n" == util.load_file(mycfg)
+        expected_conf_file = f"{mycfg}.d/50-cloud-init.conf"
+        assert not os.path.isfile(expected_conf_file)
+        assert not os.path.isfile(f"other_{mycfg}.d/50-cloud-init.conf")
+
 
 class TestBasicAuthorizedKeyParse:
     @pytest.mark.parametrize(