""" `chat`: a group's conversation — the last messages, or one more posted. `chat` alone opens the interactive chat (`tui/chat.py`): the groups, one group's conversation IRC-style, back to the groups. `chat GROUP` prints the last ten, oldest first; `-N` (as `head -N`) or `-n N` asks for another count. `chat GROUP -m TEXT` posts TEXT (`-m -` reads it from stdin), sealed and signed as the application seals and signs it. Each message is shown with its time and the name its sender sealed in it; one that cannot be read is shown as such, never left out. `chat GROUP --follow` prints the last messages, then each new one as the node pushes it, until Ctrl+C. A connection that drops — or that stops answering the ping sent every half minute, as a sleeping laptop's does — is opened again, and what was said meanwhile is read from the history: each message is printed once, whichever way it came (the node stores a device and nonce pair once). With `--json`, one object per line. """ import asyncio import contextlib import json import re import sys from ..client.chat import chat_keys, history, open_row, post from ..client.groups import my_groups, open_group from ..client.paths import group_arg, handle, resolve_group from ..client.roster import RosterCheck from ..errors import NODE, USAGE, CliError from ..output import date, err, safe, sentence from .context import Context HELP = ("chat: alone, interactively; with a group, its last 10 messages (-N for N), " "or post with -m") _COUNT = re.compile(r"^-(\d+)$") # --follow: how often the connection is asked whether it is alive, how long it # has to answer, and the waits between attempts to open it again. PING_EVERY, PING_TIMEOUT = 30, 10 RECONNECT_WAITS = (2, 5, 10, 30) # The one verdict worth a line (roster.py): the application's words. KEY_CHANGED = ("! This account is using a key you have not seen before: a new device " "added elsewhere, or the node naming the wrong person.") def add_parser(sub, parent) -> None: p = sub.add_parser("chat", parents=[parent], help=HELP, description=sentence(HELP) + ". Messages are end-to-end encrypted and signed; the node " "stores what it cannot read.") p.add_argument("group", metavar="GROUP", nargs="?", help="the group, as `groups` lists it; without it, the interactive " "chat: choose a group, talk, leave it, choose another") p.add_argument("-n", "--last", type=int, default=10, metavar="N", help="how many of the last messages to show (also -N, as -100)") p.add_argument("-m", "--message", metavar="TEXT", help="post TEXT instead of reading; - reads it from stdin") p.add_argument("-f", "--follow", action="store_true", help="after the last messages, print new ones as they arrive, until " "Ctrl+C; with --json, one object per line") def rewrite_count(argv: list[str], actions) -> list[str]: """`chat -100 GROUP` as `chat --last 100 GROUP`: argparse reads -100 as a negative number, not as an option. Only when the action is `chat`.""" action = next((i for i, a in enumerate(argv) if a in actions), None) if action is None or argv[action] != "chat": return argv out = argv[:action + 1] for a in argv[action + 1:]: m = _COUNT.match(a) out += ["--last", m.group(1)] if m else [a] return out def _text(message: str) -> str: text = sys.stdin.read() if message == "-" else message text = text.rstrip("\n") if not text.strip(): raise CliError(USAGE, "Nothing to post.") return text async def run(ctx: Context) -> int: a = ctx.args if a.last < 1: raise CliError(USAGE, "Ask for one message at least.") if a.follow and a.message is not None: raise CliError(USAGE, "Post with -m, or follow with -f; not both at once.") if a.group is None: if a.message is not None or a.follow: raise CliError(USAGE, "Name the group to post to.") if not (sys.stdin.isatty() and sys.stdout.isatty()): raise CliError(USAGE, "The interactive chat needs a terminal; name a group " "to read it, or to post with -m.") from ..tui.app import run_interactive from ..tui.chat import ChatGroupsScreen # As with no argument at all: the sign-in screen checks the hub. return await run_interactive(ctx, ChatGroupsScreen) text = _text(a.message) if a.message is not None else None await ctx.hub.check_version() account = ctx.account() session = None try: group = resolve_group(await my_groups(account), group_arg(a.group)) session = await open_group(account, group) keys = await chat_keys(session) if text is not None: await post(session, keys, text, account.username) if a.json: print(json.dumps({"posted": True, "group": handle(group)})) elif not a.quiet: err(f"Posted to {safe(handle(group))}.") return 0 check = RosterCheck(account, session) messages = [await open_row(r, keys, group["id"], check) for r in await history(session, a.last)] if a.follow: follower = Follower(ctx, account, group, session, keys, check) for m in messages: follower.emit(m) if not messages and not a.quiet: err(f"No message in {safe(handle(group))} yet; waiting for one " "(Ctrl+C to stop).") session = None # the follower owns it now await follower.run() return 0 finally: if session is not None: await session.close() await account.close() if a.json: json.dump([as_json(m) for m in messages], sys.stdout, indent=1, ensure_ascii=False) print() return 0 if not messages and not a.quiet: err(f"No message in {safe(handle(group))} yet.") for m in messages: print_message(m) return 0 def print_message(m, *, flush: bool = False) -> None: head = f"{date(int(m.timestamp))} {safe(m.sender)}: " if m.unreadable: print(f"{head}[this message cannot be read: {m.unreadable}]", flush=flush) return lines = safe_lines(m.text) print(head + lines[0]) for line in lines[1:]: print(" " * len(head) + line) if m.trust == "changed": print(" " * len(head) + KEY_CHANGED) if flush: sys.stdout.flush() def as_json(m) -> dict: return {"id": m.id, "timestamp": m.timestamp, "sender": m.sender, "text": m.text, "trust": m.trust, **({"unreadable": m.unreadable} if m.unreadable else {})} class Follower: """`--follow`: one connection at a time, opened again when it dies, and every message printed once.""" def __init__(self, ctx: Context, account, group: dict, session, keys, check): self.ctx, self.account, self.group = ctx, account, group self.session, self.keys, self.check = session, keys, check self.seen: set[str] = set() self.last_ts = 0.0 def emit(self, m) -> None: if m.key: if m.key in self.seen: return self.seen.add(m.key) self.last_ts = max(self.last_ts, m.timestamp) if self.ctx.args.json: print(json.dumps(as_json(m), ensure_ascii=False), flush=True) else: print_message(m, flush=True) async def run(self) -> None: try: while True: await self._listen() await self.session.close() self.session = None if not self.ctx.args.quiet: err("meshbay-cli: the connection to the node was lost; reconnecting...") await self._reopen() finally: if self.session is not None: await self.session.close() async def _listen(self) -> None: """Print what the node pushes until the connection dies.""" s = self.session queue: asyncio.Queue = asyncio.Queue() dead = asyncio.Event() def on_message(msg): queue.put_nowait(msg) def on_epoch(msg): # Somebody was removed: the next message is sealed under a new key. s.ack["chat_epoch"] = int(msg.get("epoch") or 0) queue.put_nowait(None) def on_state(): if s.pc is not None and s.pc.connectionState in ("failed", "closed"): dead.set() s.ch.on_push("chat_msg", on_message) s.ch.on_push("chat_epoch_ack", on_epoch) if s.pc is not None: s.pc.on("connectionstatechange", on_state) pinger = asyncio.ensure_future(self._ping(dead)) closed = asyncio.ensure_future(s.ch.closed.wait()) gone = asyncio.ensure_future(dead.wait()) try: while True: getter = asyncio.ensure_future(queue.get()) done, _ = await asyncio.wait({getter, closed, gone}, return_when=asyncio.FIRST_COMPLETED) if getter not in done: getter.cancel() return msg = getter.result() if msg is None or int(msg.get("epoch") or 0) not in self.keys.by_epoch: with contextlib.suppress(CliError): self.keys = await chat_keys(s) if msg is None: continue msg = {**msg, "timestamp": msg.get("timestamp") or 0} self.emit(await open_row(msg, self.keys, self.group["id"], self.check)) finally: for task in (pinger, closed, gone): task.cancel() s.ch.off_push("chat_msg", on_message) s.ch.off_push("chat_epoch_ack", on_epoch) async def _ping(self, dead: asyncio.Event) -> None: """A connection that stopped answering looks open to both ends.""" while True: await asyncio.sleep(PING_EVERY) try: await self.session.ch.request({"type": "ping", "token": "follow"}, "pong", timeout=PING_TIMEOUT) except CliError: dead.set() return async def _reopen(self) -> None: """A new connection, then what was said while there was none.""" attempt = 0 while True: try: self.session = await open_group(self.account, self.group) self.keys = await chat_keys(self.session) self.check = RosterCheck(self.account, self.session) rows = await history(self.session, 200) break except CliError as e: if self.session is not None: await self.session.close() self.session = None if e.code != NODE: raise await asyncio.sleep(RECONNECT_WAITS[min(attempt, len(RECONNECT_WAITS) - 1)]) attempt += 1 for row in rows: m = await open_row(row, self.keys, self.group["id"], self.check) if m.timestamp >= self.last_ts: self.emit(m) def safe_lines(text: str) -> list[str]: """A message's lines, each escaped: a line break stays a line break, every other control character is shown, never sent to the terminal.""" return [safe(line) for line in text.split("\n")] or [""]