Add basic auth to Blebox (#99320)

Co-authored-by: Robert Resch <robert@resch.dev>
This commit is contained in:
Michał Huryn 2023-10-11 16:09:56 +02:00 committed by GitHub
parent c66f0e3305
commit ddfad75eb7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 93 additions and 10 deletions

View file

@ -8,14 +8,20 @@ from blebox_uniapi.feature import Feature
from blebox_uniapi.session import ApiHost
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, Platform
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from .const import DEFAULT_SETUP_TIMEOUT, DOMAIN, PRODUCT
from .helpers import get_maybe_authenticated_session
_LOGGER = logging.getLogger(__name__)
@ -36,12 +42,16 @@ _FeatureT = TypeVar("_FeatureT", bound=Feature)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up BleBox devices from a config entry."""
websession = async_get_clientsession(hass)
host = entry.data[CONF_HOST]
port = entry.data[CONF_PORT]
username = entry.data.get(CONF_USERNAME)
password = entry.data.get(CONF_PASSWORD)
timeout = DEFAULT_SETUP_TIMEOUT
websession = get_maybe_authenticated_session(hass, password, username)
api_host = ApiHost(host, port, timeout, websession, hass.loop)
try:

View file

@ -5,16 +5,22 @@ import logging
from typing import Any
from blebox_uniapi.box import Box
from blebox_uniapi.error import Error, UnsupportedBoxResponse, UnsupportedBoxVersion
from blebox_uniapi.error import (
Error,
UnauthorizedRequest,
UnsupportedBoxResponse,
UnsupportedBoxVersion,
)
from blebox_uniapi.session import ApiHost
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import zeroconf
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from . import get_maybe_authenticated_session
from .const import (
ADDRESS_ALREADY_CONFIGURED,
CANNOT_CONNECT,
@ -46,6 +52,8 @@ def create_schema(previous_input=None):
{
vol.Required(CONF_HOST, default=host): str,
vol.Required(CONF_PORT, default=port): int,
vol.Inclusive(CONF_USERNAME, "auth"): str,
vol.Inclusive(CONF_PASSWORD, "auth"): str,
}
)
@ -153,6 +161,9 @@ class BleBoxConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
addr = host_port(user_input)
username = user_input.get(CONF_USERNAME)
password = user_input.get(CONF_PASSWORD)
for entry in self._async_current_entries():
if addr == host_port(entry.data):
host, port = addr
@ -160,7 +171,9 @@ class BleBoxConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
reason=ADDRESS_ALREADY_CONFIGURED,
description_placeholders={"address": f"{host}:{port}"},
)
websession = async_get_clientsession(hass)
websession = get_maybe_authenticated_session(hass, password, username)
api_host = ApiHost(*addr, DEFAULT_SETUP_TIMEOUT, websession, hass.loop, _LOGGER)
try:
product = await Box.async_from_host(api_host)
@ -169,6 +182,10 @@ class BleBoxConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.handle_step_exception(
"user", ex, schema, *addr, UNSUPPORTED_VERSION, _LOGGER.debug
)
except UnauthorizedRequest as ex:
return self.handle_step_exception(
"user", ex, schema, *addr, CANNOT_CONNECT, _LOGGER.error
)
except Error as ex:
return self.handle_step_exception(

View file

@ -0,0 +1,21 @@
"""Blebox helpers."""
from __future__ import annotations
import aiohttp
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import (
async_create_clientsession,
async_get_clientsession,
)
def get_maybe_authenticated_session(
hass: HomeAssistant, password: str | None, username: str | None
) -> aiohttp.ClientSession:
"""Return proper session object."""
if username and password:
auth = aiohttp.BasicAuth(login=username, password=password)
return async_create_clientsession(hass, auth=auth)
return async_get_clientsession(hass)

View file

@ -6,6 +6,6 @@
"documentation": "https://www.home-assistant.io/integrations/blebox",
"iot_class": "local_polling",
"loggers": ["blebox_uniapi"],
"requirements": ["blebox-uniapi==2.1.4"],
"requirements": ["blebox-uniapi==2.2.0"],
"zeroconf": ["_bbxsrv._tcp.local."]
}

View file

@ -530,7 +530,7 @@ bleak-retry-connector==3.2.1
bleak==0.21.1
# homeassistant.components.blebox
blebox-uniapi==2.1.4
blebox-uniapi==2.2.0
# homeassistant.components.blink
blinkpy==0.21.0

View file

@ -451,7 +451,7 @@ bleak-retry-connector==3.2.1
bleak==0.21.1
# homeassistant.components.blebox
blebox-uniapi==2.1.4
blebox-uniapi==2.2.0
# homeassistant.components.blink
blinkpy==0.21.0

View file

@ -153,6 +153,21 @@ async def test_flow_with_unsupported_version(
assert result["errors"] == {"base": "unsupported_version"}
async def test_flow_with_auth_failure(hass: HomeAssistant, product_class_mock) -> None:
"""Test that config flow works."""
with product_class_mock as products_class:
products_class.async_from_host = AsyncMock(
side_effect=blebox_uniapi.error.UnauthorizedRequest
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={config_flow.CONF_HOST: "172.2.3.4", config_flow.CONF_PORT: 80},
)
assert result["errors"] == {"base": "cannot_connect"}
async def test_async_setup(hass: HomeAssistant) -> None:
"""Test async_setup (for coverage)."""
assert await async_setup_component(hass, "blebox", {"host": "172.2.3.4"})

View file

@ -0,0 +1,20 @@
"""Blebox helpers tests."""
from aiohttp.helpers import BasicAuth
from homeassistant.components.blebox.helpers import get_maybe_authenticated_session
from homeassistant.core import HomeAssistant
async def test_get_maybe_authenticated_session_none(hass: HomeAssistant):
"""Tests if session auth is None."""
session = get_maybe_authenticated_session(hass=hass, username="", password="")
assert session.auth is None
async def test_get_maybe_authenticated_session_auth(hass: HomeAssistant):
"""Tests if session have BasicAuth."""
session = get_maybe_authenticated_session(
hass=hass, username="user", password="password"
)
assert isinstance(session.auth, BasicAuth)