feat(community): mDNS advertisement via zeroconf — defaults OFF, opt-in per a11y audit

app/services/community/mdns.py

@@ -0,0 +1,72 @@
+# app/services/community/mdns.py
+# MIT License
+
+from __future__ import annotations
+
+import logging
+import socket
+
+logger = logging.getLogger(__name__)
+
+# Import deferred to avoid hard failure when zeroconf is not installed
+try:
+    from zeroconf import ServiceInfo, Zeroconf
+    _ZEROCONF_AVAILABLE = True
+except ImportError:
+    _ZEROCONF_AVAILABLE = False
+
+
+class KiwiMDNS:
+    """Advertise this Kiwi instance on the LAN via mDNS (_kiwi._tcp.local).
+
+    Defaults to disabled (enabled=False). User must explicitly opt in via the
+    Settings page. This matches the CF a11y requirement: no surprise broadcasting.
+
+    Usage:
+        mdns = KiwiMDNS(enabled=settings.MDNS_ENABLED, port=settings.PORT,
+                        feed_url=f"http://{hostname}:{settings.PORT}/api/v1/community/local-feed")
+        mdns.start()   # in lifespan startup
+        mdns.stop()    # in lifespan shutdown
+    """
+
+    SERVICE_TYPE = "_kiwi._tcp.local."
+
+    def __init__(self, enabled: bool, port: int, feed_url: str) -> None:
+        self._enabled = enabled
+        self._port = port
+        self._feed_url = feed_url
+        self._zc: "Zeroconf | None" = None
+        self._info: "ServiceInfo | None" = None
+
+    def start(self) -> None:
+        if not self._enabled:
+            logger.debug("mDNS advertisement disabled (user has not opted in)")
+            return
+        if not _ZEROCONF_AVAILABLE:
+            logger.warning("zeroconf package not installed — mDNS advertisement unavailable")
+            return
+
+        hostname = socket.gethostname()
+        service_name = f"kiwi-{hostname}.{self.SERVICE_TYPE}"
+        self._info = ServiceInfo(
+            type_=self.SERVICE_TYPE,
+            name=service_name,
+            port=self._port,
+            properties={
+                b"feed_url": self._feed_url.encode(),
+                b"version": b"1",
+            },
+            addresses=[socket.inet_aton("127.0.0.1")],
+        )
+        self._zc = Zeroconf()
+        self._zc.register_service(self._info)
+        logger.info("mDNS: advertising %s on port %d", service_name, self._port)
+
+    def stop(self) -> None:
+        if self._zc is None or self._info is None:
+            return
+        self._zc.unregister_service(self._info)
+        self._zc.close()
+        self._zc = None
+        self._info = None
+        logger.info("mDNS: advertisement stopped")

pyproject.toml

@@ -22,6 +22,8 @@ dependencies = [
     # HTTP clients
     "httpx>=0.27",
     "requests>=2.31",
+    # mDNS advertisement (optional; user must opt in)
+    "zeroconf>=0.131",
     # CircuitForge shared scaffold
     "circuitforge-core>=0.8.0",
 ]

tests/services/community/test_mdns.py

@@ -0,0 +1,39 @@
+# tests/services/community/test_mdns.py
+import pytest
+from unittest.mock import MagicMock, patch
+from app.services.community.mdns import KiwiMDNS
+
+
+def test_mdns_does_not_advertise_when_disabled():
+    """When enabled=False, KiwiMDNS does not register any zeroconf service."""
+    with patch("app.services.community.mdns.Zeroconf") as mock_zc:
+        mdns = KiwiMDNS(enabled=False, port=8512, feed_url="http://localhost:8512/api/v1/community/local-feed")
+        mdns.start()
+        mock_zc.assert_not_called()
+
+
+def test_mdns_advertises_when_enabled():
+    with patch("app.services.community.mdns.Zeroconf") as mock_zc_cls:
+        with patch("app.services.community.mdns.ServiceInfo") as mock_si:
+            mock_zc = MagicMock()
+            mock_zc_cls.return_value = mock_zc
+            mdns = KiwiMDNS(enabled=True, port=8512, feed_url="http://localhost:8512/api/v1/community/local-feed")
+            mdns.start()
+            mock_zc.register_service.assert_called_once()
+
+
+def test_mdns_stop_unregisters_when_enabled():
+    with patch("app.services.community.mdns.Zeroconf") as mock_zc_cls:
+        with patch("app.services.community.mdns.ServiceInfo"):
+            mock_zc = MagicMock()
+            mock_zc_cls.return_value = mock_zc
+            mdns = KiwiMDNS(enabled=True, port=8512, feed_url="http://localhost:8512/api/v1/community/local-feed")
+            mdns.start()
+            mdns.stop()
+            mock_zc.unregister_service.assert_called_once()
+            mock_zc.close.assert_called_once()
+
+
+def test_mdns_stop_is_noop_when_not_started():
+    mdns = KiwiMDNS(enabled=False, port=8512, feed_url="http://localhost/feed")
+    mdns.stop()  # must not raise