Add application credentials platform

CODEOWNERS

@@ -75,6 +75,8 @@ build.json @home-assistant/supervisor
 /tests/components/api/ @home-assistant/core
 /homeassistant/components/apple_tv/ @postlund
 /tests/components/apple_tv/ @postlund
+/homeassistant/components/application_credentials/ @home-assistant/core
+/tests/components/application_credentials/ @home-assistant/core
 /homeassistant/components/apprise/ @caronc
 /tests/components/apprise/ @caronc
 /homeassistant/components/aprs/ @PhilRW

homeassistant/components/application_credentials/__init__.py

@@ -0,0 +1,236 @@
+"""The Application Credentials integration.
+
+This integration provides APIs for managing local OAuth credentials on behalf
+of other integrations. Integrations register an authorization server, and then
+the APIs are used to add one or more client credentials. Integrations may also
+provide credentials from yaml for backwards compatibility.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+import logging
+from typing import Any, Protocol
+
+import voluptuous as vol
+
+from homeassistant.components import websocket_api
+from homeassistant.components.websocket_api.connection import ActiveConnection
+from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET, CONF_DOMAIN, CONF_ID
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.generated.application_credentials import APPLICATION_CREDENTIALS
+from homeassistant.helpers import collection, config_entry_oauth2_flow
+import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.storage import Store
+from homeassistant.helpers.typing import ConfigType
+from homeassistant.loader import IntegrationNotFound, async_get_integration
+from homeassistant.util import slugify
+
+__all__ = ["ClientCredential", "AuthorizationServer", "async_import_client_credential"]
+
+_LOGGER = logging.getLogger(__name__)
+
+DOMAIN = "application_credentials"
+
+STORAGE_KEY = DOMAIN
+STORAGE_VERSION = 1
+DATA_STORAGE = "storage"
+
+CREATE_FIELDS = {
+    vol.Required(CONF_DOMAIN): cv.string,
+    vol.Required(CONF_CLIENT_ID): cv.string,
+    vol.Required(CONF_CLIENT_SECRET): cv.string,
+}
+UPDATE_FIELDS: dict = {}  # Not supported
+
+
+@dataclass
+class ClientCredential:
+    """Represent an OAuth client credential."""
+
+    client_id: str
+    client_secret: str
+
+
+@dataclass
+class AuthorizationServer:
+    """Represent an OAuth2 Authorization Server."""
+
+    authorize_url: str
+    token_url: str
+
+
+class ApplicationCredentialsStorageCollection(collection.StorageCollection):
+    """Application credential collection stored in storage."""
+
+    CREATE_SCHEMA = vol.Schema(CREATE_FIELDS)
+
+    async def _process_create_data(self, data: dict[str, str]) -> dict[str, str]:
+        """Validate the config is valid."""
+        result = self.CREATE_SCHEMA(data)
+        domain = result[CONF_DOMAIN]
+        if not await _get_platform(self.hass, domain):
+            raise ValueError(f"No application_credentials platform for {domain}")
+        return result
+
+    @callback
+    def _get_suggested_id(self, info: dict[str, str]) -> str:
+        """Suggest an ID based on the config."""
+        return f"{info[CONF_DOMAIN]}.{info[CONF_CLIENT_ID]}"
+
+    async def _update_data(
+        self, data: dict[str, str], update_data: dict[str, str]
+    ) -> dict[str, str]:
+        """Return a new updated data object."""
+        raise ValueError("Updates not supported")
+
+    async def async_delete_item(self, item_id: str) -> None:
+        """Delete item, verifying credential is not in use."""
+        if item_id not in self.data:
+            raise collection.ItemNotFound(item_id)
+
+        # Cannot delete a credential currently in use by a ConfigEntry
+        current = self.data[item_id]
+        entries = self.hass.config_entries.async_entries(current[CONF_DOMAIN])
+        for entry in entries:
+            if entry.data.get("auth_implementation") == item_id:
+                raise ValueError("Cannot delete credential in use by an integration")
+
+        await super().async_delete_item(item_id)
+
+    async def async_import_item(self, info: dict[str, str]) -> None:
+        """Import an yaml credential if it does not already exist."""
+        suggested_id = self._get_suggested_id(info)
+        if self.id_manager.has_id(slugify(suggested_id)):
+            return
+        await self.async_create_item(info)
+
+    def async_client_credentials(self, domain: str) -> dict[str, ClientCredential]:
+        """Return ClientCredentials in storage for the specified domain."""
+        credentials = {}
+        for item in self.async_items():
+            if item[CONF_DOMAIN] != domain:
+                continue
+            credentials[item[CONF_ID]] = ClientCredential(
+                item[CONF_CLIENT_ID], item[CONF_CLIENT_SECRET]
+            )
+        return credentials
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up Application Credentials."""
+    hass.data[DOMAIN] = {}
+
+    id_manager = collection.IDManager()
+    storage_collection = ApplicationCredentialsStorageCollection(
+        Store(hass, STORAGE_VERSION, STORAGE_KEY),
+        logging.getLogger(f"{__name__}.storage_collection"),
+        id_manager,
+    )
+    await storage_collection.async_load()
+    hass.data[DOMAIN][DATA_STORAGE] = storage_collection
+
+    collection.StorageCollectionWebsocket(
+        storage_collection, DOMAIN, DOMAIN, CREATE_FIELDS, UPDATE_FIELDS
+    ).async_setup(hass)
+
+    websocket_api.async_register_command(hass, handle_integration_list)
+
+    config_entry_oauth2_flow.async_add_implementation_provider(
+        hass, DOMAIN, _async_provide_implementation
+    )
+
+    return True
+
+
+async def async_import_client_credential(
+    hass: HomeAssistant, domain: str, credential: ClientCredential
+) -> None:
+    """Import an existing credential from configuration.yaml."""
+    if DOMAIN not in hass.data:
+        raise ValueError("Integration 'application_credentials' not setup")
+    storage_collection = hass.data[DOMAIN][DATA_STORAGE]
+    item = {
+        CONF_DOMAIN: domain,
+        CONF_CLIENT_ID: credential.client_id,
+        CONF_CLIENT_SECRET: credential.client_secret,
+    }
+    await storage_collection.async_import_item(item)
+
+
+class AuthImplementation(config_entry_oauth2_flow.LocalOAuth2Implementation):
+    """Application Credentials local oauth2 implementation."""
+
+    @property
+    def name(self) -> str:
+        """Name of the implementation."""
+        return self.client_id
+
+
+async def _async_provide_implementation(
+    hass: HomeAssistant, domain: str
+) -> list[config_entry_oauth2_flow.AbstractOAuth2Implementation]:
+    """Return registered OAuth implementations."""
+
+    platform = await _get_platform(hass, domain)
+    if not platform:
+        return []
+
+    authorization_server = await platform.async_get_authorization_server(hass)
+    storage_collection = hass.data[DOMAIN][DATA_STORAGE]
+    credentials = storage_collection.async_client_credentials(domain)
+    return [
+        AuthImplementation(
+            hass,
+            auth_domain,
+            credential.client_id,
+            credential.client_secret,
+            authorization_server.authorize_url,
+            authorization_server.token_url,
+        )
+        for auth_domain, credential in credentials.items()
+    ]
+
+
+class ApplicationCredentialsProtocol(Protocol):
+    """Define the format that application_credentials platforms can have."""
+
+    async def async_get_authorization_server(
+        self, hass: HomeAssistant
+    ) -> AuthorizationServer:
+        """Return authorization server."""
+
+
+async def _get_platform(
+    hass: HomeAssistant, integration_domain: str
+) -> ApplicationCredentialsProtocol | None:
+    """Register an application_credentials platform."""
+    try:
+        integration = await async_get_integration(hass, integration_domain)
+    except IntegrationNotFound as err:
+        _LOGGER.debug("Integration '%s' does not exist: %s", integration_domain, err)
+        return None
+    try:
+        platform = integration.get_platform("application_credentials")
+    except ImportError as err:
+        _LOGGER.debug(
+            "Integration '%s' does not provide application_credentials: %s",
+            integration_domain,
+            err,
+        )
+        return None
+    if not hasattr(platform, "async_get_authorization_server"):
+        raise ValueError(
+            f"Integration '{integration_domain}' platform application_credentials did not implement 'async_get_authorization_server'"
+        )
+    return platform
+
+
+@websocket_api.websocket_command(
+    {vol.Required("type"): "application_credentials/config"}
+)
+@callback
+def handle_integration_list(
+    hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
+) -> None:
+    """Handle integrations command."""
+    connection.send_result(msg["id"], {"domains": APPLICATION_CREDENTIALS})

homeassistant/components/application_credentials/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "application_credentials",
+  "name": "Application Credentials",
+  "config_flow": false,
+  "documentation": "https://www.home-assistant.io/integrations/application_credentials",
+  "dependencies": ["auth", "websocket_api"],
+  "codeowners": ["@home-assistant/core"],
+  "quality_scale": "internal"
+}

homeassistant/components/application_credentials/strings.json

@@ -0,0 +1,3 @@
+{
+  "title": "Application Credentials"
+}

homeassistant/components/cloud/account_link.py

@@ -34,9 +34,9 @@ async def async_provide_implementation(hass: HomeAssistant, domain: str):
 
     for service in services:
         if service["service"] == domain and CURRENT_VERSION >= service["min_version"]:
-            return CloudOAuth2Implementation(hass, domain)
+            return [CloudOAuth2Implementation(hass, domain)]
 
-    return
+    return []
 
 
 async def _get_services(hass):

homeassistant/components/default_config/manifest.json

@@ -3,6 +3,7 @@
   "name": "Default Config",
   "documentation": "https://www.home-assistant.io/integrations/default_config",
   "dependencies": [
+    "application_credentials",
     "automation",
     "cloud",
     "counter",

homeassistant/components/xbox/__init__.py

@@ -20,6 +20,7 @@
     SmartglassConsoleStatus,
 )
 
+from homeassistant.components import application_credentials
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET, Platform
 from homeassistant.core import HomeAssistant
@@ -31,8 +32,8 @@
 from homeassistant.helpers.typing import ConfigType
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
-from . import api, config_flow
-from .const import DOMAIN, OAUTH2_AUTHORIZE, OAUTH2_TOKEN
+from . import api
+from .const import DOMAIN
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -63,15 +64,11 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     if DOMAIN not in config:
         return True
 
-    config_flow.OAuth2FlowHandler.async_register_implementation(
+    await application_credentials.async_import_client_credential(
         hass,
-        config_entry_oauth2_flow.LocalOAuth2Implementation(
-            hass,
-            DOMAIN,
-            config[DOMAIN][CONF_CLIENT_ID],
-            config[DOMAIN][CONF_CLIENT_SECRET],
-            OAUTH2_AUTHORIZE,
-            OAUTH2_TOKEN,
+        DOMAIN,
+        application_credentials.ClientCredential(
+            config[DOMAIN][CONF_CLIENT_ID], config[DOMAIN][CONF_CLIENT_SECRET]
         ),
     )
 

homeassistant/components/xbox/application_credentials.py

@@ -0,0 +1,14 @@
+"""Application credentials platform for xbox."""
+
+from homeassistant.components.application_credentials import AuthorizationServer
+from homeassistant.core import HomeAssistant
+
+from .const import OAUTH2_AUTHORIZE, OAUTH2_TOKEN
+
+
+async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
+    """Return authorization server."""
+    return AuthorizationServer(
+        authorize_url=OAUTH2_AUTHORIZE,
+        token_url=OAUTH2_TOKEN,
+    )

homeassistant/components/xbox/manifest.json

@@ -4,7 +4,7 @@
   "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/xbox",
   "requirements": ["xbox-webapi==2.0.11"],
-  "dependencies": ["auth"],
+  "dependencies": ["auth", "application_credentials"],
   "codeowners": ["@hunterjm"],
   "iot_class": "cloud_polling"
 }

homeassistant/generated/application_credentials.py

@@ -0,0 +1,10 @@
+"""Automatically generated by hassfest.
+
+To update, run python3 -m script.hassfest
+"""
+
+# fmt: off
+
+APPLICATION_CREDENTIALS = [
+    "xbox"
+]

homeassistant/helpers/config_entry_oauth2_flow.py

@@ -347,10 +347,9 @@ async def async_get_implementations(
         return registered
 
     registered = dict(registered)
-
-    for provider_domain, get_impl in hass.data[DATA_PROVIDERS].items():
-        if (implementation := await get_impl(hass, domain)) is not None:
-            registered[provider_domain] = implementation
+    for get_impl in list(hass.data[DATA_PROVIDERS].values()):
+        for impl in await get_impl(hass, domain):
+            registered[impl.domain] = impl
 
     return registered
 
@@ -373,7 +372,7 @@ def async_add_implementation_provider(
     hass: HomeAssistant,
     provider_domain: str,
     async_provide_implementation: Callable[
-        [HomeAssistant, str], Awaitable[AbstractOAuth2Implementation | None]
+        [HomeAssistant, str], Awaitable[list[AbstractOAuth2Implementation]]
     ],
 ) -> None:
     """Add an implementation provider.

script/hassfest/__main__.py

@@ -5,6 +5,7 @@
 from time import monotonic
 
 from . import (
+    application_credentials,
     codeowners,
     config_flow,
     coverage,
@@ -25,6 +26,7 @@
 from .model import Config, Integration
 
 INTEGRATION_PLUGINS = [
+    application_credentials,
     codeowners,
     config_flow,
     dependencies,