Add data_secure flag to Telegram (#1799)

* Add data_secure flag to Telegram

* Update telegram.py

* Update telegram.py

Author: farmio

docs/changelog.md

@@ -8,6 +8,10 @@ nav_order: 2
 
 # Unreleased
 
+### Telegram
+
+- Add `data_secure` flag to Telegram to indicate if it was sent or received as Data Secure.
+
 ### Devices
 
 - ExposeSensor: `cooldown` is extended to wait for connection if not established.

test/cemi_tests/cemi_handler_test.py

@@ -45,6 +45,7 @@ async def test_wait_for_l2_confirmation(time_travel: EventLoopClockAdvancer) ->
     await task
     assert xknx.connection_manager.cemi_count_outgoing == 1
     assert xknx.connection_manager.cemi_count_outgoing_error == 0
+    assert test_telegram.data_secure is False
 
     # no L_DATA.con received -> raise ConfirmationError
     xknx.knxip_interface.send_cemi.reset_mock()

test/remote_value_tests/remote_value_test.py

@@ -275,8 +275,8 @@ def test_pre_decoded_telegram(self) -> None:
         telegram = Telegram(
             destination_address=GroupAddress("1/1/1"),
             payload=GroupValueWrite(test_payload),
-            decoded_data=TelegramDecodedData(transcoder=DPT2ByteFloat, value=3.3),
         )
+        telegram.decoded_data = TelegramDecodedData(transcoder=DPT2ByteFloat, value=3.3)
         assert remote_value.process(telegram)
         assert remote_value.value == 3.3
 

test/telegram_tests/telegram_test.py

@@ -1,7 +1,13 @@
 """Unit test for Telegram objects."""
 
-from xknx.dpt import DPTBinary
-from xknx.telegram import GroupAddress, IndividualAddress, Telegram, TelegramDirection
+from xknx.dpt import DPTBinary, DPTSwitch
+from xknx.telegram import (
+    GroupAddress,
+    IndividualAddress,
+    Telegram,
+    TelegramDecodedData,
+    TelegramDirection,
+)
 from xknx.telegram.apci import GroupValueRead, GroupValueWrite
 from xknx.telegram.tpci import TConnect, TDisconnect
 
@@ -14,9 +20,14 @@ class TestTelegram:
     #
     def test_telegram_equal(self) -> None:
         """Test equals operator."""
-        assert Telegram(GroupAddress("1/2/3"), payload=GroupValueRead()) == Telegram(
-            GroupAddress("1/2/3"), payload=GroupValueRead()
-        )
+        test = Telegram(GroupAddress("1/2/3"), payload=GroupValueRead())
+        telegram_1 = Telegram(GroupAddress("1/2/3"), payload=GroupValueRead())
+        assert test == telegram_1
+        # decoded_data and data_secure should not be considered for equality (although
+        # decoded_data doesn't make sense for a GroupValueRead, this is just for testing the equality operator)
+        telegram_1.decoded_data = TelegramDecodedData(transcoder=DPTSwitch, value=False)
+        telegram_1.data_secure = True
+        assert test == telegram_1
 
     def test_telegram_not_equal(self) -> None:
         """Test not equals operator."""

xknx/cemi/cemi_handler.py

@@ -72,6 +72,10 @@ async def send_telegram(self, telegram: Telegram) -> None:
         logger.debug("Outgoing CEMI: %s", cemi)
         if self.data_secure is not None:
             cemi.data = self.data_secure.outgoing_cemi(cemi_data=cemi_data)
+            telegram.data_secure = is_data_secure(cemi.data)
+        else:
+            telegram.data_secure = False
+
         self._l_data_confirmation_event.clear()
         try:
             await self.xknx.knxip_interface.send_cemi(cemi)
@@ -123,13 +127,14 @@ def handle_cemi_frame(self, cemi: CEMIFrame) -> None:
         logger.debug("Incoming CEMI: %s", cemi)
         self.xknx.connection_manager.cemi_count_incoming += 1
 
+        _cemi_data_is_data_secure = is_data_secure(cemi.data)
         if self.data_secure is None:
-            if is_data_secure(cemi.data):
+            if _cemi_data_is_data_secure:
                 data_secure_logger.debug(
                     "Received DataSecure encrypted CEMI frame but no keys for DataSecure are initialized: %s",
                     cemi,
                 )
-                self.handle_data_secure_key_issue(cemi.data)
+                self.handle_data_secure_key_issue(cemi.data, _cemi_data_is_data_secure)
                 return
         else:
             try:
@@ -140,11 +145,12 @@ def handle_cemi_frame(self, cemi: CEMIFrame) -> None:
                     "Could not decrypt CEMI frame: %s",
                     err,
                 )
-                self.handle_data_secure_key_issue(cemi.data)
+                self.handle_data_secure_key_issue(cemi.data, _cemi_data_is_data_secure)
                 return
 
         telegram = cemi.data.telegram()
         telegram.direction = TelegramDirection.INCOMING
+        telegram.data_secure = _cemi_data_is_data_secure
         self.telegram_received(telegram)
 
     def telegram_received(self, telegram: Telegram) -> None:
@@ -159,10 +165,13 @@ def telegram_received(self, telegram: Telegram) -> None:
             return
         self.xknx.management.process(telegram)
 
-    def handle_data_secure_key_issue(self, cemi_data: CEMILData) -> None:
+    def handle_data_secure_key_issue(
+        self, cemi_data: CEMILData, received_data_secure: bool
+    ) -> None:
         """Handle DataSecure telegrams with missing or invalid keys."""
         self.xknx.connection_manager.undecoded_data_secure += 1
         if isinstance(cemi_data.tpci, tpci.TDataGroup):
             telegram = cemi_data.telegram()
             telegram.direction = TelegramDirection.INCOMING
+            telegram.data_secure = received_data_secure
             self.xknx.telegram_queue.received_data_secure_group_key_issue(telegram)

xknx/core/telegram_queue.py

@@ -276,7 +276,8 @@ def received_data_secure_group_key_issue(self, telegram: Telegram) -> None:
         """
         Run registered callbacks for data secure group key issues.
 
-        Only TDataGroup telegrams with undecodable DataSecure payloads are forwarded.
+        TDataGroup telegrams with undecodable DataSecure payloads
+        or unexpected plain payloads are forwarded.
         """
         for data_secure_group_key_issue_cb in self._data_secure_group_key_issue_cbs:
             try:

xknx/telegram/telegram.py

@@ -1,19 +1,4 @@
-"""
-Module for KNX Telegrams.
-
-The telegram class is the lightweight data transfer object between
-
-* business logic (Lights, Covers, etc) and
-* underlying KNX/IP abstraction (CEMIHandler).
-
-It contains
-
-* the group address (e.g. GroupAddress("1/2/3"))
-* the direction (Incoming or Outgoing)
-* and the payload (e.g. GroupValueWrite(DPTBinary(False)))
-* the source address (e.g. IndividualAddress("1.2.3"))
-* the TPCI (Transport Layer Control Information) (e.g. TDataGroup())
-"""
+"""Module for KNX Telegrams."""
 
 from __future__ import annotations
 
@@ -51,16 +36,46 @@ def __str__(self) -> str:
 
 @dataclass(slots=True)
 class Telegram:
-    """Class for KNX telegrams."""
+    """
+    Data transfer object for KNX telegrams.
+
+    Represents a message exchanged on the KNX bus between the business logic
+    (Devices, Management, etc.) and the underlying KNX/IP abstraction layer.
+
+    Attributes:
+        destination_address: Target GroupAddress, IndividualAddress, or
+            InternalGroupAddress.
+        direction: Communication direction (INCOMING or OUTGOING).
+        payload: APCi payload containing the actual data (e.g., GroupValueWrite,
+            GroupValueResponse). None for control information only telegrams.
+        source_address: IndividualAddress of the sender. When default of 0.0.0 is
+            used, it will be set automatically when sent.
+        tpci: Transport Layer Control Information (TDataBroadcast, TDataGroup, or
+            TDataIndividual). If not provided, it will be automatically inferred
+            based on destination_address type.
+        decoded_data: Optional decoded version of the payload including the
+            transcoder class and decoded value. Set externally by GroupAddressDPT
+            for convenience when the payload has already been decoded.
+        data_secure: Flag indicating if the telegram was sent or received as
+            DataSecure. Set externally by CEMIHandler. None if not yet processed.
+
+    """
 
     destination_address: GroupAddress | IndividualAddress | InternalGroupAddress
     direction: TelegramDirection = TelegramDirection.OUTGOING
     payload: APCI | None = None
     source_address: IndividualAddress = field(
         default_factory=lambda: IndividualAddress(0)
     )
-    tpci: TPCI = None  # type: ignore[assignment]  # set in __post_init__
-    decoded_data: TelegramDecodedData | None = None
+    tpci: TPCI = None  # type: ignore[assignment]  # set by initializer or in __post_init__
+    # set by GroupAddressDPT
+    decoded_data: TelegramDecodedData | None = field(
+        init=False, default=None, compare=False, hash=False
+    )
+    # flag if telegram was sent or received as DataSecure, set by CEMIHandler
+    data_secure: bool | None = field(
+        init=False, default=None, compare=False, hash=False
+    )
 
     def __post_init__(self) -> None:
         """Initialize Telegram class."""
@@ -75,17 +90,6 @@ def __post_init__(self) -> None:
             else:  # InternalGroupAddress
                 self.tpci = TDataGroup()
 
-    def __eq__(self, other: object) -> bool:
-        """Equal operator. Omit decoded_data for comparison."""
-        return (
-            isinstance(other, Telegram)
-            and self.destination_address == other.destination_address
-            and self.direction == other.direction
-            and self.payload == other.payload
-            and self.source_address == other.source_address
-            and self.tpci == other.tpci
-        )
-
     def __str__(self) -> str:
         """Return object as readable string."""
         data = f'payload="{self.payload}"' if self.payload else f'tpci="{self.tpci}"'