diff --git a/app/services/community/mdns.py b/app/services/community/mdns.py new file mode 100644 index 0000000..148780a --- /dev/null +++ b/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") diff --git a/pyproject.toml b/pyproject.toml index 1928abb..736139b 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/tests/services/community/test_mdns.py b/tests/services/community/test_mdns.py new file mode 100644 index 0000000..8cfd4d3 --- /dev/null +++ b/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