fix: rebuild listen socket safely - derive family from desired set, clean up on wrap failure, defend v6only read

bdraco · bdraco · commit d05d3e2278ca · 2026-06-21T23:32:27.000-05:00
diff --git a/src/zeroconf/_engine.py b/src/zeroconf/_engine.py
@@ -74,8 +74,13 @@ def _listen_socket_supports(
     if listen_socket.family != socket.AF_INET6:
         # An IPv4 interface on an AF_INET socket.
         return True
-    # An IPv4 interface on an AF_INET6 socket: only when it is dual-stack.
-    return not listen_socket.getsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY)
+    # An IPv4 interface on an AF_INET6 socket: only when it is dual-stack. If
+    # the option can't be read (as Windows rejects some IPv6 getsockopts),
+    # assume supported so a read failure can't drive a rebuild loop.
+    try:
+        return not listen_socket.getsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY)
+    except OSError:
+        return True
 
 
 class AsyncEngine:
@@ -220,7 +225,7 @@ async def async_update_interfaces(
         if listen_socket is not None and any(
             not _listen_socket_supports(listen_socket, interface) for interface in desired.values()
         ):
-            listen_socket = await self._async_rebuild_listen_socket(ip_version, apple_p2p, desired, current)
+            listen_socket = await self._async_rebuild_listen_socket(apple_p2p, desired, current)
             listen_transport = self._listen_transport
 
         for bind_address, wrapped in current.items():
@@ -302,7 +307,6 @@ def _async_close_sender(self, wrapped: _WrappedTransport, listen_socket: socket.
 
     async def _async_rebuild_listen_socket(
         self,
-        ip_version: IPVersion,
         apple_p2p: bool,
         desired: dict[tuple[str, int], str | tuple[tuple[str, int, int], int]],
         current: dict[tuple[str, int], _WrappedTransport],
@@ -312,20 +316,38 @@ async def _async_rebuild_listen_socket(
         The listen socket's family is otherwise fixed at construction; this
         lets an instance start receiving a newly added address family. Only
         called when a desired interface cannot be joined on the current listen
-        socket. Interfaces that are staying are re-joined on the new socket,
+        socket. The replacement family is derived from the desired set (not the
+        requested ip_version, which an explicit list can contradict) so it
+        always covers every desired interface and never needs an immediate
+        re-rebuild. Interfaces that are staying are re-joined on the new socket,
         and the old socket is closed (releasing its memberships).
         """
-        new_listen = new_socket(bind_addr=("",), ip_version=ip_version, apple_p2p=apple_p2p)
+        has_v6 = any(isinstance(interface, tuple) for interface in desired.values())
+        has_v4 = any(not isinstance(interface, tuple) for interface in desired.values())
+        if has_v4 and has_v6:
+            family_version = IPVersion.All
+        elif has_v6:
+            family_version = IPVersion.V6Only
+        else:
+            family_version = IPVersion.V4Only
+        new_listen = new_socket(bind_addr=("",), ip_version=family_version, apple_p2p=apple_p2p)
         if new_listen is None:
-            raise RuntimeError("Failed to create a listen socket for the requested interface family")
-        for bind_address, interface in desired.items():
-            if bind_address in current:
-                add_multicast_member(new_listen, interface)
+            raise RuntimeError("Failed to create a listen socket for the new interface family")
+        try:
+            for bind_address, interface in desired.items():
+                if bind_address in current:
+                    add_multicast_member(new_listen, interface)
+            new_reader = await self._async_wrap_socket(new_listen, is_sender=False)
+        except Exception:
+            # Endpoint creation failed; close the unadopted socket (and its
+            # joins) rather than leak it, mirroring _async_add_interface.
+            new_listen.close()
+            raise
         # A rebuild is only entered with a live listen socket, so the old
         # transport is always present.
         old_listen_transport = self._listen_transport
         assert old_listen_transport is not None
-        self._listen_transport = await self._async_wrap_socket(new_listen, is_sender=False)
+        self._listen_transport = new_reader
         old_transport = old_listen_transport.transport
         self._async_remove_transport(old_transport)
         old_transport.close()
diff --git a/tests/test_interface_update.py b/tests/test_interface_update.py
@@ -108,6 +108,10 @@ def test_listen_socket_supports_family() -> None:
     assert _listen_socket_supports(v6_sock, "1.2.3.4") is True
     v6_sock.getsockopt.return_value = 1  # IPV6_V6ONLY on -> v6-only
     assert _listen_socket_supports(v6_sock, "1.2.3.4") is False
+    # An unreadable option (some platforms) is treated as supported so it
+    # can't drive a rebuild loop.
+    v6_sock.getsockopt.side_effect = OSError
+    assert _listen_socket_supports(v6_sock, "1.2.3.4") is True
 
 
 @pytest.mark.asyncio
@@ -544,6 +548,93 @@ async def test_update_interfaces_rebuild_failure_raises(aiozc_loopback: AsyncZer
         await engine.async_update_interfaces(["unused"], IPVersion.V6Only, False)
 
 
+@pytest.mark.asyncio
+async def test_update_interfaces_rebuild_closes_socket_on_wrap_failure(
+    aiozc_loopback: AsyncZeroconf,
+) -> None:
+    """If wrapping the new listen socket fails, it is closed rather than leaked."""
+    engine = aiozc_loopback.zeroconf.engine
+    await aiozc_loopback.zeroconf.async_wait_for_start()
+    old_listen = engine._listen_transport
+    new_listen_sock = Mock()
+    new_listen_sock.family = socket.AF_INET6
+
+    with (
+        patch.object(_engine, "normalize_interface_choice", return_value=[(("fe80::1", 0, 0), 1)]),
+        patch.object(_engine, "new_socket", return_value=new_listen_sock),
+        patch.object(_engine, "add_multicast_member", return_value=True),
+        patch.object(_engine.AsyncEngine, "_async_wrap_socket", new=AsyncMock(side_effect=OSError("boom"))),
+        pytest.raises(OSError),
+    ):
+        await engine.async_update_interfaces(["unused"], IPVersion.V6Only, False)
+
+    # The unadopted socket was closed, and the old listen socket is untouched.
+    new_listen_sock.close.assert_called_once()
+    assert engine._listen_transport is old_listen
+
+
+@pytest.mark.asyncio
+async def test_update_interfaces_rebuild_family_matches_desired_set(
+    aiozc_loopback: AsyncZeroconf,
+) -> None:
+    """The rebuilt listen socket's family is derived from the desired set, not ip_version."""
+    engine = aiozc_loopback.zeroconf.engine
+    await aiozc_loopback.zeroconf.async_wait_for_start()
+    new_listen_sock = Mock()
+    new_listen_sock.family = socket.AF_INET
+
+    async def fake_wrap(sock: object, is_sender: bool) -> _WrappedTransport:
+        wrapped = _make_wrapped(("wrapped", 0), transport=Mock())
+        (engine.senders if is_sender else engine.readers).append(wrapped)
+        return wrapped
+
+    with (
+        patch.object(_engine, "normalize_interface_choice", return_value=["192.168.1.5"]),
+        patch.object(_engine, "_listen_socket_supports", return_value=False),  # force a rebuild
+        patch.object(_engine, "new_socket", return_value=new_listen_sock) as mock_new_socket,
+        patch.object(_engine, "add_multicast_member", return_value=True),
+        patch.object(_engine, "new_respond_socket", return_value=Mock()),
+        patch.object(_engine, "drop_multicast_member"),
+        patch.object(_engine.AsyncEngine, "_async_wrap_socket", new=AsyncMock(side_effect=fake_wrap)),
+    ):
+        # ip_version says V6Only, but the desired set is all IPv4, so the
+        # rebuilt socket is IPv4 (covers the set; no immediate re-rebuild).
+        await engine.async_update_interfaces(["unused"], IPVersion.V6Only, False)
+
+    mock_new_socket.assert_called_once()
+    assert mock_new_socket.call_args.kwargs["ip_version"] is IPVersion.V4Only
+
+
+@pytest.mark.asyncio
+async def test_update_interfaces_rebuilds_real_listen_socket(aiozc_loopback: AsyncZeroconf) -> None:
+    """End to end: a family change builds a real dual-stack listen socket and closes the old one."""
+    engine = aiozc_loopback.zeroconf.engine
+    await aiozc_loopback.zeroconf.async_wait_for_start()
+    old_listen = engine._listen_transport
+    assert old_listen is not None
+    assert old_listen.sock.family == socket.AF_INET  # V4Only loopback instance
+    old_underlying = old_listen.transport
+
+    v6 = (("fe80::1", 0, 0), 1)
+    # Real new_socket + _async_wrap_socket run; only membership joins and the
+    # (unbindable) v6 responder are stubbed so no real multicast is exercised.
+    with (
+        patch.object(_engine, "normalize_interface_choice", return_value=["127.0.0.1", v6]),
+        patch.object(_engine, "add_multicast_member", return_value=True),
+        patch.object(_engine, "new_respond_socket", return_value=None),
+    ):
+        await engine.async_update_interfaces(["unused"], IPVersion.All, False)
+
+    new_listen = engine._listen_transport
+    assert new_listen is not None
+    assert new_listen is not old_listen
+    assert new_listen.sock.family == socket.AF_INET6  # rebuilt to a dual-stack socket
+    # The old listen socket was closed and removed; no duplicate remains.
+    assert old_underlying.is_closing()
+    assert old_listen not in engine.readers
+    assert sum(1 for r in engine.readers if r is new_listen) == 1
+
+
 @pytest.mark.asyncio
 async def test_close_sender_closes_transport_when_drop_raises(aiozc_loopback: AsyncZeroconf) -> None:
     """A non-benign group-leave error still releases the transport."""
