From 5fc9cdf79d8fd1808c0b2d445d9c5cf0d58ad268 Mon Sep 17 00:00:00 2001 From: Gabby Morgan Date: Thu, 28 May 2026 01:46:07 -0500 Subject: [PATCH] user can send LXMF message to create a new LXST channel --- lxst_relay.py | 458 +++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 419 insertions(+), 39 deletions(-) diff --git a/lxst_relay.py b/lxst_relay.py index c209147..eb75947 100644 --- a/lxst_relay.py +++ b/lxst_relay.py @@ -1,13 +1,16 @@ #!/usr/bin/env python3 """ -lxst-relay - Group voice channel bridge for Reticulum Telephony. +lxst-relay - Group voice channel bridge for Reticulum Telephony + with NomadNet LXMF bot for channel management. -Allows multiple lxst users to call into a shared voice channel -and talk together in a conference bridge style. Audio from each -participant is mixed and redistributed to all others. +Allows multiple users to call into a shared voice channel. +Users can create new channels by messaging the bot via LXMF +with ``/create_channel ``. Each channel is its own +Reticulum node with a unique identity. """ import RNS +import LXMF import os import time import signal as signal_module @@ -30,6 +33,10 @@ FIELD_SIGNALLING = 0x00 FIELD_FRAMES = 0x01 +# ====================================================================== +# Audio pipeline components (unchanged) +# ====================================================================== + class FrameCaptureSink(LocalSink): """Stores the latest decoded float32 audio frame from a participant.""" @@ -106,7 +113,7 @@ class ConferenceSource(LocalSource): class Participant: - """A remote caller connected to the relay.""" + """A remote caller connected to a channel.""" def __init__(self, link, identity): self.link = link @@ -224,12 +231,234 @@ class VoiceChannel: return mixed -class LXSTRelay(SignallingReceiver): - """Conference bridge for Reticulum Telephony. +# ====================================================================== +# Dynamic channel node – each is its own Reticulum identity +# ====================================================================== - Registers as a standard ``lxst.telephony`` destination so any - lxst client can call in. Callers are placed into a shared - ``VoiceChannel`` and their audio is mixed together. +class ChannelNode(SignallingReceiver): + """A voice channel that operates as its own Reticulum node. + + Has an independent ``RNS.Identity``, announces its own + ``lxst.telephony`` destination, and manages its own + ``VoiceChannel`` instance. + """ + + def __init__(self, identity, name, admins, frame_time_ms=60): + super().__init__() + self.identity = identity + self.name = name + self.admins = admins # set[bytes] – identity hashes + self.frame_time_ms = frame_time_ms + self.voice_channel = VoiceChannel(name=name, frame_time_ms=frame_time_ms) + + self.destination = RNS.Destination( + identity, + RNS.Destination.IN, + RNS.Destination.SINGLE, + APP_NAME, + PRIMITIVE_NAME, + ) + self.destination.set_proof_strategy(RNS.Destination.PROVE_NONE) + self.destination.set_link_established_callback( + self._incoming_link_established + ) + + # -- Link lifecycle ------------------------------------------------ + + def _incoming_link_established(self, link): + link.is_incoming = True + link.is_outgoing = False + link.answered = False + link.is_terminating = False + + link.set_remote_identified_callback(self._caller_identified) + link.set_link_closed_callback(self._link_closed) + + self.signal(Signalling.STATUS_AVAILABLE, link) + + def _caller_identified(self, link, identity): + participant = Participant(link, identity) + + self.signal(Signalling.STATUS_RINGING, link) + time.sleep(0.5) + + self.signal(Signalling.STATUS_CONNECTING, link) + self.voice_channel.add_participant(participant, relay=self) + self.signal(Signalling.STATUS_ESTABLISHED, link) + + RNS.log( + f"Call established on channel '{self.name}' with " + f"{RNS.prettyhexrep(identity.hash)}, " + f"{self.voice_channel.get_participant_count()} participant(s)", + RNS.LOG_NOTICE, + ) + + def _link_closed(self, link): + self.voice_channel.remove_participant(link.link_id) + + # -- SignallingReceiver ------------------------------------------- + + def signalling_received(self, signals, source): + for signal in signals: + if signal >= Signalling.PREFERRED_PROFILE: + profile = signal - Signalling.PREFERRED_PROFILE + RNS.log( + f"Channel '{self.name}' got preferred profile " + f"0x{profile:02x}", + RNS.LOG_DEBUG, + ) + + # -- Public API --------------------------------------------------- + + def announce(self): + self.destination.announce() + + def cleanup(self): + for p in list(self.voice_channel.participants.values()): + p.ended = True + if p.conf_source: + p.conf_source.stop() + if p.link_source: + p.link_source.stop() + if p.link and p.link.status == RNS.Link.ACTIVE: + p.link.teardown() + self.voice_channel.participants.clear() + + +# ====================================================================== +# Channel manager – persists channels to disk +# ====================================================================== + +class ChannelManager: + """Creates, loads, and tracks ``ChannelNode`` instances.""" + + def __init__(self, storage_dir): + self.channels_dir = os.path.join(storage_dir, "channels") + os.makedirs(self.channels_dir, exist_ok=True) + self.channels = {} # name -> ChannelNode + + # -- Creation ------------------------------------------------------ + + def create_channel(self, name, creator_identity_hash): + """Create a new channel with its own identity. + + Returns ``(ChannelNode, None)`` on success or + ``(None, error_message)`` on failure. + """ + if name in self.channels: + return None, f"Channel '{name}' already exists" + + if not name or "/" in name or ".." in name: + return None, "Invalid channel name" + + chan_dir = os.path.join(self.channels_dir, name) + try: + os.makedirs(chan_dir, exist_ok=False) + except FileExistsError: + return None, f"Channel '{name}' already exists" + + identity = RNS.Identity() + identity.to_file(os.path.join(chan_dir, "identity")) + + admins = {creator_identity_hash} + with open(os.path.join(chan_dir, "admins"), "w") as f: + f.write(RNS.hexrep(creator_identity_hash, delimit=False) + "\n") + + node = ChannelNode(identity, name, admins) + self.channels[name] = node + node.announce() + + RNS.log( + f"Created channel '{name}' with identity " + f"{RNS.hexrep(identity.hash, delimit=False)}", + RNS.LOG_NOTICE, + ) + return node, None + + # -- Persistence --------------------------------------------------- + + def load_all(self): + """Restore channels from disk (called at startup).""" + if not os.path.isdir(self.channels_dir): + return + for entry in sorted(os.listdir(self.channels_dir)): + chan_dir = os.path.join(self.channels_dir, entry) + if not os.path.isdir(chan_dir): + continue + + identity_path = os.path.join(chan_dir, "identity") + admin_path = os.path.join(chan_dir, "admins") + if not os.path.isfile(identity_path): + continue + + identity = RNS.Identity.from_file(identity_path) + if identity is None: + RNS.log( + f"Could not load identity for channel '{entry}', " + f"skipping", + RNS.LOG_ERROR, + ) + continue + + admins = set() + if os.path.isfile(admin_path): + with open(admin_path) as f: + for line in f: + line = line.strip() + if line: + try: + admins.add(bytes.fromhex(line)) + except ValueError: + pass + + node = ChannelNode(identity, entry, admins) + self.channels[entry] = node + RNS.log( + f"Restored channel '{entry}' " + f"({RNS.hexrep(identity.hash, delimit=False)})", + RNS.LOG_NOTICE, + ) + + # -- Queries ------------------------------------------------------- + + def is_admin(self, channel_name, identity_hash): + node = self.channels.get(channel_name) + if node is None: + return False + return identity_hash in node.admins + + def is_admin_of_any(self, identity_hash): + for node in self.channels.values(): + if identity_hash in node.admins: + return True + return False + + # -- Cleanup ------------------------------------------------------- + + def cleanup_all(self): + for node in self.channels.values(): + node.cleanup() + self.channels.clear() + + def announce_all(self): + for node in self.channels.values(): + node.announce() + + +# ====================================================================== +# Main relay – telephony + LXMF bot +# ====================================================================== + +class LXSTRelay(SignallingReceiver): + """Conference bridge with LXMF bot for channel management. + + Registers two destinations on the Reticulum network: + + * ``lxst.telephony`` – the default / lobby channel that anyone + can call directly. + * ``lxmf.delivery`` – an LXMF bot that responds to + ``/create_channel ``, creating a fresh Reticulum node for + each channel and assigning the creator as admin. """ def __init__( @@ -253,6 +482,7 @@ class LXSTRelay(SignallingReceiver): os.makedirs(self.configdir, exist_ok=True) + # ---- Identity ---- identity_path = os.path.join(self.configdir, "identity") if os.path.isfile(identity_path): self.identity = RNS.Identity.from_file(identity_path) @@ -277,6 +507,7 @@ class LXSTRelay(SignallingReceiver): configdir=rnsconfigdir, loglevel=3 + verbosity ) + # ---- Default telephony destination (lobby) ---- self.destination = RNS.Destination( self.identity, RNS.Destination.IN, @@ -291,15 +522,49 @@ class LXSTRelay(SignallingReceiver): self.channel = VoiceChannel(name=channel_name) - RNS.log( - f"Relay listening on " - f"{RNS.prettyhexrep(self.destination.hash)}", - RNS.LOG_NOTICE, + # ---- LXMF bot ---- + self.lxmf_storage = os.path.join(self.configdir, "lxmf") + os.makedirs(self.lxmf_storage, exist_ok=True) + + self.lxmf_router = LXMF.LXMRouter( + identity=self.identity, + storagepath=self.lxmf_storage, + ) + self.bot_destination = self.lxmf_router.register_delivery_identity( + self.identity, + display_name="LXST Channel Bot", + ) + self.lxmf_router.register_delivery_callback( + self._lxmf_delivery_callback ) - # ------------------------------------------------------------------ - # Link lifecycle - # ------------------------------------------------------------------ + # ---- Channel manager (dynamic nodes) ---- + self.channel_manager = ChannelManager(self.configdir) + self.channel_manager.load_all() + + # ---- Startup announce ---- + self.destination.announce() + self.channel_manager.announce_all() + + RNS.log( + f"Relay lobby: {RNS.prettyhexrep(self.destination.hash)}", + RNS.LOG_NOTICE, + ) + RNS.log( + f"Bot (LXMF): {RNS.prettyhexrep(self.bot_destination.hash)}", + RNS.LOG_NOTICE, + ) + if self.channel_manager.channels: + for name, node in self.channel_manager.channels.items(): + RNS.log( + f"Channel '{name}': " + f"{RNS.prettyhexrep(node.destination.hash)}", + RNS.LOG_NOTICE, + ) + + # ================================================================== + # Telephony – lobby channel + # ================================================================== def _incoming_link_established(self, link): link.is_incoming = True @@ -312,11 +577,12 @@ class LXSTRelay(SignallingReceiver): self.signal(Signalling.STATUS_AVAILABLE, link) - RNS.log("Incoming link, awaiting identification", RNS.LOG_DEBUG) + RNS.log("Incoming link (lobby), awaiting identification", RNS.LOG_DEBUG) def _caller_identified(self, link, identity): RNS.log( - f"Caller identified: {RNS.prettyhexrep(identity.hash)}", + f"Caller identified in lobby: " + f"{RNS.prettyhexrep(identity.hash)}", RNS.LOG_NOTICE, ) @@ -330,41 +596,142 @@ class LXSTRelay(SignallingReceiver): self.signal(Signalling.STATUS_ESTABLISHED, link) RNS.log( - f"Call established with " + f"Call established in lobby with " f"{RNS.prettyhexrep(identity.hash)}, " - f"{self.channel.get_participant_count()} participant(s) in " - f"'{self.channel_name}'", + f"{self.channel.get_participant_count()} participant(s)", RNS.LOG_NOTICE, ) def _link_closed(self, link): - RNS.log(f"Link closed: {link}", RNS.LOG_DEBUG) + RNS.log(f"Link closed (lobby): {link}", RNS.LOG_DEBUG) self.channel.remove_participant(link.link_id) - # ------------------------------------------------------------------ - # SignallingReceiver - # ------------------------------------------------------------------ + # ================================================================== + # LXMF bot + # ================================================================== + + def _lxmf_delivery_callback(self, message): + """Called on every incoming LXMF message addressed to the bot.""" + content = message.content_as_string().strip() + sender_hash = message.source_hash + + RNS.log( + f"Bot received from " + f"{RNS.hexrep(sender_hash, delimit=False)}: {content}", + RNS.LOG_NOTICE, + ) + + if not content: + return + + if content.startswith("/create_channel"): + self._cmd_create_channel(message, content, sender_hash) + elif content.startswith("/channels"): + self._cmd_list_channels(message) + else: + self._bot_reply( + message, + "Unknown command.\n" + "Available: /create_channel , /channels", + ) + + # -- Commands ------------------------------------------------------ + + def _cmd_create_channel(self, message, content, sender_hash): + parts = content.split(maxsplit=1) + if len(parts) < 2 or not parts[1].strip(): + self._bot_reply( + message, + "Usage: /create_channel ", + ) + return + + channel_name = parts[1].strip() + + node, error = self.channel_manager.create_channel( + channel_name, sender_hash + ) + if error: + self._bot_reply(message, error) + return + + hash_hex = RNS.hexrep(node.destination.hash, delimit=False) + self._bot_reply( + message, + f"Channel '{channel_name}' created!\n\n" + f"Call identity hash:\n{hash_hex}\n\n" + f"Share this hash for others to join.\n" + f"You are the admin of this channel.", + ) + + def _cmd_list_channels(self, message): + if not self.channel_manager.channels: + self._bot_reply(message, "No channels created yet.") + return + + lines = ["Active channels:"] + for name, node in self.channel_manager.channels.items(): + hash_hex = RNS.hexrep(node.destination.hash, delimit=False) + lines.append(f" {name} → {hash_hex}") + self._bot_reply(message, "\n".join(lines)) + + # -- Send reply ---------------------------------------------------- + + def _bot_reply(self, incoming_msg, text): + """Send a reply to the author of *incoming_msg*.""" + try: + sender_identity = RNS.Identity.recall(incoming_msg.source_hash) + except Exception: + sender_identity = None + + if sender_identity is None: + RNS.log( + "Cannot reply – sender identity not cached", + RNS.LOG_ERROR, + ) + return + + dest = RNS.Destination( + sender_identity, + RNS.Destination.OUT, + RNS.Destination.SINGLE, + "lxmf", + "delivery", + ) + + reply = LXMF.LXMessage( + dest, + self.bot_destination, + text, + desired_method=LXMF.LXMessage.DIRECT, + ) + self.lxmf_router.handle_outbound(reply) + + # ================================================================== + # SignallingReceiver (lobby) + # ================================================================== def signalling_received(self, signals, source): for signal in signals: if signal >= Signalling.PREFERRED_PROFILE: profile = signal - Signalling.PREFERRED_PROFILE RNS.log( - f"Received preferred profile: 0x{profile:02x}", + f"Lobby got preferred profile: 0x{profile:02x}", RNS.LOG_DEBUG, ) else: RNS.log( - f"Received signal: 0x{signal:02x}", + f"Lobby got signal: 0x{signal:02x}", RNS.LOG_DEBUG, ) - # ------------------------------------------------------------------ + # ================================================================== # Public API - # ------------------------------------------------------------------ + # ================================================================== def announce(self): self.destination.announce() + self.channel_manager.announce_all() RNS.log("Relay announced on network", RNS.LOG_NOTICE) def start(self): @@ -374,7 +741,6 @@ class LXSTRelay(SignallingReceiver): signal_module.signal(signal_module.SIGTERM, self._sigterm_handler) self.should_run = True - self.announce() RNS.log( f"LXST Relay ready on channel '{self.channel_name}'", @@ -384,18 +750,28 @@ class LXSTRelay(SignallingReceiver): f"Identity: {RNS.prettyhexrep(self.identity.hash)}", RNS.LOG_NOTICE, ) - RNS.log("Waiting for incoming calls...", RNS.LOG_NOTICE) + RNS.log( + f"Bot at: {RNS.prettyhexrep(self.bot_destination.hash)}", + RNS.LOG_NOTICE, + ) + RNS.log( + "Message the bot (via NomadNet / LXMF) with " + "/create_channel to create a new channel", + RNS.LOG_NOTICE, + ) + RNS.log("Waiting for calls and messages...", RNS.LOG_NOTICE) while self.should_run: time.sleep(1) def stop(self): self.should_run = False - RNS.log("Relay stopping...", RNS.LOG_NOTICE) + self.channel_manager.cleanup_all() + RNS.log("Relay stopped.", RNS.LOG_NOTICE) - # ------------------------------------------------------------------ + # ================================================================== # Signal handlers - # ------------------------------------------------------------------ + # ================================================================== def _sigint_handler(self, sig, frame): self.stop() @@ -404,11 +780,15 @@ class LXSTRelay(SignallingReceiver): self.stop() +# ====================================================================== +# Entry point +# ====================================================================== + def main(): parser = argparse.ArgumentParser( description=( - "LXST Relay - Group voice channel bridge for " - "Reticulum Telephony" + "LXST Relay – Group voice channel bridge for Reticulum " + "Telephony with NomadNet LXMF bot" ) ) @@ -429,13 +809,13 @@ def main(): "--channel", "-n", default="default", - help="voice channel name (default: 'default')", + help="lobby channel name (default: 'default')", type=str, ) parser.add_argument( "--version", action="version", - version="lxst-relay 0.1.0", + version="lxst-relay 0.2.0", ) parser.add_argument( "-v", "--verbose", action="count", default=0