diff options
Diffstat (limited to 'packages/meshbay-cli')
| -rw-r--r-- | packages/meshbay-cli/src/meshbay_cli/client/chat.py | 4 | ||||
| -rw-r--r-- | packages/meshbay-cli/src/meshbay_cli/commands/chat.py | 185 | ||||
| -rw-r--r-- | packages/meshbay-cli/src/meshbay_cli/manpage.py | 16 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/fake_node.py | 6 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_chat_follow.py | 94 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_e2e_local.py | 28 |
6 files changed, 313 insertions, 20 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/client/chat.py b/packages/meshbay-cli/src/meshbay_cli/client/chat.py index 60629e2..22a99ef 100644 --- a/packages/meshbay-cli/src/meshbay_cli/client/chat.py +++ b/packages/meshbay-cli/src/meshbay_cli/client/chat.py @@ -48,6 +48,9 @@ class Message: text: str unreadable: str = "" # why it could not be read: signature, epoch, ... trust: str = "" # roster.RosterCheck's verdict on the signing device + # Which message this is, wherever it was read from — history or a push, + # which carries no id: the node stores a (device, nonce) pair only once. + key: str = "" class ChatKeys: @@ -100,6 +103,7 @@ async def open_row(row: dict, keys: ChatKeys, group_id: str, if not (device and nonce and ct and sig): msg.unreadable = "envelope" return msg + msg.key = f"{b64(bytes(device))}:{b64(bytes(nonce))}" if not verify(group_id, epoch, bytes(device), bytes(nonce), bytes(ct), bytes(sig)): msg.unreadable = "signature" return msg diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/chat.py b/packages/meshbay-cli/src/meshbay_cli/commands/chat.py index 6fa084d..d650af3 100644 --- a/packages/meshbay-cli/src/meshbay_cli/commands/chat.py +++ b/packages/meshbay-cli/src/meshbay_cli/commands/chat.py @@ -8,8 +8,17 @@ 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 @@ -18,7 +27,7 @@ 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 USAGE, CliError +from ..errors import NODE, USAGE, CliError from ..output import date, err, safe, sentence from .context import Context @@ -26,6 +35,10 @@ HELP = ("chat: alone, interactively; with a group, its last 10 messages (-N for "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.") @@ -42,6 +55,9 @@ def add_parser(sub, parent) -> None: 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]: @@ -69,8 +85,10 @@ 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: + 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 " @@ -97,35 +115,168 @@ async def run(ctx: Context) -> int: 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([{"id": m.id, "timestamp": m.timestamp, "sender": m.sender, - "text": m.text, "trust": m.trust, - **({"unreadable": m.unreadable} if m.unreadable else {})} - for m in messages], - sys.stdout, indent=1, ensure_ascii=False) + 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: - head = f"{date(int(m.timestamp))} {safe(m.sender)}: " - if m.unreadable: - print(f"{head}[this message cannot be read: {m.unreadable}]") - continue - 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) + 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.""" diff --git a/packages/meshbay-cli/src/meshbay_cli/manpage.py b/packages/meshbay-cli/src/meshbay_cli/manpage.py index 022cd99..43a0fee 100644 --- a/packages/meshbay-cli/src/meshbay_cli/manpage.py +++ b/packages/meshbay-cli/src/meshbay_cli/manpage.py @@ -37,8 +37,8 @@ EXIT_CODES = { "accepted by the node, a folder read-only or unavailable, leaving a " "group you own, revoking this machine, deleting a file you may not " "delete.", - errors.INTERRUPTED: "Interrupted (Ctrl+C). A download resumes where it stopped " - "on the next run.", + errors.INTERRUPTED: "Interrupted (Ctrl+C), which is also how chat \\-\\-follow " + "ends. A download resumes where it stopped on the next run.", } DESCRIPTION = r""" @@ -232,6 +232,14 @@ before anything is asked or sent. shows the last messages of a group, oldest first, each with its time and the name its sender sealed in it, and posts one with .BR \-m . +With +.BR \-\-follow , +it then prints each new message as it arrives, until Ctrl+C (status 130); a +connection that drops or stops answering is opened again, and what was said +meanwhile is printed, each message once. +With +.BR \-\-json , +one object per line. Messages are encrypted and signed end to end, as the application's are; one this machine cannot verify or open is shown as unreadable, never left out. The device that signed a message is checked against the group's roster and the @@ -263,7 +271,8 @@ On Windows, .IR %LOCALAPPDATA%\emeshbay\-cli\e . Holds .B state.json -(the hub, the account name, pinned node keys, the browser access setting), +(the hub, the account name, pinned node keys, the device keys accepted for +the members whose messages were read, the browser access setting), .B secrets.enc (the session, the keys, invitation codes waiting for their node; encrypted), .B store.key @@ -321,6 +330,7 @@ meshbay\-cli search \-t audio holiday meshbay\-cli search beach 2024 | head \-1 | xargs \-d '\\n' meshbay\-cli get meshbay\-cli rm 'Family/documents/draft*.odt' meshbay\-cli chat \-100 Family +meshbay\-cli \-\-json chat Family \-\-follow | jq \-r .text echo 'On my way' | meshbay\-cli chat Family \-m \- meshbay\-cli devices approve Family@bob ZX4P\-7QRT meshbay\-cli leave \-\-yes Club@carol diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py index 8f4356b..b4e64ff 100644 --- a/packages/meshbay-cli/tests/fake_node.py +++ b/packages/meshbay-cli/tests/fake_node.py @@ -76,6 +76,7 @@ class FakeNode: "pk_x25519": self.identity.pk_x, "added_by_pk": "", "add_sig": "", "add_nonce": "", "add_ts": 0}] self.members: dict[str, Identity] = {} # sender → the device it signs with + self.unresponsive = False # pings go unanswered, as a dead link's self.admin_ops: dict[str, dict] = {} def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict: @@ -273,6 +274,11 @@ class FakeChannel: return {"type": "chat_keys_resp", "group_id": GROUP_ID, **seal(n.gek, PURPOSE_CHAT_KEYS, "chat_keys_resp", GROUP_ID, payload)} + def _ping(self, msg): + if self.node.unresponsive: + return {"type": "error", "detail": "no answer"} + return {"type": "pong", "token": msg.get("token")} + def _group_roster_req(self, msg): n = self.node return {"type": "group_roster_resp", "group_id": GROUP_ID, diff --git a/packages/meshbay-cli/tests/test_chat_follow.py b/packages/meshbay-cli/tests/test_chat_follow.py new file mode 100644 index 0000000..cc6db99 --- /dev/null +++ b/packages/meshbay-cli/tests/test_chat_follow.py @@ -0,0 +1,94 @@ +""" +`chat --follow`: new messages as the node pushes them, a dead connection opened +again, and what was said meanwhile read from the history — each message once. +""" + +import asyncio +import json + +import pytest +from fake_hub import PASSPHRASE, USERNAME, FakeHub +from fake_node import FakeNode +from meshbay_cli import __main__ as cli +from meshbay_cli.commands import chat, context + + +@pytest.fixture +def follow(monkeypatch, capsys): + hub, node = FakeHub(), FakeNode() + monkeypatch.setattr(context, "Hub", lambda url: hub) + sessions = [] + + async def open_group(account, group): + sessions.append(node.session()) + return sessions[-1] + monkeypatch.setattr(chat, "open_group", open_group) + monkeypatch.setattr(chat, "PING_EVERY", 0.05) + monkeypatch.setattr(chat, "RECONNECT_WAITS", (0.01,)) + assert cli.main(["--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, + "login"]) == 0 + capsys.readouterr() + + def start(*argv): + args = cli.build_parser().parse_args(cli.chat.rewrite_count(list(argv), cli.ACTIONS)) + return asyncio.ensure_future(chat.run(context.Context(args))) + return node, sessions, start + + +def _push(session, node): + session.ch._emit({**node.chat[-1], "type": "chat_msg"}) + + +async def _until(cond, timeout=5.0): + deadline = asyncio.get_running_loop().time() + timeout + while not cond(): + assert asyncio.get_running_loop().time() < deadline, "timed out" + await asyncio.sleep(0.02) + + +async def test_follow_prints_pushes_reconnects_and_never_repeats(follow, capsys): + node, sessions, start = follow + node.say("before") + task = start("--json", "chat", "-1", "Family", "--follow") + printed = [] + + def lines(): + printed.extend(json.loads(x)["text"] for x in capsys.readouterr().out.splitlines()) + return printed + try: + await _until(lambda: lines() == ["before"]) + + node.say("live one") + _push(sessions[-1], node) + await _until(lambda: lines()[-1:] == ["live one"]) + + # The connection drops; meanwhile two messages are stored and only the + # last of them is pushed, once the new connection is up. + sessions[-1].ch.closed.set() + node.say("while away") + await _until(lambda: len(sessions) == 2) + await _until(lambda: "while away" in lines()) + node.say("after") + _push(sessions[-1], node) + _push(sessions[-1], node) # the node said it twice + await _until(lambda: lines()[-1:] == ["after"]) + + # A connection that stops answering its ping is replaced too. + node.unresponsive = True + await _until(lambda: len(sessions) == 3) + node.unresponsive = False + node.say("on the third connection") + _push(sessions[-1], node) + await _until(lambda: lines()[-1:] == ["on the third connection"]) + await asyncio.sleep(0.2) + assert lines() == ["before", "live one", "while away", "after", + "on the third connection"] + finally: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +def test_follow_and_post_are_not_asked_at_once(follow, capsys): + assert cli.main(["chat", "Family", "-f", "-m", "x"]) == 1 + assert cli.main(["chat", "-f"]) == 1 diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py index cf6a3b9..359ab06 100644 --- a/packages/meshbay-cli/tests/test_e2e_local.py +++ b/packages/meshbay-cli/tests/test_e2e_local.py @@ -1068,3 +1068,31 @@ async def test_the_interactive_chat_against_the_real_node(world, monkeypatch, tm r = await asyncio.to_thread(cli, world, "chat", "-1", group, home="cli-chat") assert r.stdout.strip().endswith(f"{USERNAME}: typed in the room"), r.stdout + + +def test_chat_follow_against_the_real_node(world): + """`chat --follow` in its own process: messages posted from another run + arrive as the node pushes them, across more than one liveness ping, and + Ctrl+C ends it. Runs on the machine the chat tests above linked.""" + import signal + group = _handle() + env = {**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"), + "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring"} + exe = os.environ.get("MESHBAY_CLI_E2E_BIN") or str(BIN / "meshbay-cli") + follower = subprocess.Popen([exe, "--json", "chat", "-1", group, "--follow"], env=env, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + try: + first = json.loads(follower.stdout.readline()) # the last one, at once + assert first["text"] + for text, wait in (("followed one", 0), ("followed two, after a ping", 36)): + time.sleep(wait) + r = cli(world, "chat", group, "-m", text, home="cli-chat") + assert r.returncode == 0, r.stderr + got = json.loads(follower.stdout.readline()) + assert got["text"] == text and got["sender"] == USERNAME, got + assert got["trust"] in ("pinned", "first", "linked") + finally: + follower.send_signal(signal.SIGINT) + _, err = follower.communicate(timeout=20) + assert follower.returncode == 130 + assert "reconnecting" not in err, err |