Merge branch 'dev' into fail_on_templated_service_data

homeassistant/components/knx/services.py

@@ -12,7 +12,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
 from homeassistant.helpers import config_validation as cv
@@ -144,6 +144,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(

homeassistant/core.py

@@ -2624,6 +2624,9 @@ class ServiceRegistry:
 
         This method must be run in the event loop.
         """
+        from .helpers import config_validation as cv  # noqa: PLC0415
+
+        cv.raise_on_templated_service(domain, service, schema)
         domain = domain.lower()
         service = service.lower()
         service_obj = Service(

homeassistant/helpers/config_validation.py

@@ -1423,6 +1423,48 @@ 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 domain == "camera" and _service in ("record", "snapshot"):
+        return
+    if domain == "velbus" and _service == "set_memo_text":
+        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: