Repository: XKNX/xknx
Pull request #1167 (2023-02-01 22:13:40)

Data Secure support for group communication
<!--
  You are awesome! Thanks for contributing to our project!
  Please, DO NOT DELETE ANY TEXT from this template!.
-->
## Description
<!--
Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context.
List any dependencies that are required for this change.
-->
Keys for Data Secure are read form a `.knxkeys` keyring file that has to be passed to a `SecureConfig` instance.

Fixes # (issue)

## Type of change
<!--
Please tick the applicable options.
NOTE: Ticking multiple options most likely indicates
that your change is to big and it is suggested to split it into several smaller PRs.
-->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## Checklist:

- [x] The documentation has been adjusted accordingly
- [x] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog (docs/changelog.md)

codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#1167](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (3435a8b) into [main](https://codecov.io/gh/XKNX/xknx/commit/02d3a83713045d79c83ba861aa120a4b42c7718d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (02d3a83) will **decrease** coverage by `0.30%`.
> The diff coverage is `84.84%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/1167/graphs/tree.svg?width=650&height=150&src=pr&token=irWbIygS84&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main    #1167      +/-   ##
==========================================
- Coverage   96.32%   96.02%   -0.30%     
==========================================
  Files         141      143       +2     
  Lines        8703     8916     +213     
==========================================
+ Hits         8383     8562     +179     
- Misses        320      354      +34     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/exceptions/\_\_init\_\_.py](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9leGNlcHRpb25zL19faW5pdF9fLnB5) | `100.00% <ø> (ø)` | |
| [xknx/secure/security\_primitives.py](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9zZWN1cmUvc2VjdXJpdHlfcHJpbWl0aXZlcy5weQ==) | `100.00% <ø> (ø)` | |
| [xknx/telegram/apci.py](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC90ZWxlZ3JhbS9hcGNpLnB5) | `98.36% <78.94%> (-0.52%)` | :arrow_down: |
| [xknx/secure/data\_secure.py](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9zZWN1cmUvZGF0YV9zZWN1cmUucHk=) | `81.91% <81.91%> (ø)` | |
| [xknx/secure/data\_secure\_asdu.py](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9zZWN1cmUvZGF0YV9zZWN1cmVfYXNkdS5weQ==) | `84.41% <84.41%> (ø)` | |
| [xknx/cemi/cemi\_handler.py](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9jZW1pL2NlbWlfaGFuZGxlci5weQ==) | `96.49% <88.23%> (-3.51%)` | :arrow_down: |
| [xknx/exceptions/exception.py](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9leGNlcHRpb25zL2V4Y2VwdGlvbi5weQ==) | `100.00% <100.00%> (ø)` | |
| [xknx/io/ip\_secure.py](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9pcF9zZWN1cmUucHk=) | `95.44% <100.00%> (ø)` | |
| [xknx/io/knxip\_interface.py](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9rbnhpcF9pbnRlcmZhY2UucHk=) | `85.92% <100.00%> (+0.70%)` | :arrow_up: |
| [xknx/secure/keyring.py](https://codecov.io/gh/XKNX/xknx/pull/1167?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9zZWN1cmUva2V5cmluZy5weQ==) | `99.12% <100.00%> (+0.01%)` | :arrow_up: |


Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index d9314813..db688a32 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -8,11 +8,16 @@ nav_order: 2
 
 # Unreleased changes
 
+### Data Secure
+
+- Support KNX Data Secure for group communication. Keys are read from an ETS keyring export.
+
 ### Internals
 
 - Return `bytes` from `BaseAddress.to_knx()` instead of `tuple[int, int]`. This is used in `IndividualAddress` and `GroupAddress`.
 - Add `BaseAddress.from_knx()` to instantiate from `bytes`, remove instantiation form `tuple[int, int]`.
 - Refactor APCI to return complete Subclass `APCI.from_knx()` and removed `APCI.resolve_apci()`.
+- Rename `xknx.secure.ip_secure` to `xknx.secure.security_primitives`.
 - Fix wrong string length in keyfile signature verification for multi-byte UTF-8 encoded attribute values.
 
 # 2.3.0 Routing security, DPTs and CEMI-Refactoring 2023-01-10
diff --git a/docs/index.md b/docs/index.md
index abe8876f..2075abc5 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -16,12 +16,13 @@ XKNX...
 
 - ... does [cooperative multitasking via asyncio](https://github.com/XKNX/xknx/blob/main/examples/example_light_state.py) and is 100% thread safe.
 - ... provides support for KNX/IP [routing](https://github.com/XKNX/xknx/blob/main/xknx/io/routing.py) _and_ [tunneling](https://github.com/XKNX/xknx/blob/main/xknx/io/tunnel.py) devices.
+- ... supports KNX IP Secure - via tunneling or routing.
+- ... supports KNX Data Secure group communication.
 - ... has strong coverage with [unit tests](https://github.com/XKNX/xknx/tree/main/test).
 - ... automatically updates and synchronizes all devices in the background periodically.
 - ... listens for all updates of all devices on the KNX bus and updates the corresponding internal objects.
 - ... has a clear abstraction of data/network/logic-layer.
-- ... provides Heartbeat monitoring for Tunneling connections + clean reconnect if KNX/IP connection failed.
-- ... does clean [connect](https://github.com/XKNX/xknx/blob/main/xknx/io/connect.py) and [disconnect](https://github.com/XKNX/xknx/blob/main/xknx/io/disconnect.py) requests to the tunneling device.
+- ... does clean [connect](https://github.com/XKNX/xknx/blob/main/xknx/io/connect.py) and [disconnect](https://github.com/XKNX/xknx/blob/main/xknx/io/disconnect.py) requests to the tunneling device and reconnects if KNX/IP connection failed.
 - ... ships with [Home Assistant](https://home-assistant.io/).
 
 ## [](#header-2)Installation
diff --git a/xknx/cemi/cemi_handler.py b/xknx/cemi/cemi_handler.py
index 2a2f3240..71ca7623 100644
--- a/xknx/cemi/cemi_handler.py
+++ b/xknx/cemi/cemi_handler.py
@@ -11,7 +11,14 @@ import asyncio
 import logging
 from typing import TYPE_CHECKING
 
-from xknx.exceptions import CommunicationError, ConfirmationError, ConversionError
+from xknx.exceptions import (
+    CommunicationError,
+    ConfirmationError,
+    ConversionError,
+    DataSecureError,
+)
+from xknx.secure.data_secure import DataSecure
+from xknx.secure.keyring import Keyring
 from xknx.telegram import IndividualAddress, Telegram, TelegramDirection, tpci
 
 from .cemi_frame import CEMIFrame
@@ -21,6 +28,7 @@ if TYPE_CHECKING:
     from xknx.xknx import XKNX
 
 logger = logging.getLogger("xknx.cemi")
+data_secure_logger = logging.getLogger("xknx.data_secure")
 
 # See 3/6/3 EMI_IMI §4.1.5 Data Link Layer messages
 REQUEST_TO_CONFIRMATION_TIMEOUT = 3
@@ -32,8 +40,16 @@ class CEMIHandler:
     def __init__(self, xknx: XKNX) -> None:
         """Initialize CEMIHandler class."""
         self.xknx = xknx
+        self.data_secure: DataSecure | None = None
         self._l_data_confirmation_event = asyncio.Event()
 
+    def data_secure_init(self, keyring: Keyring | None) -> None:
+        """Initialize DataSecure."""
+        if keyring is None:
+            self.data_secure = None
+        else:
+            self.data_secure = DataSecure.init_from_keyring(keyring)
+
     async def send_telegram(self, telegram: Telegram) -> None:
         """Create a CEMIFrame from a Telegram and send it to the CEMI Server."""
         cemi = CEMIFrame.init_from_telegram(
@@ -43,8 +59,11 @@ class CEMIHandler:
                 self.xknx.current_address if telegram.source_address.raw == 0 else None
             ),
         )
-        self._l_data_confirmation_event.clear()
         logger.debug("Outgoing CEMI: %s", cemi)
+        if self.data_secure is not None:
+            cemi = self.data_secure.outgoing_cemi(cemi=cemi)
+
+        self._l_data_confirmation_event.clear()
         try:
             await self.xknx.knxip_interface.send_cemi(cemi)
         except (ConversionError, CommunicationError) as ex:
@@ -74,6 +93,13 @@ class CEMIHandler:
             logger.warning("Received unexpected L_DATA_REQ frame: %s", cemi)
             return
 
+        # TODO: do we have to decrypt Data Secure L_DATA_CON?
+        if self.data_secure is not None:
+            try:
+                cemi = self.data_secure.received_cemi(cemi=cemi)
+            except DataSecureError as err:
+                data_secure_logger.warning("Could not decrypt CEMI frame: %s", err)
+                return
         # TODO: remove telegram init from CEMIFrame class and move it here?
         telegram = cemi.telegram
         telegram.direction = TelegramDirection.INCOMING
diff --git a/xknx/exceptions/__init__.py b/xknx/exceptions/__init__.py
index 8ed15477..90e576cc 100644
--- a/xknx/exceptions/__init__.py
+++ b/xknx/exceptions/__init__.py
@@ -7,6 +7,7 @@ from .exception import (
     CouldNotParseAddress,
     CouldNotParseKNXIP,
     CouldNotParseTelegram,
+    DataSecureError,
     DeviceIllegalValue,
     IncompleteKNXIPFrame,
     InvalidSecureConfiguration,
@@ -27,6 +28,7 @@ __all__ = [
     "CouldNotParseAddress",
     "CouldNotParseKNXIP",
     "CouldNotParseTelegram",
+    "DataSecureError",
     "DeviceIllegalValue",
     "ManagementConnectionError",
     "ManagementConnectionRefused",
diff --git a/xknx/exceptions/exception.py b/xknx/exceptions/exception.py
index e03f6887..87a09cb1 100644
--- a/xknx/exceptions/exception.py
+++ b/xknx/exceptions/exception.py
@@ -155,8 +155,12 @@ class DeviceIllegalValue(XKNXException):
         return f'<DeviceIllegalValue description="{self.value}" value="{self.description}" />'
 
 
+class DataSecureError(XKNXException):
+    """Exception class for KNX Data Secure handling."""
+
+
 class SecureException(XKNXException):
-    """Exception class for ip secure handling."""
+    """Exception class for IP secure handling."""
 
 
 class InvalidSecureConfiguration(SecureException):
diff --git a/xknx/io/ip_secure.py b/xknx/io/ip_secure.py
index d036732c..d4c97b72 100644
--- a/xknx/io/ip_secure.py
+++ b/xknx/io/ip_secure.py
@@ -28,7 +28,7 @@ from xknx.knxip import (
     TimerNotify,
 )
 from xknx.knxip.knxip_enum import SecureSessionStatusCode
-from xknx.secure.ip_secure import (
+from xknx.secure.security_primitives import (
     calculate_message_authentication_code_cbc,
     decrypt_ctr,
     derive_device_authentication_password,
diff --git a/xknx/io/knxip_interface.py b/xknx/io/knxip_interface.py
index 0ada5357..ee06fbc6 100644
--- a/xknx/io/knxip_interface.py
+++ b/xknx/io/knxip_interface.py
@@ -21,10 +21,10 @@ from xknx.exceptions import (
     XKNXException,
 )
 from xknx.io import util
-from xknx.secure.keyring import XMLInterface, load_keyring
+from xknx.secure.keyring import Keyring, XMLInterface, load_keyring
 
 from .connection import ConnectionConfig, ConnectionType
-from .const import DEFAULT_INDIVIDUAL_ADDRESS, DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
+from .const import DEFAULT_INDIVIDUAL_ADDRESS
 from .gateway_scanner import GatewayDescriptor, GatewayScanner
 from .routing import Routing, SecureRouting
 from .self_description import request_description
@@ -72,14 +72,22 @@ class KNXIPInterface:
 
     async def _start(self) -> None:
         """Start interface. Connecting KNX/IP device with the selected method."""
+        keyring: Keyring | None = None
+        if secure_config := self.connection_config.secure_config:
+            if (
+                secure_config.knxkeys_file_path is not None
+                and secure_config.knxkeys_password is not None
+            ):
+                keyring = await load_keyring(
+                    secure_config.knxkeys_file_path,
+                    secure_config.knxkeys_password,
+                )
+        self.xknx.cemi_handler.data_secure_init(keyring=keyring)
+
         if self.connection_config.connection_type == ConnectionType.ROUTING:
-            await self._start_routing(
-                local_ip=self.connection_config.local_ip,
-                multicast_group=self.connection_config.multicast_group,
-                multicast_port=self.connection_config.multicast_port,
-            )
+            await self._start_routing()
         elif self.connection_config.connection_type == ConnectionType.ROUTING_SECURE:
-            await self._start_secure_routing()
+            await self._start_secure_routing(keyring=keyring)
         elif (
             self.connection_config.connection_type == ConnectionType.TUNNELING
             and self.connection_config.gateway_ip is not None
@@ -104,6 +112,7 @@ class KNXIPInterface:
             await self._start_secure_tunnelling_tcp(
                 gateway_ip=self.connection_config.gateway_ip,
                 gateway_port=self.connection_config.gateway_port,
+                keyring=keyring,
             )
         else:
             await self._start_automatic()
@@ -137,9 +146,7 @@ class KNXIPInterface:
                         gateway_port=gateway.port,
                     )
                 elif gateway.supports_routing and not gateway.routing_requires_secure:
-                    await self._start_routing(
-                        local_ip=self.connection_config.local_ip,
-                    )
+                    await self._start_routing()
             except CommunicationError as ex:
                 logger.debug("Could not connect to %s: %s", gateway, ex)
                 continue
@@ -176,6 +183,7 @@ class KNXIPInterface:
         gateway_ip: str,
         gateway_port: int,
         gateway_descriptor: GatewayDescriptor | None = None,
+        keyring: Keyring | None = None,
     ) -> None:
         """Start KNX/IP Secure TCP tunnel."""
         if (secure_config := self.connection_config.secure_config) is None:
@@ -190,16 +198,12 @@ class KNXIPInterface:
             device_authentication_password = (
                 secure_config.device_authentication_password
             )
-        elif (
-            secure_config.knxkeys_file_path is not None
-            and secure_config.knxkeys_password is not None
-        ):
+        elif keyring is not None:
             _gateway = gateway_descriptor or await request_description(
                 gateway_ip=gateway_ip, gateway_port=gateway_port
             )
-            xml_interface = await self._get_tunnel_interface_from_keyfile(
-                keyfile_path=secure_config.knxkeys_file_path,
-                keyfile_password=secure_config.knxkeys_password,
+            xml_interface = self._get_tunnel_interface_from_keyring(
+                keyring=keyring,
                 gateway_descriptor=_gateway,
                 config_user_id=secure_config.user_id,
             )
@@ -234,22 +238,17 @@ class KNXIPInterface:
         )
         await self._interface.connect()
 
-    async def _get_tunnel_interface_from_keyfile(
+    def _get_tunnel_interface_from_keyring(
         self,
-        keyfile_path: str,
-        keyfile_password: str,
+        keyring: Keyring,
         gateway_descriptor: GatewayDescriptor,
         config_user_id: int | None = None,
     ) -> XMLInterface:
         """
-        Get tunnel interface from keyfile.
+        Get tunnel interface from keyring.
 
         Precedence: configured individual address > configured user id > first free tunnel interface
         """
-        keyring = await load_keyring(
-            keyfile_path,
-            keyfile_password,
-        )
         if _ia := self.connection_config.individual_address:
             if xml_interface := keyring.get_tunnel_interface_by_individual_address(_ia):
                 return xml_interface
@@ -329,14 +328,11 @@ class KNXIPInterface:
         )
         await self._interface.connect()
 
-    async def _start_routing(
-        self,
-        local_ip: str | None = None,
-        multicast_group: str = DEFAULT_MCAST_GRP,
-        multicast_port: int = DEFAULT_MCAST_PORT,
-    ) -> None:
+    async def _start_routing(self) -> None:
         """Start KNX/IP Routing."""
-        local_ip = local_ip or await util.get_default_local_ip()
+        multicast_group = self.connection_config.multicast_group
+        multicast_port = self.connection_config.multicast_port
+        local_ip = self.connection_config.local_ip or await util.get_default_local_ip()
         if local_ip is None:
             raise XKNXException("No network interface found.")
         util.validate_ip(local_ip, address_name="Local IP address")
@@ -363,10 +359,7 @@ class KNXIPInterface:
 
     async def _start_secure_routing(
         self,
-        latency_ms: int | None = None,
-        local_ip: str | None = None,
-        multicast_group: str = DEFAULT_MCAST_GRP,
-        multicast_port: int = DEFAULT_MCAST_PORT,
+        keyring: Keyring | None = None,
     ) -> None:
         """Start KNX/IP Routing."""
         if self.connection_config.secure_config is None:
@@ -375,18 +368,8 @@ class KNXIPInterface:
         latency_ms = self.connection_config.secure_config.latency_ms
         multicast_group = self.connection_config.multicast_group
         multicast_port = self.connection_config.multicast_port
-        if (
-            self.connection_config.secure_config.knxkeys_file_path is not None
-            and self.connection_config.secure_config.knxkeys_password is not None
-        ):
-            keyring = await load_keyring(
-                self.connection_config.secure_config.knxkeys_file_path,
-                self.connection_config.secure_config.knxkeys_password,
-            )
-            if keyring.backbone is None:
-                raise InvalidSecureConfiguration(
-                    "No backbone key found in knxkeys file"
-                )
+
+        if keyring is not None and keyring.backbone is not None:
             # default to manually configured values
             backbone_key = backbone_key or keyring.backbone.decrypted_key
             latency_ms = latency_ms or keyring.backbone.latency
diff --git a/xknx/secure/data_secure.py b/xknx/secure/data_secure.py
new file mode 100644
index 00000000..e95ea9bb
--- /dev/null
+++ b/xknx/secure/data_secure.py
@@ -0,0 +1,244 @@
+"""Module for KNX Data Secure."""
+from __future__ import annotations
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+from copy import copy
+from datetime import datetime, timezone
+import logging
+import time
+
+from xknx.cemi import CEMIFrame
+from xknx.exceptions import DataSecureError
+from xknx.telegram.address import GroupAddress, IndividualAddress
+from xknx.telegram.apci import APCI, SecureAPDU
+
+from .data_secure_asdu import (
+    SecureData,
+    SecurityAlgorithmIdentifier,
+    SecurityALService,
+    SecurityControlField,
+)
+from .keyring import Keyring
+
+_LOGGER = logging.getLogger("xknx.data_secure")
+
+# Same timedelta in milliseconds as used in Falcon used for initial sequence_number_sending
+# py3.10 backwards compatibility - py3.11 "2018-01-05T00:00:00Z" is supported
+_SEQUENCE_NUMBER_INIT_TIMESTAMP = datetime.fromisoformat(
+    "2018-01-05T00:00:00+00:00"
+).timestamp()
+
+
+def _initial_sequence_number() -> int:
+    """Return an initial sequence number for sending Data Secure Telegrams."""
+    return int((time.time() - _SEQUENCE_NUMBER_INIT_TIMESTAMP) * 1000)
+
+
+class DataSecure:
+    """Calss for KNX Data Secure handling."""
+
+    def __init__(
+        self,
+        *,
+        group_key_table: dict[GroupAddress, bytes],
+        individual_address_table: dict[IndividualAddress, int],
+        last_sequence_number_sending: int | None = None,
+    ):
+        """Initialize DataSecure class."""
+        self._group_key_table = group_key_table
+        self._individual_address_table = individual_address_table
+        self._sequence_number_sending = (
+            last_sequence_number_sending or _initial_sequence_number()
+        )
+        # Holds the last valid sequence number for each individual address.
+        # Use sequence_number from keyfile as initial value or 0 from senders for all IAs ?
+
+        if not 0 < self._sequence_number_sending < 0xFFFFFFFFFFFF:
+            _local_time_info = (
+                f" Local time not set properly? {datetime.now(timezone.utc).isoformat()}"
+                if not last_sequence_number_sending
+                else ""
+            )
+            raise DataSecureError(
+                f"Initial sequence number out of range: {self._sequence_number_sending}"
+                f"{_local_time_info}"
+            )
+        _LOGGER.info(
+            "Data Secure initialized for %s group addresses from %s individual addresses.",
+            len(self._group_key_table),
+            len(self._individual_address_table),
+        )
+        _LOGGER.debug(
+            "Data Secure initial sequence number: %s, groups: %s, senders: %s",
+            self._sequence_number_sending,
+            [str(ga) for ga in self._group_key_table],
+            [str(ia) for ia in self._individual_address_table],
+        )
+
+    @staticmethod
+    def init_from_keyring(keyring: Keyring) -> DataSecure | None:
+        """
+        Initialize DataSecure from Keyring.
+
+        Return None if no Data Secure information is found in the Keyring.
+        """
+        ga_key_table: dict[GroupAddress, bytes] = {}
+        ia_seq_table: dict[IndividualAddress, int] = {}
+
+        for xml_ga in keyring.group_addresses:
+            if xml_ga.decrypted_key is not None:
+                ga_key_table[xml_ga.address] = xml_ga.decrypted_key
+
+        for xml_ia in keyring.devices:
+            # TODO: check if this should default to 0 or if devices without a sequence number
+            #       in keyfile should be excluded from the table
+            ia_seq_table[xml_ia.individual_address] = xml_ia.sequence_number
+        # TODO: persist local individual_address_table and update from that file on start
+        #       to have more fresh initial sequence numbers
+
+        if not ga_key_table:
+            return None
+        return DataSecure(
+            group_key_table=ga_key_table,
+            individual_address_table=ia_seq_table,
+        )
+
+    def get_sequence_number(self) -> int:
+        """Return current sequence number sending and increment local stored value."""
+        seq_nr = self._sequence_number_sending
+        self._sequence_number_sending += 1
+        return seq_nr
+
+    @contextmanager
+    def check_sequence_number(
+        self, source_address: IndividualAddress, received_sequence_number: int
+    ) -> Iterator[None]:
+        """
+        Check the last valid sequence number for incoming frames from `source_address`.
+
+        Update the Security Individual Address Table if no further exception is raised.
+        Raise `DataSecureError` if sequence number is invalid or sender is not known.
+        """
+        try:
+            last_valid_sequence_number = self._individual_address_table[source_address]
+        except KeyError:
+            raise DataSecureError(
+                f"Source address not found in Security Individual Address Table: {source_address}"
+            )
+        if not received_sequence_number > last_valid_sequence_number:
+            # TODO: implement and increment Security Failure Log counter (not when equal)
+            raise DataSecureError(
+                f"Sequence number too low for {source_address}: "
+                f"{received_sequence_number} received, {last_valid_sequence_number} last valid"
+            )
+
+        yield
+        # Don't increment sequence number if exception is raised while decrypting (yield)
+        self._individual_address_table[source_address] = received_sequence_number
+
+    def received_cemi(self, cemi: CEMIFrame) -> CEMIFrame:
+        """Handle received CEMI frame."""
+        # Data Secure frame
+        if isinstance(cemi.payload, SecureAPDU):
+            return self._received_secure_cemi(cemi, cemi.payload)
+        # Plain group communication frame
+        if isinstance(cemi.dst_addr, GroupAddress):
+            if cemi.dst_addr in self._group_key_table:
+                raise DataSecureError(
+                    f"Discarding frame with plain APDU for secure group address: {cemi}"
+                )
+            return cemi
+        # Plain point-to-point frame
+        #   No point to point key table is implemented at the moment as ETS can't even configure this
+        #   only way to communicate point-to-point with data secure currently is with tool key
+        #   - which we don't have
+        return cemi
+
+    def _received_secure_cemi(self, cemi: CEMIFrame, s_apdu: SecureAPDU) -> CEMIFrame:
+        """Handle received secured CEMI frame."""
+        if s_apdu.scf.service is not SecurityALService.S_A_DATA:
+            raise DataSecureError(f"Only SecurityALService.S_A_DATA supported {cemi}")
+        if s_apdu.scf.system_broadcast or s_apdu.scf.tool_access:
+            # TODO: handle incoming responses with tool key of sending device
+            # when we can send with tool key
+            raise DataSecureError(
+                f"System broadcast and tool access not supported {cemi}"
+            )
+
+        # Secure group communication frame
+        if isinstance(cemi.dst_addr, GroupAddress):
+            if not (key := self._group_key_table.get(cemi.dst_addr)):
+                raise DataSecureError(
+                    f"No key found for group address {cemi.dst_addr} from {cemi.src_addr}"
+                )
+        # Secure point-to-point frame
+        else:
+            # TODO: maybe possible to implement this over tool key
+            raise DataSecureError(
+                f"Secure Point-to-Point communication not supported {cemi}"
+            )
+
+        with self.check_sequence_number(
+            source_address=cemi.src_addr,
+            received_sequence_number=int.from_bytes(
+                s_apdu.secured_data.sequence_number_bytes, "big"
+            ),
+        ):
+            _address_fields_raw = cemi.src_addr.to_knx() + cemi.dst_addr.to_knx()
+            plain_apdu_raw = s_apdu.secured_data.get_plain_apdu(
+                key=key,
+                scf=s_apdu.scf,
+                address_fields_raw=_address_fields_raw,
+                frame_flags=cemi.flags,
+                tpci=cemi.tpci,
+            )
+        decrypted_payload = APCI.from_knx(plain_apdu_raw)
+        _LOGGER.debug("Unpacked APDU %s from %s", decrypted_payload, s_apdu)
+        plain_cemi = copy(cemi)
+        plain_cemi.payload = decrypted_payload
+        return plain_cemi
+
+    def outgoing_cemi(self, cemi: CEMIFrame) -> CEMIFrame:
+        """Handle outgoing CEMI frame. Pass through as plain frame or encrypt."""
+        # Outgoing  group communication frame
+        if isinstance(cemi.dst_addr, GroupAddress):
+            if key := self._group_key_table.get(cemi.dst_addr):
+                scf = SecurityControlField(
+                    algorithm=SecurityAlgorithmIdentifier.CCM_ENCRYPTION,
+                    service=SecurityALService.S_A_DATA,
+                    system_broadcast=False,
+                    tool_access=False,
+                )
+                return self._secure_data_cemi(key=key, scf=scf, cemi=cemi)
+            return cemi
+        # Outgoing secure point-to-point frames are sent plain.
+        # Data Secure point-to-point is not supported.
+        return cemi
+
+    def _secure_data_cemi(
+        self,
+        key: bytes,
+        scf: SecurityControlField,
+        cemi: CEMIFrame,
+    ) -> CEMIFrame:
+        """Wrap encrypted payload of a plain CEMIFrame in a SecureAPDU."""
+        plain_apdu_raw: bytes | bytearray
+        if cemi.payload is not None:
+            plain_apdu_raw = cemi.payload.to_knx()
+        else:
+            # TODO: test if this is correct
+            plain_apdu_raw = b""  # used ein point-to-point eg. TConnect
+        secure_asdu = SecureData.init_from_plain_apdu(
+            key=key,
+            apdu=plain_apdu_raw,
+            scf=scf,
+            sequence_number=self.get_sequence_number(),
+            address_fields_raw=cemi.src_addr.to_knx() + cemi.dst_addr.to_knx(),
+            frame_flags=cemi.flags,
+            tpci=cemi.tpci,
+        )
+        secure_cemi = copy(cemi)
+        secure_cemi.payload = SecureAPDU(scf=scf, secured_data=secure_asdu)
+        _LOGGER.debug("Secured APDU %s with %s", cemi.payload, secure_cemi.payload)
+        return secure_cemi
diff --git a/xknx/secure/data_secure_asdu.py b/xknx/secure/data_secure_asdu.py
new file mode 100644
index 00000000..7e9fa8b8
--- /dev/null
+++ b/xknx/secure/data_secure_asdu.py
@@ -0,0 +1,276 @@
+"""Class for Data Secure Application layer service data units (ASDUs)."""
+from __future__ import annotations
+
+from enum import IntEnum
+
+from xknx.exceptions import DataSecureError
+from xknx.telegram.tpci import TPCI
+
+from .security_primitives import (
+    calculate_message_authentication_code_cbc,
+    decrypt_ctr,
+    encrypt_data_ctr,
+)
+
+# Secure APCI is 0x03F1 - in block_0 it is used split into 2 octets
+_APCI_SEC_HIGH = 0x03
+_APCI_SEC_LOW = 0xF1
+# only Address Type (IA / GA) and Extended Frame format are used
+B0_AT_FIELD_FLAGS_MASK = 0b10001111
+
+
+def block_0(
+    sequence_number: bytes,
+    address_fields_raw: bytes,
+    frame_flags: int,
+    tpci_int: int,
+    payload_length: int,
+) -> bytes:
+    """Return Block 0 for KNX Data Secure."""
+    return (
+        sequence_number
+        + address_fields_raw
+        + bytes(
+            (
+                0,
+                frame_flags & B0_AT_FIELD_FLAGS_MASK,
+                (tpci_int << 2) + _APCI_SEC_HIGH,
+                _APCI_SEC_LOW,
+                0,
+                payload_length,
+            )
+        )
+    )
+
+
+def counter_0(
+    sequence_number: bytes,
+    address_fields_raw: bytes,
+) -> bytes:
+    """Return Block Counter 0 for KNX Data Secure."""
+    return sequence_number + address_fields_raw + b"\x00\x00\x00\x00\x01\x00"
+
+
+class SecurityAlgorithmIdentifier(IntEnum):
+    """Enum representing the used security algorithm."""
+
+    CCM_AUTHENTICATION = 0b000
+    CCM_ENCRYPTION = 0b001
+
+
+class SecurityALService(IntEnum):
+    """Enum representing the used security application layer service."""
+
+    S_A_DATA = 0b000
+    S_A_SYNC_REQ = 0b001
+    S_A_SYNC_RES = 0b011
+
+
+class SecurityControlField:
+    """Class for KNX Data Secure Security Control Field (SCF)."""
+
+    def __init__(
+        self,
+        tool_access: bool,
+        algorithm: SecurityAlgorithmIdentifier,
+        system_broadcast: bool,
+        service: SecurityALService,
+    ) -> None:
+        """Initialize SecurityControlField class."""
+        self.tool_access = tool_access
+        self.algorithm = algorithm
+        self.system_broadcast = system_broadcast
+        self.service = service
+
+    @staticmethod
+    def from_knx(raw: int) -> SecurityControlField:
+        """Parse/deserialize from KNX raw data."""
+        tool_access = bool(raw & 0b10000000)
+        sai = SecurityAlgorithmIdentifier(raw >> 4 & 0b111)
+        system_broadcast = bool(raw & 0b1000)
+        s_al_service = SecurityALService(raw & 0b111)
+
+        return SecurityControlField(
+            tool_access=tool_access,
+            algorithm=sai,
+            system_broadcast=system_broadcast,
+            service=s_al_service,
+        )
+
+    def to_knx(self) -> bytes:
+        """Serialize to KNX raw data."""
+        raw = 0
+        raw |= self.tool_access << 7
+        raw |= self.algorithm << 4
+        raw |= self.system_broadcast << 3
+        raw |= self.service
+        return raw.to_bytes(1, "big")
+
+    def __str__(self) -> str:
+        """Return object as readable string."""
+        return (
+            f"<SecurityControlField tool_access={self.tool_access} "
+            f"algorithm={self.algorithm.name} "
+            f"system_broadcast={self.system_broadcast} "
+            f"service={self.service.name} />"
+        )
+
+
+class SecureData:
+    """Class for KNX Data Secure ASDU for S-A_Data-service."""
+
+    def __init__(
+        self,
+        sequence_number_bytes: bytes,
+        secured_apdu: bytes,
+        message_authentication_code: bytes,
+    ) -> None:
+        """Initialize SecureData class."""
+        self.sequence_number_bytes = sequence_number_bytes
+        self.secured_apdu = secured_apdu
+        self.message_authentication_code = message_authentication_code
+
+    def __len__(self) -> int:
+        """Return length of KNX Data Secure ASDU."""
+        return 10 + len(self.secured_apdu)  # 10 = 6 bytes sequence number + 4 bytes MAC
+
+    @staticmethod
+    def init_from_plain_apdu(
+        key: bytes,
+        apdu: bytes,
+        scf: SecurityControlField,
+        sequence_number: int,
+        address_fields_raw: bytes,
+        frame_flags: int,
+        tpci: TPCI,
+    ) -> SecureData:
+        """Serialize to KNX raw data."""
+        sequence_number_bytes = sequence_number.to_bytes(6, "big")
+
+        if scf.algorithm == SecurityAlgorithmIdentifier.CCM_AUTHENTICATION:
+            mac = calculate_message_authentication_code_cbc(
+                key=key,
+                additional_data=scf.to_knx() + apdu,
+                block_0=block_0(
+                    sequence_number=sequence_number_bytes,
+                    address_fields_raw=address_fields_raw,
+                    frame_flags=frame_flags,
+                    tpci_int=tpci.to_knx(),
+                    payload_length=0,
+                ),
+            )[:4]
+            secured_apdu = apdu
+        elif scf.algorithm == SecurityAlgorithmIdentifier.CCM_ENCRYPTION:
+            mac_cbc = calculate_message_authentication_code_cbc(
+                key=key,
+                additional_data=scf.to_knx(),
+                payload=apdu,
+                block_0=block_0(
+                    sequence_number=sequence_number_bytes,
+                    address_fields_raw=address_fields_raw,
+                    frame_flags=frame_flags,
+                    tpci_int=tpci.to_knx(),
+                    payload_length=len(apdu),
+                ),
+            )[:4]
+            secured_apdu, mac = encrypt_data_ctr(
+                key=key,
+                counter_0=counter_0(
+                    sequence_number=sequence_number_bytes,
+                    address_fields_raw=address_fields_raw,
+                ),
+                mac_cbc=mac_cbc,
+                payload=apdu,
+            )
+        else:
+            raise DataSecureError(f"Unknown secure algorithm {scf.algorithm}")
+
+        return SecureData(
+            sequence_number_bytes=sequence_number_bytes,
+            secured_apdu=secured_apdu,
+            message_authentication_code=mac,  # only 4 bytes are used
+        )
+
+    def to_knx(self) -> bytes:
+        """Serialize to KNX raw data."""
+        return (
+            self.sequence_number_bytes
+            + self.secured_apdu
+            + self.message_authentication_code
+        )
+
+    @staticmethod
+    def from_knx(raw: bytes) -> SecureData:
+        """Parse/deserialize from KNX raw data."""
+        return SecureData(
+            sequence_number_bytes=raw[:6],
+            secured_apdu=raw[6:-4],
+            message_authentication_code=raw[-4:],
+        )
+
+    def get_plain_apdu(
+        self,
+        key: bytes,
+        scf: SecurityControlField,
+        address_fields_raw: bytes,
+        frame_flags: int,
+        tpci: TPCI,
+    ) -> bytes:
+        """
+        Get plain APDU as raw bytes. Decrypted or verified depending on algorithm.
+
+        Sequence number and sender individual address shall already be checked against
+        Security Individual Address Table before calling this method.
+        """
+        if scf.algorithm == SecurityAlgorithmIdentifier.CCM_ENCRYPTION:
+            dec_payload, mac_tr = decrypt_ctr(
+                key=key,
+                counter_0=counter_0(
+                    sequence_number=self.sequence_number_bytes,
+                    address_fields_raw=address_fields_raw,
+                ),
+                mac=self.message_authentication_code,
+                payload=self.secured_apdu,
+            )
+            mac_cbc = calculate_message_authentication_code_cbc(
+                key=key,
+                additional_data=scf.to_knx(),
+                payload=dec_payload,
+                block_0=block_0(
+                    sequence_number=self.sequence_number_bytes,
+                    address_fields_raw=address_fields_raw,
+                    frame_flags=frame_flags,
+                    tpci_int=tpci.to_knx(),
+                    payload_length=len(dec_payload),
+                ),
+            )[:4]
+            if mac_cbc != mac_tr:
+                raise DataSecureError("Data Secure MAC verification failed")
+            return dec_payload
+
+        if scf.algorithm == SecurityAlgorithmIdentifier.CCM_AUTHENTICATION:
+            mac = calculate_message_authentication_code_cbc(
+                key=key,
+                additional_data=scf.to_knx() + self.secured_apdu,
+                block_0=block_0(
+                    sequence_number=self.sequence_number_bytes,
+                    address_fields_raw=address_fields_raw,
+                    frame_flags=frame_flags,
+                    tpci_int=tpci.to_knx(),
+                    payload_length=0,
+                ),
+            )[:4]
+            if mac != self.message_authentication_code:
+                raise DataSecureError("Data Secure MAC verification failed.")
+            return self.secured_apdu
+
+        raise DataSecureError(f"Unknown secure algorithm {scf.algorithm}")
+
+    def __repr__(self) -> str:
+        """Return object as readable string."""
+        return (
+            "<SecureData "
+            f"sequence_number={int.from_bytes(self.sequence_number_bytes, 'big')} "
+            f'secured_apdu="{self.secured_apdu.hex()}" '
+            f'message_authentication_code="{self.message_authentication_code.hex()}" />'
+        )
diff --git a/xknx/secure/keyring.py b/xknx/secure/keyring.py
index e07998c4..4bd226f0 100644
--- a/xknx/secure/keyring.py
+++ b/xknx/secure/keyring.py
@@ -164,6 +164,7 @@ class XMLGroupAddress(AttributeReader):
     """Group Address in a knxkeys file."""
 
     address: GroupAddress
+    decrypted_key: bytes | None = None
     key: str
 
     def parse_xml(self, node: Document) -> None:
@@ -172,6 +173,15 @@ class XMLGroupAddress(AttributeReader):
         self.address = GroupAddress(self.get_attribute_value(attributes.get("Address")))
         self.key = self.get_attribute_value(attributes.get("Key"))
 
+    def decrypt_attributes(
+        self, password_hash: bytes, initialization_vector: bytes
+    ) -> None:
+        """Decrypt attribute data."""
+        if self.key:
+            self.decrypted_key = decrypt_aes128cbc(
+                base64.b64decode(self.key), password_hash, initialization_vector
+            )
+
 
 class XMLDevice(AttributeReader):
     """Device in a knxkeys file."""
diff --git a/xknx/secure/ip_secure.py b/xknx/secure/security_primitives.py
similarity index 98%
rename from xknx/secure/ip_secure.py
rename to xknx/secure/security_primitives.py
index 032d519c..87c0169c 100644
--- a/xknx/secure/ip_secure.py
+++ b/xknx/secure/security_primitives.py
@@ -1,4 +1,4 @@
-"""Encryption and Decryption functions for KNX/IP Datagrams."""
+"""Encryption and Decryption functions for KNX Secure."""
 from __future__ import annotations
 
 from cryptography.hazmat.primitives import hashes, serialization
diff --git a/xknx/telegram/apci.py b/xknx/telegram/apci.py
index e18b89ad..d06e5cda 100644
--- a/xknx/telegram/apci.py
+++ b/xknx/telegram/apci.py
@@ -15,6 +15,7 @@ from typing import ClassVar, cast
 
 from xknx.dpt import DPTArray, DPTBinary
 from xknx.exceptions import ConversionError
+from xknx.secure.data_secure_asdu import SecureData, SecurityControlField
 from xknx.telegram.address import IndividualAddress
 
 
@@ -95,6 +96,9 @@ class APCIExtendedService(Enum):
     INDIVIDUAL_ADDRESS_SERIAL_RESPONSE = 0x03DD
     INDIVIDUAL_ADDRESS_SERIAL_WRITE = 0x03DE
 
+    # DataSecure
+    APCI_SEC = 0x03F1
+
 
 class APCI(ABC):
     """
@@ -205,6 +209,8 @@ class APCI(ABC):
                 return IndividualAddressSerialResponse.from_knx(raw)
             if apci == APCIExtendedService.INDIVIDUAL_ADDRESS_SERIAL_WRITE.value:
                 return IndividualAddressSerialWrite.from_knx(raw)
+            if apci == APCIExtendedService.APCI_SEC.value:
+                return SecureAPDU.from_knx(raw)
 
         raise ConversionError(f"Class not implemented for APCI {apci:#012b}.")
 
@@ -1605,3 +1611,35 @@ class IndividualAddressSerialWrite(APCI):
     def __str__(self) -> str:
         """Return object as readable string."""
         return f'<IndividualAddressSerialWrite serial="{self.serial.hex()}" address="{self.address}" />'
+
+
+class SecureAPDU(APCI):
+    """SecureAPDU service."""
+
+    CODE = APCIExtendedService.APCI_SEC
+
+    def __init__(self, scf: SecurityControlField, secured_data: SecureData) -> None:
+        """Initialize a new instance of AuthorizeRequest."""
+        self.scf = scf
+        self.secured_data = secured_data
+
+    def calculated_length(self) -> int:
+        """Get length of APCI payload."""
+        return 2 + len(self.secured_data)
+
+    @classmethod
+    def from_knx(cls, raw: bytes) -> SecureAPDU:
+        """Parse/deserialize from KNX/IP raw data."""
+        return cls(
+            scf=SecurityControlField.from_knx(raw[2]),
+            secured_data=SecureData.from_knx(raw[3:]),
+        )
+
+    def to_knx(self) -> bytearray:
+        """Serialize to KNX/IP raw data."""
+        payload = self.scf.to_knx() + self.secured_data.to_knx()
+        return encode_cmd_and_payload(self.CODE, appended_payload=payload)
+
+    def __str__(self) -> str:
+        """Return object as readable string."""
+        return f'<SecureAPDU scf="{self.scf}" secured_data={self.secured_data!r} />'


Test patch:
diff --git a/test/io_tests/knxip_interface_test.py b/test/io_tests/knxip_interface_test.py
index 41e25e53..d1e33d3d 100644
--- a/test/io_tests/knxip_interface_test.py
+++ b/test/io_tests/knxip_interface_test.py
@@ -141,14 +141,15 @@ class TestKNXIPInterface:
         with patch("xknx.io.KNXIPInterface._start_routing") as start_routing:
             interface = knx_interface_factory(self.xknx, connection_config)
             await interface.start()
-            start_routing.assert_called_once_with(
-                local_ip=local_ip, multicast_group="224.0.23.12", multicast_port=3671
-            )
+            start_routing.assert_called_once_with()
+
         with patch("xknx.io.routing.Routing.connect") as connect_routing:
             interface = knx_interface_factory(self.xknx, connection_config)
             await interface.start()
             assert isinstance(interface._interface, Routing)
             assert interface._interface.local_ip == local_ip
+            assert interface._interface.multicast_group == "224.0.23.12"
+            assert interface._interface.multicast_port == 3671
             assert (  # pylint: disable=comparison-with-callable
                 interface._interface.cemi_received_callback == interface.cemi_received
             )
diff --git a/test/secure_tests/data_secure_test.py b/test/secure_tests/data_secure_test.py
new file mode 100644
index 00000000..0e3d3007
--- /dev/null
+++ b/test/secure_tests/data_secure_test.py
@@ -0,0 +1,387 @@
+"""Tests for KNX Data Secure."""
+import asyncio
+import os
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from xknx import XKNX
+from xknx.cemi import CEMIFrame, CEMIMessageCode
+from xknx.dpt import DPTArray
+from xknx.exceptions import DataSecureError
+from xknx.secure.data_secure_asdu import (
+    SecurityAlgorithmIdentifier,
+    SecurityALService,
+    SecurityControlField,
+)
+from xknx.secure.keyring import Keyring, _load_keyring
+from xknx.telegram import (
+    GroupAddress,
+    IndividualAddress,
+    Telegram,
+    TelegramDirection,
+    apci,
+    tpci,
+)
+
+
+@pytest.fixture
+def test_group_response_cemi():
+    """Return a CEMI frame for a group response telegram."""
+    # src = 4.0.9; dst = 0/4/0; GroupValueResponse; value=(116, 41, 41)
+    # A+C; seq_num=155806854986
+    return CEMIFrame.from_knx(
+        bytes.fromhex("29003ce0400904001103f110002446cfef4ac085e7092ab062b44d")
+    )
+
+
+@pytest.fixture
+def test_point_to_point_cemi():
+    """Return a CEMI frame for a group response telegram."""
+    # Property Value Write PID_GRP_KEY_TABLE connectionless
+    # Objet Idx = 5, PropId = 35h, Element Count = 1, Index = 1
+    # Data = 20 21 22 23 24 25 26 27 28 29 2A 2B 2C 2D 2E 2F
+    # A+C
+    # from AN158 v07 KNX Data Security AS - Annex A example
+    return CEMIFrame.from_knx(
+        bytes.fromhex(
+            "29 00 b0 60 ff 67 ff 00 22 03 f1 90 00 00 00 00"
+            "00 04 67 67 24 2a 23 08 ca 76 a1 17 74 21 4e e4"
+            "cf 5d 94 90 9f 74 3d 05 0d 8f c1 68"
+        )
+    )
+
+
+class TestDataSecure:
+    """Test class for KNX Data Secure."""
+
+    secure_test_keyring: Keyring
+
+    @classmethod
+    def setup_class(cls):
+        """Setup any state specific to the execution of the given class."""
+        secure_test_keyfile = os.path.join(
+            os.path.dirname(__file__), "resources/SecureTest.knxkeys"
+        )
+        cls.secure_test_keyring = _load_keyring(secure_test_keyfile, "test")
+
+    def setup_method(self):
+        """Setup test methods."""
+        # pylint: disable=attribute-defined-outside-init
+        self.xknx = XKNX()
+        self.xknx.knxip_interface = AsyncMock()
+        self.xknx.current_address = IndividualAddress("5.0.1")
+        self.xknx.cemi_handler.data_secure_init(TestDataSecure.secure_test_keyring)
+
+        self.data_secure = self.xknx.cemi_handler.data_secure
+
+    async def test_data_secure_init(self):
+        """Test DataSecure init and passing frames from CEMIHandler to DataSecure."""
+        assert self.data_secure is not None
+
+        assert len(self.data_secure._group_key_table) == 4
+        for ga_raw in [1024, 1027, 1028, 1029]:
+            assert GroupAddress(ga_raw) in self.data_secure._group_key_table
+
+        assert len(self.data_secure._individual_address_table) == 3
+        for ia_raw in ["4.0.0", "4.0.9", "5.0.0"]:
+            assert (
+                IndividualAddress(ia_raw) in self.data_secure._individual_address_table
+            )
+
+        # this is based on clock milliseconds
+        assert self.data_secure._sequence_number_sending > 0
+
+        test_telegram = Telegram(
+            destination_address=GroupAddress("0/4/0"),
+            payload=apci.GroupValueRead(),
+        )
+        with patch.object(self.data_secure, "outgoing_cemi") as mock_ds_outgoing_cemi:
+            task = asyncio.create_task(
+                self.xknx.cemi_handler.send_telegram(test_telegram)
+            )
+            await asyncio.sleep(0)
+            mock_ds_outgoing_cemi.assert_called_once()
+            self.xknx.cemi_handler._l_data_confirmation_event.set()
+            await task
+
+        test_cemi = CEMIFrame.init_from_telegram(test_telegram)
+        with patch.object(
+            self.data_secure, "received_cemi"
+        ) as mock_ds_received_cemi, patch.object(
+            self.xknx.cemi_handler, "telegram_received"
+        ) as mock_telegram_received:  # supress forwarding to telegras/management
+            self.xknx.cemi_handler.handle_cemi_frame(test_cemi)
+            mock_ds_received_cemi.assert_called_once()
+            mock_telegram_received.assert_called_once()
+
+    def test_data_secure_init_invalid_system_time(self):
+        """Test DataSecure init with invalid system time."""
+        with (
+            patch("time.time", return_value=1515108203.0),  # 2018-01-04T23:23:23+00:00
+            pytest.raises(DataSecureError, match=r"Initial sequence number out of .*"),
+        ):
+            self.xknx.cemi_handler.data_secure_init(TestDataSecure.secure_test_keyring)
+
+    def test_data_secure_group_send(self):
+        """Test outgoing DataSecure group communication."""
+        self.data_secure._sequence_number_sending = 160170101607
+
+        test_cemi = CEMIFrame.init_from_telegram(
+            Telegram(
+                destination_address=GroupAddress("0/4/0"),
+                payload=apci.GroupValueRead(),
+            ),
+            code=CEMIMessageCode.L_DATA_REQ,
+            src_addr=self.xknx.current_address,
+        )
+        secured_frame = self.data_secure.outgoing_cemi(test_cemi)
+        assert isinstance(secured_frame.payload, apci.SecureAPDU)
+        secured_asdu = secured_frame.payload.secured_data
+
+        assert int.from_bytes(secured_asdu.sequence_number_bytes, "big") == 160170101607
+        assert secured_asdu.secured_apdu == bytes.fromhex("cd18")
+        assert secured_asdu.message_authentication_code == bytes.fromhex("4afe5744")
+        # sequence number sending was incremented
+        assert self.data_secure._sequence_number_sending == 160170101608
+
+        assert secured_frame.to_knx() == bytes.fromhex(
+            "1100bce0500104000e03f11000254ae1cb67cd184afe5744"
+        )
+
+    def test_data_secure_group_receive(self, test_group_response_cemi):
+        """Test incoming DataSecure group communication."""
+        assert (
+            self.data_secure._individual_address_table[IndividualAddress("4.0.9")]
+            == 155806854915
+        )
+        assert test_group_response_cemi.src_addr == IndividualAddress("4.0.9")
+        assert isinstance(test_group_response_cemi.payload, apci.SecureAPDU)
+
+        plain_frame = self.data_secure.received_cemi(test_group_response_cemi)
+        assert plain_frame.payload == apci.GroupValueResponse(DPTArray((116, 41, 41)))
+        # individual_address_table sequnece number was updated
+        assert (
+            self.data_secure._individual_address_table[IndividualAddress("4.0.9")]
+            == 155806854986
+        )
+
+    def test_data_secure_individual_receive_tool_key(self, test_point_to_point_cemi):
+        """Test incoming DataSecure point-to-point communication via tool key."""
+        self.xknx.current_address = IndividualAddress("15.15.0")
+        assert isinstance(test_point_to_point_cemi.payload, apci.SecureAPDU)
+
+        with pytest.raises(
+            DataSecureError, match=r"System broadcast and tool access not supported.*"
+        ):
+            self.data_secure.received_cemi(test_point_to_point_cemi)
+        # don't raise through handle_cemi_frame()
+        assert (
+            self.xknx.cemi_handler.handle_cemi_frame(test_point_to_point_cemi) is None
+        )
+
+    def test_data_secure_individual_receive(self, test_point_to_point_cemi):
+        """Test incoming DataSecure point-to-point communication."""
+        self.xknx.current_address = IndividualAddress("15.15.0")
+        assert isinstance(test_point_to_point_cemi.payload, apci.SecureAPDU)
+        # don't use tool key or system broadcast
+        # further validation is skipped so we can use the same test data
+        test_point_to_point_cemi.payload.scf.tool_access = False
+        test_point_to_point_cemi.payload.scf.system_broadcast = False
+        with pytest.raises(
+            DataSecureError,
+            match=r"Secure Point-to-Point communication not supported.*",
+        ):
+            self.data_secure.received_cemi(test_point_to_point_cemi)
+        # don't raise through handle_cemi_frame()
+        assert (
+            self.xknx.cemi_handler.handle_cemi_frame(test_point_to_point_cemi) is None
+        )
+
+    def test_data_secure_group_receive_unknown_source(self, test_group_response_cemi):
+        """Test incoming DataSecure group communication from unknown source."""
+        test_group_response_cemi.src_addr = IndividualAddress("1.2.3")
+        with pytest.raises(
+            DataSecureError,
+            match=r"Source address not found in Security Individual Address Table.*",
+        ):
+            self.data_secure.received_cemi(test_group_response_cemi)
+
+    def test_data_secure_group_receive_unknown_destination(
+        self, test_group_response_cemi
+    ):
+        """Test incoming DataSecure group communication for unknown destination."""
+        test_group_response_cemi.dst_addr = GroupAddress("1/2/3")
+        with pytest.raises(
+            DataSecureError,
+            match=r"No key found for group address.*",
+        ):
+            self.data_secure.received_cemi(test_group_response_cemi)
+
+    def test_data_secure_group_receive_wrong_sequence_number(
+        self, test_group_response_cemi
+    ):
+        """Test incoming DataSecure group communication with wrong sequence number."""
+        seq_num = 155806854986
+        assert (
+            test_group_response_cemi.payload.secured_data.sequence_number_bytes
+            == seq_num.to_bytes(6, "big")
+        )
+        # sequence number already used
+        self.data_secure._individual_address_table[IndividualAddress("4.0.9")] = seq_num
+        with pytest.raises(
+            DataSecureError,
+            match=r"Sequence number too low.*",
+        ):
+            self.data_secure.received_cemi(test_group_response_cemi)
+
+    def test_data_secure_group_receive_wrong_mac(self, test_group_response_cemi):
+        """Test incoming DataSecure group communication with wrong MAC."""
+        test_group_response_cemi.payload.secured_data.message_authentication_code = (
+            bytes(4)
+        )
+        with pytest.raises(
+            DataSecureError,
+            match=r"Data Secure MAC verification failed.*",
+        ):
+            self.data_secure.received_cemi(test_group_response_cemi)
+
+    def test_data_secure_group_receive_plain_frame(self):
+        """Test incoming DataSecure group communication with plain frame."""
+        src_addr = IndividualAddress("4.0.9")
+        test_cemi = CEMIFrame.init_from_telegram(
+            Telegram(
+                destination_address=GroupAddress("0/4/0"),
+                direction=TelegramDirection.INCOMING,
+                payload=apci.GroupValueResponse(DPTArray((116, 41, 41))),
+            ),
+            code=CEMIMessageCode.L_DATA_IND,
+            src_addr=src_addr,
+        )
+        assert src_addr in self.data_secure._individual_address_table
+        with pytest.raises(
+            DataSecureError,
+            match=r"Discarding frame with plain APDU for secure group address.*",
+        ):
+            self.data_secure.received_cemi(test_cemi)
+
+    def test_non_secure_group_receive_plain_frame(self):
+        """Test incoming non-secure group communication with plain frame."""
+        dst_addr = GroupAddress("1/2/3")
+        test_cemi = CEMIFrame.init_from_telegram(
+            Telegram(
+                destination_address=dst_addr,
+                direction=TelegramDirection.INCOMING,
+                payload=apci.GroupValueResponse(DPTArray((116, 41, 41))),
+            ),
+            code=CEMIMessageCode.L_DATA_IND,
+            src_addr=IndividualAddress("4.0.9"),
+        )
+        assert dst_addr not in self.data_secure._group_key_table
+        assert self.data_secure.received_cemi(test_cemi) == test_cemi
+
+    def test_non_secure_group_send_plain_frame(self):
+        """Test outgoing non-secure group communication with plain frame."""
+        dst_addr = GroupAddress("1/2/3")
+        test_cemi = CEMIFrame.init_from_telegram(
+            Telegram(
+                destination_address=dst_addr,
+                direction=TelegramDirection.OUTGOING,
+                payload=apci.GroupValueResponse(DPTArray((116, 41, 41))),
+            ),
+            code=CEMIMessageCode.L_DATA_REQ,
+            src_addr=self.xknx.current_address,
+        )
+        assert dst_addr not in self.data_secure._group_key_table
+        assert self.data_secure.outgoing_cemi(test_cemi) == test_cemi
+
+    def test_non_secure_individual_receive_plain_frame(self):
+        """Test incoming non-secure group communication with plain frame."""
+        src_addr = IndividualAddress("1.2.3")
+        test_cemi = CEMIFrame.init_from_telegram(
+            Telegram(
+                destination_address=self.xknx.current_address,
+                direction=TelegramDirection.INCOMING,
+                payload=apci.PropertyValueWrite(
+                    object_index=5,
+                    property_id=0x35,
+                    count=1,
+                    start_index=1,
+                    data=bytes.fromhex(
+                        "20 21 22 23 24 25 26 27 28 29 2A 2B 2C 2D 2E 2F"
+                    ),
+                ),
+                tpci=tpci.TDataIndividual(),
+            ),
+            code=CEMIMessageCode.L_DATA_IND,
+            src_addr=IndividualAddress("4.0.9"),
+        )
+        assert src_addr not in self.data_secure._individual_address_table
+        assert self.data_secure.received_cemi(test_cemi) == test_cemi
+
+    def test_non_secure_individual_send_plain_frame(self):
+        """Test outgoing non-secure group communication with plain frame."""
+        dst_addr = IndividualAddress("1.2.3")
+        test_cemi = CEMIFrame.init_from_telegram(
+            Telegram(
+                destination_address=dst_addr,
+                direction=TelegramDirection.INCOMING,
+                payload=apci.PropertyValueWrite(
+                    object_index=5,
+                    property_id=0x35,
+                    count=1,
+                    start_index=1,
+                    data=bytes.fromhex(
+                        "20 21 22 23 24 25 26 27 28 29 2A 2B 2C 2D 2E 2F"
+                    ),
+                ),
+                tpci=tpci.TDataIndividual(),
+            ),
+            code=CEMIMessageCode.L_DATA_REQ,
+            src_addr=self.xknx.current_address,
+        )
+        assert dst_addr not in self.data_secure._individual_address_table
+        assert self.data_secure.outgoing_cemi(test_cemi) == test_cemi
+
+    def test_data_secure_authentication_only(self):
+        """Test frame de-/serialization for DataSecure authentication only."""
+        # This is currently not used from xknx and I also don't know if it is used
+        # in any ETS or runtime KNX communication. Therefore a very generic test.
+        dst_addr = GroupAddress("0/4/0")
+        test_telegram = Telegram(
+            destination_address=dst_addr,
+            direction=TelegramDirection.OUTGOING,
+            payload=apci.GroupValueWrite(DPTArray((1, 2))),
+        )
+        test_cemi = CEMIFrame.init_from_telegram(
+            telegram=test_telegram,
+            src_addr=self.xknx.current_address,
+            code=CEMIMessageCode.L_DATA_REQ,
+        )
+        scf = SecurityControlField(
+            algorithm=SecurityAlgorithmIdentifier.CCM_AUTHENTICATION,
+            service=SecurityALService.S_A_DATA,
+            system_broadcast=False,
+            tool_access=False,
+        )
+        key = self.data_secure._group_key_table[dst_addr]
+
+        outgoing_signed_cemi = self.data_secure._secure_data_cemi(
+            key=key, scf=scf, cemi=test_cemi
+        )
+        assert outgoing_signed_cemi.payload.secured_data is not None
+
+        # create new cemi to avoid mixed bytearray / byte parts
+        incoming_cemi = CEMIFrame.from_knx(outgoing_signed_cemi.to_knx())
+        # receive same cemi - fake individual address table entry
+        self.data_secure._individual_address_table[incoming_cemi.src_addr] = 1
+        assert self.data_secure.received_cemi(incoming_cemi) == test_cemi
+
+        # Test wrong MAC
+        self.data_secure._individual_address_table[incoming_cemi.src_addr] = 1
+        incoming_cemi.payload.secured_data.message_authentication_code = bytes(4)
+        with pytest.raises(
+            DataSecureError,
+            match=r"Data Secure MAC verification failed.*",
+        ):
+            self.data_secure.received_cemi(incoming_cemi)
diff --git a/test/secure_tests/ip_secure_test.py b/test/secure_tests/ip_secure_test.py
index be416c2b..b98c85e9 100644
--- a/test/secure_tests/ip_secure_test.py
+++ b/test/secure_tests/ip_secure_test.py
@@ -2,7 +2,7 @@
 
 from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
 
-from xknx.secure.ip_secure import (
+from xknx.secure.security_primitives import (
     calculate_message_authentication_code_cbc,
     decrypt_ctr,
     derive_device_authentication_password,
diff --git a/test/secure_tests/resources/SecureTest.knxkeys b/test/secure_tests/resources/SecureTest.knxkeys
new file mode 100644
index 00000000..cf919324
--- /dev/null
+++ b/test/secure_tests/resources/SecureTest.knxkeys
@@ -0,0 +1,30 @@
+﻿<?xml version="1.0" encoding="utf-8"?>
+<Keyring Project="SecureTest" CreatedBy="ETS 5.7.7 (Build 1428)" Created="2023-01-30T19:17:06" Signature="4SLrRPMpwFql5ytGzPjeBw==" xmlns="http://knx.org/xml/keyring/1">
+  <Backbone MulticastAddress="224.0.23.13" Latency="1000" Key="JxjdynGrSOB+9XV9NBaNxQ==" />
+  <Interface IndividualAddress="4.0.1" Type="Tunneling" Host="4.0.0" Authentication="KMAXnY+PvyhPVuIwClDcyEt2v9RR1cVYue8FT4QaDbU=">
+    <Group Address="1024" Senders="4.0.9 5.0.1" />
+  </Interface>
+  <Interface IndividualAddress="5.0.1" Type="Tunneling" Host="5.0.0" UserID="2" Password="sQZMSoHR/gBjj1guSY2gh4Sziv8cAiwar9F1amBRfE4=" Authentication="fEa/xofJEErwm1uzFITCerK35WrS4V0L3z8bPLBtGfg=">
+    <Group Address="1028" Senders="4.0.9" />
+    <Group Address="1027" Senders="4.0.9" />
+    <Group Address="1024" Senders="4.0.1 4.0.9" />
+  </Interface>
+  <Interface IndividualAddress="5.0.2" Type="Tunneling" Host="5.0.0" UserID="3" Password="yKolpl+K/0+sUJ4tRBKrxjZL8RAc3Zkn55CNFlHTWvQ=" Authentication="gIO2HmBA3I4nEjp3OGGsfKe9y3FiiBlPHzdk1lUf+zA=" />
+  <Interface IndividualAddress="5.0.3" Type="Tunneling" Host="5.0.0" UserID="4" Password="clJrtIVeayj7b6QpRsg2qZYW+PFyM1imJ9hLmz/wULE=" Authentication="YJ6wkBiZb0wNBCZ/G+kYrRFffxpzVSyaTYHo8yHi3DY=" />
+  <Interface IndividualAddress="5.0.4" Type="Tunneling" Host="5.0.0" UserID="5" Password="fMwapxDmIbigAP6GQEF1rk9+hSRW3SdrtN6rNRfzPeQ=" Authentication="oIDv7WGKMYUh6CYQei4ZpmogUSYoGA+v8iaSa6i8PKM=" />
+  <Interface IndividualAddress="5.0.5" Type="Tunneling" Host="5.0.0" UserID="6" Password="tbt3BABEvN49RxjaohXZAcM+JVjZc2pzbdNnrCXU5o4=" Authentication="HXdJBWnwCpDvoX+2WzhLUNjbLexUIgcwNRkPi3gMAlE=" />
+  <Interface IndividualAddress="5.0.6" Type="Tunneling" Host="5.0.0" UserID="7" Password="u5i4foIlngxrJ9gP3cEpMAXzNtkb9Qho4hKrbHqymWo=" Authentication="aNCuQJGwzryLRLFC1zPLAxtpmjloFbsGMmkjhZltSsI=" />
+  <Interface IndividualAddress="5.0.7" Type="Tunneling" Host="5.0.0" UserID="8" Password="sC8o/bioM8K0Eqm9U6Ptopa/H2IAbC8/G6mmBfpAw98=" Authentication="YoO9rL94q4J2/ing0F+qCYPumplCxO/EtdL16kCXaQE=" />
+  <Interface IndividualAddress="5.0.8" Type="Tunneling" Host="5.0.0" UserID="9" Password="sQ0XqUaOMyKmFEyYlkqAlJNW5ESWmrx87QsXLRsl7/A=" Authentication="Xaepbgxq5p3buurTduX9UkQtxyhMpsPamjdqaTN2e4I=" />
+  <GroupAddresses>
+    <Group Address="1024" Key="B8nWiBBMs/wH3I3aYKmHlw==" />
+    <Group Address="1027" Key="uoumUsZz3bG4yIu1PsBODQ==" />
+    <Group Address="1028" Key="cXzDe6WQtZJDusyfmbcMiQ==" />
+    <Group Address="1029" Key="+OAsRnyeACX+bUeja9r6mg==" />
+  </GroupAddresses>
+  <Devices>
+    <Device IndividualAddress="4.0.0" ToolKey="aTUDIKnMShsWFPjLz5co8w==" ManagementPassword="wlVWWM46kmxoPGdoYIxsmPT3SP0+SGFgh4Pai8XpR6Y=" Authentication="VYq5X3LJmoxFcf27G0OvIHJ7tbAFq4u7UTC+6tpo46s=" SequenceNumber="155806720116" />
+    <Device IndividualAddress="4.0.9" ToolKey="Ov+HnCVh/87jokTrsRnlfw==" SequenceNumber="155806854915" />
+    <Device IndividualAddress="5.0.0" ToolKey="mR+sUiCJV80vniR1/rIZVg==" ManagementPassword="5wbPUMc+XZije4vjR3WU4eqKWsPm95LRKqf5vqYvNqk=" Authentication="7a7fmoyJtRmPKx6/RPMRoKb+DxPFCn4UqoY1qplOV7A=" SequenceNumber="146212262858" />
+  </Devices>
+</Keyring>
\ No newline at end of file