diff --git a/homeassistant/components/knx/services.py b/homeassistant/components/knx/services.py index 113be9709ee..c8086c4a62a 100644 --- a/homeassistant/components/knx/services.py +++ b/homeassistant/components/knx/services.py @@ -13,7 +13,7 @@ from xknx.telegram import Telegram from xknx.telegram.address import parse_device_group_address from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite -from homeassistant.const import CONF_TYPE, SERVICE_RELOAD +from homeassistant.const import CONF_TYPE, CONF_VALUE_TEMPLATE, SERVICE_RELOAD from homeassistant.core import HomeAssistant, ServiceCall, callback from homeassistant.exceptions import HomeAssistantError, ServiceValidationError import homeassistant.helpers.config_validation as cv @@ -143,6 +143,7 @@ SERVICE_KNX_EXPOSURE_REGISTER_SCHEMA = vol.Any( ExposeSchema.EXPOSE_SENSOR_SCHEMA.extend( { vol.Optional(SERVICE_KNX_ATTR_REMOVE, default=False): cv.boolean, + vol.Optional(CONF_VALUE_TEMPLATE): cv.string, } ), vol.Schema( diff --git a/homeassistant/core.py b/homeassistant/core.py index cdfb5570b44..9b210e1ca28 100644 --- a/homeassistant/core.py +++ b/homeassistant/core.py @@ -2610,6 +2610,10 @@ class ServiceRegistry: This method must be run in the event loop. """ + # pylint: disable-next=import-outside-toplevel + from .helpers import config_validation as cv + + cv.raise_on_templated_service(domain, service, schema) domain = domain.lower() service = service.lower() service_obj = Service( diff --git a/homeassistant/helpers/config_validation.py b/homeassistant/helpers/config_validation.py index 81ac10f86cc..b540bede64d 100644 --- a/homeassistant/helpers/config_validation.py +++ b/homeassistant/helpers/config_validation.py @@ -1380,6 +1380,60 @@ def _make_entity_service_schema(schema: dict, extra: int) -> VolSchemaType: BASE_ENTITY_SCHEMA = _make_entity_service_schema({}, vol.PREVENT_EXTRA) +def raise_on_templated_service( + domain: str, _service: str, schema: VolDictType | VolSchemaType | None +) -> None: + """Raise if service schema explicitly allows templates.""" + return _raise_on_templated_service(domain, _service, schema, []) + + +def _raise_on_templated_service( + domain: str, + _service: str, + schema: Any, + _path: list[str | int], +) -> None: + """Raise if service schema explicitly allows templates.""" + if not schema: + return + if isinstance(schema, dict): + for key, val in schema.items(): + _raise_on_templated_service(domain, _service, val, [*_path, key]) + return + if isinstance(schema, list): + for pos, val in enumerate(schema): + _raise_on_templated_service(domain, _service, val, [*_path, pos]) + return + if isinstance(schema, vol.All): + for pos, val in enumerate(schema.validators): + _raise_on_templated_service(domain, _service, val, [*_path, "All", pos]) + if isinstance(schema, vol.Any): + for pos, val in enumerate(schema.validators): + _raise_on_templated_service(domain, _service, val, [*_path, "Any", pos]) + if isinstance(schema, (vol.Schema)): + _raise_on_templated_service(domain, _service, schema.schema, _path) + if _path[-5:] == ["All", 0, "entity_id", "Any", 1]: + return + if _path[-7:] == ["All", 0, "entity_id", "Any", 2, "All", 1]: + return + if _path[-10:] == ["All", 0, "device_id", "Any", 1, "All", 1, 0, "Any", 0]: + return + if _path[-10:] == ["All", 0, "area_id", "Any", 1, "All", 1, 0, "Any", 0]: + return + if _path[-10:] == ["All", 0, "floor_id", "Any", 1, "All", 1, 0, "Any", 0]: + return + if _path[-10:] == ["All", 0, "label_id", "Any", 1, "All", 1, 0, "Any", 0]: + return + if domain == "camera" and _service in ("record", "snapshot"): + return + if domain == "unifiprotect" and _service == "set_chime_paired_doorbells": + return + if schema in (dynamic_template, template, template_complex): + raise ValueError( + f"Template in service data is not allowed! {domain}.{_service}:{_path}" + ) + + def make_entity_service_schema( schema: dict | None, *, extra: int = vol.PREVENT_EXTRA ) -> VolSchemaType: