Fix ping node and retrieval of IP addresses (#585)

Co-authored-by: Stefan Agner <stefan@agner.ch>

Author: marcelveldt

matter_server/client/client.py

@@ -257,16 +257,54 @@ async def ping_node(self, node_id: int) -> NodePingResult:
             await self.send_command(APICommand.PING_NODE, node_id=node_id),
         )
 
+    async def get_node_ip_addresses(
+        self, node_id: int, prefer_cache: bool = True, scoped: bool = False
+    ) -> list[str]:
+        """Return the currently known (scoped) IP-adress(es)."""
+        if TYPE_CHECKING:
+            assert self.server_info is not None
+        if self.server_info.schema_version >= 8:
+            return cast(
+                list[str],
+                await self.send_command(
+                    APICommand.GET_NODE_IP_ADRESSES,
+                    require_schema=8,
+                    node_id=node_id,
+                    prefer_cache=prefer_cache,
+                    scoped=scoped,
+                ),
+            )
+        # alternative method of fetching ip addresses by enumerating NetworkInterfaces
+        node = self.get_node(node_id)
+        attribute = Clusters.GeneralDiagnostics.Attributes.NetworkInterfaces
+        network_interface: Clusters.GeneralDiagnostics.Structs.NetworkInterface
+        ip_addresses: list[str] = []
+        for network_interface in node.get_attribute_value(
+            0, cluster=None, attribute=attribute
+        ):
+            # ignore invalid/non-operational interfaces
+            if not network_interface.isOperational:
+                continue
+            # enumerate ipv4 and ipv6 addresses
+            for ipv4_address_hex in network_interface.IPv4Addresses:
+                ipv4_address = convert_ip_address(ipv4_address_hex)
+                ip_addresses.append(ipv4_address)
+            for ipv6_address_hex in network_interface.IPv6Addresses:
+                ipv6_address = convert_ip_address(ipv6_address_hex, True)
+                ip_addresses.append(ipv6_address)
+            break
+        return ip_addresses
+
     async def node_diagnostics(self, node_id: int) -> NodeDiagnostics:
         """Gather diagnostics for the given node."""
         # pylint: disable=too-many-statements
         node = self.get_node(node_id)
+        ip_addresses = await self.get_node_ip_addresses(node_id)
         # grab some details from the first (operational) network interface
         network_type = NetworkType.UNKNOWN
         mac_address = None
         attribute = Clusters.GeneralDiagnostics.Attributes.NetworkInterfaces
         network_interface: Clusters.GeneralDiagnostics.Structs.NetworkInterface
-        ip_addresses: list[str] = []
         for network_interface in node.get_attribute_value(
             0, cluster=None, attribute=attribute
         ):
@@ -292,13 +330,6 @@ async def node_diagnostics(self, node_id: int) -> NodeDiagnostics:
                 # unknown interface: ignore
                 continue
             mac_address = convert_mac_address(network_interface.hardwareAddress)
-            # enumerate ipv4 and ipv6 addresses
-            for ipv4_address_hex in network_interface.IPv4Addresses:
-                ipv4_address = convert_ip_address(ipv4_address_hex)
-                ip_addresses.append(ipv4_address)
-            for ipv6_address_hex in network_interface.IPv6Addresses:
-                ipv6_address = convert_ip_address(ipv6_address_hex, True)
-                ip_addresses.append(ipv6_address)
             break
         # get thread/wifi specific info
         node_type = NodeType.UNKNOWN

matter_server/common/const.py

@@ -2,4 +2,4 @@
 
 # schema version is used to determine compatibility between server and client
 # bump schema if we add new features and/or make other (breaking) changes
-SCHEMA_VERSION = 7
+SCHEMA_VERSION = 8

matter_server/common/models.py

@@ -45,6 +45,7 @@ class APICommand(str, Enum):
     READ_ATTRIBUTE = "read_attribute"
     WRITE_ATTRIBUTE = "write_attribute"
     PING_NODE = "ping_node"
+    GET_NODE_IP_ADRESSES = "get_node_ip_addresses"
 
 
 EventCallBackType = Callable[[EventType, Any], None]

matter_server/server/device_controller.py

@@ -10,7 +10,7 @@
 from functools import partial
 import logging
 import time
-from typing import TYPE_CHECKING, Any, Awaitable, Callable, Iterable, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Callable, Iterable, TypeVar, cast
 
 from chip.ChipDeviceCtrl import DeviceProxyWrapper
 from chip.clusters import Attribute, Objects as Clusters
@@ -20,7 +20,6 @@
 from zeroconf import IPVersion, ServiceStateChange, Zeroconf
 from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf
 
-from matter_server.common.helpers.util import convert_ip_address
 from matter_server.common.models import CommissionableNodeData, CommissioningParameters
 from matter_server.server.helpers.attributes import parse_attributes_from_read_result
 from matter_server.server.helpers.utils import ping_ip
@@ -38,7 +37,6 @@
     dataclass_from_dict,
     dataclass_to_dict,
     parse_attribute_path,
-    parse_value,
 )
 from ..common.models import (
     APICommand,
@@ -97,6 +95,7 @@ def __init__(
         self._nodes_in_setup: set[int] = set()
         self._mdns_last_seen: dict[int, float] = {}
         self._nodes: dict[int, MatterNodeData] = {}
+        self._last_known_ip_addresses: dict[int, list[str]] = {}
         self._last_subscription_attempt: dict[int, int] = {}
         self.wifi_credentials_set: bool = False
         self.thread_credentials_set: bool = False
@@ -679,14 +678,12 @@ async def subscribe_attribute(
     async def ping_node(self, node_id: int) -> NodePingResult:
         """Ping node on the currently known IP-adress(es)."""
         result: NodePingResult = {}
-        # the node's ip addresses are stored in the GeneralDiagnostics cluster
-        attribute = Clusters.GeneralDiagnostics.Attributes.NetworkInterfaces
-        attr_path = f"0/{attribute.cluster_id}/{attribute.attribute_id}"
         node = self._nodes.get(node_id)
         if node is None:
             raise NodeNotExists(
                 f"Node {node_id} does not exist or is not yet interviewed"
             )
+        node_logger = LOGGER.getChild(f"[node {node_id}]")
 
         battery_powered = (
             node.attributes.get(ROUTING_ROLE_ATTRIBUTE_PATH, 0)
@@ -696,41 +693,71 @@ async def ping_node(self, node_id: int) -> NodePingResult:
         async def _do_ping(ip_address: str) -> None:
             """Ping IP and add to result."""
             timeout = 10 if battery_powered else 2
-            result[ip_address] = await ping_ip(ip_address, timeout)
-
-        # The network interfaces attribute contains a list of network interfaces.
-        # For regular nodes this is just a single interface but we iterate them all anyway.
-        # Create a list of tasks so we can do multiple pings simultanuous.
-        # NOTE: Upgrade this to a TaskGroup once we bump our minimal python version.
-        attr_data = cast(list[dict[str, Any]], node.attributes.get(attr_path))
-        tasks: list[Awaitable] = []
-        for network_interface_data in attr_data:
-            network_interface: Clusters.GeneralDiagnostics.Structs.NetworkInterface = (
-                parse_value(
-                    "network_interface",
-                    network_interface_data,
-                    Clusters.GeneralDiagnostics.Structs.NetworkInterface,
+            if "%" in ip_address:
+                # ip address contains an interface index
+                clean_ip, interface_idx = ip_address.split("%", 1)
+                node_logger.debug(
+                    "Pinging address %s (using interface %s)", clean_ip, interface_idx
                 )
-            )
-            # ignore invalid/non-operational interfaces
-            if not network_interface.isOperational:
-                continue
-            if network_interface.type in (
-                Clusters.GeneralDiagnostics.Enums.InterfaceTypeEnum.kUnspecified,
-                Clusters.GeneralDiagnostics.Enums.InterfaceTypeEnum.kUnknownEnumValue,
-            ):
-                continue
+            else:
+                clean_ip = ip_address
+                node_logger.debug("Pinging address %s", clean_ip)
+            result[clean_ip] = await ping_ip(ip_address, timeout)
 
-            # enumerate ipv4 and ipv6 addresses
-            for ipv4_address_hex in network_interface.IPv4Addresses:
-                ipv4_address = convert_ip_address(ipv4_address_hex)
-                tasks.append(_do_ping(ipv4_address))
-            for ipv6_address_hex in network_interface.IPv6Addresses:
-                ipv6_address = convert_ip_address(ipv6_address_hex, True)
-                tasks.append(_do_ping(ipv6_address))
+        ip_addresses = await self.get_node_ip_addresses(
+            node_id, prefer_cache=False, scoped=True
+        )
+        tasks = [_do_ping(x) for x in ip_addresses]
+        # TODO: replace this gather with a taskgroup once we bump our py version
         await asyncio.gather(*tasks)
+
+        # retrieve the currently connected/used address which is used
+        # by the sdk for communicating with the device
+        if TYPE_CHECKING:
+            assert self.chip_controller is not None
+        if sdk_result := await self._call_sdk(
+            self.chip_controller.GetAddressAndPort, nodeid=node_id
+        ):
+            active_address = sdk_result[0]
+            node_logger.info(
+                "The SDK is communicating with the device using %s", active_address
+            )
+            if active_address not in result and node.available:
+                # if the sdk is connected to a node, treat the address as pingable
+                result[active_address] = True
+
         return result
 
+    @api_command(APICommand.GET_NODE_IP_ADRESSES)
+    async def get_node_ip_addresses(
+        self, node_id: int, prefer_cache: bool = False, scoped: bool = False
+    ) -> list[str]:
+        """Return the currently known (scoped) IP-adress(es)."""
+        cached_info = self._last_known_ip_addresses.get(node_id, [])
+        if prefer_cache and cached_info:
+            return cached_info if scoped else [x.split("%")[0] for x in cached_info]
+        node = self._nodes.get(node_id)
+        if node is None:
+            raise NodeNotExists(
+                f"Node {node_id} does not exist or is not yet interviewed"
+            )
+        node_logger = LOGGER.getChild(f"[node {node_id}]")
+        # query mdns for all IP's
+        # ensure both fabric id and node id have 16 characters (prefix with zero's)
+        mdns_name = f"{self.compressed_fabric_id:0{16}X}-{node_id:0{16}X}.{MDNS_TYPE_OPERATIONAL_NODE}"
+        info = AsyncServiceInfo(MDNS_TYPE_OPERATIONAL_NODE, mdns_name)
+        if TYPE_CHECKING:
+            assert self._aiozc is not None
+        if not await info.async_request(self._aiozc.zeroconf, 3000):
+            node_logger.info(
+                "Node could not be discovered on the network, returning cached IP's"
+            )
+            return cached_info
+        ip_adresses = info.parsed_scoped_addresses(IPVersion.All)
+        # cache this info for later use
+        self._last_known_ip_addresses[node_id] = ip_adresses
+        return ip_adresses if scoped else [x.split("%")[0] for x in ip_adresses]
+
     async def _subscribe_node(self, node_id: int) -> None:
         """
         Subscribe to all node state changes/events for an individual node.
@@ -989,6 +1016,8 @@ async def _setup_node(self, node_id: int) -> None:
             # prevent duplicate setup actions
             return
         self._nodes_in_setup.add(node_id)
+        # pre-cache ip-addresses
+        await self.get_node_ip_addresses(node_id)
         try:
             # (re)interview node (only) if needed
             node_data = self._nodes[node_id]