fix(node): chat is bounded in size and in rate

A chat message is the plainest member-supplied write there is: the node stores
it in `chat.db`, where nothing expires it — retention is a manual command
(§6.6) — relays it to every other connected member, and has the hub write a
notification for every member of the group. Nothing bounded any of it. The only
ceiling was the DataChannel frame, 64 MB once the handshake is done, so one
member in a loop filled the operator's disk and saturated everyone else's
connection, and the node's answer to each message was `ack`.

Uploads, the other member-supplied write, have carried a filename allowlist,
strict chunk ordering, a no-overwrite rule and a 4 GB cap since C5a — because
somebody asked what one member costs the others on that path. Nobody had asked
it on this one.

Two bounds, for the two halves of the question: **64 KB of ciphertext** for what
one message may cost, and **60 a minute per account per group** for how often
one member may impose it. Both are checked before anything is stored or relayed;
a refusal names itself and is audited, so "why is my disk full" has an answer.
The rate is keyed by account, not by connection: a second tab does not make
anyone type faster, and keying on the session would hand a script one budget per
socket it opens.

No node-wide ceiling beside it, deliberately. The link-preview limiter has one
because a preview spends the *node's* egress and its third-party quota, which is
one shared thing; a chat message spends the sender's own group, and a node-wide
ceiling would let a busy group silence a quiet one — this same defect one level
up. The last test in the new file is that property: a member at their limit has
not spent anybody else's.

Two things stay open on purpose and are named rather than quietly done:
retention still keeps everything, because a default that deletes people's
history is not a review's call; and the composer still offers to send an
oversized message, so this is §6.4's pattern with only the node half built.

§6.6 gains the rule, §13.5b the label — AV20, with AV21–AV23 registering the
three fixes this week that closed the same kind of gap elsewhere.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UMxEQadpzPkYLFf5CYKhpW

2 files changed, 288 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 59198d5..bc78644 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -183,6 +183,28 @@ _LINK_PREVIEW_RATE_WINDOW = 60.0
 _LINK_PREVIEW_RATE_PER_CONN = 15
 _LINK_PREVIEW_RATE_NODE = 60
 
+# Chat limits. A message is a member-supplied write onto the operator's disk
+# (`chat.db`, where retention is a manual CLI command — §6.6), relayed from there
+# to every other connected member and turned into a notification for every member
+# of the group. Nothing bounded any of it: the only ceiling was the frame size,
+# 64 MB once the handshake is done, so one member in a loop could fill the
+# operator's disk and saturate everyone else's connection. Uploads — the other
+# member-supplied write — have carried four protections and a size cap since
+# C5a; this is the same question asked of the path nobody had asked it of.
+#
+# 64 KB of ciphertext is about sixty thousand characters. The sealed payload is
+# the text, a thread id, a display name and a timestamp: an attachment is a file
+# on a root and travels as a reference (§4.5), so nothing legitimate comes close.
+MAX_CHAT_CIPHERTEXT = 64 * 1024
+# Per account per group, not per connection: a second tab does not make a person
+# type faster, and keying on the session would hand a script one budget per
+# socket. Sixty a minute is far above a human and far below a flood.
+_CHAT_RATE_WINDOW = 60.0
+_CHAT_RATE_PER_ACCOUNT = 60
+# When the map of senders grows past this, the stale entries are dropped. A node
+# with more live chatters than this in one window is not the case being bounded.
+_CHAT_RATE_MAX_TRACKED = 1000
+
 # Upload limits (finding C5a). Uploads used to land directly in the shared root under
 # a name the client chose, overwriting whatever was already there — which both violated
 # node sovereignty and defeated the delete authorization (overwrite a file, become its
@@ -4431,6 +4453,15 @@ class WebRTCPeerSession:
             self._audit("chat_refused", refusal)
             return
 
+        # Separate from the envelope check above, and deliberately: that one asks
+        # whether the message is well formed and authentic, this one asks what it
+        # costs everyone else. Before either is written to disk or relayed.
+        detail, code = self._chat_bounds_refusal(raw)
+        if detail:
+            self._send({"type": "error", "detail": detail, "code": code})
+            self._audit("chat_refused", code)
+            return
+
         if sender_name:
             self._user_names()[self._user_id] = sender_name
         if chat_store:
@@ -4538,6 +4569,56 @@ class WebRTCPeerSession:
 
         return ""
 
+    def _chat_bounds_refusal(self, ct: bytes) -> tuple[str, str]:
+        """What this message would cost the others, or ("", "") to accept it.
+
+        Two bounds, and each answers a different half of "who pays". A message
+        is written to `chat.db` on the operator's disk and kept — retention is a
+        manual command (§6.6) — then relayed to every other connected member and
+        turned into a notification for every member of the group. So **size**
+        bounds what one message costs, and **rate** bounds how often one member
+        may impose it.
+
+        There is deliberately no node-wide ceiling to go with the per-account
+        one. The link-preview limiter has both because a preview spends the
+        *node's* egress and its third-party quota, which is one shared thing; a
+        chat message spends the sender's own group. A node-wide chat ceiling
+        would let a busy group silence a quiet one, which is the same class of
+        defect this bound exists to close, one level up.
+        """
+        if len(ct) > MAX_CHAT_CIPHERTEXT:
+            return ("This message is too large to send in chat — "
+                    "send a large file as an attachment instead.",
+                    "chat_too_large")
+        if not self._chat_rate_ok():
+            return ("Too many messages just now — wait a moment.",
+                    "chat_rate_limited")
+        return ("", "")
+
+    def _chat_rate_ok(self) -> bool:
+        """True when this sender is within their window; records it when so.
+
+        Keyed by (group, account) on the transport context rather than on the
+        session: the sender is authenticated, so this is the one identifier a
+        second tab — or fifty of them — cannot multiply. The window is trimmed
+        on every call, and the map of senders is swept when it grows, so neither
+        can be the memory leak the bound was added to prevent.
+        """
+        now = time.monotonic()
+        hits: dict = self._ctx.setdefault("chat_hits", {})
+        if len(hits) > _CHAT_RATE_MAX_TRACKED:
+            for key, times in list(hits.items()):
+                if not times or now - times[-1] >= _CHAT_RATE_WINDOW:
+                    hits.pop(key, None)
+        key = (self._group_id or "", self._user_id or "")
+        mine = [t for t in hits.get(key, ()) if now - t < _CHAT_RATE_WINDOW]
+        if len(mine) >= _CHAT_RATE_PER_ACCOUNT:
+            hits[key] = mine
+            return False
+        mine.append(now)
+        hits[key] = mine
+        return True
+
     async def _store_chat_message(self, chat_store, **kwargs) -> None:
         """
         Persist one message, treating a replay as already-done.
diff --git a/packages/meshbay-node/tests/test_chat_is_bounded.py b/packages/meshbay-node/tests/test_chat_is_bounded.py
new file mode 100644
index 0000000..1af72b4
--- /dev/null
+++ b/packages/meshbay-node/tests/test_chat_is_bounded.py
@@ -0,0 +1,207 @@
+"""
+What one member's chat costs the others.
+
+This is the node-side entry of the register `docs/MESHBAY_DESIGN.md` §13.5b
+keeps — the question none of the three security reviews asked: *a participant
+supplies input; who else bears the cost?* Its sibling
+`meshbay-hub/tests/test_availability_between_members.py` holds the hub's cases
+and cannot hold this one, because the defect lives in the node's chat handler
+and needs the node's harness.
+
+A chat message is the plainest member-supplied write there is. The node stores
+it in `chat.db` on the operator's disk, where nothing expires it — retention is
+a manual CLI command (§6.6) — relays it to every other connected member, and
+has the hub write a notification for every member of the group. Uploads, the
+other member-supplied write, have carried a filename allowlist, strict chunk
+ordering, a no-overwrite rule and a 4 GB cap since C5a. Chat carried nothing:
+the only ceiling was the DataChannel frame, 64 MB once the handshake is done.
+One member in a loop filled the operator's disk and saturated everyone else's
+connection, and the node's own answer to each message was `ack`.
+
+Two bounds close it, and they answer different halves: **size** bounds what one
+message costs, **rate** bounds how often one member may impose it. There is
+deliberately no node-wide ceiling — a chat message spends the sender's own
+group, and a node-wide one would let a busy group silence a quiet one, which is
+this same defect one level up.
+
+The last test is the one that says the bound is the right shape: a member who
+has spent their budget has not spent anybody else's.
+"""
+
+import base64
+import os
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.chatbox import NONCE_LEN, SIG_LEN
+from meshbay_common.crypto import pk_to_b64
+from meshbay_node.chat import FORMAT_SEALED_V1, ChatStore
+from meshbay_node.transport import webrtc_server as ws
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+# Read with a default rather than imported. Against the source these were
+# written for there is no such bound and therefore no such constant, and a test
+# that dies of ImportError there proves only that a name is missing — the
+# assertions below are what say the node does the limiting.
+MAX_CHAT_CIPHERTEXT = getattr(ws, "MAX_CHAT_CIPHERTEXT", 64 * 1024)
+_CHAT_RATE_PER_ACCOUNT = getattr(ws, "_CHAT_RATE_PER_ACCOUNT", 60)
+
+pytestmark = pytest.mark.asyncio
+
+GROUP = "g" * 32
+
+
+@pytest.fixture
+async def store(tmp_path):
+    s = ChatStore(db_path=tmp_path / "chat.db")
+    await s.open()
+    yield s
+    await s.close()
+
+
+def _session(ctx, store, *, user: str, conn: str) -> WebRTCPeerSession:
+    """One connection, with a device already identified.
+
+    The node checks the envelope's *shape* and that the device named is the one
+    this connection proved; it verifies no signature, because the reader does
+    that (§4.5). So these tests need no real sealing to reach the bounds, which
+    is also a fair description of what the node itself knows.
+    """
+    device = pk_to_b64(Ed25519PrivateKey.generate().public_key())
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    ctx.setdefault("groups", {})[GROUP] = ctx["group_ctx"]
+    session._group_id = GROUP
+    session._user_id = user
+    session._username = user
+    session._pinned_pk = device
+    session._device_confirmed = True
+    session._registry_key = conn
+    session.sent = []
+    session._send = session.sent.append
+    session.audited = []
+    session._audit = lambda event, detail="": session.audited.append(event)
+    session._spawn = lambda coro: ctx["pending"].append(coro)
+    ctx["group_ctx"]["_peers"][conn] = session
+    return session
+
+
+@pytest.fixture
+def ctx(store):
+    return {
+        "group_ctx": {"chat_store": store, "gek": b"\x01" * 32, "_peers": {}},
+        "pending": [],
+    }
+
+
+async def _drain(ctx):
+    """Run the writes the handler spawned, as the loop would."""
+    for coro in ctx["pending"]:
+        await coro
+    ctx["pending"].clear()
+
+
+def _message(session, *, size: int):
+    return {
+        "format": FORMAT_SEALED_V1,
+        "epoch": 1,
+        "device": base64.b64decode(session._pinned_pk),
+        # A fresh nonce per message, as the real thing has (96 random bits,
+        # never a counter — §4.5). A fixed one makes the store's unique
+        # `(device, nonce)` refuse every message after the first as a replay,
+        # so a flood would look bounded here while being unbounded in
+        # production: the fixture, not the node, would be doing the limiting.
+        "nonce": os.urandom(NONCE_LEN),
+        "sig": b"\x00" * SIG_LEN,
+        "ct": b"x" * size,
+    }
+
+
+def _errors(session):
+    return [m for m in session.sent if m.get("type") == "error"]
+
+
+def _acks(session):
+    return [m for m in session.sent if m.get("type") == "ack"]
+
+
+# ── size ─────────────────────────────────────────────────────────────────────
+
+async def test_a_message_larger_than_any_text_is_refused(ctx, store):
+    alice = _session(ctx, store, user="alice", conn="c1")
+    alice._do_chat_message(_message(alice, size=1024 * 1024))
+    await _drain(ctx)
+
+    assert _errors(alice)[-1]["code"] == "chat_too_large"
+    assert not _acks(alice)
+    assert await store.message_count() == 0, (
+        "a megabyte of somebody's choosing reached the operator's disk")
+
+
+async def test_a_message_at_the_ceiling_is_still_sent(ctx, store):
+    """The bound has to admit what it claims to admit: a ceiling that refuses
+    an ordinary message is a broken feature, not a strict one."""
+    alice = _session(ctx, store, user="alice", conn="c1")
+    alice._do_chat_message(_message(alice, size=MAX_CHAT_CIPHERTEXT))
+    await _drain(ctx)
+
+    assert not _errors(alice), _errors(alice)
+    assert await store.message_count() == 1
+
+
+async def test_the_refusal_is_audited(ctx, store):
+    """An operator asking "why is my disk full" gets an answer."""
+    alice = _session(ctx, store, user="alice", conn="c1")
+    alice._do_chat_message(_message(alice, size=1024 * 1024))
+    await _drain(ctx)
+    assert "chat_refused" in alice.audited
+
+
+# ── rate ─────────────────────────────────────────────────────────────────────
+
+async def _flood(session, ctx, n):
+    for _ in range(n):
+        session._do_chat_message(_message(session, size=16))
+    await _drain(ctx)
+
+
+async def test_a_flood_stops_at_the_window(ctx, store):
+    alice = _session(ctx, store, user="alice", conn="c1")
+    await _flood(alice, ctx, _CHAT_RATE_PER_ACCOUNT + 25)
+
+    assert len(_acks(alice)) == _CHAT_RATE_PER_ACCOUNT
+    assert _errors(alice)[-1]["code"] == "chat_rate_limited"
+    assert await store.message_count() == _CHAT_RATE_PER_ACCOUNT, (
+        "the store kept growing after the bound was reached")
+
+
+async def test_a_second_tab_does_not_double_the_budget(ctx, store):
+    """Keyed by account, not by connection. A second tab does not make a person
+    type faster, and keying on the session would hand a script one budget per
+    socket it opens."""
+    first = _session(ctx, store, user="alice", conn="c1")
+    second = _session(ctx, store, user="alice", conn="c2")
+
+    await _flood(first, ctx, _CHAT_RATE_PER_ACCOUNT)
+    assert not _errors(first)
+
+    await _flood(second, ctx, 1)
+    assert _errors(second)[-1]["code"] == "chat_rate_limited"
+
+
+# ── and the reason it is per account ─────────────────────────────────────────
+
+async def test_one_member_at_their_limit_has_not_spent_anyone_elses(ctx, store):
+    """The property the whole bound exists for, and the reason there is no
+    node-wide ceiling beside it: a member who floods costs themselves their own
+    budget, and costs the group nothing it can notice."""
+    alice = _session(ctx, store, user="alice", conn="c1")
+    bob = _session(ctx, store, user="bob", conn="c2")
+
+    await _flood(alice, ctx, _CHAT_RATE_PER_ACCOUNT + 5)
+    assert _errors(alice)
+
+    await _flood(bob, ctx, 1)
+    assert not _errors(bob), "one member's flood silenced another"
+    assert _acks(bob)