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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
|
"""
Phase 11.5 security regression tests — node WebSocket registration (finding C2).
The hub relays every WebRTC offer for a node to whoever holds that node's entry in
`_connected_nodes`. That registration used to be established from a client-supplied
`node_id` with no ownership check, so any registered user could take over a victim
node's signaling and become the endpoint browsers connect to.
These exercise `_authorize_node_ws` directly rather than through a socket: it is the
function that makes the authorization decision, and the hub test harness uses
ASGITransport, which has no WebSocket support.
"""
import base64
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from meshbay_common.crypto import pk_to_b64
async def _make_user(client, username: str) -> dict:
"""Register + log in a user, returning ids, token and keys."""
sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate()
pk_ed, pk_x = pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key())
r = await client.post("/v1/users/register", json={
"username": username,
"email": f"{username}@example.test",
"auth_key": base64.b64encode(b"k" * 32).decode(),
"pk_user_ed25519": pk_ed,
"pk_user_x25519": pk_x,
})
assert r.status_code == 201, r.text
user_id = r.json()["user_id"]
r = await client.post("/v1/users/login", json={
"username": username,
"auth_key": base64.b64encode(b"k" * 32).decode(),
})
assert r.status_code == 200, r.text
return {"user_id": user_id, "token": r.json()["access_token"], "pk_ed": pk_ed}
async def _announce_node(client, user: dict) -> str:
r = await client.post(
"/v1/nodes/announce",
json={"pk_node": user["pk_ed"], "endpoint_hint": "test"},
headers={"Authorization": f"Bearer {user['token']}"},
)
assert r.status_code == 201, r.text
return r.json()["node_id"]
def _node_token(user: dict) -> str:
from meshbay_hub.auth import issue_access_token
return issue_access_token(user["user_id"], user["pk_ed"], scope="node")
@pytest.mark.asyncio
async def test_ws_rejects_user_scoped_token(client):
"""C2: a browser token must never be able to register as a node."""
from meshbay_hub.api.revocation import _authorize_node_ws
victim = await _make_user(client, "victim1")
node_id = await _announce_node(client, victim)
resolved, detail = await _authorize_node_ws(victim["token"], node_id, None)
assert resolved is None
assert "node-scoped" in detail.lower()
@pytest.mark.asyncio
async def test_ws_rejects_foreign_node_id(client):
"""
C2: the impersonation itself. An attacker with a perfectly valid node-scoped
token of their own must not be able to claim someone else's node_id.
"""
from meshbay_hub.api.revocation import _authorize_node_ws
victim = await _make_user(client, "victim2")
attacker = await _make_user(client, "attacker2")
victim_node = await _announce_node(client, victim)
await _announce_node(client, attacker)
resolved, detail = await _authorize_node_ws(
_node_token(attacker), victim_node, None)
assert resolved is None, "attacker hijacked the victim's node registration (C2)"
assert "does not belong" in detail.lower()
@pytest.mark.asyncio
async def test_ws_rejects_unknown_node_id(client):
"""C2: an invented node_id must not register either."""
from meshbay_hub.api.revocation import _authorize_node_ws
user = await _make_user(client, "user3")
resolved, _ = await _authorize_node_ws(_node_token(user), "no-such-node", None)
assert resolved is None
@pytest.mark.asyncio
async def test_ws_rejects_missing_node_id(client):
"""C2: identity may not fall back to the token subject."""
from meshbay_hub.api.revocation import _authorize_node_ws
user = await _make_user(client, "user4")
resolved, _ = await _authorize_node_ws(_node_token(user), "", None)
assert resolved is None
@pytest.mark.asyncio
async def test_ws_accepts_own_node(client):
"""The legitimate path still works."""
from meshbay_hub.api.revocation import _authorize_node_ws
user = await _make_user(client, "owner5")
node_id = await _announce_node(client, user)
resolved, groups = await _authorize_node_ws(_node_token(user), node_id, None)
assert resolved == node_id
assert groups == []
@pytest.mark.asyncio
async def test_ws_group_claims_cannot_widen_beyond_membership(client):
"""
C2: `group_ids` used to be taken verbatim, letting a node advertise itself as
an online source for any group on the hub and attract clients to it.
"""
from meshbay_hub.api.revocation import _authorize_node_ws
user = await _make_user(client, "owner6")
node_id = await _announce_node(client, user)
r = await client.post(
"/v1/groups",
json={"name": "mine", "visibility": "private", "join_policy": "invite"},
headers={"Authorization": f"Bearer {user['token']}"},
)
assert r.status_code == 201, r.text
own_group = r.json()["group_id"]
resolved, groups = await _authorize_node_ws(
_node_token(user), node_id, [own_group, "someone-elses-group"])
assert resolved == node_id
assert groups == [own_group], "node advertised a group it is not a member of"
@pytest.mark.asyncio
async def test_signaling_rejects_non_member(client):
"""
H6/H4: POST /v1/nodes/{id}/webrtc/offer was reachable by any authenticated
user for any node, with no membership check and no rate limit. Each call makes
the target node allocate an aiortc PeerConnection and gather ICE, so it was a
remote resource-exhaustion primitive against a third party's machine.
"""
from meshbay_hub.api import revocation as rev
owner = await _make_user(client, "owner8")
outsider = await _make_user(client, "outsider8")
node_id = await _announce_node(client, owner)
r = await client.post(
"/v1/groups",
json={"name": "private-g", "visibility": "private", "join_policy": "invite"},
headers={"Authorization": f"Bearer {owner['token']}"},
)
group_id = r.json()["group_id"]
# Pretend the node is connected and hosting that group.
class _FakeWS:
async def send_text(self, _):
raise AssertionError("offer relayed to node despite non-membership")
rev._connected_nodes[node_id] = _FakeWS()
rev._node_groups[node_id] = [group_id]
try:
resp = await client.post(
f"/v1/nodes/{node_id}/webrtc/offer",
json={"sdp": "v=0", "ice_candidates": []},
headers={"Authorization": f"Bearer {outsider['token']}"},
)
assert resp.status_code == 403, resp.text
finally:
rev._connected_nodes.pop(node_id, None)
rev._node_groups.pop(node_id, None)
@pytest.mark.asyncio
async def test_signaling_rejects_oversized_sdp(client):
"""H6: an SDP offer is ~2 KB; unbounded input is a memory amplifier."""
user = await _make_user(client, "user9")
resp = await client.post(
"/v1/nodes/whatever/webrtc/offer",
json={"sdp": "v=0" + ("x" * 200_000), "ice_candidates": []},
headers={"Authorization": f"Bearer {user['token']}"},
)
assert resp.status_code == 413
@pytest.mark.asyncio
async def test_incoming_rejects_foreign_peer_ip(client):
"""
H6: peer_ip was taken verbatim, letting any user make an arbitrary node emit
UDP packets to an address of their choosing — reflection via someone else's
machine. The probe target must be the caller's own address.
"""
from meshbay_hub.api import revocation as rev
owner = await _make_user(client, "owner10")
node_id = await _announce_node(client, owner)
class _FakeWS:
async def send_text(self, _):
raise AssertionError("punch relayed with attacker-chosen peer_ip")
rev._connected_nodes[node_id] = _FakeWS()
try:
resp = await client.post(
f"/v1/nodes/{node_id}/incoming",
json={"peer_ip": "198.51.100.7", "peer_port": 9999},
headers={"Authorization": f"Bearer {owner['token']}"},
)
assert resp.status_code == 403, resp.text
finally:
rev._connected_nodes.pop(node_id, None)
@pytest.mark.asyncio
async def test_ws_node_may_narrow_its_group_set(client):
"""A node hosting a subset of the operator's groups may say so."""
from meshbay_hub.api.revocation import _authorize_node_ws
user = await _make_user(client, "owner7")
node_id = await _announce_node(client, user)
created = []
for name in ("g-one", "g-two"):
r = await client.post(
"/v1/groups",
json={"name": name, "visibility": "private", "join_policy": "invite"},
headers={"Authorization": f"Bearer {user['token']}"},
)
created.append(r.json()["group_id"])
resolved, groups = await _authorize_node_ws(
_node_token(user), node_id, [created[0]])
assert resolved == node_id
assert groups == [created[0]]
|