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
|
"""
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_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]]
|