Repository: XKNX/xknx
Pull request #856 (2022-02-01 19:33:37)

support requesting Tunnel Interface informations
<!--
  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.
-->
support requesting Tunnel Interface informations - if not already known form the GatewayScanner

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:

- [ ] 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
- [ ] The changes are documented in the changelog
- [ ] The Homeassistant plugin has been adjusted in case of new config options


codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/856?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#856](https://codecov.io/gh/XKNX/xknx/pull/856?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (995d516) into [main](https://codecov.io/gh/XKNX/xknx/commit/ec122c4abc1f736b675b1498fc9c1acdc1d0632e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (ec122c4) will **decrease** coverage by `0.40%`.
> The diff coverage is `52.05%`.

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

```diff
@@            Coverage Diff             @@
##             main     #856      +/-   ##
==========================================
- Coverage   95.69%   95.28%   -0.41%     
==========================================
  Files         114      115       +1     
  Lines        6966     7024      +58     
==========================================
+ Hits         6666     6693      +27     
- Misses        300      331      +31     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/856?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/io/request\_response/connect.py](https://codecov.io/gh/XKNX/xknx/pull/856/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9yZXF1ZXN0X3Jlc3BvbnNlL2Nvbm5lY3QucHk=) | `100.00% <ø> (ø)` | |
| [xknx/io/knxip\_interface.py](https://codecov.io/gh/XKNX/xknx/pull/856/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9rbnhpcF9pbnRlcmZhY2UucHk=) | `69.23% <29.41%> (-5.24%)` | :arrow_down: |
| [xknx/io/self\_description.py](https://codecov.io/gh/XKNX/xknx/pull/856/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9zZWxmX2Rlc2NyaXB0aW9uLnB5) | `41.17% <41.17%> (ø)` | |
| [xknx/io/tunnel.py](https://codecov.io/gh/XKNX/xknx/pull/856/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby90dW5uZWwucHk=) | `73.04% <50.00%> (-0.62%)` | :arrow_down: |
| [xknx/io/\_\_init\_\_.py](https://codecov.io/gh/XKNX/xknx/pull/856/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9fX2luaXRfXy5weQ==) | `100.00% <100.00%> (ø)` | |
| [xknx/io/gateway\_scanner.py](https://codecov.io/gh/XKNX/xknx/pull/856/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9nYXRld2F5X3NjYW5uZXIucHk=) | `87.61% <100.00%> (+3.80%)` | :arrow_up: |
| [xknx/io/transport/ip\_transport.py](https://codecov.io/gh/XKNX/xknx/pull/856/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby90cmFuc3BvcnQvaXBfdHJhbnNwb3J0LnB5) | `89.58% <100.00%> (+0.22%)` | :arrow_up: |
| [xknx/io/transport/tcp\_transport.py](https://codecov.io/gh/XKNX/xknx/pull/856/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby90cmFuc3BvcnQvdGNwX3RyYW5zcG9ydC5weQ==) | `47.36% <100.00%> (+0.93%)` | :arrow_up: |


Patch:
diff --git a/xknx/io/__init__.py b/xknx/io/__init__.py
index 087d9986..f730c798 100644
--- a/xknx/io/__init__.py
+++ b/xknx/io/__init__.py
@@ -12,17 +12,19 @@ from .const import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
 from .gateway_scanner import GatewayDescriptor, GatewayScanFilter, GatewayScanner
 from .knxip_interface import KNXIPInterface, knx_interface_factory
 from .routing import Routing
+from .self_description import DescriptionQuery
 from .tunnel import TCPTunnel, UDPTunnel
 
 __all__ = [
     "DEFAULT_MCAST_GRP",
     "DEFAULT_MCAST_PORT",
+    "DescriptionQuery",
     "GatewayScanFilter",
     "GatewayScanner",
     "ConnectionConfig",
     "ConnectionType",
     "KNXIPInterface",
-    "TCPTunnel",
     "Routing",
+    "TCPTunnel",
     "UDPTunnel",
 ]
diff --git a/xknx/io/gateway_scanner.py b/xknx/io/gateway_scanner.py
index 24b4ea6c..9a98d1cf 100644
--- a/xknx/io/gateway_scanner.py
+++ b/xknx/io/gateway_scanner.py
@@ -14,6 +14,7 @@ from typing import TYPE_CHECKING
 
 import netifaces
 from xknx.knxip import (
+    DIB,
     HPAI,
     DIBDeviceInformation,
     DIBServiceFamily,
@@ -38,11 +39,11 @@ class GatewayDescriptor:
 
     def __init__(
         self,
-        name: str,
         ip_addr: str,
         port: int,
-        local_interface: str,
-        local_ip: str,
+        local_ip: str = "",
+        local_interface: str = "",
+        name: str = "UNKNOWN",
         supports_routing: bool = False,
         supports_tunnelling: bool = False,
         supports_tunnelling_tcp: bool = False,
@@ -59,6 +60,22 @@ class GatewayDescriptor:
         self.supports_tunnelling_tcp = supports_tunnelling_tcp
         self.individual_address = individual_address
 
+    def parse_dibs(self, dibs: list[DIB]) -> None:
+        """Parse DIBs for gateway information."""
+        for dib in dibs:
+            if isinstance(dib, DIBSuppSVCFamilies):
+                self.supports_routing = dib.supports(DIBServiceFamily.ROUTING)
+                if dib.supports(DIBServiceFamily.TUNNELING):
+                    self.supports_tunnelling = True
+                    self.supports_tunnelling_tcp = dib.supports(
+                        DIBServiceFamily.TUNNELING, version=2
+                    )
+                continue
+            if isinstance(dib, DIBDeviceInformation):
+                self.name = dib.name
+                self.individual_address = dib.individual_address
+                continue
+
     def __repr__(self) -> str:
         """Return object as representation string."""
         return (
@@ -223,36 +240,12 @@ class GatewayScanner:
             return
 
         gateway = GatewayDescriptor(
-            name=knx_ip_frame.body.device_name,
             ip_addr=knx_ip_frame.body.control_endpoint.ip_addr,
             port=knx_ip_frame.body.control_endpoint.port,
-            local_interface=interface,
             local_ip=udp_transport.local_addr[0],
+            local_interface=interface,
         )
-        try:
-            dib = next(
-                dib
-                for dib in knx_ip_frame.body.dibs
-                if isinstance(dib, DIBSuppSVCFamilies)
-            )
-            gateway.supports_routing = dib.supports(DIBServiceFamily.ROUTING)
-            if dib.supports(DIBServiceFamily.TUNNELING):
-                gateway.supports_tunnelling = True
-                gateway.supports_tunnelling_tcp = dib.supports(
-                    DIBServiceFamily.TUNNELING, version=2
-                )
-        except StopIteration:
-            pass
-
-        try:
-            device_infos = next(
-                device_infos
-                for device_infos in knx_ip_frame.body.dibs
-                if isinstance(device_infos, DIBDeviceInformation)
-            )
-            gateway.individual_address = device_infos.individual_address
-        except StopIteration:
-            pass
+        gateway.parse_dibs(knx_ip_frame.body.dibs)
 
         logger.debug("Found KNX/IP device at %s: %s", source, repr(gateway))
         self._add_found_gateway(gateway)
diff --git a/xknx/io/knxip_interface.py b/xknx/io/knxip_interface.py
index 93b4bdf5..5f0bd9da 100644
--- a/xknx/io/knxip_interface.py
+++ b/xknx/io/knxip_interface.py
@@ -9,18 +9,17 @@ KNXIPInterface manages KNX/IP Tunneling or Routing connections.
 from __future__ import annotations
 
 import asyncio
-import ipaddress
 import logging
 import threading
-from typing import TYPE_CHECKING, Awaitable, TypeVar, cast
+from typing import TYPE_CHECKING, Awaitable, TypeVar
 
-import netifaces
 from xknx.exceptions import CommunicationError, XKNXException
 
 from .connection import ConnectionConfig, ConnectionType
 from .gateway_scanner import GatewayDescriptor, GatewayScanFilter, GatewayScanner
 from .routing import Routing
-from .tunnel import TCPTunnel, UDPTunnel
+from .tunnel import TCPTunnel, UDPTunnel, _Tunnel
+from .util import find_local_ip, validate_ip
 
 if TYPE_CHECKING:
     import concurrent
@@ -54,8 +53,9 @@ class KNXIPInterface:
     ):
         """Initialize KNXIPInterface class."""
         self.xknx = xknx
-        self._interface: Interface | None = None
         self.connection_config = connection_config
+        self._gateway_info: GatewayDescriptor | None = None
+        self._interface: Interface | None = None
 
     async def start(self) -> None:
         """Start KNX/IP interface. Raise `CommunicationError` if connection fails."""
@@ -95,6 +95,7 @@ class KNXIPInterface:
         """Start GatewayScanner and connect to the found device."""
         scan_filter = self.connection_config.scan_filter
         gateway, local_interface_ip = await self.find_gateway(scan_filter)
+        self._gateway_info = gateway
 
         if gateway.supports_tunnelling and scan_filter.routing is not True:
             await self._start_tunnelling_udp(
@@ -174,6 +175,7 @@ class KNXIPInterface:
             scan_filter = self.connection_config.scan_filter
             scan_filter.routing = True
             _gateway, local_ip = await self.find_gateway(scan_filter)
+            self._gateway_info = _gateway
         validate_ip(local_ip, address_name="Local IP address")
         logger.debug("Starting Routing from %s as %s", local_ip, self.xknx.own_address)
         self._interface = Routing(self.xknx, self.telegram_received, local_ip)
@@ -195,6 +197,14 @@ class KNXIPInterface:
             raise CommunicationError("KNX/IP interface not connected")
         return await self._interface.send_telegram(telegram)
 
+    async def gateway_info(self) -> GatewayDescriptor | None:
+        """Get gateway descriptor from interface."""
+        if self._gateway_info is not None:
+            return self._gateway_info
+        if isinstance(self._interface, _Tunnel):
+            return await self._interface.request_description()
+        return None
+
     async def find_gateway(
         self, scan_filter: GatewayScanFilter
     ) -> tuple[GatewayDescriptor, str]:
@@ -279,50 +289,12 @@ class KNXIPInterfaceThreaded(KNXIPInterface):
             self._interface.send_telegram(telegram)
         )
 
-
-def find_local_ip(gateway_ip: str) -> str:
-    """Find local IP address on same subnet as gateway."""
-
-    def _scan_interfaces(gateway: ipaddress.IPv4Address) -> str | None:
-        """Return local IP address on same subnet as given gateway."""
-        for interface in netifaces.interfaces():
-            try:
-                af_inet = netifaces.ifaddresses(interface)[netifaces.AF_INET]
-                for link in af_inet:
-                    network = ipaddress.IPv4Network(
-                        (link["addr"], link["netmask"]), strict=False
-                    )
-                    if gateway in network:
-                        logger.debug("Using interface: %s", interface)
-                        return cast(str, link["addr"])
-            except KeyError:
-                logger.debug("Could not find IPv4 address on interface %s", interface)
-                continue
+    async def gateway_info(self) -> GatewayDescriptor | None:
+        """Get gateway descriptor from interface."""
+        if self._gateway_info is not None:
+            return self._gateway_info
+        if isinstance(self._interface, _Tunnel):
+            return await self._await_from_connection_thread(
+                self._interface.request_description()
+            )
         return None
-
-    def _find_default_gateway() -> ipaddress.IPv4Address:
-        """Return IP address of default gateway."""
-        gws = netifaces.gateways()
-        return ipaddress.IPv4Address(gws["default"][netifaces.AF_INET][0])
-
-    gateway = ipaddress.IPv4Address(gateway_ip)
-    local_ip = _scan_interfaces(gateway)
-    if local_ip is None:
-        logger.warning(
-            "No interface on same subnet as gateway found. Falling back to default gateway."
-        )
-        try:
-            default_gateway = _find_default_gateway()
-        except KeyError as err:
-            raise CommunicationError(f"No route to {gateway} found") from err
-        local_ip = _scan_interfaces(default_gateway)
-    assert isinstance(local_ip, str)
-    return local_ip
-
-
-def validate_ip(address: str, address_name: str = "IP address") -> None:
-    """Raise an exception if address cannot be parsed as IPv4 address."""
-    try:
-        ipaddress.IPv4Address(address)
-    except ipaddress.AddressValueError as ex:
-        raise XKNXException(f"{address_name} is not a valid IPv4 address.") from ex
diff --git a/xknx/io/request_response/connect.py b/xknx/io/request_response/connect.py
index 62078858..d0bc7727 100644
--- a/xknx/io/request_response/connect.py
+++ b/xknx/io/request_response/connect.py
@@ -19,7 +19,7 @@ if TYPE_CHECKING:
 
 
 class Connect(RequestResponse):
-    """Class to send a ConnectRequest and wait for ConnectResponse.."""
+    """Class to send a ConnectRequest and wait for ConnectResponse."""
 
     def __init__(self, xknx: XKNX, transport: KNXIPTransport, local_hpai: HPAI):
         """Initialize Connect class."""
diff --git a/xknx/io/request_response/request_response.py b/xknx/io/request_response/request_response.py
index 4a245734..3c047681 100644
--- a/xknx/io/request_response/request_response.py
+++ b/xknx/io/request_response/request_response.py
@@ -9,6 +9,7 @@ import asyncio
 import logging
 from typing import TYPE_CHECKING
 
+from xknx.exceptions import CommunicationError
 from xknx.io.transport import KNXIPTransport
 from xknx.knxip import HPAI, ErrorCode, KNXIPBodyResponse, KNXIPFrame
 
@@ -47,9 +48,8 @@ class RequestResponse:
         callb = self.transport.register_callback(
             self.response_rec_callback, [self.awaited_response_class.SERVICE_TYPE]
         )
-        await self.send_request()
-
         try:
+            await self.send_request()
             await asyncio.wait_for(
                 self.response_received_event.wait(),
                 timeout=self.timeout_in_seconds,
@@ -60,12 +60,16 @@ class RequestResponse:
                 self.timeout_in_seconds,
                 self.__class__.__name__,
             )
+        except CommunicationError as err:
+            logger.warning(
+                "Sending request of type '%s' failed: %s", self.__class__.__name__, err
+            )
         finally:
             # cleanup to not leave callbacks (for asyncio.CancelledError)
             self.transport.unregister_callback(callb)
 
     async def send_request(self) -> None:
-        """Build knxipframe (within derived class) and send via UDP."""
+        """Build knxipframe (within derived class) and send via transport."""
         self.transport.send(self.create_knxipframe())
 
     def response_rec_callback(
diff --git a/xknx/io/self_description.py b/xknx/io/self_description.py
new file mode 100644
index 00000000..78e22c6d
--- /dev/null
+++ b/xknx/io/self_description.py
@@ -0,0 +1,122 @@
+"""Abstraction to send DescriptionRequest and wait for DescriptionResponse."""
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import TYPE_CHECKING, Final
+
+from xknx.exceptions import CommunicationError
+from xknx.io.gateway_scanner import GatewayDescriptor
+from xknx.knxip import HPAI, DescriptionRequest, DescriptionResponse, KNXIPFrame
+
+from .const import DEFAULT_MCAST_PORT
+from .transport import UDPTransport
+from .util import find_local_ip
+
+if TYPE_CHECKING:
+    from xknx.io.transport import KNXIPTransport
+    from xknx.xknx import XKNX
+
+logger = logging.getLogger("xknx.log")
+
+DESCRIPTION_TIMEOUT: Final = 2
+
+
+async def request_description(
+    xknx: XKNX,
+    gateway_ip: str,
+    gateway_port: int = DEFAULT_MCAST_PORT,
+    local_ip: str | None = None,
+    local_port: int = 0,
+    route_back: bool = False,
+) -> GatewayDescriptor | None:
+    """Set up a UDP transport to request a description from a KNXnet/IP device."""
+    _local_ip = local_ip or find_local_ip(gateway_ip)
+    transport = UDPTransport(
+        xknx,
+        local_addr=(_local_ip, local_port),
+        remote_addr=(gateway_ip, gateway_port),
+        multicast=False,
+    )
+    try:
+        await transport.connect()
+    except OSError as err:
+        logger.warning("Could not setup socket: %s", err)
+        return None
+
+    local_hpai: HPAI
+    if route_back:
+        local_hpai = HPAI()
+    else:
+        local_addr = transport.getsockname()
+        local_hpai = HPAI(*local_addr)
+
+    description_query = DescriptionQuery(
+        xknx,
+        transport,
+        local_hpai=local_hpai,
+    )
+    await description_query.start()
+    transport.stop()
+    return description_query.gateway_descriptor
+
+
+class DescriptionQuery:
+    """Class to send a DescriptionRequest and wait for DescriptionResponse."""
+
+    def __init__(self, xknx: XKNX, transport: KNXIPTransport, local_hpai: HPAI):
+        """Initialize Description class."""
+        self.xknx = xknx
+        self.transport = transport
+        self.local_hpai = local_hpai
+
+        self.gateway_descriptor: GatewayDescriptor | None = None
+        self.response_received_event = asyncio.Event()
+
+    def create_knxipframe(self) -> KNXIPFrame:
+        """Create KNX/IP Frame object to be sent to device."""
+        description_request = DescriptionRequest(
+            self.xknx,
+            control_endpoint=self.local_hpai,
+        )
+        return KNXIPFrame.init_from_body(description_request)
+
+    async def start(self) -> None:
+        """Start. Send request and wait for an answer."""
+        callb = self.transport.register_callback(
+            self.response_rec_callback, [DescriptionResponse.SERVICE_TYPE]
+        )
+        try:
+            self.transport.send(self.create_knxipframe())
+            await asyncio.wait_for(
+                self.response_received_event.wait(),
+                timeout=DESCRIPTION_TIMEOUT,
+            )
+        except asyncio.TimeoutError:
+            logger.debug(
+                "Error: KNX bus did not respond in time (%s secs) to request of type '%s'",
+                DESCRIPTION_TIMEOUT,
+                self.__class__.__name__,
+            )
+        except CommunicationError as err:
+            logger.warning("Sending DescriptionRequest failed: %s", err)
+        finally:
+            # cleanup to not leave callbacks (for asyncio.CancelledError)
+            self.transport.unregister_callback(callb)
+
+    def response_rec_callback(
+        self, knxipframe: KNXIPFrame, source: HPAI, _: KNXIPTransport
+    ) -> None:
+        """Verify and handle knxipframe. Callback from internal transport."""
+        if not isinstance(knxipframe.body, DescriptionResponse):
+            logger.warning("Wrong knxipframe for DescriptionRequest: %s", knxipframe)
+            return
+        self.response_received_event.set()
+        # Set gateway descriptior attribute
+        gateway = GatewayDescriptor(
+            ip_addr=self.transport.remote_addr[0],
+            port=self.transport.remote_addr[1],
+            local_ip=self.transport.getsockname()[0],
+        )
+        gateway.parse_dibs(knxipframe.body.dibs)
+        self.gateway_descriptor = gateway
diff --git a/xknx/io/transport/ip_transport.py b/xknx/io/transport/ip_transport.py
index f4437162..870cdeb7 100644
--- a/xknx/io/transport/ip_transport.py
+++ b/xknx/io/transport/ip_transport.py
@@ -24,6 +24,7 @@ class KNXIPTransport(ABC):
 
     callbacks: list[KNXIPTransport.Callback]
     local_hpai: HPAI
+    remote_addr: tuple[str, int]
     transport: asyncio.BaseTransport | None
 
     class Callback:
diff --git a/xknx/io/transport/tcp_transport.py b/xknx/io/transport/tcp_transport.py
index 85b6dd91..2822ec2b 100644
--- a/xknx/io/transport/tcp_transport.py
+++ b/xknx/io/transport/tcp_transport.py
@@ -10,7 +10,7 @@ import logging
 import time
 from typing import TYPE_CHECKING, Callable, cast
 
-from xknx.exceptions import CouldNotParseKNXIP, IncompleteKNXIPFrame, XKNXException
+from xknx.exceptions import CommunicationError, CouldNotParseKNXIP, IncompleteKNXIPFrame
 from xknx.knxip import HPAI, HostProtocol, KNXIPFrame
 
 from .ip_transport import KNXIPTransport
@@ -60,6 +60,7 @@ class TCPTransport(KNXIPTransport):
     ):
         """Initialize UDPTransport class."""
         self.xknx = xknx
+        self.remote_addr = remote_addr
         self.remote_hpai = HPAI(*remote_addr, protocol=HostProtocol.IPV4_TCP)
 
         self.callbacks = []
@@ -128,6 +129,6 @@ class TCPTransport(KNXIPTransport):
             knxipframe,
         )
         if self.transport is None:
-            raise XKNXException("Transport not connected")
+            raise CommunicationError("Transport not connected")
 
         self.transport.write(bytes(knxipframe.to_knx()))
diff --git a/xknx/io/transport/udp_transport.py b/xknx/io/transport/udp_transport.py
index 42b1694f..23ed4258 100644
--- a/xknx/io/transport/udp_transport.py
+++ b/xknx/io/transport/udp_transport.py
@@ -13,7 +13,7 @@ from sys import platform
 import time
 from typing import TYPE_CHECKING, Callable, cast
 
-from xknx.exceptions import CouldNotParseKNXIP, XKNXException
+from xknx.exceptions import CommunicationError, CouldNotParseKNXIP
 from xknx.knxip import HPAI, KNXIPFrame
 
 from .ip_transport import KNXIPTransport
@@ -178,7 +178,7 @@ class UDPTransport(KNXIPTransport):
             "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")
+            raise CommunicationError("Transport not connected")
 
         if self.multicast:
             if addr is not None:
diff --git a/xknx/io/tunnel.py b/xknx/io/tunnel.py
index 273fc416..9a29be6f 100644
--- a/xknx/io/tunnel.py
+++ b/xknx/io/tunnel.py
@@ -11,7 +11,7 @@ import logging
 from typing import TYPE_CHECKING, Awaitable, Callable
 
 from xknx.core import XknxConnectionState
-from xknx.exceptions import CommunicationError, XKNXException
+from xknx.exceptions import CommunicationError
 from xknx.knxip import (
     HPAI,
     CEMIFrame,
@@ -27,8 +27,10 @@ from xknx.knxip import (
 from xknx.telegram import IndividualAddress, Telegram, TelegramDirection
 
 from .const import HEARTBEAT_RATE
+from .gateway_scanner import GatewayDescriptor
 from .interface import Interface
 from .request_response import Connect, ConnectionState, Disconnect, Tunnelling
+from .self_description import DescriptionQuery
 from .transport import KNXIPTransport, TCPTransport, UDPTransport
 
 if TYPE_CHECKING:
@@ -233,13 +235,23 @@ class _Tunnel(Interface):
             await disconnect.start()
             if not disconnect.success and not ignore_error:
                 self.communication_channel = None
-                raise XKNXException("Could not disconnect channel")
+                raise CommunicationError("Could not disconnect channel")
             logger.debug(
                 "Tunnel disconnected (communication_channel: %s)",
                 self.communication_channel,
             )
         self.communication_channel = None
 
+    async def request_description(self) -> GatewayDescriptor | None:
+        """Request description from tunneling server."""
+        description = DescriptionQuery(
+            self.xknx,
+            self.transport,
+            local_hpai=self.local_hpai,
+        )
+        await description.start()
+        return description.gateway_descriptor
+
     async def send_telegram(self, telegram: Telegram) -> None:
         """
         Send Telegram to routing tunnelling device - retry mechanism.
@@ -497,6 +509,7 @@ class UDPTunnel(_Tunnel):
             self.xknx,
             (self.local_ip, self.local_port),
             (self.gateway_ip, self.gateway_port),
+            multicast=False,
         )
 
     def _get_hpai(self) -> HPAI:
diff --git a/xknx/io/util.py b/xknx/io/util.py
new file mode 100644
index 00000000..68112381
--- /dev/null
+++ b/xknx/io/util.py
@@ -0,0 +1,59 @@
+"""Helper functions for XKNX."""
+from __future__ import annotations
+
+import ipaddress
+import logging
+from typing import cast
+
+import netifaces
+from xknx.exceptions import CommunicationError, XKNXException
+
+logger = logging.getLogger("xknx.log")
+
+
+def find_local_ip(gateway_ip: str) -> str:
+    """Find local IP address on same subnet as gateway."""
+
+    def _scan_interfaces(gateway: ipaddress.IPv4Address) -> str | None:
+        """Return local IP address on same subnet as given gateway."""
+        for interface in netifaces.interfaces():
+            try:
+                af_inet = netifaces.ifaddresses(interface)[netifaces.AF_INET]
+                for link in af_inet:
+                    network = ipaddress.IPv4Network(
+                        (link["addr"], link["netmask"]), strict=False
+                    )
+                    if gateway in network:
+                        logger.debug("Using interface: %s", interface)
+                        return cast(str, link["addr"])
+            except KeyError:
+                logger.debug("Could not find IPv4 address on interface %s", interface)
+                continue
+        return None
+
+    def _find_default_gateway() -> ipaddress.IPv4Address:
+        """Return IP address of default gateway."""
+        gws = netifaces.gateways()
+        return ipaddress.IPv4Address(gws["default"][netifaces.AF_INET][0])
+
+    gateway = ipaddress.IPv4Address(gateway_ip)
+    local_ip = _scan_interfaces(gateway)
+    if local_ip is None:
+        logger.warning(
+            "No interface on same subnet as gateway found. Falling back to default gateway."
+        )
+        try:
+            default_gateway = _find_default_gateway()
+        except KeyError as err:
+            raise CommunicationError(f"No route to {gateway} found") from err
+        local_ip = _scan_interfaces(default_gateway)
+    assert isinstance(local_ip, str)
+    return local_ip
+
+
+def validate_ip(address: str, address_name: str = "IP address") -> None:
+    """Raise an exception if address cannot be parsed as IPv4 address."""
+    try:
+        ipaddress.IPv4Address(address)
+    except ipaddress.AddressValueError as ex:
+        raise XKNXException(f"{address_name} is not a valid IPv4 address.") from ex


Test patch:
diff --git a/requirements/testing.txt b/requirements/testing.txt
index d0e3f03b..610a9577 100644
--- a/requirements/testing.txt
+++ b/requirements/testing.txt
@@ -5,9 +5,9 @@ flake8==4.0.1
 flake8-isort==4.1.1
 pydocstyle==6.1.1
 pylint==2.12.2
-pytest==7.0.0
-pytest-asyncio==0.18.0
+pytest==7.0.1
+pytest-asyncio==0.18.1
 pytest-cov==3.0.0
-setuptools==60.8.1
+setuptools==60.9.0
 tox==3.24.5
 mypy==0.931
diff --git a/test/io_tests/self_description_test.py b/test/io_tests/self_description_test.py
new file mode 100644
index 00000000..00fc51ff
--- /dev/null
+++ b/test/io_tests/self_description_test.py
@@ -0,0 +1,66 @@
+"""Unit test for KNX/IP Tunnelling Request/Response."""
+import asyncio
+from unittest.mock import Mock, patch
+
+import pytest
+from xknx import XKNX
+from xknx.io.self_description import DescriptionQuery, request_description
+from xknx.io.transport.udp_transport import UDPTransport
+from xknx.knxip import HPAI, DescriptionRequest, KNXIPFrame
+
+
+@pytest.mark.asyncio
+class TestSelfDescription:
+    """Test class for KNX/IP self description."""
+
+    async def test_description_query(self, time_travel):
+        """Test DescriptionQuery class."""
+        xknx = XKNX()
+        local_addr = ("127.0.0.1", 12345)
+        remote_addr = ("127.0.0.2", 54321)
+        transport_mocked = UDPTransport(
+            xknx, local_addr=local_addr, remote_addr=remote_addr
+        )
+        transport_mocked.getsockname = Mock(return_value=local_addr)
+        transport_mocked.send = Mock()
+
+        description_request = KNXIPFrame.init_from_body(
+            DescriptionRequest(xknx, control_endpoint=HPAI(*local_addr))
+        )
+        description_response_raw = bytes.fromhex(
+            "061002040048360102001000000000082d40834de000170c000ab3274a324b4e582f49502d526f7574657200000000000000000000000000000000000c0202020302040205020701",
+        )
+        description_query = DescriptionQuery(
+            xknx, transport=transport_mocked, local_hpai=HPAI(*local_addr)
+        )
+        task = asyncio.create_task(description_query.start())
+        await time_travel(0)
+        transport_mocked.send.assert_called_once_with(description_request)
+
+        transport_mocked.data_received_callback(
+            raw=description_response_raw, source=remote_addr
+        )
+        await task
+        assert description_query.gateway_descriptor is not None
+        assert description_query.gateway_descriptor.name == "KNX/IP-Router"
+
+    async def test_request_description(self, time_travel):
+        """Test request_description function."""
+        xknx = XKNX()
+        local_addr = ("127.0.0.1", 12345)
+        remote_addr = ("127.0.0.2", 54321)
+
+        with patch(
+            "xknx.io.transport.UDPTransport.connect"
+        ) as transport_connect_mock, patch(
+            "xknx.io.transport.UDPTransport.getsockname", return_value=local_addr
+        ) as transport_getsockname_mock, patch(
+            "xknx.io.transport.UDPTransport.stop"
+        ) as transport_stop_mock, patch(
+            "xknx.io.self_description.DescriptionQuery.start"
+        ) as description_query_start_mock:
+            await request_description(xknx, remote_addr[0])
+            transport_connect_mock.assert_called_once_with()
+            transport_getsockname_mock.assert_called_once_with()
+            transport_stop_mock.assert_called_once_with()
+            description_query_start_mock.assert_called_once_with()
diff --git a/test/io_tests/tunnel_test.py b/test/io_tests/tunnel_test.py
index 188a6ea2..b6ec0469 100644
--- a/test/io_tests/tunnel_test.py
+++ b/test/io_tests/tunnel_test.py
@@ -6,11 +6,14 @@ import pytest
 from xknx import XKNX
 from xknx.dpt import DPTArray
 from xknx.io import UDPTunnel
+from xknx.io.gateway_scanner import GatewayDescriptor
 from xknx.knxip import (
     HPAI,
     CEMIFrame,
     ConnectRequest,
     ConnectResponse,
+    DescriptionRequest,
+    DescriptionResponse,
     DisconnectRequest,
     DisconnectResponse,
     KNXIPFrame,
@@ -206,3 +209,24 @@ class TestUDPTunnel:
         await disconnection_task
         assert self.tunnel._data_endpoint_addr is None
         self.tunnel.transport.stop.assert_called_once()
+
+    async def test_tunnel_request_description(self, time_travel):
+        """Test tunnel requesting and returning description of connected interface."""
+        local_addr = ("192.168.1.1", 12345)
+        self.tunnel.transport.send = Mock()
+        self.tunnel.transport.getsockname = Mock(return_value=local_addr)
+
+        description_request = KNXIPFrame.init_from_body(
+            DescriptionRequest(self.xknx, control_endpoint=self.tunnel.local_hpai)
+        )
+        description_response = KNXIPFrame.init_from_body(DescriptionResponse(self.xknx))
+
+        task = asyncio.create_task(self.tunnel.request_description())
+        await time_travel(0)
+        self.tunnel.transport.send.assert_called_once_with(description_request)
+        self.tunnel.transport.handle_knxipframe(description_response, HPAI())
+        await time_travel(0)
+        assert task.done()
+        assert isinstance(task.result(), GatewayDescriptor)
+        assert self.tunnel.transport.send.call_count == 1
+        await task