Repository: XKNX/xknx
Pull request #1133 (2023-01-05 22:29:58)

Revert handling incoming L_Data.req frames
for incoming management requests. 
This was probably wrongly assumed to be a thing. 

I don't recall if I have read tcpdumps wrongly or had a faulty device... can't reproduce this anymore and the specs say that this is not a thing.

see #972 

<!--
  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.
-->


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.
-->

- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] 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
- [x] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog (docs/changelog.md)

codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/1133?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#1133](https://codecov.io/gh/XKNX/xknx/pull/1133?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (93b6080) into [main](https://codecov.io/gh/XKNX/xknx/commit/922e9cc93dfe00e2d3020a8b32c0782e0393cbba?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (922e9cc) will **decrease** coverage by `0.01%`.
> The diff coverage is `80.00%`.

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

```diff
@@            Coverage Diff             @@
##             main    #1133      +/-   ##
==========================================
- Coverage   96.35%   96.33%   -0.02%     
==========================================
  Files         138      138              
  Lines        8659     8645      -14     
==========================================
- Hits         8343     8328      -15     
- Misses        316      317       +1     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/1133?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/1133?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby90dW5uZWwucHk=) | `87.68% <50.00%> (-0.70%)` | :arrow_down: |
| [xknx/io/knxip\_interface.py](https://codecov.io/gh/XKNX/xknx/pull/1133?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9rbnhpcF9pbnRlcmZhY2UucHk=) | `84.76% <76.92%> (+0.47%)` | :arrow_up: |
| [xknx/management/management.py](https://codecov.io/gh/XKNX/xknx/pull/1133?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9tYW5hZ2VtZW50L21hbmFnZW1lbnQucHk=) | `88.19% <85.71%> (-0.08%)` | :arrow_down: |
| [xknx/io/interface.py](https://codecov.io/gh/XKNX/xknx/pull/1133?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9pbnRlcmZhY2UucHk=) | `100.00% <100.00%> (ø)` | |
| [xknx/io/routing.py](https://codecov.io/gh/XKNX/xknx/pull/1133?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9yb3V0aW5nLnB5) | `91.66% <100.00%> (-0.25%)` | :arrow_down: |
| [xknx/telegram/telegram.py](https://codecov.io/gh/XKNX/xknx/pull/1133?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC90ZWxlZ3JhbS90ZWxlZ3JhbS5weQ==) | `93.02% <0.00%> (-2.33%)` | :arrow_down: |


Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index d8491b85..f571dad4 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -12,6 +12,7 @@ nav_order: 2
 
 - Parse T_Data_Broadcast TPCI. Forward these telegrams to the Management class.
 - KNXIPHeader total_length is 2 bytes long. There are no reserved bytes.
+- Revert handling L_Data.req frames for incoming device management requests.
 
 # 2.2.0 Expose cooldown 2022-12-27
 
diff --git a/xknx/io/interface.py b/xknx/io/interface.py
index 020830a1..856b8601 100644
--- a/xknx/io/interface.py
+++ b/xknx/io/interface.py
@@ -8,12 +8,11 @@ Abstract base for a specific KNX/IP connection (Tunneling or Routing).
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from collections.abc import Awaitable
-from typing import Callable, Optional
+from typing import Callable
 
 from xknx.telegram import Telegram
 
-TelegramCallbackType = Callable[[Telegram], Awaitable[Optional[list[Telegram]]]]
+TelegramCallbackType = Callable[[Telegram], None]
 
 
 class Interface(ABC):
diff --git a/xknx/io/knxip_interface.py b/xknx/io/knxip_interface.py
index 5e9d67e1..778ca521 100644
--- a/xknx/io/knxip_interface.py
+++ b/xknx/io/knxip_interface.py
@@ -431,15 +431,15 @@ class KNXIPInterface:
             await self._interface.disconnect()
             self._interface = None
 
-    async def telegram_received(self, telegram: Telegram) -> list[Telegram] | None:
+    def telegram_received(self, telegram: Telegram) -> None:
         """Put received telegram into queue. Callback for having received telegram."""
         if isinstance(telegram.tpci, TDataGroup):
             self.xknx.telegrams.put_nowait(telegram)
-            return None
+            return
         if isinstance(telegram.destination_address, IndividualAddress):
             if telegram.destination_address != self.xknx.current_address:
-                return None
-        return self.xknx.management.process(telegram)
+                return
+        self.xknx.management.process(telegram)
 
     async def send_telegram(self, telegram: Telegram) -> None:
         """Send telegram to connected device (either Tunneling or Routing)."""
@@ -486,13 +486,9 @@ class KNXIPInterfaceThreaded(KNXIPInterface):
         loop_loaded.set()
         self._thread_loop.run_forever()
 
-    async def _await_from_different_thread(
-        self,
-        coro: Awaitable[T],
-        target_loop: asyncio.AbstractEventLoop,
-    ) -> T:
+    async def _await_from_connection_thread(self, coro: Awaitable[T]) -> T:
         """Await coroutine in different thread."""
-        fut = asyncio.run_coroutine_threadsafe(coro, target_loop)
+        fut = asyncio.run_coroutine_threadsafe(coro, self._thread_loop)
         finished = threading.Event()
 
         def fut_finished_cb(_: concurrent.futures.Future[T]) -> None:
@@ -500,13 +496,13 @@ class KNXIPInterfaceThreaded(KNXIPInterface):
             finished.set()
 
         fut.add_done_callback(fut_finished_cb)
-        # wait on that event in an executor, yielding control to current loop
-        await asyncio.get_running_loop().run_in_executor(None, finished.wait)
+        # wait on that event in an executor, yielding control to _main_loop
+        await self._main_loop.run_in_executor(None, finished.wait)
         return fut.result()
 
     async def start(self) -> None:
         """Start KNX/IP interface."""
-        return await self._await_from_different_thread(self._start(), self._thread_loop)
+        return await self._await_from_connection_thread(self._start())
 
     async def stop(self) -> None:
         """
@@ -515,26 +511,22 @@ class KNXIPInterfaceThreaded(KNXIPInterface):
         Can not be restarted, create a new instance instead.
         """
         if self._interface is not None:
-            await self._await_from_different_thread(
-                self._interface.disconnect(), self._thread_loop
-            )
+            await self._await_from_connection_thread(self._interface.disconnect())
             self._interface = None
         self._thread_loop.call_soon_threadsafe(self._thread_loop.stop)
         self.connection_thread.join()
 
-    async def telegram_received(self, telegram: Telegram) -> list[Telegram] | None:
+    def telegram_received(self, telegram: Telegram) -> None:
         """Put received telegram into queue. Callback for having received telegram."""
-        return await self._await_from_different_thread(
-            super().telegram_received(telegram), self._main_loop
-        )
+        self._main_loop.call_soon_threadsafe(super().telegram_received, telegram)
 
     async def send_telegram(self, telegram: Telegram) -> None:
         """Send telegram to connected device (either Tunneling or Routing)."""
         if self._interface is None:
             raise CommunicationError("KNX/IP interface not connected")
 
-        return await self._await_from_different_thread(
-            self._interface.send_telegram(telegram), self._thread_loop
+        return await self._await_from_connection_thread(
+            self._interface.send_telegram(telegram)
         )
 
     async def gateway_info(self) -> GatewayDescriptor | None:
@@ -542,7 +534,7 @@ class KNXIPInterfaceThreaded(KNXIPInterface):
         if self._gateway_info is not None:
             return self._gateway_info
         if isinstance(self._interface, _Tunnel):
-            return await self._await_from_different_thread(
-                self._interface.request_description(), self._thread_loop
+            return await self._await_from_connection_thread(
+                self._interface.request_description()
             )
         return None
diff --git a/xknx/io/routing.py b/xknx/io/routing.py
index 936e9c47..0b09b48d 100644
--- a/xknx/io/routing.py
+++ b/xknx/io/routing.py
@@ -269,17 +269,9 @@ class Routing(Interface):
         if cemi.src_addr == self.individual_address:
             logger.debug("Ignoring own packet")
             return
-        # TODO: is cemi message code L_DATA.req or .con valid for routing? if not maybe warn and ignore
-        asyncio.create_task(self.handle_cemi_frame(cemi))
-
-    async def handle_cemi_frame(self, cemi: CEMIFrame) -> None:
-        """Handle incoming telegram and send responses if applicable (device management)."""
         telegram = cemi.telegram
         telegram.direction = TelegramDirection.INCOMING
-
-        if response_tgs := await self.telegram_received_callback(telegram):
-            for response in response_tgs:
-                await self.send_telegram(response)
+        self.telegram_received_callback(telegram)
 
 
 class SecureRouting(Routing):
diff --git a/xknx/io/tunnel.py b/xknx/io/tunnel.py
index e57c871d..7ef1a05b 100644
--- a/xknx/io/tunnel.py
+++ b/xknx/io/tunnel.py
@@ -7,7 +7,6 @@ from __future__ import annotations
 
 from abc import abstractmethod
 import asyncio
-from copy import copy
 import logging
 from typing import TYPE_CHECKING
 
@@ -365,33 +364,14 @@ class _Tunnel(Interface):
             # L_DATA_CON confirmation frame signals ready to send next telegram
             self._tunnelling_request_confirmation_event.set()
             return
-
         if cemi.code is CEMIMessageCode.L_DATA_REQ:
-            confirmation = copy(cemi)
-            confirmation.code = CEMIMessageCode.L_DATA_CON
-            asyncio.create_task(self._send_cemi(confirmation))
-
-        asyncio.create_task(self.handle_cemi_frame(cemi))
+            # L_DATA_REQ frames should only be outgoing.
+            logger.warning("Tunnel received unexpected L_DATA_REQ frame: %s", cemi)
+            return
 
-    async def handle_cemi_frame(self, cemi: CEMIFrame) -> None:
-        """Handle incoming telegram and send responses if applicable (device management)."""
         telegram = cemi.telegram
         telegram.direction = TelegramDirection.INCOMING
-
-        if response_tgs := await self.telegram_received_callback(telegram):
-            response_code = (
-                CEMIMessageCode.L_DATA_IND
-                if cemi.code is CEMIMessageCode.L_DATA_REQ
-                else CEMIMessageCode.L_DATA_REQ
-            )
-            for response in response_tgs:
-                await self._send_cemi(
-                    CEMIFrame.init_from_telegram(
-                        telegram=response,
-                        code=response_code,
-                        src_addr=self._src_address,
-                    )
-                )
+        self.telegram_received_callback(telegram)
 
     def _disconnect_request_received(
         self, disconnect_request: DisconnectRequest
diff --git a/xknx/management/management.py b/xknx/management/management.py
index 86ac6875..7e265a91 100644
--- a/xknx/management/management.py
+++ b/xknx/management/management.py
@@ -34,35 +34,34 @@ class Management:
         self.xknx = xknx
         self._connections: dict[IndividualAddress, P2PConnection] = {}
 
-    def process(self, telegram: Telegram) -> list[Telegram]:
+    def process(self, telegram: Telegram) -> None:
         """Process incoming telegrams."""
-        response = []
         if isinstance(telegram.tpci, TDataConnected):
             ack = Telegram(
                 destination_address=telegram.source_address,
                 tpci=TAck(sequence_number=telegram.tpci.sequence_number),
             )
-            response.append(ack)
+            asyncio.create_task(self.xknx.knxip_interface.send_telegram(ack))
         if conn := self._connections.get(telegram.source_address):
             conn.process(telegram)
-            return response
+            return
         if telegram.tpci.numbered:
             logger.warning(
                 "No active point-to-point connection for received telegram: %s",
                 telegram,
             )
-            return response
+            return
         if isinstance(telegram.tpci, TConnect):
             # refuse incoming connections
-            # TODO: handle incoming telegrams for connections not initiated by us, or connection-less
-            # TODO: maybe use outgoing telegram queue or new task to not stall the consumer task
+            # TODO: handle incoming telegrams for connections
+            # not initiated by us, connection-less and broadcast
             disconnect = Telegram(
                 destination_address=telegram.source_address, tpci=TDisconnect()
             )
-            response.append(disconnect)
-            return response
+            asyncio.create_task(self.xknx.knxip_interface.send_telegram(disconnect))
+            return
         logger.warning("Unhandled management telegram: %r", telegram)
-        return response
+        return
 
     async def connect(self, address: IndividualAddress) -> P2PConnection:
         """Open a point-to-point connection to a KNX device."""


Test patch:
diff --git a/requirements/testing.txt b/requirements/testing.txt
index 078978df..470d71c6 100644
--- a/requirements/testing.txt
+++ b/requirements/testing.txt
@@ -9,5 +9,5 @@ pytest==7.2.0
 pytest-asyncio==0.20.3
 pytest-cov==4.0.0
 setuptools==65.6.3
-tox==4.2.1
+tox==4.2.4
 mypy==0.991
diff --git a/test/io_tests/routing_test.py b/test/io_tests/routing_test.py
index e4628f48..9d5e218a 100644
--- a/test/io_tests/routing_test.py
+++ b/test/io_tests/routing_test.py
@@ -31,9 +31,13 @@ class TestRouting:
         routing = Routing(
             self.xknx,
             individual_address=None,
-            telegram_received_callback=self.tg_received_mock,
+            # telegram_received_callback=self.tg_received_mock,
+            telegram_received_callback=self.xknx.knxip_interface.telegram_received,
             local_ip="192.168.1.1",
         )
+        self.xknx.knxip_interface._interface = routing
+        # set current address so management telegram is processed
+        self.xknx.current_address = IndividualAddress("1.0.255")
         # L_Data.ind T_Connect from 1.0.250 to 1.0.255 (xknx tunnel endpoint)
         # communication_channel_id: 0x02   sequence_counter: 0x81
         raw_ind = bytes.fromhex("0610 0530 0010 2900b06010fa10ff0080")
@@ -54,12 +58,6 @@ class TestRouting:
             RoutingIndication(raw_cemi=response_cemi.to_knx())
         )
 
-        async def tg_received_mock(telegram):
-            """Mock for telegram_received_callback."""
-            assert telegram == test_telegram
-            return [response_telegram]
-
-        routing.telegram_received_callback = tg_received_mock
         routing.transport.data_received_callback(raw_ind, ("192.168.1.2", 3671))
         await asyncio.sleep(0)
         assert send_knxipframe_mock.call_args_list == [
diff --git a/test/io_tests/tunnel_test.py b/test/io_tests/tunnel_test.py
index deb59f78..9acf5fbb 100644
--- a/test/io_tests/tunnel_test.py
+++ b/test/io_tests/tunnel_test.py
@@ -28,7 +28,13 @@ from xknx.knxip import (
     TunnellingRequest,
 )
 from xknx.knxip.knxip_enum import CEMIMessageCode, HostProtocol
-from xknx.telegram import GroupAddress, IndividualAddress, Telegram, TelegramDirection
+from xknx.telegram import (
+    GroupAddress,
+    IndividualAddress,
+    Telegram,
+    TelegramDirection,
+    tpci,
+)
 from xknx.telegram.apci import GroupValueWrite
 
 
@@ -85,42 +91,38 @@ class TestUDPTunnel:
         send_ack_mock,
     ):
         """Test Tunnel for responding to L_DATA.req with confirmation and indication."""
-        # L_Data.req T_Connect from 1.0.250 to 1.0.255 (xknx tunnel endpoint) - ETS Line-Scan
+        self.tunnel.telegram_received_callback = (
+            self.xknx.knxip_interface.telegram_received
+        )
+        self.xknx.knxip_interface._interface = self.tunnel
+        # set current address so management telegram is processed
+        self.xknx.current_address = IndividualAddress("1.0.255")
+        # L_Data.ing T_Connect from 1.0.250 to 1.0.255 (xknx tunnel endpoint) - ETS Line-Scan
         # communication_channel_id: 0x02   sequence_counter: 0x81
-        raw_req = bytes.fromhex("0610 0420 0014 04 02 81 00 1100b06010fa10ff0080")
+        raw_ind = bytes.fromhex("0610 0420 0014 04 02 81 00 2900b06010fa10ff0080")
 
         _cemi = CEMIFrame()
-        _cemi.from_knx(raw_req[10:])
+        _cemi.from_knx(raw_ind[10:])
         test_telegram = _cemi.telegram
         test_telegram.direction = TelegramDirection.INCOMING
         self.tunnel.expected_sequence_number = 0x81
 
         response_telegram = Telegram(
-            destination_address=test_telegram.source_address,
-            source_address=self.tunnel._src_address,
+            destination_address=IndividualAddress(test_telegram.source_address),
+            tpci=tpci.TDisconnect(),
         )
 
-        async def tg_received_mock(telegram):
-            """Mock for telegram_received_callback."""
-            assert telegram == test_telegram
-            return [response_telegram]
-
-        self.tunnel.telegram_received_callback = tg_received_mock
-        self.tunnel.transport.data_received_callback(raw_req, ("192.168.1.2", 3671))
-
+        self.tunnel.transport.data_received_callback(raw_ind, ("192.168.1.2", 3671))
         await asyncio.sleep(0)
-        confirmation_cemi = _cemi
-        confirmation_cemi.code = CEMIMessageCode.L_DATA_CON
         response_cemi = CEMIFrame.init_from_telegram(
             response_telegram,
-            code=CEMIMessageCode.L_DATA_IND,
+            code=CEMIMessageCode.L_DATA_REQ,
             src_addr=self.tunnel._src_address,
         )
         assert send_cemi_mock.call_args_list == [
-            call(confirmation_cemi),
             call(response_cemi),
         ]
-        send_ack_mock.assert_called_once_with(raw_req[7], raw_req[8])
+        send_ack_mock.assert_called_once_with(raw_ind[7], raw_ind[8])
 
     async def test_repeated_tunnel_request(self, time_travel):
         """Test Tunnel for receiving repeated TunnellingRequest frames."""
diff --git a/test/management_tests/management_test.py b/test/management_tests/management_test.py
index 9a5457ce..fbcc01d6 100644
--- a/test/management_tests/management_test.py
+++ b/test/management_tests/management_test.py
@@ -131,7 +131,7 @@ async def test_failed_connect_disconnect(_if_mock):
 async def test_reject_incoming_connection():
     """Test rejecting incoming transport connections."""
     # Note: incoming L_DATA.ind indication connection requests are rejected
-    # L_DATA.req frames received from a tunnelling server are not yet supported
+    # L_DATA.req frames received from a tunnelling client are not yet supported
     xknx = XKNX()
     individual_address = IndividualAddress("4.0.10")
 
@@ -146,7 +146,10 @@ async def test_reject_incoming_connection():
         destination_address=individual_address,
         tpci=tpci.TDisconnect(),
     )
-    assert await xknx.knxip_interface.telegram_received(connect) == [disconnect]
+    with patch("xknx.io.KNXIPInterface.send_telegram") as send_telegram:
+        xknx.knxip_interface.telegram_received(connect)
+        await asyncio.sleep(0)
+        assert send_telegram.call_args_list == [call(disconnect)]
 
 
 async def test_incoming_unexpected_numbered_telegram():
@@ -167,7 +170,10 @@ async def test_incoming_unexpected_numbered_telegram():
         direction=TelegramDirection.OUTGOING,
         tpci=tpci.TAck(0),
     )
-    assert await xknx.knxip_interface.telegram_received(device_desc_read) == [ack]
+    with patch("xknx.io.KNXIPInterface.send_telegram") as send_telegram:
+        xknx.knxip_interface.telegram_received(device_desc_read)
+        await asyncio.sleep(0)
+        assert send_telegram.call_args_list == [call(ack)]
 
 
 async def test_incoming_wrong_address():
@@ -195,6 +201,9 @@ async def test_incoming_wrong_address():
         direction=TelegramDirection.INCOMING,
         tpci=tpci.TDisconnect(),
     )
-    assert await xknx.knxip_interface.telegram_received(connect) is None
-    assert await xknx.knxip_interface.telegram_received(ack) is None
-    assert await xknx.knxip_interface.telegram_received(disconnect) is None
+    with patch("xknx.io.KNXIPInterface.send_telegram") as send_telegram:
+        xknx.knxip_interface.telegram_received(connect)
+        xknx.knxip_interface.telegram_received(ack)
+        xknx.knxip_interface.telegram_received(disconnect)
+        await asyncio.sleep(0)
+        send_telegram.assert_not_called()