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
|
"""
Rotating the group key and forgetting a pinned identity — `ops.set_gek` and
`ops.unpin_member`, which the Node page and the CLI reach over loopback — and
the H5 rule every signed MNP operation lives under.
`gek_rotate` and `member_unpin` were MNP messages until 6.0. No client sent
them, so they went; what they did is still tested here, at the door that is
used. **"Nothing arriving over MNP can activate a GEK" is about key material
arriving from outside** (C5b): the node generates the new key with its own
CSPRNG, which is the only thing that finishes a revocation — the ex-member
still holds the current key.
"""
import base64
from pathlib import Path
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.adminop import OP_CHAT_EPOCH, admin_transcript
from meshbay_common.crypto import pk_to_b64
from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
from meshbay_node import ops
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roster import open_roster
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
from conftest import one_root
GROUP = "g" * 32
@pytest.fixture
async def roster(tmp_path):
r = await open_roster(tmp_path)
yield r
await r.close()
def _keypair():
sk = Ed25519PrivateKey.generate()
return sk, pk_to_b64(sk.public_key())
async def _session(tmp_path: Path, roster, *, operator: bool) -> WebRTCPeerSession:
"""A peer session with an operator pinned, or deliberately without one."""
shared = tmp_path / "shared"
shared.mkdir(exist_ok=True)
index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
roots = one_root(shared)
sk_op, pk_op = _keypair()
if operator:
await roster.pin_identity("grenet", "grenet", pk_op, pk_op, "code")
await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")
group_ctx = {"gek": b"\x01" * 32, "roots": roots, "index": index,
"join_policy": "invite"}
state = {
"groups_ctx": {GROUP: group_ctx},
"roster": roster,
"indexes": {GROUP: index},
"bundle_store": _FakeBundleStore(),
"pk_x25519_raw": b"\x02" * 32,
"hub": _FakeHub(),
"node_user_id": "node-user",
}
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
session._ctx = {
"roots": roots, "index": index, "sk_node": index.sk_node,
"roster": roster, "groups": {GROUP: group_ctx},
"has_admin_authority": operator,
"daemon_state": state,
}
session._group_id = GROUP
session._user_id = "grenet" if operator else "mallory"
session._username = session._user_id
session._pk_user = ""
session._uploads = {}
session._admin_ops = {}
session._remote_ip = ""
session.sent = []
session._send = session.sent.append
session._audit = lambda *a, **k: None
session.state = state
session.sk_op = sk_op
session.spawned = []
session._spawn = session.spawned.append
return session
class _FakeBundleStore:
def __init__(self):
self.stored = []
self.deleted_keypairs = []
async def store(self, *args):
self.stored.append(args)
async def delete_keypair(self, user_id):
self.deleted_keypairs.append(user_id)
return True
class _FakeHub:
class _S:
user_id = "node-user"
_session = _S()
def _last(session):
return session.sent[-1] if session.sent else {}
async def _drain(session):
"""Await whatever `_spawn` started. The real session holds its tasks; a
hand-built one collects them here so the assertion sees the result."""
for coro in session.spawned:
await coro
session.spawned.clear()
# ── H5: a signature is for one operation ─────────────────────────────────────
async def test_a_signature_over_another_operation_does_not_count(tmp_path, roster):
"""
H5's rule: the node rebuilds the transcript from the operation it is holding
and verifies against *that*, so a signature collected for one act cannot be
presented as another.
Driven through `_do_admin_response`, deliberately. Handing a transcript
straight to `_admin_exec_*` would skip the reconstruction that is the
control, and the test would pass while proving nothing.
"""
session = await _session(tmp_path, roster, operator=True)
_, pk_bob = _keypair()
await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code")
await roster.set_member(GROUP, "bob", ROLE_MEMBER, "active", "code")
session._do_member_revoke({"user_id": "bob"})
challenge = _last(session)
assert challenge["type"] == "admin_challenge", challenge
# Signed over chat_epoch, presented against the pending member_revoke.
wrong = admin_transcript(
op=OP_CHAT_EPOCH, node_pk_b64=session._node_pk_b64(), group_id=GROUP,
subject="bob", nonce=base64.b64decode(challenge["nonce"]),
ts=challenge["ts"])
session._do_admin_response({
"op_id": challenge["op_id"],
"signature": base64.b64encode(session.sk_op.sign(wrong)).decode(),
})
await _drain(session)
assert _last(session)["type"] == "error"
assert (await roster.get_member(GROUP, "bob"))["status"] == "active"
# ── Rotating the group key ───────────────────────────────────────────────────
async def test_rotating_makes_a_new_key_on_the_node(tmp_path, roster):
session = await _session(tmp_path, roster, operator=True)
before = session.state["groups_ctx"][GROUP]["gek"]
result = await ops.set_gek(session.state, GROUP, rotate=True)
assert result["rotated"] is True
after = session.state["groups_ctx"][GROUP]["gek"]
assert after != before, "the key did not change"
assert len(after) == 32
# Produced here, not received: no key material crossed the wire (C5b).
assert session.state["bundle_store"].stored, (
"the node's own copy was not stored — the daemon could not reload it")
async def test_rotation_reaches_the_index(tmp_path, roster):
"""The index is encrypted under the GEK. Leaving the old key on it would
serve members a listing they cannot open."""
session = await _session(tmp_path, roster, operator=True)
await ops.set_gek(session.state, GROUP, rotate=True)
assert session.state["indexes"][GROUP].gek == \
session.state["groups_ctx"][GROUP]["gek"]
# ── Forgetting a pinned identity ─────────────────────────────────────────────
async def test_unpinning_forgets_the_identity_and_its_bundle(tmp_path, roster):
session = await _session(tmp_path, roster, operator=True)
_, pk_bob = _keypair()
await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code")
await ops.unpin_member(session.state, "bob")
assert await roster.get_identity("bob") is None
# The stored keypair bundle goes too — left behind it blocks the re-join
# the unpin exists to enable.
assert "bob" in session.state["bundle_store"].deleted_keypairs
async def test_unpinning_someone_unknown_says_so(tmp_path, roster):
session = await _session(tmp_path, roster, operator=True)
with pytest.raises(ops.OpError, match="No such pinned identity"):
await ops.unpin_member(session.state, "nobody")
# ── What MNP no longer carries ───────────────────────────────────────────────
@pytest.mark.parametrize("op", ["gek_rotate", "member_unpin", "transfer_limits",
"group_detach"])
def test_operations_no_client_sent_are_not_signed_ops_any_more(op):
"""Gone with MNP 6.0: a door nobody uses is an untested way in. The Node
page and the CLI do the same work over loopback."""
from meshbay_common import adminop
from meshbay_node.transport.webrtc.admin import _ADMIN_EXECUTORS
from meshbay_node.transport.webrtc.dispatch import _HANDLERS
assert op not in _HANDLERS
assert op not in _ADMIN_EXECUTORS
assert op not in vars(adminop).values()
|