Add support for fetching bindkey from Mi cloud

homeassistant/components/xiaomi_ble/config_flow.py

@@ -4,10 +4,16 @@
 
 from collections.abc import Mapping
 import dataclasses
+import logging
 from typing import Any
 
 import voluptuous as vol
-from xiaomi_ble import XiaomiBluetoothDeviceData as DeviceData
+from xiaomi_ble import (
+    XiaomiBluetoothDeviceData as DeviceData,
+    XiaomiCloudException,
+    XiaomiCloudInvalidAuthenticationException,
+    XiaomiCloudTokenFetch,
+)
 from xiaomi_ble.parser import EncryptionScheme
 
 from homeassistant.components import onboarding
@@ -18,13 +24,17 @@
     async_process_advertisements,
 )
 from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
-from homeassistant.const import CONF_ADDRESS
+from homeassistant.const import CONF_ADDRESS, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.data_entry_flow import AbortFlow
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
 from .const import DOMAIN
 
 # How long to wait for additional advertisement packets if we don't have the right ones
 ADDITIONAL_DISCOVERY_TIMEOUT = 60
 
+_LOGGER = logging.getLogger(__name__)
+
 
 @dataclasses.dataclass
 class Discovery:
@@ -104,7 +114,7 @@ async def async_step_bluetooth(
         if device.encryption_scheme == EncryptionScheme.MIBEACON_LEGACY:
             return await self.async_step_get_encryption_key_legacy()
         if device.encryption_scheme == EncryptionScheme.MIBEACON_4_5:
-            return await self.async_step_get_encryption_key_4_5()
+            return await self.async_step_get_encryption_key_4_5_choose_method()
         return await self.async_step_bluetooth_confirm()
 
     async def async_step_get_encryption_key_legacy(
@@ -175,6 +185,67 @@ async def async_step_get_encryption_key_4_5(
             errors=errors,
         )
 
+    async def async_step_cloud_auth(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the cloud auth step."""
+        assert self._discovery_info
+
+        errors: dict[str, str] = {}
+        description_placeholders: dict[str, str] = {}
+        if user_input is not None:
+            session = async_get_clientsession(self.hass)
+            fetcher = XiaomiCloudTokenFetch(
+                user_input[CONF_USERNAME], user_input[CONF_PASSWORD], session
+            )
+            try:
+                device_details = await fetcher.get_device_info(
+                    self._discovery_info.address
+                )
+            except XiaomiCloudInvalidAuthenticationException as ex:
+                _LOGGER.debug("Authentication failed: %s", ex, exc_info=True)
+                errors = {"base": "auth_failed"}
+                description_placeholders = {"error_detail": str(ex)}
+            except XiaomiCloudException as ex:
+                _LOGGER.debug("Failed to connect to MI API: %s", ex, exc_info=True)
+                raise AbortFlow(
+                    "api_error", description_placeholders={"error_detail": str(ex)}
+                ) from ex
+            else:
+                if device_details:
+                    return await self.async_step_get_encryption_key_4_5(
+                        {"bindkey": device_details.bindkey}
+                    )
+                errors = {"base": "api_device_not_found"}
+
+        user_input = user_input or {}
+        return self.async_show_form(
+            step_id="cloud_auth",
+            errors=errors,
+            data_schema=vol.Schema(
+                {
+                    vol.Required(
+                        CONF_USERNAME, default=user_input.get(CONF_USERNAME)
+                    ): str,
+                    vol.Required(CONF_PASSWORD): str,
+                }
+            ),
+            description_placeholders={
+                **self.context["title_placeholders"],
+                **description_placeholders,
+            },
+        )
+
+    async def async_step_get_encryption_key_4_5_choose_method(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Choose method to get the bind key for a version 4/5 device."""
+        return self.async_show_menu(
+            step_id="get_encryption_key_4_5_choose_method",
+            menu_options=["cloud_auth", "get_encryption_key_4_5"],
+            description_placeholders=self.context["title_placeholders"],
+        )
+
     async def async_step_bluetooth_confirm(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
@@ -231,7 +302,7 @@ async def async_step_user(
                 return await self.async_step_get_encryption_key_legacy()
 
             if discovery.device.encryption_scheme == EncryptionScheme.MIBEACON_4_5:
-                return await self.async_step_get_encryption_key_4_5()
+                return await self.async_step_get_encryption_key_4_5_choose_method()
 
             return self._async_get_or_create_entry()
 
@@ -273,7 +344,7 @@ async def async_step_reauth(
             return await self.async_step_get_encryption_key_legacy()
 
         if device.encryption_scheme == EncryptionScheme.MIBEACON_4_5:
-            return await self.async_step_get_encryption_key_4_5()
+            return await self.async_step_get_encryption_key_4_5_choose_method()
 
         # Otherwise there wasn't actually encryption so abort
         return self.async_abort(reason="reauth_successful")

homeassistant/components/xiaomi_ble/strings.json

@@ -25,18 +25,35 @@
         "data": {
           "bindkey": "Bindkey"
         }
+      },
+      "cloud_auth": {
+        "description": "Please provide your Mi app username and password. This data won't be saved and only used to retrieve the device encryption key. Usernames and passwords are case sensitive.",
+        "data": {
+          "username": "[%key:common::config_flow::data::username%]",
+          "password": "[%key:common::config_flow::data::password%]"
+        }
+      },
+      "get_encryption_key_4_5_choose_method": {
+        "description": "A Mi device can be set up in Home Assistant in two different ways.\n\nYou can enter the bindkey yourself, or Home Assistant can import them from your Mi account.",
+        "menu_options": {
+          "cloud_auth": "Mi account (recommended)",
+          "get_encryption_key_4_5": "Enter encryption key manually"
+        }
       }
     },
     "error": {
       "decryption_failed": "The provided bindkey did not work, sensor data could not be decrypted. Please check it and try again.",
       "expected_24_characters": "Expected a 24 character hexadecimal bindkey.",
-      "expected_32_characters": "Expected a 32 character hexadecimal bindkey."
+      "expected_32_characters": "Expected a 32 character hexadecimal bindkey.",
+      "auth_failed": "Authentication failed: {error_detail}",
+      "api_device_not_found": "The device was not found in your Mi account."
     },
     "abort": {
       "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
       "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
       "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
-      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
+      "api_error": "Error while communicating with Mi API: {error_detail}"
     }
   },
   "device_automation": {