SSDP response decode: replace invalid utf-8 characters (#42681)

* SSDP response decode: replace invalid utf-8 characters

* Add test to validate replaced data

Co-authored-by: Joakim Plate <elupus@ecce.se>

homeassistant/components/ssdp/__init__.py

@@ -171,13 +171,13 @@ class Scanner:
         session = self.hass.helpers.aiohttp_client.async_get_clientsession()
         try:
             resp = await session.get(xml_location, timeout=5)
-            xml = await resp.text()
+            xml = await resp.text(errors="replace")
 
             # Samsung Smart TV sometimes returns an empty document the
             # first time. Retry once.
             if not xml:
                 resp = await session.get(xml_location, timeout=5)
-                xml = await resp.text()
+                xml = await resp.text(errors="replace")
         except (aiohttp.ClientError, asyncio.TimeoutError) as err:
             _LOGGER.debug("Error fetching %s: %s", xml_location, err)
             return {}

tests/components/ssdp/test_init.py

@@ -170,3 +170,46 @@ async def test_scan_description_parse_fail(hass, aioclient_mock):
         return_value=[Mock(st="mock-st", location="http://1.1.1.1", values={})],
     ):
         await scanner.async_scan(None)
+
+
+async def test_invalid_characters(hass, aioclient_mock):
+    """Test that we replace bad characters with placeholders."""
+    aioclient_mock.get(
+        "http://1.1.1.1",
+        text="""
+<root>
+  <device>
+    <deviceType>ABC</deviceType>
+    <serialNumber>\xff\xff\xff\xff</serialNumber>
+  </device>
+</root>
+    """,
+    )
+    scanner = ssdp.Scanner(
+        hass,
+        {
+            "mock-domain": [
+                {
+                    ssdp.ATTR_UPNP_DEVICE_TYPE: "ABC",
+                }
+            ]
+        },
+    )
+
+    with patch(
+        "netdisco.ssdp.scan",
+        return_value=[Mock(st="mock-st", location="http://1.1.1.1", values={})],
+    ), patch.object(
+        hass.config_entries.flow, "async_init", return_value=mock_coro()
+    ) as mock_init:
+        await scanner.async_scan(None)
+
+    assert len(mock_init.mock_calls) == 1
+    assert mock_init.mock_calls[0][1][0] == "mock-domain"
+    assert mock_init.mock_calls[0][2]["context"] == {"source": "ssdp"}
+    assert mock_init.mock_calls[0][2]["data"] == {
+        "ssdp_location": "http://1.1.1.1",
+        "ssdp_st": "mock-st",
+        "deviceType": "ABC",
+        "serialNumber": "ÿÿÿÿ",
+    }

tests/test_util/aiohttp.py

@@ -245,9 +245,9 @@ class AiohttpClientMockResponse:
         """Return mock response."""
         return self.response
 
-    async def text(self, encoding="utf-8"):
+    async def text(self, encoding="utf-8", errors="strict"):
         """Return mock response as a string."""
-        return self.response.decode(encoding)
+        return self.response.decode(encoding, errors=errors)
 
     async def json(self, encoding="utf-8", content_type=None):
         """Return mock response as a json."""