feat: introduce otp login flow for us region

README.md

@@ -61,18 +61,18 @@ These can be access by going to the developer tools followed by actions. These c
 
 | Service                           | EU  | EU(>2023), NZ, AU | CA  | USA Kia | USA Hyundai | USA Genesis | China | India | Brazil |
 | --------------------------------- | --- | ----------------- | --- | ------- | ----------- | ----------- | ----- | ----- | ------ |
-| Update                            | ✔  | ✔                | ✔  | ✔      | ✔          | ✔          | ✔    | ✔    | ✔     |
-| Force Update                      | ✔  | not tested        | ✔  | ✔      |             |             | ✔    | ✔    | ✔     |
-| Lock Unlock                       | ✔  | ✔                | ✔  | ✔      | ✔          | ✔          | ✔    | ✔    | ✖     |
-| start stop climate                | ✔  | ✔                | ✔  | ✔      | ✔          |             | ✔    | ✔    | ✖     |
-| start stop charge                 | ✔  | ✔                | ✔  | ✔      | ✔          |             |       |       | ✖     |
-| set charge limits                 | ✔  | ✔                | ✔  | ✔      | ✔          |             |       |       | ✖     |
-| open and close charge port        | ✖  | ✔                | ✖  | ✖      | ✖          | ✖          | ✖    |       | ✖     |
-| set windows                       | ✖  | ✔                | ✖  | ✖      | ✖          | ✖          | ✖    |       | ✖     |
-| start stop hazard lights          | ✖  | ✔                | ✖  | ✖      | ✖          | ✖          | ✖    |       | ✖     |
-| start stop hazard lights and horn | ✖  | ✔                | ✖  | ✖      | ✖          | ✖          | ✖    | ✔    | ✖     |
-| schedule charging and climate     | ✖  | ✔                | ✖  | ✖      | ✖          | ✖          | ✖    |       | ✖     |
-| start stop valet_mode             | ✖  | ✔                | ✖  | ✖      | ✖          | ✖          | ✖    |       | ✖     |
+| Update                            | ✔   | ✔                 | ✔   | ✔       | ✔           | ✔           | ✔     | ✔     | ✔      |
+| Force Update                      | ✔   | not tested        | ✔   | ✔       |             |             | ✔     | ✔     | ✔      |
+| Lock Unlock                       | ✔   | ✔                 | ✔   | ✔       | ✔           | ✔           | ✔     | ✔     | ✖      |
+| start stop climate                | ✔   | ✔                 | ✔   | ✔       | ✔           |             | ✔     | ✔     | ✖      |
+| start stop charge                 | ✔   | ✔                 | ✔   | ✔       | ✔           |             |       |       | ✖      |
+| set charge limits                 | ✔   | ✔                 | ✔   | ✔       | ✔           |             |       |       | ✖      |
+| open and close charge port        | ✖   | ✔                 | ✖   | ✖       | ✖           | ✖           | ✖     |       | ✖      |
+| set windows                       | ✖   | ✔                 | ✖   | ✖       | ✖           | ✖           | ✖     |       | ✖      |
+| start stop hazard lights          | ✖   | ✔                 | ✖   | ✖       | ✖           | ✖           | ✖     |       | ✖      |
+| start stop hazard lights and horn | ✖   | ✔                 | ✖   | ✖       | ✖           | ✖           | ✖     | ✔     | ✖      |
+| schedule charging and climate     | ✖   | ✔                 | ✖   | ✖       | ✖           | ✖           | ✖     |       | ✖      |
+| start stop valet_mode             | ✖   | ✔                 | ✖   | ✖       | ✖           | ✖           | ✖     |       | ✖      |
 
 I have posted an example screenshot from my own car.
 

custom_components/kia_uvo/config_flow.py

@@ -6,7 +6,7 @@
 import logging
 from typing import Any
 
-from hyundai_kia_connect_api import Token, VehicleManager
+from hyundai_kia_connect_api import VehicleManager, Token
 from hyundai_kia_connect_api.exceptions import AuthenticationError
 import voluptuous as vol
 
@@ -35,12 +35,15 @@
     DEFAULT_PIN,
     DEFAULT_SCAN_INTERVAL,
     DOMAIN,
+    CONF_RMTOKEN,
+    CONF_DEVICE_ID,
     REGIONS,
     CONF_ENABLE_GEOLOCATION_ENTITY,
     CONF_USE_EMAIL_WITH_GEOCODE_API,
     DEFAULT_ENABLE_GEOLOCATION_ENTITY,
     DEFAULT_USE_EMAIL_WITH_GEOCODE_API,
     REGION_EUROPE,
+    REGION_USA,
     BRAND_HYUNDAI,
     BRAND_KIA,
 )
@@ -151,6 +154,10 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
     def __init__(self):
         """Initialize the config flow."""
         self._region_data = None
+        self._pending_config: dict[str, Any] | None = None
+        self._otp_notify_type: str = "EMAIL"
+        self._api = None
+        self._otp_ctx: dict[str, Any] | None = None
 
     @staticmethod
     @callback
@@ -181,13 +188,13 @@ async def async_step_credentials_password(
     ) -> FlowResult:
         """Handle the credentials step."""
         errors = {}
-
         if user_input is not None:
-            # Combine region data with credentials
             full_config = {**self._region_data, **user_input}
-
             try:
-                await validate_input(self.hass, full_config)
+                if REGIONS[self._region_data[CONF_REGION]] == REGION_USA:
+                    self._pending_config = full_config
+                    return await self.async_step_send_otp()
+                token: Token = await validate_input(self.hass, full_config)
             except InvalidAuth:
                 errors["base"] = "invalid_auth"
             except Exception:  # pylint: disable=broad-except
@@ -200,16 +207,25 @@ async def async_step_credentials_password(
                         hashlib.sha256(title.encode("utf-8")).hexdigest()
                     )
                     self._abort_if_unique_id_configured()
-                    return self.async_create_entry(title=title, data=full_config)
+                    data = {
+                        **full_config,
+                        CONF_RMTOKEN: getattr(token, "refresh_token", None),
+                        CONF_DEVICE_ID: getattr(token, "device_id", None),
+                    }
+                    return self.async_create_entry(title=title, data=data)
                 else:
+                    data = {
+                        **full_config,
+                        CONF_RMTOKEN: getattr(token, "refresh_token", None),
+                        CONF_DEVICE_ID: getattr(token, "device_id", None),
+                    }
                     self.hass.config_entries.async_update_entry(
-                        self.reauth_entry, data=full_config
+                        self.reauth_entry, data=data
                     )
                     await self.hass.config_entries.async_reload(
                         self.reauth_entry.entry_id
                     )
                     return self.async_abort(reason="reauth_successful")
-
         return self.async_show_form(
             step_id="credentials_password",
             data_schema=STEP_CREDENTIALS_DATA_SCHEMA,
@@ -221,41 +237,221 @@ async def async_step_credentials_token(
     ) -> FlowResult:
         """Handle the credentials step."""
         errors = {}
-
         if user_input is not None:
-            # Combine region data with credentials
             full_config = {**self._region_data, **user_input}
-
             try:
-                await validate_input(self.hass, full_config)
+                token: Token = await validate_input(self.hass, full_config)
             except InvalidAuth:
                 errors["base"] = "invalid_auth"
             except Exception:  # pylint: disable=broad-except
                 _LOGGER.exception("Unexpected exception")
                 errors["base"] = "unknown"
             else:
+                data = {
+                    **full_config,
+                    CONF_RMTOKEN: getattr(token, "refresh_token", None),
+                    CONF_DEVICE_ID: getattr(token, "device_id", None),
+                }
                 if self.reauth_entry is None:
                     title = f"{BRANDS[self._region_data[CONF_BRAND]]} {REGIONS[self._region_data[CONF_REGION]]} {user_input[CONF_USERNAME]}"
                     await self.async_set_unique_id(
                         hashlib.sha256(title.encode("utf-8")).hexdigest()
                     )
                     self._abort_if_unique_id_configured()
-                    return self.async_create_entry(title=title, data=full_config)
+                    return self.async_create_entry(title=title, data=data)
                 else:
                     self.hass.config_entries.async_update_entry(
-                        self.reauth_entry, data=full_config
+                        self.reauth_entry, data=data
                     )
                     await self.hass.config_entries.async_reload(
                         self.reauth_entry.entry_id
                     )
                     return self.async_abort(reason="reauth_successful")
-
         return self.async_show_form(
             step_id="credentials_token",
             data_schema=STEP_CREDENTIALS_DATA_SCHEMA,
             errors=errors,
         )
 
+    async def async_step_send_otp(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Send OTP to selected destination for USA region.
+
+        Parameters
+        ----------
+        user_input : dict[str, Any] | None
+            Form submission captured by the step, or None to render the form.
+
+        Returns
+        -------
+        FlowResult
+            The next step in the flow or a created entry.
+        """
+        errors = {}
+        schema = vol.Schema(
+            {vol.Required("notify_type", default="EMAIL"): vol.In(["EMAIL", "PHONE"])}
+        )
+        if user_input is None:
+            return self.async_show_form(
+                step_id="send_otp", data_schema=schema, errors=errors
+            )
+        self._otp_notify_type = str(user_input["notify_type"]).upper()
+        cfg = self._pending_config
+        try:
+            if self._api is None:
+                self._api = VehicleManager.get_implementation_by_region_brand(
+                    cfg[CONF_REGION],
+                    cfg[CONF_BRAND],
+                    language=self.hass.config.language,
+                )
+            token, ctx = await self.hass.async_add_executor_job(
+                self._api.start_login, cfg[CONF_USERNAME], cfg[CONF_PASSWORD]
+            )
+        except InvalidAuth:
+            errors["base"] = "invalid_auth"
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception("Unexpected exception during start_login")
+            errors["base"] = "unknown"
+        else:
+            if token is not None:
+                data = {
+                    **cfg,
+                    CONF_RMTOKEN: getattr(token, "refresh_token", None),
+                    CONF_DEVICE_ID: getattr(token, "device_id", None),
+                }
+                if self.reauth_entry is None:
+                    title = f"{BRANDS[self._region_data[CONF_BRAND]]} {REGIONS[self._region_data[CONF_REGION]]} {cfg[CONF_USERNAME]}"
+                    await self.async_set_unique_id(
+                        hashlib.sha256(title.encode("utf-8")).hexdigest()
+                    )
+                    self._abort_if_unique_id_configured()
+                    return self.async_create_entry(title=title, data=data)
+                self.hass.config_entries.async_update_entry(
+                    self.reauth_entry, data=data
+                )
+                await self.hass.config_entries.async_reload(self.reauth_entry.entry_id)
+                return self.async_abort(reason="reauth_successful")
+            self._otp_ctx = ctx or {}
+            try:
+                await self.hass.async_add_executor_job(
+                    self._api.send_otp,
+                    self._otp_ctx.get("otpKey"),
+                    self._otp_notify_type,
+                    self._otp_ctx.get("xid", ""),
+                )
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected exception during send_otp")
+                errors["base"] = "unknown"
+            else:
+                return await self.async_step_input_otp_code()
+        return self.async_show_form(
+            step_id="send_otp", data_schema=schema, errors=errors
+        )
+
+    def _login_send_otp(self, cfg: dict[str, Any], notify_type: str) -> Token | None:
+        """Call login to send OTP and stop before code verification.
+
+        Parameters
+        ----------
+        cfg : dict[str, Any]
+            Combined configuration containing region, brand, username, and password.
+        notify_type : str
+            Destination for OTP delivery, either 'EMAIL' or 'PHONE'.
+
+        Returns
+        -------
+        Token | None
+            Token if login succeeded without OTP (rmtoken reuse), otherwise None.
+        """
+        api = VehicleManager.get_implementation_by_region_brand(
+            cfg[CONF_REGION], cfg[CONF_BRAND], language=self.hass.config.language
+        )
+
+        def handler(ctx: dict) -> dict:
+            if ctx.get("stage") == "choose_destination":
+                return {"notify_type": notify_type}
+            if ctx.get("stage") == "input_code":
+                return {"otp_code": ""}
+            return {}
+
+        try:
+            return api.login(
+                cfg[CONF_USERNAME], cfg[CONF_PASSWORD], otp_handler=handler
+            )
+        except AuthenticationError:
+            return None
+
+    async def async_step_input_otp_code(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Collect OTP code from user and complete login.
+
+        Parameters
+        ----------
+        user_input : dict[str, Any] | None
+            Form submission containing the `otp_code`, or None to render the form.
+
+        Returns
+        -------
+        FlowResult
+            The next step in the flow or a created entry.
+        """
+        errors = {}
+        schema = vol.Schema({vol.Required("otp_code"): str})
+        if user_input is None:
+            return self.async_show_form(
+                step_id="input_otp_code", data_schema=schema, errors=errors
+            )
+        cfg = self._pending_config
+        if self._api is None:
+            self._api = VehicleManager.get_implementation_by_region_brand(
+                cfg[CONF_REGION], cfg[CONF_BRAND], language=self.hass.config.language
+            )
+        try:
+            token: Token = await self.hass.async_add_executor_job(
+                self._api.verify_otp_and_complete_login,
+                cfg[CONF_USERNAME],
+                cfg[CONF_PASSWORD],
+                self._otp_ctx.get("otpKey"),
+                self._otp_ctx.get("xid", ""),
+                user_input["otp_code"],
+            )
+        except InvalidAuth:
+            errors["base"] = "invalid_auth"
+        except AuthenticationError as err:
+            msg = str(err)
+            if "otp" in msg.lower():
+                errors["base"] = "invalid_otp"
+            else:
+                errors["base"] = "invalid_auth"
+        except Exception as err:  # pylint: disable=broad-except
+            msg = str(err)
+            if "otp" in msg.lower():
+                errors["base"] = "invalid_otp"
+            else:
+                _LOGGER.exception("Unexpected exception during OTP verify")
+                errors["base"] = "unknown"
+        else:
+            data = {
+                **cfg,
+                CONF_RMTOKEN: getattr(token, "refresh_token", None),
+                CONF_DEVICE_ID: getattr(token, "device_id", None),
+            }
+            if self.reauth_entry is None:
+                title = f"{BRANDS[self._region_data[CONF_BRAND]]} {REGIONS[self._region_data[CONF_REGION]]} {cfg[CONF_USERNAME]}"
+                await self.async_set_unique_id(
+                    hashlib.sha256(title.encode("utf-8")).hexdigest()
+                )
+                self._abort_if_unique_id_configured()
+                return self.async_create_entry(title=title, data=data)
+            self.hass.config_entries.async_update_entry(self.reauth_entry, data=data)
+            await self.hass.config_entries.async_reload(self.reauth_entry.entry_id)
+            return self.async_abort(reason="reauth_successful")
+        return self.async_show_form(
+            step_id="input_otp_code", data_schema=schema, errors=errors
+        )
+
     async def async_step_reauth(self, user_input=None):
         """Perform reauth upon an API authentication error."""
         self.reauth_entry = self.hass.config_entries.async_get_entry(

custom_components/kia_uvo/const.py

@@ -9,6 +9,10 @@
 CONF_ENABLE_GEOLOCATION_ENTITY: str = "enable_geolocation_entity"
 CONF_USE_EMAIL_WITH_GEOCODE_API: str = "use_email_with_geocode_api"
 
+CONF_RMTOKEN: str = "rmtoken"
+CONF_DEVICE_ID: str = "device_id"
+
+
 REGION_EUROPE: str = "Europe"
 REGION_CANADA: str = "Canada"
 REGION_USA: str = "USA"