From 36cc629fafb15738ced49a3f4e52c87db92f32ce Mon Sep 17 00:00:00 2001 From: Artur Pragacz <49985303+arturpragacz@users.noreply.github.com> Date: Tue, 12 May 2026 11:01:04 +0200 Subject: [PATCH] Validate device info string fields in the registry (#170021) --- homeassistant/components/broadlink/entity.py | 4 +- .../components/slide_local/entity.py | 4 +- .../components/smarty/coordinator.py | 4 +- homeassistant/helpers/device_registry.py | 128 ++++++++++++------ tests/components/broadlink/test_device.py | 2 +- .../slide_local/snapshots/test_init.ambr | 4 +- .../smarty/snapshots/test_init.ambr | 4 +- tests/helpers/test_device_registry.py | 55 ++++++++ 8 files changed, 153 insertions(+), 52 deletions(-) diff --git a/homeassistant/components/broadlink/entity.py b/homeassistant/components/broadlink/entity.py index a97374680f9e..14b0c6b20152 100644 --- a/homeassistant/components/broadlink/entity.py +++ b/homeassistant/components/broadlink/entity.py @@ -64,5 +64,7 @@ class BroadlinkEntity(Entity): manufacturer=device.api.manufacturer, model=device.api.model, name=device.name, - sw_version=device.fw_version, + sw_version=str(device.fw_version) + if device.fw_version is not None + else None, ) diff --git a/homeassistant/components/slide_local/entity.py b/homeassistant/components/slide_local/entity.py index 51269649add3..5b57fc03341b 100644 --- a/homeassistant/components/slide_local/entity.py +++ b/homeassistant/components/slide_local/entity.py @@ -20,8 +20,8 @@ class SlideEntity(CoordinatorEntity[SlideCoordinator]): manufacturer="Innovation in Motion", connections={(dr.CONNECTION_NETWORK_MAC, coordinator.data["mac"])}, name=coordinator.data["device_name"], - sw_version=coordinator.api_version, - hw_version=coordinator.data["board_rev"], + sw_version=str(coordinator.api_version), + hw_version=str(coordinator.data["board_rev"]), serial_number=coordinator.data["mac"], configuration_url=f"http://{coordinator.host}", ) diff --git a/homeassistant/components/smarty/coordinator.py b/homeassistant/components/smarty/coordinator.py index a55c9f2e78fb..a7075ef51476 100644 --- a/homeassistant/components/smarty/coordinator.py +++ b/homeassistant/components/smarty/coordinator.py @@ -36,8 +36,8 @@ class SmartyCoordinator(DataUpdateCoordinator[None]): async def _async_setup(self) -> None: if not await self.hass.async_add_executor_job(self.client.update): raise UpdateFailed("Failed to update Smarty data") - self.software_version = self.client.get_software_version() - self.configuration_version = self.client.get_configuration_version() + self.software_version = str(self.client.get_software_version()) + self.configuration_version = str(self.client.get_configuration_version()) async def _async_update_data(self) -> None: """Fetch data from Smarty.""" diff --git a/homeassistant/helpers/device_registry.py b/homeassistant/helpers/device_registry.py index 850b80ff617b..df7cc565279b 100644 --- a/homeassistant/helpers/device_registry.py +++ b/homeassistant/helpers/device_registry.py @@ -8,7 +8,7 @@ from enum import StrEnum from functools import lru_cache import logging import time -from typing import TYPE_CHECKING, Any, Literal, TypedDict +from typing import TYPE_CHECKING, Any, Literal, TypedDict, Unpack import attr from yarl import URL @@ -222,11 +222,11 @@ class DeviceConnectionCollisionError(DeviceCollisionError): ) -def _validate_device_info( +def _determine_device_info_type( config_entry: ConfigEntry, device_info: DeviceInfo, ) -> str: - """Process a device info.""" + """Determine the type of a device info.""" keys = set(device_info) # If no keys or not enough info to match up, abort @@ -258,22 +258,66 @@ def _validate_device_info( return device_info_type +class _ValidatedDeviceInfoFields(TypedDict): + """Device info fields validated on create and update.""" + + configuration_url: str | URL | None | UndefinedType + hw_version: str | None | UndefinedType + manufacturer: str | None | UndefinedType + model: str | None | UndefinedType + model_id: str | None | UndefinedType + serial_number: str | None | UndefinedType + sw_version: str | None | UndefinedType + + _cached_parse_url = lru_cache(maxsize=512)(URL) """Parse a URL and cache the result.""" -def _validate_configuration_url(value: Any) -> str | None: - """Validate and convert configuration_url.""" - if value is None: - return None +def _validate_str(name: str, value: Any) -> str | None | UndefinedType: + """Validate that a device registry string field has correct type.""" + if ( + value is UNDEFINED + or value is None + or type(value) is str # fast path for exact str + or isinstance(value, str) + ): + return value + report_usage( + f"passes a non-string value of type {type(value).__name__} " + f"as {name} to the device registry", + core_behavior=ReportBehavior.LOG, + breaks_in_ha_version="2026.12.0", + ) + return str(value) - url_as_str = str(value) - url = value if type(value) is URL else _cached_parse_url(url_as_str) - if url.scheme not in CONFIGURATION_URL_SCHEMES or not url.host: - raise ValueError(f"invalid configuration_url '{value}'") - - return url_as_str +def _validate_device_info_fields( + **fields: Unpack[_ValidatedDeviceInfoFields], +) -> _ValidatedDeviceInfoFields: + """Validate device-info field values.""" + configuration_url = fields["configuration_url"] + url: URL | None = None + if type(configuration_url) is URL: + url = configuration_url + configuration_url = str(configuration_url) + else: + configuration_url = _validate_str("configuration_url", configuration_url) + if isinstance(configuration_url, str): + url = _cached_parse_url(configuration_url) + if url is not None and ( + url.scheme not in CONFIGURATION_URL_SCHEMES or not url.host + ): + raise ValueError(f"invalid configuration_url '{configuration_url}'") + return { + "configuration_url": configuration_url, + "hw_version": _validate_str("hw_version", fields["hw_version"]), + "manufacturer": _validate_str("manufacturer", fields["manufacturer"]), + "model": _validate_str("model", fields["model"]), + "model_id": _validate_str("model_id", fields["model_id"]), + "serial_number": _validate_str("serial_number", fields["serial_number"]), + "sw_version": _validate_str("sw_version", fields["sw_version"]), + } @lru_cache(maxsize=512) @@ -864,8 +908,19 @@ class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]): via_device: tuple[str, str] | None | UndefinedType = UNDEFINED, ) -> DeviceEntry: """Get device. Create if it doesn't exist.""" - if configuration_url is not UNDEFINED: - configuration_url = _validate_configuration_url(configuration_url) + default_manufacturer = _validate_str( + "default_manufacturer", default_manufacturer + ) + default_model = _validate_str("default_model", default_model) + validated_fields = _validate_device_info_fields( + configuration_url=configuration_url, + hw_version=hw_version, + manufacturer=manufacturer, + model=model, + model_id=model_id, + serial_number=serial_number, + sw_version=sw_version, + ) config_entry = self.hass.config_entries.async_get_entry(config_entry_id) if config_entry is None: @@ -891,27 +946,21 @@ class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]): device_info: DeviceInfo = { # type: ignore[assignment] key: val for key, val in ( - ("configuration_url", configuration_url), ("connections", connections), ("default_manufacturer", default_manufacturer), ("default_model", default_model), ("default_name", default_name), ("entry_type", entry_type), - ("hw_version", hw_version), ("identifiers", identifiers), - ("manufacturer", manufacturer), - ("model", model), - ("model_id", model_id), ("name", name), - ("serial_number", serial_number), ("suggested_area", suggested_area), - ("sw_version", sw_version), ("via_device", via_device), + *validated_fields.items(), ) if val is not UNDEFINED } - device_info_type = _validate_device_info(config_entry, device_info) + device_info_type = _determine_device_info_type(config_entry, device_info) if identifiers is None or identifiers is UNDEFINED: identifiers = set() @@ -963,10 +1012,10 @@ class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]): name = config_entry.title if default_manufacturer is not UNDEFINED and device.manufacturer is None: - manufacturer = default_manufacturer + validated_fields["manufacturer"] = default_manufacturer if default_model is not UNDEFINED and device.model is None: - model = default_model + validated_fields["model"] = default_model if default_name is not UNDEFINED and device.name is None: name = default_name @@ -990,22 +1039,16 @@ class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]): allow_collisions=True, add_config_entry_id=config_entry_id, add_config_subentry_id=config_subentry_id, - configuration_url=configuration_url, device_info_type=device_info_type, disabled_by=disabled_by, entry_type=entry_type, - hw_version=hw_version, is_new=is_new, - manufacturer=manufacturer, merge_connections=connections or UNDEFINED, merge_identifiers=identifiers or UNDEFINED, - model=model, - model_id=model_id, name=name, - serial_number=serial_number, suggested_area=suggested_area, - sw_version=sw_version, via_device_id=via_device_id, + **validated_fields, ) # This is safe because _async_update_device will always return a device @@ -1249,9 +1292,6 @@ class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]): ) old_values["identifiers"] = old.identifiers - if configuration_url is not UNDEFINED: - configuration_url = _validate_configuration_url(configuration_url) - for attr_name, value in ( ("area_id", area_id), ("configuration_url", configuration_url), @@ -1361,32 +1401,36 @@ class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]): breaks_in_ha_version="2026.9.0", ) + validated_fields = _validate_device_info_fields( + configuration_url=configuration_url, + hw_version=hw_version, + manufacturer=manufacturer, + model=model, + model_id=model_id, + serial_number=serial_number, + sw_version=sw_version, + ) + return self._async_update_device( device_id, add_config_entry_id=add_config_entry_id, add_config_subentry_id=add_config_subentry_id, area_id=area_id, - configuration_url=configuration_url, device_info_type=device_info_type, disabled_by=disabled_by, entry_type=entry_type, - hw_version=hw_version, labels=labels, - manufacturer=manufacturer, merge_connections=merge_connections, merge_identifiers=merge_identifiers, - model=model, - model_id=model_id, name_by_user=name_by_user, name=name, new_connections=new_connections, new_identifiers=new_identifiers, remove_config_entry_id=remove_config_entry_id, remove_config_subentry_id=remove_config_subentry_id, - serial_number=serial_number, suggested_area=suggested_area, - sw_version=sw_version, via_device_id=via_device_id, + **validated_fields, ) @callback diff --git a/tests/components/broadlink/test_device.py b/tests/components/broadlink/test_device.py index c9f22dbcbf85..caaec8318189 100644 --- a/tests/components/broadlink/test_device.py +++ b/tests/components/broadlink/test_device.py @@ -262,7 +262,7 @@ async def test_device_setup_registry( assert device_entry.name == device.name assert device_entry.model == device.model assert device_entry.manufacturer == device.manufacturer - assert device_entry.sw_version == device.fwversion + assert device_entry.sw_version == str(device.fwversion) for entry in er.async_entries_for_device(entity_registry, device_entry.id): assert ( diff --git a/tests/components/slide_local/snapshots/test_init.ambr b/tests/components/slide_local/snapshots/test_init.ambr index cc93a49b98ab..8b9713cb3711 100644 --- a/tests/components/slide_local/snapshots/test_init.ambr +++ b/tests/components/slide_local/snapshots/test_init.ambr @@ -13,7 +13,7 @@ }), 'disabled_by': None, 'entry_type': None, - 'hw_version': 1, + 'hw_version': '1', 'id': , 'identifiers': set({ }), @@ -26,7 +26,7 @@ 'name_by_user': None, 'primary_config_entry': , 'serial_number': '1234567890ab', - 'sw_version': 2, + 'sw_version': '2', 'via_device_id': None, }) # --- diff --git a/tests/components/smarty/snapshots/test_init.ambr b/tests/components/smarty/snapshots/test_init.ambr index 989e95bde7e4..109fd649533e 100644 --- a/tests/components/smarty/snapshots/test_init.ambr +++ b/tests/components/smarty/snapshots/test_init.ambr @@ -9,7 +9,7 @@ }), 'disabled_by': None, 'entry_type': None, - 'hw_version': 111, + 'hw_version': '111', 'id': , 'identifiers': set({ tuple( @@ -26,7 +26,7 @@ 'name_by_user': None, 'primary_config_entry': , 'serial_number': None, - 'sw_version': 127, + 'sw_version': '127', 'via_device_id': None, }) # --- diff --git a/tests/helpers/test_device_registry.py b/tests/helpers/test_device_registry.py index b184525c47f4..e8e782b3b024 100644 --- a/tests/helpers/test_device_registry.py +++ b/tests/helpers/test_device_registry.py @@ -4914,6 +4914,61 @@ async def test_device_info_configuration_url_validation( ) +@pytest.mark.parametrize( + "field", + [ + "hw_version", + "manufacturer", + "model", + "model_id", + "serial_number", + "sw_version", + ], +) +@pytest.mark.parametrize( + ("value", "stored_value", "expected_log"), + [ + (1.0, "1.0", "passes a non-string value of type float as {field}"), + ((1, 2), "(1, 2)", "passes a non-string value of type tuple as {field}"), + ("hw-1", "hw-1", ""), + (None, None, ""), + ], +) +async def test_device_info_string_field_validation( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + caplog: pytest.LogCaptureFixture, + field: str, + value: Any, + stored_value: str | None, + expected_log: str, +) -> None: + """Test string device info fields are validated and coerced.""" + config_entry_1 = MockConfigEntry() + config_entry_1.add_to_hass(hass) + config_entry_2 = MockConfigEntry() + config_entry_2.add_to_hass(hass) + + entry = device_registry.async_get_or_create( + config_entry_id=config_entry_1.entry_id, + identifiers={("something", "1234")}, + name="name", + **{field: value}, + ) + assert getattr(entry, field) == stored_value + + update_device = device_registry.async_get_or_create( + config_entry_id=config_entry_2.entry_id, + identifiers={("something", "5678")}, + name="name", + ) + updated = device_registry.async_update_device(update_device.id, **{field: value}) + assert updated is not None + assert getattr(updated, field) == stored_value + + assert expected_log.format(field=field) in caplog.text + + @pytest.mark.parametrize("load_registries", [False]) async def test_loading_invalid_configuration_url_from_storage( hass: HomeAssistant,