210 lines
7.8 KiB
Python
210 lines
7.8 KiB
Python
"""Config flow for MB-Secure."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Any, override
|
|
|
|
import voluptuous as vol
|
|
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
|
|
from homeassistant.const import CONF_HOST, CONF_PORT
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
from homeassistant.helpers.service_info.hassio import HassioServiceInfo
|
|
|
|
from .api import (
|
|
BridgeAuthenticationError,
|
|
BridgeClient,
|
|
BridgeConnectionError,
|
|
BridgeResponseError,
|
|
BridgeVersionError,
|
|
)
|
|
from .const import (
|
|
CONF_TOKEN,
|
|
DEFAULT_HOST,
|
|
DEFAULT_PORT,
|
|
DOMAIN,
|
|
MIN_TOKEN_LENGTH,
|
|
)
|
|
from .models import BridgeInfo
|
|
|
|
|
|
class MBSecureConfigFlow(ConfigFlow, domain=DOMAIN):
|
|
"""Handle MB-Secure configuration."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize flow state."""
|
|
self._discovery_data: dict[str, Any] | None = None
|
|
|
|
@override
|
|
async def async_step_user(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Handle manual setup."""
|
|
errors: dict[str, str] = {}
|
|
if user_input is not None:
|
|
error, info = await self._async_validate(user_input)
|
|
if error is None:
|
|
assert info is not None
|
|
if info.bridge_id is None:
|
|
errors["base"] = "missing_bridge_id"
|
|
return self._show_user_form(user_input, errors)
|
|
await self.async_set_unique_id(info.bridge_id)
|
|
self._abort_if_unique_id_configured()
|
|
return self.async_create_entry(
|
|
title="MB-Secure Bridge", data=user_input
|
|
)
|
|
else:
|
|
errors["base"] = error
|
|
|
|
return self._show_user_form(user_input, errors)
|
|
|
|
def _show_user_form(
|
|
self,
|
|
user_input: dict[str, Any] | None,
|
|
errors: dict[str, str],
|
|
) -> ConfigFlowResult:
|
|
"""Show the manual configuration form."""
|
|
schema = vol.Schema(
|
|
{
|
|
vol.Required(
|
|
CONF_HOST,
|
|
default=(user_input or {}).get(CONF_HOST, DEFAULT_HOST),
|
|
): str,
|
|
vol.Required(
|
|
CONF_PORT,
|
|
default=(user_input or {}).get(CONF_PORT, DEFAULT_PORT),
|
|
): vol.All(vol.Coerce(int), vol.Range(min=1, max=65535)),
|
|
vol.Required(CONF_TOKEN): vol.All(
|
|
str, vol.Length(min=MIN_TOKEN_LENGTH)
|
|
),
|
|
}
|
|
)
|
|
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
|
|
|
|
@override
|
|
async def async_step_hassio(
|
|
self, discovery_info: HassioServiceInfo
|
|
) -> ConfigFlowResult:
|
|
"""Handle discovery from the MB-Secure Bridge app."""
|
|
config = discovery_info.config
|
|
if not all(key in config for key in (CONF_HOST, CONF_PORT, CONF_TOKEN)):
|
|
return self.async_abort(reason="invalid_discovery")
|
|
if not isinstance(config[CONF_HOST], str) or not isinstance(
|
|
config[CONF_TOKEN], str
|
|
):
|
|
return self.async_abort(reason="invalid_discovery")
|
|
if isinstance(config[CONF_PORT], bool):
|
|
return self.async_abort(reason="invalid_discovery")
|
|
try:
|
|
host = config[CONF_HOST]
|
|
port = int(config[CONF_PORT])
|
|
token = config[CONF_TOKEN]
|
|
except (TypeError, ValueError):
|
|
return self.async_abort(reason="invalid_discovery")
|
|
if not host or len(token) < MIN_TOKEN_LENGTH or not 1 <= port <= 65535:
|
|
return self.async_abort(reason="invalid_discovery")
|
|
self._discovery_data = {
|
|
CONF_HOST: host,
|
|
CONF_PORT: port,
|
|
CONF_TOKEN: token,
|
|
}
|
|
await self.async_set_unique_id(discovery_info.uuid)
|
|
self._abort_if_unique_id_configured(updates={CONF_HOST: host, CONF_PORT: port})
|
|
return await self.async_step_hassio_confirm()
|
|
|
|
async def async_step_hassio_confirm(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Confirm Supervisor discovery before creating an entry."""
|
|
if self._discovery_data is None:
|
|
return self.async_abort(reason="invalid_discovery")
|
|
errors: dict[str, str] = {}
|
|
if user_input is not None:
|
|
error, info = await self._async_validate(self._discovery_data)
|
|
if error is None:
|
|
assert info is not None
|
|
if info.bridge_id is None:
|
|
errors["base"] = "missing_bridge_id"
|
|
else:
|
|
await self.async_set_unique_id(info.bridge_id)
|
|
self._abort_if_unique_id_configured(
|
|
updates={
|
|
CONF_HOST: self._discovery_data[CONF_HOST],
|
|
CONF_PORT: self._discovery_data[CONF_PORT],
|
|
}
|
|
)
|
|
return self.async_create_entry(
|
|
title="MB-Secure Bridge", data=self._discovery_data
|
|
)
|
|
else:
|
|
errors["base"] = error
|
|
return self.async_show_form(
|
|
step_id="hassio_confirm",
|
|
data_schema=vol.Schema({}),
|
|
errors=errors,
|
|
description_placeholders={"host": self._discovery_data[CONF_HOST]},
|
|
)
|
|
|
|
async def async_step_reauth(
|
|
self, entry_data: Mapping[str, Any]
|
|
) -> ConfigFlowResult:
|
|
"""Start reauthentication."""
|
|
self._discovery_data = dict(entry_data)
|
|
return await self.async_step_reauth_confirm()
|
|
|
|
async def async_step_reauth_confirm(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Replace a rejected local bridge token."""
|
|
errors: dict[str, str] = {}
|
|
if user_input is not None:
|
|
reauth_entry = self._get_reauth_entry()
|
|
candidate = {**reauth_entry.data, CONF_TOKEN: user_input[CONF_TOKEN]}
|
|
error, info = await self._async_validate(candidate)
|
|
if error is None:
|
|
assert info is not None
|
|
if info.bridge_id is None:
|
|
errors["base"] = "missing_bridge_id"
|
|
return self._show_reauth_form(errors)
|
|
await self.async_set_unique_id(info.bridge_id)
|
|
self._abort_if_unique_id_mismatch(reason="wrong_account")
|
|
return self.async_update_reload_and_abort(
|
|
reauth_entry,
|
|
data_updates={CONF_TOKEN: user_input[CONF_TOKEN]},
|
|
)
|
|
errors["base"] = error
|
|
return self._show_reauth_form(errors)
|
|
|
|
def _show_reauth_form(self, errors: dict[str, str]) -> ConfigFlowResult:
|
|
"""Show the bridge token replacement form."""
|
|
return self.async_show_form(
|
|
step_id="reauth_confirm",
|
|
data_schema=vol.Schema(
|
|
{
|
|
vol.Required(CONF_TOKEN): vol.All(
|
|
str, vol.Length(min=MIN_TOKEN_LENGTH)
|
|
)
|
|
}
|
|
),
|
|
errors=errors,
|
|
)
|
|
|
|
async def _async_validate(
|
|
self, data: dict[str, Any]
|
|
) -> tuple[str | None, BridgeInfo | None]:
|
|
try:
|
|
client = BridgeClient(
|
|
async_get_clientsession(self.hass),
|
|
host=data[CONF_HOST],
|
|
port=data[CONF_PORT],
|
|
token=data[CONF_TOKEN],
|
|
)
|
|
info = await client.async_get_info()
|
|
except BridgeAuthenticationError:
|
|
return "invalid_auth", None
|
|
except BridgeVersionError:
|
|
return "unsupported_version", None
|
|
except (BridgeConnectionError, BridgeResponseError, ValueError):
|
|
return "cannot_connect", None
|
|
return None, info
|