Scope link-local IPv6 addresses received from WebSocket

matter_server/server/__main__.py

@@ -58,6 +58,12 @@
     default=None,
     help="Log file to write to (optional).",
 )
+parser.add_argument(
+    "--primary-interface",
+    type=str,
+    default=None,
+    help="Primary network interface for link-local addresses (optional).",
+)
 
 args = parser.parse_args()
 
@@ -85,7 +91,11 @@ def main() -> None:
 
     # Init server
     server = MatterServer(
-        args.storage_path, int(args.vendorid), int(args.fabricid), int(args.port)
+        args.storage_path,
+        int(args.vendorid),
+        int(args.fabricid),
+        int(args.port),
+        args.primary_interface,
     )
 
     async def handle_stop(loop: asyncio.AbstractEventLoop) -> None:

matter_server/server/device_controller.py

@@ -259,6 +259,8 @@ async def commission_on_network(
             raise RuntimeError("Device Controller not initialized.")
 
         node_id = self._get_next_node_id()
+        if ip_addr is not None:
+            ip_addr = self.server.scope_ipv6_lla(ip_addr)
 
         attempts = 0
         # we retry commissioning a few times as we've seen devices in the wild
@@ -281,10 +283,10 @@ async def commission_on_network(
                     filter=filter,
                 )
             else:
-                # Ip-adress provided, use CommissionIP method
                 LOGGER.info(
-                    "Starting Matter commissioning with IP using Node ID %s (attempt %s/%s).",
+                    "Starting Matter commissioning using Node ID %s and IP %s (attempt %s/%s).",
                     node_id,
+                    ip_addr,
                     attempts,
                     MAX_COMMISSION_RETRIES,
                 )

matter_server/server/server.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import asyncio
+import ipaddress
 import logging
 from typing import Any, Callable, Set
 import weakref
@@ -61,12 +62,14 @@ def __init__(
         vendor_id: int,
         fabric_id: int,
         port: int,
+        primary_interface: str | None,
     ) -> None:
         """Initialize the Matter Server."""
         self.storage_path = storage_path
         self.vendor_id = vendor_id
         self.fabric_id = fabric_id
         self.port = port
+        self.primary_interface = primary_interface
         self.logger = logging.getLogger(__name__)
         self.app = web.Application()
         self.loop: asyncio.AbstractEventLoop | None = None
@@ -166,6 +169,32 @@ def signal_event(self, evt: EventType, data: Any = None) -> None:
             else:
                 callback(evt, data)
 
+    def scope_ipv6_lla(self, ip_addr: str) -> str:
+        """Scope IPv6 link-local addresses to primary interface.
+
+        IPv6 link-local addresses received through the websocket might have no
+        scope_id or a scope_id which isn't valid on this device. Just assume the
+        device is connected on the primary interface.
+        """
+        ip_addr_parsed = ipaddress.ip_address(ip_addr)
+        if not ip_addr_parsed.is_link_local:
+            return ip_addr
+
+        if ip_addr_parsed.scope_id is not None:
+            # This type of IPv6 manipulation is not supported by the ipaddress lib
+            ip_addr = ip_addr.split("%")[0]
+
+        # Rely on host OS routing table
+        if self.primary_interface is None:
+            return ip_addr
+
+        self.logger.debug(
+            "Setting scope of link-local IP address %s to %s",
+            ip_addr,
+            self.primary_interface,
+        )
+        return ip_addr + "%" + self.primary_interface
+
     def register_api_command(
         self,
         command: str,

pyproject.toml

@@ -28,12 +28,12 @@ dependencies = [
   "coloredlogs",
   "dacite",
   "orjson",
-  "home-assistant-chip-clusters==2023.12.0"
+  "home-assistant-chip-clusters==2024.1.0"
 ]
 
 [project.optional-dependencies]
 server = [
-  "home-assistant-chip-core==2023.12.0",
+  "home-assistant-chip-core==2024.1.0",
   "cryptography==41.0.7"
 ]
 test = [

tests/server/test_server.py

@@ -99,7 +99,7 @@ def fetch_certificates_fixture() -> Generator[MagicMock, None, None]:
 @pytest.fixture(name="server")
 async def server_fixture() -> AsyncGenerator[MatterServer, None]:
     """Yield a server."""
-    server = MatterServer("test_storage_path", 1234, 5678, 5580)
+    server = MatterServer("test_storage_path", 1234, 5678, 5580, None)
     await server.start()
     yield server
     await server.stop()