starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 10:13:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add encrypted DSMR support for Luxembourg and Austria (#176058)

Browse Source 
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Paulus Schoutsen
2026-07-26 23:06:24 +02:00
committed by GitHub
co-authored by Claude
parent 5156365756
commit bce0aca97d
10 changed files with 529 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/dsmr/config_flow.py
+86 -6
View File
@@ -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."""
homeassistant/components/dsmr/const.py
+23 -1
View File
@@ -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"
homeassistant/components/dsmr/diagnostics.py
+5 -3
View File
@@ -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())
homeassistant/components/dsmr/sensor.py
+67 -16
View File
@@ -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,
)
homeassistant/components/dsmr/strings.json
+11 -1
View File
@@ -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",
tests/components/dsmr/conftest.py
+2 -2
View File
@@ -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": ""}]
tests/components/dsmr/snapshots/test_diagnostics.ambr
+1
View File
@@ -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',
tests/components/dsmr/test_config_flow.py
+168
View File
@@ -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],
tests/components/dsmr/test_diagnostics.py
+1
View File
@@ -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,
tests/components/dsmr/test_sensor.py
+165
View File
@@ -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: