Start legacy port redirect under Supervisor regardless of port

Gate the legacy port redirect and CORS relaxation purely on running under
Supervisor, instead of also requiring the active port to differ from the
previous default.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

homeassistant/bootstrap.py

@@ -47,7 +47,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
@@ -408,12 +408,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,10 +12,15 @@ 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
+from aiohttp.hdrs import (
+    ACCESS_CONTROL_ALLOW_CREDENTIALS,
+    ACCESS_CONTROL_ALLOW_ORIGIN,
+    ORIGIN,
+)
 from aiohttp.http_parser import RawRequestMessage
 from aiohttp.streams import StreamReader
 from aiohttp.typedefs import JSONDecoder, StrOrURL
@@ -37,7 +42,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 +65,30 @@ 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 async_load_config, default_server_port
+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,
+    ENV_SUPERVISOR,
+    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 +98,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"]
 
@@ -112,7 +115,7 @@ HTTP_SCHEMA: Final = vol.All(
             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_SERVER_PORT, default=default_server_port): cv.port,
             vol.Optional(CONF_BASE_URL): cv.string,
             vol.Optional(CONF_SSL_CERTIFICATE): cv.isfile,
             vol.Optional(CONF_SSL_PEER_CERTIFICATE): cv.isfile,
@@ -154,30 +157,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 +180,19 @@ 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)
+    # Deferred import: websocket_api declares http as its manifest
+    # dependency and imports back into this package at module load
+    # (websocket_api/http.py -> homeassistant.components.http). A top-level
+    # import of .websocket_api here would re-enter the still-loading
+    # websocket_api package and fail when applying its decorators
+    # (e.g. @websocket_api.require_admin).
+    websocket_api_module = await async_import_module(
+        hass, "homeassistant.components.http.websocket_api"
+    )
 
-    if conf is None:
-        conf = cast(ConfData, HTTP_SCHEMA({}))
+    conf = await async_load_config(hass, config)
+
+    websocket_api_module.async_register_websocket_commands(hass)
 
     if CONF_SERVER_HOST in conf and is_hassio(hass):
         issue_id = "server_host_deprecated_hassio"
@@ -271,9 +259,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)
 
@@ -415,6 +401,13 @@ class HomeAssistantHTTP:
         self.site: HomeAssistantTCPSite | None = None
         self.supervisor_site: HomeAssistantUnixSite | None = None
         self.context: ssl.SSLContext | None = None
+        # Under Supervisor the server listens on port 80. Redirect the previous
+        # default port (SERVER_PORT) to the active one and relax CORS for
+        # same-host requests so already-flashed devices and bookmarks keep
+        # working, until onboarding completes.
+        self._port_transition = ENV_SUPERVISOR in os.environ
+        self._port_transition_cors = self._port_transition
+        self._legacy_redirect_runner: web.AppRunner | None = None
 
     async def async_initialize(
         self,
@@ -445,6 +438,9 @@ class HomeAssistantHTTP:
         setup_headers(self.app, use_x_frame_options)
         setup_cors(self.app, cors_origins)
 
+        if self._port_transition:
+            self.app.on_response_prepare.append(self._async_add_transition_cors)
+
         if self.ssl_certificate:
             self.context = await self.hass.async_add_executor_job(
                 self._create_ssl_context
@@ -690,8 +686,96 @@ class HomeAssistantHTTP:
 
         _LOGGER.info("Now listening on port %d", self.server_port)
 
+        if self._port_transition:
+            await self._async_start_legacy_redirect()
+
+    async def _async_add_transition_cors(
+        self, request: web.Request, response: web.StreamResponse
+    ) -> None:
+        """Echo a same-host Origin during the port transition.
+
+        A cross-origin fetch that follows the legacy-port redirect stays in
+        CORS mode, so the final response on the active port also needs CORS
+        headers. Scope this to our own host (the redirect is only ever same
+        host, different port) and only until onboarding completes.
+        """
+        if not self._port_transition_cors:
+            return
+        origin = request.headers.get(ORIGIN)
+        if not origin:
+            return
+        try:
+            origin_host = URL(origin).host
+        except ValueError:
+            return
+        if origin_host and origin_host == request.url.host:
+            response.headers[ACCESS_CONTROL_ALLOW_ORIGIN] = origin
+            response.headers[ACCESS_CONTROL_ALLOW_CREDENTIALS] = "true"
+
+    async def _async_start_legacy_redirect(self) -> None:
+        """Redirect the previous default port to the active port until onboarded."""
+        target_port = self.server_port
+
+        async def _redirect(request: web.Request) -> web.StreamResponse:
+            raise web.HTTPFound(request.url.with_port(target_port))
+
+        redirect_app = web.Application()
+        redirect_app.router.add_route("*", "/{path:.*}", _redirect)
+        self._legacy_redirect_runner = web.AppRunner(redirect_app)
+        await self._legacy_redirect_runner.setup()
+        site = HomeAssistantTCPSite(
+            self._legacy_redirect_runner, self.server_host, SERVER_PORT
+        )
+        try:
+            await site.start()
+        except OSError as error:
+            _LOGGER.error(
+                "Failed to start legacy port %d redirect: %s", SERVER_PORT, error
+            )
+            await self._legacy_redirect_runner.cleanup()
+            self._legacy_redirect_runner = None
+            return
+
+        _LOGGER.info(
+            "Redirecting legacy port %d to port %d until onboarding completes",
+            SERVER_PORT,
+            target_port,
+        )
+        async_when_setup_or_start(
+            self.hass, "onboarding", self._async_register_onboarding_teardown
+        )
+
+    async def _async_register_onboarding_teardown(self, *_: Any) -> None:
+        """Tear down the port transition once onboarding completes.
+
+        async_add_listener no-ops if onboarding isn't loaded yet, so register
+        only once the onboarding component is available.
+        """
+        from homeassistant.components import onboarding  # noqa: PLC0415
+
+        @callback
+        def _end_transition() -> None:
+            self.hass.async_create_task(self._async_end_port_transition())
+
+        onboarding.async_add_listener(self.hass, _end_transition)
+
+    async def _async_end_port_transition(self) -> None:
+        """Stop the legacy redirect and CORS relaxation."""
+        self._port_transition_cors = False
+        runner = self._legacy_redirect_runner
+        if runner is None:
+            return
+        self._legacy_redirect_runner = None
+        await runner.cleanup()
+        _LOGGER.info(
+            "Onboarding complete; stopped legacy port %d redirect", SERVER_PORT
+        )
+
     async def stop(self) -> None:
         """Stop the aiohttp server."""
+        if self._legacy_redirect_runner is not None:
+            await self._legacy_redirect_runner.cleanup()
+            self._legacy_redirect_runner = None
         if self.supervisor_site is not None:
             await self.supervisor_site.stop()
             if self.supervisor_unix_socket_path is not None:
@@ -709,23 +793,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,348 @@
+"""User-managed HTTP configuration store."""
+
+import asyncio
+from ipaddress import IPv4Network, IPv6Network, ip_network
+import logging
+import os
+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,
+    ENV_SETUP_PORT,
+    ENV_SUPERVISOR,
+    NO_LOGIN_ATTEMPT_THRESHOLD,
+    SSL_INTERMEDIATE,
+    SSL_MODERN,
+    SUPERVISOR_DEFAULT_PORT,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def default_server_port() -> int:
+    """Return the default HTTP server port.
+
+    Under Supervisor the default is port 80, since Supervisor fronts Core on
+    the standard HTTP port; otherwise the default is ``SERVER_PORT``. The
+    default can be overridden via the ``SETUP_PORT`` environment variable; an
+    invalid value is ignored in favor of the default.
+    """
+    default = SUPERVISOR_DEFAULT_PORT if ENV_SUPERVISOR in os.environ else SERVER_PORT
+    if (env_value := os.environ.get(ENV_SETUP_PORT)) is None:
+        return default
+    try:
+        return cast(int, cv.port(env_value))
+    except vol.Invalid:
+        _LOGGER.warning(
+            "Invalid port %r in %s environment variable; falling back to %s",
+            env_value,
+            ENV_SETUP_PORT,
+            default,
+        )
+        return default
+
+
+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)
+
+
+class ConfData(TypedDict, total=False):
+    """Typed dict for the validated HTTP config (matches ``HTTP_STORAGE_SCHEMA``)."""
+
+    server_host: list[str]
+    server_port: int
+    ssl_certificate: str
+    ssl_peer_certificate: str
+    ssl_key: str
+    cors_allowed_origins: list[str]
+    use_x_forwarded_for: bool
+    trusted_proxies: list[IPv4Network | IPv6Network]
+    login_attempts_threshold: int
+    ip_ban_enabled: bool
+    ssl_profile: str
+    use_x_frame_options: bool
+
+
+class _HTTPStoreData(TypedDict):
+    """Data structure for HTTP config storage."""
+
+    stable: ConfData
+    pending: ConfData | None
+    yaml_migration_done: bool
+
+
+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(
+    {
+        # YAML used to allow base_url (deprecated); strip it on the way in so
+        # the stored config never contains it.
+        vol.Remove(CONF_BASE_URL): object,
+        vol.Optional(CONF_SERVER_HOST): vol.All(
+            cv.ensure_list, vol.Length(min=1), [cv.string]
+        ),
+        vol.Optional(CONF_SERVER_PORT, default=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({}))
+
+
+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:
+            # YAML is still present after migration completed; surface a repair
+            # issue so the user knows their YAML is being ignored.
+            ir.async_create_issue(
+                hass,
+                DOMAIN,
+                "yaml_still_present_after_migration",
+                is_fixable=False,
+                severity=ir.IssueSeverity.WARNING,
+                translation_key="yaml_still_present_after_migration",
+            )
+        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")
+            ir.async_delete_issue(hass, DOMAIN, "yaml_still_present_after_migration")
+    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
+
+
+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
+
+
+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
+        self._load_lock = asyncio.Lock()
+
+    @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
+        async with self._load_lock:
+            if self._loaded:
+                # Another coroutine may have loaded the config while we were waiting
+                # for the lock; check again to avoid unnecessary disk I/O.
+                return  # type: ignore[unreachable]
+            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."""
+        await self.async_load()
+        validated_config = cast(ConfData, HTTP_STORAGE_SCHEMA(config))
+        self._pending = None if validated_config == self._stable else 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,
+            }
+        )
+
+
+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 = HTTP_STORAGE_SCHEMA(old_data)
+            except vol.Invalid:
+                _LOGGER.warning(
+                    "Discarding invalid v1 HTTP config during migration; "
+                    "falling back to defaults"
+                )
+                stable = _DEFAULT_CONFIG
+            return {
+                KEY_STABLE: stable,
+                KEY_PENDING: None,
+                KEY_YAML_MIGRATION_DONE: False,
+            }
+        return old_data

homeassistant/components/http/const.py

@@ -11,6 +11,37 @@ 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"
+
+ENV_SETUP_PORT: Final = "SETUP_PORT"
+ENV_SUPERVISOR: Final = "SUPERVISOR"
+
+# Default HTTP port when running under Supervisor, which fronts Core on the
+# standard HTTP port.
+SUPERVISOR_DEFAULT_PORT: Final = 80
+
+# 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"
@@ -7,6 +15,10 @@
     "ssl_configured_without_configured_urls": {
       "description": "Home Assistant detected that SSL has been set up on your instance, however, no custom external internet URL has been set.\n\nThis may result in unexpected behavior. Text-to-speech may fail, and integrations may not be able to connect back to your instance correctly.\n\nTo address this issue, go to Settings > System > Network; under the \"Home Assistant URL\" section, configure your new \"Internet\" and \"Local network\" addresses that match your new SSL configuration.",
       "title": "SSL is configured without an external URL or internal URL"
+    },
+    "yaml_still_present_after_migration": {
+      "description": "The HTTP configuration in `configuration.yaml` has already been migrated and is now being ignored. Please remove the `http:` block from your `configuration.yaml`. Manage the HTTP configuration from the UI under **Settings** > **System** > **Network**.",
+      "title": "HTTP YAML configuration is ignored after migration"
     }
   }
 }

homeassistant/components/http/websocket_api.py

@@ -0,0 +1,101 @@
+"""WebSocket API for the HTTP integration user config."""
+
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant.components import websocket_api
+from homeassistant.components.homeassistant import (
+    DOMAIN as HASS_DOMAIN,
+    SERVICE_HOMEASSISTANT_RESTART,
+)
+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 HTTP configuration.
+
+    ``stable`` is the confirmed-working config
+    ``pending`` is an unconfirmed config awaiting promotion, or ``None``.
+    """
+    store = await async_get_and_load_store(hass)
+    connection.send_result(
+        msg["id"],
+        {"stable": store.stable, "pending": store.pending},
+    )
+
+
+@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 and restart to apply it.
+
+    Restart whenever the pending slot changes, so the runtime config is
+    refreshed. The result reports whether a restart was triggered via
+    ``{"restart": bool}``.
+    """
+    store = await async_get_and_load_store(hass)
+    previous_pending = store.pending
+    await store.async_set_pending(msg[ATTR_CONFIG])
+    restart = store.pending != previous_pending
+    connection.send_result(msg["id"], {"restart": restart})
+
+    if restart:
+        await hass.services.async_call(HASS_DOMAIN, SERVICE_HOMEASSISTANT_RESTART)
+
+
+@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"])

tests/components/conversation/test_default_agent.py

@@ -351,7 +351,7 @@ async def test_expose_flag_automatically_set(
 
     assert await async_setup_component(hass, DOMAIN, {})
     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/hassio/test_init.py

@@ -342,7 +342,7 @@ async def test_setup_api_push_api_data_default(
     assert result
     assert len(supervisor_client.mock_calls) == 16
     supervisor_client.homeassistant.set_options.assert_called_once_with(
-        HomeAssistantOptions(ssl=False, port=8123, refresh_token=ANY)
+        HomeAssistantOptions(ssl=False, port=80, refresh_token=ANY)
     )
     refresh_token = (
         supervisor_client.homeassistant.set_options.mock_calls[0].args[0].refresh_token
@@ -419,7 +419,7 @@ async def test_setup_api_existing_hassio_user(
     assert result
     assert len(supervisor_client.mock_calls) == 16
     supervisor_client.homeassistant.set_options.assert_called_once_with(
-        HomeAssistantOptions(ssl=False, port=8123, refresh_token=token.token)
+        HomeAssistantOptions(ssl=False, port=80, refresh_token=token.token)
     )
 
 

tests/components/http/test_init.py

@@ -2,31 +2,36 @@
 
 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
+from typing import Any
 from unittest.mock import ANY, Mock, patch
 
 import pytest
+from yarl import URL
 
 from homeassistant.auth.providers.homeassistant import HassAuthProvider
 from homeassistant.components import cloud, http
 from homeassistant.components.cloud import CloudNotAvailable
 from homeassistant.components.http import DOMAIN
+from homeassistant.components.http.config import (
+    _DEFAULT_CONFIG,
+    HTTP_STORAGE_SCHEMA,
+    default_server_port,
+)
+from homeassistant.components.http.const import ENV_SETUP_PORT, ENV_SUPERVISOR
 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, async_mock_service
+from tests.typing import ClientSessionGenerator, WebSocketGenerator
 
 
 @pytest.fixture(autouse=True)
@@ -315,26 +320,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[str, Any],
 ) -> None:
     """Test http starts with emergency self-signed cert on invalid cert."""
 
     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,
-            DOMAIN,
-            {
-                "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[DOMAIN] = _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()
@@ -365,7 +390,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[str, Any],
 ) -> None:
     """Test http falls back to no ssl when emergency cert creation fails.
 
@@ -374,21 +402,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[DOMAIN] = _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,
-                DOMAIN,
-                {
-                    "http": {"ssl_certificate": cert_path, "ssl_key": key_path},
-                },
-            )
-            is True
-        )
+        assert await async_setup_component(hass, DOMAIN, {}) is True
         await hass.async_start()
         await hass.async_block_till_done()
 
@@ -403,28 +425,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[str, Any],
 ) -> None:
     """Test http falls back to no ssl on emergency cert creation failure."""
 
     cert_path, key_path = await hass.async_add_executor_job(
         _setup_broken_ssl_pem_files, tmp_path
     )
+    hass_storage[DOMAIN] = _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,
-                DOMAIN,
-                {
-                    "http": {"ssl_certificate": cert_path, "ssl_key": key_path},
-                },
-            )
-            is True
-        )
+        assert await async_setup_component(hass, DOMAIN, {}) 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
@@ -434,7 +453,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[str, Any],
 ) -> None:
     """Test no-ssl fallback with peer cert when emergency cert fails.
 
@@ -447,25 +469,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[DOMAIN] = _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,
-                DOMAIN,
-                {
-                    "http": {
-                        "ssl_certificate": cert_path,
-                        "ssl_key": key_path,
-                        "ssl_peer_certificate": cert_path,
-                    },
-                },
-            )
-            is False
-        )
+        assert await async_setup_component(hass, DOMAIN, {}) 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
@@ -481,31 +497,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,
@@ -691,16 +682,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"),
+            },
         ),
     ],
 )
@@ -811,3 +810,681 @@ 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[str, Any],
+) -> 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[DOMAIN]["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[str, Any],
+) -> 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[DOMAIN] = {
+        "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[DOMAIN]["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[str, Any],
+) -> 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[DOMAIN] = {
+        "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[DOMAIN]["data"]
+    assert stored["stable"] == existing_stable
+    assert stored["pending"] == {
+        "server_port": 8765,
+        "cors_allowed_origins": ["https://cast.home-assistant.io"],
+        "login_attempts_threshold": -1,
+        "ip_ban_enabled": False,
+        "ssl_profile": "modern",
+        "use_x_frame_options": True,
+    }
+
+    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_yaml_still_present_after_migration_creates_issue(
+    hass: HomeAssistant,
+    issue_registry: ir.IssueRegistry,
+    hass_storage: dict[str, Any],
+) -> None:
+    """When YAML lingers after migration, a repair issue is surfaced and YAML is ignored."""
+    hass_storage[DOMAIN] = _stable_http_storage(
+        {"server_port": 9876}, yaml_migration_done=True
+    )
+
+    yaml_conf = {"server_port": 1234}
+    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": yaml_conf})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    # YAML must be ignored once migration is done; stable wins.
+    args, _ = mock_create_server.call_args
+    assert args[2] == 9876
+
+    issue = issue_registry.async_get_issue("http", "yaml_still_present_after_migration")
+    assert issue is not None
+    assert issue.severity is ir.IssueSeverity.WARNING
+
+
+async def test_yaml_still_present_issue_cleared_when_yaml_removed(
+    hass: HomeAssistant,
+    issue_registry: ir.IssueRegistry,
+    hass_storage: dict[str, Any],
+) -> None:
+    """A previously created leftover-YAML issue is cleared once YAML is removed."""
+    hass_storage[DOMAIN] = _stable_http_storage(
+        {"server_port": 9876}, yaml_migration_done=True
+    )
+    ir.async_create_issue(
+        hass,
+        "http",
+        "yaml_still_present_after_migration",
+        is_fixable=False,
+        severity=ir.IssueSeverity.WARNING,
+        translation_key="yaml_still_present_after_migration",
+    )
+
+    with patch("asyncio.BaseEventLoop.create_server", return_value=Mock()):
+        assert await async_setup_component(hass, "http", {})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    assert (
+        issue_registry.async_get_issue("http", "yaml_still_present_after_migration")
+        is None
+    )
+
+
+async def test_setup_uses_stable_config_when_no_yaml(
+    hass: HomeAssistant,
+    issue_registry: ir.IssueRegistry,
+    hass_storage: dict[str, Any],
+) -> None:
+    """Test HTTP config is loaded from the stable slot when no YAML or pending is set."""
+    hass_storage[DOMAIN] = _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[str, Any],
+) -> None:
+    """Pending overrides stable on a normal boot so the new config gets tested."""
+    hass_storage[DOMAIN] = _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[str, Any],
+) -> None:
+    """In recovery mode the pending config is ignored to keep HA reachable."""
+    hass_storage[DOMAIN] = _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[str, Any],
+    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[str, Any],
+    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[DOMAIN] = _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[DOMAIN]["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[str, Any],
+) -> None:
+    """An existing v1 store is migrated into the stable slot."""
+    hass_storage[DOMAIN] = {
+        "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()
+
+    # The migrated v1 store config is only used in recovery mode. Since this
+    # test isn't running in recovery mode, the YAML migration runs on first
+    # boot after store migration. With no YAML http config, the default config is migrated to the pending slot and used. Therefore we assert below the default port (8123)
+    args, _ = mock_create_server.call_args
+    assert args[2] == 8123
+    assert hass_storage[DOMAIN]["version"] == 2
+    data = hass_storage[DOMAIN]["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
+
+
+@pytest.mark.parametrize(
+    ("env", "expected_port"),
+    [
+        pytest.param({}, 8123, id="unset"),
+        pytest.param({ENV_SETUP_PORT: "80"}, 80, id="valid"),
+        pytest.param({ENV_SETUP_PORT: "0"}, 8123, id="out-of-range"),
+        pytest.param({ENV_SETUP_PORT: "notaport"}, 8123, id="not-a-number"),
+        pytest.param({ENV_SETUP_PORT: ""}, 8123, id="empty"),
+        pytest.param({ENV_SUPERVISOR: "core"}, 80, id="supervisor"),
+        pytest.param(
+            {ENV_SUPERVISOR: "core", ENV_SETUP_PORT: "8080"},
+            8080,
+            id="supervisor-setup-port-override",
+        ),
+        pytest.param(
+            {ENV_SUPERVISOR: "core", ENV_SETUP_PORT: "notaport"},
+            80,
+            id="supervisor-invalid-setup-port",
+        ),
+    ],
+)
+def test_default_server_port(
+    env: dict[str, str],
+    expected_port: int,
+) -> None:
+    """Test the default port, including Supervisor and SETUP_PORT overrides."""
+    with patch.dict(os.environ, env, clear=True):
+        assert default_server_port() == expected_port
+
+
+async def test_setup_port_env_var_used_as_default(
+    hass: HomeAssistant,
+    hass_storage: dict[str, Any],
+) -> None:
+    """Test SETUP_PORT is used as the default server port without YAML config."""
+    mock_server = Mock()
+    with (
+        patch.dict(os.environ, {ENV_SETUP_PORT: "80"}),
+        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] == 80
+    assert hass_storage[DOMAIN]["data"]["pending"]["server_port"] == 80
+
+
+async def test_supervisor_defaults_to_port_80(
+    hass: HomeAssistant,
+    hass_storage: dict[str, Any],
+) -> None:
+    """Test that under Supervisor the server uses port 80 and redirects 8123."""
+    mock_server = Mock()
+    with (
+        patch.dict(os.environ, {ENV_SUPERVISOR: "core"}),
+        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()
+
+    # The main server binds port 80 and a legacy redirect binds port 8123.
+    bound_ports = {call.args[2] for call in mock_create_server.call_args_list}
+    assert bound_ports == {80, 8123}
+    assert hass_storage[DOMAIN]["data"]["pending"]["server_port"] == 80
+
+
+async def test_no_legacy_redirect_without_supervisor(
+    hass: HomeAssistant,
+    hass_storage: dict[str, Any],
+) -> None:
+    """Test that no legacy redirect is started when not running under Supervisor."""
+    mock_server = Mock()
+    with (
+        patch.dict(os.environ, {ENV_SETUP_PORT: "80"}),
+        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()
+
+    bound_ports = {call.args[2] for call in mock_create_server.call_args_list}
+    assert bound_ports == {80}
+    assert hass.http._legacy_redirect_runner is None
+
+
+@pytest.mark.parametrize(
+    ("origin", "cors_enabled", "expect_header"),
+    [
+        pytest.param("http://homeassistant.local:8123", True, True, id="same-host"),
+        pytest.param("http://evil.example.com", True, False, id="foreign-host"),
+        pytest.param("http://homeassistant.local:8123", False, False, id="onboarded"),
+    ],
+)
+async def test_transition_cors_header(
+    hass: HomeAssistant,
+    hass_storage: dict[str, Any],
+    origin: str,
+    cors_enabled: bool,
+    expect_header: bool,
+) -> None:
+    """Test the transition CORS hook only echoes same-host origins while active."""
+    mock_server = Mock()
+    with (
+        patch.dict(os.environ, {ENV_SUPERVISOR: "core"}),
+        patch("asyncio.BaseEventLoop.create_server", return_value=mock_server),
+    ):
+        assert await async_setup_component(hass, "http", {})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    server = hass.http
+    server._port_transition_cors = cors_enabled
+    request = Mock()
+    request.headers = {"Origin": origin}
+    request.url = URL("http://homeassistant.local/manifest.json")
+    response = Mock()
+    response.headers = {}
+
+    await server._async_add_transition_cors(request, response)
+
+    assert ("Access-Control-Allow-Origin" in response.headers) is expect_header
+    if expect_header:
+        assert response.headers["Access-Control-Allow-Origin"] == origin
+        assert response.headers["Access-Control-Allow-Credentials"] == "true"
+
+
+async def test_port_transition_ends_on_onboarding(
+    hass: HomeAssistant,
+    hass_storage: dict[str, Any],
+) -> None:
+    """Test the legacy redirect and CORS relaxation stop once onboarded."""
+    mock_server = Mock()
+    with (
+        patch.dict(os.environ, {ENV_SUPERVISOR: "core"}),
+        patch("asyncio.BaseEventLoop.create_server", return_value=mock_server),
+    ):
+        assert await async_setup_component(hass, "http", {})
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+        server = hass.http
+        assert server._legacy_redirect_runner is not None
+        assert server._port_transition_cors is True
+
+        await server._async_end_port_transition()
+
+    assert server._legacy_redirect_runner is None
+    assert server._port_transition_cors is False
+
+
+async def test_websocket_http_config(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    hass_storage: dict[str, Any],
+) -> 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)
+
+    # Staging a new config triggers a restart so the pending config is applied.
+    restart_calls = async_mock_service(hass, "homeassistant", "restart")
+
+    # On a fresh setup the stable slot is seeded with the schema defaults and
+    # there is no pending config.
+    await ws_client.send_json_auto_id({"type": "http/config"})
+    response = await ws_client.receive_json()
+    assert response["success"]
+    assert response["result"] == {"stable": _DEFAULT_CONFIG, "pending": None}
+
+    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"]
+    assert response["result"] == {"restart": True}
+    pending = hass_storage[DOMAIN]["data"]["pending"]
+    assert pending["server_port"] == 9123
+    assert pending["trusted_proxies"] == ["127.0.0.0/8"]
+    await hass.async_block_till_done()
+    assert len(restart_calls) == 1
+
+    # Stable is unchanged until the user promotes, but the pending config is
+    # now returned alongside it.
+    await ws_client.send_json_auto_id({"type": "http/config"})
+    response = await ws_client.receive_json()
+    assert response["success"]
+    assert response["result"] == {"stable": _DEFAULT_CONFIG, "pending": new_config}
+
+    # 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[DOMAIN]["data"]["pending"] is None
+    assert hass_storage[DOMAIN]["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"] == {"stable": new_config, "pending": None}
+
+    # 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"
+
+    # Staging a different config again changes the pending slot -> restart.
+    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 response["result"] == {"restart": True}
+    assert hass_storage[DOMAIN]["data"]["pending"]["server_port"] == 7000
+    await hass.async_block_till_done()
+    assert len(restart_calls) == 2
+
+    # Clearing a previously staged config also changes the active config back
+    # to stable, so it must trigger a restart too.
+    await ws_client.send_json_auto_id({"type": "http/config/configure", "config": None})
+    response = await ws_client.receive_json()
+    assert response["success"]
+    assert response["result"] == {"restart": True}
+    assert hass_storage[DOMAIN]["data"]["pending"] is None
+    assert hass_storage[DOMAIN]["data"]["stable"]["server_port"] == 9123
+    await hass.async_block_till_done()
+    assert len(restart_calls) == 3
+
+    # Clearing again when there is no pending config is a no-op -> no restart.
+    await ws_client.send_json_auto_id({"type": "http/config/configure", "config": None})
+    response = await ws_client.receive_json()
+    assert response["success"]
+    assert response["result"] == {"restart": False}
+    await hass.async_block_till_done()
+    assert len(restart_calls) == 3
+
+
+@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

@@ -2263,5 +2263,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