From c53f04aeb2acf9526a2ebf3d3320f149ac46caa6 Mon Sep 17 00:00:00 2001
From: Ani Sinha <anisinha@redhat.com>
Date: Tue, 2 May 2023 20:35:45 +0530
Subject: [PATCH] Do not generate dsa and ed25519 key types when crypto FIPS
 mode is enabled (#2142)

Reference:https://github.com/canonical/cloud-init/commit/c53f04aeb2acf9526a2ebf3d3320f149ac46caa6
Conflict:(1)Add extra information to cc_ssh.py:
+import logging
+LOG = logging.getLogger(__name__)
(2)format diffs.
(3)add 'M_PATH = "cloudinit.util."' in test_util.py

DSA and ED25519 key types are not supported when FIPS is enabled in crypto.
Check if FIPS has been enabled on the system and if so, do not generate those
key types. Presently the check is only available on Linux systems.

LP: 2017761
RHBZ: 2187164

Signed-off-by: Ani Sinha <anisinha@redhat.com>
---
 cloudinit/config/cc_ssh.py         | 23 ++++++++++++++++-
 cloudinit/util.py                  | 12 +++++++++
 cloudinit/config/tests/test_ssh.py | 40 ++++++++++++++++++++++--------
 tests/unittests/test_util.py       | 26 +++++++++++++++++++
 4 files changed, 90 insertions(+), 11 deletions(-)

diff --git a/cloudinit/config/cc_ssh.py b/cloudinit/config/cc_ssh.py
index 1053ab6..2fda565 100755
--- a/cloudinit/config/cc_ssh.py
+++ b/cloudinit/config/cc_ssh.py
@@ -167,8 +167,12 @@ from cloudinit import ssh_util
 from cloudinit import subp
 from cloudinit import util
 
+import logging
+LOG = logging.getLogger(__name__)
 
 GENERATE_KEY_NAMES = ['rsa', 'dsa', 'ecdsa', 'ed25519']
+FIPS_UNSUPPORTED_KEY_NAMES = ["dsa", "ed25519"]
+
 KEY_FILE_TPL = '/etc/ssh/ssh_host_%s_key'
 PUBLISH_HOST_KEYS = True
 # Don't publish the dsa hostkey by default since OpenSSH recommends not using
@@ -231,9 +235,26 @@ def handle(_name, cfg, cloud, log, _args):
         genkeys = util.get_cfg_option_list(cfg,
                                            'ssh_genkeytypes',
                                            GENERATE_KEY_NAMES)
+        # remove keys that are not supported in fips mode if its enabled
+        key_names = (
+            genkeys
+            if not util.fips_enabled()
+            else [
+                names
+                for names in genkeys
+                if names not in FIPS_UNSUPPORTED_KEY_NAMES
+            ]
+        )
+        skipped_keys = set(genkeys).difference(key_names)
+        if skipped_keys:
+            LOG.debug(
+                "skipping keys that are not supported in fips mode: %s",
+                ",".join(skipped_keys),
+            )
+
         lang_c = os.environ.copy()
         lang_c['LANG'] = 'C'
-        for keytype in genkeys:
+        for keytype in key_names:
             keyfile = KEY_FILE_TPL % (keytype)
             if os.path.exists(keyfile):
                 continue
diff --git a/cloudinit/util.py b/cloudinit/util.py
index 78164de..c18aecf 100644
--- a/cloudinit/util.py
+++ b/cloudinit/util.py
@@ -1418,6 +1418,18 @@ def get_cmdline():
     return _get_cmdline()


+def fips_enabled() -> bool:
+    fips_proc = "/proc/sys/crypto/fips_enabled"
+    try:
+        contents = load_file(fips_proc).strip()
+        return contents == "1"
+    except (IOError, OSError):
+        # for BSD systems and Linux systems where the proc entry is not
+        # available, we assume FIPS is disabled to retain the old behavior
+        # for now.
+        return False
+
+
 def pipe_in_out(in_fh, out_fh, chunk_size=1024, chunk_cb=None):
     bytes_piped = 0
     while True:
diff --git a/cloudinit/config/tests/test_ssh.py b/cloudinit/config/tests/test_ssh.py
index 714949c..b8fb610 100644
--- a/cloudinit/config/tests/test_ssh.py
+++ b/cloudinit/config/tests/test_ssh.py
@@ -91,12 +91,16 @@ class TestHandleSsh:
             expected_calls = [mock.call(set(keys), user)] + expected_calls
         assert expected_calls == m_setup_keys.call_args_list
 
+    @pytest.mark.parametrize("fips_enabled", (True, False))
     @mock.patch(MODPATH + "glob.glob")
     @mock.patch(MODPATH + "ug_util.normalize_users_groups")
     @mock.patch(MODPATH + "os.path.exists")
-    def test_handle_no_cfg(self, m_path_exists, m_nug,
-                           m_glob, m_setup_keys):
+    @mock.patch(MODPATH + "util.fips_enabled")
+    def test_handle_no_cfg(
+        self, m_fips, m_path_exists, m_nug, m_glob, m_setup_keys, fips_enabled
+    ):
         """Test handle with no config ignores generating existing keyfiles."""
+        m_fips.return_value = fips_enabled
         cfg = {}
         keys = ["key1"]
         m_glob.return_value = []  # Return no matching keys to prevent removal
@@ -109,12 +113,22 @@ class TestHandleSsh:
         options = ssh_util.DISABLE_USER_OPTS.replace("$USER", "NONE")
         options = options.replace("$DISABLE_USER", "root")
         m_glob.assert_called_once_with('/etc/ssh/ssh_host_*key*')
-        assert [
-            mock.call("/etc/ssh/ssh_host_rsa_key"),
-            mock.call("/etc/ssh/ssh_host_dsa_key"),
-            mock.call("/etc/ssh/ssh_host_ecdsa_key"),
-            mock.call("/etc/ssh/ssh_host_ed25519_key"),
-        ] in m_path_exists.call_args_list
+        m_fips.assert_called_once()
+
+        if not m_fips():
+            expected_calls = [
+                mock.call("/etc/ssh/ssh_host_rsa_key"),
+                mock.call("/etc/ssh/ssh_host_dsa_key"),
+                mock.call("/etc/ssh/ssh_host_ecdsa_key"),
+                mock.call("/etc/ssh/ssh_host_ed25519_key"),
+            ]
+        else:
+            # Enabled fips doesn't generate dsa or ed25519
+            expected_calls = [
+                mock.call("/etc/ssh/ssh_host_rsa_key"),
+                mock.call("/etc/ssh/ssh_host_ecdsa_key"),
+            ]
+        assert expected_calls in m_path_exists.call_args_list
         assert [
             mock.call(set(keys), "root", options=options)
         ] == m_setup_keys.call_args_list
@@ -122,8 +136,10 @@ class TestHandleSsh:
     @mock.patch(MODPATH + "glob.glob")
     @mock.patch(MODPATH + "ug_util.normalize_users_groups")
     @mock.patch(MODPATH + "os.path.exists")
-    def test_dont_allow_public_ssh_keys(self, m_path_exists, m_nug,
-                                        m_glob, m_setup_keys):
+    @mock.patch(MODPATH + "util.fips_enabled", return_value=False)
+    def test_dont_allow_public_ssh_keys(
+        self, m_fips, m_path_exists, m_nug, m_glob, m_setup_keys
+    ):
         """Test allow_public_ssh_keys=False ignores ssh public keys from
            platform.
         """
@@ -166,8 +182,10 @@ class TestHandleSsh:
     @mock.patch(MODPATH + "glob.glob")
     @mock.patch(MODPATH + "ug_util.normalize_users_groups")
     @mock.patch(MODPATH + "os.path.exists")
+    @mock.patch(MODPATH + "util.fips_enabled", return_value=False)
     def test_handle_default_root(
         self,
+        m_fips,
         m_path_exists,
         m_nug,
         m_glob,
@@ -232,8 +250,10 @@ class TestHandleSsh:
     @mock.patch(MODPATH + "glob.glob")
     @mock.patch(MODPATH + "ug_util.normalize_users_groups")
     @mock.patch(MODPATH + "os.path.exists")
+    @mock.patch(MODPATH + "util.fips_enabled", return_value=False)
     def test_handle_publish_hostkeys(
         self,
+        m_fips,
         m_path_exists,
         m_nug,
         m_glob,
diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py
index 2ab3bad..7368b3b 100644
--- a/tests/unittests/test_util.py
+++ b/tests/unittests/test_util.py
@@ -16,6 +16,7 @@ from cloudinit import subp
 from cloudinit import importer, util
 from cloudinit.tests import helpers

+M_PATH = "cloudinit.util."

 class FakeSelinux(object):

@@ -287,6 +287,31 @@ class TestGetCmdline(helpers.TestCase):
         self.assertEqual("abcd 123", ret)
 
 
+class TestFipsEnabled:
+    @pytest.mark.parametrize(
+        "fips_enabled_content,expected",
+        (
+            pytest.param(None, False, id="false_when_no_fips_enabled_file"),
+            pytest.param("0\n", False, id="false_when_fips_disabled"),
+            pytest.param("1\n", True, id="true_when_fips_enabled"),
+            pytest.param("1", True, id="true_when_fips_enabled_no_newline"),
+        ),
+    )
+    @mock.patch(M_PATH + "load_file")
+    def test_fips_enabled_based_on_proc_crypto(
+        self, load_file, fips_enabled_content, expected, tmpdir
+    ):
+        def fake_load_file(path):
+            assert path == "/proc/sys/crypto/fips_enabled"
+            if fips_enabled_content is None:
+                raise IOError("No file exists Bob")
+            return fips_enabled_content
+
+        load_file.side_effect = fake_load_file
+
+        assert expected is util.fips_enabled()
+
+
 class TestLoadYaml(helpers.CiTestCase):
     mydefault = "7b03a8ebace993d806255121073fed52"
     with_logs = True
-- 
2.33.0


