Improve discovering upnp/igd device by always using the SSDP-discovery for the Unique Device Name

homeassistant/components/upnp/__init__.py

@@ -79,6 +79,7 @@ async def device_discovered(
 
     # 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:
@@ -117,7 +118,9 @@ async def device_discovered(
     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))
 

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(config_entries.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]:
@@ -241,9 +240,9 @@ async def async_step_ignore(self, user_input: dict[str, Any]) -> FlowResult:
         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),
@@ -265,9 +264,9 @@ async def _async_create_entry_from_discovery(
         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"],

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,
+    }

tests/components/upnp/test_init.py

@@ -1,10 +1,12 @@
 """Test UPnP/IGD setup process."""
 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,
@@ -15,7 +17,14 @@
 )
 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
 
@@ -94,3 +103,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)