Handle cname records in dnsip (#175313)

homeassistant/components/dnsip/sensor.py

@@ -35,14 +35,14 @@ _LOGGER = logging.getLogger(__name__)
 SCAN_INTERVAL = timedelta(seconds=120)
 
 
-def sort_ips(ips: list, querytype: Literal["A", "AAAA"]) -> list:
+def sort_ips(ips: list[str], querytype: Literal["A", "AAAA"]) -> list[str]:
     """Join IPs into a single string."""
-
+    _ips: list[IPv4Address | IPv6Address]
     if querytype == "AAAA":
-        ips = [IPv6Address(ip) for ip in ips]
+        _ips = [IPv6Address(ip) for ip in ips]
     else:
-        ips = [IPv4Address(ip) for ip in ips]
-    return [str(ip) for ip in sorted(ips)][:MAX_RESULTS]
+        _ips = [IPv4Address(ip) for ip in ips]
+    return [str(ip) for ip in sorted(_ips)][:MAX_RESULTS]
 
 
 async def async_setup_entry(
@@ -142,35 +142,34 @@ class WanIpSensor(SensorEntity):
         else:
             self.entry.runtime_data.resolver_ipv4 = new_resolver
 
+    async def async_resolve(self, hostname: str) -> list[str]:
+        """Resolve a hostname to its IP addresses."""
+        ips: list[str] = []
+        try:
+            async with asyncio.timeout(10):
+                response = await self._resolver.query_dns(hostname, self.querytype)
+        except TimeoutError as err:
+            _LOGGER.debug("Timeout while resolving host: %s", err)
+            await self._resolver.close()
+            return ips
+        except DNSError as err:
+            _LOGGER.warning("Exception while resolving host: %s", err)
+            await self._resolver.close()
+            return ips
+
+        for res in response.answer:
+            if isinstance(res.data, (pycares.AAAARecordData, pycares.ARecordData)):
+                addr = res.data.addr
+                ips.append(addr)
+        return ips
+
     async def async_update(self) -> None:
         """Get the current DNS IP address for hostname."""
         if self._resolver._closed:  # noqa: SLF001
             self.create_dns_resolver()
-        response = None
-        try:
-            async with asyncio.timeout(10):
-                response = await self._resolver.query_dns(self.hostname, self.querytype)
-        except TimeoutError as err:
-            _LOGGER.debug("Timeout while resolving host: %s", err)
-            await self._resolver.close()
-        except DNSError as err:
-            _LOGGER.warning("Exception while resolving host: %s", err)
-            await self._resolver.close()
 
-        if response:
-            if TYPE_CHECKING:
-                assert all(
-                    isinstance(res.data, (pycares.ARecordData, pycares.AAAARecordData))
-                    for res in response.answer
-                )
-            _ips = []
-            for res in response.answer:
-                if TYPE_CHECKING:
-                    assert isinstance(
-                        res.data, (pycares.ARecordData, pycares.AAAARecordData)
-                    )
-                _ips.append(res.data.addr)
-            sorted_ips = sort_ips(_ips, querytype=self.querytype)
+        if ips := await self.async_resolve(self.hostname):
+            sorted_ips = sort_ips(ips, querytype=self.querytype)
             self._attr_native_value = sorted_ips[0]
             self._attr_extra_state_attributes["ip_addresses"] = sorted_ips
             self._attr_available = True

tests/components/dnsip/__init__.py

@@ -69,6 +69,13 @@ class RetrieveDNS:
         else:
             results = pycares.DNSResult(
                 answer=[
+                    pycares.DNSRecord(
+                        name="test",
+                        type=pycares.QUERY_TYPE_CNAME,
+                        record_class=pycares.QUERY_CLASS_IN,
+                        data=pycares.CNAMERecordData(cname="test.testing.com"),
+                        ttl=60,
+                    ),
                     pycares.DNSRecord(
                         name="test",
                         type=pycares.QUERY_TYPE_A,

tests/components/dnsip/test_sensor.py

@@ -234,3 +234,38 @@ async def test_sensor_timeout(
 
     state = hass.states.get("sensor.home_assistant_io")
     assert state.state == STATE_UNAVAILABLE
+
+
+async def test_handle_cname(hass: HomeAssistant) -> None:
+    """Test CNAME records are dropped."""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        source=SOURCE_USER,
+        data={
+            CONF_HOSTNAME: "home-assistant.io",
+            CONF_NAME: "home-assistant.io",
+            CONF_IPV4: True,
+            CONF_IPV6: False,
+        },
+        options={
+            CONF_RESOLVER: "208.67.222.222",
+            CONF_RESOLVER_IPV6: "2620:119:53::53",
+            CONF_PORT: 53,
+            CONF_PORT_IPV6: 53,
+        },
+        entry_id="1",
+        unique_id="home-assistant.io",
+    )
+    entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.dnsip.aiodns.DNSResolver",
+        return_value=RetrieveDNS(),
+    ):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    state1 = hass.states.get("sensor.home_assistant_io")
+
+    assert state1.state == "1.1.1.1"
+    assert state1.attributes["ip_addresses"] == ["1.1.1.1", "1.2.3.4"]