Add support cloud verify via phone and captcha

Author: AlexxIT

custom_components/xiaomi_gateway3/config_flow.py

@@ -1,4 +1,5 @@
 import base64
+import logging
 
 import homeassistant.helpers.config_validation as cv
 import voluptuous as vol
@@ -11,6 +12,8 @@
 from .core.xiaomi_cloud import AuthResult, MiCloud
 from .hass import hass_utils
 
+_LOGGER = logging.getLogger(__name__)
+
 SERVERS = {
     "cn": "China",
     "de": "Europe",
@@ -85,7 +88,7 @@ def _show_cloud_form(self, defaults: dict, errors: dict = None):
             errors=errors,
         )
 
-    async def _process_cloud_result(self, result: AuthResult | None):
+    async def _process_cloud_result(self, result: AuthResult):
         if result["ok"]:
             return self.async_create_entry(
                 title=self.cloud_user_input["username"],
@@ -104,13 +107,15 @@ async def _process_cloud_result(self, result: AuthResult | None):
                 description_placeholders={"image": image},
             )
 
-        if email := result.get("email"):
+        if verify := result.get("verify"):
             return self.async_show_form(
-                step_id="cloud_email",
+                step_id="cloud_verify",
                 data_schema=vol_schema({vol.Required("code"): str}),
-                description_placeholders={"email": email},
+                description_placeholders={"address": verify},
             )
 
+        _LOGGER.error("Can't login", exc_info=result["exception"])
+
         return self._show_cloud_form(
             self.cloud_user_input, errors={"base": "cant_login"}
         )
@@ -136,8 +141,8 @@ async def async_step_cloud_captcha(self, user_input: dict = None):
         result = await self.cloud.login_captcha(user_input["code"])
         return await self._process_cloud_result(result)
 
-    async def async_step_cloud_email(self, user_input: dict = None):
-        result = await self.cloud.verify_email(user_input["code"])
+    async def async_step_cloud_verify(self, user_input: dict = None):
+        result = await self.cloud.login_verify(user_input["code"])
         return await self._process_cloud_result(result)
 
     async def async_step_token(self, user_input: dict = None):

custom_components/xiaomi_gateway3/core/xiaomi_cloud.py

@@ -15,11 +15,14 @@
 SERVERS = ["cn", "de", "i2", "ru", "sg", "us"]
 SID = "xiaomiio"
 
+FLAG_PHONE = 4
+FLAG_EMAIL = 8
+
 
 class AuthResult(TypedDict, total=False):
     ok: bool
     captcha: bytes | None
-    email: str | None
+    verify: str | None  # phone or email address
     token: str | None
     exception: Exception | None
 
@@ -54,9 +57,7 @@ async def login(self, username: str, password: str, **kwargs) -> AuthResult:
                 return {"ok": False, "captcha": data["image"]}
 
             if notification_url := res1.get("notificationUrl"):
-                data = await self._send_email_ticket(notification_url)
-                self.auth = {"identity_session": data["identity_session"]}
-                return {"ok": False, "email": data["email"]}
+                return await self._get_notification_url(notification_url)
 
             res2 = await self._get_location(res1["location"])
 
@@ -90,15 +91,23 @@ async def login_token(self, token: str) -> AuthResult:
             return {"ok": False, "exception": e}
 
     async def login_captcha(self, code: str) -> AuthResult:
+        if "flag" in self.auth and "identity_session" in self.auth:
+            return await self._send_ticket(
+                self.auth["flag"], self.auth["identity_session"], code
+            )
+
         return await self.login(
             self.auth["username"], self.auth["password"], captcha_code=code
         )
 
-    async def verify_email(self, ticket: str) -> AuthResult:
+    async def login_verify(self, ticket: str) -> AuthResult:
+        flag = self.auth["flag"]
+        key = "Phone" if flag == FLAG_PHONE else "Email"
+
         r = await self.session.post(
-            "https://account.xiaomi.com/identity/auth/verifyEmail",
+            f"https://account.xiaomi.com/identity/auth/verify{key}",
             cookies={"identity_session": self.auth["identity_session"]},
-            params={"_flag": 8, "ticket": ticket, "trust": "false", "_json": "true"},
+            params={"_flag": flag, "ticket": ticket, "trust": "false", "_json": "true"},
         )
         res1 = parse_auth_response(await r.read())
         assert res1["code"] == 0, res1
@@ -162,33 +171,56 @@ async def _get_captcha_url(self, captcha_url: str) -> dict:
         body = await r.read()
         return {"image": body, "ick": r.cookies["ick"]}
 
-    async def _send_email_ticket(self, notification_url: str) -> dict:
+    async def _get_notification_url(self, notification_url: str) -> AuthResult:
         assert "/identity/authStart" in notification_url, notification_url
         notification_url = notification_url.replace("authStart", "list")
 
         r = await self.session.get(notification_url)
         res1 = parse_auth_response(await r.read())
         assert res1["code"] == 2, res1
 
-        identity_session = r.cookies["identity_session"]
+        flag = res1["flag"]
+        assert flag in (FLAG_EMAIL, FLAG_PHONE), res1
+
+        return await self._send_ticket(flag, r.cookies["identity_session"])
+
+    async def _send_ticket(
+        self, flag: int, identity_session, captcha_code: str = None
+    ) -> AuthResult:
+        key = "Phone" if flag == FLAG_PHONE else "Email"
 
         r = await self.session.get(
-            "https://account.xiaomi.com/identity/auth/verifyEmail",
+            f"https://account.xiaomi.com/identity/auth/verify{key}",
             cookies={"identity_session": identity_session},
-            params={"_flag": 8, "_json": "true"},
+            params={"_flag": flag, "_json": "true"},
         )
-        res2 = parse_auth_response(await r.read())
-        assert res2["code"] == 0, res2
+        res1 = parse_auth_response(await r.read())
+        assert res1["code"] == 0, res1
+
+        if captcha_code:
+            cookies = {"identity_session": identity_session, "ick": self.auth["ick"]}
+            data = {"retry": 0, "icode": captcha_code, "_json": "true"}
+        else:
+            cookies = {"identity_session": identity_session}
+            data = {"retry": 0, "icode": "", "_json": "true"}
 
         r = await self.session.post(
-            "https://account.xiaomi.com/identity/auth/sendEmailTicket",
-            cookies={"identity_session": identity_session},
-            data={"retry": 0, "icode": "", "_json": "true"},
+            f"https://account.xiaomi.com/identity/auth/send{key}Ticket",
+            cookies=cookies,
+            data=data,
         )
-        res3 = parse_auth_response(await r.read())
-        assert res3["code"] == 0, res3
+        res2 = parse_auth_response(await r.read())
+
+        self.auth = {"flag": flag, "identity_session": identity_session}
+
+        if captcha_url := res2.get("captchaUrl"):
+            data = await self._get_captcha_url(captcha_url)
+            self.auth["ick"] = data["ick"]
+            return {"ok": False, "captcha": data["image"]}
+
+        assert res2["code"] == 0, res2
 
-        return {"email": res2["maskedEmail"], "identity_session": identity_session}
+        return {"ok": False, "verify": res1[f"masked{key}"]}
 
     async def request(self, server: str, path: str, params: dict) -> dict:
         form: dict[str, str] = {"data": json.dumps(params, separators=(",", ":"))}

custom_components/xiaomi_gateway3/translations/en.json

@@ -28,14 +28,14 @@
       },
       "cloud_captcha": {
         "title": "Verify account",
-        "description": "Enter code from image: {image}",
+        "description": "Enter code from: ![]({image})",
         "data": {
           "code": "Code"
         }
       },
-      "cloud_email": {
+      "cloud_verify": {
         "title": "Verify account",
-        "description": "Enter code from email: {email}",
+        "description": "Enter code from: {address}",
         "data": {
           "code": "Code"
         }

custom_components/xiaomi_gateway3/translations/hu.json

@@ -27,14 +27,14 @@
       },
       "cloud_captcha": {
         "title": "Verify account",
-        "description": "Enter code from image: {image}",
+        "description": "Enter code from: ![]({image})",
         "data": {
           "code": "Code"
         }
       },
-      "cloud_email": {
+      "cloud_verify": {
         "title": "Verify account",
-        "description": "Enter code from email: {email}",
+        "description": "Enter code from: {address}",
         "data": {
           "code": "Code"
         }

custom_components/xiaomi_gateway3/translations/pl.json

@@ -27,14 +27,14 @@
       },
       "cloud_captcha": {
         "title": "Verify account",
-        "description": "Enter code from image: {image}",
+        "description": "Enter code from: ![]({image})",
         "data": {
           "code": "Code"
         }
       },
-      "cloud_email": {
+      "cloud_verify": {
         "title": "Verify account",
-        "description": "Enter code from email: {email}",
+        "description": "Enter code from: {address}",
         "data": {
           "code": "Code"
         }

custom_components/xiaomi_gateway3/translations/pt-BR.json

@@ -27,14 +27,14 @@
       },
       "cloud_captcha": {
         "title": "Verify account",
-        "description": "Enter code from image: {image}",
+        "description": "Enter code from: ![]({image})",
         "data": {
           "code": "Code"
         }
       },
-      "cloud_email": {
+      "cloud_verify": {
         "title": "Verify account",
-        "description": "Enter code from email: {email}",
+        "description": "Enter code from: {address}",
         "data": {
           "code": "Code"
         }

custom_components/xiaomi_gateway3/translations/ro.json

@@ -27,14 +27,14 @@
       },
       "cloud_captcha": {
         "title": "Verify account",
-        "description": "Enter code from image: {image}",
+        "description": "Enter code from: ![]({image})",
         "data": {
           "code": "Code"
         }
       },
-      "cloud_email": {
+      "cloud_verify": {
         "title": "Verify account",
-        "description": "Enter code from email: {email}",
+        "description": "Enter code from: {address}",
         "data": {
           "code": "Code"
         }

custom_components/xiaomi_gateway3/translations/ru.json

@@ -27,14 +27,14 @@
       },
       "cloud_captcha": {
         "title": "Проверка аккаунта",
-        "description": "Введите код с картинки: {image}",
+        "description": "Введите код из: ![]({image})",
         "data": {
           "code": "Код"
         }
       },
-      "cloud_email": {
+      "cloud_verify": {
         "title": "Проверка аккаунта",
-        "description": "Введите код из e-mail: {email}",
+        "description": "Введите код из: {address}",
         "data": {
           "code": "Код"
         }

custom_components/xiaomi_gateway3/translations/ua.json

@@ -27,14 +27,14 @@
       },
       "cloud_captcha": {
         "title": "Verify account",
-        "description": "Enter code from image: {image}",
+        "description": "Enter code from: ![]({image})",
         "data": {
           "code": "Code"
         }
       },
-      "cloud_email": {
+      "cloud_verify": {
         "title": "Verify account",
-        "description": "Enter code from email: {email}",
+        "description": "Enter code from: {address}",
         "data": {
           "code": "Code"
         }

custom_components/xiaomi_gateway3/translations/zh-Hans.json

@@ -27,14 +27,14 @@
       },
       "cloud_captcha": {
         "title": "Verify account",
-        "description": "Enter code from image: {image}",
+        "description": "Enter code from: ![]({image})",
         "data": {
           "code": "Code"
         }
       },
-      "cloud_email": {
+      "cloud_verify": {
         "title": "Verify account",
-        "description": "Enter code from email: {email}",
+        "description": "Enter code from: {address}",
         "data": {
           "code": "Code"
         }