feat: add async_update_interfaces to rescan network interfaces at runtime

bdraco · bdraco · commit cc6dd24cf055 · 2026-06-21T18:22:47.000-05:00
diff --git a/src/zeroconf/_core.py b/src/zeroconf/_core.py
@@ -199,6 +199,11 @@ def __init__(
 
         self.unicast = unicast
         self._use_asyncio = use_asyncio
+        # Retained so async_update_interfaces can re-run create_sockets /
+        # normalize_interface_choice against the live interface set later.
+        self._interfaces = interfaces
+        self._ip_version = ip_version
+        self._apple_p2p = apple_p2p
         listen_socket, respond_sockets = create_sockets(interfaces, unicast, ip_version, apple_p2p=apple_p2p)
         log.debug("Listen socket %s, respond sockets %s", listen_socket, respond_sockets)
 
@@ -406,6 +411,39 @@ async def async_update_service(self, info: ServiceInfo) -> Awaitable:
         self.registry.async_update(info)
         return asyncio.ensure_future(self._async_broadcast_service(info, _REGISTER_TIME, None))
 
+    def update_interfaces(self, interfaces: InterfacesType | None = None) -> None:
+        """Rescan network interfaces and reconcile the sockets in use.
+
+        While it is not expected during normal operation,
+        this function may raise EventLoopBlocked if the underlying
+        call to `async_update_interfaces` cannot be completed.
+        """
+        assert self.loop is not None
+        run_coro_with_timeout(
+            self.async_update_interfaces(interfaces),
+            self.loop,
+            _REGISTER_TIME * _REGISTER_BROADCASTS,
+        )
+
+    async def async_update_interfaces(self, interfaces: InterfacesType | None = None) -> None:
+        """Rescan network interfaces and reconcile the sockets in use.
+
+        Adds sockets for interfaces that appeared, drops sockets for
+        interfaces that disappeared, and re-announces existing
+        registrations on the resulting senders. ``interfaces`` defaults to
+        the choice passed at construction; pass a new value to switch.
+        """
+        if interfaces is not None:
+            self._interfaces = interfaces
+        await self.async_wait_for_start()
+        await self.engine.async_update_interfaces(self._interfaces, self._ip_version, self._apple_p2p)
+        await asyncio.gather(
+            *[
+                self._async_broadcast_service(info, _REGISTER_TIME, None)
+                for info in self.registry.async_get_service_infos()
+            ]
+        )
+
     async def async_get_service_info(
         self,
         type_: str,
diff --git a/src/zeroconf/_engine.py b/src/zeroconf/_engine.py
@@ -30,6 +30,14 @@
 
 from ._record_update import RecordUpdate
 from ._utils.asyncio import get_running_loop, run_coro_with_timeout
+from ._utils.net import (
+    InterfacesType,
+    IPVersion,
+    add_multicast_member,
+    drop_multicast_member,
+    new_respond_socket,
+    normalize_interface_choice,
+)
 from ._utils.time import current_time_millis
 from .const import _CACHE_CLEANUP_INTERVAL
 
@@ -43,12 +51,40 @@
 _CLOSE_TIMEOUT = 3000  # ms
 
 
+def _strip_zone(address: str) -> str:
+    """Drop a ``%zone`` suffix from an IPv6 address string."""
+    percent = address.find("%")
+    return address[:percent] if percent != -1 else address
+
+
+def _interface_bind_address(interface: str | tuple[tuple[str, int, int], int]) -> str:
+    """Return the bind address an interface choice maps to, for diffing."""
+    if isinstance(interface, tuple):
+        return _strip_zone(interface[0][0])
+    return interface
+
+
+def _wrapped_bind_address(wrapped: _WrappedTransport) -> str:
+    """Return the bound address of a sender transport, for diffing."""
+    return _strip_zone(wrapped.sock_name[0])
+
+
+def _wrapped_interface(wrapped: _WrappedTransport) -> str | tuple[tuple[str, int, int], int]:
+    """Reconstruct the interface representation for a sender, for membership drops."""
+    sock_name = wrapped.sock_name
+    if wrapped.is_ipv6:
+        scope_id = cast(int, sock_name[3]) if len(sock_name) > 3 else 0
+        return ((_strip_zone(sock_name[0]), 0, 0), scope_id)
+    return cast(str, sock_name[0])
+
+
 class AsyncEngine:
     """An engine wraps sockets in the event loop."""
 
     __slots__ = (
         "_cleanup_timer",
         "_listen_socket",
+        "_listen_transport",
         "_respond_sockets",
         "_setup_task",
         "loop",
@@ -72,6 +108,7 @@ def __init__(
         self.senders: list[_WrappedTransport] = []
         self.running_future: asyncio.Future[bool | None] | None = None
         self._listen_socket = listen_socket
+        self._listen_transport: _WrappedTransport | None = None
         self._respond_sockets = respond_sockets
         self._cleanup_timer: asyncio.TimerHandle | None = None
         self._setup_task: asyncio.Task[None] | None = None
@@ -114,14 +151,19 @@ async def _async_create_endpoints(self) -> None:
                 lambda: AsyncListener(self.zc),  # type: ignore[arg-type, return-value]
                 sock=s,
             )
+            datagram_transport = cast(asyncio.DatagramTransport, transport)
+            reader = make_wrapped_transport(datagram_transport)
             # Register the wrapped transport before releasing the engine's
             # handle so a concurrent shutdown always sees ``s`` in exactly
             # one place; do not add an ``await`` between these two steps.
             self.protocols.append(cast(AsyncListener, protocol))
-            self.readers.append(make_wrapped_transport(cast(asyncio.DatagramTransport, transport)))
+            self.readers.append(reader)
             if s in sender_sockets:
-                self.senders.append(make_wrapped_transport(cast(asyncio.DatagramTransport, transport)))
+                self.senders.append(make_wrapped_transport(datagram_transport))
             if s is self._listen_socket:
+                # Keep a handle to the shared listen socket so interface
+                # rescans can add/drop multicast memberships on it.
+                self._listen_transport = reader
                 self._listen_socket = None
             if s in self._respond_sockets:
                 self._respond_sockets.remove(s)
@@ -134,12 +176,93 @@ def _async_remove_listener(self, listener: AsyncListener) -> None:
         instead of raising ``EHOSTUNREACH`` on every send forever.
         """
         wrapped = listener.transport
-        transport = wrapped.transport if wrapped is not None else None
         if listener in self.protocols:
             self.protocols.remove(listener)
-        if transport is not None:
-            self.readers = [w for w in self.readers if w.transport is not transport]
-            self.senders = [w for w in self.senders if w.transport is not transport]
+        if wrapped is not None:
+            self._async_drop_transport_wrappers(wrapped.transport)
+
+    def _async_drop_transport_wrappers(self, transport: asyncio.DatagramTransport) -> None:
+        """Drop every reader/sender wrapper that holds ``transport``."""
+        self.readers = [w for w in self.readers if w.transport is not transport]
+        self.senders = [w for w in self.senders if w.transport is not transport]
+
+    async def async_update_interfaces(
+        self,
+        interfaces: InterfacesType,
+        ip_version: IPVersion,
+        apple_p2p: bool,
+    ) -> None:
+        """Reconcile sender/reader sockets to the live interface set.
+
+        Adds a per-interface responder socket for each interface that
+        appeared and tears down the socket for each interface that
+        disappeared, diffing on the bound address. The shared listen
+        socket (including the Default single-family dual-use socket) is
+        never torn down here.
+        """
+        assert self.loop is not None
+        normalized = normalize_interface_choice(interfaces, ip_version)
+        desired = {_interface_bind_address(interface): interface for interface in normalized}
+        current = {_wrapped_bind_address(wrapped): wrapped for wrapped in self.senders}
+        listen_transport = self._listen_transport
+        listen_socket = listen_transport.sock if listen_transport is not None else None
+
+        for bind_address, wrapped in current.items():
+            if bind_address in desired:
+                continue
+            if listen_transport is not None and wrapped.transport is listen_transport.transport:
+                # The shared listen / dual-use socket is not a per-interface
+                # sender; leaving the group or closing it would break receive.
+                continue
+            self._async_close_sender(wrapped, listen_socket)
+
+        for bind_address, interface in desired.items():
+            if bind_address in current:
+                continue
+            await self._async_add_interface(interface, listen_socket, apple_p2p)
+
+    async def _async_add_interface(
+        self,
+        interface: str | tuple[tuple[str, int, int], int],
+        listen_socket: socket.socket | None,
+        apple_p2p: bool,
+    ) -> None:
+        """Join the multicast group and adopt a responder socket for one interface."""
+        # A unicast instance has no listen socket, so membership is only
+        # ever managed when ``listen_socket`` is present.
+        if listen_socket is not None and not add_multicast_member(listen_socket, interface):
+            return
+        respond_socket = new_respond_socket(interface, apple_p2p=apple_p2p, unicast=self.zc.unicast)
+        if respond_socket is None:
+            if listen_socket is not None:
+                drop_multicast_member(listen_socket, interface)
+            return
+        await self._async_adopt_respond_socket(respond_socket)
+
+    async def _async_adopt_respond_socket(self, sock: socket.socket) -> None:
+        """Wrap a freshly created per-interface responder socket as reader + sender."""
+        assert self.loop is not None
+        transport, protocol = await self.loop.create_datagram_endpoint(  # type: ignore[type-var]
+            lambda: AsyncListener(self.zc),  # type: ignore[arg-type, return-value]
+            sock=sock,
+        )
+        datagram_transport = cast(asyncio.DatagramTransport, transport)
+        # No ``await`` between wrapping and registering so a concurrent
+        # shutdown always sees the transport in exactly one place.
+        self.protocols.append(cast(AsyncListener, protocol))
+        self.readers.append(make_wrapped_transport(datagram_transport))
+        self.senders.append(make_wrapped_transport(datagram_transport))
+
+    def _async_close_sender(self, wrapped: _WrappedTransport, listen_socket: socket.socket | None) -> None:
+        """Drop a per-interface sender's wrappers/protocol and close its transport."""
+        transport = wrapped.transport
+        self.protocols = [
+            p for p in self.protocols if p.transport is None or p.transport.transport is not transport
+        ]
+        self._async_drop_transport_wrappers(transport)
+        if listen_socket is not None:
+            drop_multicast_member(listen_socket, _wrapped_interface(wrapped))
+        transport.close()
 
     def _async_cache_cleanup(self) -> None:
         """Periodic cache cleanup."""
diff --git a/src/zeroconf/_utils/net.py b/src/zeroconf/_utils/net.py
@@ -401,6 +401,34 @@ def add_multicast_member(
     return True
 
 
+def drop_multicast_member(
+    listen_socket: socket.socket,
+    interface: str | tuple[tuple[str, int, int], int],
+) -> bool:
+    """Leave the mDNS multicast group on an interface; inverse of add_multicast_member."""
+    # This is based on assumptions in normalize_interface_choice
+    is_v6 = isinstance(interface, tuple)
+    log.debug("Dropping %r (socket %d) from multicast group", interface, listen_socket.fileno())
+    try:
+        if is_v6:
+            try:
+                mdns_addr6_bytes = socket.inet_pton(socket.AF_INET6, _MDNS_ADDR6)
+            except OSError:
+                return False
+            iface_bin = struct.pack("@I", cast(int, interface[1]))
+            listen_socket.setsockopt(_IPPROTO_IPV6, socket.IPV6_LEAVE_GROUP, mdns_addr6_bytes + iface_bin)
+        else:
+            _value = socket.inet_aton(_MDNS_ADDR) + socket.inet_aton(cast(str, interface))
+            listen_socket.setsockopt(socket.IPPROTO_IP, socket.IP_DROP_MEMBERSHIP, _value)
+    except OSError as e:
+        # The kernel drops memberships automatically when an interface
+        # disappears, so a stale leave is expected to fail benignly.
+        if get_errno(e) in {errno.EADDRNOTAVAIL, errno.EINVAL, errno.ENODEV, errno.ENOPROTOOPT}:
+            return False
+        raise
+    return True
+
+
 def new_respond_socket(
     interface: str | tuple[tuple[str, int, int], int],
     apple_p2p: bool = False,
diff --git a/src/zeroconf/asyncio.py b/src/zeroconf/asyncio.py
@@ -224,6 +224,15 @@ async def async_update_service(self, info: ServiceInfo) -> Awaitable:
         """
         return await self.zeroconf.async_update_service(info)
 
+    async def async_update_interfaces(self, interfaces: InterfacesType | None = None) -> None:
+        """Rescan network interfaces and reconcile the sockets in use.
+
+        Adds sockets for interfaces that appeared, drops sockets for
+        interfaces that disappeared, and re-announces existing
+        registrations on the resulting senders.
+        """
+        await self.zeroconf.async_update_interfaces(interfaces)
+
     async def async_close(self) -> None:
         """Ends the background threads, and prevent this instance from
         servicing further queries."""
diff --git a/tests/test_interface_update.py b/tests/test_interface_update.py
diff --git a/tests/utils/test_net.py b/tests/utils/test_net.py
