From 2bea77549d02c61211decd62b7a5a0c8ee2e293a Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Mon, 7 Nov 2022 05:09:36 -0600 Subject: [PATCH] Fix creating multiple ElkM1 systems with TLS 1.2 (#81627) fixes https://github.com/home-assistant/core/issues/81516 --- homeassistant/components/elkm1/__init__.py | 9 +- homeassistant/components/elkm1/config_flow.py | 13 +- tests/components/elkm1/test_config_flow.py | 146 ++++++++++++++++++ 3 files changed, 159 insertions(+), 9 deletions(-) diff --git a/homeassistant/components/elkm1/__init__.py b/homeassistant/components/elkm1/__init__.py index 7833bfd66b3..7b14c7e85ed 100644 --- a/homeassistant/components/elkm1/__init__.py +++ b/homeassistant/components/elkm1/__init__.py @@ -7,11 +7,11 @@ import logging import re from types import MappingProxyType from typing import Any, cast -from urllib.parse import urlparse import async_timeout from elkm1_lib.elements import Element from elkm1_lib.elk import Elk +from elkm1_lib.util import parse_url import voluptuous as vol from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry @@ -96,6 +96,11 @@ SET_TIME_SERVICE_SCHEMA = vol.Schema( ) +def hostname_from_url(url: str) -> str: + """Return the hostname from a url.""" + return parse_url(url)[1] + + def _host_validator(config: dict[str, str]) -> dict[str, str]: """Validate that a host is properly configured.""" if config[CONF_HOST].startswith("elks://"): @@ -231,7 +236,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up Elk-M1 Control from a config entry.""" conf: MappingProxyType[str, Any] = entry.data - host = urlparse(entry.data[CONF_HOST]).hostname + host = hostname_from_url(entry.data[CONF_HOST]) _LOGGER.debug("Setting up elkm1 %s", conf["host"]) diff --git a/homeassistant/components/elkm1/config_flow.py b/homeassistant/components/elkm1/config_flow.py index 8675ff45ee7..ac7fc903330 100644 --- a/homeassistant/components/elkm1/config_flow.py +++ b/homeassistant/components/elkm1/config_flow.py @@ -4,7 +4,6 @@ from __future__ import annotations import asyncio import logging from typing import Any -from urllib.parse import urlparse from elkm1_lib.discovery import ElkSystem from elkm1_lib.elk import Elk @@ -26,7 +25,7 @@ from homeassistant.helpers.typing import DiscoveryInfoType from homeassistant.util import slugify from homeassistant.util.network import is_ip_address -from . import async_wait_for_elk_to_sync +from . import async_wait_for_elk_to_sync, hostname_from_url from .const import CONF_AUTO_CONFIGURE, DISCOVER_SCAN_TIMEOUT, DOMAIN, LOGIN_TIMEOUT from .discovery import ( _short_mac, @@ -170,7 +169,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): for entry in self._async_current_entries(include_ignore=False): if ( entry.unique_id == mac - or urlparse(entry.data[CONF_HOST]).hostname == host + or hostname_from_url(entry.data[CONF_HOST]) == host ): if async_update_entry_from_discovery(self.hass, entry, device): self.hass.async_create_task( @@ -214,7 +213,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): current_unique_ids = self._async_current_ids() current_hosts = { - urlparse(entry.data[CONF_HOST]).hostname + hostname_from_url(entry.data[CONF_HOST]) for entry in self._async_current_entries(include_ignore=False) } discovered_devices = await async_discover_devices( @@ -344,7 +343,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): if self._url_already_configured(url): return self.async_abort(reason="address_already_configured") - host = urlparse(url).hostname + host = hostname_from_url(url) _LOGGER.debug( "Importing is trying to fill unique id from discovery for %s", host ) @@ -367,10 +366,10 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): def _url_already_configured(self, url: str) -> bool: """See if we already have a elkm1 matching user input configured.""" existing_hosts = { - urlparse(entry.data[CONF_HOST]).hostname + hostname_from_url(entry.data[CONF_HOST]) for entry in self._async_current_entries() } - return urlparse(url).hostname in existing_hosts + return hostname_from_url(url) in existing_hosts class InvalidAuth(exceptions.HomeAssistantError): diff --git a/tests/components/elkm1/test_config_flow.py b/tests/components/elkm1/test_config_flow.py index 7ce0e2163ac..a58528b700a 100644 --- a/tests/components/elkm1/test_config_flow.py +++ b/tests/components/elkm1/test_config_flow.py @@ -1454,3 +1454,149 @@ async def test_multiple_instances_with_discovery(hass): "password": "", } assert len(mock_setup_entry.mock_calls) == 1 + + +async def test_multiple_instances_with_tls_v12(hass): + """Test we can setup a secure elk with tls v1_2.""" + + elk_discovery_1 = ElkSystem("aa:bb:cc:dd:ee:ff", "127.0.0.1", 2601) + elk_discovery_2 = ElkSystem("aa:bb:cc:dd:ee:fe", "127.0.0.2", 2601) + + with _patch_discovery(device=elk_discovery_1): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + await hass.async_block_till_done() + + assert result["type"] == "form" + assert not result["errors"] + assert result["step_id"] == "user" + + mocked_elk = mock_elk(invalid_auth=False, sync_complete=True) + + with _patch_elk(elk=mocked_elk): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"device": elk_discovery_1.mac_address}, + ) + await hass.async_block_till_done() + + assert result2["type"] == "form" + assert not result["errors"] + assert result2["step_id"] == "discovered_connection" + with _patch_discovery(device=elk_discovery_1), _patch_elk(elk=mocked_elk), patch( + "homeassistant.components.elkm1.async_setup", return_value=True + ) as mock_setup, patch( + "homeassistant.components.elkm1.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + { + "protocol": "TLS 1.2", + "username": "test-username", + "password": "test-password", + }, + ) + await hass.async_block_till_done() + + assert result3["type"] == "create_entry" + assert result3["title"] == "ElkM1 ddeeff" + assert result3["data"] == { + "auto_configure": True, + "host": "elksv1_2://127.0.0.1", + "password": "test-password", + "prefix": "", + "username": "test-username", + } + assert len(mock_setup.mock_calls) == 1 + assert len(mock_setup_entry.mock_calls) == 1 + + # Now try to add another instance with the different discovery info + with _patch_discovery(device=elk_discovery_2): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + await hass.async_block_till_done() + + assert result["type"] == "form" + assert not result["errors"] + assert result["step_id"] == "user" + + mocked_elk = mock_elk(invalid_auth=False, sync_complete=True) + + with _patch_elk(elk=mocked_elk): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"device": elk_discovery_2.mac_address}, + ) + await hass.async_block_till_done() + + with _patch_discovery(device=elk_discovery_2), _patch_elk(elk=mocked_elk), patch( + "homeassistant.components.elkm1.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + { + "protocol": "TLS 1.2", + "username": "test-username", + "password": "test-password", + }, + ) + await hass.async_block_till_done() + + assert result3["type"] == "create_entry" + assert result3["title"] == "ElkM1 ddeefe" + assert result3["data"] == { + "auto_configure": True, + "host": "elksv1_2://127.0.0.2", + "password": "test-password", + "prefix": "ddeefe", + "username": "test-username", + } + assert len(mock_setup_entry.mock_calls) == 1 + + # Finally, try to add another instance manually with no discovery info + + with _patch_discovery(no_device=True): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + await hass.async_block_till_done() + + assert result["type"] == "form" + assert result["errors"] == {} + assert result["step_id"] == "manual_connection" + + mocked_elk = mock_elk(invalid_auth=False, sync_complete=True) + + with _patch_discovery(no_device=True), _patch_elk(elk=mocked_elk), patch( + "homeassistant.components.elkm1.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "protocol": "TLS 1.2", + "address": "1.2.3.4", + "prefix": "guest_house", + "password": "test-password", + "username": "test-username", + }, + ) + await hass.async_block_till_done() + + import pprint + + pprint.pprint(result2) + assert result2["type"] == "create_entry" + assert result2["title"] == "guest_house" + assert result2["data"] == { + "auto_configure": True, + "host": "elksv1_2://1.2.3.4", + "prefix": "guest_house", + "password": "test-password", + "username": "test-username", + } + assert len(mock_setup_entry.mock_calls) == 1