diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_admin_ops_mnp.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_admin_ops_mnp.py | 174 |
1 files changed, 49 insertions, 125 deletions
diff --git a/packages/meshbay-node/tests/test_admin_ops_mnp.py b/packages/meshbay-node/tests/test_admin_ops_mnp.py index 7fd1c2b..898228e 100644 --- a/packages/meshbay-node/tests/test_admin_ops_mnp.py +++ b/packages/meshbay-node/tests/test_admin_ops_mnp.py @@ -1,17 +1,14 @@ """ -`gek_rotate` and `member_unpin` over MNP. +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. -Both are destructive and both are new, so the tests are negative assertions: -nobody without the operator's pinned key can reach them, a signature over the -wrong transcript does not count, and the operation cannot be triggered by the -request message alone. - -The rule these live under is worth restating, because it is easy to read -draft-v5 §5.1 as forbidding them: **"nothing arriving over MNP can activate a -GEK" is about key material arriving from outside** (C5b — a member handing the -node a key of their choosing). An operator-signed instruction where the node -generates the key with its own CSPRNG is a different shape, and it is the only -thing that finishes a revocation: the ex-member still holds the current key. +`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 @@ -19,14 +16,10 @@ from pathlib import Path import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey -from meshbay_common.adminop import ( - OP_GEK_ROTATE, - OP_MEMBER_UNPIN, - admin_transcript, -) +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_common.protocol import MNP +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 @@ -127,58 +120,7 @@ async def _drain(session): session.spawned.clear() -async def _sign_and_exec(session, op: str, subject: str, sk, exec_fn): - challenge = _last(session) - assert challenge["type"] == "admin_challenge", challenge - transcript = admin_transcript( - op=op, node_pk_b64=session._node_pk_b64(), group_id=GROUP, - subject=subject, nonce=base64.b64decode(challenge["nonce"]), - ts=challenge["ts"]) - pending = session._admin_ops.get(challenge["op_id"]) or { - "op": op, "subject": subject} - await exec_fn(pending, transcript, sk.sign(transcript)) - - -# ── gek_rotate ─────────────────────────────────────────────────────────────── - -async def test_rotation_needs_an_operator(tmp_path, roster): - """Without a paired operator there is nobody who could sign, so the node - fails closed and says why rather than issuing a challenge nobody can meet.""" - session = await _session(tmp_path, roster, operator=False) - - session._do_gek_rotate({}) - - assert _last(session)["type"] == "error" - assert "authorized key" in _last(session)["detail"] - assert not session._admin_ops - - -async def test_the_request_alone_rotates_nothing(tmp_path, roster): - """The message asks; only a signature acts. A node that rotated here would - let any member lock the group out.""" - session = await _session(tmp_path, roster, operator=True) - before = session._ctx["groups"][GROUP]["gek"] - - session._do_gek_rotate({}) - - assert _last(session)["type"] == "admin_challenge" - assert session._ctx["groups"][GROUP]["gek"] == before - - -async def test_a_members_signature_does_not_rotate(tmp_path, roster): - session = await _session(tmp_path, roster, operator=True) - sk_mallory, pk_mallory = _keypair() - await roster.pin_identity("mallory", "mallory", pk_mallory, pk_mallory, "code") - await roster.set_member(GROUP, "mallory", ROLE_MEMBER, "active", "grenet") - before = session._ctx["groups"][GROUP]["gek"] - - session._do_gek_rotate({}) - await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, sk_mallory, - session._admin_exec_gek_rotate) - - assert _last(session)["type"] == "error" - assert session._ctx["groups"][GROUP]["gek"] == before - +# ── H5: a signature is for one operation ───────────────────────────────────── async def test_a_signature_over_another_operation_does_not_count(tmp_path, roster): """ @@ -191,15 +133,18 @@ async def test_a_signature_over_another_operation_does_not_count(tmp_path, roste control, and the test would pass while proving nothing. """ session = await _session(tmp_path, roster, operator=True) - before = session._ctx["groups"][GROUP]["gek"] + _, 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_gek_rotate({}) + session._do_member_revoke({"user_id": "bob"}) challenge = _last(session) + assert challenge["type"] == "admin_challenge", challenge - # Signed over member_unpin, presented against the pending gek_rotate. + # Signed over chat_epoch, presented against the pending member_revoke. wrong = admin_transcript( - op=OP_MEMBER_UNPIN, node_pk_b64=session._node_pk_b64(), group_id=GROUP, - subject=GROUP, nonce=base64.b64decode(challenge["nonce"]), + 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"], @@ -208,19 +153,18 @@ async def test_a_signature_over_another_operation_does_not_count(tmp_path, roste await _drain(session) assert _last(session)["type"] == "error" - assert session.state["groups_ctx"][GROUP]["gek"] == before + assert (await roster.get_member(GROUP, "bob"))["status"] == "active" -async def test_the_operator_rotates_and_the_node_makes_the_key(tmp_path, roster): +# ── 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._ctx["groups"][GROUP]["gek"] + before = session.state["groups_ctx"][GROUP]["gek"] - session._do_gek_rotate({}) - await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, session.sk_op, - session._admin_exec_gek_rotate) + result = await ops.set_gek(session.state, GROUP, rotate=True) - ack = _last(session) - assert ack["type"] == MNP.GEK_ROTATE_ACK, ack + assert result["rotated"] is True after = session.state["groups_ctx"][GROUP]["gek"] assert after != before, "the key did not change" assert len(after) == 32 @@ -234,54 +178,21 @@ async def test_rotation_reaches_the_index(tmp_path, roster): serve members a listing they cannot open.""" session = await _session(tmp_path, roster, operator=True) - session._do_gek_rotate({}) - await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, session.sk_op, - session._admin_exec_gek_rotate) + await ops.set_gek(session.state, GROUP, rotate=True) assert session.state["indexes"][GROUP].gek == \ session.state["groups_ctx"][GROUP]["gek"] -# ── member_unpin ───────────────────────────────────────────────────────────── - -async def test_unpinning_needs_an_operator(tmp_path, roster): - session = await _session(tmp_path, roster, operator=False) - session._do_member_unpin({"user_id": "bob"}) - assert _last(session)["type"] == "error" - - -async def test_unpinning_yourself_is_refused(tmp_path, roster): - """It would end the authority of the connection performing the operation, - halfway through it.""" - session = await _session(tmp_path, roster, operator=True) - session._do_member_unpin({"user_id": "grenet"}) - assert _last(session)["detail"] == "Cannot unpin yourself" - - -async def test_a_members_signature_does_not_unpin(tmp_path, roster): - session = await _session(tmp_path, roster, operator=True) - sk_bob, pk_bob = _keypair() - await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code") - - session._do_member_unpin({"user_id": "bob"}) - await _sign_and_exec(session, OP_MEMBER_UNPIN, "bob", sk_bob, - session._admin_exec_member_unpin) - - assert _last(session)["type"] == "error" - assert await roster.get_identity("bob") is not None, ( - "a member removed their own pin — only the operator may") - +# ── Forgetting a pinned identity ───────────────────────────────────────────── -async def test_the_operator_unpins(tmp_path, roster): +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") - session._do_member_unpin({"user_id": "bob"}) - await _sign_and_exec(session, OP_MEMBER_UNPIN, "bob", session.sk_op, - session._admin_exec_member_unpin) + await ops.unpin_member(session.state, "bob") - assert _last(session)["type"] == MNP.MEMBER_UNPIN_ACK 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. @@ -290,8 +201,21 @@ async def test_the_operator_unpins(tmp_path, roster): async def test_unpinning_someone_unknown_says_so(tmp_path, roster): session = await _session(tmp_path, roster, operator=True) - session._do_member_unpin({"user_id": "nobody"}) - await _sign_and_exec(session, OP_MEMBER_UNPIN, "nobody", session.sk_op, - session._admin_exec_member_unpin) - assert _last(session)["type"] == "error" - assert "No such pinned identity" in _last(session)["detail"] + 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() |