diff --git a/esphome/components/dlms_meter/__init__.py b/esphome/components/dlms_meter/__init__.py index 7094699b0b..b747f73a14 100644 --- a/esphome/components/dlms_meter/__init__.py +++ b/esphome/components/dlms_meter/__init__.py @@ -136,12 +136,8 @@ CONFIG_SCHEMA = cv.All( cv.Schema( { cv.GenerateID(): cv.declare_id(DlmsMeterComponent), - cv.Optional(CONF_DECRYPTION_KEY): lambda value: cv.bind_key( - value, name="Decryption key" - ), - cv.Optional(CONF_AUTH_KEY): lambda value: cv.bind_key( - value, name="Authentication key" - ), + cv.Optional(CONF_DECRYPTION_KEY): cv.bind_key(name="Decryption key"), + cv.Optional(CONF_AUTH_KEY): cv.bind_key(name="Authentication key"), cv.Optional(CONF_CUSTOM_PATTERNS): cv.ensure_list(CUSTOM_PATTERN_SCHEMA), cv.Optional(CONF_SKIP_CRC, default=False): cv.boolean, cv.Optional(CONF_PROVIDER): cv.string, diff --git a/esphome/components/dsmr/__init__.py b/esphome/components/dsmr/__init__.py index 1dc3664602..34f37ace35 100644 --- a/esphome/components/dsmr/__init__.py +++ b/esphome/components/dsmr/__init__.py @@ -40,9 +40,7 @@ CONFIG_SCHEMA = cv.All( cv.Schema( { cv.GenerateID(): cv.declare_id(Dsmr), - cv.Optional(CONF_DECRYPTION_KEY): lambda value: cv.bind_key( - value, name="Decryption key" - ), + cv.Optional(CONF_DECRYPTION_KEY): cv.bind_key(name="Decryption key"), cv.Optional(CONF_CRC_CHECK, default=True): cv.boolean, cv.Optional(CONF_GAS_MBUS_ID, default=1): cv.int_, cv.Optional(CONF_WATER_MBUS_ID, default=2): cv.int_, diff --git a/esphome/config_validation.py b/esphome/config_validation.py index 0ef6d212fe..0fdce85dc3 100644 --- a/esphome/config_validation.py +++ b/esphome/config_validation.py @@ -1220,21 +1220,60 @@ def mac_address(value): return core.MACAddress(*parts_int) -def bind_key(value, *, name="Bind key"): - value = string_strict(value) - parts = [value[i : i + 2] for i in range(0, len(value), 2)] - if len(parts) != 16: - raise Invalid(f"{name} must consist of 16 hexadecimal numbers") - parts_int = [] - if any(len(part) != 2 for part in parts): - raise Invalid(f"{name} must be format XX") - for part in parts: - try: - parts_int.append(int(part, 16)) - except ValueError: - raise Invalid(f"{name} must be hex values from 00 to FF") from None +_BIND_KEY_MISSING = object() - return "".join(f"{part:02X}" for part in parts_int) + +class BindKeyValidator(SensitiveValidator): + """Sensitive validator for a 16-byte hex bind/encryption key. + + Use bare as a validator (``cv.bind_key``) for the default error wording, or + call it with a custom ``name`` (``cv.bind_key(name="Decryption key")``) to + get a validator with tailored error messages. Either way the value is marked + sensitive so frontends mask it and dump tooling redacts it. + """ + + def __init__(self, name: str = "Bind key") -> None: + self._name = name + super().__init__(self._validate) + + def _validate(self, value: typing.Any) -> str: + value = string_strict(value) + parts = [value[i : i + 2] for i in range(0, len(value), 2)] + if len(parts) != 16: + raise Invalid(f"{self._name} must consist of 16 hexadecimal numbers") + parts_int = [] + if any(len(part) != 2 for part in parts): + raise Invalid(f"{self._name} must be format XX") + for part in parts: + try: + parts_int.append(int(part, 16)) + except ValueError: + raise Invalid( + f"{self._name} must be hex values from 00 to FF" + ) from None + + return "".join(f"{part:02X}" for part in parts_int) + + def __call__( + self, value: typing.Any = _BIND_KEY_MISSING, *, name: str | None = None + ) -> typing.Any: + if value is _BIND_KEY_MISSING: + # Factory usage: return a validator with customized error wording. + return BindKeyValidator(name if name is not None else self._name) + if name is not None and name != self._name: + # Direct validation with a one-off custom name. + return BindKeyValidator(name)(value) + return super().__call__(value) + + def __repr__(self) -> str: + # ``self.inner`` is a bound method of this instance, so the inherited + # ``SensitiveValidator.__repr__`` (which returns ``repr(self.inner)``) + # would recurse infinitely. Provide a stable, name-keyed repr instead so + # ``build_language_schema`` dedup and voluptuous errors stay sane. + return f"bind_key({self._name!r})" + + +bind_key = BindKeyValidator() def uuid(value): diff --git a/tests/unit_tests/test_config_validation.py b/tests/unit_tests/test_config_validation.py index f1a6118870..9b9f003b0d 100644 --- a/tests/unit_tests/test_config_validation.py +++ b/tests/unit_tests/test_config_validation.py @@ -188,6 +188,92 @@ def test_sensitive__is_detectable_via_isinstance() -> None: assert isinstance(validator, config_validation.SensitiveValidator) +def test_bind_key__bare_usage_validates_and_is_sensitive() -> None: + # Used bare (cv.bind_key) it is itself a sensitive validator: detectable for + # frontend masking and validating a value directly tags the result. + assert isinstance(config_validation.bind_key, config_validation.SensitiveValidator) + + result = config_validation.bind_key("0123456789ABCDEF0123456789ABCDEF") + + assert isinstance(result, SensitiveStr) + assert result == "0123456789ABCDEF0123456789ABCDEF" + + +def test_bind_key__bare_usage_in_schema() -> None: + # Voluptuous calls the bare validator with the config value; the result must + # come through tagged sensitive. + schema = config_validation.Schema( + {config_validation.Required("key"): config_validation.bind_key} + ) + out = schema({"key": "0123456789ABCDEF0123456789ABCDEF"}) + + assert isinstance(out["key"], SensitiveStr) + + +def test_bind_key__factory_returns_sensitive_validator() -> None: + # Called with a name (cv.bind_key(name=...)) it returns a new sensitive + # validator rather than validating. + validator = config_validation.bind_key(name="Decryption key") + + assert isinstance(validator, config_validation.SensitiveValidator) + assert validator is not config_validation.bind_key + assert isinstance(validator("0123456789ABCDEF0123456789ABCDEF"), SensitiveStr) + + +@pytest.mark.parametrize( + ("value", "error"), + ( + ("00", "Decryption key must consist of 16 hexadecimal numbers"), + ("0123456789ABCDEF0123456789ABCDEG", "Decryption key must be hex values"), + ), +) +def test_bind_key__custom_name_in_error(value: str, error: str) -> None: + # The ``name`` argument (used by dsmr/dlms_meter) customizes error messages. + validator = config_validation.bind_key(name="Decryption key") + with pytest.raises(Invalid, match=error): + validator(value) + + +def test_bind_key__rejects_non_hex_pair_length() -> None: + # Odd-length input yields a trailing single-char part, hitting the + # "format XX" branch rather than the hex-value branch. + with pytest.raises(Invalid, match="Bind key must be format XX"): + config_validation.bind_key("0123456789ABCDEF0123456789ABCDE") + + +def test_bind_key__direct_call_with_name_validates_with_that_name() -> None: + # Passing both a value and a name validates immediately using the custom + # name for error wording, and still tags the result sensitive. + result = config_validation.bind_key( + "0123456789ABCDEF0123456789ABCDEF", name="Decryption key" + ) + assert isinstance(result, SensitiveStr) + + with pytest.raises(Invalid, match="Decryption key must consist of"): + config_validation.bind_key("00", name="Decryption key") + + +def test_bind_key__factory_without_name_keeps_existing_name() -> None: + # Re-invoking a named validator without a name preserves its name rather + # than resetting to the default. + named = config_validation.bind_key(name="Decryption key") + rederived = named() + + with pytest.raises(Invalid, match="Decryption key must consist of"): + rederived("00") + + +def test_bind_key__repr_is_name_keyed_and_non_recursive() -> None: + # ``self.inner`` is a bound method of the instance, so the inherited + # ``repr(self.inner)`` would recurse infinitely; the override keeps repr + # finite and keyed on the name for schema-dump dedup. + assert repr(config_validation.bind_key) == "bind_key('Bind key')" + assert ( + repr(config_validation.bind_key(name="Decryption key")) + == "bind_key('Decryption key')" + ) + + def test_sensitive__repr_mirrors_inner() -> None: # The schema dump dedups on ``repr(schema)``; mirroring the inner # validator's repr keeps two ``cv.sensitive(cv.string)`` wrappers