diff --git a/libervia/backend/plugins/plugin_misc_serverless/__init__.py b/libervia/backend/plugins/plugin_misc_serverless/__init__.py new file mode 100644 index 00000000..b962381c --- /dev/null +++ b/libervia/backend/plugins/plugin_misc_serverless/__init__.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 + +# Libervia plugin for Serverless Connections. +# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +from typing import TYPE_CHECKING, cast + +from libp2p.crypto.ed25519 import Ed25519PrivateKey +from libp2p.peer.id import ID +from twisted.internet import defer +from twisted.internet import reactor +from twisted.internet.interfaces import IReactorCore + +from libervia.backend.core.constants import Const as C +from libervia.backend.core.i18n import _ +from libervia.backend.core.log import getLogger +from libervia.backend.memory import persistent + +from .mdns import DEFAULT_PORT, MDNSManager +from .models import DiscoverExtra, DiscoverNodes, DiscoveredPeerInfo, Node, ServerlessInfo + +if TYPE_CHECKING: + from libervia.backend.core.main import LiberviaBackend + +log = getLogger(__name__) + +IMPORT_NAME = "SERVERLESS" + +PLUGIN_INFO = { + C.PI_NAME: "Serverless XMPP", + C.PI_IMPORT_NAME: IMPORT_NAME, + C.PI_TYPE: C.PLUG_TYPE_MISC, + C.PI_MODES: [C.PLUG_MODE_CLIENT], + C.PI_PROTOCOLS: [], + C.PI_DEPENDENCIES: [], + C.PI_RECOMMENDATIONS: [], + C.PI_MAIN: "Serverless", + C.PI_HANDLER: "no", + C.PI_DESCRIPTION: _("""Run Libervia without server."""), +} + + +class Serverless: + """Run Libervia without server.""" + + def __init__(self, host: "LiberviaBackend") -> None: + self.host = host + log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") + + self.peer_id: str = "" + self.mdns: MDNSManager | None = None + self.discovered_peers: dict[str, DiscoveredPeerInfo] = {} + + host.bridge.add_method( + "serverless_discover", + ".plugin", + in_sign="s", + out_sign="s", + method=self._discover, + async_=True, + ) + host.bridge.add_method( + "serverless_infos_get", + ".plugin", + in_sign="", + out_sign="s", + method=self._infos, + async_=True, + ) + # args: Serialised Node + host.bridge.add_signal("serverless_peer_found", ".plugin", signature="s") + # args: device_id + host.bridge.add_signal("serverless_peer_lost", ".plugin", signature="s") + + cast(IReactorCore, reactor).callLater( + 0, lambda: defer.ensureDeferred(self.post_init()) + ) + + async def post_init(self) -> None: + """Load or generate the device key and start mDNS discovery.""" + storage = persistent.PersistentBinaryDict(IMPORT_NAME) + await storage.load() + + device_key_hex = storage.get("device_key") + if device_key_hex is not None: + log.debug("Loaded existing device key from storage.") + device_key = bytes.fromhex(device_key_hex) + else: + log.info("No device key found, generating a new one…") + private_key = Ed25519PrivateKey.new() + device_key = private_key.to_bytes() + device_key_hex = device_key.hex() + await storage.aset("device_key", device_key_hex) + + private_key = Ed25519PrivateKey.from_bytes(device_key) + device_id = ID.from_pubkey(private_key.get_public_key()).to_bytes() + self.peer_id = ID(device_id).to_base58() + log.info(f"Device ID: {self.peer_id}") + + self.mdns = MDNSManager( + self.peer_id, + port=DEFAULT_PORT, + on_peer_found=self._on_peer_found, + on_peer_lost=self._on_peer_lost, + ) + log.info("Starting mDNS manager in background…") + cast(IReactorCore, reactor).callInThread(self.mdns.start) + + def _infos(self) -> defer.Deferred[str]: + service_name = self.mdns.service_name if self.mdns else "" + peer_count = len(self.discovered_peers) + return defer.succeed( + ServerlessInfo( + device_id=self.peer_id, service_name=service_name, peer_count=peer_count + ).model_dump_json() + ) + + def _on_peer_found( + self, device_id: str, name: str, addrs: list[str], metadata: dict[str, str] + ) -> None: + log.info(f"Peer found: {name} ({device_id})") + self.discovered_peers[device_id] = { + "name": name, + "device_id": device_id, + "addrs": addrs, + "metadata": metadata, + } + node = Node(device_id=device_id, name=name, addrs=addrs, metadata=metadata) + self.host.bridge.serverless_peer_found(node.model_dump_json()) + + def _on_peer_lost(self, device_id: str) -> None: + log.info(f"Peer lost: {device_id}") + self.discovered_peers.pop(device_id, None) + self.host.bridge.serverless_peer_lost(device_id) + + async def unload(self) -> None: + """Stop mDNS discovery and clean up resources.""" + log.info("Unloading serverless plugin…") + if self.mdns is not None: + log.info("Stopping mDNS manager…") + await self.mdns.stop() + log.info("mDNS manager stopped.") + + def _discover(self, extra_s: str) -> defer.Deferred[str]: + extra = DiscoverExtra.model_validate_json(extra_s) + d = defer.ensureDeferred(self.discover(extra)) + d.addCallback(lambda result: result.model_dump_json()) + d = cast(defer.Deferred[str], d) + return d + + async def discover(self, extra: DiscoverExtra) -> DiscoverNodes: + nodes = [ + Node( + device_id=device_id, + name=info["name"], + addrs=info["addrs"], + metadata=info["metadata"], + ) + for device_id, info in self.discovered_peers.items() + ] + return DiscoverNodes(nodes) diff --git a/libervia/backend/plugins/plugin_misc_serverless/mdns.py b/libervia/backend/plugins/plugin_misc_serverless/mdns.py new file mode 100644 index 00000000..1002bcae --- /dev/null +++ b/libervia/backend/plugins/plugin_misc_serverless/mdns.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 + +# Libervia plugin for Serverless XMPP +# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +from collections.abc import Callable +import socket +from typing import cast + +from twisted.internet import reactor +from twisted.internet.interfaces import IReactorCore +from zeroconf import ServiceBrowser, ServiceInfo, ServiceListener, Zeroconf +from zeroconf.asyncio import AsyncZeroconf + +from libervia.backend.core.log import getLogger +from libervia.backend.plugins.plugin_misc_serverless.models import DiscoveredPeerInfo +from libervia.backend.tools.utils import aio + +log = getLogger(__name__) + +SERVICE_TYPE = "_xmpp-serverless._tcp.local." +DEFAULT_PORT = 5299 + + +class MDNSManager: + """mDNS-based peer discovery. + + Broadcasts our peer on the local network and listens for others. + """ + + def __init__( + self, + peer_id: str, + port: int = DEFAULT_PORT, + on_peer_found: ( + Callable[[str, str, list[str], dict[str, str]], None] | None + ) = None, + on_peer_lost: Callable[[str], None] | None = None, + ): + """Init the manager. + + @param peer_id: Peer ID (from libP2P). + @param port: TCP port to advertise. + @param on_peer_found: callback called when a peer is found. + @param on_peer_lost: callback called when a peer is lost. + """ + self.peer_id = peer_id + self.port = port + self.on_peer_found = on_peer_found + self.on_peer_lost = on_peer_lost + self.zeroconf = Zeroconf() + # Human-readable device name; zeroconf will auto-rename on collision + device_name = f"Libervia-{socket.gethostname()}" + self.service_name = f"{device_name}.{SERVICE_TYPE}" + log.info( + f"MDNSManager created: peer_id={peer_id}, port={port}, " + f"service_name={self.service_name}" + ) + + def start(self) -> None: + """Register this peer and start listening for others.""" + log.info(f"Registering mDNS service: {self.service_name} port={self.port}") + self.service_info = ServiceInfo( + type_=SERVICE_TYPE, + name=self.service_name, + port=self.port, + properties={b"id": self.peer_id.encode()}, + server=f"{socket.gethostname()}.local.", + ) + self.zeroconf.register_service(self.service_info, allow_name_change=True) + # Zeroconf may have renamed on collision, we have to check the final name. + self.service_name = self.service_info.name + log.info(f"mDNS service registered as: {self.service_info.name}") + + self.listener = PeersListener( + zeroconf=self.zeroconf, + service_type=SERVICE_TYPE, + service_name=self.service_name, + on_peer_found=self.on_peer_found, + on_peer_lost=self.on_peer_lost, + ) + log.info("ServiceBrowser started for peer discovery.") + + @aio + async def stop(self) -> None: + """Unregister this peer and clean up resources. + + Uses AsyncZeroconf to await the unregistration (goodbye broadcasts), + then calls close() to shut down the zeroconf instance. + Decorated with @aio so asyncio.sleep() inside the broadcasts works + correctly (scheduled via asyncio.ensure_future, not ensureDeferred). + Returns a Deferred. + """ + log.info("Cleaning mDNS services.") + if hasattr(self, "listener"): + self.listener.browser.cancel() + log.debug("ServiceBrowser cancelled.") + if hasattr(self, "service_info"): + async_zc = AsyncZeroconf(zc=self.zeroconf) + # async_unregister_service returns an Awaitable (a Task) that must be + # double-awaited: first await retrieves the Task, second awaits it. + unregister_task = await async_zc.async_unregister_service( + self.service_info, + ) + await unregister_task + log.debug("mDNS service unregistered via AsyncZeroconf.") + self.zeroconf.close() + log.debug("Zeroconf closed.") + + +class PeersListener(ServiceListener): + """Listen for other libp2p peers on the local network. + + Callbacks are fired from zeroconf's ServiceBrowser thread. All model updates must be + sent to the Twisted reactor thread with reactor.callFromThread(). + """ + + def __init__( + self, + zeroconf: Zeroconf, + service_type: str, + service_name: str, + on_peer_found: ( + Callable[[str, str, list[str], dict[str, str]], None] | None + ) = None, + on_peer_lost: Callable[[str], None] | None = None, + ) -> None: + """Initialize the listener. + + @param zeroconf: Zeroconf instance. + @param service_type: mDNS service type to listen for. + @param service_name: our own service name (skipped on discovery). + @param on_peer_found: callback called when a peer is found. + @param on_peer_lost: callback called when a peer is lost. + """ + self.zeroconf = zeroconf + self.service_type = service_type + self.service_name = service_name + self.on_peer_found = on_peer_found + self.on_peer_lost = on_peer_lost + self.browser = ServiceBrowser(zeroconf, service_type, listener=self) + + def add_service(self, zc: Zeroconf, type_: str, name: str) -> None: + """Handle a newly discovered mDNS service. + + @param zc: Zeroconf instance. + @param type_: service type. + @param name: fully qualified service name. + """ + if name == self.service_name: + return + peer_info = self._extract_peer_info(zc, type_, name) + if peer_info and self.on_peer_found: + log.debug(f"mDNS peer discovered: {peer_info['device_id']}") + cast(IReactorCore, reactor).callFromThread( + self.on_peer_found, + peer_info["device_id"], + peer_info["name"], + peer_info["addrs"], + peer_info["metadata"], + ) + + def remove_service(self, zc: Zeroconf, type_: str, name: str) -> None: + """Handle a removed mDNS service. + + @param zc: Zeroconf instance. + @param type_: service type. + @param name: fully qualified service name. + """ + if name == self.service_name: + return + peer_info = self._extract_peer_info(zc, type_, name) + device_id = peer_info["device_id"] if peer_info else None + if device_id is not None and self.on_peer_lost: + log.debug(f"mDNS peer removed: {device_id}") + cast(IReactorCore, reactor).callFromThread(self.on_peer_lost, device_id) + + def update_service(self, zc: Zeroconf, type_: str, name: str) -> None: + """Handle an updated mDNS service by removing and re-adding it. + + @param zc: Zeroconf instance. + @param type_: service type. + @param name: fully qualified service name. + """ + self.remove_service(zc, type_, name) + self.add_service(zc, type_, name) + + def _extract_peer_info( + self, zc: Zeroconf, type_: str, name: str + ) -> DiscoveredPeerInfo | None: + """Extract peer info from a zeroconf service. + + Handles both IPv4 and IPv6 addresses. + Deduplicates addresses that may appear from multiple interfaces. + + @return: mDNS peer infos, or None on failure. + """ + try: + info = zc.get_service_info(type_, name, timeout=5000) + if not info: + log.debug(f"No service info for {name!r}") + return None + pid_bytes = info.properties.get(b"id") + if not pid_bytes: + log.warning(f'No "id" property for {name!r}, {info.properties=}') + return None + # Extract human-readable name from full service name + # e.g. "Libervia-example-2._xmpp-serverless._tcp.local." -> "Libervia-example" + peer_name = name.split(".")[0] + # Build libp2p multiaddr strings, deduplicating by IP string. + # See https://github.com/multiformats/multiaddr + seen_ips: set[str] = set() + addrs = [] + for addr in info.addresses: + if len(addr) == 16: + # IPv6 + ip_str = socket.inet_ntop(socket.AF_INET6, addr) + else: + # 4 bytes = IPv4 + ip_str = socket.inet_ntop(socket.AF_INET, addr) + if ip_str not in seen_ips: + seen_ips.add(ip_str) + af = "ip6" if len(addr) == 16 else "ip4" + addrs.append(f"/{af}/{ip_str}/tcp/{info.port}") + metadata = { + key.decode(): value.decode() + for key, value in info.properties.items() + if value is not None + } + return { + "name": peer_name, + "device_id": pid_bytes.decode(), + "addrs": addrs, + "metadata": metadata, + } + except Exception: + log.exception(f"extract_peer_info failed for {name!r}") + return None diff --git a/libervia/backend/plugins/plugin_misc_serverless/models.py b/libervia/backend/plugins/plugin_misc_serverless/models.py new file mode 100644 index 00000000..e61dbd26 --- /dev/null +++ b/libervia/backend/plugins/plugin_misc_serverless/models.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python3 + +# Libervia plugin for Serverless XMPP +# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +from collections.abc import Iterator +from typing import TypedDict + +from pydantic import BaseModel, Field, RootModel + + +class DiscoverExtra(BaseModel): + # XXX: only a placeholder for now + pass + + +class DiscoveredPeerInfo(TypedDict): + """Info stored for a discovered peer. + + @ivar addrs: list of libp2p multiaddr strings, e.g. "/ip4/192.168.1.10/tcp/5299". + See https://github.com/multiformats/multiaddr + """ + + name: str + device_id: str + addrs: list[str] + metadata: dict[str, str] + + +class ServerlessInfo(BaseModel): + """Serverless device information.""" + + device_id: str = Field(description="ID of the device") + service_name: str = Field(description="mDNS service name") + peer_count: int = Field(description="Number of discovered peers") + + +class Node(BaseModel): + """Discovered peer node. + + @param device_id: Identifier for the discovered device. + @param name: Human-readable name of the peer. + @param addrs: List of addresses to reach the peer. + @param metadata: metadata extracted from the discovery protocol. + """ + + device_id: str = Field(description="Identifier for the discovered device") + name: str = Field(description="Human-readable name of the peer") + addrs: list[str] = Field( + description="List of addresses (multiaddr format), see https://github.com/multiformats/multiaddr" + ) + metadata: dict[str, str] = Field( + default_factory=dict, description="Key-value metadata from the discovery protocol" + ) + + +class DiscoverNodes(RootModel): + """List of discovered nodes.""" + + root: list[Node] + + def __iter__(self) -> Iterator[Node]: # type: ignore + return iter(self.root) + + def __len__(self) -> int: + return len(self.root) + + def __getitem__(self, index: int) -> Node: + return self.root[index]