lxst-client/client.py

#!/usr/bin/env python3

import json
import os
import sys
import time
import threading
import signal
import argparse

import RNS
import LXMF
from LXST.Primitives.Telephony import Telephone


class Terminal:
    if not RNS.vendor.platformutils.is_windows():
        UNDERLINE = "\033[4m"
        BOLD = "\033[1m"
        END = "\033[0m"
    else:
        UNDERLINE = ""
        BOLD = ""
        END = ""


class ChannelClient:
    STATE_IDLE = 0
    STATE_DISCOVERING = 1
    STATE_READY = 2
    STATE_CALLING = 3
    STATE_IN_CALL = 4

    def __init__(self, configdir=None, rnsconfigdir=None, whitelist_path=None, verbosity=0):
        self.state = self.STATE_IDLE
        self.channels = []
        self.telephone = None
        self.should_run = False
        self._responses = []
        self._response_event = threading.Event()

        if configdir is None:
            if os.path.isdir("/etc/lxst-client"):
                self.configdir = "/etc/lxst-client"
            else:
                userdir = RNS.Reticulum.userdir
                self.configdir = os.path.join(userdir, ".lxst-client")
        else:
            self.configdir = configdir
        os.makedirs(self.configdir, exist_ok=True)

        self.whitelist_path = whitelist_path
        if self.whitelist_path is None:
            candidates = [
                os.path.join(os.getcwd(), "bridge_whitelist.json"),
                os.path.join(self.configdir, "bridge_whitelist.json"),
            ]
            for p in candidates:
                if os.path.isfile(p):
                    self.whitelist_path = p
                    break
            if self.whitelist_path is None:
                self.whitelist_path = candidates[0]

        identity_path = os.path.join(self.configdir, "identity")
        if os.path.isfile(identity_path):
            self.identity = RNS.Identity.from_file(identity_path)
            if self.identity is None:
                self.identity = RNS.Identity()
                self.identity.to_file(identity_path)
        else:
            RNS.log("No identity found, generating new", RNS.LOG_NOTICE)
            self.identity = RNS.Identity()
            self.identity.to_file(identity_path)
            RNS.log(
                f"Generated identity: {RNS.prettyhexrep(self.identity.hash)}",
                RNS.LOG_NOTICE,
            )

        self.reticulum = RNS.Reticulum(
            configdir=rnsconfigdir, loglevel=3 + verbosity
        )

        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.client_destination = self.lxmf_router.register_delivery_identity(
            self.identity,
            display_name="LXST Channel Client",
        )
        self.lxmf_router.register_delivery_callback(self._delivery_callback)

    def _delivery_callback(self, message):
        content = message.content_as_string().strip()
        sender_hex = RNS.hexrep(message.source_hash, delimit=False)
        try:
            data = json.loads(content)
            if isinstance(data, list):
                for h in data:
                    if isinstance(h, str) and len(h) in (32, 64):
                        self._responses.append({
                            "hash": h,
                            "relay": sender_hex,
                        })
        except (json.JSONDecodeError, ValueError):
            pass
        self._response_event.set()

    def _discover_from_relay(self, address):
        try:
            addr_bytes = bytes.fromhex(address)
            identity = RNS.Identity.recall(addr_bytes)
            if identity is None:
                RNS.Transport.request_path(addr_bytes)
                for _ in range(50):
                    identity = RNS.Identity.recall(addr_bytes)
                    if identity is not None:
                        break
                    time.sleep(0.1)

            if identity is None:
                print(f"  Could not discover relay {address[:16]}...")
                return

            dest = RNS.Destination(
                identity,
                RNS.Destination.OUT,
                RNS.Destination.SINGLE,
                "lxmf",
                "delivery",
            )
            msg = LXMF.LXMessage(
                dest,
                self.client_destination,
                "/channels_json",
                desired_method=LXMF.LXMessage.DIRECT,
            )
            self.lxmf_router.handle_outbound(msg)

        except Exception as e:
            RNS.log(f"Error sending to {address[:16]}...: {e}", RNS.LOG_ERROR)

    def discover(self, timeout=30):
        if not os.path.isfile(self.whitelist_path):
            print(f"bridge_whitelist.json not found at {self.whitelist_path}")
            return []

        with open(self.whitelist_path) as f:
            whitelist = json.load(f)

        if not isinstance(whitelist, list):
            print("bridge_whitelist.json must contain a list of LXMF addresses")
            return []

        self.state = self.STATE_DISCOVERING
        self._responses = []
        self._response_event.clear()

        print(f"\nQuerying {len(whitelist)} relay(s) for channels...", flush=True)
        for addr in whitelist:
            self._discover_from_relay(addr)

        deadline = time.time() + timeout
        while time.time() < deadline:
            remaining = deadline - time.time()
            self._response_event.wait(timeout=min(remaining, 1))
            if self._responses:
                break

        seen = set()
        deduped = []
        for r in self._responses:
            if r["hash"] not in seen:
                seen.add(r["hash"])
                deduped.append(r)

        self.channels = deduped
        print(
            f"  Got {len(self.channels)} unique channel(s) "
            f"from {len(self._responses)} response(s)"
        )

        self.state = self.STATE_READY
        return self.channels

    def _init_telephone(self):
        if self.telephone is None:
            self.telephone = Telephone(self.identity)
            self.telephone.set_established_callback(self._on_call_established)
            self.telephone.set_ended_callback(self._on_call_ended)
            self.telephone.announce()

    def _on_call_established(self, remote_identity):
        self.state = self.STATE_IN_CALL
        print(
            f"\nCall established with "
            f"{RNS.prettyhexrep(remote_identity.hash)}"
        )

    def _on_call_ended(self, remote_identity):
        was_in_call = self.state == self.STATE_IN_CALL
        was_calling = self.state == self.STATE_CALLING
        self.state = self.STATE_READY
        if was_in_call:
            print(f"\nCall ended")
        elif was_calling:
            print(f"\nCall could not be connected")

    def call_channel(self, index):
        if index < 0 or index >= len(self.channels):
            print("Invalid channel index")
            return

        channel = self.channels[index]
        dest_hash_hex = channel["hash"]
        dest_hash = bytes.fromhex(dest_hash_hex)

        print(
            f"\nConnecting to channel {index + 1}: "
            f"{dest_hash_hex[:16]}..."
        )

        if not RNS.Transport.has_path(dest_hash):
            RNS.Transport.request_path(dest_hash)
            print("  Requesting path...", end="", flush=True)
            for _ in range(100):
                if RNS.Transport.has_path(dest_hash):
                    break
                time.sleep(0.1)
            print()

        if not RNS.Transport.has_path(dest_hash):
            print("  Could not find path to channel")
            return

        hops = RNS.Transport.hops_to(dest_hash)
        print(f"  Path found ({hops} hop{'s' if hops != 1 else ''})")

        identity = RNS.Identity.recall(dest_hash)
        if identity is None:
            print("  Could not discover channel identity")
            return

        self._init_telephone()
        self.state = self.STATE_CALLING
        self.telephone.call(identity)

    def show_channels(self):
        if not self.channels:
            print("No channels discovered. Run 'discover' first.")
            return

        print(
            f"\n{Terminal.BOLD}"
            f"Available channels ({len(self.channels)}):"
            f"{Terminal.END}"
        )
        print(f"  {'#':<4} {'Destination Hash':<48}  {'Relay':<16}")
        print(f"  {'-'*4} {'-'*48}  {'-'*16}")
        for i, ch in enumerate(self.channels):
            relay_short = (
                ch["relay"][:14] + ".."
                if len(ch["relay"]) > 16
                else ch["relay"]
            )
            print(f"  {i + 1:<4} {ch['hash']:<48}  {relay_short}")

    def show_help(self):
        print("")
        print(f"{Terminal.BOLD}Commands:{Terminal.END}")
        print(f"  <number>  Call the channel at that number")
        print(f"  d         Discover channels (query relays)")
        print(f"  l         List channels")
        print(f"  h / ?     Show this help")
        print(f"  q         Quit")

    def run(self):
        signal.signal(signal.SIGINT, self._sigint_handler)
        signal.signal(signal.SIGTERM, self._sigterm_handler)
        self.should_run = True

        print(
            f"\n{Terminal.BOLD}"
            f"LXST Channel Client"
            f"{Terminal.END}"
        )
        print(f"  Identity: {RNS.prettyhexrep(self.identity.hash)}")
        print(f"  Whitelist: {self.whitelist_path}")
        print()

        self.discover(timeout=15)
        self.show_channels()
        self.show_help()

        while self.should_run:
            if self.state == self.STATE_IN_CALL:
                try:
                    input("  In call, press Enter to hang up > ")
                    self.telephone.hangup()
                    self.state = self.STATE_READY
                except (EOFError, KeyboardInterrupt):
                    break
                continue

            if self.state == self.STATE_CALLING:
                time.sleep(0.5)
                continue

            if self.state == self.STATE_DISCOVERING:
                time.sleep(0.5)
                continue

            try:
                inp = input("> ").strip().lower()
            except (EOFError, KeyboardInterrupt):
                print()
                break

            if not inp:
                continue

            if inp == "q":
                break
            elif inp in ("h", "?"):
                self.show_help()
            elif inp == "d":
                self.discover()
                self.show_channels()
            elif inp == "l":
                self.show_channels()
            else:
                try:
                    idx = int(inp) - 1
                    if 0 <= idx < len(self.channels):
                        self.call_channel(idx)
                    else:
                        print(
                            f"Invalid index. "
                            f"Enter a number between 1 and {len(self.channels)}"
                        )
                except ValueError:
                    print(f"Unknown command: {inp}. Type 'h' for help.")

    def cleanup(self):
        if self.telephone and self.state == self.STATE_IN_CALL:
            self.telephone.hangup()
        self.should_run = False

    def _sigint_handler(self, sig, frame):
        self.cleanup()
        sys.exit(0)

    def _sigterm_handler(self, sig, frame):
        self.cleanup()
        sys.exit(0)


def main():
    parser = argparse.ArgumentParser(description="LXST Channel Client")
    parser.add_argument(
        "--config", "-c", default=None, help="config directory",
    )
    parser.add_argument(
        "--rnsconfig", default=None,
        help="path to alternative Reticulum config directory",
    )
    parser.add_argument(
        "--whitelist", "-w", default=None,
        help="path to bridge_whitelist.json",
    )
    parser.add_argument(
        "--version",
        action="version",
        version="lxst-client 0.1.0",
    )
    parser.add_argument("-v", "--verbose", action="count", default=0)

    args = parser.parse_args()

    client = ChannelClient(
        configdir=args.config,
        rnsconfigdir=args.rnsconfig,
        whitelist_path=args.whitelist,
        verbosity=args.verbose,
    )

    try:
        client.run()
    except KeyboardInterrupt:
        client.cleanup()
        print()


if __name__ == "__main__":
    main()