From cd4e8707eaf23e9ce13ba4ef0669da2f65cd16e6 Mon Sep 17 00:00:00 2001 From: Steven Looman Date: Tue, 12 Mar 2024 17:38:09 +0100 Subject: [PATCH] Improve discovering upnp/igd device by always using the SSDP-discovery for the Unique Device Name (#111487) * Always use the UDN found in the SSDP discovery, instead of the device description * Ensure existing DeviceEntries are still matched --- homeassistant/components/upnp/__init__.py | 5 +- homeassistant/components/upnp/config_flow.py | 15 +++--- tests/components/upnp/test_config_flow.py | 41 ++++++++++++++- tests/components/upnp/test_init.py | 54 +++++++++++++++++++- 4 files changed, 103 insertions(+), 12 deletions(-) diff --git a/homeassistant/components/upnp/__init__.py b/homeassistant/components/upnp/__init__.py index 893b86cd1b2..f2f3ffd0a1b 100644 --- a/homeassistant/components/upnp/__init__.py +++ b/homeassistant/components/upnp/__init__.py @@ -80,6 +80,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: # Create device. assert discovery_info is not None + assert discovery_info.ssdp_udn assert discovery_info.ssdp_all_locations location = get_preferred_location(discovery_info.ssdp_all_locations) try: @@ -118,7 +119,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: if device.serial_number: identifiers.add((IDENTIFIER_SERIAL_NUMBER, device.serial_number)) - connections = {(dr.CONNECTION_UPNP, device.udn)} + connections = {(dr.CONNECTION_UPNP, discovery_info.ssdp_udn)} + if discovery_info.ssdp_udn != device.udn: + connections.add((dr.CONNECTION_UPNP, device.udn)) if device_mac_address: connections.add((dr.CONNECTION_NETWORK_MAC, device_mac_address)) diff --git a/homeassistant/components/upnp/config_flow.py b/homeassistant/components/upnp/config_flow.py index d02877ccd05..a708403b6f2 100644 --- a/homeassistant/components/upnp/config_flow.py +++ b/homeassistant/components/upnp/config_flow.py @@ -42,7 +42,7 @@ def _friendly_name_from_discovery(discovery_info: ssdp.SsdpServiceInfo) -> str: def _is_complete_discovery(discovery_info: ssdp.SsdpServiceInfo) -> bool: """Test if discovery is complete and usable.""" return bool( - ssdp.ATTR_UPNP_UDN in discovery_info.upnp + discovery_info.ssdp_udn and discovery_info.ssdp_st and discovery_info.ssdp_all_locations and discovery_info.ssdp_usn @@ -80,9 +80,8 @@ class UpnpFlowHandler(ConfigFlow, domain=DOMAIN): VERSION = 1 # Paths: - # - ssdp(discovery_info) --> ssdp_confirm(None) - # --> ssdp_confirm({}) --> create_entry() - # - user(None): scan --> user({...}) --> create_entry() + # 1: ssdp(discovery_info) --> ssdp_confirm(None) --> ssdp_confirm({}) --> create_entry() + # 2: user(None): scan --> user({...}) --> create_entry() @property def _discoveries(self) -> dict[str, SsdpServiceInfo]: @@ -243,9 +242,9 @@ class UpnpFlowHandler(ConfigFlow, domain=DOMAIN): discovery = self._remove_discovery(usn) mac_address = await _async_mac_address_from_discovery(self.hass, discovery) data = { - CONFIG_ENTRY_UDN: discovery.upnp[ssdp.ATTR_UPNP_UDN], + CONFIG_ENTRY_UDN: discovery.ssdp_udn, CONFIG_ENTRY_ST: discovery.ssdp_st, - CONFIG_ENTRY_ORIGINAL_UDN: discovery.upnp[ssdp.ATTR_UPNP_UDN], + CONFIG_ENTRY_ORIGINAL_UDN: discovery.ssdp_udn, CONFIG_ENTRY_MAC_ADDRESS: mac_address, CONFIG_ENTRY_HOST: discovery.ssdp_headers["_host"], CONFIG_ENTRY_LOCATION: get_preferred_location(discovery.ssdp_all_locations), @@ -267,9 +266,9 @@ class UpnpFlowHandler(ConfigFlow, domain=DOMAIN): title = _friendly_name_from_discovery(discovery) mac_address = await _async_mac_address_from_discovery(self.hass, discovery) data = { - CONFIG_ENTRY_UDN: discovery.upnp[ssdp.ATTR_UPNP_UDN], + CONFIG_ENTRY_UDN: discovery.ssdp_udn, CONFIG_ENTRY_ST: discovery.ssdp_st, - CONFIG_ENTRY_ORIGINAL_UDN: discovery.upnp[ssdp.ATTR_UPNP_UDN], + CONFIG_ENTRY_ORIGINAL_UDN: discovery.ssdp_udn, CONFIG_ENTRY_LOCATION: get_preferred_location(discovery.ssdp_all_locations), CONFIG_ENTRY_MAC_ADDRESS: mac_address, CONFIG_ENTRY_HOST: discovery.ssdp_headers["_host"], diff --git a/tests/components/upnp/test_config_flow.py b/tests/components/upnp/test_config_flow.py index 7c542e33c9d..67b4e5b10e6 100644 --- a/tests/components/upnp/test_config_flow.py +++ b/tests/components/upnp/test_config_flow.py @@ -1,5 +1,6 @@ """Test UPnP/IGD config flow.""" +import copy from copy import deepcopy from unittest.mock import patch @@ -111,6 +112,7 @@ async def test_flow_ssdp_incomplete_discovery(hass: HomeAssistant) -> None: context={"source": config_entries.SOURCE_SSDP}, data=ssdp.SsdpServiceInfo( ssdp_usn=TEST_USN, + # ssdp_udn=TEST_UDN, # Not provided. ssdp_st=TEST_ST, ssdp_location=TEST_LOCATION, upnp={ @@ -132,12 +134,12 @@ async def test_flow_ssdp_non_igd_device(hass: HomeAssistant) -> None: context={"source": config_entries.SOURCE_SSDP}, data=ssdp.SsdpServiceInfo( ssdp_usn=TEST_USN, + ssdp_udn=TEST_UDN, ssdp_st=TEST_ST, ssdp_location=TEST_LOCATION, ssdp_all_locations=[TEST_LOCATION], upnp={ ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:WFADevice:1", # Non-IGD - ssdp.ATTR_UPNP_UDN: TEST_UDN, }, ), ) @@ -442,3 +444,40 @@ async def test_flow_user_no_discovery(hass: HomeAssistant) -> None: ) assert result["type"] == data_entry_flow.FlowResultType.ABORT assert result["reason"] == "no_devices_found" + + +@pytest.mark.usefixtures( + "ssdp_instant_discovery", + "mock_setup_entry", + "mock_get_source_ip", + "mock_mac_address_from_host", +) +async def test_flow_ssdp_with_mismatched_udn(hass: HomeAssistant) -> None: + """Test config flow: discovered + configured through ssdp, where the UDN differs in the SSDP-discovery vs device description.""" + # Discovered via step ssdp. + test_discovery = copy.deepcopy(TEST_DISCOVERY) + test_discovery.upnp[ssdp.ATTR_UPNP_UDN] = "uuid:another_udn" + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_SSDP}, + data=test_discovery, + ) + assert result["type"] == data_entry_flow.FlowResultType.FORM + assert result["step_id"] == "ssdp_confirm" + + # Confirm via step ssdp_confirm. + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY + assert result["title"] == TEST_FRIENDLY_NAME + assert result["data"] == { + CONFIG_ENTRY_ST: TEST_ST, + CONFIG_ENTRY_UDN: TEST_UDN, + CONFIG_ENTRY_ORIGINAL_UDN: TEST_UDN, + CONFIG_ENTRY_LOCATION: TEST_LOCATION, + CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS, + CONFIG_ENTRY_HOST: TEST_HOST, + } diff --git a/tests/components/upnp/test_init.py b/tests/components/upnp/test_init.py index c46f3187b21..5c59ddd9da7 100644 --- a/tests/components/upnp/test_init.py +++ b/tests/components/upnp/test_init.py @@ -2,10 +2,12 @@ from __future__ import annotations -from unittest.mock import AsyncMock +import copy +from unittest.mock import AsyncMock, MagicMock, patch import pytest +from homeassistant.components import ssdp from homeassistant.components.upnp.const import ( CONFIG_ENTRY_LOCATION, CONFIG_ENTRY_MAC_ADDRESS, @@ -16,7 +18,14 @@ from homeassistant.components.upnp.const import ( ) from homeassistant.core import HomeAssistant -from .conftest import TEST_LOCATION, TEST_MAC_ADDRESS, TEST_ST, TEST_UDN, TEST_USN +from .conftest import ( + TEST_DISCOVERY, + TEST_LOCATION, + TEST_MAC_ADDRESS, + TEST_ST, + TEST_UDN, + TEST_USN, +) from tests.common import MockConfigEntry @@ -95,3 +104,44 @@ async def test_async_setup_entry_multi_location( # Ensure that the IPv4 location is used. mock_async_create_device.assert_called_once_with(TEST_LOCATION) + + +@pytest.mark.usefixtures("mock_get_source_ip", "mock_mac_address_from_host") +async def test_async_setup_udn_mismatch( + hass: HomeAssistant, mock_async_create_device: AsyncMock +) -> None: + """Test async_setup_entry for a device which reports a different UDN from SSDP-discovery and device description.""" + test_discovery = copy.deepcopy(TEST_DISCOVERY) + test_discovery.upnp[ssdp.ATTR_UPNP_UDN] = "uuid:another_udn" + + entry = MockConfigEntry( + domain=DOMAIN, + unique_id=TEST_USN, + data={ + CONFIG_ENTRY_ST: TEST_ST, + CONFIG_ENTRY_UDN: TEST_UDN, + CONFIG_ENTRY_ORIGINAL_UDN: TEST_UDN, + CONFIG_ENTRY_LOCATION: TEST_LOCATION, + CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS, + }, + ) + + # Set up device discovery callback. + async def register_callback(hass, callback, match_dict): + """Immediately do callback.""" + await callback(test_discovery, ssdp.SsdpChange.ALIVE) + return MagicMock() + + with patch( + "homeassistant.components.ssdp.async_register_callback", + side_effect=register_callback, + ), patch( + "homeassistant.components.ssdp.async_get_discovery_info_by_st", + return_value=[test_discovery], + ): + # Load config_entry. + entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(entry.entry_id) is True + + # Ensure that the IPv4 location is used. + mock_async_create_device.assert_called_once_with(TEST_LOCATION)