erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin serverless: first draft, peer discovery on LAN:

Browse source 
Implement peer discovery on LAN using mDNS (via `zeroconf`).
A Peer ID is generated (via `libP2P`) per instance, and announced on backend startup.
Meanwhile, other peers are discovered on LAN. A signal is send on peer found and another
on peer lost.

There a bridge methods to find already discovered peers, and to get infos on serverless
(our own ID, adverised service name, and number of peers found).

`Zeroconf` runs in background in its own thread.

rel 479
This commit is contained in:
Goffi 2026-07-16 19:38:22 +02:00
commit 2ae02c8e4a
3 changed files with 507 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

174
libervia/backend/plugins/plugin_misc_serverless/__init__.py Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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)

251
libervia/backend/plugins/plugin_misc_serverless/mdns.py Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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

82
libervia/backend/plugins/plugin_misc_serverless/models.py Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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]