feat(cli): chat — read a group's last messages, post one

chat GROUP prints the last ten messages, oldest first (-N, as head -N, or
-n N for another count, paging back past the node's 200), and -m posts one
(- reads it from stdin). The application's construction: every epoch key
from chat_keys_req, a post sealed under the current epoch through the
device subkey and signed with this machine's key on the node, the sender's
name inside the envelope; nothing sent past the node's ciphertext bound.
A stored message is read signature first, then key, then content, and one
that fails is shown as unreadable, never dropped. Not done here: the Tier 2
roster check on the signing device.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

3 files changed, 215 insertions, 12 deletions

diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py
index 73d6f72..f5fe631 100644
--- a/packages/meshbay-cli/tests/fake_node.py
+++ b/packages/meshbay-cli/tests/fake_node.py
@@ -22,13 +22,20 @@ from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
 from fake_hub import USER_ID
 from meshbay_cli.client.keys import Identity
 from meshbay_cli.client.session import GroupSession, NodeRefusal
+from meshbay_common import chatbox
 from meshbay_common.adminop import OP_FILE_DELETE, admin_transcript
 from meshbay_common.device import (
     device_add_transcript,
     device_code_hash,
     device_revoke_transcript,
 )
-from meshbay_common.groupbox import PURPOSE_INDEX, PURPOSE_UPLOAD, seal, unseal
+from meshbay_common.groupbox import (
+    PURPOSE_CHAT_KEYS,
+    PURPOSE_INDEX,
+    PURPOSE_UPLOAD,
+    seal,
+    unseal,
+)
 from meshbay_common.protocol import UPLOAD_PROBE_INDEX, file_upload_ack_wire
 from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
 
@@ -59,7 +66,9 @@ class FakeNode:
                          "pinned_via": "code"}]
         self.requests: list[dict] = []
         self.role = ""                       # "operator": this key was paired here
-        self.ack: dict = {}                  # the sealed configuration, opened
+        self.ack: dict = {"chat_epoch": 1}   # the sealed configuration, opened
+        self.chat_epochs = {1: secrets.token_bytes(32)}
+        self.chat: list[dict] = []           # stored rows, oldest first
         self.admin_ops: dict[str, dict] = {}
 
     def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict:
@@ -77,6 +86,25 @@ class FakeNode:
         return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek,
                             dict(self.ack), self.identity, self.nonce, self.role)
 
+    def say(self, text: str, *, sender: str = "bob-test", epoch: int = 1,
+            forge: bool = False) -> None:
+        """Another member's message, sealed and signed as a client does; `forge`
+        signs it with a key other than the device it names."""
+        other = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate())
+        sealed = chatbox.seal(self.chat_epochs[epoch], GROUP_ID, epoch, other.pk_ed,
+                              base64.b64decode(other.pk_ed), other.sk_ed,
+                              {"text": text, "sender_name": sender})
+        if forge:
+            sealed["sig"] = Ed25519PrivateKey.generate().sign(b"something else")
+        self._store(base64.b64decode(other.pk_ed), epoch, sealed, "u-" + sender)
+
+    def _store(self, device: bytes, epoch: int, sealed: dict, sender_id: str) -> None:
+        self.chat.append({"id": len(self.chat) + 1, "sender_id": sender_id,
+                          "sender_name": "", "timestamp": 1_700_000_000 + len(self.chat),
+                          "thread_id": None, "format": 1, "epoch": epoch,
+                          "device": device, "nonce": sealed["nonce"], "sig": sealed["sig"],
+                          "payload": "", "ct": sealed["ct"]})
+
     def request_device(self, code: str) -> Identity:
         """Another machine files a request with `code`, as it would before its proof."""
         other = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate())
@@ -199,6 +227,35 @@ class FakeChannel:
                                     dir=folder)
 
 
+    # ── Chat ─────────────────────────────────────────────────────────────────
+
+    def _chat_keys_req(self, msg):
+        n = self.node
+        payload = {"epochs": [{"epoch": e, "key": k} for e, k in sorted(n.chat_epochs.items())],
+                   "current": max(n.chat_epochs)}
+        return {"type": "chat_keys_resp", "group_id": GROUP_ID,
+                **seal(n.gek, PURPOSE_CHAT_KEYS, "chat_keys_resp", GROUP_ID, payload)}
+
+    def _chat_hist(self, msg):
+        rows = self.node.chat
+        if msg.get("before"):
+            rows = [r for r in rows if r["id"] < msg["before"]]
+        page = rows[-min(int(msg["limit"]), 200):]
+        return {"type": "chat_hist_resp", "messages": page,
+                "has_more": bool(page) and page[0]["id"] > rows[0]["id"]}
+
+    def _chat_msg(self, msg):
+        n = self.node
+        if bytes(msg["device"]) != base64.b64decode(n.identity.pk_ed):
+            return {"type": "error", "detail": "That is not the device on this connection"}
+        if msg["epoch"] != max(n.chat_epochs) or msg["format"] != 1:
+            return {"type": "error", "detail": "stale epoch or plaintext"}
+        if not chatbox.verify(GROUP_ID, msg["epoch"], bytes(msg["device"]), msg["nonce"],
+                              msg["ct"], msg["sig"]):
+            return {"type": "error", "detail": "bad signature"}
+        n._store(bytes(msg["device"]), msg["epoch"], msg, USER_ID)
+        return {"type": "ack"}
+
     # ── Deleting: a challenge, signed with this machine's key ──────────────
 
     def _file_delete(self, msg):
diff --git a/packages/meshbay-cli/tests/test_cli_actions.py b/packages/meshbay-cli/tests/test_cli_actions.py
index e06fb9c..5443d9b 100644
--- a/packages/meshbay-cli/tests/test_cli_actions.py
+++ b/packages/meshbay-cli/tests/test_cli_actions.py
@@ -510,6 +510,51 @@ def test_search_finds_across_groups_once_per_file(hub, monkeypatch, capsys):
     assert main("search", "beach") == 0
 
 
+def test_chat_reads_the_last_messages_and_posts(node, monkeypatch, capsys):
+    from meshbay_cli.commands import chat
+
+    async def open_group(account, group):
+        return node.session()
+    monkeypatch.setattr(chat, "open_group", open_group)
+
+    for i in range(1, 231):                       # more than one page of the node's
+        node.say(f"message {i}")
+    node.say("forged", forge=True)
+    node.say("two lines\nand an \x1b[31m escape")
+    node.chat_epochs[2] = b"\x09" * 32
+    node.say("from an epoch this member was not given", epoch=2)
+    del node.chat_epochs[2]
+
+    assert main("chat", "Family") == 0
+    lines = capsys.readouterr().out.splitlines()
+    assert len(lines) == 11 and lines[0].endswith("bob-test: message 224")
+    assert lines[7].endswith("bob-test: [this message cannot be read: signature]")
+    assert lines[8].endswith("bob-test: two lines")
+    assert lines[9].strip() == "and an \\x1b[31m escape"            # shown, never run
+    assert "cannot be read: no key for its epoch" in lines[10]
+
+    assert main("chat", "-5", "Family") == 0
+    assert capsys.readouterr().out.splitlines()[0].endswith("message 229")
+    assert main("--json", "chat", "Family", "-n", "300") == 0     # paged back
+    got = json.loads(capsys.readouterr().out)
+    assert len(got) == 233 and got[0]["text"] == "message 1"
+    assert got[230]["unreadable"] == "signature"
+
+    assert main("chat", "Family", "-m", "hello from the terminal") == 0
+    assert "Posted to Family@alice-test" in capsys.readouterr().err
+    monkeypatch.setattr("sys.stdin", io.StringIO("piped\n"))
+    assert main("chat", "Family", "-m", "-") == 0
+    capsys.readouterr()
+    assert main("chat", "-2", "Family") == 0
+    out = capsys.readouterr().out.splitlines()
+    assert out[0].endswith(f"{USERNAME}: hello from the terminal")
+    assert out[1].endswith(f"{USERNAME}: piped")
+
+    assert main("chat", "Family", "-m", "   ") == 1
+    assert main("chat", "Family", "-m", "x" * 70_000) == 1          # over the node's bound
+    assert main("chat", "-0", "Family") == 1
+
+
 def test_every_action_is_run_here():
     """Runs last in this module: every action the parser knows was run above."""
     assert RUN == set(cli.ACTIONS), f"never run: {set(cli.ACTIONS) - RUN}"
diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py
index 8ce3ebc..19f5ca9 100644
--- a/packages/meshbay-cli/tests/test_e2e_local.py
+++ b/packages/meshbay-cli/tests/test_e2e_local.py
@@ -312,6 +312,27 @@ class App:
         assert r.get("detail") == "keypair_bundle_deleted", r
         await close()
 
+    async def revoke_the_others(self) -> None:
+        """Retire every device of the account but this one: the node admits five,
+        and the tests link more machines than that between them."""
+        from meshbay_cli.client.keys import b64
+        from meshbay_common import MNP_VERSION
+        from meshbay_common.device import device_revoke_transcript
+        ch, close, node_pk, nonce_s, account = await self.session()
+        listed = await ch.request({"type": "device_list", "v": MNP_VERSION},
+                                  "device_list_result")
+        for d in listed["devices"]:
+            if d["pk_ed25519"] == self.ident.pk_ed:
+                continue
+            ts = int(time.time())
+            sig = self.ident.sk_ed.sign(device_revoke_transcript(
+                node_pk_b64=node_pk, user_id=account.user_id,
+                pk_ed25519_b64=d["pk_ed25519"], nonce_node=nonce_s, ts=ts))
+            await ch.request({"type": "device_revoke", "v": MNP_VERSION,
+                              "pk_ed25519": d["pk_ed25519"], "ts": ts, "sig": b64(sig)},
+                             "device_add_ack")
+        await close()
+
     async def approve(self, code: str) -> None:
         """Approve a device by the code it shows, as the group's Settings tab does."""
         from meshbay_cli.client.keys import b64
@@ -668,7 +689,9 @@ _LINK_FORM = r"""(() => {
   return { form: Boolean(form), button: Boolean(form && form.querySelector('button')),
            code: code || '', minted: window.__minted,
            error: (document.querySelector('.error-msg') || {}).innerText || '',
-           opened: /No messages yet/.test(page), page: page.slice(-400) };
+           // The group open on its Chat tab, the composer ready.
+           opened: Boolean(document.querySelector('.chat-input:not([disabled])')),
+           page: page.slice(-400) };
 })()"""
 
 
@@ -730,12 +753,10 @@ class _Chrome:
         self.profile.cleanup()
 
 
-@pytest.mark.skipif(not shutil.which("google-chrome"),
-                    reason="google-chrome is not available")
-def test_the_application_links_itself_from_its_group_page(world):
+def _session_for_the_page(world: dict) -> dict:
+    """What the page keeps in `mb_auth` after signing in, for the account."""
     from meshbay_cli.client.keys import auth_key
-    app, hub_url = world["app"], world["hub_url"]
-    asyncio.run(app.withdraw_bundle())
+    hub_url = world["hub_url"]
     deadline = time.monotonic() + 90
     while (r := httpx.post(f"{hub_url}/v1/users/login", json={
             "username": USERNAME, "auth_key": auth_key(USERNAME, PASSPHRASE)})
@@ -745,12 +766,19 @@ def test_the_application_links_itself_from_its_group_page(world):
     tokens = r.json()
     me = httpx.get(f"{hub_url}/v1/users/me",
                    headers={"Authorization": f"Bearer {tokens['access_token']}"}).json()
-    auth = {"username": USERNAME, "userId": me["user_id"], "role": "user",
+    return {"username": USERNAME, "userId": me["user_id"], "role": "user",
             "token": tokens["access_token"], "refreshToken": tokens.get("refresh_token")}
 
-    chrome = _Chrome(_KEYS_STUB.replace("__AUTH__", json.dumps(auth)))
+
+def _linked_group_page(world: dict) -> "_Chrome":
+    """The group page in Chrome as the desktop application, linked as a new
+    device by its own "Get a linking code" and open. The caller closes it."""
+    app = world["app"]
+    asyncio.run(app.withdraw_bundle())
+    chrome = _Chrome(_KEYS_STUB.replace("__AUTH__", json.dumps(_session_for_the_page(world))))
     try:
-        chrome.call("Page.navigate", url=f"{hub_url}/#/group/{world['group_id']}")
+        chrome.call("Page.navigate",
+                    url=f"{world['hub_url']}/#/group/{world['group_id']}")
         # Refused as an unknown device: the form, and nothing shown as an error.
         s = chrome.until(_LINK_FORM, lambda s: s["button"])
         assert not s["error"], s
@@ -763,9 +791,16 @@ def test_the_application_links_itself_from_its_group_page(world):
         s = chrome.until(_LINK_FORM, lambda s: s["opened"] and not s["form"], timeout=60)
         assert not s["error"], s
         assert s["minted"] == 1, "the identity linked is not the one kept"
-    finally:
+        return chrome
+    except BaseException:
         chrome.close()
+        raise
+
 
+@pytest.mark.skipif(not shutil.which("google-chrome"),
+                    reason="google-chrome is not available")
+def test_the_application_links_itself_from_its_group_page(world):
+    _linked_group_page(world).close()
 
 
 # ── A second account: invited, joined, its devices, browser access, leaving ──
@@ -916,3 +951,69 @@ def test_search_finds_a_file_and_get_takes_its_path(world, tmp_path):
     assert (tmp_path / "plage de l'ouest.jpg").stat().st_size == 3000
     r = cli(world, "search", "no-such-thing-anywhere", home="cli-search")
     assert r.returncode == 3 and "Nothing found" in r.stderr
+
+
+def _cli_signed_in(world: dict, home: str) -> None:
+    deadline = time.monotonic() + 90
+    while True:
+        r = cli(world, "--hub", world["hub_url"], "-u", USERNAME, "login",
+                "--password-file", "-", stdin=PASSPHRASE + "\n", home=home)
+        if r.returncode == 0:
+            return
+        assert "Too many attempts" in r.stderr and time.monotonic() < deadline, r.stderr
+        time.sleep(5)
+
+
+@pytest.mark.skipif(not shutil.which("google-chrome"),
+                    reason="google-chrome is not available")
+def test_chat_between_the_command_line_and_the_application(world):
+    """Posted here, read on the group page in Chrome; answered there, read here —
+    the Python seal and the page's crypto.js, on a real node."""
+    group = _handle()
+    asyncio.run(world["app"].revoke_the_others())    # room for this machine and the page
+    _cli_signed_in(world, "cli-chat")
+    r = cli(world, "chat", group, "-m", "hello from the terminal 1", home="cli-chat")
+    if r.returncode == 5:                         # a new device here: approved first
+        found = re.search(r"enter\s+([0-9A-Z]{4}-[0-9A-Z]{4})", r.stderr)
+        assert found, r.stderr
+        code = found.group(1)
+        asyncio.run(world["app"].approve(code))
+        r = cli(world, "chat", group, "-m", "hello from the terminal 1", home="cli-chat")
+    assert r.returncode == 0, r.stderr
+    for i in range(2, 13):
+        r = cli(world, "chat", group, "-m", f"hello from the terminal {i}", home="cli-chat")
+        assert r.returncode == 0, r.stderr
+
+    r = cli(world, "chat", group, home="cli-chat")
+    assert r.returncode == 0, r.stderr
+    lines = r.stdout.splitlines()
+    assert len(lines) == 10
+    assert lines[0].endswith(f"{USERNAME}: hello from the terminal 3")
+    assert lines[-1].endswith(f"{USERNAME}: hello from the terminal 12")
+    r = cli(world, "chat", "-100", group, home="cli-chat")
+    assert len(r.stdout.splitlines()) == 12, r.stdout
+
+    chrome = _linked_group_page(world)
+    try:
+        # The page opens the terminal's messages: signature, epoch key, content.
+        chrome.until("document.body.innerText",
+                     lambda text: "hello from the terminal 12" in (text or ""))
+        assert "cannot be read" not in chrome.js("document.body.innerText").lower()
+        chrome.js("""(() => {
+            const box = document.querySelector('.chat-input');
+            box.value = 'hello from the browser';
+            box.dispatchEvent(new Event('input', { bubbles: true }));
+        })()""")
+        chrome.until("!document.querySelector('.chat-send').disabled", bool)
+        chrome.js("document.querySelector('.chat-send').click()")
+        chrome.until("document.querySelector('.chat-input').value === ''", bool)
+    finally:
+        chrome.close()
+
+    deadline = time.monotonic() + 30
+    while True:
+        r = cli(world, "chat", "-1", group, home="cli-chat")
+        if r.stdout.strip().endswith(f"{USERNAME}: hello from the browser"):
+            break
+        assert time.monotonic() < deadline, (r.stdout, r.stderr)
+        time.sleep(1)