Enforce top level keys in autoinstall user data

doc/reference/autoinstall-reference.rst

@@ -3,7 +3,9 @@
 Autoinstall configuration reference manual
 ==========================================
 
-The autoinstall file uses the YAML format. At the top level, it must be a mapping containing the keys described in this document. Unrecognised keys are ignored.
+The autoinstall file uses the YAML format. At the top level, it must be a
+mapping containing the keys described in this document. Unrecognised keys are
+currently ignored, but may cause a run-time failure in future versions.
 
 .. _ai-schema:
 

subiquity/server/autoinstall.py

@@ -13,6 +13,7 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 import logging
+from typing import Optional
 
 log = logging.getLogger("subiquity.server.autoinstall")
 
@@ -24,8 +25,13 @@ class AutoinstallError(Exception):
 class AutoinstallValidationError(AutoinstallError):
     def __init__(
         self,
-        owner: str,
+        section: str,
+        message: Optional[str] = None,
     ):
-        self.message = f"Malformed autoinstall in {owner!r} section"
-        self.owner = owner
+        if not message:
+            self.message: str = f"Malformed autoinstall in {section!r} section"
+        else:
+            self.message: str = message
+
+        self.section: str = section
         super().__init__(self.message)

subiquity/server/server.py

@@ -14,6 +14,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import asyncio
+import copy
 import logging
 import os
 import sys
@@ -475,14 +476,73 @@ async def apply_autoinstall_config(self, context):
             await controller.apply_autoinstall_config()
             await controller.configured()
 
+    def _strip_controller_keys(self, autoinstall_config: dict) -> dict:
+        return_dict: dict = copy.deepcopy(autoinstall_config)
+
+        for controller in self.controllers.instances:
+            return_dict.pop(controller.autoinstall_key, None)
+
+        return return_dict
+
+    def _enforce_top_level_keys(self, base_config: dict) -> dict:
+        # Assumes base_config is the result of stripping away all
+        # controller related sections (via _strip_controller_keys)
+
+        top_level_keys: set[str] = set(base_config.keys())
+        valid_keys: set[str] = set(self.base_schema["properties"].keys())
+        bad_keys: set[str] = top_level_keys - valid_keys
+
+        # Return early if no bad keys
+        if not bad_keys:
+            return base_config
+
+        # If version early enough, only warn
+        version: int = base_config.get("version", None)
+
+        if not version:
+            raise AutoinstallValidationError(
+                section="top-level keys",
+                message="top-level key 'version' is missing",
+            )
+
+        if version <= 1:
+            for key in bad_keys:
+                log.warning(f"Unrecognized top-level autoinstall key {key!r}")
+
+            log.warning(
+                "Unrecognized keys may cause autoinstall to crash in future versions"
+            )
+
+            # Clean out bad keys
+            for key in bad_keys:
+                base_config.pop(key)
+
+            return base_config
+
+        # Otherwise, error
+        else:
+            for key in bad_keys:
+                log.error(f"Unrecognized top-level autoinstall key {key!r}")
+
+            raw_keys = (f"{key!r}" for key in bad_keys)
+            message: str = (
+                f"Unrecognized top-level autoinstall key(s): {', '.join(raw_keys)}",
+            )
+            raise AutoinstallValidationError(
+                section="top-level keys",
+                message=message,
+            )
+
     def validate_autoinstall(self):
         with self.context.child("core_validation", level="INFO"):
             try:
-                jsonschema.validate(self.autoinstall_config, self.base_schema)
+                stripped_config: dict = self._strip_controller_keys(
+                    self.autoinstall_config
+                )
+                base_config: dict = self._enforce_top_level_keys(stripped_config)
+                jsonschema.validate(base_config, self.base_schema)
             except ValidationError as original_exception:
-                # SubiquityServer currently only checks for these sections
-                # of autoinstall. Hardcode until we have better validation.
-                section = "version or interative-sessions"
+                section: str = "top-level keys"
                 new_exception: AutoinstallValidationError = AutoinstallValidationError(
                     section,
                 )

subiquity/server/tests/test_controller.py

@@ -83,7 +83,7 @@ def test_autoinstall_validation(self):
         exception = ctx.exception
 
         # Assert error section is based on autoinstall_key
-        self.assertEquals(exception.owner, "some-key")
+        self.assertEquals(exception.section, "some-key")
 
         # Assert apport report is not created
         # This only checks that controllers do not manually create an apport

subiquity/server/tests/test_server.py

@@ -121,7 +121,9 @@ def test_bogus_autoinstall_argument(self):
         with self.assertRaises(Exception):
             self.server.select_autoinstall()
 
-    def test_early_commands_changes_autoinstall(self):
+    # Only care about changes to autoinstall, not validity
+    @patch("subiquity.server.server.SubiquityServer.validate_autoinstall")
+    def test_early_commands_changes_autoinstall(self, mocked_validator):
         self.server.controllers = Mock()
         self.server.controllers.instances = []
         rootpath = self.path(root_autoinstall_path)
@@ -162,6 +164,17 @@ async def asyncSetUp(self):
         }
         self.server.make_apport_report = Mock()
 
+    # Pseudo Load controllers to avoid patching loading logic for each controller
+    # Only need to access class attributes
+    def pseudo_load_controllers(self):
+        controller_classes = []
+        for prefix in self.server.controllers.controller_names:
+            controller_classes.append(
+                self.server.controllers._get_controller_class(prefix)
+            )
+
+        self.server.controllers.instances = controller_classes
+
     def test_valid_schema(self):
         """Test that the expected autoinstall JSON schema is valid"""
 
@@ -194,6 +207,67 @@ async def test_autoinstall_validation__no_error_report(self):
 
         self.server.make_apport_report.assert_not_called()
 
+    @patch("subiquity.server.server.log")
+    def test_autoinstall_validation__enforce_top_level_keys(self, log_mock):
+        """Test strict top level keys"""
+
+        # Reset base schema
+        self.server.base_schema = SubiquityServer.base_schema
+
+        # "apt" should be known by the MirrorController and not considered
+        # by the server's validation
+        bad_ai_data = {
+            "version": 1,
+            "apt": "Invalid but deferred",
+            "literally-anything": "lmao",
+        }
+
+        self.server.autoinstall_config = bad_ai_data
+
+        # Load the controllers
+        self.pseudo_load_controllers()
+
+        # OK in Version 1 but ensure warnings
+        self.server.validate_autoinstall()
+        log_mock.warning.assert_called()
+
+        # Not OK in Versions >=2
+        bad_ai_data["version"] = 2
+        self.server.autoinstall_config = bad_ai_data
+
+        log_mock.reset()
+
+        with self.assertRaises(AutoinstallValidationError) as ctx:
+            self.server.validate_autoinstall()
+
+        exception = ctx.exception
+
+        self.assertIn("literally-anything", str(exception))
+        self.assertNotIn("apt", str(exception))
+
+        log_mock.error.assert_called()
+
+    def test_autoinstall__strip_controller_keys(self):
+        """Test only controller keys are stripped"""
+
+        # Mixed data: Has base sections, controller section, and a bad key
+        autoinstall_config = {
+            "version": 1,  # Should stay
+            "interactive-sections": ["identity"],  # Should stay
+            "apt": "...",  # Should be stripped
+            "invalid_key": "...",  # Should stay
+        }
+
+        # Load the controllers
+        self.pseudo_load_controllers()
+
+        result = self.server._strip_controller_keys(autoinstall_config)
+
+        self.assertIn("version", result)
+        self.assertIn("interactive-sections", result)
+        self.assertNotIn("apt", result)
+        self.assertIn("invalid_key", result)
+
 
 class TestMetaController(SubiTestCase):
     async def test_interactive_sections_not_present(self):