Repository: XKNX/xknx (python)

Issue:
Handle different Tunneling data and control endpoints correctly
KNX tunneling servers can have different data and control endpoints (eg. Calimero does).
`DisconnectRequest`s should be sent to the other device’s control endpoint instead of to the data endpoint. (see KNX specification 03_08_04 Tunneling §2.6 Frame confirmation)

I have not found an other usecase for the control endpoint for us yet.

Pull request description:
Handle separate Tunneling control and data endpoints
<!--
  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.
-->
Handle different Tunneling data and control endpoints

Closes #816
may solve https://github.com/calimero-project/calimero-server/issues/12 cc @aschamberger 
## 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:

- [ ] I have performed a self-review of my own code
- [ ] The documentation has been adjusted accordingly
- [ ] The changes generate no new warnings
- [ ] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog
- [ ] The Homeassistant plugin has been adjusted in case of new config options


Patch:
diff --git a/changelog.md b/changelog.md
index e312e426..35e3c87c 100644
--- a/changelog.md
+++ b/changelog.md
@@ -4,7 +4,8 @@
 
 ### Connection
 
-- Fix rate limiter wait time. Don't add time waiting for ACK or L_DATA.con frames to the rate_limit.
+- Handle separate Tunneling control and data endpoints
+- Fix rate limiter wait time: don't add time waiting for ACK or L_DATA.con frames to the rate_limit
 
 ## Internals
 
diff --git a/xknx/io/request_response/connect.py b/xknx/io/request_response/connect.py
index ec97d58a..b5ae8696 100644
--- a/xknx/io/request_response/connect.py
+++ b/xknx/io/request_response/connect.py
@@ -27,6 +27,7 @@ class Connect(RequestResponse):
         self.route_back = route_back
         super().__init__(xknx, self.udp_client, ConnectResponse)
         self.communication_channel = 0
+        self.data_endpoint = HPAI()
         self.identifier = 0
 
     def create_knxipframe(self) -> KNXIPFrame:
@@ -48,6 +49,6 @@ class Connect(RequestResponse):
     def on_success_hook(self, knxipframe: KNXIPFrame) -> None:
         """Set communication channel and identifier after having received a valid answer."""
         assert isinstance(knxipframe.body, ConnectResponse)
-        assert isinstance(knxipframe.body.identifier, int)
         self.communication_channel = knxipframe.body.communication_channel
+        self.data_endpoint = knxipframe.body.data_endpoint
         self.identifier = knxipframe.body.identifier
diff --git a/xknx/io/request_response/tunnelling.py b/xknx/io/request_response/tunnelling.py
index 6a295bea..f7fc0e60 100644
--- a/xknx/io/request_response/tunnelling.py
+++ b/xknx/io/request_response/tunnelling.py
@@ -26,6 +26,7 @@ class Tunnelling(RequestResponse):
         self,
         xknx: XKNX,
         udp_client: UDPClient,
+        data_endpoint: tuple[str, int] | None,
         telegram: Telegram,
         src_address: IndividualAddress,
         sequence_counter: int,
@@ -34,6 +35,7 @@ class Tunnelling(RequestResponse):
         """Initialize Tunnelling class."""
         self.xknx = xknx
         self.udp_client = udp_client
+        self.data_endpoint_addr = data_endpoint
         self.src_address = src_address
 
         super().__init__(xknx, self.udp_client, TunnellingAck)
@@ -42,6 +44,10 @@ class Tunnelling(RequestResponse):
         self.sequence_counter = sequence_counter
         self.communication_channel_id = communication_channel_id
 
+    async def send_request(self) -> None:
+        """Build knxipframe (within derived class) and send via UDP."""
+        self.udpclient.send(self.create_knxipframe(), addr=self.data_endpoint_addr)
+
     def create_knxipframe(self) -> KNXIPFrame:
         """Create KNX/IP Frame object to be sent to device."""
         cemi = CEMIFrame.init_from_telegram(
diff --git a/xknx/io/tunnel.py b/xknx/io/tunnel.py
index 65752e0a..cd555a13 100644
--- a/xknx/io/tunnel.py
+++ b/xknx/io/tunnel.py
@@ -69,6 +69,7 @@ class Tunnel(Interface):
         self.init_udp_client()
 
         self._src_address = xknx.own_address
+        self._data_endpoint_addr: tuple[str, int] | None = None
         self.sequence_number = 0
         self.communication_channel: int | None = None
         self.number_heartbeat_failed = 0
@@ -144,6 +145,7 @@ class Tunnel(Interface):
             )
         )
         self.stop_heartbeat()
+        self._data_endpoint_addr = None
         if self.auto_reconnect:
             self._reconnect_task = asyncio.create_task(self._reconnect())
         else:
@@ -172,6 +174,7 @@ class Tunnel(Interface):
             XknxConnectionState.DISCONNECTED
         )
         self.stop_heartbeat()
+        self._data_endpoint_addr = None
         self._stop_reconnect()
         await self._disconnect_request(False)
         await self.udp_client.stop()
@@ -188,6 +191,11 @@ class Tunnel(Interface):
         await connect.start()
         if connect.success:
             self.communication_channel = connect.communication_channel
+            # assign data_endpoint received from server
+            self._data_endpoint_addr = (
+                connect.data_endpoint.ip_addr,
+                connect.data_endpoint.port,
+            )
             # Use the individual address provided by the tunnelling server
             self._src_address = IndividualAddress(connect.identifier)
             self.xknx.current_address = self._src_address
@@ -274,6 +282,7 @@ class Tunnel(Interface):
         tunnelling = Tunnelling(
             self.xknx,
             self.udp_client,
+            self._data_endpoint_addr,
             telegram,
             self._src_address,
             self.sequence_number,
@@ -355,7 +364,9 @@ class Tunnel(Interface):
             communication_channel_id=communication_channel_id,
             sequence_counter=sequence_counter,
         )
-        self.udp_client.send(KNXIPFrame.init_from_body(ack))
+        self.udp_client.send(
+            KNXIPFrame.init_from_body(ack), addr=self._data_endpoint_addr
+        )
 
     def _disconnect_request_received(
         self, disconnect_request: DisconnectRequest
diff --git a/xknx/io/udp_client.py b/xknx/io/udp_client.py
index 2637a73f..1aeee391 100644
--- a/xknx/io/udp_client.py
+++ b/xknx/io/udp_client.py
@@ -116,7 +116,7 @@ class UDPClient:
                 )
             else:
                 knx_logger.debug(
-                    "Received from %s:%s at %s:\n%s",
+                    "Received from %s:%s at %s:\n %s",
                     source[0],
                     source[1],
                     time.time(),
@@ -213,27 +213,28 @@ class UDPClient:
             (transport, _) = await loop.create_datagram_endpoint(
                 lambda: udp_client_factory,
                 local_addr=self.local_addr,
-                remote_addr=self.remote_addr,
             )
             # TODO: typing - remove cast - loop.create_datagram_endpoint should return a DatagramTransport
             self.transport = cast(asyncio.DatagramTransport, transport)
 
-    def send(self, knxipframe: KNXIPFrame) -> None:
+    def send(self, knxipframe: KNXIPFrame, addr: tuple[str, int] | None = None) -> None:
         """Send KNXIPFrame to socket."""
+        _addr = addr or self.remote_addr
         knx_logger.debug(
-            "Sending to %s:%s at %s:\n%s",
-            self.remote_addr[0],
-            self.remote_addr[1],
-            time.time(),
-            knxipframe,
+            "Sending to %s:%s at %s:\n %s", _addr[0], _addr[1], time.time(), knxipframe
         )
         if self.transport is None:
             raise XKNXException("Transport not connected")
 
         if self.multicast:
+            if addr is not None:
+                logger.warning(
+                    "Multicast send to specific address is invalid. %s",
+                    knxipframe,
+                )
             self.transport.sendto(bytes(knxipframe.to_knx()), self.remote_addr)
         else:
-            self.transport.sendto(bytes(knxipframe.to_knx()))
+            self.transport.sendto(bytes(knxipframe.to_knx()), addr=_addr)
 
     def getsockname(self) -> tuple[str, int]:
         """Return socket IP and port."""
diff --git a/xknx/knxip/connect_response.py b/xknx/knxip/connect_response.py
index 3f9c3835..865923ef 100644
--- a/xknx/knxip/connect_response.py
+++ b/xknx/knxip/connect_response.py
@@ -32,8 +32,8 @@ class ConnectResponse(KNXIPBodyResponse):
         communication_channel: int = 0,
         status_code: ErrorCode = ErrorCode.E_NO_ERROR,
         request_type: ConnectRequestType = ConnectRequestType.TUNNEL_CONNECTION,
-        control_endpoint: HPAI = HPAI(),
-        identifier: int | None = None,
+        data_endpoint: HPAI = HPAI(),
+        identifier: int = 0,
     ):
         """Initialize ConnectResponse class."""
         super().__init__(xknx)
@@ -41,7 +41,7 @@ class ConnectResponse(KNXIPBodyResponse):
         self.communication_channel = communication_channel
         self.status_code = status_code
         self.request_type = request_type
-        self.control_endpoint = control_endpoint
+        self.data_endpoint = data_endpoint
         # identifier shall contain KNX Individual Address assigned to this KNXnet/IP Tunnelling connection
         self.identifier = identifier
 
@@ -67,11 +67,11 @@ class ConnectResponse(KNXIPBodyResponse):
         pos = 2
 
         if self.status_code == ErrorCode.E_NO_ERROR:
-            pos += self.control_endpoint.from_knx(raw[pos:])
+            pos += self.data_endpoint.from_knx(raw[pos:])
             pos += crd_from_knx(raw[pos:])
         else:
             # do not parse HPAI and CRD in case of errors - just check length
-            pos += len(raw[pos:])
+            pos = len(raw)
         return pos
 
     def to_knx(self) -> list[int]:
@@ -91,7 +91,7 @@ class ConnectResponse(KNXIPBodyResponse):
         data = []
         data.append(self.communication_channel)
         data.append(self.status_code.value)
-        data.extend(self.control_endpoint.to_knx())
+        data.extend(self.data_endpoint.to_knx())
         data.extend(crd_to_knx())
 
         return data
@@ -102,7 +102,7 @@ class ConnectResponse(KNXIPBodyResponse):
             "<ConnectResponse "
             f'communication_channel="{self.communication_channel}" '
             f'status_code="{self.status_code}" '
-            f'control_endpoint="{self.control_endpoint}" '
+            f'data_endpoint="{self.data_endpoint}" '
             f'request_type="{self.request_type}" '
             f'identifier="{self.identifier}" />'
         )
diff --git a/xknx/knxip/knxip.py b/xknx/knxip/knxip.py
index b5425a86..ed333e0a 100644
--- a/xknx/knxip/knxip.py
+++ b/xknx/knxip/knxip.py
@@ -101,7 +101,7 @@ class KNXIPFrame:
         data.extend(self.body.to_knx())
         return data
 
-    def __str__(self) -> str:
+    def __repr__(self) -> str:
         """Return object as readable string."""
         return f'<KNXIPFrame {self.header}\n body="{self.body}" />'
 


Test patch:
diff --git a/test/io_tests/tunnel_test.py b/test/io_tests/tunnel_test.py
index 3615b9cd..d9d8aa48 100644
--- a/test/io_tests/tunnel_test.py
+++ b/test/io_tests/tunnel_test.py
@@ -1,14 +1,24 @@
 """Unit test for KNX/IP Tunnelling Request/Response."""
 import asyncio
-from unittest.mock import Mock, patch
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 from xknx import XKNX
 from xknx.dpt import DPTArray
 from xknx.io import Tunnel
-from xknx.knxip import HPAI, CEMIFrame, KNXIPFrame, TunnellingAck, TunnellingRequest
+from xknx.knxip import (
+    HPAI,
+    CEMIFrame,
+    ConnectRequest,
+    ConnectResponse,
+    DisconnectRequest,
+    DisconnectResponse,
+    KNXIPFrame,
+    TunnellingAck,
+    TunnellingRequest,
+)
 from xknx.knxip.knxip_enum import CEMIMessageCode
-from xknx.telegram import Telegram, TelegramDirection
+from xknx.telegram import IndividualAddress, Telegram, TelegramDirection
 from xknx.telegram.apci import GroupValueWrite
 
 
@@ -103,3 +113,88 @@ class TestTunnel:
         # one call for the outgoing request and one for the ACK for the confirmation
         assert self.tunnel.udp_client.send.call_count == 2
         await task
+
+    async def test_tunnel_connect_send_disconnect(self, time_travel):
+        """Test initiating a tunnelling connection."""
+        local_addr = ("192.168.1.1", 12345)
+        gateway_control_addr = ("192.168.1.2", 3671)
+        gateway_data_addr = ("192.168.1.2", 56789)
+        self.tunnel.udp_client.connect = AsyncMock()
+        self.tunnel.udp_client.getsockname = Mock(return_value=local_addr)
+        self.tunnel.udp_client.send = Mock()
+        self.tunnel.udp_client.stop = AsyncMock()
+
+        # Connect
+        connect_request = ConnectRequest(
+            self.xknx,
+            control_endpoint=HPAI(*local_addr),
+            data_endpoint=HPAI(*local_addr),
+        )
+        connect_frame = KNXIPFrame.init_from_body(connect_request)
+
+        connection_task = asyncio.create_task(self.tunnel.connect())
+        await time_travel(0)
+        self.tunnel.udp_client.connect.assert_called_once()
+        self.tunnel.udp_client.send.assert_called_once_with(connect_frame)
+
+        connect_response_frame = KNXIPFrame.init_from_body(
+            ConnectResponse(
+                self.xknx,
+                communication_channel=23,
+                data_endpoint=HPAI(*gateway_data_addr),
+                identifier=7,
+            )
+        )
+        self.tunnel.udp_client.handle_knxipframe(
+            connect_response_frame, gateway_control_addr
+        )
+        await connection_task
+        assert self.tunnel._data_endpoint_addr == gateway_data_addr
+        assert self.tunnel._src_address == IndividualAddress(7)
+
+        # Send - use data endpoint
+        self.tunnel.udp_client.send.reset_mock()
+        test_telegram = Telegram(payload=GroupValueWrite(DPTArray((1,))))
+        test_telegram_frame = KNXIPFrame.init_from_body(
+            TunnellingRequest(
+                self.xknx,
+                communication_channel_id=23,
+                sequence_counter=0,
+                cemi=CEMIFrame.init_from_telegram(
+                    self.xknx,
+                    test_telegram,
+                    code=CEMIMessageCode.L_DATA_REQ,
+                    src_addr=IndividualAddress(7),
+                ),
+            )
+        )
+        asyncio.create_task(self.tunnel.send_telegram(test_telegram))
+        await time_travel(0)
+        self.tunnel.udp_client.send.assert_called_once_with(
+            test_telegram_frame, addr=gateway_data_addr
+        )
+        # skip ack and confirmation
+
+        # Disconnect
+        self.tunnel.udp_client.send.reset_mock()
+        disconnect_request = DisconnectRequest(
+            self.xknx, communication_channel_id=23, control_endpoint=HPAI(*local_addr)
+        )
+        disconnect_frame = KNXIPFrame.init_from_body(disconnect_request)
+
+        disconnection_task = asyncio.create_task(self.tunnel.disconnect())
+        await time_travel(0)
+        self.tunnel.udp_client.send.assert_called_once_with(disconnect_frame)
+
+        disconnect_response_frame = KNXIPFrame.init_from_body(
+            DisconnectResponse(
+                self.xknx,
+                communication_channel_id=23,
+            )
+        )
+        self.tunnel.udp_client.handle_knxipframe(
+            disconnect_response_frame, gateway_control_addr
+        )
+        await disconnection_task
+        assert self.tunnel._data_endpoint_addr is None
+        self.tunnel.udp_client.stop.assert_called_once()
diff --git a/test/io_tests/tunnelling_test.py b/test/io_tests/tunnelling_test.py
index bfb4142c..b893e3cc 100644
--- a/test/io_tests/tunnelling_test.py
+++ b/test/io_tests/tunnelling_test.py
@@ -26,6 +26,7 @@ class TestTunnelling:
         """Test tunnelling from KNX bus."""
         xknx = XKNX()
         communication_channel_id = 23
+        data_endpoint = ("192.168.1.2", 4567)
         udp_client = UDPClient(xknx, ("192.168.1.1", 0), ("192.168.1.2", 1234))
         telegram = Telegram(
             destination_address=GroupAddress("1/2/3"),
@@ -36,6 +37,7 @@ class TestTunnelling:
         tunnelling = Tunnelling(
             xknx,
             udp_client,
+            data_endpoint,
             telegram,
             src_address,
             sequence_counter,
@@ -60,7 +62,7 @@ class TestTunnelling:
         ) as mock_udp_getsockname:
             mock_udp_getsockname.return_value = ("192.168.1.3", 4321)
             await tunnelling.start()
-            mock_udp_send.assert_called_with(exp_knxipframe)
+            mock_udp_send.assert_called_with(exp_knxipframe, addr=data_endpoint)
 
         # Response KNX/IP-Frame with wrong type
         wrong_knxipframe = KNXIPFrame(xknx)
diff --git a/test/knxip_tests/connect_response_test.py b/test/knxip_tests/connect_response_test.py
index 71f61a4c..3bd85054 100644
--- a/test/knxip_tests/connect_response_test.py
+++ b/test/knxip_tests/connect_response_test.py
@@ -38,9 +38,7 @@ class TestKNXIPConnectResponse:
         assert isinstance(knxipframe.body, ConnectResponse)
         assert knxipframe.body.communication_channel == 1
         assert knxipframe.body.status_code == ErrorCode.E_NO_ERROR
-        assert knxipframe.body.control_endpoint == HPAI(
-            ip_addr="192.168.42.10", port=3671
-        )
+        assert knxipframe.body.data_endpoint == HPAI(ip_addr="192.168.42.10", port=3671)
         assert knxipframe.body.request_type == ConnectRequestType.TUNNEL_CONNECTION
         assert knxipframe.body.identifier == 4607
 
@@ -49,7 +47,7 @@ class TestKNXIPConnectResponse:
             communication_channel=1,
             status_code=ErrorCode.E_NO_ERROR,
             request_type=ConnectRequestType.TUNNEL_CONNECTION,
-            control_endpoint=HPAI(ip_addr="192.168.42.10", port=3671),
+            data_endpoint=HPAI(ip_addr="192.168.42.10", port=3671),
             identifier=4607,
         )
         knxipframe2 = KNXIPFrame.init_from_body(connect_response)
@@ -151,7 +149,7 @@ class TestKNXIPConnectResponse:
             communication_channel=192,
             status_code=ErrorCode.E_NO_MORE_CONNECTIONS,
             request_type=ConnectRequestType.TUNNEL_CONNECTION,
-            control_endpoint=HPAI(ip_addr="10.1.0.41", port=3671),
+            data_endpoint=HPAI(ip_addr="10.1.0.41", port=3671),
             identifier=0,
         )
         knxipframe2 = KNXIPFrame.init_from_body(connect_response)
diff --git a/test/str_test.py b/test/str_test.py
index 72c9ef56..3d34719f 100644
--- a/test/str_test.py
+++ b/test/str_test.py
@@ -526,12 +526,12 @@ class TestStringRepresentations:
         connect_response = ConnectResponse(xknx)
         connect_response.communication_channel = 13
         connect_response.request_type = ConnectRequestType.TUNNEL_CONNECTION
-        connect_response.control_endpoint = HPAI(ip_addr="192.168.42.1", port=33941)
+        connect_response.data_endpoint = HPAI(ip_addr="192.168.42.1", port=33941)
         connect_response.identifier = 42
         assert (
             str(connect_response)
             == '<ConnectResponse communication_channel="13" status_code="ErrorCode.E_NO_ERROR" '
-            'control_endpoint="192.168.42.1:33941" '
+            'data_endpoint="192.168.42.1:33941" '
             'request_type="ConnectRequestType.TUNNEL_CONNECTION" identifier="42" />'
         )