1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
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)
|