user can send LXMF message to create a new LXST channel

2026-05-28 01:46:07 -05:00
parent f073ebe106
commit 5fc9cdf79d
1 changed files with 419 additions and 39 deletions
@@ -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 <name>``. 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 <name>``, 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)

+        # ---- 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
+        )
+
+        # ---- 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 listening on "
-            f"{RNS.prettyhexrep(self.destination.hash)}",
+            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,
                )

-    # ------------------------------------------------------------------
-    #  Link lifecycle
-    # ------------------------------------------------------------------
+    # ==================================================================
+    #  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 <name>, /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 <name>",
+            )
+            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 <name> 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