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
|
"""A node-scoped daemon token must never reach the hub admin/moderator surface.
A node's authority and a hub role are different notions (docs/MESHBAY_DESIGN.md
§7.1, NS4): what a node may do is decided by its operator's roster pin on the
node and by the deliberately narrow node scope; being an admin or moderator is a
hub role on a *person's* account, exercised from a browser with a user-scoped
token. When the operator's account also holds a hub role — which is the case on
the reference deployment, where the operator is an admin and runs a node — the
`scope:"node"` token the daemon keeps in memory used to pass `require_admin` and
`require_moderator`, because those checked only the role and not the scope. The
scope gate was wired onto `require_user_scope` (group mutation) alone.
These are refusals: each asserts the node token is turned away with 403, and the
same account's user token is let through, so the guard is proven to bite on the
scope and not on the account.
"""
import base64
import time
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_hub.api.deps import set_admin_usernames
def _gen_ed25519():
sk = Ed25519PrivateKey.generate()
pk_raw = sk.public_key().public_bytes(
serialization.Encoding.Raw, serialization.PublicFormat.Raw)
return sk, base64.b64encode(pk_raw).decode()
async def _register(client, username):
r = await client.post("/v1/users/register", json={
"username": username, "email": f"{username}@test.local",
"auth_key": "k" * 44})
assert r.status_code == 201
async def _user_login(client, username):
r = await client.post("/v1/users/login", json={
"username": username, "auth_key": "k" * 44})
assert r.status_code == 200
return r.json()["access_token"]
async def _node_token(client, username, user_token):
"""Link a node key for `username` and return a scope:"node" token for it."""
sk_node, pk_node = _gen_ed25519()
r = await client.put("/v1/users/me/node_key",
json={"pk_node_ed25519": pk_node},
headers={"Authorization": f"Bearer {user_token}"})
assert r.status_code == 200
ts = int(time.time())
sig = sk_node.sign(f"meshbay:node_auth:{username}:{ts}".encode())
r = await client.post("/v1/nodes/auth", json={
"username": username, "timestamp": ts,
"signature": base64.b64encode(sig).decode()})
assert r.status_code == 200
data = r.json()
# Confirm we really are holding a node-scoped token.
import jwt as _jwt
assert _jwt.decode(data["access_token"], options={"verify_signature": False}
)["scope"] == "node"
return data["access_token"]
async def _admin_with_node(client, username="op_admin_test"):
"""An admin account that also runs a node — the reference-deployment case.
Returns (user_token, node_token) for the same account.
"""
await _register(client, username)
set_admin_usernames([username])
user_token = await _user_login(client, username)
node_token = await _node_token(client, username, user_token)
return user_token, node_token
def _bearer(token):
return {"Authorization": f"Bearer {token}"}
# ── require_admin ────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_node_token_cannot_reach_admin_stats(client):
user_token, node_token = await _admin_with_node(client)
assert (await client.get("/v1/admin/stats", headers=_bearer(node_token))
).status_code == 403
# The same account, from a browser, is admin and gets in.
assert (await client.get("/v1/admin/stats", headers=_bearer(user_token))
).status_code == 200
@pytest.mark.asyncio
async def test_node_token_cannot_revoke(client):
"""The sharpest one: revoke is signed and broadcast to every node."""
_, node_token = await _admin_with_node(client)
r = await client.post("/v1/admin/revoke", headers=_bearer(node_token),
json={"target": "group", "target_id": "whatever"})
assert r.status_code == 403
@pytest.mark.asyncio
async def test_node_token_cannot_change_instance_policy(client):
_, node_token = await _admin_with_node(client)
r = await client.patch("/v1/admin/settings", headers=_bearer(node_token),
json={"allow_public_groups": True})
assert r.status_code == 403
@pytest.mark.asyncio
async def test_node_token_cannot_delete_account(client):
_, node_token = await _admin_with_node(client)
r = await client.delete("/v1/admin/users/some-id", headers=_bearer(node_token))
assert r.status_code == 403
# ── require_moderator (a wider set of accounts, including the IP audit log) ───
@pytest.mark.asyncio
async def test_node_token_cannot_read_ip_audit_log(client):
user_token, node_token = await _admin_with_node(client)
assert (await client.get("/v1/admin/logs", headers=_bearer(node_token))
).status_code == 403
assert (await client.get("/v1/admin/logs", headers=_bearer(user_token))
).status_code == 200
@pytest.mark.asyncio
async def test_node_token_cannot_list_nodes(client):
_, node_token = await _admin_with_node(client)
assert (await client.get("/v1/admin/nodes", headers=_bearer(node_token))
).status_code == 403
@pytest.mark.asyncio
async def test_a_moderators_node_token_is_also_refused(client):
"""A moderator (not admin) who runs a node: the moderation surface is still
the person's, not the machine's."""
# An admin promotes a second account to moderator, which then links a node.
admin_user_token, _ = await _admin_with_node(client, "boss_admin_test")
await _register(client, "mod_test")
mod_user_token = await _user_login(client, "mod_test")
# promote
import jwt as _jwt
mod_id = _jwt.decode(mod_user_token, options={"verify_signature": False})["sub"]
r = await client.patch(f"/v1/admin/users/{mod_id}", json={"role": "moderator"},
headers=_bearer(admin_user_token))
assert r.status_code == 200
mod_node_token = await _node_token(client, "mod_test", mod_user_token)
# user-scoped moderator token gets in; node-scoped one does not
assert (await client.get("/v1/admin/stats", headers=_bearer(mod_user_token))
).status_code == 200
assert (await client.get("/v1/admin/stats", headers=_bearer(mod_node_token))
).status_code == 403
|