config_flow.py

Go to the documentation of this file.

"""Config flow to configure StarLine component."""
 
from __future__ import annotations
 
from starline import StarlineAuth
import voluptuous as vol
 
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import callback
 
from .const import (
    _LOGGER,
    CONF_APP_ID,
    CONF_APP_SECRET,
    CONF_CAPTCHA_CODE,
    CONF_MFA_CODE,
    DATA_EXPIRES,
    DATA_SLID_TOKEN,
    DATA_SLNET_TOKEN,
    DATA_USER_ID,
    DOMAIN,
    ERROR_AUTH_APP,
    ERROR_AUTH_MFA,
    ERROR_AUTH_USER,
)
 
 
class StarlineFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle a StarLine config flow."""
 
    VERSION = 1
 
    _app_code: str
    _app_token: str
    _captcha_image: str
    _phone_number: str
 
    def __init__(self) -> None:
        """Initialize flow."""
        self._app_id_app_id: str | None = None
        self._app_secret_app_secret: str | None = None
        self._username_username: str | None = None
        self._password_password: str | None = None
        self._mfa_code_mfa_code: str | None = None
 
        self._user_slid_user_slid = None
        self._user_id_user_id = None
        self._slnet_token_slnet_token = None
        self._slnet_token_expires_slnet_token_expires = None
        self._captcha_sid_captcha_sid: str | None = None
        self._captcha_code_captcha_code: str | None = None
 
        self._auth_auth = StarlineAuth()
 
    async def async_step_user(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initialized by the user."""
        return await self.async_step_auth_appasync_step_auth_app(user_input)
 
    async def async_step_auth_app(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Authenticate application step."""
        if user_input is not None:
            self._app_id_app_id = user_input[CONF_APP_ID]
            self._app_secret_app_secret = user_input[CONF_APP_SECRET]
            return await self._async_authenticate_app_async_authenticate_app()
        return self._async_form_auth_app_async_form_auth_app()
 
    async def async_step_auth_user(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Authenticate user step."""
        if user_input is not None:
            self._username_username = user_input[CONF_USERNAME]
            self._password_password = user_input[CONF_PASSWORD]
            return await self._async_authenticate_user_async_authenticate_user()
        return self._async_form_auth_user_async_form_auth_user()
 
    async def async_step_auth_mfa(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Authenticate mfa step."""
        if user_input is not None:
            self._mfa_code_mfa_code = user_input[CONF_MFA_CODE]
            return await self._async_authenticate_user_async_authenticate_user()
        return self._async_form_auth_mfa_async_form_auth_mfa()
 
    async def async_step_auth_captcha(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Captcha verification step."""
        if user_input is not None:
            self._captcha_code_captcha_code = user_input[CONF_CAPTCHA_CODE]
            return await self._async_authenticate_user_async_authenticate_user()
        return self._async_form_auth_captcha_async_form_auth_captcha()
 
    @callback
    def _async_form_auth_app(self, error: str | None = None) -> ConfigFlowResult:
        """Authenticate application form."""
        errors: dict[str, str] = {}
        if error is not None:
            errors["base"] = error
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="auth_app",
            data_schema=vol.Schema(
                {
                    vol.Required(
                        CONF_APP_ID, default=self._app_id_app_id or vol.UNDEFINED
                    ): str,
                    vol.Required(
                        CONF_APP_SECRET, default=self._app_secret_app_secret or vol.UNDEFINED
                    ): str,
                }
            ),
            errors=errors,
        )
 
    @callback
    def _async_form_auth_user(self, error: str | None = None) -> ConfigFlowResult:
        """Authenticate user form."""
        errors = {}
        if error is not None:
            errors["base"] = error
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="auth_user",
            data_schema=vol.Schema(
                {
                    vol.Required(
                        CONF_USERNAME, default=self._username_username or vol.UNDEFINED
                    ): str,
                    vol.Required(
                        CONF_PASSWORD, default=self._password_password or vol.UNDEFINED
                    ): str,
                }
            ),
            errors=errors,
        )
 
    @callback
    def _async_form_auth_mfa(self, error: str | None = None) -> ConfigFlowResult:
        """Authenticate mfa form."""
        errors = {}
        if error is not None:
            errors["base"] = error
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="auth_mfa",
            data_schema=vol.Schema(
                {
                    vol.Required(
                        CONF_MFA_CODE, default=self._mfa_code_mfa_code or vol.UNDEFINED
                    ): str
                }
            ),
            errors=errors,
            description_placeholders={"phone_number": self._phone_number_phone_number},
        )
 
    @callback
    def _async_form_auth_captcha(self, error: str | None = None) -> ConfigFlowResult:
        """Captcha verification form."""
        errors = {}
        if error is not None:
            errors["base"] = error
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="auth_captcha",
            data_schema=vol.Schema(
                {
                    vol.Required(
                        CONF_CAPTCHA_CODE, default=self._captcha_code_captcha_code or vol.UNDEFINED
                    ): str
                }
            ),
            errors=errors,
            description_placeholders={
                "captcha_img": '<img src="' + self._captcha_image_captcha_image + '"/>'
            },
        )
 
    async def _async_authenticate_app(
        self, error: str | None = None
    ) -> ConfigFlowResult:
        """Authenticate application."""
        try:
            self._app_code_app_code = await self.hass.async_add_executor_job(
                self._auth_auth.get_app_code, self._app_id_app_id, self._app_secret_app_secret
            )
            self._app_token_app_token = await self.hass.async_add_executor_job(
                self._auth_auth.get_app_token, self._app_id_app_id, self._app_secret_app_secret, self._app_code_app_code
            )
            return self._async_form_auth_user_async_form_auth_user(error)
        except Exception as err:  # noqa: BLE001
            _LOGGER.error("Error auth StarLine: %s", err)
            return self._async_form_auth_app_async_form_auth_app(ERROR_AUTH_APP)
 
    async def _async_authenticate_user(
        self, error: str | None = None
    ) -> ConfigFlowResult:
        """Authenticate user."""
        try:
            state, data = await self.hass.async_add_executor_job(
                self._auth_auth.get_slid_user_token,
                self._app_token_app_token,
                self._username_username,
                self._password_password,
                self._mfa_code_mfa_code,
                self._captcha_sid_captcha_sid,
                self._captcha_code_captcha_code,
            )
 
            if state == 1:
                self._user_slid_user_slid = data["user_token"]
                return await self._async_get_entry_async_get_entry()
 
            if "phone" in data:
                self._phone_number_phone_number = data["phone"]
                if state == 0:
                    error = ERROR_AUTH_MFA
                return self._async_form_auth_mfa_async_form_auth_mfa(error)
 
            if "captchaSid" in data:
                self._captcha_sid_captcha_sid = data["captchaSid"]
                self._captcha_image_captcha_image = data["captchaImg"]
                return self._async_form_auth_captcha_async_form_auth_captcha(error)
 
            raise Exception(data)  # noqa: TRY002, TRY301
        except Exception as err:  # noqa: BLE001
            _LOGGER.error("Error auth user: %s", err)
            return self._async_form_auth_user_async_form_auth_user(ERROR_AUTH_USER)
 
    async def _async_get_entry(self) -> ConfigFlowResult:
        """Create entry."""
        (
            self._slnet_token_slnet_token,
            self._slnet_token_expires_slnet_token_expires,
            self._user_id_user_id,
        ) = await self.hass.async_add_executor_job(
            self._auth_auth.get_user_id, self._user_slid_user_slid
        )
 
        return self.async_create_entryasync_create_entryasync_create_entry(
            title=f"Application {self._app_id}",
            data={
                DATA_USER_ID: self._user_id_user_id,
                DATA_SLNET_TOKEN: self._slnet_token_slnet_token,
                DATA_SLID_TOKEN: self._user_slid_user_slid,
                DATA_EXPIRES: self._slnet_token_expires_slnet_token_expires,
            },
        )