diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py index 5a5daa5008a..f58df9dc11e 100644 --- a/homeassistant/components/knx/__init__.py +++ b/homeassistant/components/knx/__init__.py @@ -2,7 +2,9 @@ from __future__ import annotations import asyncio +import contextlib import logging +from pathlib import Path from typing import Final import voluptuous as vol @@ -335,6 +337,21 @@ async def async_update_entry(hass: HomeAssistant, entry: ConfigEntry) -> None: await hass.config_entries.async_reload(entry.entry_id) +async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None: + """Remove a config entry.""" + + def remove_keyring_files(file_path: Path) -> None: + """Remove keyring files.""" + with contextlib.suppress(FileNotFoundError): + file_path.unlink() + with contextlib.suppress(FileNotFoundError, OSError): + file_path.parent.rmdir() + + if (_knxkeys_file := entry.data.get(CONF_KNX_KNXKEY_FILENAME)) is not None: + file_path = Path(hass.config.path(STORAGE_DIR)) / _knxkeys_file + await hass.async_add_executor_job(remove_keyring_files, file_path) + + class KNXModule: """Representation of KNX Object.""" @@ -384,6 +401,14 @@ class KNXModule: def connection_config(self) -> ConnectionConfig: """Return the connection_config.""" _conn_type: str = self.entry.data[CONF_KNX_CONNECTION_TYPE] + _knxkeys_file: str | None = ( + self.hass.config.path( + STORAGE_DIR, + self.entry.data[CONF_KNX_KNXKEY_FILENAME], + ) + if self.entry.data.get(CONF_KNX_KNXKEY_FILENAME) is not None + else None + ) if _conn_type == CONF_KNX_ROUTING: return ConnectionConfig( connection_type=ConnectionType.ROUTING, @@ -392,6 +417,10 @@ class KNXModule: multicast_port=self.entry.data[CONF_KNX_MCAST_PORT], local_ip=self.entry.data.get(CONF_KNX_LOCAL_IP), auto_reconnect=True, + secure_config=SecureConfig( + knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD), + knxkeys_file_path=_knxkeys_file, + ), threaded=True, ) if _conn_type == CONF_KNX_TUNNELING: @@ -402,6 +431,10 @@ class KNXModule: local_ip=self.entry.data.get(CONF_KNX_LOCAL_IP), route_back=self.entry.data.get(CONF_KNX_ROUTE_BACK, False), auto_reconnect=True, + secure_config=SecureConfig( + knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD), + knxkeys_file_path=_knxkeys_file, + ), threaded=True, ) if _conn_type == CONF_KNX_TUNNELING_TCP: @@ -410,16 +443,12 @@ class KNXModule: gateway_ip=self.entry.data[CONF_HOST], gateway_port=self.entry.data[CONF_PORT], auto_reconnect=True, + secure_config=SecureConfig( + knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD), + knxkeys_file_path=_knxkeys_file, + ), threaded=True, ) - knxkeys_file: str | None = ( - self.hass.config.path( - STORAGE_DIR, - self.entry.data[CONF_KNX_KNXKEY_FILENAME], - ) - if self.entry.data.get(CONF_KNX_KNXKEY_FILENAME) is not None - else None - ) if _conn_type == CONF_KNX_TUNNELING_TCP_SECURE: return ConnectionConfig( connection_type=ConnectionType.TUNNELING_TCP_SECURE, @@ -432,7 +461,7 @@ class KNXModule: CONF_KNX_SECURE_DEVICE_AUTHENTICATION ), knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD), - knxkeys_file_path=knxkeys_file, + knxkeys_file_path=_knxkeys_file, ), auto_reconnect=True, threaded=True, @@ -450,13 +479,17 @@ class KNXModule: CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE ), knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD), - knxkeys_file_path=knxkeys_file, + knxkeys_file_path=_knxkeys_file, ), auto_reconnect=True, threaded=True, ) return ConnectionConfig( auto_reconnect=True, + secure_config=SecureConfig( + knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD), + knxkeys_file_path=_knxkeys_file, + ), threaded=True, ) diff --git a/homeassistant/components/knx/config_flow.py b/homeassistant/components/knx/config_flow.py index 7465c394dd1..ec79dbc5f9a 100644 --- a/homeassistant/components/knx/config_flow.py +++ b/homeassistant/components/knx/config_flow.py @@ -3,6 +3,7 @@ from __future__ import annotations from abc import ABC, abstractmethod from collections.abc import AsyncGenerator +from pathlib import Path from typing import Any, Final import voluptuous as vol @@ -11,8 +12,9 @@ from xknx.exceptions.exception import CommunicationError, InvalidSecureConfigura from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT from xknx.io.gateway_scanner import GatewayDescriptor, GatewayScanner from xknx.io.self_description import request_description -from xknx.secure.keyring import XMLInterface, load_keyring +from xknx.secure.keyring import Keyring, XMLInterface, sync_load_keyring +from homeassistant.components.file_upload import process_uploaded_file from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow from homeassistant.const import CONF_HOST, CONF_PORT from homeassistant.core import callback @@ -27,7 +29,6 @@ from .const import ( CONF_KNX_DEFAULT_RATE_LIMIT, CONF_KNX_DEFAULT_STATE_UPDATER, CONF_KNX_INDIVIDUAL_ADDRESS, - CONF_KNX_KNXKEY_FILENAME, CONF_KNX_KNXKEY_PASSWORD, CONF_KNX_LOCAL_IP, CONF_KNX_MCAST_GRP, @@ -42,10 +43,10 @@ from .const import ( CONF_KNX_SECURE_USER_ID, CONF_KNX_SECURE_USER_PASSWORD, CONF_KNX_STATE_UPDATER, + CONF_KNX_TUNNEL_ENDPOINT_IA, CONF_KNX_TUNNELING, CONF_KNX_TUNNELING_TCP, CONF_KNX_TUNNELING_TCP_SECURE, - CONST_KNX_STORAGE_KEY, DEFAULT_ROUTING_IA, DOMAIN, KNXConfigEntryData, @@ -65,6 +66,9 @@ DEFAULT_ENTRY_DATA = KNXConfigEntryData( state_updater=CONF_KNX_DEFAULT_STATE_UPDATER, ) +CONF_KEYRING_FILE: Final = "knxkeys_file" +DEFAULT_KNX_KEYRING_FILENAME: Final = "keyring.knxkeys" + CONF_KNX_TUNNELING_TYPE: Final = "tunneling_type" CONF_KNX_TUNNELING_TYPE_LABELS: Final = { CONF_KNX_TUNNELING: "UDP (Tunnelling v1)", @@ -93,6 +97,9 @@ class KNXCommonFlow(ABC, FlowHandler): """Initialize KNXCommonFlow.""" self.initial_data = initial_data self.new_entry_data = KNXConfigEntryData() + self.new_title: str | None = None + + self._keyring: Keyring | None = None self._found_gateways: list[GatewayDescriptor] = [] self._found_tunnels: list[GatewayDescriptor] = [] self._selected_tunnel: GatewayDescriptor | None = None @@ -102,9 +109,25 @@ class KNXCommonFlow(ABC, FlowHandler): self._async_scan_gen: AsyncGenerator[GatewayDescriptor, None] | None = None @abstractmethod - def finish_flow(self, title: str) -> FlowResult: + def finish_flow(self) -> FlowResult: """Finish the flow.""" + @property + def connection_type(self) -> str: + """Return the configured connection type.""" + _new_type = self.new_entry_data.get(CONF_KNX_CONNECTION_TYPE) + if _new_type is None: + return self.initial_data[CONF_KNX_CONNECTION_TYPE] + return _new_type + + @property + def tunnel_endpoint_ia(self) -> str | None: + """Return the configured tunnel endpoint individual address.""" + return self.new_entry_data.get( + CONF_KNX_TUNNEL_ENDPOINT_IA, + self.initial_data.get(CONF_KNX_TUNNEL_ENDPOINT_IA), + ) + async def async_step_connection_type( self, user_input: dict | None = None ) -> FlowResult: @@ -135,8 +158,12 @@ class KNXCommonFlow(ABC, FlowHandler): return await self.async_step_tunnel() # Automatic connection type - self.new_entry_data = KNXConfigEntryData(connection_type=CONF_KNX_AUTOMATIC) - return self.finish_flow(title=CONF_KNX_AUTOMATIC.capitalize()) + self.new_entry_data = KNXConfigEntryData( + connection_type=CONF_KNX_AUTOMATIC, + tunnel_endpoint_ia=None, + ) + self.new_title = CONF_KNX_AUTOMATIC.capitalize() + return self.finish_flow() supported_connection_types = { CONF_KNX_TUNNELING: CONF_KNX_TUNNELING.capitalize(), @@ -194,13 +221,18 @@ class KNXCommonFlow(ABC, FlowHandler): port=self._selected_tunnel.port, route_back=False, connection_type=connection_type, + device_authentication=None, + user_id=None, + user_password=None, + tunnel_endpoint_ia=None, ) if connection_type == CONF_KNX_TUNNELING_TCP_SECURE: return self.async_show_menu( step_id="secure_key_source", menu_options=["secure_knxkeys", "secure_tunnel_manual"], ) - return self.finish_flow(title=f"Tunneling @ {self._selected_tunnel}") + self.new_title = f"Tunneling @ {self._selected_tunnel}" + return self.finish_flow() if not self._found_tunnels: return await self.async_step_manual_tunnel() @@ -264,6 +296,10 @@ class KNXCommonFlow(ABC, FlowHandler): port=user_input[CONF_PORT], route_back=user_input[CONF_KNX_ROUTE_BACK], local_ip=_local_ip, + device_authentication=None, + user_id=None, + user_password=None, + tunnel_endpoint_ia=None, ) if selected_tunnelling_type == CONF_KNX_TUNNELING_TCP_SECURE: @@ -271,7 +307,12 @@ class KNXCommonFlow(ABC, FlowHandler): step_id="secure_key_source", menu_options=["secure_knxkeys", "secure_tunnel_manual"], ) - return self.finish_flow(title=f"Tunneling @ {_host}") + self.new_title = ( + "Tunneling " + f"{'UDP' if selected_tunnelling_type == CONF_KNX_TUNNELING else 'TCP'} " + f"@ {_host}" + ) + return self.finish_flow() _reconfiguring_existing_tunnel = ( self.initial_data.get(CONF_KNX_CONNECTION_TYPE) @@ -342,10 +383,10 @@ class KNXCommonFlow(ABC, FlowHandler): device_authentication=user_input[CONF_KNX_SECURE_DEVICE_AUTHENTICATION], user_id=user_input[CONF_KNX_SECURE_USER_ID], user_password=user_input[CONF_KNX_SECURE_USER_PASSWORD], + tunnel_endpoint_ia=None, ) - return self.finish_flow( - title=f"Secure Tunneling @ {self.new_entry_data[CONF_HOST]}" - ) + self.new_title = f"Secure Tunneling @ {self.new_entry_data[CONF_HOST]}" + return self.finish_flow() fields = { vol.Required( @@ -399,12 +440,8 @@ class KNXCommonFlow(ABC, FlowHandler): CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE ], ) - return self.finish_flow( - title=( - "Secure Routing as" - f" {self.new_entry_data[CONF_KNX_INDIVIDUAL_ADDRESS]}" - ) - ) + self.new_title = f"Secure Routing as {self.new_entry_data[CONF_KNX_INDIVIDUAL_ADDRESS]}" + return self.finish_flow() fields = { vol.Required( @@ -437,92 +474,101 @@ class KNXCommonFlow(ABC, FlowHandler): ) async def async_step_secure_knxkeys( - self, user_input: dict | None = None + self, user_input: dict[str, Any] | None = None ) -> FlowResult: - """Configure secure knxkeys used to authenticate.""" - errors = {} - description_placeholders = {} + """Manage upload of new KNX Keyring file.""" + errors: dict[str, str] = {} if user_input is not None: - connection_type = self.new_entry_data[CONF_KNX_CONNECTION_TYPE] - storage_key = CONST_KNX_STORAGE_KEY + user_input[CONF_KNX_KNXKEY_FILENAME] - try: - keyring = await load_keyring( - path=self.hass.config.path(STORAGE_DIR, storage_key), - password=user_input[CONF_KNX_KNXKEY_PASSWORD], - ) - except FileNotFoundError: - errors[CONF_KNX_KNXKEY_FILENAME] = "keyfile_not_found" - except InvalidSecureConfiguration: - errors[CONF_KNX_KNXKEY_PASSWORD] = "keyfile_invalid_signature" - else: - if ( - connection_type == CONF_KNX_TUNNELING_TCP_SECURE - and self._selected_tunnel is not None - ): - if host_ia := self._selected_tunnel.individual_address: - self._tunnel_endpoints = keyring.get_tunnel_interfaces_by_host( - host=host_ia - ) - if not self._tunnel_endpoints: - errors["base"] = "keyfile_no_tunnel_for_host" - description_placeholders = {CONF_HOST: str(host_ia)} - - if connection_type == CONF_KNX_ROUTING_SECURE: - if not (keyring.backbone is not None and keyring.backbone.key): - errors["base"] = "keyfile_no_backbone_key" - - if not errors: + password = user_input[CONF_KNX_KNXKEY_PASSWORD] + errors = await self._save_uploaded_knxkeys_file( + uploaded_file_id=user_input[CONF_KEYRING_FILE], + password=password, + ) + if not errors and self._keyring: self.new_entry_data |= KNXConfigEntryData( - knxkeys_filename=storage_key, - knxkeys_password=user_input[CONF_KNX_KNXKEY_PASSWORD], + knxkeys_filename=f"{DOMAIN}/{DEFAULT_KNX_KEYRING_FILENAME}", + knxkeys_password=password, backbone_key=None, sync_latency_tolerance=None, - device_authentication=None, - user_id=None, - user_password=None, ) - if connection_type == CONF_KNX_ROUTING_SECURE: - return self.finish_flow( - title=( - "Secure Routing as" - f" {self.new_entry_data[CONF_KNX_INDIVIDUAL_ADDRESS]}" - ) - ) - return await self.async_step_knxkeys_tunnel_select() + # Routing + if self.connection_type in (CONF_KNX_ROUTING, CONF_KNX_ROUTING_SECURE): + return self.finish_flow() + + # Tunneling / Automatic + # skip selection step if we have a keyfile update that includes a configured tunnel + if self.tunnel_endpoint_ia is not None and self.tunnel_endpoint_ia in [ + str(_if.individual_address) for _if in self._keyring.interfaces + ]: + return self.finish_flow() + if not errors: + return await self.async_step_knxkeys_tunnel_select() - if _default_filename := self.initial_data.get(CONF_KNX_KNXKEY_FILENAME): - _default_filename = _default_filename.lstrip(CONST_KNX_STORAGE_KEY) fields = { - vol.Required( - CONF_KNX_KNXKEY_FILENAME, default=_default_filename - ): selector.TextSelector(), + vol.Required(CONF_KEYRING_FILE): selector.FileSelector( + config=selector.FileSelectorConfig(accept=".knxkeys") + ), vol.Required( CONF_KNX_KNXKEY_PASSWORD, default=self.initial_data.get(CONF_KNX_KNXKEY_PASSWORD), ): selector.TextSelector(), } - return self.async_show_form( step_id="secure_knxkeys", data_schema=vol.Schema(fields), errors=errors, - description_placeholders=description_placeholders, ) async def async_step_knxkeys_tunnel_select( self, user_input: dict | None = None ) -> FlowResult: """Select if a specific tunnel should be used from knxkeys file.""" + errors = {} + description_placeholders = {} if user_input is not None: - if user_input[CONF_KNX_SECURE_USER_ID] == CONF_KNX_AUTOMATIC: - selected_user_id = None + selected_tunnel_ia: str | None = None + _if_user_id: int | None = None + if user_input[CONF_KNX_TUNNEL_ENDPOINT_IA] == CONF_KNX_AUTOMATIC: + self.new_entry_data |= KNXConfigEntryData( + tunnel_endpoint_ia=None, + ) else: - selected_user_id = int(user_input[CONF_KNX_SECURE_USER_ID]) - self.new_entry_data |= KNXConfigEntryData(user_id=selected_user_id) - return self.finish_flow( - title=f"Secure Tunneling @ {self.new_entry_data[CONF_HOST]}" + selected_tunnel_ia = user_input[CONF_KNX_TUNNEL_ENDPOINT_IA] + self.new_entry_data |= KNXConfigEntryData( + tunnel_endpoint_ia=selected_tunnel_ia, + user_id=None, + user_password=None, + device_authentication=None, + ) + _if_user_id = next( + ( + _if.user_id + for _if in self._tunnel_endpoints + if str(_if.individual_address) == selected_tunnel_ia + ), + None, + ) + self.new_title = ( + f"{'Secure ' if _if_user_id else ''}" + f"Tunneling @ {selected_tunnel_ia or self.new_entry_data[CONF_HOST]}" ) + return self.finish_flow() + + # this step is only called from async_step_secure_knxkeys so self._keyring is always set + assert self._keyring + + # Filter for selected tunnel + if self._selected_tunnel is not None: + if host_ia := self._selected_tunnel.individual_address: + self._tunnel_endpoints = self._keyring.get_tunnel_interfaces_by_host( + host=host_ia + ) + if not self._tunnel_endpoints: + errors["base"] = "keyfile_no_tunnel_for_host" + description_placeholders = {CONF_HOST: str(host_ia)} + else: + self._tunnel_endpoints = self._keyring.interfaces tunnel_endpoint_options = [ selector.SelectOptionDict( @@ -532,8 +578,12 @@ class KNXCommonFlow(ABC, FlowHandler): for endpoint in self._tunnel_endpoints: tunnel_endpoint_options.append( selector.SelectOptionDict( - value=str(endpoint.user_id), - label=f"{endpoint.individual_address} (User ID: {endpoint.user_id})", + value=str(endpoint.individual_address), + label=( + f"{endpoint.individual_address} " + f"{'🔐 ' if endpoint.user_id else ''}" + f"(Data Secure GAs: {len(endpoint.group_addresses)})" + ), ) ) return self.async_show_form( @@ -541,7 +591,7 @@ class KNXCommonFlow(ABC, FlowHandler): data_schema=vol.Schema( { vol.Required( - CONF_KNX_SECURE_USER_ID, default=CONF_KNX_AUTOMATIC + CONF_KNX_TUNNEL_ENDPOINT_IA, default=CONF_KNX_AUTOMATIC ): selector.SelectSelector( selector.SelectSelectorConfig( options=tunnel_endpoint_options, @@ -550,6 +600,8 @@ class KNXCommonFlow(ABC, FlowHandler): ), } ), + errors=errors, + description_placeholders=description_placeholders, ) async def async_step_routing(self, user_input: dict | None = None) -> FlowResult: @@ -598,13 +650,19 @@ class KNXCommonFlow(ABC, FlowHandler): multicast_group=_multicast_group, multicast_port=_multicast_port, local_ip=_local_ip, + device_authentication=None, + user_id=None, + user_password=None, + tunnel_endpoint_ia=None, ) if connection_type == CONF_KNX_ROUTING_SECURE: + self.new_title = f"Secure Routing as {_individual_address}" return self.async_show_menu( step_id="secure_key_source", menu_options=["secure_knxkeys", "secure_routing_manual"], ) - return self.finish_flow(title=f"Routing as {_individual_address}") + self.new_title = f"Routing as {_individual_address}" + return self.finish_flow() routers = [router for router in self._found_gateways if router.supports_routing] if not routers: @@ -631,6 +689,32 @@ class KNXCommonFlow(ABC, FlowHandler): step_id="routing", data_schema=vol.Schema(fields), errors=errors ) + async def _save_uploaded_knxkeys_file( + self, uploaded_file_id: str, password: str + ) -> dict[str, str]: + """Validate the uploaded file and move it to the storage directory. Return errors.""" + + def _process_upload() -> tuple[Keyring | None, dict[str, str]]: + keyring: Keyring | None = None + errors = {} + with process_uploaded_file(self.hass, uploaded_file_id) as file_path: + try: + keyring = sync_load_keyring( + path=file_path, + password=password, + ) + except InvalidSecureConfiguration: + errors[CONF_KNX_KNXKEY_PASSWORD] = "keyfile_invalid_signature" + else: + dest_path = Path(self.hass.config.path(STORAGE_DIR, DOMAIN)) + dest_path.mkdir(exist_ok=True) + file_path.rename(dest_path / DEFAULT_KNX_KEYRING_FILENAME) + return keyring, errors + + keyring, errors = await self.hass.async_add_executor_job(_process_upload) + self._keyring = keyring + return errors + class KNXConfigFlow(KNXCommonFlow, ConfigFlow, domain=DOMAIN): """Handle a KNX config flow.""" @@ -648,8 +732,9 @@ class KNXConfigFlow(KNXCommonFlow, ConfigFlow, domain=DOMAIN): return KNXOptionsFlow(config_entry) @callback - def finish_flow(self, title: str) -> FlowResult: + def finish_flow(self) -> FlowResult: """Create the ConfigEntry.""" + title = self.new_title or f"KNX {self.new_entry_data[CONF_KNX_CONNECTION_TYPE]}" return self.async_create_entry( title=title, data=DEFAULT_ENTRY_DATA | self.new_entry_data, @@ -673,13 +758,13 @@ class KNXOptionsFlow(KNXCommonFlow, OptionsFlow): super().__init__(initial_data=config_entry.data) # type: ignore[arg-type] @callback - def finish_flow(self, title: str | None) -> FlowResult: + def finish_flow(self) -> FlowResult: """Update the ConfigEntry and finish the flow.""" new_data = DEFAULT_ENTRY_DATA | self.initial_data | self.new_entry_data self.hass.config_entries.async_update_entry( self.config_entry, data=new_data, - title=title or UNDEFINED, + title=self.new_title or UNDEFINED, ) return self.async_create_entry(title="", data={}) @@ -689,7 +774,11 @@ class KNXOptionsFlow(KNXCommonFlow, OptionsFlow): """Manage KNX options.""" return self.async_show_menu( step_id="options_init", - menu_options=["connection_type", "communication_settings"], + menu_options=[ + "connection_type", + "communication_settings", + "secure_knxkeys", + ], ) async def async_step_communication_settings( @@ -701,7 +790,7 @@ class KNXOptionsFlow(KNXCommonFlow, OptionsFlow): state_updater=user_input[CONF_KNX_STATE_UPDATER], rate_limit=user_input[CONF_KNX_RATE_LIMIT], ) - return self.finish_flow(title=None) + return self.finish_flow() data_schema = { vol.Required( diff --git a/homeassistant/components/knx/const.py b/homeassistant/components/knx/const.py index 058223bfaa1..d006637abd1 100644 --- a/homeassistant/components/knx/const.py +++ b/homeassistant/components/knx/const.py @@ -39,6 +39,7 @@ CONF_KNX_TUNNELING_TCP_SECURE: Final = "tunneling_tcp_secure" CONF_KNX_LOCAL_IP: Final = "local_ip" CONF_KNX_MCAST_GRP: Final = "multicast_group" CONF_KNX_MCAST_PORT: Final = "multicast_port" +CONF_KNX_TUNNEL_ENDPOINT_IA: Final = "tunnel_endpoint_ia" CONF_KNX_RATE_LIMIT: Final = "rate_limit" CONF_KNX_ROUTE_BACK: Final = "route_back" @@ -89,6 +90,7 @@ class KNXConfigEntryData(TypedDict, total=False): rate_limit: int host: str port: int + tunnel_endpoint_ia: str | None user_id: int | None user_password: str | None diff --git a/homeassistant/components/knx/manifest.json b/homeassistant/components/knx/manifest.json index c8c5c37a480..9bf2731b3d9 100644 --- a/homeassistant/components/knx/manifest.json +++ b/homeassistant/components/knx/manifest.json @@ -3,6 +3,7 @@ "name": "KNX", "codeowners": ["@Julius2342", "@farmio", "@marvin-w"], "config_flow": true, + "dependencies": ["file_upload"], "documentation": "https://www.home-assistant.io/integrations/knx", "integration_type": "hub", "iot_class": "local_push", diff --git a/homeassistant/components/knx/strings.json b/homeassistant/components/knx/strings.json index d1ac3793c05..a781f9d73cc 100644 --- a/homeassistant/components/knx/strings.json +++ b/homeassistant/components/knx/strings.json @@ -42,14 +42,13 @@ } }, "secure_knxkeys": { - "title": "Keyfile", - "description": "Please enter the information for your `.knxkeys` file.", + "title": "Import KNX Keyring", + "description": "Please select a `.knxkeys` file to import.", "data": { - "knxkeys_filename": "The filename of your `.knxkeys` file (including extension)", + "knxkeys_file": "Keyring file", "knxkeys_password": "The password to decrypt the `.knxkeys` file" }, "data_description": { - "knxkeys_filename": "The file is expected to be found in your config directory in `.storage/knx/`.\nIn Home Assistant OS this would be `/config/.storage/knx/`\nExample: `my_project.knxkeys`", "knxkeys_password": "This was set when exporting the file from ETS." } }, @@ -126,7 +125,8 @@ "title": "KNX Settings", "menu_options": { "connection_type": "Configure KNX interface", - "communication_settings": "Communication settings" + "communication_settings": "Communication settings", + "secure_knxkeys": "Import a `.knxkeys` file" } }, "communication_settings": { @@ -184,11 +184,10 @@ "title": "[%key:component::knx::config::step::secure_knxkeys::title%]", "description": "[%key:component::knx::config::step::secure_knxkeys::description%]", "data": { - "knxkeys_filename": "[%key:component::knx::config::step::secure_knxkeys::data::knxkeys_filename%]", + "knxkeys_file": "[%key:component::knx::config::step::secure_knxkeys::data::knxkeys_file%]", "knxkeys_password": "[%key:component::knx::config::step::secure_knxkeys::data::knxkeys_password%]" }, "data_description": { - "knxkeys_filename": "[%key:component::knx::config::step::secure_knxkeys::data_description::knxkeys_filename%]", "knxkeys_password": "[%key:component::knx::config::step::secure_knxkeys::data_description::knxkeys_password%]" } }, diff --git a/tests/components/knx/test_config_flow.py b/tests/components/knx/test_config_flow.py index 4466db2b8c3..4ac6a366119 100644 --- a/tests/components/knx/test_config_flow.py +++ b/tests/components/knx/test_config_flow.py @@ -1,4 +1,5 @@ """Test the KNX config flow.""" +from contextlib import contextmanager from unittest.mock import Mock, patch import pytest @@ -10,6 +11,7 @@ from xknx.telegram import IndividualAddress from homeassistant import config_entries from homeassistant.components.knx.config_flow import ( + CONF_KEYRING_FILE, CONF_KNX_GATEWAY, CONF_KNX_TUNNELING_TYPE, DEFAULT_ENTRY_DATA, @@ -35,6 +37,7 @@ from homeassistant.components.knx.const import ( CONF_KNX_SECURE_USER_ID, CONF_KNX_SECURE_USER_PASSWORD, CONF_KNX_STATE_UPDATER, + CONF_KNX_TUNNEL_ENDPOINT_IA, CONF_KNX_TUNNELING, CONF_KNX_TUNNELING_TCP, CONF_KNX_TUNNELING_TCP_SECURE, @@ -47,6 +50,10 @@ from homeassistant.data_entry_flow import FlowResult, FlowResultType from tests.common import MockConfigEntry, get_fixture_path FIXTURE_KNXKEYS_PASSWORD = "test" +FIXTURE_KEYRING = sync_load_keyring( + get_fixture_path("fixture.knxkeys", DOMAIN), FIXTURE_KNXKEYS_PASSWORD +) +FIXTURE_UPLOAD_UUID = "0123456789abcdef0123456789abcdef" GATEWAY_INDIVIDUAL_ADDRESS = IndividualAddress("1.0.0") @@ -59,6 +66,29 @@ def fixture_knx_setup(): yield mock_async_setup_entry +@contextmanager +def patch_file_upload(return_value=FIXTURE_KEYRING, side_effect=None): + """Patch file upload. Yields the Keyring instance (return_value).""" + with patch( + "homeassistant.components.knx.config_flow.process_uploaded_file" + ) as file_upload_mock, patch( + "homeassistant.components.knx.config_flow.sync_load_keyring", + return_value=return_value, + side_effect=side_effect, + ), patch( + "pathlib.Path.mkdir" + ) as mkdir_mock: + file_path_mock = Mock() + file_upload_mock.return_value.__enter__.return_value = file_path_mock + yield return_value + if side_effect: + mkdir_mock.assert_not_called() + file_path_mock.rename.assert_not_called() + else: + mkdir_mock.assert_called_once() + file_path_mock.rename.assert_called_once() + + def _gateway_descriptor( ip: str, port: int, @@ -153,6 +183,10 @@ async def test_routing_setup( CONF_KNX_MCAST_PORT: 3675, CONF_KNX_LOCAL_IP: None, CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110", + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, } knx_setup.assert_called_once() @@ -224,6 +258,10 @@ async def test_routing_setup_advanced( CONF_KNX_MCAST_PORT: 3675, CONF_KNX_LOCAL_IP: "192.168.1.112", CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110", + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, } knx_setup.assert_called_once() @@ -310,6 +348,10 @@ async def test_routing_secure_manual_setup( CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44bbaacc44bbaacc44", CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.123", + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, } knx_setup.assert_called_once() @@ -358,29 +400,11 @@ async def test_routing_secure_keyfile( assert result4["step_id"] == "secure_knxkeys" assert not result4["errors"] - # test file without backbone key - with patch( - "homeassistant.components.knx.config_flow.load_keyring" - ) as mock_load_keyring: - mock_keyring = Mock() - mock_keyring.backbone.key = None - mock_load_keyring.return_value = mock_keyring - secure_knxkeys = await hass.config_entries.flow.async_configure( - result["flow_id"], - { - CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys", - CONF_KNX_KNXKEY_PASSWORD: "password", - }, - ) - assert secure_knxkeys["type"] == FlowResultType.FORM - assert secure_knxkeys["errors"] == {"base": "keyfile_no_backbone_key"} - - # test valid file - with patch("homeassistant.components.knx.config_flow.load_keyring"): + with patch_file_upload(): routing_secure_knxkeys = await hass.config_entries.flow.async_configure( result4["flow_id"], { - CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys", + CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "password", }, ) @@ -390,20 +414,21 @@ async def test_routing_secure_keyfile( assert routing_secure_knxkeys["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING_SECURE, - CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys", + CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", CONF_KNX_KNXKEY_PASSWORD: "password", CONF_KNX_ROUTING_BACKBONE_KEY: None, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.123", } knx_setup.assert_called_once() @pytest.mark.parametrize( - ("user_input", "config_entry_data"), + ("user_input", "title", "config_entry_data"), [ ( { @@ -412,6 +437,7 @@ async def test_routing_secure_keyfile( CONF_PORT: 3675, CONF_KNX_ROUTE_BACK: False, }, + "Tunneling UDP @ 192.168.0.1", { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, @@ -420,6 +446,10 @@ async def test_routing_secure_keyfile( CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, }, ), ( @@ -429,6 +459,7 @@ async def test_routing_secure_keyfile( CONF_PORT: 3675, CONF_KNX_ROUTE_BACK: False, }, + "Tunneling TCP @ 192.168.0.1", { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, @@ -437,6 +468,10 @@ async def test_routing_secure_keyfile( CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, }, ), ( @@ -446,6 +481,7 @@ async def test_routing_secure_keyfile( CONF_PORT: 3675, CONF_KNX_ROUTE_BACK: True, }, + "Tunneling UDP @ 192.168.0.1", { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, @@ -454,6 +490,10 @@ async def test_routing_secure_keyfile( CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: True, CONF_KNX_LOCAL_IP: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, }, ), ], @@ -467,6 +507,7 @@ async def test_tunneling_setup_manual( hass: HomeAssistant, knx_setup, user_input, + title, config_entry_data, ) -> None: """Test tunneling if no gateway was found found (or `manual` option was chosen).""" @@ -502,7 +543,7 @@ async def test_tunneling_setup_manual( ) await hass.async_block_till_done() assert result3["type"] == FlowResultType.CREATE_ENTRY - assert result3["title"] == "Tunneling @ 192.168.0.1" + assert result3["title"] == title assert result3["data"] == config_entry_data knx_setup.assert_called_once() @@ -631,12 +672,16 @@ async def test_tunneling_setup_manual_request_description_error( ) await hass.async_block_till_done() assert result["type"] == FlowResultType.CREATE_ENTRY - assert result["title"] == "Tunneling @ 192.168.0.1" + assert result["title"] == "Tunneling TCP @ 192.168.0.1" assert result["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, CONF_HOST: "192.168.0.1", CONF_PORT: 3671, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, } knx_setup.assert_called_once() @@ -718,7 +763,7 @@ async def test_tunneling_setup_for_local_ip( ) await hass.async_block_till_done() assert result3["type"] == FlowResultType.CREATE_ENTRY - assert result3["title"] == "Tunneling @ 192.168.0.2" + assert result3["title"] == "Tunneling UDP @ 192.168.0.2" assert result3["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, @@ -727,6 +772,10 @@ async def test_tunneling_setup_for_local_ip( CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: "192.168.1.112", + CONF_KNX_TUNNEL_ENDPOINT_IA: None, + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, } knx_setup.assert_called_once() @@ -771,6 +820,10 @@ async def test_tunneling_setup_for_multiple_found_gateways( CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, } knx_setup.assert_called_once() @@ -848,11 +901,12 @@ async def test_form_with_automatic_connection_handling( assert result2["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, } knx_setup.assert_called_once() -async def _get_menu_step(hass: HomeAssistant) -> FlowResult: +async def _get_menu_step_secure_tunnel(hass: HomeAssistant) -> FlowResult: """Return flow in secure_tunnelling menu step.""" gateway = _gateway_descriptor( "192.168.0.1", @@ -950,7 +1004,7 @@ async def test_get_secure_menu_step_manual_tunnelling( async def test_configure_secure_tunnel_manual(hass: HomeAssistant, knx_setup) -> None: """Test configure tunnelling secure keys manually.""" - menu_step = await _get_menu_step(hass) + menu_step = await _get_menu_step_secure_tunnel(hass) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], @@ -981,13 +1035,14 @@ async def test_configure_secure_tunnel_manual(hass: HomeAssistant, knx_setup) -> CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, } knx_setup.assert_called_once() async def test_configure_secure_knxkeys(hass: HomeAssistant, knx_setup) -> None: """Test configure secure knxkeys.""" - menu_step = await _get_menu_step(hass) + menu_step = await _get_menu_step_secure_tunnel(hass) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], @@ -997,17 +1052,11 @@ async def test_configure_secure_knxkeys(hass: HomeAssistant, knx_setup) -> None: assert result["step_id"] == "secure_knxkeys" assert not result["errors"] - with patch( - "xknx.secure.keyring.sync_load_keyring", - return_value=sync_load_keyring( - str(get_fixture_path("fixture.knxkeys", DOMAIN).absolute()), - FIXTURE_KNXKEYS_PASSWORD, - ), - ): + with patch_file_upload(): secure_knxkeys = await hass.config_entries.flow.async_configure( result["flow_id"], { - CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys", + CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "test", }, ) @@ -1016,7 +1065,7 @@ async def test_configure_secure_knxkeys(hass: HomeAssistant, knx_setup) -> None: assert not result["errors"] secure_knxkeys = await hass.config_entries.flow.async_configure( secure_knxkeys["flow_id"], - {CONF_KNX_SECURE_USER_ID: CONF_KNX_AUTOMATIC}, + {CONF_KNX_TUNNEL_ENDPOINT_IA: CONF_KNX_AUTOMATIC}, ) await hass.async_block_till_done() @@ -1024,13 +1073,14 @@ async def test_configure_secure_knxkeys(hass: HomeAssistant, knx_setup) -> None: assert secure_knxkeys["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, - CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys", + CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", CONF_KNX_KNXKEY_PASSWORD: "test", CONF_KNX_ROUTING_BACKBONE_KEY: None, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", @@ -1040,37 +1090,9 @@ async def test_configure_secure_knxkeys(hass: HomeAssistant, knx_setup) -> None: knx_setup.assert_called_once() -async def test_configure_secure_knxkeys_file_not_found(hass: HomeAssistant) -> None: - """Test configure secure knxkeys but file was not found.""" - menu_step = await _get_menu_step(hass) - - result = await hass.config_entries.flow.async_configure( - menu_step["flow_id"], - {"next_step_id": "secure_knxkeys"}, - ) - assert result["type"] == FlowResultType.FORM - assert result["step_id"] == "secure_knxkeys" - assert not result["errors"] - - with patch( - "homeassistant.components.knx.config_flow.load_keyring", - side_effect=FileNotFoundError(), - ): - secure_knxkeys = await hass.config_entries.flow.async_configure( - result["flow_id"], - { - CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys", - CONF_KNX_KNXKEY_PASSWORD: "password", - }, - ) - assert secure_knxkeys["type"] == FlowResultType.FORM - assert secure_knxkeys["errors"] - assert secure_knxkeys["errors"][CONF_KNX_KNXKEY_FILENAME] == "keyfile_not_found" - - async def test_configure_secure_knxkeys_invalid_signature(hass: HomeAssistant) -> None: """Test configure secure knxkeys but file was not found.""" - menu_step = await _get_menu_step(hass) + menu_step = await _get_menu_step_secure_tunnel(hass) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], @@ -1080,14 +1102,13 @@ async def test_configure_secure_knxkeys_invalid_signature(hass: HomeAssistant) - assert result["step_id"] == "secure_knxkeys" assert not result["errors"] - with patch( - "homeassistant.components.knx.config_flow.load_keyring", + with patch_file_upload( side_effect=InvalidSecureConfiguration(), ): secure_knxkeys = await hass.config_entries.flow.async_configure( result["flow_id"], { - CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys", + CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "password", }, ) @@ -1101,7 +1122,7 @@ async def test_configure_secure_knxkeys_invalid_signature(hass: HomeAssistant) - async def test_configure_secure_knxkeys_no_tunnel_for_host(hass: HomeAssistant) -> None: """Test configure secure knxkeys but file was not found.""" - menu_step = await _get_menu_step(hass) + menu_step = await _get_menu_step_secure_tunnel(hass) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], @@ -1111,16 +1132,12 @@ async def test_configure_secure_knxkeys_no_tunnel_for_host(hass: HomeAssistant) assert result["step_id"] == "secure_knxkeys" assert not result["errors"] - with patch( - "homeassistant.components.knx.config_flow.load_keyring" - ) as mock_load_keyring: - mock_keyring = Mock() + with patch_file_upload(return_value=Mock()) as mock_keyring: mock_keyring.get_tunnel_interfaces_by_host.return_value = [] - mock_load_keyring.return_value = mock_keyring secure_knxkeys = await hass.config_entries.flow.async_configure( result["flow_id"], { - CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys", + CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "password", }, ) @@ -1180,6 +1197,10 @@ async def test_options_flow_connection_type( CONF_KNX_RATE_LIMIT: 0, CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER, CONF_KNX_ROUTE_BACK: False, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, } @@ -1251,17 +1272,11 @@ async def test_options_flow_secure_manual_to_keyfile( assert result4["step_id"] == "secure_knxkeys" assert not result4["errors"] - with patch( - "xknx.secure.keyring.sync_load_keyring", - return_value=sync_load_keyring( - str(get_fixture_path("fixture.knxkeys", DOMAIN).absolute()), - FIXTURE_KNXKEYS_PASSWORD, - ), - ): + with patch_file_upload(): secure_knxkeys = await hass.config_entries.options.async_configure( result4["flow_id"], { - CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys", + CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "test", }, ) @@ -1270,7 +1285,7 @@ async def test_options_flow_secure_manual_to_keyfile( assert not result["errors"] secure_knxkeys = await hass.config_entries.options.async_configure( secure_knxkeys["flow_id"], - {CONF_KNX_SECURE_USER_ID: "2"}, + {CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1"}, ) await hass.async_block_till_done() @@ -1278,11 +1293,12 @@ async def test_options_flow_secure_manual_to_keyfile( assert mock_config_entry.data == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, - CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys", + CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", CONF_KNX_KNXKEY_PASSWORD: "test", CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, - CONF_KNX_SECURE_USER_ID: 2, + CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1", CONF_KNX_ROUTING_BACKBONE_KEY: None, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, CONF_HOST: "192.168.0.1", @@ -1326,3 +1342,118 @@ async def test_options_communication_settings( CONF_KNX_RATE_LIMIT: 40, } knx_setup.assert_called_once() + + +async def test_options_update_keyfile(hass: HomeAssistant, knx_setup) -> None: + """Test options flow updating keyfile when tunnel endpoint is already configured.""" + start_data = { + **DEFAULT_ENTRY_DATA, + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, + CONF_KNX_SECURE_USER_ID: 2, + CONF_KNX_SECURE_USER_PASSWORD: "password", + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: "device_auth", + CONF_KNX_KNXKEY_PASSWORD: "old_password", + CONF_HOST: "192.168.0.1", + CONF_PORT: 3675, + CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", + CONF_KNX_ROUTE_BACK: False, + CONF_KNX_LOCAL_IP: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1", + } + mock_config_entry = MockConfigEntry( + title="KNX", + domain="knx", + data=start_data, + ) + mock_config_entry.add_to_hass(hass) + await hass.config_entries.async_setup(mock_config_entry.entry_id) + menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id) + + result = await hass.config_entries.options.async_configure( + menu_step["flow_id"], + {"next_step_id": "secure_knxkeys"}, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "secure_knxkeys" + + with patch_file_upload(): + result2 = await hass.config_entries.options.async_configure( + result["flow_id"], + { + CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, + CONF_KNX_KNXKEY_PASSWORD: "password", + }, + ) + await hass.async_block_till_done() + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert not result2.get("data") + assert mock_config_entry.data == { + **start_data, + CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", + CONF_KNX_KNXKEY_PASSWORD: "password", + CONF_KNX_ROUTING_BACKBONE_KEY: None, + CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, + } + knx_setup.assert_called_once() + + +async def test_options_keyfile_upload(hass: HomeAssistant, knx_setup) -> None: + """Test options flow uploading a keyfile for the first time.""" + start_data = { + **DEFAULT_ENTRY_DATA, + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3675, + CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", + CONF_KNX_ROUTE_BACK: False, + CONF_KNX_LOCAL_IP: None, + } + mock_config_entry = MockConfigEntry( + title="KNX", + domain="knx", + data=start_data, + ) + mock_config_entry.add_to_hass(hass) + await hass.config_entries.async_setup(mock_config_entry.entry_id) + menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id) + + result = await hass.config_entries.options.async_configure( + menu_step["flow_id"], + {"next_step_id": "secure_knxkeys"}, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "secure_knxkeys" + + with patch_file_upload(): + result2 = await hass.config_entries.options.async_configure( + result["flow_id"], + { + CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, + CONF_KNX_KNXKEY_PASSWORD: "password", + }, + ) + + assert result2["type"] == FlowResultType.FORM + assert result2["step_id"] == "knxkeys_tunnel_select" + + result3 = await hass.config_entries.options.async_configure( + result["flow_id"], + user_input={ + CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1", + }, + ) + await hass.async_block_till_done() + assert result3["type"] == FlowResultType.CREATE_ENTRY + assert not result3.get("data") + assert mock_config_entry.data == { + **start_data, + CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", + CONF_KNX_KNXKEY_PASSWORD: "password", + CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1", + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_ROUTING_BACKBONE_KEY: None, + CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, + } + knx_setup.assert_called_once() diff --git a/tests/components/knx/test_init.py b/tests/components/knx/test_init.py index ff763f94b67..785ff9d8317 100644 --- a/tests/components/knx/test_init.py +++ b/tests/components/knx/test_init.py @@ -1,4 +1,6 @@ """Test KNX init.""" +from unittest.mock import patch + import pytest from xknx.io import ( DEFAULT_MCAST_GRP, @@ -8,6 +10,7 @@ from xknx.io import ( SecureConfig, ) +from homeassistant import config_entries from homeassistant.components.knx.config_flow import DEFAULT_ROUTING_IA from homeassistant.components.knx.const import ( CONF_KNX_AUTOMATIC, @@ -110,12 +113,17 @@ from tests.common import MockConfigEntry CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_INDIVIDUAL_ADDRESS: DEFAULT_ROUTING_IA, + CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", + CONF_KNX_KNXKEY_PASSWORD: "password", }, ConnectionConfig( connection_type=ConnectionType.TUNNELING_TCP, gateway_ip="192.168.0.2", gateway_port=3675, auto_reconnect=True, + secure_config=SecureConfig( + knxkeys_file_path="keyring.knxkeys", knxkeys_password="password" + ), threaded=True, ), ), @@ -251,3 +259,30 @@ async def test_init_connection_handling( .connection_config() .secure_config.knxkeys_file_path ) + + +async def test_async_remove_entry( + hass: HomeAssistant, + knx: KNXTestKit, +) -> None: + """Test async_setup_entry (for coverage).""" + config_entry = MockConfigEntry( + title="KNX", + domain=KNX_DOMAIN, + data={ + CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys", + }, + ) + knx.mock_config_entry = config_entry + await knx.setup_integration({}) + + with patch("pathlib.Path.unlink") as unlink_mock, patch( + "pathlib.Path.rmdir" + ) as rmdir_mock: + assert await hass.config_entries.async_remove(config_entry.entry_id) + unlink_mock.assert_called_once() + rmdir_mock.assert_called_once() + await hass.async_block_till_done() + + assert hass.config_entries.async_entries() == [] + assert config_entry.state is config_entries.ConfigEntryState.NOT_LOADED