Fix circular import

homeassistant/bootstrap.py

@@ -46,7 +46,7 @@ from .components import (
     file_upload as file_upload_pre_import,  # noqa: F401
     group as group_pre_import,  # noqa: F401
     history as history_pre_import,  # noqa: F401
-    http,  # not named pre_import since it has requirements
+    http as http_import,  # noqa: F401 - not named pre_import since it has requirements
     image_upload as image_upload_import,  # noqa: F401 - not named pre_import since it has requirements
     logbook as logbook_pre_import,  # noqa: F401
     lovelace as lovelace_pre_import,  # noqa: F401
@@ -403,12 +403,7 @@ async def async_setup_hass(
         _LOGGER.info("Starting in recovery mode")
         hass.config.recovery_mode = True
 
-        http_conf = (await http.async_get_last_config(hass)) or {}
-
-        await async_from_config_dict(
-            {"recovery_mode": {}, "http": http_conf},
-            hass,
-        )
+        await async_from_config_dict({"recovery_mode": {}}, hass)
 
     if runtime_config.open_ui:
         hass.add_job(open_hass_ui, hass)

homeassistant/components/http/__init__.py

@@ -12,7 +12,7 @@ from pathlib import Path
 import socket
 import ssl
 from tempfile import NamedTemporaryFile
-from typing import Any, Final, TypedDict, cast
+from typing import Any, Final, cast
 
 from aiohttp import web
 from aiohttp.abc import AbstractStreamWriter
@@ -37,7 +37,7 @@ from homeassistant.const import (
 )
 from homeassistant.core import Event, HomeAssistant, callback
 from homeassistant.exceptions import HomeAssistantError
-from homeassistant.helpers import config_validation as cv, issue_registry as ir, storage
+from homeassistant.helpers import config_validation as cv, issue_registry as ir
 from homeassistant.helpers.hassio import is_hassio
 from homeassistant.helpers.http import (
     KEY_ALLOW_CONFIGURED_CORS,
@@ -60,7 +60,29 @@ from homeassistant.util.json import json_loads
 
 from .auth import async_setup_auth
 from .ban import setup_bans
-from .const import DOMAIN, KEY_HASS_REFRESH_TOKEN_ID, KEY_HASS_USER  # noqa: F401
+from .config import ConfData, async_load_config  # noqa: F401
+from .const import (  # noqa: F401
+    CONF_BASE_URL,
+    CONF_CORS_ORIGINS,
+    CONF_IP_BAN_ENABLED,
+    CONF_LOGIN_ATTEMPTS_THRESHOLD,
+    CONF_SERVER_HOST,
+    CONF_SERVER_PORT,
+    CONF_SSL_CERTIFICATE,
+    CONF_SSL_KEY,
+    CONF_SSL_PEER_CERTIFICATE,
+    CONF_SSL_PROFILE,
+    CONF_TRUSTED_PROXIES,
+    CONF_USE_X_FORWARDED_FOR,
+    CONF_USE_X_FRAME_OPTIONS,
+    DEFAULT_CORS,
+    DOMAIN,
+    KEY_HASS_REFRESH_TOKEN_ID,
+    KEY_HASS_USER,
+    NO_LOGIN_ATTEMPT_THRESHOLD,
+    SSL_INTERMEDIATE,
+    SSL_MODERN,
+)
 from .cors import setup_cors
 from .decorators import require_admin  # noqa: F401
 from .forwarded import async_setup_forwarded
@@ -70,38 +92,13 @@ from .security_filter import setup_security_filter
 from .static import CACHE_HEADERS, CachingStaticResource
 from .web_runner import HomeAssistantTCPSite, HomeAssistantUnixSite
 
-CONF_SERVER_HOST: Final = "server_host"
-CONF_SERVER_PORT: Final = "server_port"
-CONF_BASE_URL: Final = "base_url"
-CONF_SSL_CERTIFICATE: Final = "ssl_certificate"
-CONF_SSL_PEER_CERTIFICATE: Final = "ssl_peer_certificate"
-CONF_SSL_KEY: Final = "ssl_key"
-CONF_CORS_ORIGINS: Final = "cors_allowed_origins"
-CONF_USE_X_FORWARDED_FOR: Final = "use_x_forwarded_for"
-CONF_USE_X_FRAME_OPTIONS: Final = "use_x_frame_options"
-CONF_TRUSTED_PROXIES: Final = "trusted_proxies"
-CONF_LOGIN_ATTEMPTS_THRESHOLD: Final = "login_attempts_threshold"
-CONF_IP_BAN_ENABLED: Final = "ip_ban_enabled"
-CONF_SSL_PROFILE: Final = "ssl_profile"
-
-SSL_MODERN: Final = "modern"
-SSL_INTERMEDIATE: Final = "intermediate"
-
 _LOGGER: Final = logging.getLogger(__name__)
 
 DEFAULT_DEVELOPMENT: Final = "0"
-# Cast to be able to load custom cards.
-# My to be able to check url and version info.
-DEFAULT_CORS: Final[list[str]] = ["https://cast.home-assistant.io"]
-NO_LOGIN_ATTEMPT_THRESHOLD: Final = -1
 
 MAX_CLIENT_SIZE: Final = 1024**2 * 16
 MAX_LINE_SIZE: Final = 24570
 
-STORAGE_KEY: Final = DOMAIN
-STORAGE_VERSION: Final = 1
-SAVE_DELAY: Final = 180
-
 _HAS_IPV6 = hasattr(socket, "AF_INET6")
 _DEFAULT_BIND = ["0.0.0.0", "::"] if _HAS_IPV6 else ["0.0.0.0"]
 
@@ -154,30 +151,6 @@ _STATIC_CLASSES = {
 }
 
 
-class ConfData(TypedDict, total=False):
-    """Typed dict for config data."""
-
-    server_host: list[str]
-    server_port: int
-    base_url: str
-    ssl_certificate: str
-    ssl_peer_certificate: str
-    ssl_key: str
-    cors_allowed_origins: list[str]
-    use_x_forwarded_for: bool
-    use_x_frame_options: bool
-    trusted_proxies: list[IPv4Network | IPv6Network]
-    login_attempts_threshold: int
-    ip_ban_enabled: bool
-    ssl_profile: str
-
-
-async def async_get_last_config(hass: HomeAssistant) -> dict[str, Any] | None:
-    """Return the last known working config."""
-    store = storage.Store[dict[str, Any]](hass, STORAGE_VERSION, STORAGE_KEY)
-    return await store.async_load()
-
-
 class ApiConfig:
     """Configuration settings for API server."""
 
@@ -201,10 +174,13 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     # we import aiohttp_fast_zlib
     (await async_import_module(hass, "aiohttp_fast_zlib")).enable()
 
-    conf: ConfData | None = config.get(DOMAIN)
+    # Local import to avoid a circular import with websocket_api,
+    # which imports from homeassistant.components.http at module load.
+    from .websocket_api import async_register_websocket_commands  # noqa: PLC0415
 
-    if conf is None:
-        conf = cast(ConfData, HTTP_SCHEMA({}))
+    conf = await async_load_config(hass, config)
+
+    async_register_websocket_commands(hass)
 
     if CONF_SERVER_HOST in conf and is_hassio(hass):
         issue_id = "server_host_deprecated_hassio"
@@ -271,9 +247,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         """Start the server."""
         with async_start_setup(hass, integration="http", phase=SetupPhases.SETUP):
             hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_server)
-            # We already checked it's not None.
-            assert conf is not None
-            await start_http_server_and_save_config(hass, dict(conf), server)
+            await server.start()
 
     async_when_setup_or_start(hass, "frontend", start_server)
 
@@ -709,23 +683,3 @@ class HomeAssistantHTTP:
             await self.site.stop()
         if self.runner is not None:
             await self.runner.cleanup()
-
-
-async def start_http_server_and_save_config(
-    hass: HomeAssistant, conf: dict, server: HomeAssistantHTTP
-) -> None:
-    """Startup the http server and save the config."""
-    await server.start()
-
-    # If we are set up successful, we store the HTTP settings for recovery mode.
-    store: storage.Store[dict[str, Any]] = storage.Store(
-        hass, STORAGE_VERSION, STORAGE_KEY
-    )
-
-    if CONF_TRUSTED_PROXIES in conf:
-        conf[CONF_TRUSTED_PROXIES] = [
-            str(cast(IPv4Network | IPv6Network, ip).network_address)
-            for ip in conf[CONF_TRUSTED_PROXIES]
-        ]
-
-    store.async_delay_save(lambda: conf, SAVE_DELAY)

homeassistant/components/http/config.py

@@ -0,0 +1,314 @@
+"""User-managed HTTP configuration store."""
+
+from ipaddress import IPv4Network, IPv6Network, ip_network
+import logging
+from typing import Any, Final, TypedDict, cast
+
+import voluptuous as vol
+
+from homeassistant.const import SERVER_PORT
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import config_validation as cv, issue_registry as ir
+from homeassistant.helpers.storage import Store
+from homeassistant.helpers.typing import ConfigType
+from homeassistant.util.hass_dict import HassKey
+
+from .const import (
+    CONF_BASE_URL,
+    CONF_CORS_ORIGINS,
+    CONF_IP_BAN_ENABLED,
+    CONF_LOGIN_ATTEMPTS_THRESHOLD,
+    CONF_SERVER_HOST,
+    CONF_SERVER_PORT,
+    CONF_SSL_CERTIFICATE,
+    CONF_SSL_KEY,
+    CONF_SSL_PEER_CERTIFICATE,
+    CONF_SSL_PROFILE,
+    CONF_TRUSTED_PROXIES,
+    CONF_USE_X_FORWARDED_FOR,
+    CONF_USE_X_FRAME_OPTIONS,
+    DEFAULT_CORS,
+    DOMAIN,
+    NO_LOGIN_ATTEMPT_THRESHOLD,
+    SSL_INTERMEDIATE,
+    SSL_MODERN,
+)
+
+
+class ConfData(TypedDict, total=False):
+    """Typed dict for config data."""
+
+    server_host: list[str]
+    server_port: int
+    base_url: str
+    ssl_certificate: str
+    ssl_peer_certificate: str
+    ssl_key: str
+    cors_allowed_origins: list[str]
+    use_x_forwarded_for: bool
+    use_x_frame_options: bool
+    trusted_proxies: list[IPv4Network | IPv6Network]
+    login_attempts_threshold: int
+    ip_ban_enabled: bool
+    ssl_profile: str
+
+
+_LOGGER = logging.getLogger(__name__)
+
+STORAGE_KEY: Final = DOMAIN
+STORAGE_VERSION: Final = 2
+
+KEY_STABLE: Final = "stable"
+KEY_PENDING: Final = "pending"
+KEY_YAML_MIGRATION_DONE: Final = "yaml_migration_done"
+
+DATA_STORE: HassKey[HTTPConfigStore] = HassKey(STORAGE_KEY)
+
+
+def _ip_network_str(value: Any) -> str:
+    """Validate the value is a valid IP network and return its string form."""
+    return str(ip_network(value))
+
+
+HTTP_STORAGE_SCHEMA: Final = vol.Schema(
+    {
+        vol.Optional(CONF_SERVER_HOST): vol.All(
+            cv.ensure_list, vol.Length(min=1), [cv.string]
+        ),
+        vol.Optional(CONF_SERVER_PORT, default=SERVER_PORT): cv.port,
+        vol.Optional(CONF_SSL_CERTIFICATE): cv.isfile,
+        vol.Optional(CONF_SSL_PEER_CERTIFICATE): cv.isfile,
+        vol.Optional(CONF_SSL_KEY): cv.isfile,
+        vol.Optional(CONF_CORS_ORIGINS, default=DEFAULT_CORS): vol.All(
+            cv.ensure_list, [cv.string]
+        ),
+        vol.Inclusive(CONF_USE_X_FORWARDED_FOR, "proxy"): cv.boolean,
+        vol.Inclusive(CONF_TRUSTED_PROXIES, "proxy"): vol.All(
+            cv.ensure_list, [_ip_network_str]
+        ),
+        vol.Optional(
+            CONF_LOGIN_ATTEMPTS_THRESHOLD, default=NO_LOGIN_ATTEMPT_THRESHOLD
+        ): vol.Any(cv.positive_int, NO_LOGIN_ATTEMPT_THRESHOLD),
+        vol.Optional(CONF_IP_BAN_ENABLED, default=True): cv.boolean,
+        vol.Optional(CONF_SSL_PROFILE, default=SSL_MODERN): vol.In(
+            [SSL_INTERMEDIATE, SSL_MODERN]
+        ),
+        vol.Optional(CONF_USE_X_FRAME_OPTIONS, default=True): cv.boolean,
+    }
+)
+_DEFAULT_CONFIG: Final[ConfData] = cast(ConfData, HTTP_STORAGE_SCHEMA({}))
+
+
+def yaml_config_to_storage(conf: dict[str, Any]) -> dict[str, Any]:
+    """Convert a validated HTTP_SCHEMA config to a JSON-serializable storage dict."""
+    storage_conf: dict[str, Any] = dict(conf)
+    storage_conf.pop(CONF_BASE_URL, None)
+    if CONF_TRUSTED_PROXIES in storage_conf:
+        storage_conf[CONF_TRUSTED_PROXIES] = [
+            str(network) for network in storage_conf[CONF_TRUSTED_PROXIES]
+        ]
+    return storage_conf
+
+
+class _HTTPStoreData(TypedDict):
+    """Data structure for HTTP config storage."""
+
+    stable: ConfData
+    pending: ConfData | None
+    yaml_migration_done: bool
+
+
+class _HTTPStore(Store[_HTTPStoreData]):
+    """Http store."""
+
+    async def _async_migrate_func(
+        self,
+        old_major_version: int,
+        old_minor_version: int,
+        old_data: dict[str, Any],
+    ) -> dict[str, Any]:
+        if old_major_version == 1:
+            # Run the v1 payload through the storage schema so the v2 ``stable``
+            # slot is well-formed (all keys present, values normalised) and the
+            # load step can rely on direct key access.
+            try:
+                stable = dict(HTTP_STORAGE_SCHEMA(old_data))
+            except vol.Invalid:
+                _LOGGER.warning(
+                    "Discarding invalid v1 HTTP config during migration; "
+                    "falling back to defaults"
+                )
+                stable = dict(HTTP_STORAGE_SCHEMA({}))
+            return {
+                KEY_STABLE: stable,
+                KEY_PENDING: None,
+                KEY_YAML_MIGRATION_DONE: False,
+            }
+        return old_data
+
+
+class HTTPConfigStore:
+    """Persist HTTP config as a stable/pending pair.
+
+    ``stable`` holds the last config the user confirmed as working;
+    ``pending`` holds an unconfirmed config the user wants to try on
+    the next start. Normal startup prefers ``pending`` so the new
+    config gets exercised; recovery mode falls back to ``stable`` so
+    Home Assistant can still come up after a bad config.
+    """
+
+    def __init__(self, hass: HomeAssistant) -> None:
+        """Initialize the store."""
+        self._hass = hass
+        self._store = _HTTPStore(
+            hass,
+            STORAGE_VERSION,
+            STORAGE_KEY,
+            private=True,
+            atomic_writes=True,
+        )
+        self._stable: ConfData = _DEFAULT_CONFIG
+        self._pending: ConfData | None = None
+        self._yaml_migration_done = False
+        self._loaded = False
+
+    @property
+    def stable(self) -> ConfData:
+        """Return the last confirmed-working config."""
+        return self._stable
+
+    @property
+    def pending(self) -> ConfData | None:
+        """Return the unconfirmed config awaiting promotion, if any."""
+        return self._pending
+
+    @property
+    def yaml_migration_done(self) -> bool:
+        """Return whether the YAML migration has been completed."""
+        return self._yaml_migration_done
+
+    async def async_load(self) -> None:
+        """Load the stable and pending configs from disk."""
+        if self._loaded:
+            return
+        raw = await self._store.async_load()
+        if raw is not None:
+            self._stable = raw[KEY_STABLE]
+            self._pending = raw[KEY_PENDING]
+            self._yaml_migration_done = raw[KEY_YAML_MIGRATION_DONE]
+        self._loaded = True
+
+    async def async_set_pending(self, config: ConfData | None) -> None:
+        """Set (or clear) the pending config."""
+        await self.async_load()
+        if config == self.stable:
+            # No need to save a pending config that is the same as stable.
+            config = None
+        self._pending = config
+        await self._async_persist()
+
+    async def async_promote_pending(self) -> None:
+        """Promote the pending config to stable.
+
+        Raises ``HomeAssistantError`` if there is nothing to promote.
+        """
+        await self.async_load()
+        if self._pending is None:
+            raise HomeAssistantError("No pending HTTP config to promote")
+        self._stable = self._pending
+        self._pending = None
+        await self._async_persist()
+
+    async def async_migrate_yaml(self, config: ConfData) -> None:
+        """Migrate YAML config to storage as pending if not the same as the config used for recovery."""
+        validated_config = cast(ConfData, HTTP_STORAGE_SCHEMA(config))
+        await self.async_set_pending(validated_config)
+        self._yaml_migration_done = True
+        await self._async_persist()
+
+    async def _async_persist(self) -> None:
+        """Write the current state to disk (or remove the file if empty)."""
+        await self._store.async_save(
+            {
+                KEY_STABLE: self._stable,
+                KEY_PENDING: self._pending,
+                KEY_YAML_MIGRATION_DONE: self._yaml_migration_done,
+            }
+        )
+
+
+async def async_get_and_load_store(hass: HomeAssistant) -> HTTPConfigStore:
+    """Return the singleton HTTP config store and load it."""
+    if (store := hass.data.get(DATA_STORE)) is None:
+        store = HTTPConfigStore(hass)
+        hass.data[DATA_STORE] = store
+    await store.async_load()
+    return store
+
+
+async def async_load_config(hass: HomeAssistant, config: ConfigType) -> ConfData:
+    """Load the HTTP config to apply on this startup.
+
+    YAML config is only migrated once. Subsequent boots will ignore YAML and
+    use the store exclusively.
+
+    Resolution order:
+    - Recovery mode: always use ``stable`` so HA stays reachable after a bad
+      config; YAML is ignored entirely (any pending YAML migration is
+      deferred to the next normal boot).
+    - Normal mode: prefer ``pending`` if set, otherwise ``stable``.
+    """
+    store = await async_get_and_load_store(hass)
+    if hass.config.recovery_mode:
+        _LOGGER.info("Recovery mode active; using stable HTTP config")
+        return store.stable
+
+    yaml_conf: ConfData | None = config.get(DOMAIN)
+    if store.yaml_migration_done:
+        if yaml_conf is not None:
+            # todo create repair issue about YAML still being present after migration completed
+            # it will be ignored
+            pass
+        else:
+            # Clear any leftover deprecation issues if YAML was removed after migration.
+            ir.async_delete_issue(hass, DOMAIN, "deprecated_yaml_import_error")
+            ir.async_delete_issue(hass, DOMAIN, "deprecated_yaml")
+    else:
+        # Migrate YAML to storage and use it directly for this start. The
+        # migration function also marks the migration as done so future
+        # starts will ignore any remaining YAML.
+        conf_in_yaml = yaml_conf is not None
+        if yaml_conf is None:
+            yaml_conf = cast(ConfData, HTTP_STORAGE_SCHEMA({}))
+
+        try:
+            await store.async_migrate_yaml(yaml_conf)
+        except Exception:
+            _LOGGER.exception("Failed to migrate HTTP YAML configuration to storage")
+            ir.async_create_issue(
+                hass,
+                DOMAIN,
+                "deprecated_yaml_import_error",
+                is_fixable=False,
+                severity=ir.IssueSeverity.ERROR,
+                translation_key="deprecated_yaml_import_error",
+            )
+        else:
+            if conf_in_yaml:
+                ir.async_create_issue(
+                    hass,
+                    DOMAIN,
+                    "deprecated_yaml",
+                    breaks_in_ha_version="2027.6.0",
+                    is_fixable=False,
+                    severity=ir.IssueSeverity.WARNING,
+                    translation_key="deprecated_yaml",
+                )
+
+    if store.pending is not None:
+        _LOGGER.info("Using pending HTTP config")
+        return store.pending
+
+    _LOGGER.info("Using stable HTTP config")
+    return store.stable

homeassistant/components/http/const.py

@@ -11,6 +11,30 @@ DOMAIN: Final = "http"
 KEY_HASS_USER: Final = "hass_user"
 KEY_HASS_REFRESH_TOKEN_ID: Final = "hass_refresh_token_id"
 
+CONF_SERVER_HOST: Final = "server_host"
+CONF_SERVER_PORT: Final = "server_port"
+CONF_BASE_URL: Final = "base_url"
+CONF_SSL_CERTIFICATE: Final = "ssl_certificate"
+CONF_SSL_PEER_CERTIFICATE: Final = "ssl_peer_certificate"
+CONF_SSL_KEY: Final = "ssl_key"
+CONF_CORS_ORIGINS: Final = "cors_allowed_origins"
+CONF_USE_X_FORWARDED_FOR: Final = "use_x_forwarded_for"
+CONF_USE_X_FRAME_OPTIONS: Final = "use_x_frame_options"
+CONF_TRUSTED_PROXIES: Final = "trusted_proxies"
+CONF_LOGIN_ATTEMPTS_THRESHOLD: Final = "login_attempts_threshold"
+CONF_IP_BAN_ENABLED: Final = "ip_ban_enabled"
+CONF_SSL_PROFILE: Final = "ssl_profile"
+
+SSL_MODERN: Final = "modern"
+SSL_INTERMEDIATE: Final = "intermediate"
+
+# Cast to be able to load custom cards.
+# My to be able to check url and version info.
+DEFAULT_CORS: Final[list[str]] = ["https://cast.home-assistant.io"]
+NO_LOGIN_ATTEMPT_THRESHOLD: Final = -1
+
+ATTR_CONFIG = "config"
+
 
 def is_supervisor_unix_socket_request(request: Request) -> bool:
     """Check if request arrived over the Supervisor Unix socket."""

homeassistant/components/http/strings.json

@@ -1,5 +1,13 @@
 {
   "issues": {
+    "deprecated_yaml": {
+      "description": "Your existing HTTP configuration from `configuration.yaml` has been imported. The `http` integration is now configured from the UI under **Settings** > **System** > **Network**.\n\nPlease remove the `http:` block from your `configuration.yaml` and restart Home Assistant.",
+      "title": "The HTTP YAML configuration is deprecated"
+    },
+    "deprecated_yaml_import_error": {
+      "description": "Migrating the `http` configuration from `configuration.yaml` to the integration's storage failed. Please check the logs for details and configure the `http` integration from the UI under **Settings** > **System** > **Network**.",
+      "title": "Failed to import HTTP YAML configuration"
+    },
     "server_host_deprecated_hassio": {
       "description": "The deprecated `server_host` configuration option in the HTTP integration is prone to break the communication between Home Assistant Core and Supervisor, and will be removed.\n\nIf you are using this option to bind Home Assistant to specific network interfaces, please remove it from your configuration. Home Assistant will automatically bind to all available interfaces by default.\n\nIf you have specific networking requirements, consider using firewall rules or other network configuration to control access to Home Assistant.",
       "title": "The `server_host` HTTP configuration may break Home Assistant Core - Supervisor communication"

homeassistant/components/http/websocket_api.py

@@ -0,0 +1,85 @@
+"""WebSocket API for the HTTP integration user config."""
+
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant.components import websocket_api
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.exceptions import HomeAssistantError
+
+from .config import HTTP_STORAGE_SCHEMA, async_get_and_load_store
+from .const import ATTR_CONFIG
+
+
+@callback
+def async_register_websocket_commands(hass: HomeAssistant) -> None:
+    """Register the HTTP config websocket commands."""
+    websocket_api.async_register_command(hass, websocket_get_config)
+    websocket_api.async_register_command(hass, websocket_set_config)
+    websocket_api.async_register_command(hass, websocket_promote_config)
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command({vol.Required("type"): "http/config"})
+@websocket_api.async_response
+async def websocket_get_config(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict[str, Any],
+) -> None:
+    """Return the active HTTP configuration (the confirmed-working ``stable`` slot)."""
+    store = await async_get_and_load_store(hass)
+    connection.send_result(msg["id"], store.stable)
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "http/config/configure",
+        vol.Required(ATTR_CONFIG): vol.Any(None, HTTP_STORAGE_SCHEMA),
+    }
+)
+@websocket_api.async_response
+async def websocket_set_config(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict[str, Any],
+) -> None:
+    """Store a new pending HTTP configuration.
+
+    The new config is not applied until Home Assistant is restarted
+    and the user promotes it via ``http/config/promote``. Until then
+    the existing ``stable`` config remains the recovery fallback.
+    """
+    store = await async_get_and_load_store(hass)
+    await store.async_set_pending(msg[ATTR_CONFIG])
+    connection.send_result(msg["id"])
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command({vol.Required("type"): "http/config/promote"})
+@websocket_api.async_response
+async def websocket_promote_config(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict[str, Any],
+) -> None:
+    """Promote the pending HTTP config to stable.
+
+    Called by the user after they have verified Home Assistant is
+    working correctly with the pending config. The stable config is
+    the one used by recovery mode, so promotion must be explicit.
+    """
+    store = await async_get_and_load_store(hass)
+    try:
+        await store.async_promote_pending()
+    except HomeAssistantError as err:
+        connection.send_error(
+            msg["id"],
+            websocket_api.const.ERR_NOT_ALLOWED,
+            str(err),
+        )
+        return
+
+    connection.send_result(msg["id"])

homeassistant/components/recovery_mode/manifest.json

@@ -3,6 +3,7 @@
   "name": "Recovery Mode",
   "codeowners": ["@home-assistant/core"],
   "config_flow": false,
+  "dependencies": ["http"],
   "documentation": "https://www.home-assistant.io/integrations/recovery_mode",
   "integration_type": "system",
   "quality_scale": "internal"

tests/components/conversation/test_default_agent.py

@@ -278,7 +278,7 @@ async def test_expose_flag_automatically_set(
 
     assert await async_setup_component(hass, "conversation", {})
     await hass.async_block_till_done()
-    with patch("homeassistant.components.http.start_http_server_and_save_config"):
+    with patch("homeassistant.components.http.HomeAssistantHTTP.start"):
         await hass.async_start()
 
     # After setting up conversation, the expose flag should now be set on all entities

tests/components/http/test_init.py

@@ -2,9 +2,7 @@
 
 import asyncio
 from collections.abc import Callable
-from datetime import timedelta
 from http import HTTPStatus
-from ipaddress import ip_network
 import logging
 import os
 from pathlib import Path
@@ -15,17 +13,17 @@ import pytest
 from homeassistant.auth.providers.homeassistant import HassAuthProvider
 from homeassistant.components import cloud, http
 from homeassistant.components.cloud import CloudNotAvailable
+from homeassistant.components.http.config import _DEFAULT_CONFIG, HTTP_STORAGE_SCHEMA
 from homeassistant.const import HASSIO_USER_NAME
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import issue_registry as ir
 from homeassistant.helpers.http import KEY_HASS
 from homeassistant.helpers.network import NoURLAvailableError
 from homeassistant.setup import async_setup_component
-from homeassistant.util import dt as dt_util
 from homeassistant.util.ssl import server_context_intermediate, server_context_modern
 
-from tests.common import async_call_logger_set_level, async_fire_time_changed
-from tests.typing import ClientSessionGenerator
+from tests.common import async_call_logger_set_level
+from tests.typing import ClientSessionGenerator, WebSocketGenerator
 
 
 @pytest.fixture(autouse=True)
@@ -314,26 +312,46 @@ async def test_peer_cert(hass: HomeAssistant, tmp_path: Path) -> None:
     assert len(mock_load_verify_locations.mock_calls) == 1
 
 
+def _stable_http_storage(
+    stable: dict, *, pending: dict | None = None, yaml_migration_done: bool = True
+) -> dict:
+    """Build a hass_storage entry seeded with a confirmed-working stable config.
+
+    ``stable`` (and ``pending`` if given) are normalised through the storage
+    schema, matching what real users have on disk after migration / writes —
+    the load path does direct key access and assumes the payload is complete.
+    """
+    normalised_stable = dict(HTTP_STORAGE_SCHEMA(stable))
+    normalised_pending = dict(HTTP_STORAGE_SCHEMA(pending)) if pending else None
+    return {
+        "version": 2,
+        "key": "http",
+        "data": {
+            "stable": normalised_stable,
+            "pending": normalised_pending,
+            "yaml_migration_done": yaml_migration_done,
+        },
+    }
+
+
 async def test_emergency_ssl_certificate_when_invalid(
-    hass: HomeAssistant, tmp_path: Path, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+    hass_storage: dict,
 ) -> None:
     """Test http can startup with an emergency self signed cert when the current one is broken."""
 
     cert_path, key_path = await hass.async_add_executor_job(
         _setup_broken_ssl_pem_files, tmp_path
     )
-
-    hass.config.recovery_mode = True
-    assert (
-        await async_setup_component(
-            hass,
-            "http",
-            {
-                "http": {"ssl_certificate": cert_path, "ssl_key": key_path},
-            },
-        )
-        is True
+    # In recovery mode YAML is ignored, so seed the broken SSL paths into the
+    # store's stable slot — that's the only config recovery mode will look at.
+    hass_storage["http"] = _stable_http_storage(
+        {"ssl_certificate": str(cert_path), "ssl_key": str(key_path)}
     )
+    hass.config.recovery_mode = True
+    assert await async_setup_component(hass, "http", {}) is True
 
     await hass.async_start()
     await hass.async_block_till_done()
@@ -363,7 +381,10 @@ async def test_emergency_ssl_certificate_not_used_when_not_recovery_mode(
 
 
 async def test_emergency_ssl_certificate_when_invalid_get_url_fails(
-    hass: HomeAssistant, tmp_path: Path, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+    hass_storage: dict,
 ) -> None:
     """Test http falls back to no ssl when an emergency cert cannot be created when the configured one is broken.
 
@@ -372,21 +393,15 @@ async def test_emergency_ssl_certificate_when_invalid_get_url_fails(
     cert_path, key_path = await hass.async_add_executor_job(
         _setup_broken_ssl_pem_files, tmp_path
     )
+    hass_storage["http"] = _stable_http_storage(
+        {"ssl_certificate": str(cert_path), "ssl_key": str(key_path)}
+    )
     hass.config.recovery_mode = True
 
     with patch(
         "homeassistant.components.http.get_url", side_effect=NoURLAvailableError
     ) as mock_get_url:
-        assert (
-            await async_setup_component(
-                hass,
-                "http",
-                {
-                    "http": {"ssl_certificate": cert_path, "ssl_key": key_path},
-                },
-            )
-            is True
-        )
+        assert await async_setup_component(hass, "http", {}) is True
         await hass.async_start()
         await hass.async_block_till_done()
 
@@ -400,28 +415,25 @@ async def test_emergency_ssl_certificate_when_invalid_get_url_fails(
 
 
 async def test_invalid_ssl_and_cannot_create_emergency_cert(
-    hass: HomeAssistant, tmp_path: Path, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+    hass_storage: dict,
 ) -> None:
     """Test http falls back to no ssl when an emergency cert cannot be created when the configured one is broken."""
 
     cert_path, key_path = await hass.async_add_executor_job(
         _setup_broken_ssl_pem_files, tmp_path
     )
+    hass_storage["http"] = _stable_http_storage(
+        {"ssl_certificate": str(cert_path), "ssl_key": str(key_path)}
+    )
     hass.config.recovery_mode = True
 
     with patch(
         "homeassistant.components.http.x509.CertificateBuilder", side_effect=OSError
     ) as mock_builder:
-        assert (
-            await async_setup_component(
-                hass,
-                "http",
-                {
-                    "http": {"ssl_certificate": cert_path, "ssl_key": key_path},
-                },
-            )
-            is True
-        )
+        assert await async_setup_component(hass, "http", {}) is True
         await hass.async_start()
         await hass.async_block_till_done()
     assert "Could not create an emergency self signed ssl certificate" in caplog.text
@@ -431,7 +443,10 @@ async def test_invalid_ssl_and_cannot_create_emergency_cert(
 
 
 async def test_invalid_ssl_and_cannot_create_emergency_cert_with_ssl_peer_cert(
-    hass: HomeAssistant, tmp_path: Path, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+    hass_storage: dict,
 ) -> None:
     """Test http falls back to no ssl when an emergency cert cannot be created when the configured one is broken.
 
@@ -444,25 +459,19 @@ async def test_invalid_ssl_and_cannot_create_emergency_cert_with_ssl_peer_cert(
     cert_path, key_path = await hass.async_add_executor_job(
         _setup_broken_ssl_pem_files, tmp_path
     )
+    hass_storage["http"] = _stable_http_storage(
+        {
+            "ssl_certificate": str(cert_path),
+            "ssl_key": str(key_path),
+            "ssl_peer_certificate": str(cert_path),
+        }
+    )
     hass.config.recovery_mode = True
 
     with patch(
         "homeassistant.components.http.x509.CertificateBuilder", side_effect=OSError
     ) as mock_builder:
-        assert (
-            await async_setup_component(
-                hass,
-                "http",
-                {
-                    "http": {
-                        "ssl_certificate": cert_path,
-                        "ssl_key": key_path,
-                        "ssl_peer_certificate": cert_path,
-                    },
-                },
-            )
-            is False
-        )
+        assert await async_setup_component(hass, "http", {}) is False
         await hass.async_start()
         await hass.async_block_till_done()
     assert "Could not create an emergency self signed ssl certificate" in caplog.text
@@ -478,31 +487,6 @@ async def test_cors_defaults(hass: HomeAssistant) -> None:
     assert mock_setup.mock_calls[0][1][1] == ["https://cast.home-assistant.io"]
 
 
-async def test_storing_config(
-    hass: HomeAssistant,
-    aiohttp_client: ClientSessionGenerator,
-    unused_tcp_port_factory: Callable[[], int],
-) -> None:
-    """Test that we store last working config."""
-    config = {
-        http.CONF_SERVER_PORT: unused_tcp_port_factory(),
-        "use_x_forwarded_for": True,
-        "trusted_proxies": ["192.168.1.100"],
-    }
-
-    assert await async_setup_component(hass, http.DOMAIN, {http.DOMAIN: config})
-
-    await hass.async_start()
-
-    async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=200))
-    await hass.async_block_till_done()
-
-    restored = await http.async_get_last_config(hass)
-    restored["trusted_proxies"][0] = ip_network(restored["trusted_proxies"][0])
-
-    assert restored == http.HTTP_SCHEMA(config)
-
-
 async def test_logging(
     hass: HomeAssistant,
     hass_client: ClientSessionGenerator,
@@ -688,16 +672,24 @@ async def test_ssl_issue_urls_configured(
         "expected_issues",
     ),
     [
-        (False, {}, ["0.0.0.0", "::"], set()),
-        (False, {"server_host": "0.0.0.0"}, ["0.0.0.0"], set()),
-        (True, {}, ["0.0.0.0", "::"], set()),
+        (False, {}, ["0.0.0.0", "::"], {("http", "deprecated_yaml")}),
+        (
+            False,
+            {"server_host": "0.0.0.0"},
+            ["0.0.0.0"],
+            {("http", "deprecated_yaml")},
+        ),
+        (True, {}, ["0.0.0.0", "::"], {("http", "deprecated_yaml")}),
         (
             True,
             {"server_host": "0.0.0.0"},
             [
                 "0.0.0.0",
             ],
-            {("http", "server_host_deprecated_hassio")},
+            {
+                ("http", "server_host_deprecated_hassio"),
+                ("http", "deprecated_yaml"),
+            },
         ),
     ],
 )
@@ -808,3 +800,438 @@ async def test_unix_socket_rejected_relative_path(
 
     assert hass.http.supervisor_site is None
     assert "path must be absolute" in caplog.text
+
+
+async def test_yaml_migration_to_storage(
+    hass: HomeAssistant,
+    issue_registry: ir.IssueRegistry,
+    hass_storage: dict,
+) -> None:
+    """Test YAML config is migrated to the HTTP config store with a deprecation issue.
+
+    With no prior store, the migration stages YAML in ``pending`` (stable stays
+    on the schema defaults). The pending slot is what HA boots from until the
+    user confirms / promotes it via the UI.
+    """
+    yaml_conf = {
+        "server_port": 9123,
+        "cors_allowed_origins": ["https://example.com"],
+        "use_x_forwarded_for": True,
+        "trusted_proxies": ["127.0.0.0/8"],
+        "ip_ban_enabled": False,
+    }
+    with patch("asyncio.BaseEventLoop.create_server", return_value=Mock()):
+        assert await async_setup_component(hass, "http", {"http": yaml_conf})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    issue = issue_registry.async_get_issue("http", "deprecated_yaml")
+    assert issue is not None
+    assert issue.severity is ir.IssueSeverity.WARNING
+
+    assert (
+        issue_registry.async_get_issue("http", "deprecated_yaml_import_error") is None
+    )
+
+    stored = hass_storage["http"]["data"]
+    assert stored["yaml_migration_done"] is True
+    assert stored["stable"]["server_port"] == 8123  # untouched defaults
+    pending = stored["pending"]
+    assert pending is not None
+    assert pending["server_port"] == 9123
+    assert pending["cors_allowed_origins"] == ["https://example.com"]
+    assert pending["trusted_proxies"] == ["127.0.0.0/8"]
+    assert pending["ip_ban_enabled"] is False
+
+
+async def test_yaml_migration_matches_stable_no_pending(
+    hass: HomeAssistant,
+    issue_registry: ir.IssueRegistry,
+    hass_storage: dict,
+) -> None:
+    """If the YAML matches the existing stable config, no pending should be created."""
+    existing_stable = {
+        "server_port": 9123,
+        "cors_allowed_origins": ["https://example.com"],
+        "use_x_forwarded_for": True,
+        "trusted_proxies": ["127.0.0.0/8"],
+        "ip_ban_enabled": False,
+        "login_attempts_threshold": -1,
+        "ssl_profile": "modern",
+        "use_x_frame_options": True,
+    }
+    hass_storage["http"] = {
+        "version": 2,
+        "key": "http",
+        "data": {
+            "stable": existing_stable,
+            "pending": None,
+            "yaml_migration_done": False,
+        },
+    }
+
+    yaml_conf = {
+        "server_port": 9123,
+        "cors_allowed_origins": ["https://example.com"],
+        "use_x_forwarded_for": True,
+        "trusted_proxies": ["127.0.0.0/8"],
+        "ip_ban_enabled": False,
+    }
+    with patch("asyncio.BaseEventLoop.create_server", return_value=Mock()):
+        assert await async_setup_component(hass, "http", {"http": yaml_conf})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    stored = hass_storage["http"]["data"]
+    assert stored["pending"] is None
+    assert stored["stable"] == existing_stable
+
+    issue = issue_registry.async_get_issue("http", "deprecated_yaml")
+    assert issue is not None
+
+
+async def test_yaml_migration_differs_from_stable_creates_pending(
+    hass: HomeAssistant,
+    issue_registry: ir.IssueRegistry,
+    hass_storage: dict,
+) -> None:
+    """If the YAML differs from the existing stable config, it must be stored as pending."""
+    existing_stable = {
+        "server_port": 9123,
+        "cors_allowed_origins": ["https://example.com"],
+        "login_attempts_threshold": -1,
+        "ip_ban_enabled": True,
+        "ssl_profile": "modern",
+        "use_x_frame_options": True,
+    }
+    hass_storage["http"] = {
+        "version": 2,
+        "key": "http",
+        "data": {
+            "stable": existing_stable,
+            "pending": None,
+            "yaml_migration_done": False,
+        },
+    }
+
+    yaml_conf = {"server_port": 8765, "ip_ban_enabled": False}
+    with patch("asyncio.BaseEventLoop.create_server", return_value=Mock()):
+        assert await async_setup_component(hass, "http", {"http": yaml_conf})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    stored = hass_storage["http"]["data"]
+    assert stored["stable"] == existing_stable
+    pending = stored["pending"]
+    assert pending is not None
+    assert pending["server_port"] == 8765
+    assert pending["ip_ban_enabled"] is False
+
+    issue = issue_registry.async_get_issue("http", "deprecated_yaml")
+    assert issue is not None
+
+
+async def test_yaml_migration_failure_creates_error_issue(
+    hass: HomeAssistant,
+    issue_registry: ir.IssueRegistry,
+) -> None:
+    """Test that an error during YAML migration creates an error issue."""
+    yaml_conf = {"server_port": 9123}
+
+    with (
+        patch("asyncio.BaseEventLoop.create_server", return_value=Mock()),
+        patch(
+            "homeassistant.components.http.config.HTTPConfigStore.async_migrate_yaml",
+            side_effect=RuntimeError("boom"),
+        ),
+    ):
+        assert await async_setup_component(hass, "http", {"http": yaml_conf})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    issue = issue_registry.async_get_issue("http", "deprecated_yaml_import_error")
+    assert issue is not None
+    assert issue.severity is ir.IssueSeverity.ERROR
+
+    assert issue_registry.async_get_issue("http", "deprecated_yaml") is None
+
+
+async def test_setup_uses_stable_config_when_no_yaml(
+    hass: HomeAssistant,
+    issue_registry: ir.IssueRegistry,
+    hass_storage: dict,
+) -> None:
+    """Test HTTP config is loaded from the stable slot when no YAML or pending is set."""
+    hass_storage["http"] = _stable_http_storage(
+        {
+            "server_port": 9876,
+            "cors_allowed_origins": ["https://stored.example.com"],
+        }
+    )
+
+    mock_server = Mock()
+    with patch(
+        "asyncio.BaseEventLoop.create_server", return_value=mock_server
+    ) as mock_create_server:
+        assert await async_setup_component(hass, "http", {})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    args, _ = mock_create_server.call_args
+    assert args[2] == 9876
+
+    assert issue_registry.async_get_issue("http", "deprecated_yaml") is None
+    assert (
+        issue_registry.async_get_issue("http", "deprecated_yaml_import_error") is None
+    )
+
+
+async def test_setup_prefers_pending_over_stable_in_normal_mode(
+    hass: HomeAssistant,
+    hass_storage: dict,
+) -> None:
+    """Pending overrides stable on a normal boot so the new config gets tested."""
+    hass_storage["http"] = _stable_http_storage(
+        {"server_port": 9876}, pending={"server_port": 9999}
+    )
+
+    mock_server = Mock()
+    with patch(
+        "asyncio.BaseEventLoop.create_server", return_value=mock_server
+    ) as mock_create_server:
+        assert await async_setup_component(hass, "http", {})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    args, _ = mock_create_server.call_args
+    assert args[2] == 9999
+
+
+async def test_recovery_mode_falls_back_to_stable(
+    hass: HomeAssistant,
+    hass_storage: dict,
+) -> None:
+    """In recovery mode the pending config is ignored to keep HA reachable."""
+    hass_storage["http"] = _stable_http_storage(
+        {"server_port": 9876}, pending={"server_port": 9999}
+    )
+    hass.config.recovery_mode = True
+
+    mock_server = Mock()
+    with patch(
+        "asyncio.BaseEventLoop.create_server", return_value=mock_server
+    ) as mock_create_server:
+        assert await async_setup_component(hass, "http", {})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    args, _ = mock_create_server.call_args
+    assert args[2] == 9876
+
+
+async def test_recovery_mode_with_no_storage(
+    hass: HomeAssistant,
+    hass_storage: dict,
+    issue_registry: ir.IssueRegistry,
+) -> None:
+    """Recovery mode with no prior storage starts HTTP on the schema defaults.
+
+    This covers the first-ever-boot-into-recovery-mode case: bootstrap fell
+    through to recovery before HTTP ever had a chance to migrate YAML, so the
+    store is empty and we must come up cleanly on defaults.
+    """
+    assert "http" not in hass_storage
+    hass.config.recovery_mode = True
+
+    mock_server = Mock()
+    with patch(
+        "asyncio.BaseEventLoop.create_server", return_value=mock_server
+    ) as mock_create_server:
+        assert await async_setup_component(hass, "http", {})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    args, _ = mock_create_server.call_args
+    assert args[2] == 8123
+    # Recovery mode must not trigger YAML migration side effects.
+    assert issue_registry.async_get_issue("http", "deprecated_yaml") is None
+
+
+async def test_recovery_mode_ignores_yaml(
+    hass: HomeAssistant,
+    hass_storage: dict,
+    issue_registry: ir.IssueRegistry,
+) -> None:
+    """YAML config must not be applied or migrated while in recovery mode.
+
+    The whole point of recovery mode is to ignore the user's (possibly bad)
+    config and fall back to the last known good ``stable`` slot. Migrating
+    YAML here would defeat that and could re-introduce the broken config.
+    """
+    hass_storage["http"] = _stable_http_storage(
+        {"server_port": 5555}, yaml_migration_done=False
+    )
+    hass.config.recovery_mode = True
+
+    mock_server = Mock()
+    with patch(
+        "asyncio.BaseEventLoop.create_server", return_value=mock_server
+    ) as mock_create_server:
+        assert await async_setup_component(
+            hass, "http", {"http": {"server_port": 1234}}
+        )
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    args, _ = mock_create_server.call_args
+    # YAML's port must NOT win: stable is the only source of truth in recovery.
+    assert args[2] == 5555
+    # The migration must not run in recovery mode, so its flag stays untouched
+    # and no deprecation issue is created on this boot.
+    assert hass_storage["http"]["data"]["yaml_migration_done"] is False
+    assert issue_registry.async_get_issue("http", "deprecated_yaml") is None
+
+
+async def test_setup_migrates_v1_storage_to_v2(
+    hass: HomeAssistant,
+    hass_storage: dict,
+) -> None:
+    """An existing v1 store is migrated into the stable slot."""
+    hass_storage["http"] = {
+        "version": 1,
+        "key": "http",
+        "data": {"server_port": 9876},
+    }
+
+    mock_server = Mock()
+    with patch(
+        "asyncio.BaseEventLoop.create_server", return_value=mock_server
+    ) as mock_create_server:
+        assert await async_setup_component(hass, "http", {})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    # We do a yaml migration on first boot after store mirgation as v1 store is just used for recovery mode
+    # therefore default config is used
+
+    args, _ = mock_create_server.call_args
+    assert args[2] == 8123
+    assert hass_storage["http"]["version"] == 2
+    data = hass_storage["http"]["data"]
+    # The v1→v2 migration normalises the payload through the storage schema,
+    # so the v2 stable slot is well-formed (all keys present) on disk.
+    assert data["stable"]["server_port"] == 9876
+    assert data["stable"]["ip_ban_enabled"] is True
+    assert data["pending"] == _DEFAULT_CONFIG
+    assert data["yaml_migration_done"] is True
+
+
+async def test_websocket_http_config(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    hass_storage: dict,
+) -> None:
+    """Test the http/config, configure and promote websocket commands."""
+    with patch("asyncio.BaseEventLoop.create_server", return_value=Mock()):
+        assert await async_setup_component(hass, "http", {})
+        await async_setup_component(hass, "websocket_api", {})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    ws_client = await hass_ws_client(hass)
+
+    # On a fresh setup the stable slot is seeded with the schema defaults.
+    await ws_client.send_json_auto_id({"type": "http/config"})
+    response = await ws_client.receive_json()
+    assert response["success"]
+    assert response["result"]["server_port"] == 8123
+    assert response["result"]["ip_ban_enabled"] is True
+
+    new_config = {
+        "server_port": 9123,
+        "cors_allowed_origins": ["https://example.com"],
+        "use_x_forwarded_for": True,
+        "trusted_proxies": ["127.0.0.0/8"],
+        "ip_ban_enabled": False,
+        "login_attempts_threshold": 5,
+        "ssl_profile": "modern",
+        "use_x_frame_options": True,
+    }
+    await ws_client.send_json_auto_id(
+        {"type": "http/config/configure", "config": new_config}
+    )
+    response = await ws_client.receive_json()
+    assert response["success"]
+    # Configure is an ack; verify pending state via storage.
+    pending = hass_storage["http"]["data"]["pending"]
+    assert pending["server_port"] == 9123
+    assert pending["trusted_proxies"] == ["127.0.0.0/8"]
+    # Stable is unchanged until the user promotes.
+    await ws_client.send_json_auto_id({"type": "http/config"})
+    response = await ws_client.receive_json()
+    assert response["success"]
+    assert response["result"]["server_port"] == 8123
+
+    # Promote: pending becomes stable, pending is cleared.
+    await ws_client.send_json_auto_id({"type": "http/config/promote"})
+    response = await ws_client.receive_json()
+    assert response["success"]
+    assert hass_storage["http"]["data"]["pending"] is None
+    assert hass_storage["http"]["data"]["stable"]["server_port"] == 9123
+
+    await ws_client.send_json_auto_id({"type": "http/config"})
+    response = await ws_client.receive_json()
+    assert response["success"]
+    assert response["result"]["server_port"] == 9123
+
+    # Promoting again with no pending is rejected.
+    await ws_client.send_json_auto_id({"type": "http/config/promote"})
+    response = await ws_client.receive_json()
+    assert not response["success"]
+    assert response["error"]["code"] == "not_allowed"
+
+    # Clearing pending leaves stable untouched.
+    await ws_client.send_json_auto_id(
+        {"type": "http/config/configure", "config": {"server_port": 7000}}
+    )
+    response = await ws_client.receive_json()
+    assert response["success"]
+    assert hass_storage["http"]["data"]["pending"]["server_port"] == 7000
+
+    await ws_client.send_json_auto_id({"type": "http/config/configure", "config": None})
+    response = await ws_client.receive_json()
+    assert response["success"]
+    assert hass_storage["http"]["data"]["pending"] is None
+    assert hass_storage["http"]["data"]["stable"]["server_port"] == 9123
+
+
+@pytest.mark.parametrize(
+    "config",
+    [
+        {"server_port": "not-a-port"},
+        {
+            "use_x_forwarded_for": True,
+            "trusted_proxies": ["not-an-ip-network"],
+        },
+    ],
+)
+async def test_websocket_http_config_invalid(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    config: dict,
+) -> None:
+    """Test that an invalid HTTP config is rejected."""
+    with patch("asyncio.BaseEventLoop.create_server", return_value=Mock()):
+        assert await async_setup_component(hass, "http", {})
+        await async_setup_component(hass, "websocket_api", {})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    ws_client = await hass_ws_client(hass)
+
+    await ws_client.send_json_auto_id(
+        {"type": "http/config/configure", "config": config}
+    )
+    response = await ws_client.receive_json()
+    assert not response["success"]
+    assert response["error"]["code"] == "invalid_format"

tests/conftest.py

@@ -2240,5 +2240,5 @@ def disable_http_server() -> Generator[None]:
     This prevents the HTTP server from starting in tests that setup
     integrations which depend on the HTTP component.
     """
-    with patch("homeassistant.components.http.start_http_server_and_save_config"):
+    with patch("homeassistant.components.http.HomeAssistantHTTP.start"):
         yield

tests/test_bootstrap.py

@@ -76,7 +76,7 @@ def disable_block_async_io(disable_block_async_io):
 def mock_http_start_stop() -> Generator[None]:
     """Mock HTTP start and stop."""
     with (
-        patch("homeassistant.components.http.start_http_server_and_save_config"),
+        patch("homeassistant.components.http.HomeAssistantHTTP.start"),
         patch("homeassistant.components.http.HomeAssistantHTTP.stop"),
     ):
         yield