""" Roster and operator pairing (M3, and the mechanism that will close H3). Negative assertions, per the posture set in Phase 11.5: each test states an attack or a mistake that must not work. The one to keep an eye on is `test_daemon_does_not_auto_pin_keystore_key` — the auto-pin is what made node sovereignty inert as shipped, and it fails closed, so nothing else in the suite notices if it comes back. See `docs/invite-pairing-v1.md`. """ import base64 import time from pathlib import Path import pytest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from meshbay_common.crypto import generate_gek, pk_to_b64, unwrap_gek_aes from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.roster import Roster, hash_code, normalize_code from meshbay_node.transport.webrtc_server import WebRTCPeerSession # ── Fixtures ────────────────────────────────────────────────────────────────── @pytest.fixture async def roster(tmp_path): r = Roster(db_path=tmp_path / "roster.db") await r.open() yield r await r.close() def _keypair_full(): """(sk_ed, pk_ed_b64, pk_x_b64, sk_x) — the X25519 secret is needed to unwrap.""" sk_ed = Ed25519PrivateKey.generate() sk_x = X25519PrivateKey.generate() pk_ed_b64 = pk_to_b64(sk_ed.public_key()) pk_x_b64 = base64.b64encode( sk_x.public_key().public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) ).decode() return sk_ed, pk_ed_b64, pk_x_b64, sk_x def _keypair(): sk_ed, pk_ed_b64, pk_x_b64, _ = _keypair_full() return sk_ed, pk_ed_b64, pk_x_b64 def _session(tmp_path: Path, roster, user_id: str = "grenet", group_id: str | None = None, gek: bytes | None = None, join_policy: str = "invite") -> WebRTCPeerSession: """A peer session with the join path wired and sending stubbed out.""" shared_root = tmp_path / "shared" shared_root.mkdir(exist_ok=True) index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = { "shared_root": shared_root, "index": index, "sk_node": index.sk_node, "roster": roster, } if group_id: session._ctx["groups"] = { group_id: { "gek": gek, "shared_root": shared_root, "index": index, "join_policy": join_policy, }, } session._group_id = group_id session._user_id = user_id session._username = user_id session._pk_user = "" session._uploads = {} session._join_attempts = 0 session._nonce_node = b"\x11" * 32 session._remote_ip = "" session.sent = [] session._send = session.sent.append session._audit = lambda *a, **k: None return session def _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="", user_id="grenet", group_id="", nonce=None, ts=None): ts = int(time.time()) if ts is None else ts transcript = join_transcript( node_pk_b64=session._node_pk_b64(), group_id=group_id, user_id=user_id, pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64, nonce_node=nonce if nonce is not None else session._nonce_node, ts=ts, ) return { "type": "join_request", "group_id": group_id, "pk_ed25519": pk_ed_b64, "pk_x25519": pk_x_b64, "code": code, "ts": ts, "sig": base64.b64encode(sk_ed.sign(transcript)).decode(), } def _last(session): return session.sent[-1] if session.sent else {} # ── Roster ──────────────────────────────────────────────────────────────────── async def test_invite_is_single_use(roster): code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") assert await roster.consume_invite(code, "grenet") is not None assert await roster.consume_invite(code, "grenet") is None, ( "a pairing code must not be redeemable twice") async def test_invite_is_bound_to_one_account(roster): """A leaked code must be useless to whoever finds it.""" code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") assert await roster.consume_invite(code, "eve") is None assert await roster.consume_invite(code, "grenet") is not None async def test_expired_invite_is_refused(roster): code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli", ttl=-1) assert await roster.consume_invite(code, "grenet") is None async def test_reinvite_supersedes_the_previous_code(roster): first = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") second = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") assert await roster.consume_invite(first, "grenet") is None assert await roster.consume_invite(second, "grenet") is not None async def test_codes_are_not_stored_in_the_clear(roster, tmp_path): code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") rows = await roster.list_invites() assert rows and rows[0]["code_hash"] != normalize_code(code) assert rows[0]["code_hash"] == hash_code(code) def test_code_normalization_absorbs_human_error(): """Someone reading a code aloud must not be able to get it wrong.""" assert normalize_code("k7m2-qx4p") == normalize_code("K7M2QX4P") assert normalize_code("O1IL") == "0111" assert normalize_code(" k7m2 qx4p ") == "K7M2QX4P" async def test_operator_pks_reflect_unpinning(roster): _, pk_ed_b64, pk_x_b64 = _keypair() await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code") await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli") assert await roster.operator_pks() == [pk_ed_b64] await roster.unpin("grenet") assert await roster.operator_pks() == [], ( "authority must disappear with the pin, without a daemon restart") # ── Join / pairing over MNP ─────────────────────────────────────────────────── async def test_pairing_with_a_valid_code_pins_the_identity(tmp_path, roster): session = _session(tmp_path, roster) sk_ed, pk_ed_b64, pk_x_b64 = _keypair() code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") await session._do_join_request( _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code)) assert _last(session).get("ok") is True pinned = await roster.get_identity("grenet") assert pinned["pk_ed25519"] == pk_ed_b64 assert await roster.operator_pks() == [pk_ed_b64] async def test_pairing_without_a_code_is_refused(tmp_path, roster): """Fails closed: an unknown identity gets nothing until someone authorizes it.""" session = _session(tmp_path, roster) sk_ed, pk_ed_b64, pk_x_b64 = _keypair() await session._do_join_request(_join_msg(session, sk_ed, pk_ed_b64, pk_x_b64)) assert _last(session).get("ok") is False assert _last(session).get("reason") == "code_required" assert await roster.get_identity("grenet") is None async def test_wrong_code_pins_nothing(tmp_path, roster): session = _session(tmp_path, roster) sk_ed, pk_ed_b64, pk_x_b64 = _keypair() await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") await session._do_join_request( _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="ZZZZ-ZZZZ")) assert _last(session).get("reason") == "code_invalid" assert await roster.get_identity("grenet") is None async def test_signature_must_cover_the_presented_keys(tmp_path, roster): """ The heart of it: the X25519 key is only trustworthy because the Ed25519 identity signed it. Swapping in another encryption key after signing must fail. """ session = _session(tmp_path, roster) sk_ed, pk_ed_b64, pk_x_b64 = _keypair() code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") msg = _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code) _, _, attacker_pk_x = _keypair() msg["pk_x25519"] = attacker_pk_x await session._do_join_request(msg) assert _last(session).get("reason") == "signature_invalid" assert await roster.get_identity("grenet") is None async def test_join_cannot_be_replayed_onto_another_connection(tmp_path, roster): session = _session(tmp_path, roster) sk_ed, pk_ed_b64, pk_x_b64 = _keypair() code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") # Signed against a nonce this connection never issued. msg = _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code, nonce=b"\x99" * 32) await session._do_join_request(msg) assert _last(session).get("reason") == "signature_invalid" assert await roster.get_identity("grenet") is None async def test_pinned_identity_presenting_a_new_key_is_refused(tmp_path, roster): """ 11.5.8's rule, applied to people: a changed key is refused outright rather than warned about, and clearing it is a deliberate operator action. """ session = _session(tmp_path, roster) _, old_pk_ed, old_pk_x = _keypair() await roster.pin_identity("grenet", "grenet", old_pk_ed, old_pk_x, "code") sk_ed2, new_pk_ed, new_pk_x = _keypair() await session._do_join_request( _join_msg(session, sk_ed2, new_pk_ed, new_pk_x, code="ANY-CODE")) assert _last(session).get("reason") == "key_changed" assert (await roster.get_identity("grenet"))["pk_ed25519"] == old_pk_ed async def test_attempts_are_bounded(tmp_path, roster): session = _session(tmp_path, roster) sk_ed, pk_ed_b64, pk_x_b64 = _keypair() await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli") for _ in range(6): await session._do_join_request( _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="AAAA-AAAA")) assert any(m.get("detail") == "Too many attempts" for m in session.sent), ( "a connection must not be able to sit there guessing codes") async def test_failures_are_counted_across_connections(tmp_path, roster): """ The adversary who can mint a token for any account is the hub, and it can reconnect at will — so a per-connection budget alone would bound nothing. """ shared_ctx = None for _ in range(6): session = _session(tmp_path, roster) if shared_ctx is None: shared_ctx = session._ctx else: session._ctx = shared_ctx # same node, new connection sk_ed, pk_ed_b64, pk_x_b64 = _keypair() for _ in range(4): await session._do_join_request( _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="AAAA-AAAA")) assert any(m.get("detail") == "Pairing temporarily locked" for m in session.sent), ( "reconnecting must not reset the pairing budget") async def test_group_id_cannot_name_another_group(tmp_path, roster): session = _session(tmp_path, roster) session._group_id = "a" * 32 sk_ed, pk_ed_b64, pk_x_b64 = _keypair() await session._do_join_request( _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, group_id="b" * 32)) assert _last(session).get("reason") == "group_mismatch" # ── H3: the node wraps the group key, and only for people it admitted ───────── GROUP = "g" * 32 async def test_node_wraps_the_gek_for_the_key_the_member_proved(tmp_path, roster): """ The H3 fix. Nobody fetches a public key from the hub: the node encrypts the group key for the X25519 key the joiner signed with their pinned identity, so a hub substituting a key of its own has nothing to substitute into. """ gek = generate_gek() session = _session(tmp_path, roster, user_id="bob", group_id=GROUP, gek=gek) sk_ed, pk_ed_b64, pk_x_b64, sk_x = _keypair_full() code = await roster.create_invite(GROUP, "bob", ROLE_MEMBER, "grenet") await session._do_join_request( _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code, user_id="bob", group_id=GROUP)) reply = _last(session) assert reply["ok"] is True and reply["gek"] is True pk_x_raw = base64.b64decode(pk_x_b64) sk_x_raw = sk_x.private_bytes( encoding=serialization.Encoding.Raw, format=serialization.PrivateFormat.Raw, encryption_algorithm=serialization.NoEncryption(), ) assert unwrap_gek_aes(reply, sk_x_raw, pk_x_raw) == gek async def test_hub_membership_alone_yields_no_key(tmp_path, roster): """ A hub can invent an account, add it to a group and mint it a token. What it cannot do is put it on the node's roster — so the key never leaves. """ gek = generate_gek() session = _session(tmp_path, roster, user_id="eve", group_id=GROUP, gek=gek) sk_ed, pk_ed_b64, pk_x_b64 = _keypair() # Pinned on this node (say, for another group) but never admitted to this one. await roster.pin_identity("eve", "eve", pk_ed_b64, pk_x_b64, "code") await session._do_join_request( _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, user_id="eve", group_id=GROUP)) reply = _last(session) assert reply.get("gek") is False assert reply.get("reason") == "not_authorized_for_group" assert "wrapped_b64" not in reply async def test_open_join_group_admits_without_a_code(tmp_path, roster): """§3.4: where anyone may join, a code protects nothing and is not required.""" gek = generate_gek() session = _session(tmp_path, roster, user_id="newcomer", group_id=GROUP, gek=gek, join_policy="open") sk_ed, pk_ed_b64, pk_x_b64 = _keypair() await session._do_join_request( _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, user_id="newcomer", group_id=GROUP)) reply = _last(session) assert reply["ok"] is True and reply["gek"] is True pinned = await roster.get_identity("newcomer") assert pinned["pinned_via"] == "tofu" async def test_invite_only_group_still_demands_a_code(tmp_path, roster): """Being public (discoverable) is not being open (admitting anyone).""" gek = generate_gek() session = _session(tmp_path, roster, user_id="newcomer", group_id=GROUP, gek=gek, join_policy="invite") sk_ed, pk_ed_b64, pk_x_b64 = _keypair() await session._do_join_request( _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, user_id="newcomer", group_id=GROUP)) assert _last(session).get("reason") == "code_required" assert await roster.get_identity("newcomer") is None async def test_unknown_group_is_invite_only(tmp_path, roster): """ Fail closed: a group whose policy the node cannot read is treated as invite-only, never as open. """ session = _session(tmp_path, roster, user_id="newcomer") session._group_id = "unconfigured-group" assert session._group_join_policy("unconfigured-group") == "invite" assert session._group_join_policy("") == "invite" def test_join_policy_is_carried_from_node_config(): """ The policy reaches the transport from node.toml. If it ever came from the hub instead, a hub could declare any group open and be handed its key. """ daemon_src = (Path(__file__).parent.parent / "src" / "meshbay_node" / "daemon.py").read_text() assert '"join_policy": group_cfg.join_policy' in daemon_src config_src = (Path(__file__).parent.parent / "src" / "meshbay_node" / "config.py").read_text() assert "join_policy" in config_src, "GroupConfig must carry the admission policy" async def test_revoked_member_stops_receiving_the_key(tmp_path, roster): """ Wrapping on demand is what makes revocation work. A stored bundle survived revocation; this does not. (Rotating the GEK is still required — the ex-member has the old one.) """ gek = generate_gek() session = _session(tmp_path, roster, user_id="bob", group_id=GROUP, gek=gek) sk_ed, pk_ed_b64, pk_x_b64 = _keypair() await roster.pin_identity("bob", "bob", pk_ed_b64, pk_x_b64, "code") await roster.set_member(GROUP, "bob", ROLE_MEMBER, "active", "grenet") await session._do_join_request( _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, user_id="bob", group_id=GROUP)) assert _last(session)["gek"] is True await roster.set_status(GROUP, "bob", "revoked") session.sent.clear() await session._do_join_request( _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, user_id="bob", group_id=GROUP)) assert _last(session).get("gek") is False # ── M3: where node authority comes from ─────────────────────────────────────── async def test_admin_signature_verified_against_the_paired_key(tmp_path, roster): session = _session(tmp_path, roster) sk_ed, pk_ed_b64, pk_x_b64 = _keypair() await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code") await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli") transcript = b"meshbay:admin:v1 whatever" assert await session._verify_admin_sig(transcript, sk_ed.sign(transcript)) stranger = Ed25519PrivateKey.generate() assert not await session._verify_admin_sig( transcript, stranger.sign(transcript)) async def test_unpinned_operator_loses_authority_immediately(tmp_path, roster): """No caching: revoking a paired browser must not need a daemon restart.""" session = _session(tmp_path, roster) sk_ed, pk_ed_b64, pk_x_b64 = _keypair() await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code") await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli") transcript = b"meshbay:admin:v1 whatever" assert await session._verify_admin_sig(transcript, sk_ed.sign(transcript)) await roster.unpin("grenet") assert not await session._verify_admin_sig(transcript, sk_ed.sign(transcript)) def test_daemon_does_not_auto_pin_keystore_key(): """ M3: the daemon used to auto-pin its own keystore key as the admin key, while the browser signs with the user's identity key. Different keys, so every privileged operation failed closed with a signature error that looked like a bug elsewhere — and the demo only worked because a deploy script overwrote it. Authority now comes from the roster, or from an explicit node.toml value. """ source = (Path(__file__).parent.parent / "src" / "meshbay_node" / "daemon.py").read_text() assert "Auto-pinning admin key" not in source assert "_resolve_admin_pk" not in source, ( "the auto-pin resolver is back — node authority must be established " "locally by pairing, never inferred from the node's own keystore (M3)") def test_admin_authority_is_never_fetched_from_the_hub(): """ The fix M3 invites: ask the hub which key belongs to the operator. That would hand a malicious hub the node — the same substitution as H3, one level deeper. """ source = (Path(__file__).parent.parent / "src" / "meshbay_node" / "daemon.py").read_text() admin_region = source[source.find("_legacy_admin_pk"):] assert "pubkeys" not in admin_region.split("def ")[1], ( "node authority must never be resolved through a hub lookup")