core/homeassistant/core_config.py

"""Module to help with parsing and generating configuration files."""

from __future__ import annotations

from collections import OrderedDict
from collections.abc import Sequence
from contextlib import suppress
import enum
import logging
import os
import pathlib
from typing import TYPE_CHECKING, Any, Final
from urllib.parse import urlparse

import voluptuous as vol
from webrtc_models import RTCConfiguration, RTCIceServer
import yarl

from . import auth
from .auth import mfa_modules as auth_mfa_modules, providers as auth_providers
from .const import (
    ATTR_ASSUMED_STATE,
    ATTR_FRIENDLY_NAME,
    ATTR_HIDDEN,
    BASE_PLATFORMS,
    CONF_ALLOWLIST_EXTERNAL_DIRS,
    CONF_ALLOWLIST_EXTERNAL_URLS,
    CONF_AUTH_MFA_MODULES,
    CONF_AUTH_PROVIDERS,
    CONF_COUNTRY,
    CONF_CURRENCY,
    CONF_CUSTOMIZE,
    CONF_CUSTOMIZE_DOMAIN,
    CONF_CUSTOMIZE_GLOB,
    CONF_DEBUG,
    CONF_ELEVATION,
    CONF_EXTERNAL_URL,
    CONF_ID,
    CONF_INTERNAL_URL,
    CONF_LANGUAGE,
    CONF_LATITUDE,
    CONF_LEGACY_TEMPLATES,
    CONF_LONGITUDE,
    CONF_MEDIA_DIRS,
    CONF_NAME,
    CONF_PACKAGES,
    CONF_RADIUS,
    CONF_TEMPERATURE_UNIT,
    CONF_TIME_ZONE,
    CONF_TYPE,
    CONF_UNIT_SYSTEM,
    CONF_URL,
    CONF_USERNAME,
    EVENT_CORE_CONFIG_UPDATE,
    LEGACY_CONF_WHITELIST_EXTERNAL_DIRS,
    UnitOfLength,
    __version__,
)
from .core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from .generated.currencies import HISTORIC_CURRENCIES
from .helpers import config_validation as cv, issue_registry as ir
from .helpers.entity_values import EntityValues
from .helpers.frame import ReportBehavior, report_usage
from .helpers.storage import Store
from .helpers.typing import UNDEFINED, UndefinedType
from .util import dt as dt_util, location
from .util.hass_dict import HassKey
from .util.package import is_docker_env
from .util.unit_system import (
    _CONF_UNIT_SYSTEM_IMPERIAL,
    _CONF_UNIT_SYSTEM_METRIC,
    _CONF_UNIT_SYSTEM_US_CUSTOMARY,
    METRIC_SYSTEM,
    UnitSystem,
    get_unit_system,
)

# Typing imports that create a circular dependency
if TYPE_CHECKING:
    from .components.http import ApiConfig

_LOGGER = logging.getLogger(__name__)

DATA_CUSTOMIZE: HassKey[EntityValues] = HassKey("hass_customize")

CONF_CREDENTIAL: Final = "credential"
CONF_ICE_SERVERS: Final = "ice_servers"
CONF_WEBRTC: Final = "webrtc"

CORE_STORAGE_KEY = "core.config"
CORE_STORAGE_VERSION = 1
CORE_STORAGE_MINOR_VERSION = 4


class ConfigSource(enum.StrEnum):
    """Source of core configuration."""

    DEFAULT = "default"
    DISCOVERED = "discovered"
    STORAGE = "storage"
    YAML = "yaml"


def _no_duplicate_auth_provider(
    configs: Sequence[dict[str, Any]],
) -> Sequence[dict[str, Any]]:
    """No duplicate auth provider config allowed in a list.

    Each type of auth provider can only have one config without optional id.
    Unique id is required if same type of auth provider used multiple times.
    """
    config_keys: set[tuple[str, str | None]] = set()
    for config in configs:
        key = (config[CONF_TYPE], config.get(CONF_ID))
        if key in config_keys:
            raise vol.Invalid(
                f"Duplicate auth provider {config[CONF_TYPE]} found. "
                "Please add unique IDs "
                "if you want to have the same auth provider twice"
            )
        config_keys.add(key)
    return configs


def _no_duplicate_auth_mfa_module(
    configs: Sequence[dict[str, Any]],
) -> Sequence[dict[str, Any]]:
    """No duplicate auth mfa module item allowed in a list.

    Each type of mfa module can only have one config without optional id.
    A global unique id is required if same type of mfa module used multiple
    times.
    Note: this is different than auth provider
    """
    config_keys: set[str] = set()
    for config in configs:
        key = config.get(CONF_ID, config[CONF_TYPE])
        if key in config_keys:
            raise vol.Invalid(
                f"Duplicate mfa module {config[CONF_TYPE]} found. "
                "Please add unique IDs "
                "if you want to have the same mfa module twice"
            )
        config_keys.add(key)
    return configs


def _filter_bad_internal_external_urls(conf: dict) -> dict:
    """Filter internal/external URL with a path."""
    for key in CONF_INTERNAL_URL, CONF_EXTERNAL_URL:
        if key in conf and urlparse(conf[key]).path not in ("", "/"):
            # We warn but do not fix, because if this was incorrectly configured,
            # adjusting this value might impact security.
            _LOGGER.warning(
                "Invalid %s set. It's not allowed to have a path (/bla)", key
            )

    return conf


# Schema for all packages element
_PACKAGES_CONFIG_SCHEMA = vol.Schema({cv.string: vol.Any(dict, list)})

# Schema for individual package definition
_PACKAGE_DEFINITION_SCHEMA = vol.Schema({cv.string: vol.Any(dict, list, None)})

_CUSTOMIZE_DICT_SCHEMA = vol.Schema(
    {
        vol.Optional(ATTR_FRIENDLY_NAME): cv.string,
        vol.Optional(ATTR_HIDDEN): cv.boolean,
        vol.Optional(ATTR_ASSUMED_STATE): cv.boolean,
    },
    extra=vol.ALLOW_EXTRA,
)

_CUSTOMIZE_CONFIG_SCHEMA = vol.Schema(
    {
        vol.Optional(CONF_CUSTOMIZE, default={}): vol.Schema(
            {cv.entity_id: _CUSTOMIZE_DICT_SCHEMA}
        ),
        vol.Optional(CONF_CUSTOMIZE_DOMAIN, default={}): vol.Schema(
            {cv.string: _CUSTOMIZE_DICT_SCHEMA}
        ),
        vol.Optional(CONF_CUSTOMIZE_GLOB, default={}): vol.Schema(
            {cv.string: _CUSTOMIZE_DICT_SCHEMA}
        ),
    }
)


def _raise_issue_if_imperial_unit_system(
    hass: HomeAssistant, config: dict[str, Any]
) -> dict[str, Any]:
    if config.get(CONF_UNIT_SYSTEM) == _CONF_UNIT_SYSTEM_IMPERIAL:
        ir.async_create_issue(
            hass,
            HOMEASSISTANT_DOMAIN,
            "imperial_unit_system",
            is_fixable=False,
            learn_more_url="homeassistant://config/general",
            severity=ir.IssueSeverity.WARNING,
            translation_key="imperial_unit_system",
        )
        config[CONF_UNIT_SYSTEM] = _CONF_UNIT_SYSTEM_US_CUSTOMARY
    else:
        ir.async_delete_issue(hass, HOMEASSISTANT_DOMAIN, "imperial_unit_system")

    return config


def _raise_issue_if_historic_currency(hass: HomeAssistant, currency: str) -> None:
    if currency not in HISTORIC_CURRENCIES:
        ir.async_delete_issue(hass, HOMEASSISTANT_DOMAIN, "historic_currency")
        return

    ir.async_create_issue(
        hass,
        HOMEASSISTANT_DOMAIN,
        "historic_currency",
        is_fixable=False,
        learn_more_url="homeassistant://config/general",
        severity=ir.IssueSeverity.WARNING,
        translation_key="historic_currency",
        translation_placeholders={"currency": currency},
    )


def _raise_issue_if_no_country(hass: HomeAssistant, country: str | None) -> None:
    if country is not None:
        ir.async_delete_issue(hass, HOMEASSISTANT_DOMAIN, "country_not_configured")
        return

    ir.async_create_issue(
        hass,
        HOMEASSISTANT_DOMAIN,
        "country_not_configured",
        is_fixable=False,
        learn_more_url="homeassistant://config/general",
        severity=ir.IssueSeverity.WARNING,
        translation_key="country_not_configured",
    )


def _validate_currency(data: Any) -> Any:
    try:
        return cv.currency(data)
    except vol.InInvalid:
        with suppress(vol.InInvalid):
            return cv.historic_currency(data)
        raise


def _validate_stun_or_turn_url(value: Any) -> str:
    """Validate an URL."""
    url_in = str(value)
    url = urlparse(url_in)

    if url.scheme not in ("stun", "stuns", "turn", "turns"):
        raise vol.Invalid("invalid url")
    return url_in


CORE_CONFIG_SCHEMA = vol.All(
    _CUSTOMIZE_CONFIG_SCHEMA.extend(
        {
            CONF_NAME: vol.Coerce(str),
            CONF_LATITUDE: cv.latitude,
            CONF_LONGITUDE: cv.longitude,
            CONF_ELEVATION: vol.Coerce(int),
            CONF_RADIUS: cv.positive_int,
            vol.Remove(CONF_TEMPERATURE_UNIT): cv.temperature_unit,
            CONF_UNIT_SYSTEM: vol.Any(
                _CONF_UNIT_SYSTEM_METRIC,
                _CONF_UNIT_SYSTEM_US_CUSTOMARY,
                _CONF_UNIT_SYSTEM_IMPERIAL,
            ),
            CONF_TIME_ZONE: cv.time_zone,
            vol.Optional(CONF_INTERNAL_URL): cv.url,
            vol.Optional(CONF_EXTERNAL_URL): cv.url,
            vol.Optional(CONF_ALLOWLIST_EXTERNAL_DIRS): vol.All(
                cv.ensure_list, [vol.IsDir()]
            ),
            vol.Optional(LEGACY_CONF_WHITELIST_EXTERNAL_DIRS): vol.All(
                cv.ensure_list, [vol.IsDir()]
            ),
            vol.Optional(CONF_ALLOWLIST_EXTERNAL_URLS): vol.All(
                cv.ensure_list, [cv.url]
            ),
            vol.Optional(CONF_PACKAGES, default={}): _PACKAGES_CONFIG_SCHEMA,
            vol.Optional(CONF_AUTH_PROVIDERS): vol.All(
                cv.ensure_list,
                [
                    auth_providers.AUTH_PROVIDER_SCHEMA.extend(
                        {
                            CONF_TYPE: vol.NotIn(
                                ["insecure_example"],
                                (
                                    "The insecure_example auth provider"
                                    " is for testing only."
                                ),
                            )
                        }
                    )
                ],
                _no_duplicate_auth_provider,
            ),
            vol.Optional(CONF_AUTH_MFA_MODULES): vol.All(
                cv.ensure_list,
                [
                    auth_mfa_modules.MULTI_FACTOR_AUTH_MODULE_SCHEMA.extend(
                        {
                            CONF_TYPE: vol.NotIn(
                                ["insecure_example"],
                                "The insecure_example mfa module is for testing only.",
                            )
                        }
                    )
                ],
                _no_duplicate_auth_mfa_module,
            ),
            vol.Optional(CONF_MEDIA_DIRS): cv.schema_with_slug_keys(vol.IsDir()),
            vol.Remove(CONF_LEGACY_TEMPLATES): cv.boolean,
            vol.Optional(CONF_CURRENCY): _validate_currency,
            vol.Optional(CONF_COUNTRY): cv.country,
            vol.Optional(CONF_LANGUAGE): cv.language,
            vol.Optional(CONF_DEBUG): cv.boolean,
            vol.Optional(CONF_WEBRTC): vol.Schema(
                {
                    vol.Required(CONF_ICE_SERVERS): vol.All(
                        cv.ensure_list,
                        [
                            vol.Schema(
                                {
                                    vol.Required(CONF_URL): vol.All(
                                        cv.ensure_list, [_validate_stun_or_turn_url]
                                    ),
                                    vol.Optional(CONF_USERNAME): cv.string,
                                    vol.Optional(CONF_CREDENTIAL): cv.string,
                                }
                            )
                        ],
                    )
                }
            ),
        }
    ),
    _filter_bad_internal_external_urls,
)


async def async_process_ha_core_config(hass: HomeAssistant, config: dict) -> None:
    """Process the [homeassistant] section from the configuration.

    This method is a coroutine.
    """
    # CORE_CONFIG_SCHEMA is not async safe since it uses vol.IsDir
    # so we need to run it in an executor job.
    config = await hass.async_add_executor_job(CORE_CONFIG_SCHEMA, config)

    # Check if we need to raise an issue for imperial unit system
    config = _raise_issue_if_imperial_unit_system(hass, config)

    # Only load auth during startup.
    if not hasattr(hass, "auth"):
        if (auth_conf := config.get(CONF_AUTH_PROVIDERS)) is None:
            auth_conf = [{"type": "homeassistant"}]

        mfa_conf = config.get(
            CONF_AUTH_MFA_MODULES,
            [{"type": "totp", "id": "totp", "name": "Authenticator app"}],
        )

        setattr(
            hass, "auth", await auth.auth_manager_from_config(hass, auth_conf, mfa_conf)
        )

    await hass.config.async_load()

    hac = hass.config

    if any(
        k in config
        for k in (
            CONF_COUNTRY,
            CONF_CURRENCY,
            CONF_ELEVATION,
            CONF_EXTERNAL_URL,
            CONF_INTERNAL_URL,
            CONF_LANGUAGE,
            CONF_LATITUDE,
            CONF_LONGITUDE,
            CONF_NAME,
            CONF_RADIUS,
            CONF_TIME_ZONE,
            CONF_UNIT_SYSTEM,
        )
    ):
        hac.config_source = ConfigSource.YAML

    for key, attr in (
        (CONF_COUNTRY, "country"),
        (CONF_CURRENCY, "currency"),
        (CONF_ELEVATION, "elevation"),
        (CONF_EXTERNAL_URL, "external_url"),
        (CONF_INTERNAL_URL, "internal_url"),
        (CONF_LANGUAGE, "language"),
        (CONF_LATITUDE, "latitude"),
        (CONF_LONGITUDE, "longitude"),
        (CONF_MEDIA_DIRS, "media_dirs"),
        (CONF_NAME, "location_name"),
        (CONF_RADIUS, "radius"),
    ):
        if key in config:
            setattr(hac, attr, config[key])

    if config.get(CONF_DEBUG):
        hac.debug = True

    if CONF_WEBRTC in config:
        hac.webrtc.ice_servers = [
            RTCIceServer(
                server[CONF_URL],
                server.get(CONF_USERNAME),
                server.get(CONF_CREDENTIAL),
            )
            for server in config[CONF_WEBRTC][CONF_ICE_SERVERS]
        ]

    _raise_issue_if_historic_currency(hass, hass.config.currency)
    _raise_issue_if_no_country(hass, hass.config.country)

    if CONF_TIME_ZONE in config:
        await hac.async_set_time_zone(config[CONF_TIME_ZONE])

    if CONF_MEDIA_DIRS not in config:
        if is_docker_env():
            hac.media_dirs = {"local": "/media"}
        else:
            hac.media_dirs = {"local": hass.config.path("media")}

    # Init whitelist external dir
    hac.allowlist_external_dirs = {hass.config.path("www"), *hac.media_dirs.values()}
    if CONF_ALLOWLIST_EXTERNAL_DIRS in config:
        hac.allowlist_external_dirs.update(set(config[CONF_ALLOWLIST_EXTERNAL_DIRS]))

    elif LEGACY_CONF_WHITELIST_EXTERNAL_DIRS in config:
        _LOGGER.warning(
            "Key %s has been replaced with %s. Please update your config",
            LEGACY_CONF_WHITELIST_EXTERNAL_DIRS,
            CONF_ALLOWLIST_EXTERNAL_DIRS,
        )
        hac.allowlist_external_dirs.update(
            set(config[LEGACY_CONF_WHITELIST_EXTERNAL_DIRS])
        )

    # Init whitelist external URL list – make sure to add / to every URL that doesn't
    # already have it so that we can properly test "path ownership"
    if CONF_ALLOWLIST_EXTERNAL_URLS in config:
        hac.allowlist_external_urls.update(
            url if url.endswith("/") else f"{url}/"
            for url in config[CONF_ALLOWLIST_EXTERNAL_URLS]
        )

    # Customize
    cust_exact = dict(config[CONF_CUSTOMIZE])
    cust_domain = dict(config[CONF_CUSTOMIZE_DOMAIN])
    cust_glob = OrderedDict(config[CONF_CUSTOMIZE_GLOB])

    for name, pkg in config[CONF_PACKAGES].items():
        if (pkg_cust := pkg.get(HOMEASSISTANT_DOMAIN)) is None:
            continue

        try:
            pkg_cust = _CUSTOMIZE_CONFIG_SCHEMA(pkg_cust)
        except vol.Invalid:
            _LOGGER.warning("Package %s contains invalid customize", name)
            continue

        cust_exact.update(pkg_cust[CONF_CUSTOMIZE])
        cust_domain.update(pkg_cust[CONF_CUSTOMIZE_DOMAIN])
        cust_glob.update(pkg_cust[CONF_CUSTOMIZE_GLOB])

    hass.data[DATA_CUSTOMIZE] = EntityValues(cust_exact, cust_domain, cust_glob)

    if CONF_UNIT_SYSTEM in config:
        hac.units = get_unit_system(config[CONF_UNIT_SYSTEM])


class _ComponentSet(set[str]):
    """Set of loaded components.

    This set contains both top level components and platforms.

    Examples:
    `light`, `switch`, `hue`, `mjpeg.camera`, `universal.media_player`,
    `homeassistant.scene`

    The top level components set only contains the top level components.

    The all components set contains all components, including platform
    based components.

    """

    def __init__(
        self, top_level_components: set[str], all_components: set[str]
    ) -> None:
        """Initialize the component set."""
        self._top_level_components = top_level_components
        self._all_components = all_components

    def add(self, value: str) -> None:
        """Add a component to the store."""
        if "." not in value:
            self._top_level_components.add(value)
            self._all_components.add(value)
        else:
            platform, _, domain = value.partition(".")
            if domain in BASE_PLATFORMS:
                self._all_components.add(platform)
        return super().add(value)

    def remove(self, value: str) -> None:
        """Remove a component from the store."""
        if "." in value:
            raise ValueError("_ComponentSet does not support removing sub-components")
        self._top_level_components.remove(value)
        return super().remove(value)

    def discard(self, value: str) -> None:
        """Remove a component from the store."""
        raise NotImplementedError("_ComponentSet does not support discard, use remove")


class Config:
    """Configuration settings for Home Assistant."""

    _store: Config._ConfigStore

    def __init__(self, hass: HomeAssistant, config_dir: str) -> None:
        """Initialize a new config object."""
        # pylint: disable-next=import-outside-toplevel
        from .components.zone import DEFAULT_RADIUS

        self.hass = hass

        self.latitude: float = 0
        self.longitude: float = 0

        self.elevation: int = 0
        """Elevation (always in meters regardless of the unit system)."""

        self.radius: int = DEFAULT_RADIUS
        """Radius of the Home Zone (always in meters regardless of the unit system)."""

        self.debug: bool = False
        self.location_name: str = "Home"
        self.time_zone: str = "UTC"
        self.units: UnitSystem = METRIC_SYSTEM
        self.internal_url: str | None = None
        self.external_url: str | None = None
        self.currency: str = "EUR"
        self.country: str | None = None
        self.language: str = "en"

        self.config_source: ConfigSource = ConfigSource.DEFAULT

        # If True, pip install is skipped for requirements on startup
        self.skip_pip: bool = False

        # List of packages to skip when installing requirements on startup
        self.skip_pip_packages: list[str] = []

        # Set of loaded top level components
        # This set is updated by _ComponentSet
        # and should not be modified directly
        self.top_level_components: set[str] = set()

        # Set of all loaded components including platform
        # based components
        self.all_components: set[str] = set()

        # Set of loaded components
        self.components: _ComponentSet = _ComponentSet(
            self.top_level_components, self.all_components
        )

        # API (HTTP) server configuration
        self.api: ApiConfig | None = None

        # Directory that holds the configuration
        self.config_dir: str = config_dir

        # List of allowed external dirs to access
        self.allowlist_external_dirs: set[str] = set()

        # List of allowed external URLs that integrations may use
        self.allowlist_external_urls: set[str] = set()

        # Dictionary of Media folders that integrations may use
        self.media_dirs: dict[str, str] = {}

        # If Home Assistant is running in recovery mode
        self.recovery_mode: bool = False

        # Use legacy template behavior
        self.legacy_templates: bool = False

        # If Home Assistant is running in safe mode
        self.safe_mode: bool = False

        self.webrtc = RTCConfiguration()

    def async_initialize(self) -> None:
        """Finish initializing a config object.

        This must be called before the config object is used.
        """
        self._store = self._ConfigStore(self.hass)

    def distance(self, lat: float, lon: float) -> float | None:
        """Calculate distance from Home Assistant.

        Async friendly.
        """
        return self.units.length(
            location.distance(self.latitude, self.longitude, lat, lon),
            UnitOfLength.METERS,
        )

    def path(self, *path: str) -> str:
        """Generate path to the file within the configuration directory.

        Async friendly.
        """
        return os.path.join(self.config_dir, *path)

    def is_allowed_external_url(self, url: str) -> bool:
        """Check if an external URL is allowed."""
        parsed_url = f"{yarl.URL(url)!s}/"

        return any(
            allowed
            for allowed in self.allowlist_external_urls
            if parsed_url.startswith(allowed)
        )

    def is_allowed_path(self, path: str) -> bool:
        """Check if the path is valid for access from outside.

        This function does blocking I/O and should not be called from the event loop.
        Use hass.async_add_executor_job to schedule it on the executor.
        """
        assert path is not None

        thepath = pathlib.Path(path)
        try:
            # The file path does not have to exist (it's parent should)
            if thepath.exists():
                thepath = thepath.resolve()
            else:
                thepath = thepath.parent.resolve()
        except (FileNotFoundError, RuntimeError, PermissionError):
            return False

        for allowed_path in self.allowlist_external_dirs:
            try:
                thepath.relative_to(allowed_path)
            except ValueError:
                pass
            else:
                return True

        return False

    def as_dict(self) -> dict[str, Any]:
        """Return a dictionary representation of the configuration.

        Async friendly.
        """
        allowlist_external_dirs = list(self.allowlist_external_dirs)
        return {
            "allowlist_external_dirs": allowlist_external_dirs,
            "allowlist_external_urls": list(self.allowlist_external_urls),
            "components": list(self.components),
            "config_dir": self.config_dir,
            "config_source": self.config_source,
            "country": self.country,
            "currency": self.currency,
            "debug": self.debug,
            "elevation": self.elevation,
            "external_url": self.external_url,
            "internal_url": self.internal_url,
            "language": self.language,
            "latitude": self.latitude,
            "location_name": self.location_name,
            "longitude": self.longitude,
            "radius": self.radius,
            "recovery_mode": self.recovery_mode,
            "safe_mode": self.safe_mode,
            "state": self.hass.state.value,
            "time_zone": self.time_zone,
            "unit_system": self.units.as_dict(),
            "version": __version__,
            # legacy, backwards compat
            "whitelist_external_dirs": allowlist_external_dirs,
        }

    async def async_set_time_zone(self, time_zone_str: str) -> None:
        """Help to set the time zone."""
        if time_zone := await dt_util.async_get_time_zone(time_zone_str):
            self.time_zone = time_zone_str
            dt_util.set_default_time_zone(time_zone)
        else:
            raise ValueError(f"Received invalid time zone {time_zone_str}")

    def set_time_zone(self, time_zone_str: str) -> None:
        """Set the time zone.

        This is a legacy method that should not be used in new code.
        Use async_set_time_zone instead.

        It will be removed in Home Assistant 2025.6.
        """
        report_usage(
            "sets the time zone using set_time_zone instead of async_set_time_zone",
            core_integration_behavior=ReportBehavior.ERROR,
            custom_integration_behavior=ReportBehavior.ERROR,
            breaks_in_ha_version="2025.6",
        )
        if time_zone := dt_util.get_time_zone(time_zone_str):
            self.time_zone = time_zone_str
            dt_util.set_default_time_zone(time_zone)
        else:
            raise ValueError(f"Received invalid time zone {time_zone_str}")

    async def _async_update(
        self,
        *,
        country: str | UndefinedType | None = UNDEFINED,
        currency: str | None = None,
        elevation: int | None = None,
        external_url: str | UndefinedType | None = UNDEFINED,
        internal_url: str | UndefinedType | None = UNDEFINED,
        language: str | None = None,
        latitude: float | None = None,
        location_name: str | None = None,
        longitude: float | None = None,
        radius: int | None = None,
        source: ConfigSource,
        time_zone: str | None = None,
        unit_system: str | None = None,
    ) -> None:
        """Update the configuration from a dictionary."""
        self.config_source = source
        if country is not UNDEFINED:
            self.country = country
        if currency is not None:
            self.currency = currency
        if elevation is not None:
            self.elevation = elevation
        if external_url is not UNDEFINED:
            self.external_url = external_url
        if internal_url is not UNDEFINED:
            self.internal_url = internal_url
        if language is not None:
            self.language = language
        if latitude is not None:
            self.latitude = latitude
        if location_name is not None:
            self.location_name = location_name
        if longitude is not None:
            self.longitude = longitude
        if radius is not None:
            self.radius = radius
        if time_zone is not None:
            await self.async_set_time_zone(time_zone)
        if unit_system is not None:
            try:
                self.units = get_unit_system(unit_system)
            except ValueError:
                self.units = METRIC_SYSTEM

    async def async_update(self, **kwargs: Any) -> None:
        """Update the configuration from a dictionary."""
        await self._async_update(source=ConfigSource.STORAGE, **kwargs)
        await self._async_store()
        self.hass.bus.async_fire_internal(EVENT_CORE_CONFIG_UPDATE, kwargs)

        _raise_issue_if_historic_currency(self.hass, self.currency)
        _raise_issue_if_no_country(self.hass, self.country)

    async def async_load(self) -> None:
        """Load [homeassistant] core config."""
        if not (data := await self._store.async_load()):
            return

        # In 2021.9 we fixed validation to disallow a path (because that's never
        # correct) but this data still lives in storage, so we print a warning.
        if data.get("external_url") and urlparse(data["external_url"]).path not in (
            "",
            "/",
        ):
            _LOGGER.warning("Invalid external_url set. It's not allowed to have a path")

        if data.get("internal_url") and urlparse(data["internal_url"]).path not in (
            "",
            "/",
        ):
            _LOGGER.warning("Invalid internal_url set. It's not allowed to have a path")

        await self._async_update(
            source=ConfigSource.STORAGE,
            latitude=data.get("latitude"),
            longitude=data.get("longitude"),
            elevation=data.get("elevation"),
            unit_system=data.get("unit_system_v2"),
            location_name=data.get("location_name"),
            time_zone=data.get("time_zone"),
            external_url=data.get("external_url", UNDEFINED),
            internal_url=data.get("internal_url", UNDEFINED),
            currency=data.get("currency"),
            country=data.get("country"),
            language=data.get("language"),
            radius=data["radius"],
        )

    async def _async_store(self) -> None:
        """Store [homeassistant] core config."""
        data = {
            "latitude": self.latitude,
            "longitude": self.longitude,
            "elevation": self.elevation,
            # We don't want any integrations to use the name of the unit system
            # so we are using the private attribute here
            "unit_system_v2": self.units._name,  # noqa: SLF001
            "location_name": self.location_name,
            "time_zone": self.time_zone,
            "external_url": self.external_url,
            "internal_url": self.internal_url,
            "currency": self.currency,
            "country": self.country,
            "language": self.language,
            "radius": self.radius,
        }
        await self._store.async_save(data)

    class _ConfigStore(Store[dict[str, Any]]):
        """Class to help storing Config data."""

        def __init__(self, hass: HomeAssistant) -> None:
            """Initialize storage class."""
            super().__init__(
                hass,
                CORE_STORAGE_VERSION,
                CORE_STORAGE_KEY,
                private=True,
                atomic_writes=True,
                minor_version=CORE_STORAGE_MINOR_VERSION,
            )
            self._original_unit_system: str | None = None  # from old store 1.1

        async def _async_migrate_func(
            self,
            old_major_version: int,
            old_minor_version: int,
            old_data: dict[str, Any],
        ) -> dict[str, Any]:
            """Migrate to the new version."""

            # pylint: disable-next=import-outside-toplevel
            from .components.zone import DEFAULT_RADIUS

            data = old_data
            if old_major_version == 1 and old_minor_version < 2:
                # In 1.2, we remove support for "imperial", replaced by "us_customary"
                # Using a new key to allow rollback
                self._original_unit_system = data.get("unit_system")
                data["unit_system_v2"] = self._original_unit_system
                if data["unit_system_v2"] == _CONF_UNIT_SYSTEM_IMPERIAL:
                    data["unit_system_v2"] = _CONF_UNIT_SYSTEM_US_CUSTOMARY
            if old_major_version == 1 and old_minor_version < 3:
                # In 1.3, we add the key "language", initialize it from the
                # owner account.
                data["language"] = "en"
                try:
                    owner = await self.hass.auth.async_get_owner()
                    if owner is not None:
                        # pylint: disable-next=import-outside-toplevel
                        from .components.frontend import storage as frontend_store

                        _, owner_data = await frontend_store.async_user_store(
                            self.hass, owner.id
                        )

                        if (
                            "language" in owner_data
                            and "language" in owner_data["language"]
                        ):
                            with suppress(vol.InInvalid):
                                data["language"] = cv.language(
                                    owner_data["language"]["language"]
                                )
                # pylint: disable-next=broad-except
                except Exception:
                    _LOGGER.exception("Unexpected error during core config migration")
            if old_major_version == 1 and old_minor_version < 4:
                # In 1.4, we add the key "radius", initialize it with the default.
                data.setdefault("radius", DEFAULT_RADIUS)

            if old_major_version > 1:
                raise NotImplementedError
            return data

        async def async_save(self, data: dict[str, Any]) -> None:
            if self._original_unit_system:
                data["unit_system"] = self._original_unit_system
            return await super().async_save(data)