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
|
"""
Availability, on the node: what the people you invited can cost you.
The message budget for an unauthenticated peer (`PRE_HANDSHAKE_MAX_MSG`, H6)
bounds what *one* of them spends. Nothing bounded how many there could be, or
how long one could stay without ever proving anything — and the hub's cap is
three offers in flight **per account**, which is a limit on each caller rather
than on this machine, so an operator's exposure grew with the number of people
in their groups. A member who never meant any harm — a tab left open through a
sleep, a client retrying a connection it cannot complete — arrives here the
same way as one who does.
"""
import asyncio
from unittest.mock import MagicMock
import pytest
from meshbay_node.transport import webrtc_server as ws_mod
from meshbay_node.transport.webrtc_server import (
MAX_PEER_SESSIONS,
UNAUTHENTICATED_SESSION_TIMEOUT,
WebRTCTransport,
)
def _transport() -> WebRTCTransport:
return WebRTCTransport(sk_node=MagicMock(), hub_pk_pem=b"", gek=None,
roots=None, index=None)
class _FakeSession:
"""Stands in for a peer that connected and then said nothing."""
def __init__(self, user_id=None):
self._user_id = user_id
self.closed = False
async def close(self):
self.closed = True
@pytest.mark.asyncio
async def test_the_node_refuses_more_peers_than_it_will_hold():
"""The refusal comes before an RTCPeerConnection is allocated, or the cap
would be counting the thing it is meant to prevent."""
tp = _transport()
for i in range(MAX_PEER_SESSIONS):
tp._sessions[f"peer-{i}"] = _FakeSession(user_id="someone")
with pytest.raises(RuntimeError, match="peer-connection limit"):
await tp.handle_offer("v=0", "one-too-many")
assert "one-too-many" not in tp._sessions
@pytest.mark.asyncio
async def test_a_peer_that_never_handshakes_is_closed(monkeypatch):
"""`connectionstatechange` reaps a connection that *fails*. One that
succeeds and stays silent was held for the life of the daemon."""
monkeypatch.setattr(ws_mod, "UNAUTHENTICATED_SESSION_TIMEOUT", 0.05)
tp = _transport()
session = _FakeSession(user_id=None)
tp._sessions["quiet"] = session
tp._reap_if_unauthenticated("quiet")
await asyncio.sleep(0.2)
assert session.closed, "a peer that never proved anything was kept"
assert "quiet" not in tp._sessions
@pytest.mark.asyncio
async def test_a_peer_that_handshaked_is_left_alone(monkeypatch):
"""`_user_id` is set by the GEK proof, and is the one honest test of
whether this peer ever became anybody."""
monkeypatch.setattr(ws_mod, "UNAUTHENTICATED_SESSION_TIMEOUT", 0.05)
tp = _transport()
session = _FakeSession(user_id=None)
tp._sessions["real"] = session
tp._reap_if_unauthenticated("real")
session._user_id = "a-real-member" # the handshake completes
await asyncio.sleep(0.2)
assert not session.closed, "a member who completed the handshake was cut off"
assert tp._sessions["real"] is session
@pytest.mark.asyncio
async def test_handle_offer_arms_the_reaper(monkeypatch):
"""
The seam, driven rather than described. The three tests above call
`_reap_if_unauthenticated` themselves, so every one of them would still
pass with the call removed from `handle_offer` and no peer reaped at all —
which is the defect, not the helper. aiortc is stubbed because a real
RTCPeerConnection wants a real SDP; everything else here is the shipped
code path.
"""
class _FakePC:
def __init__(self, *a, **kw):
self.localDescription = MagicMock(sdp="v=0\r\n")
self.remoteDescription = None
def on(self, _event):
return lambda fn: fn
async def setRemoteDescription(self, _d): pass
async def createAnswer(self): return MagicMock()
async def setLocalDescription(self, _d): pass
async def close(self): pass
monkeypatch.setattr(ws_mod, "RTCPeerConnection", _FakePC)
monkeypatch.setattr(ws_mod, "RTCSessionDescription",
lambda **kw: MagicMock(**kw))
tp = _transport()
await tp.handle_offer("v=0\r\n", "fresh-peer")
try:
assert tp._reapers, (
"handle_offer allocated a peer session and armed nothing to "
"close it if the handshake never comes")
finally:
await tp.close_all()
class _StatefulPC:
"""An RTCPeerConnection that keeps the handlers the transport registers, so
a test can drive the connection through its states."""
def __init__(self, *a, **kw):
self.localDescription = MagicMock(sdp="v=0\r\n")
self.remoteDescription = None
self.connectionState = "new"
self.handlers = {}
def on(self, event):
def register(fn):
self.handlers[event] = fn
return fn
return register
async def setRemoteDescription(self, _d): pass
async def createAnswer(self): return MagicMock()
async def setLocalDescription(self, _d): pass
async def close(self): pass
@pytest.mark.asyncio
@pytest.mark.parametrize("state", ["closed", "failed"])
async def test_a_connection_that_ends_leaves_its_groups_peer_set(monkeypatch, state):
"""
A closed tab, a phone gone to sleep and a dead network all arrive at
`connectionstatechange`, which dropped the session from the transport and
stopped its tasks but left it in its group's peer set: only `close()` took
it out, and nothing on this path called it. Every broadcast to the group
was then written to a closed channel ("WebRTC send skipped", thirteen times
per settings change on a node up for a day), and every reconnect left one
more dead session held until the node restarted.
"""
pcs = []
def make_pc(*a, **kw):
pcs.append(_StatefulPC())
return pcs[-1]
monkeypatch.setattr(ws_mod, "RTCPeerConnection", make_pc)
monkeypatch.setattr(ws_mod, "RTCSessionDescription",
lambda **kw: MagicMock(**kw))
tp = _transport()
tp._ctx["groups"] = {"g1": {"name": "g1"}}
try:
await tp.handle_offer("v=0\r\n", "peer-1")
session = tp._sessions["peer-1"]
session._user_id, session._group_id = "a-member", "g1" # handshake done
session._register_peer()
peers = tp._ctx["groups"]["g1"]["_peers"]
assert list(peers.values()) == [session]
pcs[0].connectionState = state
await pcs[0].handlers["connectionstatechange"]()
assert "peer-1" not in tp._sessions
assert not peers, "the group still broadcasts to a connection that is gone"
finally:
await tp.close_all()
@pytest.mark.asyncio
async def test_the_reaper_is_held_and_cancelled_with_the_transport():
"""asyncio keeps only a weak reference to a task, and a reaper collected
mid-sleep reaps nothing — the same trap as every other task in this file."""
tp = _transport()
tp._sessions["held"] = _FakeSession()
tp._reap_if_unauthenticated("held")
assert tp._reapers, "the reaper was fired and forgotten"
await tp.close_all()
assert not tp._reapers
# The timeout is a real duration, not something a test has to wait out.
assert UNAUTHENTICATED_SESSION_TIMEOUT >= 30
|