diff --git a/subiquity/cloudinit.py b/subiquity/cloudinit.py
index 168d2f763..bd0e03199 100644
--- a/subiquity/cloudinit.py
+++ b/subiquity/cloudinit.py
@@ -4,7 +4,9 @@
 import json
 import logging
 import re
-from typing import Optional
+from collections.abc import Awaitable
+from subprocess import CompletedProcess
+from typing import Any, Optional
 
 from subiquitycore.utils import arun_command, run_command
 
@@ -67,6 +69,56 @@ def read_legacy_status(stream):
     return None
 
 
+def read_recoverable_errors(
+    stream: bytes | str,
+    level: Optional[str] = None,
+) -> list[str]:
+    """Retrieve recoverable errors from cloud-init status."""
+
+    try:
+        status: dict[str, Any] = json.loads(stream)
+    except json.JSONDecodeError:
+        return []
+
+    errors_by_level: dict[str, list[str]] = status.get("recoverable_errors", {})
+
+    if level is not None:
+        return errors_by_level.get(level, [])
+
+    all_errors: list[str] = []
+    for _level in errors_by_level:
+        all_errors.extend(errors_by_level[_level])
+
+    return all_errors
+
+
+async def get_schema_failure_sources() -> list[str]:
+    """Retrieve the keys causing schema failure."""
+
+    cmd: list[str] = ["cloud-init", "schema", "--system"]
+    status_coro: Awaitable = arun_command(cmd)
+    try:
+        sp: CompletedProcess = await asyncio.wait_for(status_coro, 10)
+    except asyncio.TimeoutError:
+        return []
+
+    error: str = sp.stderr  # Relies on arun_command decoding to utf-8 str by default
+
+    # Matches:
+    # ('some-key' was unexpected)
+    # ('some-key', 'another-key' were unexpected)
+    pattern = r"\((?P<args>'[\w\-]+'(,\s'[\w\-]+')*)+ (?:was|were) unexpected\)"
+    search_result = re.search(pattern, error)
+
+    if search_result is None:
+        return []
+
+    args_list: list[str] = search_result.group("args").split(", ")
+    no_quotes: list[str] = [arg.strip("'") for arg in args_list]
+
+    return no_quotes
+
+
 async def cloud_init_status_wait() -> (bool, Optional[str]):
     """Wait for cloud-init completion, and return if timeout ocurred and best
     available status information.
@@ -86,3 +138,31 @@ async def cloud_init_status_wait() -> (bool, Optional[str]):
     else:
         status = read_legacy_status(sp.stdout)
     return (True, status)
+
+
+async def validate_cloud_init_schema() -> tuple[bool, Optional[list[str]]]:
+    """Check for cloud-init schema errors.
+
+    Returns True if validated OK or can't tell. Otherwise, False and
+    a list of bad keys found.
+    Requires cloud-init supporting recoverable errors and extended status.
+    :return: tuple of (ok, list of bad keys or None)
+    """
+
+    if not supports_recoverable_errors():
+        return (True, None)
+
+    # Recoverable errors support implies extended status support
+    cmd: list[str] = ["cloud-init", "status", "--format=json"]
+    status_coro: Awaitable = arun_command(cmd)
+    try:
+        sp: CompletedProcess = await asyncio.wait_for(status_coro, 10)
+    except asyncio.TimeoutError:
+        return (True, None)
+
+    errors: list[str] = read_recoverable_errors(sp.stdout)
+
+    if not errors or not any("schema errors" in err for err in errors):
+        return (True, None)
+
+    return (False, await get_schema_failure_sources())
diff --git a/subiquity/server/server.py b/subiquity/server/server.py
index 6f68be4fa..a77a797d1 100644
--- a/subiquity/server/server.py
+++ b/subiquity/server/server.py
@@ -28,7 +28,11 @@
 from jsonschema.exceptions import ValidationError
 from systemd import journal
 
-from subiquity.cloudinit import cloud_init_status_wait, get_host_combined_cloud_config
+from subiquity.cloudinit import (
+    cloud_init_status_wait,
+    get_host_combined_cloud_config,
+    validate_cloud_init_schema,
+)
 from subiquity.common.api.server import bind, controller_for_request
 from subiquity.common.apidef import API
 from subiquity.common.errorreport import ErrorReport, ErrorReporter, ErrorReportKind
@@ -43,7 +47,7 @@
     PasswordKind,
 )
 from subiquity.models.subiquity import ModelNames, SubiquityModel
-from subiquity.server.autoinstall import AutoinstallValidationError
+from subiquity.server.autoinstall import AutoinstallError, AutoinstallValidationError
 from subiquity.server.controller import SubiquityController
 from subiquity.server.dryrun import DRConfig
 from subiquity.server.errors import ErrorController
@@ -723,7 +727,66 @@ async def start_site(self, runner: web.AppRunner):
     def base_relative(self, path):
         return os.path.join(self.base_model.root, path)
 
-    def load_cloud_config(self):
+    @with_context(name="extract_autoinstall")
+    async def _extract_autoinstall_from_cloud_config(
+        self,
+        *,
+        cloud_cfg: dict[str, Any],
+        context: Context,
+    ) -> dict[str, Any]:
+        """Extract autoinstall passed via cloud config."""
+
+        # Not really is-install-context but set to force event reporting
+        context.set("is-install-context", True)
+        context.enter()  # publish start event
+
+        if "autoinstall" not in cloud_cfg:
+            return {}
+
+        # Get the explicitly passed autoinstall data
+        cfg: dict[str, Any] = cloud_cfg["autoinstall"]
+
+        # Check for keys mistakenly placed in regular cloud init and add
+        # them to the autoinstall data
+        ok, bad_keys = await validate_cloud_init_schema()
+
+        if ok:
+            return cfg
+        else:
+            raw_keys: list[str] = [f"{key!r}" for key in bad_keys]
+            context.warning(
+                f"cloud-init schema validation failure for: {', '.join(raw_keys)}",
+                log=log,
+            )
+
+            # Only filter on the bad_keys
+            potential_autoinstall: dict[str, Any] = dict(
+                ((key, cloud_cfg[key]) for key in bad_keys)
+            )
+            autoinstall, other = self.filter_autoinstall(potential_autoinstall)
+
+            if len(autoinstall) == 0:
+                log.debug(
+                    "No autoinstall keys found among bad cloud config. Continuing."
+                )
+
+                return cfg
+
+            for key in autoinstall:
+                context.error(
+                    f"{key!r} is valid autoinstall but not found under 'autoinstall'",
+                    log=log,
+                )
+
+            raise AutoinstallError(
+                (
+                    "Misplaced autoinstall directives resulted in a cloud-init "
+                    "schema validation failure."
+                )
+            )
+
+    @with_context()
+    async def load_cloud_config(self, *, context: Context):
         # cloud-init 23.3 introduced combined-cloud-config, which helps to
         # prevent subiquity from having to go load cloudinit modules.
         # This matters because a downgrade pickle deserialization issue may
@@ -755,13 +818,16 @@ def load_cloud_config(self):
             users = ug_util.normalize_users_groups(cloud_cfg, cloud.distro)[0]
             self.installer_user_name = ug_util.extract_default(users)[0]
 
-        if "autoinstall" in cloud_cfg:
+        autoinstall = await self._extract_autoinstall_from_cloud_config(
+            cloud_cfg=cloud_cfg, context=context
+        )
+
+        if autoinstall != {}:
             log.debug("autoinstall found in cloud-config")
-            cfg = cloud_cfg["autoinstall"]
             target = self.base_relative(cloud_autoinstall_path)
             from cloudinit import safeyaml
 
-            write_file(target, safeyaml.dumps(cfg))
+            write_file(target, safeyaml.dumps(autoinstall))
         else:
             log.debug("no autoinstall found in cloud-config")
 
@@ -778,7 +844,7 @@ async def wait_for_cloudinit(self):
             if "disabled" in status:
                 log.debug("Skip cloud-init autoinstall, cloud-init is disabled")
             else:
-                self.load_cloud_config()
+                await self.load_cloud_config()
 
     def select_autoinstall(self):
         # precedence
diff --git a/subiquity/server/tests/test_server.py b/subiquity/server/tests/test_server.py
index a6a199f3e..853d83a4c 100644
--- a/subiquity/server/tests/test_server.py
+++ b/subiquity/server/tests/test_server.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 copy
 import os
 import shlex
 from typing import Any
@@ -22,7 +23,7 @@
 from jsonschema.validators import validator_for
 
 from subiquity.common.types import NonReportableError, PasswordKind
-from subiquity.server.autoinstall import AutoinstallValidationError
+from subiquity.server.autoinstall import AutoinstallError, AutoinstallValidationError
 from subiquity.server.nonreportable import NonReportableException
 from subiquity.server.server import (
     NOPROBERARG,
@@ -357,6 +358,82 @@ async def test_autoinstall_validation__filter_autoinstall(self, config, good, ba
         self.assertEqual(valid, good)
         self.assertEqual(invalid, bad)
 
+    @parameterized.expand(
+        (
+            # Has valid cloud config, no autoinstall
+            ({"valid-cloud": "data"}, {}, False),
+            # Has valid cloud config and autoinstall, no valid ai in cloud cfg
+            (
+                {
+                    "valid-cloud": "data",
+                    "autoinstall": {
+                        "version": 1,
+                        "interactive-sections": ["identity"],
+                    },
+                },
+                {
+                    "version": 1,
+                    "interactive-sections": ["identity"],
+                },
+                False,
+            ),
+            # Has valid autoinstall directive in cloud config
+            (
+                {
+                    "interactive-sections": "data",
+                    "autoinstall": {
+                        "version": 1,
+                        "interactive-sections": ["identity"],
+                    },
+                },
+                None,  # Doesn't return
+                True,
+            ),
+            # Has invalid cloud config key but is not valid autoinstall either
+            (
+                {
+                    "something-else": "data",
+                    "autoinstall": {
+                        "version": 1,
+                        "interactive-sections": ["identity"],
+                    },
+                },
+                {
+                    "version": 1,
+                    "interactive-sections": ["identity"],
+                },
+                False,
+            ),
+        )
+    )
+    async def test_autoinstall_from_cloud_config(self, cloud_cfg, expected, throws):
+        """Test autoinstall extract from cloud config."""
+
+        self.server.base_schema = SubiquityServer.base_schema
+        self.pseudo_load_controllers()
+
+        cloud_data = copy.copy(cloud_cfg)
+        cloud_data.pop("valid-cloud", None)
+        cloud_data.pop("autoinstall", None)
+
+        with patch("subiquity.server.server.validate_cloud_init_schema") as val_mock:
+            if len(cloud_data) == 0:
+                val_mock.return_value = (True, None)
+            else:
+                val_mock.return_value = (False, set(cloud_data.keys()))
+
+            if throws:
+                with self.assertRaises(AutoinstallError):
+                    cfg = await self.server._extract_autoinstall_from_cloud_config(
+                        cloud_cfg=cloud_cfg
+                    )
+            else:
+                cfg = await self.server._extract_autoinstall_from_cloud_config(
+                    cloud_cfg=cloud_cfg
+                )
+
+                self.assertEqual(cfg, expected)
+
 
 class TestMetaController(SubiTestCase):
     async def test_interactive_sections_not_present(self):
diff --git a/subiquity/tests/test_cloudinit.py b/subiquity/tests/test_cloudinit.py
index 8f4b3ee6f..b74e5e64b 100644
--- a/subiquity/tests/test_cloudinit.py
+++ b/subiquity/tests/test_cloudinit.py
@@ -14,6 +14,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import asyncio
+import json
 from subprocess import CompletedProcess
 from unittest import skipIf
 from unittest.mock import Mock, patch
@@ -21,10 +22,13 @@
 from subiquity.cloudinit import (
     cloud_init_status_wait,
     cloud_init_version,
+    get_schema_failure_sources,
     read_json_extended_status,
     read_legacy_status,
+    read_recoverable_errors,
     supports_format_json,
     supports_recoverable_errors,
+    validate_cloud_init_schema,
 )
 from subiquitycore.tests import SubiTestCase
 from subiquitycore.tests.parameterized import parameterized
@@ -132,3 +136,115 @@ async def test_cloud_init_status_wait_legacy(self, m_wait_for):
             args=[], returncode=0, stdout="status: done\n"
         )
         self.assertEqual((True, "done"), await cloud_init_status_wait())
+
+
+class TestCloudInitUtil(SubiTestCase):
+    @parameterized.expand(
+        (
+            ("WARNING", ["warning"]),
+            ("DEPRECATED", ["deprecated"]),
+            ("ERROR", ["error"]),
+            ("CRITICAL", ["critical"]),
+            (None, ["warning", "deprecated", "error", "critical"]),
+        )
+    )
+    def test_read_recoverable_errors(self, level, expected):
+        command_output = {
+            "recoverable_errors": {
+                "WARNING": ["warning"],
+                "DEPRECATED": ["deprecated"],
+                "ERROR": ["error"],
+                "CRITICAL": ["critical"],
+            }
+        }
+
+        errors = read_recoverable_errors(json.dumps(command_output), level=level)
+        self.assertEqual(errors, expected)
+
+
+class TestCloudInitSchemaValidation(SubiTestCase):
+    @parameterized.expand(
+        (
+            (
+                (
+                    "  Error: Cloud config schema errors: : Additional "
+                    "properties are not allowed ('bad-key', 'late-commands' "
+                    "were unexpected)\n\nError: Invalid schema: user-data\n\n"
+                ),
+                ["bad-key", "late-commands"],
+            ),
+            (
+                (
+                    "  Error: Cloud config schema errors: : Additional "
+                    "properties are not allowed ('bad-key' "
+                    "was unexpected)\n\nError: Invalid schema: user-data\n\n"
+                ),
+                ["bad-key"],
+            ),
+            ("('key_1', 'key-2', 'KEY3' were unexpected)", ["key_1", "key-2", "KEY3"]),
+        )
+    )
+    async def test_get_schema_failure_sources(self, msg, expected):
+        """Test 1 or more keys are extracted correctly."""
+
+        with (
+            patch("subiquity.cloudinit.arun_command", new=Mock()),
+            patch("subiquity.cloudinit.asyncio.wait_for") as wait_for_mock,
+        ):
+            wait_for_mock.return_value = CompletedProcess(
+                args=[], returncode=1, stderr=msg
+            )
+
+            bad_keys = await get_schema_failure_sources()
+
+        self.assertEqual(bad_keys, expected)
+
+    @patch("subiquity.cloudinit.arun_command", new=Mock())
+    @patch("subiquity.cloudinit.asyncio.wait_for")
+    async def test_get_schema_failure_malformed(self, wait_for_mock):
+        """Test graceful failure if output changes."""
+
+        error_msg = "('key_1', 'key-2', 'KEY3', were unexpected)"
+
+        wait_for_mock.return_value = CompletedProcess(
+            args=[], returncode=1, stderr=error_msg
+        )
+
+        bad_keys = await get_schema_failure_sources()
+
+        self.assertEqual(bad_keys, [])
+
+    async def test_no_support(self):
+        with patch("subiquity.cloudinit.cloud_init_version") as civ:
+            civ.return_value = "23.3"  # Starts support as 23.4
+            self.assertEqual((True, None), await validate_cloud_init_schema())
+
+    @patch("subiquity.cloudinit.supports_recoverable_errors", return_value=True)
+    @patch("subiquity.cloudinit.arun_command", new=Mock())
+    @patch("subiquity.cloudinit.asyncio.wait_for")
+    @patch("subiquity.cloudinit.read_recoverable_errors")
+    async def test_errors_but_not_schema(self, errors_mock, wait_for_mock, recov_mock):
+        errors_mock.return_value = ["Error but not a schema one"]
+
+        self.assertEqual((True, None), await validate_cloud_init_schema())
+
+    @patch("subiquity.cloudinit.supports_recoverable_errors", return_value=True)
+    @patch("subiquity.cloudinit.arun_command", new=Mock())
+    @patch("subiquity.cloudinit.asyncio.wait_for")
+    @patch("subiquity.cloudinit.read_recoverable_errors")
+    @patch("subiquity.cloudinit.get_schema_failure_sources")
+    async def test_validate_cloud_init_schema(
+        self, sources_mock, errors_mock, wait_for_mock, recov_mock
+    ):
+        errors_mock.return_value = [
+            (
+                "Invalid cloud-config provided: Please run 'sudo cloud-init "
+                "schema --system' to see the schema errors."
+            ),
+            "Other errors",
+        ]
+
+        mock_keys = ["key1", "key2"]
+        sources_mock.return_value = mock_keys
+
+        self.assertEqual((False, mock_keys), await validate_cloud_init_schema())