1 files changed, 207 insertions, 0 deletions
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)