Merge pull request #1928 from dbungert/allow-skip-identity

Allow skip identity

subiquity/models/subiquity.py

@@ -207,7 +207,7 @@ def __init__(self, root, hub, install_model_names, postinstall_model_names):
         self.timezone = TimeZoneModel()
         self.ubuntu_pro = UbuntuProModel()
         self.updates = UpdatesModel()
-        self.userdata = {}
+        self.userdata = None
 
         self._confirmation = asyncio.Event()
         self._confirmation_task = None
@@ -378,6 +378,13 @@ def _cloud_init_config(self):
                 user_info["ssh_authorized_keys"] = self.ssh.authorized_keys
             config["users"] = [user_info]
         else:
+            if self.userdata is None:
+                config["users"] = []
+                if self.ssh.authorized_keys:
+                    log.warning(
+                        "likely configuration error: "
+                        "authorized_keys supplied but no known user login"
+                    )
             if self.ssh.authorized_keys:
                 config["ssh_authorized_keys"] = self.ssh.authorized_keys
         if self.ssh.install_server:
@@ -388,7 +395,8 @@ def _cloud_init_config(self):
                 merge_config(config, model.make_cloudconfig())
         for package in self.cloud_init_packages:
             merge_config(config, {"packages": list(self.cloud_init_packages)})
-        merge_cloud_init_config(config, self.userdata)
+        if self.userdata is not None:
+            merge_cloud_init_config(config, self.userdata)
         if lsb_release()["release"] not in ("20.04", "22.04"):
             config.setdefault("write_files", []).append(CLOUDINIT_DISABLE_AFTER_INSTALL)
         self.validate_cloudconfig_schema(data=config, data_source="system install")

subiquity/models/tests/test_subiquity.py

@@ -411,3 +411,112 @@ def __init__(self, schema_errors=(), schema_deprecations=()):
                 " type 'array'"
             )
             self.assertEqual(expected_error, str(ctx.exception))
+
+
+class TestUserCreationFlows(unittest.IsolatedAsyncioTestCase):
+    """live-server and desktop have a key behavior difference: desktop will
+    permit user creation on first boot, while server will do no such thing.
+    When combined with documented autoinstall behaviors for the `identity`
+    section and allowing `user-data` to mean that the `identity` section may be
+    skipped, the following use cases need to be supported:
+
+    1. PASS - interactive, UI triggers user creation (`identity_POST` called)
+    2. PASS - interactive, if `identity` `mark_configured`, create nothing
+    3. in autoinstall, supply an `identity` section
+      a. PASS - and omit `user-data`
+      b. PASS - and supply `user-data` but no `user-data.users`
+      c. PASS - and supply `user-data` including `user-data.users`
+    4. in autoinstall, omit an `identity` section
+      a. and omit `user-data`
+        1. FAIL - live-server - failure mode is autoinstall schema validation
+        2. PASS - desktop
+      b. PASS - and supply `user-data` but no `user-data.users`
+        - cloud-init defaults
+      c. PASS - and supply `user-data` including `user-data.users`
+
+    The distinction of a user being created by autoinstall or not is not
+    visible here, so some interactive and autoinstall test cases are equivalent
+    at this abstraction layer (and are merged in the actual tests)."""
+
+    def setUp(self):
+        install = ModelNames(set())
+        postinstall = ModelNames({"userdata"})
+        self.model = SubiquityModel("test", MessageHub(), install, postinstall)
+        self.user = dict(name="user", passwd="passw0rd")
+        self.user_identity = IdentityData(
+            username=self.user["name"], crypted_password=self.user["passwd"]
+        )
+        self.foobar = dict(name="foobar", passwd="foobarpassw0rd")
+
+    def assertDictSubset(self, expected, actual):
+        for key in expected.keys():
+            msg = f"expected[{key}] != actual[{key}]"
+            self.assertEqual(expected[key], actual[key], msg)
+
+    def test_create_user_cases_1_3a(self):
+        self.assertIsNone(self.model.userdata)
+        self.model.identity.add_user(self.user_identity)
+        cloud_cfg = self.model._cloud_init_config()
+        [actual] = cloud_cfg["users"]
+        self.assertDictSubset(self.user, actual)
+
+    def test_assert_no_default_user_cases_2_4a2(self):
+        self.assertIsNone(self.model.userdata)
+        cloud_cfg = self.model._cloud_init_config()
+        self.assertEqual([], cloud_cfg["users"])
+
+    def test_create_user_but_no_merge_case_3b(self):
+        # near identical to cases 1 / 3a but user-data is present in
+        # autoinstall, however this doesn't change the outcome.
+        self.model.userdata = {}
+        self.model.identity.add_user(self.user_identity)
+        cloud_cfg = self.model._cloud_init_config()
+        [actual] = cloud_cfg["users"]
+        self.assertDictSubset(self.user, actual)
+
+    def test_create_user_and_merge_case_3c(self):
+        # now we have more user info to merge in
+        self.model.userdata = {"users": ["default", self.foobar]}
+        self.model.identity.add_user(self.user_identity)
+        cloud_cfg = self.model._cloud_init_config()
+        for actual in cloud_cfg["users"]:
+            if isinstance(actual, str):
+                self.assertEqual("default", actual)
+            else:
+                if actual["name"] == self.user["name"]:
+                    expected = self.user
+                else:
+                    expected = self.foobar
+                self.assertDictSubset(expected, actual)
+
+    def test_create_user_and_merge_case_3c_empty(self):
+        # another merge case, but a merge of an empty list, so we
+        # have just supplied the `identity` user with extra steps
+        self.model.userdata = {"users": []}
+        self.model.identity.add_user(self.user_identity)
+        cloud_cfg = self.model._cloud_init_config()
+        [actual] = cloud_cfg["users"]
+        self.assertDictSubset(self.user, actual)
+
+    # 4a1 fails before we get here, see TestControllerUserCreationFlows in
+    # subiquity/server/controllers/tests/test_identity.py for details.
+
+    def test_create_nothing_case_4b(self):
+        self.model.userdata = {}
+        cloud_cfg = self.model._cloud_init_config()
+        self.assertNotIn("users", cloud_cfg)
+
+    def test_create_only_merge_4c(self):
+        self.model.userdata = {"users": ["default", self.foobar]}
+        cloud_cfg = self.model._cloud_init_config()
+        for actual in cloud_cfg["users"]:
+            if isinstance(actual, str):
+                self.assertEqual("default", actual)
+            else:
+                self.assertDictSubset(self.foobar, actual)
+
+    def test_create_only_merge_4c_empty(self):
+        # explicitly saying no (additional) users, thank you very much
+        self.model.userdata = {"users": []}
+        cloud_cfg = self.model._cloud_init_config()
+        self.assertEqual([], cloud_cfg["users"])

subiquity/server/controllers/identity.py

@@ -88,7 +88,9 @@ async def apply_autoinstall_config(self, context=None):
             return
         if self.app.base_model.target is None:
             return
-        raise Exception("no identity data provided")
+        if self.app.base_model.source.current.variant != "server":
+            return
+        raise Exception("neither identity nor user-data provided")
 
     def make_autoinstall(self):
         if self.model.user is None:

subiquity/server/controllers/tests/test_identity.py

@@ -18,6 +18,7 @@
 
 from subiquity.server.controllers.identity import IdentityController
 from subiquitycore.tests import SubiTestCase
+from subiquitycore.tests.mocks import make_app
 
 
 class TestIdentityController(SubiTestCase):
@@ -29,3 +30,22 @@ def test_valid_schema(self):
         )
 
         JsonValidator.check_schema(IdentityController.autoinstall_schema)
+
+
+class TestControllerUserCreationFlows(SubiTestCase):
+    # TestUserCreationFlows has more information about user flow use cases.
+    # See subiquity/models/tests/test_subiquity.py for details.
+    def setUp(self):
+        self.app = make_app()
+        self.ic = IdentityController(self.app)
+        self.ic.model.user = None
+
+    async def test_server_requires_identity_case_4a1(self):
+        self.app.base_model.source.current.variant = "server"
+        with self.assertRaises(Exception):
+            await self.ic.apply_autoinstall_config()
+
+    async def test_desktop_does_not_require_identity_case_4a2(self):
+        self.app.base_model.source.current.variant = "desktop"
+        await self.ic.apply_autoinstall_config()
+        # should not raise

subiquity/server/controllers/tests/test_userdata.py

@@ -33,6 +33,7 @@ def SchemaProblem(x, y):
 class TestUserdataController(unittest.TestCase):
     def setUp(self):
         self.controller = UserdataController(make_app())
+        self.controller.model = None
 
     def test_load_autoinstall_data(self):
         with self.subTest("Valid user-data resets userdata model"):
@@ -69,3 +70,15 @@ def test_valid_schema(self):
         )
 
         JsonValidator.check_schema(UserdataController.autoinstall_schema)
+
+    def test_load_none(self):
+        self.controller.load_autoinstall_data(None)
+        self.assertIsNone(self.controller.model)
+
+    def test_load_empty(self):
+        self.controller.load_autoinstall_data({})
+        self.assertEqual({}, self.controller.model)
+
+    def test_load_some(self):
+        self.controller.load_autoinstall_data({"stuff": "things"})
+        self.assertEqual({"stuff": "things"}, self.controller.model)

subiquity/server/controllers/userdata.py

@@ -23,19 +23,20 @@
 class UserdataController(NonInteractiveController):
     model_name = "userdata"
     autoinstall_key = "user-data"
-    autoinstall_default = {}
+    autoinstall_default = None
     autoinstall_schema = {
         "type": "object",
     }
 
     def load_autoinstall_data(self, data):
-        self.model.clear()
+        if data is None:
+            return
         if data:
             self.app.base_model.validate_cloudconfig_schema(
                 data=data,
                 data_source="autoinstall.user-data",
             )
-        self.model.update(data)
+        self.app.base_model.userdata = self.model = data.copy()
 
     def make_autoinstall(self):
-        return self.app.base_model.userdata
+        return self.app.base_model.userdata or {}