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
|
"""
The operator decides which group "applications" (Chat, Files, ...) are shown.
Same shape as `test_root_writable_policy.py`, because it is the same kind of
setting: changed by a signed operator instruction, stored on the node rather
than the hub, and safe for an existing group to have never heard of. The two
things specific to this one: the whole set is signed in one message rather
than one op per app, and an empty or unrecognised set is refused before a
challenge is ever issued — there is no file write to refuse afterwards the
way an unsigned upload is refused, so the check has to happen up front.
"""
from pathlib import Path
import pytest
from meshbay_common.adminop import OP_APPS_ENABLED
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roster import Roster
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from conftest import one_root
pytestmark = pytest.mark.asyncio
def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession:
shared_root = tmp_path / "shared"
shared_root.mkdir(exist_ok=True)
index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
ctx = {
"roots": one_root(shared_root),
"index": index,
"sk_node": index.sk_node,
"node_user_id": operator,
}
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
session._ctx = ctx
session._group_id = None
session._user_id = user_id
session._pk_user = ""
session.sent = []
session._send = session.sent.append
session._audit = lambda *a, **k: None
return session
# ── Refused before a challenge is even issued ───────────────────────────────
async def test_an_empty_set_is_refused(tmp_path):
"""Never let the operator lock a group down to nothing — no round trip
to the operator's browser needed to learn that."""
session = _session(tmp_path, "op", operator="op")
session._has_admin_authority = lambda: True
issued = []
session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
session._do_apps_enabled({"apps": []})
assert not issued
assert [m for m in session.sent if m.get("type") == "error"]
async def test_an_unknown_app_is_refused(tmp_path):
session = _session(tmp_path, "op", operator="op")
session._has_admin_authority = lambda: True
issued = []
session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
session._do_apps_enabled({"apps": ["chat", "videos"]})
assert not issued, "videos is not registered yet — accepting it would " \
"silently store a setting no client can act on"
assert [m for m in session.sent if m.get("type") == "error"]
async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
session = _session(tmp_path, "member-1", operator="the-operator")
session._has_admin_authority = lambda: False
session._do_apps_enabled({"apps": ["chat"]})
assert [m for m in session.sent if m.get("type") == "error"]
# ── Who may change it ───────────────────────────────────────────────────────
async def test_changing_it_needs_a_signature(tmp_path):
"""The request only ever produces a challenge. Nothing is applied until a
signature over the transcript verifies — the same path as the root ops."""
session = _session(tmp_path, "op", operator="op")
session._has_admin_authority = lambda: True
issued = []
session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
session._do_apps_enabled({"apps": ["files"]})
assert issued == [(OP_APPS_ENABLED, "files")]
async def test_the_subject_is_the_sorted_set_so_both_sides_build_the_same_transcript(tmp_path):
"""The operator's browser and the node must independently arrive at the
same subject string to sign/verify — order in the request must not
matter."""
session = _session(tmp_path, "op", operator="op")
session._has_admin_authority = lambda: True
issued = []
session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
session._do_apps_enabled({"apps": ["files", "chat"]})
assert issued == [(OP_APPS_ENABLED, "chat,files")]
# ── Where it is stored ──────────────────────────────────────────────────────
async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
roster = Roster(db_path=tmp_path / "roster.db")
await roster.open()
try:
assert sorted(await roster.enabled_apps("g1")) == ["chat", "files"], (
"absent must mean every registered app, or an upgrade hides one "
"for every existing group")
await roster.set_enabled_apps("g1", ["chat"], set_by="op")
# Files comes back whatever was stored: `enabled_apps` inserts it at
# the front on read, and `ops.set_enabled_apps` does the same on write,
# because Settings is the one way back if everything else were turned
# off. The assertion predates that guard -- the code is right and the
# test was describing the older behaviour.
assert await roster.enabled_apps("g1") == ["files", "chat"]
finally:
await roster.close()
reopened = Roster(db_path=tmp_path / "roster.db")
await reopened.open()
try:
assert await reopened.enabled_apps("g1") == ["files", "chat"]
assert sorted(await reopened.enabled_apps("g2")) == ["chat", "files"], (
"one group's setting must not answer for another")
finally:
await reopened.close()
|