Repository: XKNX/xknx
Pull request #1045 (2022-10-03 21:57:19)

Add IP-Secure Routing support
<!--
  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.
-->
Add IP-Secure Routing support

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:

- [ ] The documentation has been adjusted accordingly
- [ ] 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/1045?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#1045](https://codecov.io/gh/XKNX/xknx/pull/1045?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (ad9f5a6) into [main](https://codecov.io/gh/XKNX/xknx/commit/58348de59662222cc63b498d52bdb2faccbaa894?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (58348de) will **decrease** coverage by `1.75%`.
> The diff coverage is `36.11%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/1045/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/1045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main    #1045      +/-   ##
==========================================
- Coverage   96.42%   94.67%   -1.76%     
==========================================
  Files         138      138              
  Lines        8363     8575     +212     
==========================================
+ Hits         8064     8118      +54     
- Misses        299      457     +158     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/1045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/io/tunnel.py](https://codecov.io/gh/XKNX/xknx/pull/1045/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby90dW5uZWwucHk=) | `88.72% <ø> (ø)` | |
| [xknx/io/knxip\_interface.py](https://codecov.io/gh/XKNX/xknx/pull/1045/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9rbnhpcF9pbnRlcmZhY2UucHk=) | `76.92% <21.73%> (-8.27%)` | :arrow_down: |
| [xknx/io/secure\_session.py](https://codecov.io/gh/XKNX/xknx/pull/1045/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9zZWN1cmVfc2Vzc2lvbi5weQ==) | `54.80% <32.17%> (-41.57%)` | :arrow_down: |
| [xknx/io/routing.py](https://codecov.io/gh/XKNX/xknx/pull/1045/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9yb3V0aW5nLnB5) | `87.40% <68.42%> (-2.08%)` | :arrow_down: |
| [xknx/io/connection.py](https://codecov.io/gh/XKNX/xknx/pull/1045/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9jb25uZWN0aW9uLnB5) | `97.29% <100.00%> (+0.23%)` | :arrow_up: |
| [xknx/secure/keyring.py](https://codecov.io/gh/XKNX/xknx/pull/1045/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9zZWN1cmUva2V5cmluZy5weQ==) | `99.11% <100.00%> (+<0.01%)` | :arrow_up: |

farmio: Cool! I'll try to add some tests before merging. 

Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index 0bc6d12d..31ec1409 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -8,6 +8,12 @@ nav_order: 2
 
 # Unreleased changes
 
+### Features
+
+- We now support KNXnet/IP Secure multicast communication (secure routing) in addition to tunnelling!
+  Thanks to Weinzierl for providing us a router for testing purposes!
+- Parse `latency` from a .knxkeys keyring files `backbone` tag.
+
 ### Protocol
 
 - Add TimerNotify frame parser
diff --git a/xknx/io/connection.py b/xknx/io/connection.py
index 44c36f6f..96297a37 100644
--- a/xknx/io/connection.py
+++ b/xknx/io/connection.py
@@ -12,6 +12,7 @@ class ConnectionType(Enum):
 
     AUTOMATIC = auto()
     ROUTING = auto()
+    ROUTING_SECURE = auto()
     TUNNELING = auto()
     TUNNELING_TCP = auto()
     TUNNELING_TCP_SECURE = auto()
@@ -78,6 +79,8 @@ class SecureConfig:
     Secure configuration.
 
     Handles:
+    * backbone_key: Key used for KNX Secure Routing in hex representation.
+    * latency_ms: Latency in milliseconds for KNX Secure Routing.
     * user_id: The user id to use when initializing the secure tunnel.
     * device_authentication_password: the authentication password to use when connecting to the tunnel.
     * user_password: the user password for knx secure.
@@ -88,6 +91,8 @@ class SecureConfig:
     def __init__(
         self,
         *,
+        backbone_key: str | None = None,
+        latency_ms: int | None = None,
         user_id: int | None = None,
         device_authentication_password: str | None = None,
         user_password: str | None = None,
@@ -95,6 +100,8 @@ class SecureConfig:
         knxkeys_password: str | None = None,
     ):
         """Initialize SecureConfig class."""
+        self.backbone_key = bytes.fromhex(backbone_key) if backbone_key else None
+        self.latency_ms = latency_ms
         self.user_id = user_id
         self.device_authentication_password = device_authentication_password
         self.user_password = user_password
diff --git a/xknx/io/ip_secure.py b/xknx/io/ip_secure.py
new file mode 100644
index 00000000..d7781892
--- /dev/null
+++ b/xknx/io/ip_secure.py
@@ -0,0 +1,726 @@
+"""IPSecure is an abstraction for handling a KNXnet/IP Secure layer."""
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+import asyncio
+import logging
+import random
+from typing import Callable, Final
+
+from cryptography.hazmat.primitives.asymmetric.x25519 import (
+    X25519PrivateKey,
+    X25519PublicKey,
+)
+
+from xknx.exceptions import (
+    CommunicationError,
+    CouldNotParseKNXIP,
+    KNXSecureValidationError,
+)
+from xknx.knxip import (
+    HPAI,
+    KNXIPFrame,
+    KNXIPServiceType,
+    SecureWrapper,
+    SessionResponse,
+    SessionStatus,
+    TimerNotify,
+)
+from xknx.knxip.knxip_enum import SecureSessionStatusCode
+from xknx.secure.ip_secure import (
+    calculate_message_authentication_code_cbc,
+    decrypt_ctr,
+    derive_device_authentication_password,
+    derive_user_password,
+    encrypt_data_ctr,
+    generate_ecdh_key_pair,
+)
+from xknx.secure.util import bytes_xor, sha256_hash
+
+from .const import SESSION_KEEPALIVE_RATE, XKNX_SERIAL_NUMBER
+from .request_response import Authenticate, Session
+from .transport import KNXIPTransport, TCPTransport, UDPTransport
+
+logger = logging.getLogger("xknx.log")
+knx_logger = logging.getLogger("xknx.knx")
+ip_secure_logger = logging.getLogger("xknx.ip_secure")
+
+
+COUNTER_0_HANDSHAKE = (  # used in SessionResponse and SessionAuthenticate
+    b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\x00"
+)
+MESSAGE_TAG_TUNNELLING = bytes.fromhex("00 00")  # use 0x00 0x00 for tunneling
+
+
+class _IPSecureTransportLayer(ABC):
+    """Abstract for Secure transport layer."""
+
+    session_id: int
+    _key: bytes
+
+    @abstractmethod
+    def get_sequence_information(self) -> bytes:
+        """Return byte representation of sequence information."""
+
+    @abstractmethod
+    def get_message_tag(self) -> bytes:
+        """Return byte representation of message tag."""
+
+    def decrypt_frame(self, encrypted_frame: KNXIPFrame) -> KNXIPFrame:
+        """Unwrap and verify KNX/IP frame from SecureWrapper."""
+        # TODO: get raw data from KNXIPFrame class directly instead of recalculating it with to_knx()
+        # TODO: refactor so assert isn't needed (maybe subclass SecureWrapper from KNXIPFrame instead of being an attribute)
+        assert isinstance(encrypted_frame.body, SecureWrapper)
+        if encrypted_frame.body.secure_session_id != self.session_id:
+            raise KNXSecureValidationError("Invalid secure session id")
+
+        session_id_bytes = encrypted_frame.body.secure_session_id.to_bytes(2, "big")
+        wrapper_header = encrypted_frame.header.to_knx()
+
+        dec_frame, mac_tr = decrypt_ctr(
+            key=self._key,
+            counter_0=(
+                encrypted_frame.body.sequence_information
+                + encrypted_frame.body.serial_number
+                + encrypted_frame.body.message_tag
+                + b"\xff\x00"
+            ),
+            mac=encrypted_frame.body.message_authentication_code,
+            payload=encrypted_frame.body.encrypted_data,
+        )
+        mac_cbc = calculate_message_authentication_code_cbc(
+            key=self._key,
+            additional_data=wrapper_header + session_id_bytes,
+            payload=dec_frame,
+            block_0=(
+                encrypted_frame.body.sequence_information
+                + encrypted_frame.body.serial_number
+                + encrypted_frame.body.message_tag
+                + len(dec_frame).to_bytes(2, "big")
+            ),
+        )
+        if mac_cbc != mac_tr:
+            raise KNXSecureValidationError(
+                "Verification of message authentication code failed"
+            )
+
+        knxipframe = KNXIPFrame()
+        knxipframe.from_knx(dec_frame)
+        return knxipframe
+
+    def encrypt_frame(self, plain_frame: KNXIPFrame) -> KNXIPFrame:
+        """Wrap KNX/IP frame in SecureWrapper."""
+        sequence_information = self.get_sequence_information()
+        message_tag = self.get_message_tag()
+        plain_payload = plain_frame.to_knx()  # P
+        payload_length = len(plain_payload)  # Q
+        # 6 KNXnet/IP header, 2 session_id, 6 sequence_number, 6 serial_number, 2 message_tag, 16 MAC = 38
+        total_length = 38 + payload_length
+        # TODO: get header data and total_length from SecureWrapper class
+        wrapper_header = bytes.fromhex("06 10 09 50") + total_length.to_bytes(2, "big")
+
+        mac_cbc = calculate_message_authentication_code_cbc(
+            key=self._key,
+            additional_data=wrapper_header + self.session_id.to_bytes(2, "big"),
+            payload=plain_payload,
+            block_0=(
+                sequence_information
+                + XKNX_SERIAL_NUMBER
+                + message_tag
+                + payload_length.to_bytes(2, "big")
+            ),
+        )
+        encrypted_data, mac = encrypt_data_ctr(
+            key=self._key,
+            counter_0=(
+                sequence_information + XKNX_SERIAL_NUMBER + message_tag + b"\xff\x00"
+            ),
+            mac_cbc=mac_cbc,
+            payload=plain_payload,
+        )
+        return KNXIPFrame.init_from_body(
+            SecureWrapper(
+                secure_session_id=self.session_id,
+                sequence_information=sequence_information,
+                serial_number=XKNX_SERIAL_NUMBER,
+                message_tag=message_tag,
+                encrypted_data=encrypted_data,
+                message_authentication_code=mac,
+            )
+        )
+
+
+class SecureSession(TCPTransport, _IPSecureTransportLayer):
+    """Class for handling a KNXnet/IP Secure tunnelling session."""
+
+    def __init__(
+        self,
+        remote_addr: tuple[str, int],
+        user_id: int,
+        user_password: str,
+        device_authentication_password: str | None = None,
+        connection_lost_cb: Callable[[], None] | None = None,
+    ) -> None:
+        """Initialize SecureSession class."""
+        super().__init__(
+            remote_addr=remote_addr,
+            connection_lost_cb=connection_lost_cb,
+        )
+        self._device_authentication_code: bytes | None = (
+            derive_device_authentication_password(device_authentication_password)
+            if device_authentication_password
+            else None
+        )
+        self.user_id = user_id
+        self._user_password = derive_user_password(user_password)
+
+        self._private_key: X25519PrivateKey
+        self.public_key: bytes
+        self._peer_public_key: X25519PublicKey
+        self._key: bytes  # Session Key
+        self.session_id: int
+
+        self._sequence_number: int = 0
+        self._sequence_number_received: int = -1
+        self.initialized: bool = False
+        self._keepalive_task: asyncio.Task[None] | None = None
+        self._session_status_handler: KNXIPTransport.Callback | None = None
+
+    async def connect(self) -> None:
+        """Connect transport."""
+        await super().connect()
+        self._private_key, self.public_key = generate_ecdh_key_pair()
+        self._sequence_number = 0
+        self._sequence_number_received = -1
+        request_session = Session(
+            transport=self,
+            ecdh_client_public_key=self.public_key,
+        )
+        await request_session.start()
+        if request_session.response is None:
+            raise CommunicationError(
+                "Secure session could not be established. No response received."
+            )
+        # SessionAuthenticate and everything else after now shall be wrapped in SecureWrapper
+        authenticate_mac = self.handshake(request_session.response)
+        self.initialized = True
+
+        request_authentication = Authenticate(
+            transport=self,
+            user_id=self.user_id,
+            message_authentication_code=authenticate_mac,
+        )
+        await request_authentication.start()
+        if request_authentication.response is None:
+            raise CommunicationError(
+                "Secure session could not be established. No response received."
+            )
+        if (  # TODO: look for status in request/response and use `success` instead of response ?
+            request_authentication.response.status
+            != SecureSessionStatusCode.STATUS_AUTHENTICATION_SUCCESS
+        ):
+            raise CommunicationError(
+                f"Secure session authentication failed: {request_authentication.response.status}"
+            )
+        self._session_status_handler = self.register_callback(
+            self._handle_session_status, [KNXIPServiceType.SESSION_STATUS]
+        )
+
+    def handshake(self, session_response: SessionResponse) -> bytes:
+        """
+        Handshake with device.
+
+        Returns a SessionAuthenticate KNX/IP body.
+        """
+        self._peer_public_key = X25519PublicKey.from_public_bytes(
+            session_response.ecdh_server_public_key
+        )
+        self.session_id = session_response.secure_session_id
+        # verify SessionResponse MAC
+        # TODO: get header data from actual KNX/IP frame
+        response_header_data = bytes.fromhex("06 10 09 52 00 38")
+        pub_keys_xor = bytes_xor(
+            self.public_key,
+            session_response.ecdh_server_public_key,
+        )
+        if self._device_authentication_code:
+            response_mac_cbc = calculate_message_authentication_code_cbc(
+                key=self._device_authentication_code,
+                additional_data=response_header_data
+                + self.session_id.to_bytes(2, "big")
+                + pub_keys_xor,  # knx_ip_header + secure_session_id + bytes_xor(client_pub_key, server_pub_key)
+            )
+            _, mac_tr = decrypt_ctr(
+                key=self._device_authentication_code,
+                counter_0=COUNTER_0_HANDSHAKE,
+                mac=session_response.message_authentication_code,
+            )
+            if mac_tr != response_mac_cbc:
+                raise CommunicationError("SessionResponse MAC verification failed.")
+        # calculate session key
+        ecdh_shared_secret = self._private_key.exchange(self._peer_public_key)
+        self._key = sha256_hash(ecdh_shared_secret)[:16]
+        # generate SessionAuthenticate MAC
+        authenticate_header_data = bytes.fromhex("06 10 09 53 00 18")
+        authenticate_mac_cbc = calculate_message_authentication_code_cbc(
+            key=self._user_password,
+            additional_data=authenticate_header_data
+            + bytes(1)  # reserved
+            + self.user_id.to_bytes(1, "big")
+            + pub_keys_xor,
+            block_0=bytes(16),
+        )
+        _, authenticate_mac = encrypt_data_ctr(
+            key=self._user_password,
+            counter_0=COUNTER_0_HANDSHAKE,
+            mac_cbc=authenticate_mac_cbc,
+        )
+        return authenticate_mac
+
+    def stop(self) -> None:
+        """Stop session and transport."""
+        if self._session_status_handler:
+            self.unregister_callback(self._session_status_handler)
+            self._session_status_handler = None
+        if self.transport and self.initialized:
+            self.send(
+                KNXIPFrame.init_from_body(
+                    SessionStatus(status=SecureSessionStatusCode.STATUS_CLOSE)
+                )
+            )
+        self.stop_keepalive_task()
+        self.initialized = False
+        super().stop()
+
+    def handle_knxipframe(self, knxipframe: KNXIPFrame, source: HPAI) -> None:
+        """Handle KNXIP Frame and call all callbacks matching the service type ident."""
+        # TODO: disallow unencrypted frames with exceptions for discovery etc. eg. DescriptionResponse
+        if isinstance(knxipframe.body, SecureWrapper):
+            if not self.initialized:
+                raise CouldNotParseKNXIP(
+                    "Received SecureWrapper while Secure session not initialized"
+                )
+            new_sequence_number = int.from_bytes(
+                knxipframe.body.sequence_information, "big"
+            )
+            if not new_sequence_number > self._sequence_number_received:
+                ip_secure_logger.warning(
+                    "Discarding SecureWrapper with invalid sequence number: %s",
+                    knxipframe,
+                )
+                return
+            try:
+                knxipframe = self.decrypt_frame(knxipframe)
+            except KNXSecureValidationError as err:
+                ip_secure_logger.warning("Could not decrypt KNXIPFrame: %s", err)
+                # Frame shall be discarded
+                return
+            except CouldNotParseKNXIP as couldnotparseknxip:
+                knx_logger.debug(
+                    "Unsupported encrypted KNXIPFrame: %s",
+                    couldnotparseknxip.description,
+                )
+                return
+            self._sequence_number_received = new_sequence_number
+            knx_logger.debug("Decrypted frame: %s", knxipframe)
+        super().handle_knxipframe(knxipframe, source)
+
+    async def _session_keepalive(self) -> None:
+        """Keep session alive."""
+        await asyncio.sleep(SESSION_KEEPALIVE_RATE)
+        self.send(
+            KNXIPFrame.init_from_body(
+                SessionStatus(status=SecureSessionStatusCode.STATUS_KEEPALIVE)
+            )
+        )
+
+    def send(self, knxipframe: KNXIPFrame, addr: tuple[str, int] | None = None) -> None:
+        """Send KNXIPFrame to socket. `addr` is ignored on TCP."""
+        if self.initialized:
+            knx_logger.debug("Encrypting frame: %s", knxipframe)
+            knxipframe = self.encrypt_frame(plain_frame=knxipframe)
+            # keepalive timer is started with first and resetted with every other
+            # SecureWrapper frame (including wrapped keepalive frames themselves)
+            self.start_keepalive_task()
+        # TODO: disallow sending unencrypted frames over non-initialized session with
+        # exceptions for discovery and SessionRequest
+        super().send(knxipframe, addr)
+
+    def get_sequence_information(self) -> bytes:
+        """Increment sequence number. Return byte representation of current sequence number."""
+        next_sn = self._sequence_number.to_bytes(6, "big")
+        self._sequence_number += 1
+        return next_sn
+
+    def get_message_tag(self) -> bytes:
+        """Return byte representation of current message tag."""
+        return MESSAGE_TAG_TUNNELLING
+
+    def start_keepalive_task(self) -> None:
+        """Start or restart session keepalive task."""
+        if self._keepalive_task:
+            self._keepalive_task.cancel()
+        self._keepalive_task = asyncio.create_task(self._session_keepalive())
+
+    def stop_keepalive_task(self) -> None:
+        """Stop keepalive task."""
+        if self._keepalive_task:
+            self._keepalive_task.cancel()
+            self._keepalive_task = None
+
+    def _handle_session_status(
+        self, knxipframe: KNXIPFrame, source: HPAI, transport: KNXIPTransport
+    ) -> None:
+        """Handle session status."""
+        assert isinstance(knxipframe.body, SessionStatus)
+        if knxipframe.body.status in (
+            SecureSessionStatusCode.STATUS_CLOSE,
+            SecureSessionStatusCode.STATUS_TIMEOUT,
+            SecureSessionStatusCode.STATUS_UNAUTHENTICATED,
+        ):
+            logger.info("Secure session closed by server: %s.", knxipframe.body.status)
+            if self.transport:
+                # closing transport will call `asyncio.Protocol.connection_lost`
+                # and its callback from SecureTunnel
+                self.transport.close()
+
+
+class SecureGroup(UDPTransport, _IPSecureTransportLayer):
+    """Class for secure KNXnet/IP multicast communication."""
+
+    session_id = 0  # Routing uses fixed session id 0
+
+    def __init__(
+        self,
+        local_addr: tuple[str, int],
+        remote_addr: tuple[str, int],
+        backbone_key: bytes,
+        latency_ms: int = 1000,
+    ):
+        """Initialize SecureGroup class."""
+        super().__init__(
+            local_addr=local_addr,
+            remote_addr=remote_addr,
+            multicast=True,
+        )
+        self._key = backbone_key
+        self.secure_timer = SecureSequenceTimer(
+            backbone_key=backbone_key,
+            latency_ms=latency_ms,
+            transport_send=super().send,
+        )
+
+    async def connect(self) -> None:
+        """Connect transport."""
+        await super().connect()
+        await self.secure_timer.synchronize()
+
+    def stop(self) -> None:
+        """Stop tasks and transport."""
+        self.secure_timer.stop()
+        super().stop()
+
+    def handle_knxipframe(self, knxipframe: KNXIPFrame, source: HPAI) -> None:
+        """Handle KNXIP Frame and call all callbacks matching the service type ident."""
+        # TODO: disallow unencrypted frames with exceptions for discovery etc. eg. SearchRequest
+        if isinstance(knxipframe.body, TimerNotify):
+            self.secure_timer.handle_timer_notify(knxipframe.body)
+            return
+        if isinstance(knxipframe.body, SecureWrapper):
+            if not self.secure_timer.validate_secure_wrapper(knxipframe.body):
+                ip_secure_logger.warning(
+                    "Discarding SecureWrapper with invalid timer value: %s",
+                    knxipframe,
+                )
+                return
+            if not self.secure_timer.timer_authenticated:
+                ip_secure_logger.debug(
+                    "Discarding received SecureWrapper before timer synchronisazion finished: %s",
+                    knxipframe,
+                )
+                return
+            try:
+                knxipframe = self.decrypt_frame(knxipframe)
+            except KNXSecureValidationError as err:
+                ip_secure_logger.warning("Could not decrypt KNXIPFrame: %s", err)
+                # Frame shall be discarded
+                return
+            except CouldNotParseKNXIP as couldnotparseknxip:
+                knx_logger.debug(
+                    "Unsupported encrypted KNXIPFrame: %s",
+                    couldnotparseknxip.description,
+                )
+                return
+            knx_logger.debug("Decrypted frame: %s", knxipframe)
+        super().handle_knxipframe(knxipframe, source)
+
+    def send(self, knxipframe: KNXIPFrame, addr: tuple[str, int] | None = None) -> None:
+        """Send KNXIPFrame to socket. `addr` is ignored on TCP."""
+        knx_logger.debug("Encrypting frame: %s", knxipframe)
+        knxipframe = self.encrypt_frame(plain_frame=knxipframe)
+        # TODO: disallow sending unencrypted frames over non-initialized session with
+        # exceptions for discovery and SessionRequest
+        super().send(knxipframe, addr)
+
+    def get_sequence_information(self) -> bytes:
+        """Return byte representation of current timer value."""
+        return self.secure_timer.get_for_outgoing_secure_wrapper().to_bytes(6, "big")
+
+    def get_message_tag(self) -> bytes:
+        """Return byte representation of current message tag."""
+        return random.randbytes(2)
+
+
+class SecureSequenceTimer:
+    """
+    Class for holding and synchronizing the timer for secure sequence information.
+
+    According to AN159 v06 KNXnet-IP Secure AS §2.2.2.3 Timer synchronizing
+    """
+
+    min_delay_time_keeper_periodic_notify: Final = 10  # pylint: disable=invalid-name
+    min_delay_time_keeper_update_notify: Final = 0.1  # pylint: disable=invalid-name
+    sync_latency_fraction: int = 10  # 10%
+
+    def __init__(
+        self,
+        backbone_key: bytes,
+        latency_ms: int,
+        transport_send: Callable[[KNXIPFrame, tuple[str, int] | None], None],
+    ) -> None:
+        """Initialize SecureSequenceTimer class."""
+        self._backbone_key = backbone_key
+        self._clock_difference: int = 0
+        self._expected_notify_handler: tuple[
+            bytes, asyncio.Future[int]  # message_tag, synchronization future
+        ] | None = None
+        self._loop = asyncio.get_running_loop()
+        self._notify_timer_handle: asyncio.TimerHandle | None = None
+        self._transport_send = transport_send
+
+        self.sched_update: bool = False
+        self.timekeeper: bool = False
+        self.timer_authenticated: bool = False
+
+        self.latency_tolerance_ms = latency_ms
+        self.sync_latency_tolerance_ms: int = round(
+            latency_ms / 100 * self.sync_latency_fraction
+        )
+        _sync_latency_tolerance_seconds = self.sync_latency_tolerance_ms / 1000
+        self.max_delay_time_keeper_periodic_notify = (
+            self.min_delay_time_keeper_periodic_notify
+            + _sync_latency_tolerance_seconds * 3
+        )
+        self.min_delay_time_follower_periodic_notify = (
+            self.max_delay_time_keeper_periodic_notify
+            + _sync_latency_tolerance_seconds * 1
+        )
+        self.max_delay_time_follower_periodic_notify = (
+            self.min_delay_time_follower_periodic_notify
+            + _sync_latency_tolerance_seconds * 10
+        )
+        self.max_delay_time_keeper_update_notify = (
+            self.min_delay_time_keeper_update_notify
+            + _sync_latency_tolerance_seconds * 1
+        )
+        self.min_delay_time_follower_update_notify = (
+            self.max_delay_time_keeper_update_notify
+            + _sync_latency_tolerance_seconds * 1
+        )
+        self.max_delay_time_follower_update_notify = (
+            self.min_delay_time_follower_update_notify
+            + _sync_latency_tolerance_seconds * 10
+        )
+
+    def stop(self) -> None:
+        """Cancel notify timer."""
+        if self._notify_timer_handle:
+            self._notify_timer_handle.cancel()
+            self._notify_timer_handle = None
+        if self._expected_notify_handler:
+            self._expected_notify_handler[1].cancel()
+
+    def _monotonic_ms(self) -> int:
+        """Return current monotonic time in milliseconds."""
+        return int(self._loop.time() * 1000.0)
+
+    def current_timer_value(self) -> int:
+        """Return current timer value in milliseconds."""
+        return self._monotonic_ms() + self._clock_difference
+
+    def update(self, new_value: int) -> None:
+        """Update timer value."""
+        self._clock_difference = new_value - self._monotonic_ms()
+
+    def reschedule(self, update: tuple[bytes, bytes] | None = None) -> None:
+        """
+        Reschedule notify timer.
+
+        If `update` is set, the timer is rescheduled for an update notify,
+        else for a periodic notify.
+        `update` is expected to be a tuple of (message_tag, serial_number).
+        """
+        if self._notify_timer_handle is not None:
+            self._notify_timer_handle.cancel()
+        self.sched_update = bool(update)
+        min_delay: float
+        max_delay: float
+        if self.sched_update:
+            if self.timekeeper:
+                min_delay = self.min_delay_time_keeper_update_notify
+                max_delay = self.max_delay_time_keeper_update_notify
+            else:
+                min_delay = self.min_delay_time_follower_update_notify
+                max_delay = self.max_delay_time_follower_update_notify
+        else:
+            if self.timekeeper:
+                min_delay = self.min_delay_time_keeper_periodic_notify
+                max_delay = self.max_delay_time_keeper_periodic_notify
+            else:
+                min_delay = self.min_delay_time_follower_periodic_notify
+                max_delay = self.max_delay_time_follower_periodic_notify
+        self._notify_timer_handle = self._loop.call_later(
+            random.uniform(min_delay, max_delay), self._notify_timer_expired, update
+        )
+
+    def _notify_timer_expired(self, update: tuple[bytes, bytes] | None) -> None:
+        """Notify timer expired."""
+        if update:
+            self.send_timer_notify(message_tag=update[0], serial_number=update[1])
+        else:
+            self.send_timer_notify()
+        if not self.timekeeper:
+            self.timekeeper = True
+            ip_secure_logger.debug("Becoming timekeeper")
+        self.reschedule()
+
+    def send_timer_notify(
+        self,
+        message_tag: bytes | None = None,
+        serial_number: bytes = XKNX_SERIAL_NUMBER,
+    ) -> None:
+        """Send a TimerNotify frame."""
+        timer_value = self.current_timer_value()
+        timer_bytes = timer_value.to_bytes(6, "big")
+        _message_tag = message_tag or random.randbytes(2)
+        b_0 = timer_bytes + serial_number + _message_tag + b"\x00\x00"
+        # TODO: get header data and total_length from TimerNotify class
+        #       or handle mac calculation in TimerNotify class directly
+        timer_notify_header = bytes.fromhex("06 10 09 55 00 24")
+        mac_cbc = calculate_message_authentication_code_cbc(
+            key=self._backbone_key,
+            additional_data=timer_notify_header,
+            block_0=b_0,
+        )
+        c_0 = timer_bytes + serial_number + _message_tag + b"\xff\x00"
+        _, mac = encrypt_data_ctr(
+            key=self._backbone_key,
+            counter_0=c_0,
+            mac_cbc=mac_cbc,
+        )
+        self._transport_send(
+            KNXIPFrame.init_from_body(
+                TimerNotify(
+                    timer_value=timer_value,
+                    serial_number=serial_number,
+                    message_tag=_message_tag,
+                    message_authentication_code=mac,
+                )
+            ),
+            None,
+        )
+
+    async def synchronize(self) -> None:
+        """Synchronize timer with remote time keeper."""
+        message_tag = random.randbytes(2)
+        waiter_fut: asyncio.Future[int] = self._loop.create_future()
+        self._expected_notify_handler = message_tag, waiter_fut
+        self.send_timer_notify(message_tag=message_tag)
+        try:
+            timer_value = await asyncio.wait_for(
+                waiter_fut,
+                timeout=(  # 3.3 seconds at latency_ms=1000, sync_latency_fraction=10%
+                    self.max_delay_time_follower_update_notify
+                    + 2 * self.latency_tolerance_ms / 1000
+                ),
+            )
+            self.update(new_value=timer_value)
+        except asyncio.TimeoutError:
+            # use highest received timer value of TimerNotify or SecureWrapper frames
+            ip_secure_logger.warning(
+                "Timer synchronization not answered. Becoming time keeper."
+            )
+            self.timekeeper = True
+        except asyncio.CancelledError:
+            return
+        finally:
+            self._expected_notify_handler = None
+        self.timer_authenticated = True
+        self.reschedule()
+
+    def validate_secure_wrapper(self, secure_wrapper: SecureWrapper) -> bool:
+        """Validate a SecureWrapper frames and handle timer update schedule."""
+        local_timer_value = self.current_timer_value()
+        received_timer_value = int.from_bytes(
+            secure_wrapper.sequence_information, "big"
+        )
+        # §2.2.2.3.2.5 Events: E5 - E8
+        if received_timer_value > local_timer_value:
+            self._clock_difference += received_timer_value - local_timer_value
+            if not self.sched_update:
+                self.reschedule()
+            return True
+        if received_timer_value > local_timer_value - self.sync_latency_tolerance_ms:
+            if not self.sched_update:
+                self.reschedule()
+            return True
+        if received_timer_value > local_timer_value - self.latency_tolerance_ms:
+            return True
+        if received_timer_value <= local_timer_value - self.latency_tolerance_ms:
+            if not self.sched_update:
+                self.reschedule(
+                    update=(secure_wrapper.message_tag, secure_wrapper.serial_number)
+                )
+        return False
+
+    def handle_timer_notify(self, timer_notify: TimerNotify) -> None:
+        """Handle received TimerNotify frame."""
+        local_timer_value = self.current_timer_value()
+        received_timer_value = timer_notify.timer_value
+        # §2.2.2.3.2.5 Events: E11
+        if (
+            self._expected_notify_handler is not None
+            and timer_notify.serial_number == XKNX_SERIAL_NUMBER
+            and timer_notify.message_tag == self._expected_notify_handler[0]
+        ):
+            fut = self._expected_notify_handler[1]
+            fut.set_result(timer_notify.timer_value)
+            return
+        # §2.2.2.3.2.5 Events: E1 - E4
+        if received_timer_value > local_timer_value:
+            self._clock_difference += received_timer_value - local_timer_value
+            if self.timekeeper:
+                ip_secure_logger.debug("Becoming time follower")
+                self.timekeeper = False
+            self.reschedule()
+            return
+        if received_timer_value > local_timer_value - self.sync_latency_tolerance_ms:
+            if self.timekeeper:
+                ip_secure_logger.debug("Becoming time follower")
+                self.timekeeper = False
+            self.reschedule()
+            return
+        if received_timer_value > local_timer_value - self.latency_tolerance_ms:
+            return
+        if received_timer_value <= local_timer_value - self.latency_tolerance_ms:
+            if not self.sched_update:
+                self.reschedule(
+                    update=(timer_notify.message_tag, timer_notify.serial_number)
+                )
+
+    def get_for_outgoing_secure_wrapper(self) -> int:
+        """Return current timer value and handle timer update schedule."""
+        # §2.2.2.3.2.5 Events: E9
+        if not self.sched_update:
+            self.reschedule()
+        return self.current_timer_value()
diff --git a/xknx/io/knxip_interface.py b/xknx/io/knxip_interface.py
index 816cc8da..e165600a 100644
--- a/xknx/io/knxip_interface.py
+++ b/xknx/io/knxip_interface.py
@@ -21,12 +21,12 @@ from xknx.exceptions import (
     XKNXException,
 )
 from xknx.io import util
-from xknx.secure import Keyring, load_key_ring
+from xknx.secure import load_key_ring
 from xknx.telegram import IndividualAddress, Telegram
 
 from .connection import ConnectionConfig, ConnectionType
 from .gateway_scanner import GatewayDescriptor, GatewayScanner
-from .routing import Routing
+from .routing import Routing, SecureRouting
 from .tunnel import SecureTunnel, TCPTunnel, UDPTunnel, _Tunnel
 
 if TYPE_CHECKING:
@@ -72,6 +72,35 @@ class KNXIPInterface:
         """Start interface. Connecting KNX/IP device with the selected method."""
         if self.connection_config.connection_type == ConnectionType.ROUTING:
             await self._start_routing(local_ip=self.connection_config.local_ip)
+        elif (
+            self.connection_config.connection_type == ConnectionType.ROUTING_SECURE
+            and self.connection_config.secure_config is not None
+        ):
+            secure_config = self.connection_config.secure_config
+            backbone_key = secure_config.backbone_key
+            latency_ms = secure_config.latency_ms
+            if (
+                secure_config.knxkeys_file_path is not None
+                and secure_config.knxkeys_password is not None
+            ):
+                keyring = load_key_ring(
+                    secure_config.knxkeys_file_path, secure_config.knxkeys_password
+                )
+                if keyring.backbone is None:
+                    raise InvalidSecureConfiguration(
+                        "No backbone key found in knxkeys file"
+                    )
+                backbone_key = backbone_key or keyring.backbone.decrypted_key
+                latency_ms = latency_ms or keyring.backbone.latency
+            if not backbone_key:
+                raise InvalidSecureConfiguration(
+                    "No backbone key found in secure configuration"
+                )
+            await self._start_secure_routing(
+                backbone_key,
+                latency_ms=latency_ms,
+                local_ip=self.connection_config.local_ip,
+            )
         elif (
             self.connection_config.connection_type == ConnectionType.TUNNELING
             and self.connection_config.gateway_ip is not None
@@ -102,7 +131,7 @@ class KNXIPInterface:
                 secure_config.knxkeys_file_path is not None
                 and secure_config.knxkeys_password is not None
             ):
-                keyring: Keyring = load_key_ring(
+                keyring = load_key_ring(
                     secure_config.knxkeys_file_path, secure_config.knxkeys_password
                 )
                 if secure_config.user_id is not None:
@@ -270,6 +299,26 @@ class KNXIPInterface:
         self._interface = Routing(self.xknx, self.telegram_received, local_ip)
         await self._interface.connect()
 
+    async def _start_secure_routing(
+        self,
+        backbone_key: bytes,
+        latency_ms: int | None = None,
+        local_ip: str | None = None,
+    ) -> None:
+        """Start KNX/IP Routing."""
+        local_ip = 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")
+
+        logger.debug(
+            "Starting Secure Routing from %s as %s", local_ip, self.xknx.own_address
+        )
+        self._interface = SecureRouting(
+            self.xknx, self.telegram_received, local_ip, backbone_key, latency_ms
+        )
+        await self._interface.connect()
+
     async def stop(self) -> None:
         """Stop connected interfae (either Tunneling or Routing)."""
         if self._interface is not None:
diff --git a/xknx/io/routing.py b/xknx/io/routing.py
index 1e5699f5..7c9573c2 100644
--- a/xknx/io/routing.py
+++ b/xknx/io/routing.py
@@ -1,7 +1,7 @@
 """
-Abstraction for handling KNX/IP routing.
+Abstraction for handling KNXnet/IP routing.
 
-Routing uses UDP Multicast to broadcast and receive KNX/IP messages.
+Routing uses UDP Multicast to send and receive KNXnet/IP messages.
 """
 from __future__ import annotations
 
@@ -27,6 +27,7 @@ from xknx.knxip import (
 from xknx.telegram import Telegram, TelegramDirection
 
 from .interface import Interface, TelegramCallbackType
+from .ip_secure import SecureGroup
 from .transport import KNXIPTransport, UDPTransport
 
 if TYPE_CHECKING:
@@ -40,10 +41,12 @@ BUSY_RANDOM_TIME_FACTOR: Final = 0.05  # 50 ms
 BUSY_SLOWDURATION_TIME_FACTOR: Final = 0.1  # 100 ms
 ROUTING_INDICATION_WAIT_TIME: Final = 0.02  # 20 ms
 
+DEFAULT_LATENCY_TOLERANCE_MS: Final = 1000
+
 
 class _RoutingFlowControl:
     """
-    Class for hanling KNX/IP routing flow control.
+    Class for hanling KNXnet/IP routing flow control.
 
     See KNX Specifications 3.8.5 Routing §2.3.5 Flow control handling
     """
@@ -127,7 +130,9 @@ class _RoutingFlowControl:
 
 
 class Routing(Interface):
-    """Class for handling KNX/IP routing."""
+    """Class for handling KNXnet/IP multicast communication."""
+
+    transport: UDPTransport
 
     def __init__(
         self,
@@ -140,12 +145,8 @@ class Routing(Interface):
         self.telegram_received_callback = telegram_received_callback
         self.local_ip = local_ip
 
-        self.udp_transport = UDPTransport(
-            local_addr=(local_ip, 0),
-            remote_addr=(self.xknx.multicast_group, self.xknx.multicast_port),
-            multicast=True,
-        )
-        self.udp_transport.register_callback(
+        self._init_transport()
+        self.transport.register_callback(
             self._handle_frame,
             [
                 KNXIPServiceType.ROUTING_INDICATION,
@@ -155,6 +156,14 @@ class Routing(Interface):
         )
         self._flow_control = _RoutingFlowControl()
 
+    def _init_transport(self) -> None:
+        """Initialize transport."""
+        self.transport = UDPTransport(
+            local_addr=(self.local_ip, 0),
+            remote_addr=(self.xknx.multicast_group, self.xknx.multicast_port),
+            multicast=True,
+        )
+
     ####################
     #
     # CONNECT DISCONNECT
@@ -167,10 +176,10 @@ class Routing(Interface):
             XknxConnectionState.CONNECTING
         )
         try:
-            await self.udp_transport.connect()
+            await self.transport.connect()
         except OSError as ex:
             logger.debug(
-                "Could not establish connection to KNX/IP network. %s: %s",
+                "Could not establish connection to KNXnet/IP network. %s: %s",
                 type(ex).__name__,
                 ex,
             )
@@ -178,7 +187,7 @@ class Routing(Interface):
                 XknxConnectionState.DISCONNECTED
             )
             # close udp transport to prevent open file descriptors
-            self.udp_transport.stop()
+            self.transport.stop()
             raise CommunicationError("Routing could not be started") from ex
         await self.xknx.connection_manager.connection_state_changed(
             XknxConnectionState.CONNECTED
@@ -187,7 +196,7 @@ class Routing(Interface):
 
     async def disconnect(self) -> None:
         """Stop routing."""
-        self.udp_transport.stop()
+        self.transport.stop()
         await self.xknx.connection_manager.connection_state_changed(
             XknxConnectionState.DISCONNECTED
         )
@@ -213,7 +222,7 @@ class Routing(Interface):
 
     def _send_knxipframe(self, knxipframe: KNXIPFrame) -> None:
         """Send KNXIPFrame to connected routing device."""
-        self.udp_transport.send(knxipframe)
+        self.transport.send(knxipframe)
 
     ##################
     #
@@ -224,7 +233,7 @@ class Routing(Interface):
     def _handle_frame(
         self, knxipframe: KNXIPFrame, source: HPAI, _: KNXIPTransport
     ) -> None:
-        """Handle incoming KNXIPFrames. Callback from internal udp_transport."""
+        """Handle incoming KNXIPFrames. Callback from internal transport."""
         if isinstance(knxipframe.body, RoutingIndication):
             self._handle_routing_indication(knxipframe.body)
         elif isinstance(knxipframe.body, RoutingBusy):
@@ -258,3 +267,31 @@ class Routing(Interface):
         if response_tgs := await self.telegram_received_callback(telegram):
             for response in response_tgs:
                 await self.send_telegram(response)
+
+
+class SecureRouting(Routing):
+    """Class for handling KNXnet/IP secure multicast communication."""
+
+    transport: SecureGroup
+
+    def __init__(
+        self,
+        xknx: XKNX,
+        telegram_received_callback: TelegramCallbackType,
+        local_ip: str,
+        backbone_key: bytes,
+        latency_ms: int | None = None,
+    ):
+        """Initialize SecureRouting class."""
+        self.backbone_key = backbone_key
+        self.latency_ms = latency_ms or DEFAULT_LATENCY_TOLERANCE_MS
+        super().__init__(xknx, telegram_received_callback, local_ip)
+
+    def _init_transport(self) -> None:
+        """Initialize transport."""
+        self.transport = SecureGroup(
+            local_addr=(self.local_ip, 0),
+            remote_addr=(self.xknx.multicast_group, self.xknx.multicast_port),
+            backbone_key=self.backbone_key,
+            latency_ms=self.latency_ms,
+        )
diff --git a/xknx/io/secure_session.py b/xknx/io/secure_session.py
deleted file mode 100644
index a9da671f..00000000
--- a/xknx/io/secure_session.py
+++ /dev/null
@@ -1,364 +0,0 @@
-"""SecureSession is an abstraction for handling a KNXnet/IP Secure session."""
-from __future__ import annotations
-
-import asyncio
-import logging
-from typing import Callable
-
-from cryptography.hazmat.primitives.asymmetric.x25519 import (
-    X25519PrivateKey,
-    X25519PublicKey,
-)
-
-from xknx.exceptions import (
-    CommunicationError,
-    CouldNotParseKNXIP,
-    KNXSecureValidationError,
-)
-from xknx.knxip import (
-    HPAI,
-    KNXIPFrame,
-    KNXIPServiceType,
-    SecureWrapper,
-    SessionResponse,
-    SessionStatus,
-)
-from xknx.knxip.knxip_enum import SecureSessionStatusCode
-from xknx.secure.ip_secure import (
-    calculate_message_authentication_code_cbc,
-    decrypt_ctr,
-    derive_device_authentication_password,
-    derive_user_password,
-    encrypt_data_ctr,
-    generate_ecdh_key_pair,
-)
-from xknx.secure.util import bytes_xor, sha256_hash
-
-from .const import SESSION_KEEPALIVE_RATE, XKNX_SERIAL_NUMBER
-from .request_response import Authenticate, Session
-from .transport import KNXIPTransport, TCPTransport
-
-logger = logging.getLogger("xknx.log")
-knx_logger = logging.getLogger("xknx.knx")
-
-
-COUNTER_0_HANDSHAKE = (  # used in SessionResponse and SessionAuthenticate
-    b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\x00"
-)
-MESSAGE_TAG_TUNNELLING = bytes.fromhex("00 00")  # use 0x00 0x00 for tunneling
-
-
-class SecureSession(TCPTransport):
-    """Class for handling a KNXnet/IP Secure session."""
-
-    def __init__(
-        self,
-        remote_addr: tuple[str, int],
-        user_id: int,
-        user_password: str,
-        device_authentication_password: str | None = None,
-        connection_lost_cb: Callable[[], None] | None = None,
-    ) -> None:
-        """Initialize SecureSession class."""
-        super().__init__(
-            remote_addr=remote_addr,
-            connection_lost_cb=connection_lost_cb,
-        )
-        self._device_authentication_code: bytes | None = (
-            derive_device_authentication_password(device_authentication_password)
-            if device_authentication_password
-            else None
-        )
-        self.user_id = user_id
-        self._user_password = derive_user_password(user_password)
-
-        self._private_key: X25519PrivateKey
-        self.public_key: bytes
-        self._peer_public_key: X25519PublicKey
-        self._session_key: bytes
-        self.session_id: int
-
-        self._sequence_number = 0
-        self._sequence_number_received = -1
-        self.initialized = False
-        self._keepalive_task: asyncio.Task[None] | None = None
-        self._session_status_handler: KNXIPTransport.Callback | None = None
-
-    async def connect(self) -> None:
-        """Connect transport."""
-        await super().connect()
-        self._private_key, self.public_key = generate_ecdh_key_pair()
-        self._sequence_number = 0
-        self._sequence_number_received = -1
-        request_session = Session(
-            transport=self,
-            ecdh_client_public_key=self.public_key,
-        )
-        await request_session.start()
-        if request_session.response is None:
-            raise CommunicationError(
-                "Secure session could not be established. No response received."
-            )
-        # SessionAuthenticate and everything else after now shall be wrapped in SecureWrapper
-        authenticate_mac = self.handshake(request_session.response)
-        self.initialized = True
-
-        request_authentication = Authenticate(
-            transport=self,
-            user_id=self.user_id,
-            message_authentication_code=authenticate_mac,
-        )
-        await request_authentication.start()
-        if request_authentication.response is None:
-            raise CommunicationError(
-                "Secure session could not be established. No response received."
-            )
-        if (  # TODO: look for status in request/response and use `success` instead of response ?
-            request_authentication.response.status
-            != SecureSessionStatusCode.STATUS_AUTHENTICATION_SUCCESS
-        ):
-            raise CommunicationError(
-                f"Secure session authentication failed: {request_authentication.response.status}"
-            )
-        self._session_status_handler = self.register_callback(
-            self._handle_session_status, [KNXIPServiceType.SESSION_STATUS]
-        )
-
-    def handshake(self, session_response: SessionResponse) -> bytes:
-        """
-        Handshake with device.
-
-        Returns a SessionAuthenticate KNX/IP body.
-        """
-        self._peer_public_key = X25519PublicKey.from_public_bytes(
-            session_response.ecdh_server_public_key
-        )
-        self.session_id = session_response.secure_session_id
-        # verify SessionResponse MAC
-        # TODO: get header data from actual KNX/IP frame
-        response_header_data = bytes.fromhex("06 10 09 52 00 38")
-        pub_keys_xor = bytes_xor(
-            self.public_key,
-            session_response.ecdh_server_public_key,
-        )
-        if self._device_authentication_code:
-            response_mac_cbc = calculate_message_authentication_code_cbc(
-                key=self._device_authentication_code,
-                additional_data=response_header_data
-                + self.session_id.to_bytes(2, "big")
-                + pub_keys_xor,  # knx_ip_header + secure_session_id + bytes_xor(client_pub_key, server_pub_key)
-            )
-            _, mac_tr = decrypt_ctr(
-                key=self._device_authentication_code,
-                counter_0=COUNTER_0_HANDSHAKE,
-                mac=session_response.message_authentication_code,
-            )
-            if mac_tr != response_mac_cbc:
-                raise CommunicationError("SessionResponse MAC verification failed.")
-        # calculate session key
-        ecdh_shared_secret = self._private_key.exchange(self._peer_public_key)
-        self._session_key = sha256_hash(ecdh_shared_secret)[:16]
-        # generate SessionAuthenticate MAC
-        authenticate_header_data = bytes.fromhex("06 10 09 53 00 18")
-        authenticate_mac_cbc = calculate_message_authentication_code_cbc(
-            key=self._user_password,
-            additional_data=authenticate_header_data
-            + bytes(1)  # reserved
-            + self.user_id.to_bytes(1, "big")
-            + pub_keys_xor,
-            block_0=bytes(16),
-        )
-        _, authenticate_mac = encrypt_data_ctr(
-            key=self._user_password,
-            counter_0=COUNTER_0_HANDSHAKE,
-            mac_cbc=authenticate_mac_cbc,
-        )
-        return authenticate_mac
-
-    def stop(self) -> None:
-        """Stop session and transport."""
-        if self._session_status_handler:
-            self.unregister_callback(self._session_status_handler)
-            self._session_status_handler = None
-        if self.transport and self.initialized:
-            self.send(
-                KNXIPFrame.init_from_body(
-                    SessionStatus(status=SecureSessionStatusCode.STATUS_CLOSE)
-                )
-            )
-        self.stop_keepalive_task()
-        self.initialized = False
-        super().stop()
-
-    def handle_knxipframe(self, knxipframe: KNXIPFrame, source: HPAI) -> None:
-        """Handle KNXIP Frame and call all callbacks matching the service type ident."""
-        # TODO: disallow unencrypted frames with exceptions for discovery etc. eg. DescriptionResponse
-        if isinstance(knxipframe.body, SecureWrapper):
-            if not self.initialized:
-                raise CouldNotParseKNXIP(
-                    "Received SecureWrapper while Secure session not initialized"
-                )
-            new_sequence_number = int.from_bytes(
-                knxipframe.body.sequence_information, "big"
-            )
-            if not new_sequence_number > self._sequence_number_received:
-                knx_logger.warning(
-                    "Discarding SecureWrapper with invalid sequence number: %s",
-                    knxipframe,
-                )
-                return
-            try:
-                knxipframe = self.decrypt_frame(knxipframe)
-            except KNXSecureValidationError as err:
-                knx_logger.warning("Could not decrypt KNXIPFrame: %s", err)
-                # Frame shall be discarded
-                return
-            except CouldNotParseKNXIP as couldnotparseknxip:
-                knx_logger.debug(
-                    "Unsupported encrypted KNXIPFrame: %s",
-                    couldnotparseknxip.description,
-                )
-                return
-            self._sequence_number_received = new_sequence_number
-            knx_logger.debug("Decrypted frame: %s", knxipframe)
-        super().handle_knxipframe(knxipframe, source)
-
-    def decrypt_frame(self, encrypted_frame: KNXIPFrame) -> KNXIPFrame:
-        """Unwrap and verify KNX/IP frame from SecureWrapper."""
-        # TODO: get raw data from KNXIPFrame class directly instead of recalculating it with to_knx()
-        # TODO: refactor so assert isn't needed (maybe subclass SecureWrapper from KNXIPFrame instead of being an attribute)
-        assert isinstance(encrypted_frame.body, SecureWrapper)
-        if encrypted_frame.body.secure_session_id != self.session_id:
-            raise KNXSecureValidationError("Wrong secure session id")
-
-        session_id_bytes = encrypted_frame.body.secure_session_id.to_bytes(2, "big")
-        wrapper_header = encrypted_frame.header.to_knx()
-
-        dec_frame, mac_tr = decrypt_ctr(
-            key=self._session_key,
-            counter_0=(
-                encrypted_frame.body.sequence_information
-                + encrypted_frame.body.serial_number
-                + encrypted_frame.body.message_tag
-                + bytes.fromhex("ff 00")
-            ),
-            mac=encrypted_frame.body.message_authentication_code,
-            payload=encrypted_frame.body.encrypted_data,
-        )
-        mac_cbc = calculate_message_authentication_code_cbc(
-            key=self._session_key,
-            additional_data=wrapper_header + session_id_bytes,
-            payload=dec_frame,
-            block_0=(
-                encrypted_frame.body.sequence_information
-                + encrypted_frame.body.serial_number
-                + encrypted_frame.body.message_tag
-                + len(dec_frame).to_bytes(2, "big")
-            ),
-        )
-        if mac_cbc != mac_tr:
-            raise KNXSecureValidationError(
-                "Verification of message authentication code failed"
-            )
-
-        knxipframe = KNXIPFrame()
-        knxipframe.from_knx(dec_frame)
-        return knxipframe
-
-    async def _session_keepalive(self) -> None:
-        """Keep session alive."""
-        await asyncio.sleep(SESSION_KEEPALIVE_RATE)
-        self.send(
-            KNXIPFrame.init_from_body(
-                SessionStatus(status=SecureSessionStatusCode.STATUS_KEEPALIVE)
-            )
-        )
-
-    def send(self, knxipframe: KNXIPFrame, addr: tuple[str, int] | None = None) -> None:
-        """Send KNXIPFrame to socket. `addr` is ignored on TCP."""
-        if self.initialized:
-            knx_logger.debug("Encrypting frame: %s", knxipframe)
-            knxipframe = self.encrypt_frame(plain_frame=knxipframe)
-            # keepalive timer is started with first and resetted with every other
-            # SecureWrapper frame (including wrapped keepalive frames themselves)
-            self.start_keepalive_task()
-        # TODO: disallow sending unencrypted frames over non-initialized session with
-        # exceptions for discovery and SessionRequest
-        super().send(knxipframe, addr)
-
-    def encrypt_frame(self, plain_frame: KNXIPFrame) -> KNXIPFrame:
-        """Wrap KNX/IP frame in SecureWrapper."""
-        sequence_information = self.increment_sequence_number()
-        plain_payload = plain_frame.to_knx()  # P
-        payload_length = len(plain_payload)  # Q
-        # 6 KNXnet/IP header, 2 session_id, 6 sequence_number, 6 serial_number, 2 message_tag, 16 MAC = 38
-        total_length = 38 + payload_length
-        # TODO: get header data and total_length from SecureWrapper class
-        wrapper_header = bytes.fromhex("06 10 09 50") + total_length.to_bytes(2, "big")
-
-        mac_cbc = calculate_message_authentication_code_cbc(
-            key=self._session_key,
-            additional_data=wrapper_header + self.session_id.to_bytes(2, "big"),
-            payload=plain_payload,
-            block_0=(
-                sequence_information
-                + XKNX_SERIAL_NUMBER
-                + MESSAGE_TAG_TUNNELLING
-                + payload_length.to_bytes(2, "big")
-            ),
-        )
-        encrypted_data, mac = encrypt_data_ctr(
-            key=self._session_key,
-            counter_0=(
-                sequence_information
-                + XKNX_SERIAL_NUMBER
-                + MESSAGE_TAG_TUNNELLING
-                + bytes.fromhex("ff 00")
-            ),
-            mac_cbc=mac_cbc,
-            payload=plain_payload,
-        )
-        return KNXIPFrame.init_from_body(
-            SecureWrapper(
-                secure_session_id=self.session_id,
-                sequence_information=sequence_information,
-                serial_number=XKNX_SERIAL_NUMBER,
-                message_tag=MESSAGE_TAG_TUNNELLING,
-                encrypted_data=encrypted_data,
-                message_authentication_code=mac,
-            )
-        )
-
-    def increment_sequence_number(self) -> bytes:
-        """Increment sequence number. Return byte representation of current sequence number."""
-        next_sn = self._sequence_number.to_bytes(6, "big")
-        self._sequence_number += 1
-        return next_sn
-
-    def start_keepalive_task(self) -> None:
-        """Start or restart session keepalive task."""
-        if self._keepalive_task:
-            self._keepalive_task.cancel()
-        self._keepalive_task = asyncio.create_task(self._session_keepalive())
-
-    def stop_keepalive_task(self) -> None:
-        """Stop keepalive task."""
-        if self._keepalive_task:
-            self._keepalive_task.cancel()
-            self._keepalive_task = None
-
-    def _handle_session_status(
-        self, knxipframe: KNXIPFrame, source: HPAI, transport: KNXIPTransport
-    ) -> None:
-        """Handle session status."""
-        assert isinstance(knxipframe.body, SessionStatus)
-        if knxipframe.body.status in (
-            SecureSessionStatusCode.STATUS_CLOSE,
-            SecureSessionStatusCode.STATUS_TIMEOUT,
-            SecureSessionStatusCode.STATUS_UNAUTHENTICATED,
-        ):
-            logger.info("Secure session closed by server: %s.", knxipframe.body.status)
-            if self.transport:
-                # closing transport will call `asyncio.Protocol.connection_lost`
-                # and its callback from SecureTunnel
-                self.transport.close()
diff --git a/xknx/io/tunnel.py b/xknx/io/tunnel.py
index bedc0284..10e68b05 100644
--- a/xknx/io/tunnel.py
+++ b/xknx/io/tunnel.py
@@ -30,8 +30,8 @@ from xknx.telegram import IndividualAddress, Telegram, TelegramDirection
 from .const import HEARTBEAT_RATE
 from .gateway_scanner import GatewayDescriptor
 from .interface import Interface, TelegramCallbackType
+from .ip_secure import SecureSession
 from .request_response import Connect, ConnectionState, Disconnect, Tunnelling
-from .secure_session import SecureSession
 from .self_description import DescriptionQuery
 from .transport import KNXIPTransport, TCPTransport, UDPTransport
 
@@ -85,7 +85,7 @@ class _Tunnel(Interface):
 
     @abstractmethod
     def _init_transport(self) -> None:
-        """Initialize transport transport."""
+        """Initialize transport."""
         # set up self.transport
 
     @abstractmethod
diff --git a/xknx/secure/keyring.py b/xknx/secure/keyring.py
index 1727e5e9..2459def8 100644
--- a/xknx/secure/keyring.py
+++ b/xknx/secure/keyring.py
@@ -129,17 +129,19 @@ class XMLInterface(AttributeReader):
 class XMLBackbone(AttributeReader):
     """Backbone in a knxkeys file."""
 
-    multicast_address: str
-    key: str
     decrypted_key: bytes
+    key: str
+    latency: int
+    multicast_address: str
 
     def parse_xml(self, node: Document) -> None:
         """Parse all needed attributes from the given node map."""
         attributes = node.attributes
+        self.key = self.get_attribute_value(attributes.get("Key"))
+        self.latency = int(self.get_attribute_value(attributes.get("Latency")))
         self.multicast_address = self.get_attribute_value(
             attributes.get("MulticastAddress")
         )
-        self.key = self.get_attribute_value(attributes.get("Key"))
 
     def decrypt_attributes(
         self, password_hash: bytes, initialization_vector: bytes


Test patch:
diff --git a/requirements/testing.txt b/requirements/testing.txt
index 6c1e47ae..fe2ea675 100644
--- a/requirements/testing.txt
+++ b/requirements/testing.txt
@@ -10,4 +10,4 @@ pytest-asyncio==0.19.0
 pytest-cov==4.0.0
 setuptools==65.4.1
 tox==3.26.0
-mypy==0.981
+mypy==0.982
diff --git a/test/io_tests/knxip_interface_test.py b/test/io_tests/knxip_interface_test.py
index 58d00aab..6c03fe65 100644
--- a/test/io_tests/knxip_interface_test.py
+++ b/test/io_tests/knxip_interface_test.py
@@ -18,7 +18,7 @@ from xknx.io import (
     SecureConfig,
     knx_interface_factory,
 )
-from xknx.io.routing import Routing
+from xknx.io.routing import Routing, SecureGroup, SecureRouting
 from xknx.io.tunnel import SecureTunnel, TCPTunnel, UDPTunnel
 
 
@@ -373,3 +373,73 @@ class TestKNXIPInterface:
         with pytest.raises(InvalidSecureConfiguration):
             interface = knx_interface_factory(self.xknx, connection_config)
             await interface.start()
+
+    async def test_start_secure_routing_knx_keys(self):
+        """Test starting a secure routing connection from a knxkeys file."""
+        backbone_key = bytes.fromhex("cf89fd0f18f4889783c7ef44ee1f5e14")
+        knxkeys_file = os.path.join(
+            os.path.dirname(__file__), "resources/testcase.knxkeys"
+        )
+        connection_config = ConnectionConfig(
+            connection_type=ConnectionType.ROUTING_SECURE,
+            secure_config=SecureConfig(
+                knxkeys_file_path=knxkeys_file, knxkeys_password="password"
+            ),
+        )
+        with patch(
+            "xknx.io.KNXIPInterface._start_secure_routing"
+        ) as start_secure_tunnel:
+            interface = knx_interface_factory(self.xknx, connection_config)
+            await interface.start()
+            start_secure_tunnel.assert_called_once_with(
+                backbone_key, latency_ms=1000, local_ip=None
+            )
+        with patch("xknx.io.routing.SecureRouting.connect") as connect_secure:
+            interface = knx_interface_factory(self.xknx, connection_config)
+            await interface.start()
+            assert isinstance(interface._interface, SecureRouting)
+            assert interface._interface.backbone_key == backbone_key
+            assert interface._interface.latency_ms == 1000
+            assert isinstance(interface._interface.transport, SecureGroup)
+            assert interface._interface.transport.remote_addr == (
+                self.xknx.multicast_group,
+                3671,
+            )
+            assert (  # pylint: disable=comparison-with-callable
+                interface._interface.telegram_received_callback
+                == interface.telegram_received
+            )
+            connect_secure.assert_called_once_with()
+
+    async def test_start_secure_routing_manual(self):
+        """Test starting a secure routing connection from a knxkeys file."""
+        backbone_key_str = "cf89fd0f18f4889783c7ef44ee1f5e14"
+        backbone_key = bytes.fromhex(backbone_key_str)
+        connection_config = ConnectionConfig(
+            connection_type=ConnectionType.ROUTING_SECURE,
+            secure_config=SecureConfig(backbone_key=backbone_key_str, latency_ms=2000),
+        )
+        with patch(
+            "xknx.io.KNXIPInterface._start_secure_routing"
+        ) as start_secure_tunnel:
+            interface = knx_interface_factory(self.xknx, connection_config)
+            await interface.start()
+            start_secure_tunnel.assert_called_once_with(
+                backbone_key, latency_ms=2000, local_ip=None
+            )
+        with patch("xknx.io.routing.SecureRouting.connect") as connect_secure:
+            interface = knx_interface_factory(self.xknx, connection_config)
+            await interface.start()
+            assert isinstance(interface._interface, SecureRouting)
+            assert interface._interface.backbone_key == backbone_key
+            assert interface._interface.latency_ms == 2000
+            assert isinstance(interface._interface.transport, SecureGroup)
+            assert interface._interface.transport.remote_addr == (
+                self.xknx.multicast_group,
+                3671,
+            )
+            assert (  # pylint: disable=comparison-with-callable
+                interface._interface.telegram_received_callback
+                == interface.telegram_received
+            )
+            connect_secure.assert_called_once_with()
diff --git a/test/io_tests/routing_test.py b/test/io_tests/routing_test.py
index f80fd56c..6739a23f 100644
--- a/test/io_tests/routing_test.py
+++ b/test/io_tests/routing_test.py
@@ -56,7 +56,7 @@ class TestRouting:
             return [response_telegram]
 
         routing.telegram_received_callback = tg_received_mock
-        routing.udp_transport.data_received_callback(raw_ind, ("192.168.1.2", 3671))
+        routing.transport.data_received_callback(raw_ind, ("192.168.1.2", 3671))
         await asyncio.sleep(0)
         assert send_knxipframe_mock.call_args_list == [
             call(response_frame),
@@ -71,7 +71,7 @@ class TestRouting:
             local_ip="192.168.1.1",
         )
         raw = bytes((0x06, 0x10, 0x05, 0x31, 0x00, 0x0A, 0x04, 0x00, 0x00, 0x05))
-        routing.udp_transport.data_received_callback(raw, ("192.168.1.2", 3671))
+        routing.transport.data_received_callback(raw, ("192.168.1.2", 3671))
         assert logging_mock.called_once_with(
             "RoutingLostMessage received from %s - %s lost messages.",
             "192.168.1.2",
diff --git a/test/io_tests/secure_group_test.py b/test/io_tests/secure_group_test.py
new file mode 100644
index 00000000..45fe7012
--- /dev/null
+++ b/test/io_tests/secure_group_test.py
@@ -0,0 +1,432 @@
+"""Test Secure Group."""
+import asyncio
+from unittest.mock import patch
+
+from xknx.io.const import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT, XKNX_SERIAL_NUMBER
+from xknx.io.ip_secure import SecureGroup
+from xknx.knxip import HPAI, KNXIPFrame, SecureWrapper, TimerNotify
+from xknx.knxip.cemi_frame import CEMIFrame
+from xknx.knxip.routing_indication import RoutingIndication
+from xknx.telegram import GroupAddress, Telegram, apci
+
+ONE_HOUR_MS = 60 * 60 * 1000
+
+
+@patch("xknx.io.transport.udp_transport.UDPTransport.connect")
+@patch("xknx.io.transport.udp_transport.UDPTransport.send")
+class TestSecureGroup:
+    """Test class for xknx/io/SecureGroup objects."""
+
+    mock_addr = ("127.0.0.1", 12345)
+    mock_backbone_key = bytes.fromhex(
+        "0a a2 27 b4 fd 7a 32 31 9b a9 96 0a c0 36 ce 0e"
+        "5c 45 07 b5 ae 55 16 1f 10 78 b1 dc fb 3c b6 31"
+    )
+
+    def assert_timer_notify(
+        self,
+        knxipframe,
+        timer_value=None,
+        serial_number=None,
+        message_tag=None,
+        mac=None,
+    ) -> bytes:
+        """Assert that knxipframe is a TimerNotify."""
+        assert isinstance(knxipframe.body, TimerNotify)
+        if timer_value is not None:
+            assert knxipframe.body.timer_value == timer_value
+        if serial_number is not None:
+            assert knxipframe.body.serial_number == serial_number
+        if message_tag is not None:
+            assert knxipframe.body.message_tag == message_tag
+        if mac is not None:
+            assert knxipframe.body.message_authentication_code == mac
+        return knxipframe.body.message_tag
+
+    async def test_no_synchronize(
+        self,
+        mock_super_send,
+        mock_super_connect,
+        time_travel,
+    ):
+        """Test synchronisazion not answered."""
+        secure_group = SecureGroup(
+            local_addr=self.mock_addr,
+            remote_addr=(DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT),
+            backbone_key=self.mock_backbone_key,
+            latency_ms=1000,
+        )
+        secure_timer = secure_group.secure_timer
+        connect_task = asyncio.create_task(secure_group.connect())
+        await time_travel(0)
+        mock_super_connect.assert_called_once()
+        self.assert_timer_notify(
+            mock_super_send.call_args[0][0],
+            serial_number=XKNX_SERIAL_NUMBER,
+        )
+        mock_super_send.reset_mock()
+        # synchronize timed out
+        await time_travel(
+            secure_timer.max_delay_time_follower_update_notify
+            + 2 * secure_timer.latency_tolerance_ms / 1000
+        )
+        assert connect_task.done()
+        # we are timekeeper so we send TimerNotify after time_keeper_periodic
+        assert secure_timer.timekeeper
+        assert not secure_timer.sched_update
+        await time_travel(secure_timer.min_delay_time_keeper_periodic_notify - 0.01)
+        mock_super_send.assert_not_called()
+        await time_travel(secure_timer.sync_latency_tolerance_ms / 1000 * 3 + 0.02)
+        self.assert_timer_notify(
+            mock_super_send.call_args[0][0],
+            serial_number=XKNX_SERIAL_NUMBER,
+        )
+        mock_super_send.reset_mock()
+        # test response to invalid timer as timekeeper
+        timer_invalid = KNXIPFrame.init_from_body(
+            TimerNotify(
+                timer_value=0,
+                serial_number=bytes.fromhex("00 fa 12 34 56 78"),
+                message_tag=bytes.fromhex("12 34"),
+            )
+        )
+        secure_group.handle_knxipframe(timer_invalid, HPAI(*self.mock_addr))
+        assert secure_timer.timekeeper
+        assert secure_timer.sched_update
+        await time_travel(secure_timer.min_delay_time_keeper_update_notify - 0.01)
+        mock_super_send.assert_not_called()
+        await time_travel(secure_timer.sync_latency_tolerance_ms / 1000 * 1 + 0.02)
+        self.assert_timer_notify(
+            mock_super_send.call_args[0][0],
+            message_tag=bytes.fromhex("12 34"),
+            serial_number=bytes.fromhex("00 fa 12 34 56 78"),
+        )
+        # stop
+        secure_group.stop()
+        assert secure_timer._notify_timer_handle is None
+
+    async def test_synchronize(
+        self,
+        mock_super_send,
+        mock_super_connect,
+        time_travel,
+    ):
+        """Test synchronisazion."""
+        secure_group = SecureGroup(
+            local_addr=self.mock_addr,
+            remote_addr=(DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT),
+            backbone_key=self.mock_backbone_key,
+            latency_ms=1000,
+        )
+        secure_timer = secure_group.secure_timer
+        connect_task = asyncio.create_task(secure_group.connect())
+        await time_travel(0)
+        mock_super_connect.assert_called_once()
+        _message_tag = self.assert_timer_notify(
+            mock_super_send.call_args[0][0],
+            serial_number=XKNX_SERIAL_NUMBER,
+        )
+        mock_super_send.reset_mock()
+        # incoming
+        timer_update = KNXIPFrame.init_from_body(
+            TimerNotify(
+                timer_value=ONE_HOUR_MS,
+                serial_number=XKNX_SERIAL_NUMBER,
+                message_tag=_message_tag,
+            )
+        )
+        secure_group.handle_knxipframe(timer_update, HPAI(*self.mock_addr))
+        await time_travel(0)
+        _leeway_for_ci = 50  # ms
+        assert (
+            ONE_HOUR_MS
+            <= secure_timer.current_timer_value()
+            <= ONE_HOUR_MS + _leeway_for_ci
+        )
+        assert connect_task.done()
+        # nothing sent until time_follower_periodic
+        assert not secure_timer.timekeeper
+        await time_travel(secure_timer.min_delay_time_follower_periodic_notify - 0.01)
+        mock_super_send.assert_not_called()
+        await time_travel(secure_timer.sync_latency_tolerance_ms / 1000 * 10 + 0.02)
+        self.assert_timer_notify(
+            mock_super_send.call_args[0][0],
+            serial_number=XKNX_SERIAL_NUMBER,
+        )
+        mock_super_send.reset_mock()
+        secure_group.stop()
+        assert secure_timer._notify_timer_handle is None
+
+    @patch("xknx.io.ip_secure.SecureSequenceTimer._notify_timer_expired")
+    @patch("xknx.io.ip_secure.SecureSequenceTimer._monotonic_ms")
+    async def test_received_timer_notify(
+        self,
+        mock_monotonic_ms,
+        _mock_notify_timer_expired,  # we don't want to actually send here
+        mock_super_send,
+        mock_super_connect,
+        time_travel,
+    ):
+        """Test handling of received TimerNotify frames."""
+        mock_monotonic_ms.return_value = 0
+        secure_group = SecureGroup(
+            local_addr=self.mock_addr,
+            remote_addr=(DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT),
+            backbone_key=self.mock_backbone_key,
+            latency_ms=1000,
+        )
+        secure_timer = secure_group.secure_timer
+
+        with patch.object(
+            secure_timer, "reschedule", wraps=secure_timer.reschedule
+        ) as mock_reschedule:
+            # E1
+            timer_newer = KNXIPFrame.init_from_body(
+                TimerNotify(
+                    timer_value=ONE_HOUR_MS,
+                    serial_number=bytes.fromhex("00 fa 12 34 56 78"),
+                    message_tag=bytes.fromhex("12 34"),
+                )
+            )
+            secure_group.handle_knxipframe(timer_newer, HPAI(*self.mock_addr))
+            mock_reschedule.assert_called_once()
+            assert ONE_HOUR_MS == secure_timer.current_timer_value()
+            assert not secure_timer.timekeeper
+            assert not secure_timer.sched_update
+            mock_reschedule.reset_mock()
+            # E2
+            timer_exact = KNXIPFrame.init_from_body(
+                TimerNotify(
+                    timer_value=ONE_HOUR_MS,
+                    serial_number=bytes.fromhex("00 fa 12 34 56 78"),
+                    message_tag=bytes.fromhex("12 34"),
+                )
+            )
+            secure_group.handle_knxipframe(timer_exact, HPAI(*self.mock_addr))
+            mock_reschedule.assert_called_once()
+            assert ONE_HOUR_MS == secure_timer.current_timer_value()
+            assert not secure_timer.timekeeper
+            assert not secure_timer.sched_update
+            mock_reschedule.reset_mock()
+
+            timer_valid = KNXIPFrame.init_from_body(
+                TimerNotify(
+                    timer_value=ONE_HOUR_MS
+                    - secure_timer.sync_latency_tolerance_ms
+                    + 1,
+                    serial_number=bytes.fromhex("00 fa 12 34 56 78"),
+                    message_tag=bytes.fromhex("12 34"),
+                )
+            )
+            secure_group.handle_knxipframe(timer_valid, HPAI(*self.mock_addr))
+            mock_reschedule.assert_called_once()
+            assert ONE_HOUR_MS == secure_timer.current_timer_value()
+            assert not secure_timer.timekeeper
+            assert not secure_timer.sched_update
+            mock_reschedule.reset_mock()
+            # E3
+            timer_tolerable_1 = KNXIPFrame.init_from_body(
+                TimerNotify(
+                    timer_value=ONE_HOUR_MS - secure_timer.sync_latency_tolerance_ms,
+                    serial_number=bytes.fromhex("00 fa 12 34 56 78"),
+                    message_tag=bytes.fromhex("12 34"),
+                )
+            )
+            secure_group.handle_knxipframe(timer_tolerable_1, HPAI(*self.mock_addr))
+            mock_reschedule.assert_not_called()
+
+            timer_tolerable_2 = KNXIPFrame.init_from_body(
+                TimerNotify(
+                    timer_value=ONE_HOUR_MS - secure_timer.latency_tolerance_ms + 1,
+                    serial_number=bytes.fromhex("00 fa 12 34 56 78"),
+                    message_tag=bytes.fromhex("12 34"),
+                )
+            )
+            secure_group.handle_knxipframe(timer_tolerable_2, HPAI(*self.mock_addr))
+            mock_reschedule.assert_not_called()
+            # E4
+            timer_invalid = KNXIPFrame.init_from_body(
+                TimerNotify(
+                    timer_value=ONE_HOUR_MS - secure_timer.latency_tolerance_ms,
+                    serial_number=bytes.fromhex("00 fa 12 34 56 78"),
+                    message_tag=bytes.fromhex("12 34"),
+                )
+            )
+            secure_group.handle_knxipframe(timer_invalid, HPAI(*self.mock_addr))
+            mock_reschedule.assert_called_once_with(
+                update=(bytes.fromhex("12 34"), bytes.fromhex("00 fa 12 34 56 78"))
+            )
+            assert not secure_timer.timekeeper
+            assert secure_timer.sched_update
+            mock_reschedule.reset_mock()
+            # E4 from sched_update
+            secure_group.handle_knxipframe(timer_invalid, HPAI(*self.mock_addr))
+            mock_reschedule.assert_not_called()
+
+    @patch("xknx.io.ip_secure.SecureSequenceTimer._notify_timer_expired")
+    @patch("xknx.io.ip_secure.SecureSequenceTimer._monotonic_ms")
+    @patch("xknx.io.transport.udp_transport.UDPTransport.handle_knxipframe")
+    async def test_received_secure_wrapper(
+        self,
+        mock_super_handle_knxipframe,
+        mock_monotonic_ms,
+        _mock_notify_timer_expired,  # we don't want to actually send here
+        mock_super_send,
+        mock_super_connect,
+        time_travel,
+    ):
+        """Test handling of received TimerNotify frames."""
+        mock_monotonic_ms.return_value = 0
+        secure_group = SecureGroup(
+            local_addr=self.mock_addr,
+            remote_addr=(DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT),
+            backbone_key=self.mock_backbone_key,
+            latency_ms=1000,
+        )
+        secure_timer = secure_group.secure_timer
+        secure_timer.timer_authenticated = True
+
+        with patch.object(
+            secure_timer, "reschedule", wraps=secure_timer.reschedule
+        ) as mock_reschedule, patch.object(
+            secure_group, "decrypt_frame"
+        ) as mock_decrypt:
+            mock_decrypt.return_value = True  # we are only interested in timer value
+            # E5
+            wrapper_newer = KNXIPFrame.init_from_body(
+                SecureWrapper(
+                    sequence_information=ONE_HOUR_MS.to_bytes(6, "big"),
+                )
+            )
+
+            secure_group.handle_knxipframe(wrapper_newer, HPAI(*self.mock_addr))
+            mock_super_handle_knxipframe.assert_called_once()
+            mock_super_handle_knxipframe.reset_mock()
+            mock_reschedule.assert_called_once()
+            assert ONE_HOUR_MS == secure_timer.current_timer_value()
+            assert not secure_timer.sched_update
+            mock_reschedule.reset_mock()
+            # E6
+            wrapper_exact = KNXIPFrame.init_from_body(
+                SecureWrapper(sequence_information=ONE_HOUR_MS.to_bytes(6, "big"))
+            )
+
+            secure_group.handle_knxipframe(wrapper_exact, HPAI(*self.mock_addr))
+            mock_super_handle_knxipframe.assert_called_once()
+            mock_super_handle_knxipframe.reset_mock()
+            mock_reschedule.assert_called_once()
+            assert ONE_HOUR_MS == secure_timer.current_timer_value()
+            assert not secure_timer.timekeeper
+            assert not secure_timer.sched_update
+            mock_reschedule.reset_mock()
+
+            wrapper_valid = KNXIPFrame.init_from_body(
+                SecureWrapper(
+                    sequence_information=(
+                        ONE_HOUR_MS - secure_timer.sync_latency_tolerance_ms + 1
+                    ).to_bytes(6, "big")
+                )
+            )
+
+            secure_group.handle_knxipframe(wrapper_valid, HPAI(*self.mock_addr))
+            mock_super_handle_knxipframe.assert_called_once()
+            mock_super_handle_knxipframe.reset_mock()
+            mock_reschedule.assert_called_once()
+            assert ONE_HOUR_MS == secure_timer.current_timer_value()
+            assert not secure_timer.timekeeper
+            assert not secure_timer.sched_update
+            mock_reschedule.reset_mock()
+            # E7
+            wrapper_tolerable_1 = KNXIPFrame.init_from_body(
+                SecureWrapper(
+                    sequence_information=(
+                        ONE_HOUR_MS - secure_timer.sync_latency_tolerance_ms
+                    ).to_bytes(6, "big"),
+                )
+            )
+
+            secure_group.handle_knxipframe(wrapper_tolerable_1, HPAI(*self.mock_addr))
+            mock_super_handle_knxipframe.assert_called_once()
+            mock_super_handle_knxipframe.reset_mock()
+            mock_reschedule.assert_not_called()
+
+            wrapper_tolerable_2 = KNXIPFrame.init_from_body(
+                SecureWrapper(
+                    sequence_information=(
+                        ONE_HOUR_MS - secure_timer.latency_tolerance_ms + 1
+                    ).to_bytes(6, "big"),
+                    serial_number=bytes.fromhex("00 fa 12 34 56 78"),
+                    message_tag=bytes.fromhex("12 34"),
+                )
+            )
+
+            secure_group.handle_knxipframe(wrapper_tolerable_2, HPAI(*self.mock_addr))
+            mock_super_handle_knxipframe.assert_called_once()
+            mock_super_handle_knxipframe.reset_mock()
+            mock_reschedule.assert_not_called()
+            # E8
+            wrapper_invalid = KNXIPFrame.init_from_body(
+                SecureWrapper(
+                    sequence_information=(
+                        ONE_HOUR_MS - secure_timer.latency_tolerance_ms
+                    ).to_bytes(6, "big"),
+                    serial_number=bytes.fromhex("00 fa 12 34 56 78"),
+                    message_tag=bytes.fromhex("12 34"),
+                )
+            )
+
+            secure_group.handle_knxipframe(wrapper_invalid, HPAI(*self.mock_addr))
+            mock_reschedule.assert_called_once_with(
+                update=(bytes.fromhex("12 34"), bytes.fromhex("00 fa 12 34 56 78"))
+            )
+            assert not secure_timer.timekeeper
+            assert secure_timer.sched_update
+            mock_reschedule.reset_mock()
+            # E8 from sched_update
+            secure_group.handle_knxipframe(wrapper_invalid, HPAI(*self.mock_addr))
+            mock_super_handle_knxipframe.assert_not_called()
+            mock_reschedule.assert_not_called()
+
+    @patch("xknx.io.ip_secure.SecureSequenceTimer._notify_timer_expired")
+    @patch("xknx.io.ip_secure.SecureSequenceTimer._monotonic_ms")
+    @patch("xknx.io.transport.udp_transport.UDPTransport.handle_knxipframe")
+    async def test_send_secure_wrapper(
+        self,
+        mock_super_handle_knxipframe,
+        mock_monotonic_ms,
+        _mock_notify_timer_expired,  # we don't want to actually send here
+        mock_super_send,
+        mock_super_connect,
+        time_travel,
+    ):
+        """Test handling of received TimerNotify frames."""
+        mock_monotonic_ms.return_value = 0
+        secure_group = SecureGroup(
+            local_addr=self.mock_addr,
+            remote_addr=(DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT),
+            backbone_key=self.mock_backbone_key,
+            latency_ms=1000,
+        )
+        secure_timer = secure_group.secure_timer
+        secure_timer.timer_authenticated = True
+
+        test_cemi = CEMIFrame.init_from_telegram(
+            Telegram(
+                destination_address=GroupAddress("1/2/3"),
+                payload=apci.GroupValueRead(),
+            )
+        )
+
+        with patch.object(
+            secure_timer, "reschedule", wraps=secure_timer.reschedule
+        ) as mock_reschedule:
+            assert not secure_timer.sched_update
+            secure_group.send(KNXIPFrame.init_from_body(RoutingIndication(test_cemi)))
+            mock_reschedule.assert_called_once()
+            mock_reschedule.reset_mock()
+            # no reschedule when sched_update is true
+            secure_timer.reschedule(update=(bytes(2), bytes(6)))
+            assert secure_timer.sched_update
+            mock_reschedule.reset_mock()
+            secure_group.send(KNXIPFrame.init_from_body(RoutingIndication(test_cemi)))
+            mock_reschedule.assert_not_called()
diff --git a/test/io_tests/secure_session_test.py b/test/io_tests/secure_session_test.py
index b92b3876..0e10e3d5 100644
--- a/test/io_tests/secure_session_test.py
+++ b/test/io_tests/secure_session_test.py
@@ -7,7 +7,7 @@ import pytest
 
 from xknx.exceptions import CommunicationError, CouldNotParseKNXIP
 from xknx.io.const import SESSION_KEEPALIVE_RATE
-from xknx.io.secure_session import SecureSession
+from xknx.io.ip_secure import SecureSession
 from xknx.knxip import (
     HPAI,
     KNXIPFrame,
@@ -48,11 +48,11 @@ class TestSecureSession:
         """Set up test class."""
         # pylint: disable=attribute-defined-outside-init
         self.patch_serial_number = patch(
-            "xknx.io.secure_session.XKNX_SERIAL_NUMBER", self.mock_serial_number
+            "xknx.io.ip_secure.XKNX_SERIAL_NUMBER", self.mock_serial_number
         )
         self.patch_serial_number.start()
         self.patch_message_tag = patch(
-            "xknx.io.secure_session.MESSAGE_TAG_TUNNELLING", self.mock_message_tag
+            "xknx.io.ip_secure.MESSAGE_TAG_TUNNELLING", self.mock_message_tag
         )
         self.patch_message_tag.start()
 
@@ -72,7 +72,7 @@ class TestSecureSession:
     @patch("xknx.io.transport.tcp_transport.TCPTransport.connect")
     @patch("xknx.io.transport.tcp_transport.TCPTransport.send")
     @patch(
-        "xknx.io.secure_session.generate_ecdh_key_pair",
+        "xknx.io.ip_secure.generate_ecdh_key_pair",
         return_value=(mock_private_key, mock_public_key),
     )
     async def test_lifecycle(
@@ -193,7 +193,7 @@ class TestSecureSession:
     @patch("xknx.io.transport.tcp_transport.TCPTransport.connect")
     @patch("xknx.io.transport.tcp_transport.TCPTransport.send")
     @patch(
-        "xknx.io.secure_session.generate_ecdh_key_pair",
+        "xknx.io.ip_secure.generate_ecdh_key_pair",
         return_value=(mock_private_key, mock_public_key),
     )
     async def test_invalid_frames(
@@ -271,7 +271,7 @@ class TestSecureSession:
     @patch("xknx.io.transport.tcp_transport.TCPTransport.connect")
     @patch("xknx.io.transport.tcp_transport.TCPTransport.send")
     @patch(
-        "xknx.io.secure_session.generate_ecdh_key_pair",
+        "xknx.io.ip_secure.generate_ecdh_key_pair",
         return_value=(mock_private_key, mock_public_key),
     )
     async def test_invalid_session_response(
@@ -304,7 +304,7 @@ class TestSecureSession:
     @patch("xknx.io.transport.tcp_transport.TCPTransport.connect")
     @patch("xknx.io.transport.tcp_transport.TCPTransport.send")
     @patch(
-        "xknx.io.secure_session.generate_ecdh_key_pair",
+        "xknx.io.ip_secure.generate_ecdh_key_pair",
         return_value=(mock_private_key, mock_public_key),
     )
     async def test_no_authentication(
@@ -374,7 +374,7 @@ class TestSecureSession:
     @patch("xknx.io.transport.tcp_transport.TCPTransport.connect")
     @patch("xknx.io.transport.tcp_transport.TCPTransport.send")
     @patch(
-        "xknx.io.secure_session.generate_ecdh_key_pair",
+        "xknx.io.ip_secure.generate_ecdh_key_pair",
         return_value=(mock_private_key, mock_public_key),
     )
     async def test_invalid_authentication(