Repository: XKNX/xknx
Pull request #1183 (2023-02-07 20:06:38)

Fix Data Secure Keyring parsing
<!--
  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.
-->
- Fix Data Secure Keyring parsing for non-IP-Secure interfaces and when no devices are exported (only `senders` list).
- Accept `str | os.PathLike` for Keyring path. Previously only `str`.
- Rename `_load_keyring` to `sync_load_keyring` to make it public e.g. when it should be used from an executor.
- Use only Interfaces listed in Keyring when `ConnectionType.AUTOMATIC` is used and a Keyring is configured.

## 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)
- [ ] Code quality improvements to existing code or addition of tests

## 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/1183?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#1183](https://codecov.io/gh/XKNX/xknx/pull/1183?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (aaaa747) into [main](https://codecov.io/gh/XKNX/xknx/commit/cad292f2bb3d646af206974c5bac4ca6841290f7?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (cad292f) will **increase** coverage by `0.03%`.
> The diff coverage is `98.36%`.

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

```diff
@@            Coverage Diff             @@
##             main    #1183      +/-   ##
==========================================
+ Coverage   96.40%   96.43%   +0.03%     
==========================================
  Files         143      143              
  Lines        8873     8900      +27     
==========================================
+ Hits         8554     8583      +29     
+ Misses        319      317       -2     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/1183?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/secure/keyring.py](https://codecov.io/gh/XKNX/xknx/pull/1183?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9zZWN1cmUva2V5cmluZy5weQ==) | `98.79% <97.43%> (-0.33%)` | :arrow_down: |
| [xknx/io/connection.py](https://codecov.io/gh/XKNX/xknx/pull/1183?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9jb25uZWN0aW9uLnB5) | `100.00% <100.00%> (ø)` | |
| [xknx/io/knxip\_interface.py](https://codecov.io/gh/XKNX/xknx/pull/1183?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9rbnhpcF9pbnRlcmZhY2UucHk=) | `87.90% <100.00%> (+1.98%)` | :arrow_up: |
| [xknx/secure/data\_secure.py](https://codecov.io/gh/XKNX/xknx/pull/1183?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9zZWN1cmUvZGF0YV9zZWN1cmUucHk=) | `97.75% <100.00%> (-0.12%)` | :arrow_down: |


Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index e6bd62c5..a88744f0 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -6,6 +6,23 @@ nav_order: 2
 
 # Changelog
 
+# Unreleased changes
+
+### Connection
+
+- Use only Interfaces listed in Keyring when `ConnectionType.AUTOMATIC` is used and a Keyring is configured.
+
+### Bugfixes
+
+- Parse Data Secure credentials form Keyring from non-IP-Secure interfaces.
+- Parse Data Secure credentials from Keyrings exported for specific interfaces.
+
+### Cleanups
+
+- Accept `str | os.PathLike` for Keyring path. Previously only `str`.
+- Rename `_load_keyring` to `sync_load_keyring` to make it public e.g. when it should be used from an executor.
+- Update CI. Use `codespell` and `flake8-print`.
+
 # 2.4.0 Data Secure 2023-02-05
 
 ### Data Secure
diff --git a/xknx/io/connection.py b/xknx/io/connection.py
index c6ad6a51..7cd0bf8f 100644
--- a/xknx/io/connection.py
+++ b/xknx/io/connection.py
@@ -2,6 +2,8 @@
 from __future__ import annotations
 
 from enum import Enum, auto
+import os
+from typing import Any
 
 from xknx.telegram.address import IndividualAddress, IndividualAddressableType
 
@@ -111,7 +113,7 @@ class SecureConfig:
         user_id: int | None = None,
         device_authentication_password: str | None = None,
         user_password: str | None = None,
-        knxkeys_file_path: str | None = None,
+        knxkeys_file_path: str | os.PathLike[Any] | None = None,
         knxkeys_password: str | None = None,
     ):
         """Initialize SecureConfig class."""
diff --git a/xknx/io/knxip_interface.py b/xknx/io/knxip_interface.py
index ee06fbc6..c04eb054 100644
--- a/xknx/io/knxip_interface.py
+++ b/xknx/io/knxip_interface.py
@@ -21,7 +21,8 @@ from xknx.exceptions import (
     XKNXException,
 )
 from xknx.io import util
-from xknx.secure.keyring import Keyring, XMLInterface, load_keyring
+from xknx.secure.keyring import InterfaceType, Keyring, XMLInterface, load_keyring
+from xknx.telegram import IndividualAddress
 
 from .connection import ConnectionConfig, ConnectionType
 from .const import DEFAULT_INDIVIDUAL_ADDRESS
@@ -115,15 +116,27 @@ class KNXIPInterface:
                 keyring=keyring,
             )
         else:
-            await self._start_automatic()
+            await self._start_automatic(keyring=keyring)
 
-    async def _start_automatic(self) -> None:
+    async def _start_automatic(self, keyring: Keyring | None) -> None:
         """Start GatewayScanner and connect to the found device."""
+        keyring_host_ias: set[IndividualAddress] | None = None
+        if keyring and keyring.interfaces:
+            keyring_host_ias = {
+                interface.host
+                for interface in keyring.interfaces
+                if interface.host is not None
+                and interface.type is InterfaceType.TUNNELING
+            }
+
         async for gateway in GatewayScanner(
             self.xknx,
             local_ip=self.connection_config.local_ip,
             scan_filter=self.connection_config.scan_filter,
         ).async_scan():
+            if keyring_host_ias and gateway.individual_address not in keyring_host_ias:
+                logger.debug("Skipping %s. No match in keyring file", gateway)
+                continue
             try:
                 if gateway.supports_tunnelling_tcp:
                     if gateway.tunnelling_requires_secure:
@@ -131,6 +144,7 @@ class KNXIPInterface:
                             gateway_ip=gateway.ip_addr,
                             gateway_port=gateway.port,
                             gateway_descriptor=gateway,
+                            keyring=keyring,
                         )
                     else:
                         await self._start_tunnelling_tcp(
@@ -148,13 +162,20 @@ class KNXIPInterface:
                 elif gateway.supports_routing and not gateway.routing_requires_secure:
                     await self._start_routing()
             except CommunicationError as ex:
-                logger.debug("Could not connect to %s: %s", gateway, ex)
+                logger.debug("Skipping %s. Could not connect: %s", gateway, ex)
+                continue
+            except InvalidSecureConfiguration as ex:
+                logger.debug(
+                    "Skipping %s. Invalid secure configuration: %s", gateway, ex
+                )
                 continue
             else:
                 self._gateway_info = gateway
                 break
         else:
-            raise CommunicationError("No usable KNX/IP device found.")
+            raise CommunicationError(
+                f"No usable KNX/IP device found{' in keyring file' if keyring_host_ias else ''}."
+            )
 
     async def _start_tunnelling_tcp(
         self,
@@ -205,14 +226,18 @@ class KNXIPInterface:
             xml_interface = self._get_tunnel_interface_from_keyring(
                 keyring=keyring,
                 gateway_descriptor=_gateway,
+                individual_address=self.connection_config.individual_address,
                 config_user_id=secure_config.user_id,
             )
-            user_id = xml_interface.user_id
-            if (_user_password := xml_interface.decrypted_password) is None:
+            if (
+                xml_interface.user_id is None
+                or xml_interface.decrypted_password is None
+            ):
                 raise InvalidSecureConfiguration(
-                    f"No password found for tunnel {xml_interface.individual_address} user_id {user_id}"
+                    f"No user_id or password found for tunnel {xml_interface.individual_address}"
                 )
-            user_password = _user_password
+            user_id = xml_interface.user_id
+            user_password = xml_interface.decrypted_password
             device_authentication_password = xml_interface.decrypted_authentication
         else:
             raise InvalidSecureConfiguration(
@@ -238,10 +263,11 @@ class KNXIPInterface:
         )
         await self._interface.connect()
 
+    @staticmethod
     def _get_tunnel_interface_from_keyring(
-        self,
         keyring: Keyring,
         gateway_descriptor: GatewayDescriptor,
+        individual_address: IndividualAddress | None = None,
         config_user_id: int | None = None,
     ) -> XMLInterface:
         """
@@ -249,11 +275,13 @@ class KNXIPInterface:
 
         Precedence: configured individual address > configured user id > first free tunnel interface
         """
-        if _ia := self.connection_config.individual_address:
-            if xml_interface := keyring.get_tunnel_interface_by_individual_address(_ia):
+        if individual_address:
+            if xml_interface := keyring.get_tunnel_interface_by_individual_address(
+                individual_address
+            ):
                 return xml_interface
             raise InvalidSecureConfiguration(
-                f"Interface with individual address {_ia} not found in keyfile"
+                f"Interface with individual address {individual_address} not found in keyfile"
             )
 
         if not (host_ia := gateway_descriptor.individual_address):
diff --git a/xknx/secure/data_secure.py b/xknx/secure/data_secure.py
index 249f84b5..2b723b5a 100644
--- a/xknx/secure/data_secure.py
+++ b/xknx/secure/data_secure.py
@@ -83,20 +83,10 @@ class DataSecure:
 
         Return None if no Data Secure information is found in the Keyring.
         """
-        ga_key_table: dict[GroupAddress, bytes] = {}
-        ia_seq_table: dict[IndividualAddress, int] = {}
-
-        for xml_ga in keyring.group_addresses:
-            if xml_ga.decrypted_key is not None:
-                ga_key_table[xml_ga.address] = xml_ga.decrypted_key
-
-        for xml_ia in keyring.devices:
-            # TODO: check if this should default to 0 or if devices without a sequence number
-            #       in keyfile should be excluded from the table
-            ia_seq_table[xml_ia.individual_address] = xml_ia.sequence_number
+        ga_key_table = keyring.get_data_secure_group_keys()
+        ia_seq_table = keyring.get_data_secure_senders()
         # TODO: persist local individual_address_table and update from that file on start
         #       to have more fresh initial sequence numbers
-
         if not ga_key_table:
             return None
         return DataSecure(
diff --git a/xknx/secure/keyring.py b/xknx/secure/keyring.py
index 4bd226f0..42980632 100644
--- a/xknx/secure/keyring.py
+++ b/xknx/secure/keyring.py
@@ -7,6 +7,7 @@ import base64
 import enum
 from itertools import chain
 import logging
+import os
 from typing import Any
 from xml.dom.minidom import Attr, Document, parse
 from xml.etree.ElementTree import Element, ElementTree
@@ -59,7 +60,7 @@ class XMLAssignedGroupAddress(AttributeReader):
     """Assigned Group Addresses to an interface in a knxkeys file."""
 
     address: GroupAddress
-    senders: list[str]
+    senders: list[IndividualAddress]
 
     def parse_xml(self, node: Document) -> None:
         """Parse all needed attributes from the given node map."""
@@ -67,40 +68,46 @@ class XMLAssignedGroupAddress(AttributeReader):
         self.address = GroupAddress(
             self.get_attribute_value(attributes.get("Address", None))
         )
-        self.senders = str(
-            self.get_attribute_value(attributes.get("Senders", ""))
-        ).split(" ")
+        self.senders = [
+            IndividualAddress(sender)
+            for sender in (
+                self.get_attribute_value(attributes.get("Senders", ""))
+            ).split()
+        ]
 
 
 class XMLInterface(AttributeReader):
     """Interface in a knxkeys file."""
 
     type: InterfaceType
-    host: IndividualAddress
-    user_id: int
-    password: str
+    individual_address: IndividualAddress
+    host: IndividualAddress | None = None
+    user_id: int | None = None
+    password: str | None = None
     decrypted_password: str | None = None
     decrypted_authentication: str | None = None
-    individual_address: IndividualAddress
-    authentication: str
-    group_addresses: list[XMLAssignedGroupAddress] = []
+    authentication: str | None = None
+    group_addresses: dict[GroupAddress, list[IndividualAddress]]
 
     def parse_xml(self, node: Document) -> None:
         """Parse all needed attributes from the given node map."""
         attributes = node.attributes
         self.type = InterfaceType(self.get_attribute_value(attributes.get("Type")))
-        self.host = IndividualAddress(self.get_attribute_value(attributes.get("Host")))
-        self.user_id = int(self.get_attribute_value(attributes.get("UserID")) or 2)
-        self.password = self.get_attribute_value(attributes.get("Password"))
         self.individual_address = IndividualAddress(
             self.get_attribute_value(attributes.get("IndividualAddress"))
         )
+        _host = self.get_attribute_value(attributes.get("Host"))
+        self.host = IndividualAddress(_host) if _host else None
+        _user_id = self.get_attribute_value(attributes.get("UserID"))
+        self.user_id = int(_user_id) if _user_id else None
+        self.password = self.get_attribute_value(attributes.get("Password"))
         self.authentication = self.get_attribute_value(attributes.get("Authentication"))
 
+        self.group_addresses = {}
         for assigned_ga in filter(lambda x: x.nodeType != 3, node.childNodes):
-            group_address: XMLAssignedGroupAddress = XMLAssignedGroupAddress()
-            group_address.parse_xml(assigned_ga)
-            self.group_addresses.append(group_address)
+            xml_group_address: XMLAssignedGroupAddress = XMLAssignedGroupAddress()
+            xml_group_address.parse_xml(assigned_ga)
+            self.group_addresses[xml_group_address.address] = xml_group_address.senders
 
     def decrypt_attributes(
         self, password_hash: bytes, initialization_vector: bytes
@@ -311,6 +318,64 @@ class Keyring(AttributeReader):
             None,
         )
 
+    def get_interface_by_individual_address(
+        self, individual_address: IndividualAddress
+    ) -> XMLInterface | None:
+        """Get the interface with the given individual address. Any interface type."""
+        return next(
+            (
+                interface
+                for interface in self.interfaces
+                if interface.individual_address == individual_address
+            ),
+            None,
+        )
+
+    def get_data_secure_group_keys(
+        self, receiver: IndividualAddress | None = None
+    ) -> dict[GroupAddress, bytes]:
+        """
+        Get data secure group keys.
+
+        If `receiver` is None, all data secure sending devices are returned.
+        Else the result is filtered by the given receiver.
+        """
+        ga_key_table = {
+            group_address.address: group_address.decrypted_key
+            for group_address in self.group_addresses
+            if group_address.decrypted_key is not None
+        }
+        if receiver is None:
+            return ga_key_table
+
+        rcv_interface = self.get_interface_by_individual_address(
+            individual_address=receiver
+        )
+        if rcv_interface is None:
+            return {}
+        return {
+            ga: key
+            for ga, key in ga_key_table.items()
+            if ga in rcv_interface.group_addresses
+        }
+
+    def get_data_secure_senders(self) -> dict[IndividualAddress, int]:
+        """
+        Get all data secure sending device addresses.
+
+        Sequence numbers are sourced from devices list or default to 0.
+        """
+        ia_seq_table: dict[IndividualAddress, int] = {}
+        for interface in self.interfaces:
+            for senders in interface.group_addresses.values():
+                ia_seq_table |= {sender: 0 for sender in senders}
+        # devices are only available if the full project was exported
+        for device in self.devices:
+            ia_seq_table[device.individual_address] = device.sequence_number
+        # TODO: check if this should default to 0 or if devices without a sequence number
+        # in keyfile should be excluded from the table (are there non-secure devices listed?)
+        return ia_seq_table
+
     def parse_xml(self, node: Document) -> None:
         """Parse all needed attributes from the given node map."""
         attributes = node.attributes
@@ -325,8 +390,7 @@ class Keyring(AttributeReader):
             if sub_node.nodeName == "Interface":
                 interface: XMLInterface = XMLInterface()
                 interface.parse_xml(sub_node)
-                if interface.password is not None:
-                    self.interfaces.append(interface)
+                self.interfaces.append(interface)
             if sub_node.nodeName == "Backbone":
                 backbone: XMLBackbone = XMLBackbone()
                 backbone.parse_xml(sub_node)
@@ -360,18 +424,20 @@ class Keyring(AttributeReader):
 
 
 async def load_keyring(
-    path: str, password: str, validate_signature: bool = True
+    path: str | os.PathLike[Any], password: str, validate_signature: bool = True
 ) -> Keyring:
     """Load a .knxkeys file from the given path in an executor."""
     return await asyncio.to_thread(
-        _load_keyring,
+        sync_load_keyring,
         path,
         password,
         validate_signature=validate_signature,
     )
 
 
-def _load_keyring(path: str, password: str, validate_signature: bool = True) -> Keyring:
+def sync_load_keyring(
+    path: str | os.PathLike[Any], password: str, validate_signature: bool = True
+) -> Keyring:
     """Load a .knxkeys file from the given path."""
 
     if validate_signature and not verify_keyring_signature(path, password):
@@ -432,7 +498,7 @@ class KeyringSAXContentHandler(ContentHandler):
         self.output.extend(value)
 
 
-def verify_keyring_signature(path: str, password: str) -> bool:
+def verify_keyring_signature(path: str | os.PathLike[Any], password: str) -> bool:
     """Verify the signature of the given knxkeys file."""
     handler = KeyringSAXContentHandler(password)
     signature: bytes


Test patch:
diff --git a/test/io_tests/knxip_interface_test.py b/test/io_tests/knxip_interface_test.py
index d1e33d3d..c1e0dc20 100644
--- a/test/io_tests/knxip_interface_test.py
+++ b/test/io_tests/knxip_interface_test.py
@@ -37,10 +37,18 @@ class TestKNXIPInterface:
         interface = knx_interface_factory(self.xknx, connection_config)
         with patch("xknx.io.KNXIPInterface._start_automatic") as start_automatic_mock:
             await interface.start()
-            start_automatic_mock.assert_called_once_with()
+            start_automatic_mock.assert_called_once_with(keyring=None)
             assert threading.active_count() == 1
 
         async def gateway_generator_mock(_):
+            secure_interface = GatewayDescriptor(
+                ip_addr="10.1.2.3",
+                port=3671,
+                supports_tunnelling_tcp=True,
+                supports_secure=True,
+            )
+            secure_interface.tunnelling_requires_secure = True
+            yield secure_interface
             yield GatewayDescriptor(
                 ip_addr="10.1.2.3", port=3671, supports_tunnelling_tcp=True
             )
@@ -55,6 +63,9 @@ class TestKNXIPInterface:
             "xknx.io.knxip_interface.GatewayScanner.async_scan",
             new=gateway_generator_mock,
         ), patch(
+            "xknx.io.KNXIPInterface._start_secure_tunnelling_tcp",
+            side_effect=InvalidSecureConfiguration("Error"),
+        ) as start_secure_tunnelling_tcp, patch(
             "xknx.io.KNXIPInterface._start_tunnelling_tcp",
             side_effect=CommunicationError("Error"),
         ) as start_tunnelling_tcp_mock, patch(
@@ -66,10 +77,49 @@ class TestKNXIPInterface:
         ) as start_routing_mock:
             with pytest.raises(CommunicationError):
                 await interface.start()
+            start_secure_tunnelling_tcp.assert_called_once()
             start_tunnelling_tcp_mock.assert_called_once()
             start_tunnelling_udp_mock.assert_called_once()
             start_routing_mock.assert_called_once()
 
+    async def test_start_automatic_with_keyring(self):
+        """Test starting with automatic mode and keyring."""
+        connection_config = ConnectionConfig(
+            secure_config=SecureConfig(
+                knxkeys_file_path=self.knxkeys_file,
+                knxkeys_password="password",
+            )
+        )
+        assert connection_config.connection_type == ConnectionType.AUTOMATIC
+        interface = knx_interface_factory(self.xknx, connection_config)
+
+        # in the test keyfile the only host is 1.0.0 - others shall be skipped
+        async def gateway_generator_mock(_):
+            yield GatewayDescriptor(
+                ip_addr="10.1.5.5",
+                port=3671,
+                supports_tunnelling_tcp=True,
+                individual_address=IndividualAddress("5.0.0"),
+            )
+            yield GatewayDescriptor(
+                ip_addr="10.1.0.0",
+                port=3671,
+                supports_tunnelling_tcp=True,
+                individual_address=IndividualAddress("1.0.0"),
+            )
+
+        with patch(
+            "xknx.io.knxip_interface.GatewayScanner.async_scan",
+            new=gateway_generator_mock,
+        ), patch(
+            "xknx.io.KNXIPInterface._start_tunnelling_tcp",
+        ) as start_tunnelling_tcp_mock:
+            await interface.start()
+            start_tunnelling_tcp_mock.assert_called_once_with(
+                gateway_ip="10.1.0.0",
+                gateway_port=3671,
+            )
+
     async def test_start_udp_tunnel_connection(self):
         """Test starting UDP tunnel connection."""
         # without gateway_ip automatic is called
@@ -171,7 +221,7 @@ class TestKNXIPInterface:
         ) as start_automatic_mock:
             interface = knx_interface_factory(self.xknx, connection_config)
             await interface.start()
-            start_automatic_mock.assert_called_once_with()
+            start_automatic_mock.assert_called_once_with(keyring=None)
 
     async def test_threaded_send_cemi(self):
         """Test sending cemi with threaded connection."""
diff --git a/test/secure_tests/data_secure_test.py b/test/secure_tests/data_secure_test.py
index 81339851..1c91fa13 100644
--- a/test/secure_tests/data_secure_test.py
+++ b/test/secure_tests/data_secure_test.py
@@ -14,7 +14,7 @@ from xknx.secure.data_secure_asdu import (
     SecurityALService,
     SecurityControlField,
 )
-from xknx.secure.keyring import Keyring, _load_keyring
+from xknx.secure.keyring import Keyring, sync_load_keyring
 from xknx.telegram import (
     GroupAddress,
     IndividualAddress,
@@ -63,7 +63,7 @@ class TestDataSecure:
         secure_test_keyfile = os.path.join(
             os.path.dirname(__file__), "resources/SecureTest.knxkeys"
         )
-        cls.secure_test_keyring = _load_keyring(secure_test_keyfile, "test")
+        cls.secure_test_keyring = sync_load_keyring(secure_test_keyfile, "test")
 
     def setup_method(self):
         """Setup test methods."""
@@ -83,8 +83,8 @@ class TestDataSecure:
         for ga_raw in [1024, 1027, 1028, 1029]:
             assert GroupAddress(ga_raw) in self.data_secure._group_key_table
 
-        assert len(self.data_secure._individual_address_table) == 3
-        for ia_raw in ["4.0.0", "4.0.9", "5.0.0"]:
+        assert len(self.data_secure._individual_address_table) == 5
+        for ia_raw in ["4.0.0", "4.0.1", "4.0.9", "5.0.0", "5.0.1"]:
             assert (
                 IndividualAddress(ia_raw) in self.data_secure._individual_address_table
             )
diff --git a/test/secure_tests/keyring_test.py b/test/secure_tests/keyring_test.py
index e70c9c9a..65c6b3f2 100644
--- a/test/secure_tests/keyring_test.py
+++ b/test/secure_tests/keyring_test.py
@@ -1,29 +1,35 @@
 """Unit test for keyring reader."""
 import os
+from pathlib import Path
 
 import pytest
 
 from xknx.exceptions.exception import InvalidSecureConfiguration
 from xknx.secure.keyring import (
+    InterfaceType,
     Keyring,
     XMLDevice,
     XMLInterface,
-    _load_keyring,
+    sync_load_keyring,
     verify_keyring_signature,
 )
-from xknx.telegram import IndividualAddress
+from xknx.telegram import GroupAddress, IndividualAddress
 
 
 class TestKeyRing:
     """Test class for keyring."""
 
-    keyring_test_file = os.path.join(
-        os.path.dirname(__file__), "resources/keyring.knxkeys"
-    )
-
-    testcase_file = os.path.join(
+    keyring_test_file = Path(__file__).parent / "resources/keyring.knxkeys"
+    testcase_file: str = os.path.join(
         os.path.dirname(__file__), "resources/testcase.knxkeys"
     )
+    special_chars_file = (
+        Path(__file__).parent / "resources/special_chars_secure_tunnel.knxkeys"
+    )
+    data_secure_ip = (
+        Path(__file__).parent / "resources/DataSecure_only_one_interface.knxkeys"
+    )
+    data_secure_usb = Path(__file__).parent / "resources/DataSecure_usb.knxkeys"
 
     @staticmethod
     def assert_interface(
@@ -37,9 +43,9 @@ class TestKeyRing:
 
         assert matched
 
-    def test_load_keyring(self):
+    def test_load_keyring_test(self):
         """Test load keyring from knxkeys file."""
-        keyring: Keyring = _load_keyring(self.keyring_test_file, "pwd")
+        keyring = sync_load_keyring(self.keyring_test_file, "pwd")
         TestKeyRing.assert_interface(keyring, "user4", IndividualAddress("1.1.4"))
         TestKeyRing.assert_interface(keyring, "@zvI1G&_", IndividualAddress("1.1.6"))
         TestKeyRing.assert_interface(keyring, "ZvDY-:g#", IndividualAddress("1.1.7"))
@@ -50,9 +56,9 @@ class TestKeyRing:
             "96f034fccf510760cbd63da0f70d4a9d"
         )
 
-    def test_load_keyring_real(self):
+    def test_load_testcase_file(self):
         """Test load keyring from knxkeys file."""
-        keyring: Keyring = _load_keyring(self.testcase_file, "password")
+        keyring = sync_load_keyring(self.testcase_file, "password")
         TestKeyRing.assert_interface(keyring, "user1", IndividualAddress("1.0.1"))
         TestKeyRing.assert_interface(keyring, "user2", IndividualAddress("1.0.11"))
         TestKeyRing.assert_interface(keyring, "user3", IndividualAddress("1.0.12"))
@@ -69,19 +75,131 @@ class TestKeyRing:
         assert device is not None
         assert device.decrypted_authentication == "authenticationcode"
 
+    def test_load_special_chars_file(self):
+        """Test load keyring from knxkeys file."""
+        keyring = sync_load_keyring(self.special_chars_file, "test")
+        TestKeyRing.assert_interface(keyring, "tunnel_2", IndividualAddress("1.0.2"))
+        TestKeyRing.assert_interface(keyring, "tunnel_3", IndividualAddress("1.0.3"))
+        TestKeyRing.assert_interface(keyring, "tunnel_4", IndividualAddress("1.0.4"))
+        TestKeyRing.assert_interface(keyring, "tunnel_5", IndividualAddress("1.0.5"))
+        TestKeyRing.assert_interface(keyring, "tunnel_6", IndividualAddress("1.0.6"))
+        assert keyring.backbone is None
+
+    def test_load_data_secure_ip(self):
+        """Test load keyring from knxkeys file."""
+        keyring = sync_load_keyring(self.data_secure_ip, "test")
+        assert len(keyring.interfaces) == 1
+        tunnel = keyring.interfaces[0]
+        assert tunnel is not None
+        assert tunnel.password is None
+        assert tunnel.decrypted_password is None
+        assert tunnel.user_id is None
+        assert tunnel.type is InterfaceType.TUNNELING
+        assert len(tunnel.group_addresses) == 3
+        assert tunnel.group_addresses[GroupAddress("0/0/1")] == [
+            IndividualAddress("1.0.1"),
+            IndividualAddress("1.0.2"),
+        ]
+        assert tunnel.group_addresses[GroupAddress("0/0/3")] == [
+            IndividualAddress("1.0.1"),
+            IndividualAddress("1.0.2"),
+        ]
+        assert tunnel.group_addresses[GroupAddress("31/7/255")] == []
+        assert keyring.backbone is None
+
+    def test_load_data_secure_usb(self):
+        """Test load keyring from knxkeys file."""
+        keyring = sync_load_keyring(self.data_secure_usb, "test")
+        assert len(keyring.interfaces) == 1
+        interface = keyring.interfaces[0]
+        assert interface is not None
+        assert interface.password is None
+        assert interface.decrypted_password is None
+        assert interface.user_id is None
+        assert interface.host is None
+        assert interface.type is InterfaceType.USB
+        assert len(interface.group_addresses) == 1
+        assert interface.group_addresses[GroupAddress("31/7/255")] == [
+            IndividualAddress("1.0.4")
+        ]
+        assert keyring.backbone is None
+
     def test_verify_signature(self):
         """Test signature verification."""
         assert verify_keyring_signature(self.keyring_test_file, "pwd")
         assert verify_keyring_signature(self.testcase_file, "password")
+        assert verify_keyring_signature(self.special_chars_file, "test")
 
     def test_invalid_signature(self):
         """Test invalid signature throws error."""
         with pytest.raises(InvalidSecureConfiguration):
-            _load_keyring(self.testcase_file, "wrong_password")
+            sync_load_keyring(self.testcase_file, "wrong_password")
 
     def test_raises_error(self):
         """Test raises error if password is wrong."""
         with pytest.raises(InvalidSecureConfiguration):
-            _load_keyring(
+            sync_load_keyring(
                 self.testcase_file, "wrong_password", validate_signature=False
             )
+
+    def test_keyring_get_methods_full(self):
+        """Test keyring get_* methods for full project export."""
+        keyring = sync_load_keyring(self.keyring_test_file, "pwd")
+        test_interfaces = keyring.get_tunnel_interfaces_by_host(
+            host=IndividualAddress("1.1.10")
+        )
+        assert len(test_interfaces) == 1
+        test_interface = test_interfaces[0]
+
+        test_device = keyring.get_device_by_interface(interface=test_interface)
+        assert test_device.individual_address == IndividualAddress("1.1.10")
+
+        test_interface = keyring.get_tunnel_interface_by_host_and_user_id(
+            host=IndividualAddress("1.1.0"), user_id=4
+        )
+        assert test_interface.individual_address == IndividualAddress("1.1.7")
+
+        test_interface = keyring.get_tunnel_interface_by_individual_address(
+            tunnelling_slot=IndividualAddress("1.1.8")
+        )
+        assert test_interface.user_id == 8
+
+        test_interface = keyring.get_tunnel_interface_by_individual_address(
+            tunnelling_slot=IndividualAddress("1.1.20")
+        )
+        assert test_interface.user_id is None
+        assert test_interface.host == IndividualAddress("1.1.10")
+        # this doesn't check for `type`, but there are no other than TUNNELLING interfaces in this keyring
+        test_interface = keyring.get_interface_by_individual_address(
+            individual_address=IndividualAddress("1.1.20")
+        )
+        assert test_interface.host == IndividualAddress("1.1.10")
+
+        full_ga_key_table = keyring.get_data_secure_group_keys()
+        assert len(full_ga_key_table) == 1
+
+        individual_ga_key_table = keyring.get_data_secure_group_keys(
+            receiver=IndividualAddress("1.1.7")
+        )
+        assert len(individual_ga_key_table) == 0
+
+        ia_seq_nums = keyring.get_data_secure_senders()
+        assert len(ia_seq_nums) == 5
+
+    def test_keyring_get_methods_one_interface(self):
+        """Test keyring get_* methods for partial export."""
+        keyring = sync_load_keyring(self.data_secure_ip, "test")
+
+        full_ga_key_table = keyring.get_data_secure_group_keys()
+        assert len(full_ga_key_table) == 3
+
+        individual_ga_key_table = keyring.get_data_secure_group_keys(
+            receiver=IndividualAddress("1.0.4")
+        )
+        assert len(individual_ga_key_table) == 3
+
+        ia_seq_nums = keyring.get_data_secure_senders()
+        assert ia_seq_nums == {
+            IndividualAddress("1.0.1"): 0,
+            IndividualAddress("1.0.2"): 0,
+        }
diff --git a/test/secure_tests/resources/DataSecure_only_one_interface.knxkeys b/test/secure_tests/resources/DataSecure_only_one_interface.knxkeys
new file mode 100644
index 00000000..79c9aaba
--- /dev/null
+++ b/test/secure_tests/resources/DataSecure_only_one_interface.knxkeys
@@ -0,0 +1,13 @@
+﻿<?xml version="1.0" encoding="utf-8"?>
+<Keyring Project="DataSecure_only" CreatedBy="ETS 5.7.7 (Build 1428)" Created="2023-02-06T21:17:09" Signature="FRVrpISVDd9IPYmEPHakqA==" xmlns="http://knx.org/xml/keyring/1">
+  <Interface IndividualAddress="1.0.4" Type="Tunneling" Host="1.0.3">
+    <Group Address="65535" Senders="" />
+    <Group Address="1" Senders="1.0.1 1.0.2" />
+    <Group Address="3" Senders="1.0.1 1.0.2" />
+  </Interface>
+  <GroupAddresses>
+    <Group Address="1" Key="gBIFuQX8V+WXe1rCpK4mZA==" />
+    <Group Address="3" Key="qZit1ss5dw6QP9JkX0mW3A==" />
+    <Group Address="65535" Key="JdtAoNfDRKQqkhnnRzw/hw==" />
+  </GroupAddresses>
+</Keyring>
\ No newline at end of file
diff --git a/test/secure_tests/resources/DataSecure_usb.knxkeys b/test/secure_tests/resources/DataSecure_usb.knxkeys
new file mode 100644
index 00000000..e1d86d0d
--- /dev/null
+++ b/test/secure_tests/resources/DataSecure_usb.knxkeys
@@ -0,0 +1,9 @@
+﻿<?xml version="1.0" encoding="utf-8"?>
+<Keyring Project="DataSecure_only" CreatedBy="ETS 5.7.7 (Build 1428)" Created="2023-02-07T15:07:21" Signature="3XSm3p3KJlIj428Ke26HNg==" xmlns="http://knx.org/xml/keyring/1">
+  <Interface IndividualAddress="1.0.12" Type="USB">
+    <Group Address="65535" Senders="1.0.4" />
+  </Interface>
+  <GroupAddresses>
+    <Group Address="65535" Key="71Mp7thP+Wac9vmynVke8Q==" />
+  </GroupAddresses>
+</Keyring>
\ No newline at end of file
diff --git a/test/secure_tests/resources/special_chars_secure_tunnel.knxkeys b/test/secure_tests/resources/special_chars_secure_tunnel.knxkeys
new file mode 100644
index 00000000..de8acfd8
--- /dev/null
+++ b/test/secure_tests/resources/special_chars_secure_tunnel.knxkeys
@@ -0,0 +1,11 @@
+﻿<?xml version="1.0" encoding="utf-8"?>
+<Keyring Project="Project name with special chars äüöÄÜÖßáâéèê?()|{}" CreatedBy="ETS 5.7.7 (Build 1428)" Created="2023-02-06T21:10:09" Signature="t2dSfNq331qP/90hMT6/KQ==" xmlns="http://knx.org/xml/keyring/1">
+  <Interface IndividualAddress="1.0.2" Type="Tunneling" Host="1.0.1" UserID="2" Password="l6dgsbqEJIXIGRVwlCHIVFErkI0G4k6Z+AJ2fDTgync=" Authentication="0nclVWobmO62CYaaGpDBk9ZB3yr6Az3uQfKxwgOtPj4=" />
+  <Interface IndividualAddress="1.0.3" Type="Tunneling" Host="1.0.1" UserID="3" Password="27vW02U989lmx7Wx7/Y8dKu3kaGJqNUrm036G86GkWE=" Authentication="Ge+fgeIQUHwR+EqOY4QJnNJ+sDbnZFePCV+C3SmHgt8=" />
+  <Interface IndividualAddress="1.0.4" Type="Tunneling" Host="1.0.1" UserID="4" Password="mRFZ0pI4v9To9df61RZEi6DDDeqybRf/Hi/K5ioCo+M=" Authentication="sOebBH7nNSKFzbO5JNUmgacB9swH8MoR6DyMfA/IQyM=" />
+  <Interface IndividualAddress="1.0.5" Type="Tunneling" Host="1.0.1" UserID="5" Password="pPstJS+Nrn4K5yWgADJev31am6u2Uv8w7lgdbcNNGMM=" Authentication="hoxZzhaqZslMzdBkEDYO+ZHl2xwOUvTULjjWFD6NgkA=" />
+  <Interface IndividualAddress="1.0.6" Type="Tunneling" Host="1.0.1" UserID="6" Password="w/xgBR2Qa+esLjBpzoGTuxtCHjbHWHFlIUWXSPijQ7o=" Authentication="DCXk2BUsPDd1xK/OU0Hv/3HJCDjVQEYS83yDeLUb32s=" />
+  <Devices>
+    <Device IndividualAddress="1.0.1" ToolKey="DZu9XTOfcfgw3ku3by4p0Q==" ManagementPassword="MCoY1uZUTW7YD5u2WBnsIC82feoSfeFUS58ScVKnuPU=" Authentication="srLCgGq2b1GrFk3hBogXpKJSBc6kTyt5QpKVUBo/ei8=" />
+  </Devices>
+</Keyring>
\ No newline at end of file