fix(ug_util): prioritize user-data.users over the default user config

cloudinit/distros/ug_util.py

@@ -134,13 +134,14 @@ def _normalize_users(u_cfg, def_user_cfg=None):
             # Now merge the extracted groups with the default config provided
             users_groups = util.uniq_merge_sorted(parsed_groups, def_groups)
             parsed_config["groups"] = ",".join(users_groups)
-            # The real config for the default user is the combination of the
-            # default user config provided by the distro, the default user
-            # config provided by the above merging for the user 'default' and
-            # then the parsed config from the user's 'real name' which does not
-            # have to be 'default' (but could be)
+            # The real config for the default user is the combination of:
+            # - the parsed config from the user's 'real name' which does
+            #   not have to be 'default' (but could be)
+            # - then the default user config provided by the above merging
+            #   for the user 'default'
+            # - then the default user config provided by the distro
             users[def_user] = util.mergemanydict(
-                [def_user_cfg, def_config, parsed_config]
+                [parsed_config, def_config, def_user_cfg]
             )
 
     # Ensure that only the default user that we found (if any) is actually

tests/integration_tests/modules/test_set_password.py

@@ -14,7 +14,10 @@
 
 from tests.integration_tests.decorators import retry
 from tests.integration_tests.releases import CURRENT_RELEASE, IS_UBUNTU
-from tests.integration_tests.util import get_console_log
+from tests.integration_tests.util import (
+    fetch_and_parse_etc_shadow,
+    get_console_log,
+)
 
 COMMON_USER_DATA = """\
 #cloud-config
@@ -38,6 +41,9 @@
     # sha256 gojanego
     passwd: "$5$iW$XsxmWCdpwIW8Yhv.Jn/R3uk6A4UaicfW5Xp7C9p9pg."
     lock_passwd: false
+  - name: sally
+    # sha256 gosallygo
+    passwd: "$5$bA$KBMTe8lXf0e8lE4f4hYPU0h6h0HzQX4vHpnq6xHn9Q2"
   - name: "mikey"
     lock_passwd: false
 """
@@ -94,41 +100,33 @@
 class Mixin:
     """Shared test definitions."""
 
-    def _fetch_and_parse_etc_shadow(self, class_client):
-        """Fetch /etc/shadow and parse it into Python data structures
-
-        Returns: ({user: password}, [duplicate, users])
-        """
-        shadow_content = class_client.read_from_file("/etc/shadow")
-        users = {}
-        dupes = []
-        for line in shadow_content.splitlines():
-            user, encpw = line.split(":")[0:2]
-            if user in users:
-                dupes.append(user)
-            users[user] = encpw
-        return users, dupes
-
     def test_no_duplicate_users_in_shadow(self, class_client):
         """Confirm that set_passwords has not added duplicate shadow entries"""
-        _, dupes = self._fetch_and_parse_etc_shadow(class_client)
+        _, dupes = fetch_and_parse_etc_shadow(class_client)
 
         assert [] == dupes
 
     def test_password_in_users_dict_set_correctly(self, class_client):
         """Test that the password specified in the users dict is set."""
-        shadow_users, _ = self._fetch_and_parse_etc_shadow(class_client)
+        shadow_users, _ = fetch_and_parse_etc_shadow(class_client)
         assert USERS_PASSWD_VALUES["jane"] == shadow_users["jane"]
 
+    def test_hashed_password_without_lock_passwd_override_is_locked(
+        self, class_client
+    ):
+        """Hashed passwords are locked when lock_passwd is not set."""
+        shadow_users, _ = fetch_and_parse_etc_shadow(class_client)
+        assert f"!{USERS_PASSWD_VALUES['sally']}" == shadow_users["sally"]
+
     def test_password_in_chpasswd_list_set_correctly(self, class_client):
         """Test that a chpasswd password overrides one in the users dict."""
-        shadow_users, _ = self._fetch_and_parse_etc_shadow(class_client)
+        shadow_users, _ = fetch_and_parse_etc_shadow(class_client)
         mikey_hash = "$5$xZ$B2YGGEx2AOf4PeW48KC6.QyT1W2B4rZ9Qbltudtha89"
         assert mikey_hash == shadow_users["mikey"]
 
     def test_random_passwords_set_correctly(self, class_client):
         """Test that RANDOM chpasswd entries replace users dict passwords."""
-        shadow_users, _ = self._fetch_and_parse_etc_shadow(class_client)
+        shadow_users, _ = fetch_and_parse_etc_shadow(class_client)
 
         # These should have been changed
         assert shadow_users["harry"] != USERS_PASSWD_VALUES["harry"]
@@ -170,7 +168,7 @@ def test_explicit_password_set_correctly(self, class_client):
         )
         if minor_version > 12:
             pytest.xfail("Instance under test doesn't have 'crypt' in stdlib")
-        shadow_users, _ = self._fetch_and_parse_etc_shadow(class_client)
+        shadow_users, _ = fetch_and_parse_etc_shadow(class_client)
 
         fmt_and_salt = shadow_users["tom"].rsplit("$", 1)[0]
 

tests/integration_tests/modules/test_users_groups.py

@@ -10,13 +10,17 @@
 import pytest
 
 from tests.integration_tests.instances import IntegrationInstance
+from tests.integration_tests.integration_settings import CLOUD_INIT_SOURCE
 from tests.integration_tests.releases import (
     CURRENT_RELEASE,
     IS_UBUNTU,
     JAMMY,
     NOBLE,
 )
-from tests.integration_tests.util import verify_clean_boot
+from tests.integration_tests.util import (
+    fetch_and_parse_etc_shadow,
+    verify_clean_boot,
+)
 
 USER_DATA = """\
 #cloud-config
@@ -192,3 +196,62 @@ def test_sudoers_includedir(client: IntegrationInstance):
         "/etc/sudoers.d/90-cloud-init-users"
     ).splitlines()[1:]
     assert sudoers_content_before == sudoers_content_after
+
+
+USER_DATA_OVERRIDE = """\
+#cloud-config
+users:
+  - default
+  - name: ubuntu
+    shell: /bin/sh
+    lock_passwd: false
+    hashed_passwd: $5$xZ$B2YGGEx2AOf4PeW48KC6.QyT1W2B4rZ9Qbltudtha89
+"""
+
+
+@pytest.mark.ci
+@pytest.mark.skipif(not IS_UBUNTU, reason="Test is Ubuntu specific")
+@pytest.mark.user_data(USER_DATA_OVERRIDE)
+def test_default_user_settings_override(client: IntegrationInstance):
+    """
+    Test that the default user settings are correctly overridden.
+    """
+    # Check shell
+    shell_set = (
+        client.execute(["getent", "passwd", "ubuntu"])
+        .stdout.strip()
+        .split(":")[-1]
+    )
+    if CLOUD_INIT_SOURCE in ["NONE", "IN_PLACE"]:
+        assert (
+            "/bin/sh" == shell_set
+        ), "Shell setting not overriden even though the user is new"
+    else:
+        assert (
+            "/bin/bash" == shell_set
+        ), "Shell setting overriden even though user already exists"
+    # Check password is not locked
+    passwd_status = client.execute(["passwd", "-S", "ubuntu"]).stdout
+    assert re.search(r"^ubuntu\s+P\b", passwd_status)
+    exepected_passwd_hash = "$5$xZ$B2YGGEx2AOf4PeW48KC6.QyT1W2B4rZ9Qbltudtha89"
+    parsed_shadow, _ = fetch_and_parse_etc_shadow(client)
+    assert parsed_shadow["ubuntu"] == exepected_passwd_hash
+
+
+@pytest.mark.skipif(not IS_UBUNTU, reason="Test is Ubuntu specific")
+def test_default_user_settings(client: IntegrationInstance):
+    """
+    This test serves as a "negative control" for
+    test_default_user_settings_override, confirming the default
+    user settings are as expected when not overridden by user-data.
+    """
+    # Check shel
+    shell_set = (
+        client.execute(["getent", "passwd", "ubuntu"])
+        .stdout.strip()
+        .split(":")[-1]
+    )
+    assert "/bin/bash" == shell_set
+    # Check password is not locked
+    passwd_status = client.execute(["passwd", "-S", "ubuntu"]).stdout
+    assert re.search(r"^ubuntu\s+L\b", passwd_status)

tests/integration_tests/util.py

@@ -692,3 +692,19 @@ def get_datetime_from_string(
             )
         )
     return converted_datetime
+
+
+def fetch_and_parse_etc_shadow(client):
+    """Fetch /etc/shadow and parse it into Python data structures
+
+    Returns: ({user: password}, [duplicate, users])
+    """
+    shadow_content = client.read_from_file("/etc/shadow")
+    users = {}
+    dupes = []
+    for line in shadow_content.splitlines():
+        user, encpw = line.split(":")[0:2]
+        if user in users:
+            dupes.append(user)
+        users[user] = encpw
+    return users, dupes

tests/unittests/distros/test_user_data_normalize.py

@@ -196,6 +196,30 @@ def test_users_dict_default_additional(self):
         assert users["bob"]["blah"] is True
         assert users["bob"]["default"] is True
 
+    def test_users_dict_override_default_attribute(self):
+        distro = self._make_distro("ubuntu", bcfg)
+        ug_cfg = {
+            "users": ["default", {"name": "bob", "lock_passwd": False}],
+        }
+        users, _ = self._norm(ug_cfg, distro)
+
+        assert "bob" in users
+        assert "name" not in users["bob"]
+
+        for key, val in bcfg.items():
+            if key == "lock_passwd":
+                # Assert that the default user config is True
+                assert val is True
+                # Assert that the resolved value
+                # matches the passed config: False
+                assert users["bob"][key] is False
+            elif key == "groups":
+                assert users["bob"][key] == ",".join(val)
+            elif key != "name":
+                assert users["bob"][key] == val
+
+        assert users["bob"]["default"] is True
+
     def test_users_dict_extract(self):
         distro = self._make_distro("ubuntu", bcfg)
         ug_cfg = {