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
|
"""
The operator can close uploading to everyone but themselves.
The point of these tests is the difference between a hidden button and a closed
door. The interface stops offering the control, which is a courtesy to the
people who are not trying; **the node refuses the upload**, which is the part
that holds against someone who is. A member who kept an old tab open, or who
speaks MNP directly, gets the same answer as everyone else.
Two further things are worth holding:
* the setting is changed by a **signed** operator instruction. A node that took
it from an unsigned message would let any member turn it back on, and the
control would be a suggestion;
* it is stored on the **node**, not the hub. A hub that could decide who may
write to the operator's disk is a hub with authority over the node, which is
the thing this whole design is arranged to avoid.
"""
import base64
from pathlib import Path
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.adminop import OP_MEMBER_UPLOAD
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roster import Roster
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
from conftest import one_root
pytestmark = pytest.mark.asyncio
def _session(tmp_path: Path, user_id: str, *, member_upload: bool,
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,
"member_upload": member_upload,
"node_user_id": operator,
}
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
session._ctx = ctx
session._group_id = None
session._user_id = user_id
session._pk_user = ""
session._uploads = {}
session.sent = []
session._send = session.sent.append
session._audit = lambda *a, **k: None
return session
def _upload(session, filename="clip.mp4", body=b"bytes"):
session._do_file_upload({
"filename": filename, "chunk_index": 0, "total_chunks": 1,
"data": base64.b64encode(body).decode(),
})
def _uploads_dir(session) -> Path:
return session._ctx["roots"].upload_root.path / "uploads"
# ── The door, not the button ────────────────────────────────────────────────
async def test_a_member_cannot_upload_when_it_is_turned_off(tmp_path):
session = _session(tmp_path, "member-1", member_upload=False,
operator="the-operator")
_upload(session)
assert not (_uploads_dir(session) / "clip.mp4").exists(), (
"the file was written even though uploading is off — the setting is "
"decorative and the hidden button was the whole control")
refusal = [m for m in session.sent if m.get("type") == "error"]
assert refusal and refusal[0].get("code") == "member_upload_off"
async def test_the_operator_can_still_upload(tmp_path):
"""Otherwise turning it off locks the operator out of their own node, and
the only way back is a config file and a restart."""
session = _session(tmp_path, "the-operator", member_upload=False,
operator="the-operator")
_upload(session)
assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
async def test_members_upload_normally_when_it_is_on(tmp_path):
session = _session(tmp_path, "member-1", member_upload=True,
operator="the-operator")
_upload(session)
assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
async def test_a_node_that_never_heard_of_the_setting_still_accepts_uploads(tmp_path):
"""An existing node's context has no such key. The absence must read as
"allowed", or upgrading the node silently closes every group."""
session = _session(tmp_path, "member-1", member_upload=True,
operator="the-operator")
del session._ctx["member_upload"]
_upload(session)
assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
# ── 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 removing a member.
"""
session = _session(tmp_path, "member-1", member_upload=True,
operator="the-operator")
session._has_admin_authority = lambda: True
issued = []
session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
session._do_member_upload({"allowed": False})
assert issued == [(OP_MEMBER_UPLOAD, "off")]
assert session._ctx["member_upload"] is True, "applied before it was signed"
async def test_the_subject_names_the_outcome_not_the_operation(tmp_path):
"""The operator is shown the subject before signing. "member_upload" tells
them nothing; "off" tells them what they are about to do."""
session = _session(tmp_path, "op", member_upload=False, operator="op")
session._has_admin_authority = lambda: True
issued = []
session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
session._do_member_upload({"allowed": True})
assert issued == [(OP_MEMBER_UPLOAD, "on")]
async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
session = _session(tmp_path, "member-1", member_upload=True,
operator="the-operator")
session._has_admin_authority = lambda: False
session._do_member_upload({"allowed": False})
assert [m for m in session.sent if m.get("type") == "error"]
# ── 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 await roster.member_upload_allowed("g1") is True, (
"absent must mean allowed, or an upgrade closes every group")
await roster.set_member_upload("g1", False, set_by="op")
assert await roster.member_upload_allowed("g1") is False
finally:
await roster.close()
reopened = Roster(db_path=tmp_path / "roster.db")
await reopened.open()
try:
assert await reopened.member_upload_allowed("g1") is False
assert await reopened.member_upload_allowed("g2") is True, (
"one group's setting must not answer for another")
finally:
await reopened.close()
|