From bce0aca97df0c15f47aa0ed32e23fb34d19fa2f8 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Sun, 26 Jul 2026 23:06:24 +0200 Subject: [PATCH] Add encrypted DSMR support for Luxembourg and Austria (#176058) Co-authored-by: Claude --- homeassistant/components/dsmr/config_flow.py | 92 +++++++++- homeassistant/components/dsmr/const.py | 24 ++- homeassistant/components/dsmr/diagnostics.py | 8 +- homeassistant/components/dsmr/sensor.py | 83 +++++++-- homeassistant/components/dsmr/strings.json | 12 +- tests/components/dsmr/conftest.py | 4 +- .../dsmr/snapshots/test_diagnostics.ambr | 1 + tests/components/dsmr/test_config_flow.py | 168 ++++++++++++++++++ tests/components/dsmr/test_diagnostics.py | 1 + tests/components/dsmr/test_sensor.py | 165 +++++++++++++++++ 10 files changed, 529 insertions(+), 29 deletions(-) diff --git a/homeassistant/components/dsmr/config_flow.py b/homeassistant/components/dsmr/config_flow.py index 439307d8dbac..9130accf24ca 100644 --- a/homeassistant/components/dsmr/config_flow.py +++ b/homeassistant/components/dsmr/config_flow.py @@ -2,6 +2,7 @@ import asyncio from functools import partial +import re from typing import Any, override from dsmr_parser import obis_references as obis_ref @@ -23,6 +24,7 @@ from homeassistant.helpers.selector import SerialPortSelector from .const import ( CONF_DSMR_VERSION, + CONF_ENCRYPTION_KEY, CONF_SERIAL_ID, CONF_SERIAL_ID_GAS, CONF_TIME_BETWEEN_UPDATE, @@ -30,24 +32,36 @@ from .const import ( DOMAIN, DSMR_PROTOCOL, DSMR_VERSIONS, + DSMR_VERSIONS_WITHOUT_EQUIPMENT_ID, + ENCRYPTED_DSMR_VERSIONS, LOGGER, RFXTRX_DSMR_PROTOCOL, ) +ENCRYPTION_KEY_PATTERN = re.compile(r"[0-9a-fA-F]{32}") + class DSMRConnection: """Test the connection to DSMR and receive telegram to read serial ids.""" - def __init__(self, port: str, dsmr_version: str, protocol: str) -> None: + def __init__( + self, + port: str, + dsmr_version: str, + protocol: str, + encryption_key: str = "", + ) -> None: """Initialize.""" self._port = port self._dsmr_version = dsmr_version self._protocol = protocol + self._encryption_key = encryption_key + self._decryption_failed = False self._telegram: dict[str, DSMRObject] = {} self._equipment_identifier = obis_ref.EQUIPMENT_IDENTIFIER if dsmr_version == "5B": self._equipment_identifier = obis_ref.BELGIUM_EQUIPMENT_IDENTIFIER - if dsmr_version in ("5L", "5EONHU"): + if dsmr_version in ("5L", "5EONHU", "MSn"): self._equipment_identifier = obis_ref.LUXEMBOURG_EQUIPMENT_IDENTIFIER if dsmr_version == "Q3D": self._equipment_identifier = obis_ref.Q3D_EQUIPMENT_IDENTIFIER @@ -75,13 +89,24 @@ class DSMRConnection: if self._equipment_identifier in telegram: self._telegram = telegram transport.close() - # Swedish meters have no equipment identifier - if self._dsmr_version == "5S" and obis_ref.P1_MESSAGE_TIMESTAMP in telegram: + # Meters without an equipment identifier fall back to the timestamp + if ( + self._dsmr_version in DSMR_VERSIONS_WITHOUT_EQUIPMENT_ID + and obis_ref.P1_MESSAGE_TIMESTAMP in telegram + ): self._telegram = telegram transport.close() + # Only the standard DSMR reader supports encryption. authentication_key= + # None decrypts without verifying the GCM authentication tag; the + # telegram CRC still catches transmission errors (but not tampering). + key_kwargs: dict[str, Any] = {} if self._protocol == DSMR_PROTOCOL: create_reader = create_dsmr_reader + key_kwargs = { + "encryption_key": self._encryption_key, + "authentication_key": None, + } else: create_reader = create_rfxtrx_dsmr_reader reader_factory = partial( @@ -90,6 +115,7 @@ class DSMRConnection: self._dsmr_version, update_telegram, loop=hass.loop, + **key_kwargs, ) try: @@ -108,8 +134,15 @@ class DSMRConnection: # result in CannotCommunicate error) transport.close() await protocol.wait_closed() + # A wrong key closes the transport with a DecryptionError + if getattr(protocol, "decryption_error", None) is not None: + self._decryption_failed = True return True + def decryption_failed(self) -> bool: + """Return whether decryption failed (wrong key).""" + return self._decryption_failed + async def _validate_dsmr_connection( hass: HomeAssistant, data: dict[str, Any], protocol: str @@ -119,16 +152,23 @@ async def _validate_dsmr_connection( data[CONF_PORT], data[CONF_DSMR_VERSION], protocol, + data.get(CONF_ENCRYPTION_KEY, ""), ) if not await conn.validate_connect(hass): raise CannotConnect + if conn.decryption_failed(): + raise InvalidKey + equipment_identifier = conn.equipment_identifier() equipment_identifier_gas = conn.equipment_identifier_gas() # Check only for equipment identifier in case no gas meter is connected - if equipment_identifier is None and data[CONF_DSMR_VERSION] != "5S": + if ( + equipment_identifier is None + and data[CONF_DSMR_VERSION] not in DSMR_VERSIONS_WITHOUT_EQUIPMENT_ID + ): raise CannotCommunicate return { @@ -142,6 +182,9 @@ class DSMRFlowHandler(ConfigFlow, domain=DOMAIN): VERSION = 1 + _pending_data: dict[str, Any] + _pending_title: str + @staticmethod @callback @override @@ -163,6 +206,11 @@ class DSMRFlowHandler(ConfigFlow, domain=DOMAIN): """ errors: dict[str, str] = {} if user_input is not None: + if user_input[CONF_DSMR_VERSION] in ENCRYPTED_DSMR_VERSIONS: + self._pending_data = user_input + self._pending_title = user_input[CONF_PORT] + return await self.async_step_encryption_key() + data = await self.async_validate_dsmr(user_input, errors) if not errors: return self.async_create_entry(title=data[CONF_PORT], data=data) @@ -179,6 +227,29 @@ class DSMRFlowHandler(ConfigFlow, domain=DOMAIN): errors=errors, ) + async def async_step_encryption_key( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Ask for the encryption key of an encrypted meter.""" + errors: dict[str, str] = {} + if user_input is not None: + # DLMS uses an AES-128 key: 32 hex characters. Reject malformed keys + # here so the user gets immediate feedback instead of a timeout. + if ENCRYPTION_KEY_PATTERN.fullmatch(user_input[CONF_ENCRYPTION_KEY]): + data = await self.async_validate_dsmr( + {**self._pending_data, **user_input}, errors + ) + if not errors: + return self.async_create_entry(title=self._pending_title, data=data) + else: + errors["base"] = "invalid_key" + + return self.async_show_form( + step_id="encryption_key", + data_schema=vol.Schema({vol.Required(CONF_ENCRYPTION_KEY): str}), + errors=errors, + ) + async def async_validate_dsmr( self, input_data: dict[str, Any], errors: dict[str, str] ) -> dict[str, Any]: @@ -190,6 +261,9 @@ class DSMRFlowHandler(ConfigFlow, domain=DOMAIN): protocol = DSMR_PROTOCOL info = await _validate_dsmr_connection(self.hass, data, protocol) except CannotCommunicate: + # Encrypted meters don't support the RFXtrx fallback + if data[CONF_DSMR_VERSION] in ENCRYPTED_DSMR_VERSIONS: + raise protocol = RFXTRX_DSMR_PROTOCOL info = await _validate_dsmr_connection(self.hass, data, protocol) @@ -202,6 +276,8 @@ class DSMRFlowHandler(ConfigFlow, domain=DOMAIN): errors["base"] = "cannot_connect" except CannotCommunicate: errors["base"] = "cannot_communicate" + except InvalidKey: + errors["base"] = "invalid_key" return data @@ -236,4 +312,8 @@ class CannotConnect(HomeAssistantError): class CannotCommunicate(HomeAssistantError): - """Error to indicate we cannot connect.""" + """Error to indicate we cannot communicate with the device.""" + + +class InvalidKey(HomeAssistantError): + """Error to indicate the decryption key is invalid.""" diff --git a/homeassistant/components/dsmr/const.py b/homeassistant/components/dsmr/const.py index 8dbcef1ffc75..ea791ea57f03 100644 --- a/homeassistant/components/dsmr/const.py +++ b/homeassistant/components/dsmr/const.py @@ -11,6 +11,7 @@ LOGGER = logging.getLogger(__package__) PLATFORMS = [Platform.SENSOR] CONF_DSMR_VERSION = "dsmr_version" CONF_TIME_BETWEEN_UPDATE = "time_between_update" +CONF_ENCRYPTION_KEY = "encryption_key" CONF_SERIAL_ID = "serial_id" CONF_SERIAL_ID_GAS = "serial_id_gas" @@ -26,7 +27,28 @@ DEVICE_NAME_GAS = "Gas Meter" DEVICE_NAME_WATER = "Water Meter" DEVICE_NAME_HEAT = "Heat Meter" -DSMR_VERSIONS = {"2.2", "4", "5", "5B", "5L", "5S", "Q3D", "5EONHU"} +# Maps each dsmr_version token to the label shown in the config-flow picker; the +# label disambiguates the Luxembourg Smarty (MSn) from the Austrian Sagemcom. +# Labels are hardcoded rather than translated because the tokens (e.g. "2.2", +# "MSn") are not valid translation keys. +DSMR_VERSIONS = { + "5": "DSMR 5", + "MSn": "Luxembourg Smarty / Sagemcom T210-D, encrypted (Creos)", + "SAGEMCOM_T210_D_R": "Sagemcom T210-D-R, encrypted (Austria, Energienetze Steiermark)", + "5B": "DSMR 5B (Belgium, Fluvius)", + "5L": "DSMR 5L (Luxembourg, unencrypted)", + "5S": "DSMR 5S (Sweden)", + "Q3D": "Q3D (Austria)", + "5EONHU": "DSMR 5 (E.ON Hungary)", + "4": "DSMR 4", + "2.2": "DSMR 2.2", +} + +# Versions with AES-128-GCM encrypted telegrams that require an encryption key. +ENCRYPTED_DSMR_VERSIONS = {"MSn", "SAGEMCOM_T210_D_R"} + +# Versions whose telegrams carry no equipment identifier. +DSMR_VERSIONS_WITHOUT_EQUIPMENT_ID = {"5S", "SAGEMCOM_T210_D_R"} DSMR_PROTOCOL = "dsmr_protocol" RFXTRX_DSMR_PROTOCOL = "rfxtrx_dsmr_protocol" diff --git a/homeassistant/components/dsmr/diagnostics.py b/homeassistant/components/dsmr/diagnostics.py index bdda8ca56423..ef66ecc77cdd 100644 --- a/homeassistant/components/dsmr/diagnostics.py +++ b/homeassistant/components/dsmr/diagnostics.py @@ -2,10 +2,14 @@ from typing import Any +from homeassistant.components.diagnostics import async_redact_data from homeassistant.core import HomeAssistant from homeassistant.util.json import json_loads from . import DsmrConfigEntry +from .const import CONF_ENCRYPTION_KEY + +TO_REDACT = {CONF_ENCRYPTION_KEY} async def async_get_config_entry_diagnostics( @@ -15,9 +19,7 @@ async def async_get_config_entry_diagnostics( return { "entry": { - "data": { - **config_entry.data, - }, + "data": async_redact_data(config_entry.data, TO_REDACT), "unique_id": config_entry.unique_id, }, "data": json_loads(config_entry.runtime_data.telegram.to_json()) diff --git a/homeassistant/components/dsmr/sensor.py b/homeassistant/components/dsmr/sensor.py index 71e39af38713..fa007fe6ac97 100644 --- a/homeassistant/components/dsmr/sensor.py +++ b/homeassistant/components/dsmr/sensor.py @@ -49,6 +49,7 @@ from homeassistant.util import Throttle from . import DsmrConfigEntry from .const import ( CONF_DSMR_VERSION, + CONF_ENCRYPTION_KEY, CONF_SERIAL_ID, CONF_SERIAL_ID_GAS, CONF_TIME_BETWEEN_UPDATE, @@ -116,7 +117,16 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="electricity_active_tariff", translation_key="electricity_active_tariff", obis_reference="ELECTRICITY_ACTIVE_TARIFF", - dsmr_versions={"2.2", "4", "5", "5B", "5L", "5EONHU"}, + dsmr_versions={ + "2.2", + "4", + "5", + "5B", + "5L", + "5EONHU", + "MSn", + "SAGEMCOM_T210_D_R", + }, device_class=SensorDeviceClass.ENUM, options=["low", "normal"], ), @@ -124,7 +134,16 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="electricity_used_tariff_1", translation_key="electricity_used_tariff_1", obis_reference="ELECTRICITY_USED_TARIFF_1", - dsmr_versions={"2.2", "4", "5", "5B", "5L", "5EONHU"}, + dsmr_versions={ + "2.2", + "4", + "5", + "5B", + "5L", + "5EONHU", + "MSn", + "SAGEMCOM_T210_D_R", + }, device_class=SensorDeviceClass.ENERGY, state_class=SensorStateClass.TOTAL_INCREASING, ), @@ -132,7 +151,16 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="electricity_used_tariff_2", translation_key="electricity_used_tariff_2", obis_reference="ELECTRICITY_USED_TARIFF_2", - dsmr_versions={"2.2", "4", "5", "5B", "5L", "5EONHU"}, + dsmr_versions={ + "2.2", + "4", + "5", + "5B", + "5L", + "5EONHU", + "MSn", + "SAGEMCOM_T210_D_R", + }, device_class=SensorDeviceClass.ENERGY, state_class=SensorStateClass.TOTAL_INCREASING, ), @@ -158,7 +186,16 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="electricity_delivered_tariff_1", translation_key="electricity_delivered_tariff_1", obis_reference="ELECTRICITY_DELIVERED_TARIFF_1", - dsmr_versions={"2.2", "4", "5", "5B", "5L", "5EONHU"}, + dsmr_versions={ + "2.2", + "4", + "5", + "5B", + "5L", + "5EONHU", + "MSn", + "SAGEMCOM_T210_D_R", + }, device_class=SensorDeviceClass.ENERGY, state_class=SensorStateClass.TOTAL_INCREASING, ), @@ -166,7 +203,16 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="electricity_delivered_tariff_2", translation_key="electricity_delivered_tariff_2", obis_reference="ELECTRICITY_DELIVERED_TARIFF_2", - dsmr_versions={"2.2", "4", "5", "5B", "5L", "5EONHU"}, + dsmr_versions={ + "2.2", + "4", + "5", + "5B", + "5L", + "5EONHU", + "MSn", + "SAGEMCOM_T210_D_R", + }, device_class=SensorDeviceClass.ENERGY, state_class=SensorStateClass.TOTAL_INCREASING, ), @@ -240,7 +286,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="short_power_failure_count", translation_key="short_power_failure_count", obis_reference="SHORT_POWER_FAILURE_COUNT", - dsmr_versions={"2.2", "4", "5", "5L"}, + dsmr_versions={"2.2", "4", "5", "5L", "MSn"}, entity_registry_enabled_default=False, state_class=SensorStateClass.TOTAL_INCREASING, entity_category=EntityCategory.DIAGNOSTIC, @@ -249,7 +295,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="long_power_failure_count", translation_key="long_power_failure_count", obis_reference="LONG_POWER_FAILURE_COUNT", - dsmr_versions={"2.2", "4", "5", "5L"}, + dsmr_versions={"2.2", "4", "5", "5L", "MSn"}, entity_registry_enabled_default=False, state_class=SensorStateClass.TOTAL_INCREASING, entity_category=EntityCategory.DIAGNOSTIC, @@ -258,7 +304,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="voltage_sag_l1_count", translation_key="voltage_sag_l1_count", obis_reference="VOLTAGE_SAG_L1_COUNT", - dsmr_versions={"2.2", "4", "5", "5L"}, + dsmr_versions={"2.2", "4", "5", "5L", "MSn"}, entity_registry_enabled_default=False, state_class=SensorStateClass.TOTAL_INCREASING, entity_category=EntityCategory.DIAGNOSTIC, @@ -267,7 +313,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="voltage_sag_l2_count", translation_key="voltage_sag_l2_count", obis_reference="VOLTAGE_SAG_L2_COUNT", - dsmr_versions={"2.2", "4", "5", "5L"}, + dsmr_versions={"2.2", "4", "5", "5L", "MSn"}, entity_registry_enabled_default=False, state_class=SensorStateClass.TOTAL_INCREASING, entity_category=EntityCategory.DIAGNOSTIC, @@ -276,7 +322,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="voltage_sag_l3_count", translation_key="voltage_sag_l3_count", obis_reference="VOLTAGE_SAG_L3_COUNT", - dsmr_versions={"2.2", "4", "5", "5L"}, + dsmr_versions={"2.2", "4", "5", "5L", "MSn"}, entity_registry_enabled_default=False, state_class=SensorStateClass.TOTAL_INCREASING, entity_category=EntityCategory.DIAGNOSTIC, @@ -285,7 +331,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="voltage_swell_l1_count", translation_key="voltage_swell_l1_count", obis_reference="VOLTAGE_SWELL_L1_COUNT", - dsmr_versions={"2.2", "4", "5", "5L"}, + dsmr_versions={"2.2", "4", "5", "5L", "MSn"}, entity_registry_enabled_default=False, state_class=SensorStateClass.TOTAL_INCREASING, entity_category=EntityCategory.DIAGNOSTIC, @@ -294,7 +340,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="voltage_swell_l2_count", translation_key="voltage_swell_l2_count", obis_reference="VOLTAGE_SWELL_L2_COUNT", - dsmr_versions={"2.2", "4", "5", "5L"}, + dsmr_versions={"2.2", "4", "5", "5L", "MSn"}, entity_registry_enabled_default=False, state_class=SensorStateClass.TOTAL_INCREASING, entity_category=EntityCategory.DIAGNOSTIC, @@ -303,7 +349,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="voltage_swell_l3_count", translation_key="voltage_swell_l3_count", obis_reference="VOLTAGE_SWELL_L3_COUNT", - dsmr_versions={"2.2", "4", "5", "5L"}, + dsmr_versions={"2.2", "4", "5", "5L", "MSn"}, entity_registry_enabled_default=False, state_class=SensorStateClass.TOTAL_INCREASING, entity_category=EntityCategory.DIAGNOSTIC, @@ -386,7 +432,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="electricity_imported_total", translation_key="electricity_imported_total", obis_reference="ELECTRICITY_IMPORTED_TOTAL", - dsmr_versions={"5L", "5S", "Q3D", "5EONHU"}, + dsmr_versions={"5L", "5S", "Q3D", "5EONHU", "MSn", "SAGEMCOM_T210_D_R"}, device_class=SensorDeviceClass.ENERGY, state_class=SensorStateClass.TOTAL_INCREASING, ), @@ -394,7 +440,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="electricity_exported_total", translation_key="electricity_exported_total", obis_reference="ELECTRICITY_EXPORTED_TOTAL", - dsmr_versions={"5L", "5S", "Q3D", "5EONHU"}, + dsmr_versions={"5L", "5S", "Q3D", "5EONHU", "MSn", "SAGEMCOM_T210_D_R"}, device_class=SensorDeviceClass.ENERGY, state_class=SensorStateClass.TOTAL_INCREASING, ), @@ -418,7 +464,7 @@ SENSORS: tuple[DSMRSensorEntityDescription, ...] = ( key="hourly_gas_meter_reading", translation_key="gas_meter_reading", obis_reference="HOURLY_GAS_METER_READING", - dsmr_versions={"4", "5", "5L"}, + dsmr_versions={"4", "5", "5L", "MSn"}, is_gas=True, device_class=SensorDeviceClass.GAS, state_class=SensorStateClass.TOTAL_INCREASING, @@ -815,12 +861,17 @@ async def async_setup_entry( # create_dsmr_reader opens both local devices and any URL (socket://, # esphome://, ...); the only difference is the keep-alive watchdog. keep_alive = {} if port.startswith("/") else {"keep_alive_interval": 60} + # authentication_key=None decrypts without verifying the GCM + # authentication tag; an empty encryption key leaves plain telegrams + # untouched. reader_factory = partial( create_dsmr_reader, port, dsmr_version, update_entities_telegram, loop=hass.loop, + encryption_key=entry.data.get(CONF_ENCRYPTION_KEY, ""), + authentication_key=None, **keep_alive, ) diff --git a/homeassistant/components/dsmr/strings.json b/homeassistant/components/dsmr/strings.json index f8fe86fa4368..b54c5e21f584 100644 --- a/homeassistant/components/dsmr/strings.json +++ b/homeassistant/components/dsmr/strings.json @@ -8,9 +8,19 @@ "error": { "already_configured": "[%key:common::config_flow::abort::already_configured_device%]", "cannot_communicate": "Failed to communicate", - "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]" + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "invalid_key": "Failed to decrypt the telegram, check the encryption key" }, "step": { + "encryption_key": { + "data": { + "encryption_key": "Encryption key" + }, + "data_description": { + "encryption_key": "The encryption key of your meter, provided by your grid operator (32 hexadecimal characters)" + }, + "title": "Encryption key" + }, "user": { "data": { "dsmr_version": "Select DSMR version", diff --git a/tests/components/dsmr/conftest.py b/tests/components/dsmr/conftest.py index 6629f18032c8..ceae5772329a 100644 --- a/tests/components/dsmr/conftest.py +++ b/tests/components/dsmr/conftest.py @@ -106,7 +106,7 @@ def dsmr_connection_send_validate_fixture() -> Generator[ EQUIPMENT_IDENTIFIER_GAS, [{"value": "123456789", "unit": ""}] ), } - if args[1] == "5L": + if args[1] in ("5L", "MSn"): protocol.telegram = { LUXEMBOURG_EQUIPMENT_IDENTIFIER: CosemObject( LUXEMBOURG_EQUIPMENT_IDENTIFIER, [{"value": "12345678", "unit": ""}] @@ -121,7 +121,7 @@ def dsmr_connection_send_validate_fixture() -> Generator[ LUXEMBOURG_EQUIPMENT_IDENTIFIER, [{"value": "12345678", "unit": ""}] ), } - if args[1] == "5S": + if args[1] in ("5S", "SAGEMCOM_T210_D_R"): protocol.telegram = { P1_MESSAGE_TIMESTAMP: CosemObject( P1_MESSAGE_TIMESTAMP, [{"value": "12345678", "unit": ""}] diff --git a/tests/components/dsmr/snapshots/test_diagnostics.ambr b/tests/components/dsmr/snapshots/test_diagnostics.ambr index ec2dc274efab..b845864a597b 100644 --- a/tests/components/dsmr/snapshots/test_diagnostics.ambr +++ b/tests/components/dsmr/snapshots/test_diagnostics.ambr @@ -19,6 +19,7 @@ 'entry': dict({ 'data': dict({ 'dsmr_version': '2.2', + 'encryption_key': '**REDACTED**', 'port': '/dev/ttyUSB0', 'serial_id': '1234', 'serial_id_gas': '5678', diff --git a/tests/components/dsmr/test_config_flow.py b/tests/components/dsmr/test_config_flow.py index 222e24971c2a..4b5d323b03ad 100644 --- a/tests/components/dsmr/test_config_flow.py +++ b/tests/components/dsmr/test_config_flow.py @@ -4,9 +4,11 @@ from itertools import chain, repeat from typing import Any from unittest.mock import DEFAULT, AsyncMock, MagicMock, patch +from dsmr_parser.exceptions import DecryptionError import pytest from homeassistant import config_entries +from homeassistant.components.dsmr.config_flow import CannotCommunicate from homeassistant.components.dsmr.const import DOMAIN from homeassistant.components.usb import SerialDevice from homeassistant.core import HomeAssistant @@ -199,6 +201,172 @@ async def test_setup_serial( assert result["data"] == entry_data +@pytest.mark.parametrize( + ("version", "serial_data"), + [ + ("MSn", SERIAL_DATA), + ("SAGEMCOM_T210_D_R", SERIAL_DATA_SWEDEN), + ], +) +async def test_setup_serial_encrypted( + hass: HomeAssistant, + dsmr_connection_send_validate_fixture: tuple[MagicMock, MagicMock, MagicMock], + version: str, + serial_data: dict[str, str | None], +) -> None: + """Test we can setup an encrypted meter that asks for an encryption key.""" + (connection_factory, _transport, _protocol) = dsmr_connection_send_validate_fixture + port = com_port() + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"port": port.device, "dsmr_version": version}, + ) + + # An encrypted version asks for the encryption key in a second step + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "encryption_key" + + with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"encryption_key": "aabbccddeeff00112233445566778899"}, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == port.device + assert result["data"] == { + "port": port.device, + "dsmr_version": version, + "protocol": "dsmr_protocol", + "encryption_key": "aabbccddeeff00112233445566778899", + **serial_data, + } + # The key is decrypted without verifying the GCM authentication tag + assert ( + connection_factory.call_args.kwargs["encryption_key"] + == "aabbccddeeff00112233445566778899" + ) + assert connection_factory.call_args.kwargs["authentication_key"] is None + + +async def test_setup_serial_encrypted_invalid_key( + hass: HomeAssistant, + dsmr_connection_send_validate_fixture: tuple[MagicMock, MagicMock, MagicMock], +) -> None: + """Test an encrypted meter with a wrong encryption key reports an error.""" + (_connection_factory, _transport, protocol) = dsmr_connection_send_validate_fixture + port = com_port() + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"port": port.device, "dsmr_version": "MSn"}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "encryption_key" + + # A wrong key makes the protocol report a decryption error + protocol.decryption_error = DecryptionError("wrong key") + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"encryption_key": "00000000000000000000000000000000"}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "encryption_key" + assert result["errors"] == {"base": "invalid_key"} + + +@pytest.mark.parametrize( + "encryption_key", + [ + "tooshort", + "nothexnothexnothexnothexnothexgg", + "aabbccddeeff00112233445566778899ff", + ], + ids=["too_short", "non_hex", "too_long"], +) +async def test_setup_serial_encrypted_malformed_key( + hass: HomeAssistant, + dsmr_connection_send_validate_fixture: tuple[MagicMock, MagicMock, MagicMock], + encryption_key: str, +) -> None: + """Test a malformed encryption key is rejected without a connection attempt.""" + (connection_factory, _transport, _protocol) = dsmr_connection_send_validate_fixture + port = com_port() + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"port": port.device, "dsmr_version": "MSn"}, + ) + + assert result["step_id"] == "encryption_key" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"encryption_key": encryption_key}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "encryption_key" + assert result["errors"] == {"base": "invalid_key"} + # A malformed key must not reach the reader + connection_factory.assert_not_called() + + +@pytest.mark.usefixtures("dsmr_connection_send_validate_fixture") +async def test_setup_serial_encrypted_cannot_communicate( + hass: HomeAssistant, +) -> None: + """Test an encrypted meter does not fall back to RFXtrx when it stays silent.""" + port = com_port() + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"port": port.device, "dsmr_version": "MSn"}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "encryption_key" + + with patch( + "homeassistant.components.dsmr.config_flow._validate_dsmr_connection", + side_effect=CannotCommunicate, + ) as validate: + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"encryption_key": "aabbccddeeff00112233445566778899"}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "encryption_key" + assert result["errors"] == {"base": "cannot_communicate"} + # Encrypted meters must not retry over the RFXtrx protocol + assert validate.call_count == 1 + + async def test_setup_serial_rfxtrx( hass: HomeAssistant, dsmr_connection_send_validate_fixture: tuple[MagicMock, MagicMock, MagicMock], diff --git a/tests/components/dsmr/test_diagnostics.py b/tests/components/dsmr/test_diagnostics.py index f2a475097ae6..e6af3dfd52f3 100644 --- a/tests/components/dsmr/test_diagnostics.py +++ b/tests/components/dsmr/test_diagnostics.py @@ -33,6 +33,7 @@ async def test_diagnostics( "dsmr_version": "2.2", "serial_id": "1234", "serial_id_gas": "5678", + "encryption_key": "aabbccddeeff00112233445566778899", } entry_options = { "time_between_update": 0, diff --git a/tests/components/dsmr/test_sensor.py b/tests/components/dsmr/test_sensor.py index 60792ace9f23..474d3d1ab0b5 100644 --- a/tests/components/dsmr/test_sensor.py +++ b/tests/components/dsmr/test_sensor.py @@ -512,6 +512,171 @@ async def test_luxembourg_meter( ) +async def test_luxembourg_smarty_encrypted_meter( + hass: HomeAssistant, dsmr_connection_fixture: tuple[MagicMock, MagicMock, MagicMock] +) -> None: + """Test if an encrypted Luxembourg Smarty (MSn) meter is correctly parsed.""" + (connection_factory, _transport, _protocol) = dsmr_connection_fixture + + entry_data = { + "port": "/dev/ttyUSB0", + "dsmr_version": "MSn", + "serial_id": "1234", + "serial_id_gas": "5678", + "encryption_key": "aabbccddeeff00112233445566778899", + } + entry_options = { + "time_between_update": 0, + } + + telegram = Telegram() + telegram.add( + HOURLY_GAS_METER_READING, + MBusObject( + (0, 0), + [ + {"value": datetime.datetime.fromtimestamp(1551642213)}, + {"value": Decimal("745.695"), "unit": "m3"}, + ], + ), + "HOURLY_GAS_METER_READING", + ) + telegram.add( + ELECTRICITY_IMPORTED_TOTAL, + CosemObject( + (0, 0), + [{"value": Decimal("123.456"), "unit": UnitOfEnergy.KILO_WATT_HOUR}], + ), + "ELECTRICITY_IMPORTED_TOTAL", + ) + telegram.add( + ELECTRICITY_EXPORTED_TOTAL, + CosemObject( + (0, 0), + [{"value": Decimal("654.321"), "unit": UnitOfEnergy.KILO_WATT_HOUR}], + ), + "ELECTRICITY_EXPORTED_TOTAL", + ) + + mock_entry = MockConfigEntry( + domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options + ) + + mock_entry.add_to_hass(hass) + + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + + # The key is decrypted without verifying the GCM authentication tag + assert ( + connection_factory.call_args.kwargs["encryption_key"] + == "aabbccddeeff00112233445566778899" + ) + assert connection_factory.call_args.kwargs["authentication_key"] is None + + telegram_callback = connection_factory.call_args_list[0][0][2] + + # simulate a telegram pushed from the smartmeter and parsed by dsmr_parser + telegram_callback(telegram) + + # after receiving telegram entities need to have the chance to be created + await hass.async_block_till_done() + + consumption = hass.states.get("sensor.electricity_meter_energy_consumption_total") + assert consumption.state == "123.456" + assert consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENERGY + assert ( + consumption.attributes.get(ATTR_UNIT_OF_MEASUREMENT) + == UnitOfEnergy.KILO_WATT_HOUR + ) + + production = hass.states.get("sensor.electricity_meter_energy_production_total") + assert production.state == "654.321" + + gas_consumption = hass.states.get("sensor.gas_meter_gas_consumption") + assert gas_consumption.state == "745.695" + assert gas_consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.GAS + + +async def test_austrian_sagemcom_encrypted_meter( + hass: HomeAssistant, dsmr_connection_fixture: tuple[MagicMock, MagicMock, MagicMock] +) -> None: + """Test if an encrypted Austrian Sagemcom (T210-D-R) meter is correctly parsed.""" + (connection_factory, _transport, _protocol) = dsmr_connection_fixture + + entry_data = { + "port": "/dev/ttyUSB0", + "dsmr_version": "SAGEMCOM_T210_D_R", + "serial_id": None, + "serial_id_gas": None, + "encryption_key": "aabbccddeeff00112233445566778899", + } + entry_options = { + "time_between_update": 0, + } + + telegram = Telegram() + telegram.add( + ELECTRICITY_IMPORTED_TOTAL, + CosemObject( + (0, 0), + [{"value": Decimal("123.456"), "unit": UnitOfEnergy.KILO_WATT_HOUR}], + ), + "ELECTRICITY_IMPORTED_TOTAL", + ) + telegram.add( + ELECTRICITY_EXPORTED_TOTAL, + CosemObject( + (0, 0), + [{"value": Decimal("654.321"), "unit": UnitOfEnergy.KILO_WATT_HOUR}], + ), + "ELECTRICITY_EXPORTED_TOTAL", + ) + telegram.add( + obis_references.ELECTRICITY_USED_TARIFF_1, + CosemObject( + (0, 0), + [{"value": Decimal("11.111"), "unit": UnitOfEnergy.KILO_WATT_HOUR}], + ), + "ELECTRICITY_USED_TARIFF_1", + ) + + mock_entry = MockConfigEntry( + domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data, options=entry_options + ) + + mock_entry.add_to_hass(hass) + + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + + # The key is decrypted without verifying the GCM authentication tag + assert ( + connection_factory.call_args.kwargs["encryption_key"] + == "aabbccddeeff00112233445566778899" + ) + assert connection_factory.call_args.kwargs["authentication_key"] is None + + telegram_callback = connection_factory.call_args_list[0][0][2] + + # simulate a telegram pushed from the smartmeter and parsed by dsmr_parser + telegram_callback(telegram) + + # after receiving telegram entities need to have the chance to be created + await hass.async_block_till_done() + + consumption = hass.states.get("sensor.electricity_meter_energy_consumption_total") + assert consumption.state == "123.456" + assert consumption.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENERGY + + production = hass.states.get("sensor.electricity_meter_energy_production_total") + assert production.state == "654.321" + + tariff_1 = hass.states.get("sensor.electricity_meter_energy_consumption_tarif_1") + assert tariff_1.state == "11.111" + assert tariff_1.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.ENERGY + + async def test_eonhu_meter( hass: HomeAssistant, dsmr_connection_fixture: tuple[MagicMock, MagicMock, MagicMock] ) -> None: