path: root/packages/meshbay-node/tests/test_webrtc_transport.py

"""
Integration test: WebRTC DataChannel transport for browser clients.

Phase 9 milestone 9.1 — spike: validate aiortc WebRTC DataChannel works
for MNP protocol exchange (handshake, index_sync, file_request, file_chunk).

Uses local loopback (no STUN/ICE needed for localhost).
"""

import asyncio
import base64
import hashlib
import hmac
import os
import struct
import time
from contextlib import asynccontextmanager

import jwt
import msgpack
import pytest
from aiortc import RTCPeerConnection, RTCSessionDescription
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)
from meshbay_common import MNP_VERSION
from meshbay_common.adminop import (
    OP_FILE_DELETE,
    OP_INVITE_CANCEL,
    OP_INVITE_CREATE,
    OP_INVITE_LINK_CREATE,
    admin_transcript,
)
from meshbay_common.crypto import (
    generate_gek,
    pk_to_b64,
    unwrap_gek,
    unwrap_gek_aes,
    wrap_gek,
    wrap_gek_aes,
)
from meshbay_common.groupbox import PURPOSE_ACK, PURPOSE_INDEX, unseal
from meshbay_common.handshake import (
    NONCE_LEN,
    ROLE_CLIENT,
    ROLE_NODE,
    challenge_transcript,
    handshake_transcript,
    make_proof,
    verify_proof,
    webrtc_binding,
)
from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript
from meshbay_common.protocol import MNP
from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
from meshbay_node.bundle_store import BundleStore
from meshbay_node.indexer import DirectoryIndexer
from meshbay_node.roster import Roster
from meshbay_node.transport.webrtc_server import WebRTCTransport

from conftest import one_root

TEST_GROUP = "g"


@pytest.fixture
def sk_node():
    return Ed25519PrivateKey.generate()


@pytest.fixture
def sk_hub():
    return Ed25519PrivateKey.generate()


@pytest.fixture
def sk_user():
    return Ed25519PrivateKey.generate()


@pytest.fixture
def gek():
    return generate_gek()


@pytest.fixture
def shared_dir(tmp_path):
    d = tmp_path / "shared"
    d.mkdir()
    (d / "test.bin").write_bytes(os.urandom(2048))
    (d / "hello.txt").write_bytes(b"hello webrtc " * 50)
    return d


def _hub_pk_pem(sk_hub):
    return sk_hub.public_key().public_bytes(
        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)


def _make_jwt(sk_hub, groups=None, pk_user="test"):
    # group_id is mandatory now (M1), so the default token must be a member
    # of the group the tests connect to. Tests that exercise refusal pass
    # groups=[...] explicitly.
    sk_pem = sk_hub.private_bytes(
        serialization.Encoding.PEM,
        serialization.PrivateFormat.PKCS8,
        serialization.NoEncryption(),
    )
    now = int(time.time())
    return jwt.encode({
        "iss": "test-hub", "sub": "user-001",
        "pk_user": pk_user, "hub_id": "test-hub",
        "jti": "test-jti-webrtc", "iat": now, "exp": now + 3600,
        "groups": groups if groups is not None else [TEST_GROUP],
    }, sk_pem, algorithm="EdDSA")


def _transcript_from(challenge_msg: dict) -> bytes:
    """
    Rebuild the signed transcript from an admin_challenge, the way a real client
    does — from the announced fields, never from opaque bytes on the wire (H5).
    """
    return admin_transcript(
        op=challenge_msg["op"],
        node_pk_b64=challenge_msg["node_pk"],
        group_id=challenge_msg["group_id"],
        subject=challenge_msg["subject"],
        nonce=base64.b64decode(challenge_msg["nonce"]),
        ts=challenge_msg["ts"],
    )


def _pack(obj: dict) -> bytes:
    data = msgpack.packb(obj, use_bin_type=True)
    return struct.pack(">I", len(data)) + data


def _unpack(raw: bytes) -> dict:
    length = struct.unpack(">I", raw[:4])[0]
    return msgpack.unpackb(raw[4:4 + length], raw=False)


def _extract_dtls_fp(sdp: str) -> bytes:
    for line in sdp.splitlines():
        if line.startswith("a=fingerprint:sha-256 "):
            return bytes.fromhex(line.split(" ", 1)[1].replace(":", ""))
    return b""


async def _do_mnp_handshake(channel, received, token, gek, pc, group_id):
    """
    Client half of the unified handshake (11.5.4): client nonce, length-prefixed
    role-bound transcript, and verification of the node's own proof + signature.
    """
    nonce_c = os.urandom(NONCE_LEN)
    channel.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION,
        "token": token, "group_id": group_id,
        "nonce": base64.b64encode(nonce_c).decode(),
    }))
    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    if msg["type"] != MNP.HANDSHAKE_CHALLENGE:
        return msg

    nonce_s = base64.b64decode(msg["nonce"])
    binding = webrtc_binding(
        _extract_dtls_fp(pc.localDescription.sdp),
        _extract_dtls_fp(pc.remoteDescription.sdp),
    )
    # MNP 3.4: every challenge in this suite must carry a signature by the key
    # it announces, over this very connection.
    Ed25519PublicKey.from_public_bytes(
        base64.b64decode(msg["node_pk"])
    ).verify(base64.b64decode(msg["sig"]),
             challenge_transcript(group_id, nonce_c, nonce_s, binding))
    proof = make_proof(gek, ROLE_CLIENT, group_id, nonce_c, nonce_s, binding)
    channel.send(_pack({
        "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION,
        "proof": base64.b64encode(proof).decode(),
    }))
    ack = await asyncio.wait_for(received.get(), timeout=5.0)

    if ack.get("type") == MNP.HANDSHAKE_ACK:
        # The client must authenticate the node too (C3).
        assert verify_proof(
            gek, base64.b64decode(ack["proof"]), ROLE_NODE,
            group_id, nonce_c, nonce_s, binding), "node proof invalid"
        transcript = handshake_transcript(
            ROLE_NODE, group_id, nonce_c, nonce_s, binding)
        Ed25519PublicKey.from_public_bytes(
            base64.b64decode(ack["node_pk"])
        ).verify(base64.b64decode(ack["sig"]), transcript)
    return ack


async def _handshake_with_gek_proof(channel, received, sk_hub, gek, groups=None,
                                    browser_pc=None, group_id=TEST_GROUP):
    """Send handshake, handle GEK challenge, return handshake_ack."""
    token = _make_jwt(sk_hub, groups=groups or [group_id])
    msg = await _do_mnp_handshake(
        channel, received, token, gek, browser_pc, group_id)
    assert msg["type"] == MNP.HANDSHAKE_ACK
    return msg


def _token(sk_hub, jwt_sub, peer_id, group_id, pk_user="test"):
    """A hub-issued user token, as the browser would present it."""
    sk_h_pem = sk_hub.private_bytes(
        serialization.Encoding.PEM,
        serialization.PrivateFormat.PKCS8,
        serialization.NoEncryption(),
    )
    now = int(time.time())
    return jwt.encode({
        "iss": "test-hub", "sub": jwt_sub,
        "pk_user": pk_user, "hub_id": "test-hub",
        "jti": f"jti-{peer_id}", "iat": now, "exp": now + 3600,
        "groups": [group_id], "scope": "user",
    }, sk_h_pem, algorithm="EdDSA")


async def _open_channel(transport, peer_id):
    """
    Signaling only: a live DataChannel with no MNP handshake performed.

    Separate from `_setup_peer` because someone joining a group for the first time
    cannot complete the handshake — they have no GEK to prove — and the join has to
    happen in that window.
    """
    pc = RTCPeerConnection()
    q = asyncio.Queue()
    buf = bytearray()
    ch = pc.createDataChannel("mnp")
    ready = asyncio.Event()

    @ch.on("open")
    def on_open():
        ready.set()

    @ch.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        buf.extend(message)
        while len(buf) >= 4:
            length = struct.unpack(">I", buf[:4])[0]
            if len(buf) < 4 + length:
                break
            msg_bytes = bytes(buf[4:4 + length])
            del buf[:4 + length]
            q.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

    offer = await pc.createOffer()
    await pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(pc.localDescription.sdp, peer_id)
    await pc.setRemoteDescription(RTCSessionDescription(sdp=answer_sdp, type="answer"))
    await asyncio.wait_for(ready.wait(), timeout=5.0)
    return pc, ch, q


def _sealed_chat(session, text: bytes = b"ciphertext") -> dict:
    """
    A chat message in the shape MNP 2.0 requires, on a live session.

    There is no plaintext chat any more, so a test that wants to exercise
    delivery has to send a real envelope. The bytes need not be a real
    ciphertext — the node never opens one — but the envelope's shape and the
    device claim are checked, and the device must be the one this connection
    identified itself as. Identifying it here is what `device_hello` does over
    the wire; doing it directly keeps this test about chat rather than about
    device linking, which `test_device_on_connection.py` covers.
    """
    device = hashlib.sha256(session._registry_key.encode()).digest()
    session._pinned_pk = base64.b64encode(device).decode()
    session._device_confirmed = True
    return {
        "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION,
        "format": 1, "epoch": 1, "device": device, "ct": text,
        "nonce": b"\x02" * 12, "sig": b"\x03" * 64,
    }


def _only_session(transport):
    """The one live peer session on a transport, for tests that made one."""
    sessions = list(transport._sessions.values())
    assert len(sessions) == 1, f"expected one session, got {len(sessions)}"
    return sessions[0]


async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_user=None,
                      group_id=TEST_GROUP):
    """Create a peer connection, perform handshake with GEK proof, return (pc, channel, queue)."""
    pc, ch, q = await _open_channel(transport, peer_id)

    pk_user = "test"
    if sk_user:
        pk_user = base64.b64encode(
            sk_user.public_key().public_bytes(
                serialization.Encoding.Raw, serialization.PublicFormat.Raw)
        ).decode()

    token = _token(sk_hub, jwt_sub, peer_id, group_id, pk_user)

    msg = await _do_mnp_handshake(ch, q, token, gek, pc, group_id)
    assert msg["type"] == MNP.HANDSHAKE_ACK
    return pc, ch, q


@pytest.mark.asyncio
async def test_webrtc_datachannel_handshake(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: browser sends MNP handshake, node responds with handshake_ack."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)

    answer_sdp, ice_candidates = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-001")

    answer = RTCSessionDescription(sdp=answer_sdp, type="answer")
    await browser_pc.setRemoteDescription(answer)

    await asyncio.sleep(0.5)

    msg = await _handshake_with_gek_proof(channel, received, sk_hub, gek,
                                          browser_pc=browser_pc)
    assert msg["v"] == MNP_VERSION
    assert "node_pk" in msg

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: full file transfer — handshake, index, fetch chunk, decrypt."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()

    channel = browser_pc.createDataChannel("mnp")
    channel_ready = asyncio.Event()

    @channel.on("open")
    def on_open():
        channel_ready.set()

    buf = bytearray()

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        buf.extend(message)
        while len(buf) >= 4:
            length = struct.unpack(">I", buf[:4])[0]
            if len(buf) < 4 + length:
                break
            msg_bytes = bytes(buf[4:4 + length])
            del buf[:4 + length]
            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)

    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-002")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    await asyncio.wait_for(channel_ready.wait(), timeout=5.0)

    # 1) Handshake with GEK proof
    ack = await _handshake_with_gek_proof(channel, received, sk_hub, gek,
                                          browser_pc=browser_pc)
    assert ack["type"] == MNP.HANDSHAKE_ACK
    # 1b) The ack's configuration is sealed under the group key (MNP 1.0), and the
    # signed handshake transcript names no ack field — so this envelope is the only
    # thing authenticating `is_node_admin` and the rest.
    config = unseal(gek, PURPOSE_ACK, MNP.HANDSHAKE_ACK, TEST_GROUP, ack)
    assert "is_node_admin" in config
    assert "is_node_admin" not in ack

    # 2) Request index
    channel.send(_pack({"type": MNP.INDEX_SYNC, "v": MNP_VERSION}))
    idx_msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert idx_msg["type"] == MNP.INDEX_SYNC
    assert "entries" not in idx_msg, "the index travels in the clear"
    payload = unseal(gek, PURPOSE_INDEX, MNP.INDEX_SYNC, TEST_GROUP, idx_msg)
    assert len(payload["entries"]) > 0

    # 3) Request file chunk
    entry = next(e for e in indexer.index.entries if e.name == "test.bin")
    channel.send(_pack({
        "type": MNP.FILE_REQUEST,
        "v": MNP_VERSION,
        "file_id": entry.id,
        "chunk_index": 0,
    }))

    chunk_msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert chunk_msg["type"] == MNP.FILE_CHUNK

    # 4) Verify and decrypt (binary fields — no base64, minimal envelope)
    ct = chunk_msg["ct"]
    nonce = chunk_msg["nonce"]
    file_hash = bytes.fromhex(entry.id)

    ckey = chunk_key_aes(gek, file_hash, 0)
    plaintext = decrypt_chunk_aes(ckey, nonce, ct)

    original = (shared_dir / "test.bin").read_bytes()
    assert plaintext == original

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: invalid JWT is rejected with error."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)

    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-003")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    await asyncio.sleep(0.5)

    channel.send(_pack({
        "type": MNP.HANDSHAKE,
        "v": MNP_VERSION,
        "token": "invalid.jwt.token",
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "JWT" in msg["detail"] or "Invalid" in msg["detail"]

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_old_client_is_refused_with_a_code(sk_node, sk_hub, gek, shared_dir):
    """
    A version mismatch must present as a refusal, not as a missing field.

    An 0.x client reaching a 1.0 node would otherwise get a `handshake_ack` with
    no `enabled_apps` and apply its documented fallback — show every app — and an
    `index_sync` with no `entries` it would read as an empty group. Both are
    confident wrong answers. The check runs *before* the token, so it costs
    nothing and reports the real reason (L2).
    """
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id=TEST_GROUP,
                               sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index, stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-old")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))
    await asyncio.sleep(0.5)

    # A perfectly valid token — the refusal must not depend on it, and must not
    # be reported as an authorization problem either.
    channel.send(_pack({
        "type": MNP.HANDSHAKE,
        "v": "0.15",
        "token": _make_jwt(sk_hub, groups=[TEST_GROUP]),
        "group_id": TEST_GROUP,
        "nonce": base64.b64encode(os.urandom(32)).decode(),
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    # The client matches on the code; the text may be reworded.
    assert msg["code"] == "version_too_old"
    assert "0.15" in msg["detail"]

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_request_before_handshake_rejected(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: request without handshake is rejected."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)

    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-004")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    await asyncio.sleep(0.5)

    channel.send(_pack({"type": MNP.INDEX_SYNC, "v": MNP_VERSION}))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "Handshake required" in msg["detail"]

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tmp_path):
    """WebRTC DataChannel: send chat message, then retrieve history."""
    from meshbay_node.chat.store import ChatStore

    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    chat_store = ChatStore(db_path=tmp_path / "chat_test.db")
    await chat_store.open()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["chat_store"] = chat_store

    browser_pc, channel, received = await _setup_peer(
        transport, sk_hub, gek, "peer-chat")

    channel.send(_pack(_sealed_chat(_only_session(transport),
                                    b"hello from browser")))
    chat_ack = await asyncio.wait_for(received.get(), timeout=5.0)
    assert chat_ack["type"] == "ack"

    await asyncio.sleep(0.2)

    channel.send(_pack({
        "type": MNP.CHAT_HISTORY,
        "v": MNP_VERSION,
        "since": 0,
        "limit": 50,
    }))
    hist = await asyncio.wait_for(received.get(), timeout=5.0)
    assert hist["type"] == MNP.CHAT_HISTORY_RESPONSE
    assert len(hist["messages"]) == 1
    # The ciphertext comes back under `ct`, byte for byte — `payload` is the
    # plaintext field and stays empty for a sealed row. Decoding a ciphertext
    # as UTF-8, which the history path used to do, would mangle it.
    assert hist["messages"][0]["ct"] == b"hello from browser"
    assert hist["messages"][0]["payload"] == ""
    assert hist["messages"][0]["format"] == 1
    assert hist["messages"][0]["sender_id"] == "user-001"

    await chat_store.close()
    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_chat_history_no_store(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: chat history without chat_store returns empty list."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )

    browser_pc, channel, received = await _setup_peer(
        transport, sk_hub, gek, "peer-no-store")

    channel.send(_pack({
        "type": MNP.CHAT_HISTORY, "v": MNP_VERSION, "since": 0, "limit": 50,
    }))
    hist = await asyncio.wait_for(received.get(), timeout=5.0)
    assert hist["type"] == MNP.CHAT_HISTORY_RESPONSE
    assert hist["messages"] == []

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path):
    """WebRTC DataChannel: chat message from peer A is broadcast to peer B."""
    from meshbay_node.chat.store import ChatStore

    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    chat_store = ChatStore(db_path=tmp_path / "chat_bc.db")
    await chat_store.open()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["chat_store"] = chat_store

    pc_a, ch_a, q_a = await _setup_peer(transport, sk_hub, gek, "peer-A", "user-A")
    pc_b, ch_b, q_b = await _setup_peer(transport, sk_hub, gek, "peer-B", "user-B")

    session_a = next(s for s in transport._sessions.values()
                     if s._user_id == "user-A")
    ch_a.send(_pack(_sealed_chat(session_a, b"hi from A")))

    ack_a = await asyncio.wait_for(q_a.get(), timeout=5.0)
    assert ack_a["type"] == "ack"

    broadcast = await asyncio.wait_for(q_b.get(), timeout=5.0)
    assert broadcast["type"] == MNP.CHAT_MESSAGE
    # `sender_id` is still the node's, from the authenticated session (NS6).
    # What it now carries beside it is the sending device and a signature over
    # the ciphertext, which is what makes the claim checkable by the receiver
    # rather than taken on the node's word.
    assert broadcast["sender_id"] == "user-A"
    assert broadcast["ct"] == b"hi from A"
    assert broadcast["device"] == base64.b64decode(session_a._pinned_pk)

    await chat_store.close()
    await pc_a.close()
    await pc_b.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_group_membership_enforced(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: JWT without matching group claim is rejected."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-group-test")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    await asyncio.sleep(0.5)

    token = _make_jwt(sk_hub, groups=["other-group"])
    channel.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION,
        "token": token, "group_id": "my-group",
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "Not a member" in msg["detail"]

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: peer removed from _peers dict on session close."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )

    browser_pc, channel, received = await _setup_peer(
        transport, sk_hub, gek, "peer-cleanup")

    # Keyed per connection, not per account (finding F7,
    # docs/MESHBAY_DESIGN.md §13.6), so
    # membership is asserted by the session object rather than by user_id —
    # one account may hold several entries here.
    peers = transport._ctx["_peers"]
    assert [s._user_id for s in peers.values()] == ["user-001"]
    assert transport.active_peers == 1

    await transport.close_peer("peer-cleanup")

    assert transport._ctx["_peers"] == {}
    assert transport.active_peers == 0

    await browser_pc.close()


@pytest.mark.asyncio
async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: wrong GEK proof is rejected — hub admin can't fake membership."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-fake")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    await asyncio.sleep(0.5)

    token = _make_jwt(sk_hub)
    channel.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
        "group_id": TEST_GROUP,
        "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(),
    }))

    challenge = await asyncio.wait_for(received.get(), timeout=5.0)
    assert challenge["type"] == MNP.HANDSHAKE_CHALLENGE

    fake_gek = os.urandom(32)
    nonce = base64.b64decode(challenge["nonce"])
    offer_fp = _extract_dtls_fp(browser_pc.localDescription.sdp)
    answer_fp = _extract_dtls_fp(browser_pc.remoteDescription.sdp)
    bad_proof = hmac.new(fake_gek, nonce + offer_fp + answer_fp, hashlib.sha256).digest()
    channel.send(_pack({
        "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION,
        "proof": base64.b64encode(bad_proof).decode(),
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "GEK proof failed" in msg["detail"]

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_dtls_channel_binding_detects_mitm(sk_node, sk_hub, gek, shared_dir):
    """WebRTC: DTLS channel binding detects fingerprint substitution (simulated MitM)."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-mitm")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    await asyncio.sleep(0.5)

    token = _make_jwt(sk_hub)
    channel.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
        "group_id": TEST_GROUP,
        "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(),
    }))

    challenge = await asyncio.wait_for(received.get(), timeout=5.0)
    assert challenge["type"] == MNP.HANDSHAKE_CHALLENGE

    nonce = base64.b64decode(challenge["nonce"])
    # Correct GEK but fake fingerprints — simulates MitM substituting DTLS certs
    fake_fp = os.urandom(32)
    proof = hmac.new(gek, nonce + fake_fp + fake_fp, hashlib.sha256).digest()
    channel.send(_pack({
        "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION,
        "proof": base64.b64encode(proof).decode(),
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "GEK proof failed" in msg["detail"]

    await browser_pc.close()
    await transport.close_all()



@pytest.mark.asyncio
async def test_the_challenge_signature_is_bound_to_this_connection(
        sk_node, sk_hub, gek, shared_dir):
    """
    MNP 3.4. The node signs its challenge so a client can check `node_pk` before
    a join — which goes out before the ack that used to be the only proof. That
    is only worth anything if the signature cannot be carried elsewhere: under a
    substituted fingerprint (a relay in the middle), another client nonce (a
    recording replayed) or another group, it must not verify.
    """
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()
    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=_hub_pk_pem(sk_hub), gek=gek,
        roots=one_root(shared_dir), index=indexer.index, stun_servers=[],
    )
    pc, ch, q = await _open_channel(transport, "peer-sig")
    try:
        nonce_c = os.urandom(NONCE_LEN)
        ch.send(_pack({
            "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": _make_jwt(sk_hub),
            "group_id": TEST_GROUP, "nonce": base64.b64encode(nonce_c).decode(),
        }))
        challenge = await asyncio.wait_for(q.get(), timeout=5.0)
        assert challenge["type"] == MNP.HANDSHAKE_CHALLENGE
        assert challenge["node_pk"] == pk_to_b64(sk_node.public_key())

        nonce_s = base64.b64decode(challenge["nonce"])
        offer_fp = _extract_dtls_fp(pc.localDescription.sdp)
        answer_fp = _extract_dtls_fp(pc.remoteDescription.sdp)
        pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(challenge["node_pk"]))
        sig = base64.b64decode(challenge["sig"])

        pk.verify(sig, challenge_transcript(
            TEST_GROUP, nonce_c, nonce_s, webrtc_binding(offer_fp, answer_fp)))
        for label, transcript in (
            ("once relayed", challenge_transcript(
                TEST_GROUP, nonce_c, nonce_s, webrtc_binding(offer_fp, os.urandom(32)))),
            ("once replayed", challenge_transcript(
                TEST_GROUP, os.urandom(NONCE_LEN), nonce_s,
                webrtc_binding(offer_fp, answer_fp))),
            ("for another group", challenge_transcript(
                "other-group", nonce_c, nonce_s, webrtc_binding(offer_fp, answer_fp))),
        ):
            with pytest.raises(InvalidSignature):
                pk.verify(sig, transcript)
                pytest.fail(f"the challenge signature verified {label}")
    finally:
        await pc.close()
        await transport.close_all()

async def _paired_operator_roster(tmp_path, sk_admin):
    """
    A roster holding one operator, which is the only thing that authorizes an
    admin operation now. It used to be enough to name a key in node.toml; that
    path is gone, so these tests build the authority the way an operator does —
    by pairing.
    """
    from meshbay_node.roster import Roster

    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    await roster.pin_identity(
        user_id="user-001", username="operator",
        pk_ed25519=base64.b64encode(sk_admin.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw)).decode(),
        pk_x25519="", via="test")
    await roster.set_member(group_id="", user_id="user-001", role="operator",
                            status="active", approved_by="test")
    return roster


@pytest.mark.asyncio
async def test_webrtc_admin_challenge_response(sk_node, sk_hub, gek, shared_dir,
                                              tmp_path):
    """WebRTC DataChannel: admin file delete requires Ed25519 challenge-response."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    sk_admin = Ed25519PrivateKey.generate()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["roster"] = await _paired_operator_roster(tmp_path, sk_admin)
    transport._ctx["has_admin_authority"] = True
    transport._ctx["node_user_id"] = "user-001"

    browser_pc, channel, received = await _setup_peer(
        transport, sk_hub, gek, "peer-admin")

    entry = indexer.index.entries[0]
    channel.send(_pack({
        "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id,
    }))

    challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
    assert challenge_msg["op"] == OP_FILE_DELETE
    assert challenge_msg["subject"] == entry.id

    signature = sk_admin.sign(_transcript_from(challenge_msg))
    channel.send(_pack({
        "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
        "op_id": challenge_msg["op_id"],
        "signature": base64.b64encode(signature).decode(),
    }))

    ack = await asyncio.wait_for(received.get(), timeout=5.0)
    assert ack["type"] == MNP.FILE_DELETE_ACK
    assert ack["file_id"] == entry.id

    assert indexer.index.get_entry(entry.id) is None

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_admin_bad_signature_rejected(sk_node, sk_hub, gek, shared_dir,
                                                  tmp_path):
    """WebRTC DataChannel: wrong Ed25519 signature is rejected — hub can't fake admin."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    sk_admin = Ed25519PrivateKey.generate()
    sk_attacker = Ed25519PrivateKey.generate()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["roster"] = await _paired_operator_roster(tmp_path, sk_admin)
    transport._ctx["has_admin_authority"] = True
    transport._ctx["node_user_id"] = "user-001"

    browser_pc, channel, received = await _setup_peer(
        transport, sk_hub, gek, "peer-attacker")

    entry = indexer.index.entries[0]
    channel.send(_pack({
        "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id,
    }))

    challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE

    bad_sig = sk_attacker.sign(_transcript_from(challenge_msg))
    channel.send(_pack({
        "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
        "op_id": challenge_msg["op_id"],
        "signature": base64.b64encode(bad_sig).decode(),
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "signature" in msg["detail"].lower() or "verification" in msg["detail"].lower()

    assert indexer.index.get_entry(entry.id) is not None

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: stream_request for non-existent file returns error."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )

    browser_pc, channel, received = await _setup_peer(
        transport, sk_hub, gek, "peer-mse")

    channel.send(_pack({
        "type": MNP.STREAM_REQUEST, "v": MNP_VERSION,
        "file_id": "nonexistent-file-id",
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "not found" in msg["detail"].lower()

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_uploader_delete_requires_challenge(sk_node, sk_hub, gek, shared_dir):
    """Uploader must prove Ed25519 key ownership to delete — no uploader shortcut."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    sk_uploader = Ed25519PrivateKey.generate()
    pk_uploader_b64 = base64.b64encode(
        sk_uploader.public_key().public_bytes(
            serialization.Encoding.Raw, serialization.PublicFormat.Raw)
    ).decode()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    # No admin_pk configured — only uploader_pk should authorize deletion

    browser_pc, channel, received = await _setup_peer(
        transport, sk_hub, gek, "peer-uploader-del", sk_user=sk_uploader)

    # Tag an existing entry with the uploader's public key
    entry = indexer.index.entries[0]
    entry.uploader_id = "user-001"
    entry.uploader_pk = pk_uploader_b64

    # Request deletion — should get a challenge (no shortcut)
    channel.send(_pack({
        "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id,
    }))

    challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
    assert challenge_msg["op"] == OP_FILE_DELETE
    assert challenge_msg["subject"] == entry.id

    # Sign with uploader's Ed25519 key
    signature = sk_uploader.sign(_transcript_from(challenge_msg))
    channel.send(_pack({
        "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
        "op_id": challenge_msg["op_id"],
        "signature": base64.b64encode(signature).decode(),
    }))

    ack = await asyncio.wait_for(received.get(), timeout=5.0)
    assert ack["type"] == MNP.FILE_DELETE_ACK
    assert ack["file_id"] == entry.id

    # Verify file was removed from index
    assert indexer.index.get_entry(entry.id) is None

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_uploader_impersonation_blocked(sk_node, sk_hub, gek, shared_dir):
    """Hub-forged JWT with same sub cannot delete — wrong Ed25519 key is rejected."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    # User A uploaded the file
    sk_user_a = Ed25519PrivateKey.generate()
    pk_a_b64 = base64.b64encode(
        sk_user_a.public_key().public_bytes(
            serialization.Encoding.Raw, serialization.PublicFormat.Raw)
    ).decode()

    # User B is the attacker (different Ed25519 key, but hub forges JWT with same sub)
    sk_user_b = Ed25519PrivateKey.generate()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    # No admin_pk — only uploader_pk matters

    # Tag entry with user A's public key
    entry = indexer.index.entries[0]
    entry.uploader_id = "user-001"
    entry.uploader_pk = pk_a_b64

    # Connect as user B (same jwt_sub "user-001" via hub forgery, but B's Ed25519 key)
    browser_pc, channel, received = await _setup_peer(
        transport, sk_hub, gek, "peer-impersonator",
        jwt_sub="user-001", sk_user=sk_user_b)

    # Request deletion — should get a challenge
    channel.send(_pack({
        "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id,
    }))

    challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE

    # Sign with user B's key (wrong key)
    bad_sig = sk_user_b.sign(_transcript_from(challenge_msg))
    channel.send(_pack({
        "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
        "op_id": challenge_msg["op_id"],
        "signature": base64.b64encode(bad_sig).decode(),
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "verification" in msg["detail"].lower() or "signature" in msg["detail"].lower()

    # File must still exist in the index
    assert indexer.index.get_entry(entry.id) is not None

    await browser_pc.close()
    await transport.close_all()


# ── GEK bundle P2P exchange tests ──────────────────────────────────────────


@pytest.fixture
def x25519_keypair():
    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
    sk = X25519PrivateKey.generate()
    sk_raw = sk.private_bytes(
        serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
        serialization.NoEncryption())
    pk_raw = sk.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
    return sk_raw, pk_raw


class _InviteHub:
    """Just enough hub for `ops.create_invite`: a live session, and a member
    registration that records what it was asked to do."""

    class _S:
        user_id = "node-user"

    _session = _S()

    def __init__(self):
        self.added = []

    async def add_group_member(self, group_id, username):
        self.added.append((group_id, username))
        return {"status": "stored"}


@pytest.mark.asyncio
async def test_invite_then_join_delivers_the_gek(sk_node, sk_hub, gek, shared_dir,
                                                 tmp_path, x25519_keypair):
    """
    The whole invite flow over a real DataChannel, end to end.

    The operator asks for a code; the invitee — who has never held the group key
    and therefore cannot complete the GEK proof — redeems it in the pre-proof
    window and the node wraps the key for the X25519 key they just proved they
    hold. At no point is a public key fetched from the hub, which is the point:
    that lookup was H3.
    """
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["roster"] = roster
    transport._ctx["has_admin_authority"] = True
    transport._ctx["groups"] = {
        # A RootSet, like the transport two lines up and like the code under
        # test expects: a group's content became several named roots (draft v6,
        # change 1) and this one line kept passing the bare Path. The handshake
        # died on `'PosixPath' object has no attribute 'describe'` and answered
        # `error` instead of `handshake_ack`, which is a scaffolding that never
        # followed the change, not a defect in the flow being tested.
        TEST_GROUP: {"gek": gek, "roots": one_root(shared_dir),
                     "index": indexer.index},
    }
    # `create_invite` registers the invitee as a hub member *before* writing the
    # invite, and fails the whole operation if it cannot: `/v1/groups/mine`
    # joins `GroupMember`, so someone never registered does not see the group
    # and could never redeem the code. Without a hub here the operation is
    # correctly refused — this test used to have none, and passed only because
    # the registration failure was swallowed and the unredeemable code returned
    # anyway.
    transport._ctx["daemon_state"] = {
        "roster": roster,
        "groups_ctx": transport._ctx["groups"],
        "hub": _InviteHub(),
    }

    # A paired operator, as `meshbay-node operator pair` would have left it.
    sk_admin = Ed25519PrivateKey.generate()
    admin_pk_b64 = pk_to_b64(sk_admin.public_key())
    await roster.pin_identity("user-001", "grenet", admin_pk_b64, "AA==", "code")
    await roster.set_member("", "user-001", ROLE_OPERATOR, "active", "local-cli")

    pc_admin, ch_admin, q_admin = await _setup_peer(
        transport, sk_hub, gek, "peer-admin")

    # 1. The operator asks the node for an invitation code.
    ch_admin.send(_pack({
        "type": MNP.INVITE_CREATE, "v": MNP_VERSION,
        "user_id": "user-002", "group_id": TEST_GROUP, "username": "bob",
    }))
    challenge_msg = await asyncio.wait_for(q_admin.get(), timeout=5.0)
    assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
    assert challenge_msg["op"] == OP_INVITE_CREATE
    assert challenge_msg["subject"] == "user-002"

    ch_admin.send(_pack({
        "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
        "op_id": challenge_msg["op_id"],
        "signature": base64.b64encode(
            sk_admin.sign(_transcript_from(challenge_msg))).decode(),
    }))
    invite = await asyncio.wait_for(q_admin.get(), timeout=5.0)
    assert invite["type"] == MNP.INVITE_RESULT
    code = invite["code"]
    assert code and len(code) == 9        # XXXX-XXXX

    # 2. Bob connects. He cannot prove GEK possession — he has never had it — so
    #    he redeems the code in the pre-proof window instead.
    sk_x_raw, pk_x_raw = x25519_keypair
    sk_bob_ed = Ed25519PrivateKey.generate()
    pc_bob, ch_bob, q_bob = await _open_channel(transport, "peer-bob")

    nonce_c = os.urandom(NONCE_LEN)
    ch_bob.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION,
        "token": _token(sk_hub, "user-002", "peer-bob", TEST_GROUP),
        "group_id": TEST_GROUP,
        "nonce": base64.b64encode(nonce_c).decode(),
    }))
    challenge = await asyncio.wait_for(q_bob.get(), timeout=5.0)
    assert challenge["type"] == MNP.HANDSHAKE_CHALLENGE
    nonce_s = base64.b64decode(challenge["nonce"])

    # Bob signs a transcript naming the node, and he cannot complete the handshake
    # that would prove its key — he has no GEK yet. So he has to be able to learn
    # it from the challenge; taking it from the test's own knowledge of sk_node
    # would hide the fact that a real client cannot.
    assert challenge["node_pk"] == pk_to_b64(sk_node.public_key()), (
        "the challenge must announce the node key to a first-time joiner")
    node_pk_b64 = challenge["node_pk"]

    pk_ed_b64 = pk_to_b64(sk_bob_ed.public_key())
    pk_x_b64 = base64.b64encode(pk_x_raw).decode()
    ts = int(time.time())
    transcript = join_transcript(
        node_pk_b64=node_pk_b64,
        group_id=TEST_GROUP, user_id="user-002",
        pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64,
        nonce_node=nonce_s, ts=ts,
    )
    ch_bob.send(_pack({
        "type": MNP.JOIN_REQUEST, "v": MNP_VERSION,
        "group_id": TEST_GROUP,
        "pk_ed25519": pk_ed_b64, "pk_x25519": pk_x_b64,
        "code": code, "ts": ts,
        "sig": base64.b64encode(sk_bob_ed.sign(transcript)).decode(),
    }))

    result = await asyncio.wait_for(q_bob.get(), timeout=5.0)
    assert result["type"] == MNP.JOIN_RESULT
    assert result["ok"] is True
    assert result["gek"] is True
    assert result["role"] == ROLE_MEMBER

    # 3. The key really is the group key, and only Bob's secret opens it.
    assert unwrap_gek_aes(result, sk_x_raw, pk_x_raw) == gek

    # 4. The code is spent.
    assert await roster.consume_invite(code, "user-002") is None

    await roster.close()
    await pc_admin.close()
    await pc_bob.close()
    await transport.close_all()


async def _bearer_join(transport, sk_hub, user_id, peer_id, code, x25519_keypair,
                       expect_node_pk):
    """A newcomer's first connection with a link code, as the browser makes it:
    the challenge must prove the node key the link named before the code goes."""
    sk_x_raw, pk_x_raw = x25519_keypair
    sk_ed = Ed25519PrivateKey.generate()
    pc, ch, q = await _open_channel(transport, peer_id)
    nonce_c = os.urandom(NONCE_LEN)
    ch.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION,
        "token": _token(sk_hub, user_id, peer_id, TEST_GROUP),
        "group_id": TEST_GROUP, "nonce": base64.b64encode(nonce_c).decode(),
    }))
    challenge = await asyncio.wait_for(q.get(), timeout=5.0)
    nonce_s = base64.b64decode(challenge["nonce"])
    assert challenge["node_pk"] == expect_node_pk
    Ed25519PublicKey.from_public_bytes(base64.b64decode(expect_node_pk)).verify(
        base64.b64decode(challenge["sig"]),
        challenge_transcript(TEST_GROUP, nonce_c, nonce_s, webrtc_binding(
            _extract_dtls_fp(pc.localDescription.sdp),
            _extract_dtls_fp(pc.remoteDescription.sdp))))

    pk_ed_b64 = pk_to_b64(sk_ed.public_key())
    pk_x_b64 = base64.b64encode(pk_x_raw).decode()
    ts = int(time.time())
    ch.send(_pack({
        "type": MNP.JOIN_REQUEST, "v": MNP_VERSION, "group_id": TEST_GROUP,
        "pk_ed25519": pk_ed_b64, "pk_x25519": pk_x_b64, "code": code, "ts": ts,
        "sig": base64.b64encode(sk_ed.sign(join_transcript(
            node_pk_b64=challenge["node_pk"], group_id=TEST_GROUP, user_id=user_id,
            pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64,
            nonce_node=nonce_s, ts=ts))).decode(),
    }))
    result = await asyncio.wait_for(q.get(), timeout=5.0)
    return pc, result, sk_x_raw, pk_x_raw


@pytest.mark.asyncio
async def test_a_link_is_issued_signed_redeemed_once_and_cancellable(
        sk_node, sk_hub, gek, shared_dir, tmp_path, x25519_keypair):
    """
    Invitation links over a real DataChannel (docs/MESHBAY_NODE_PROTOCOL.md §8.6).

    The operator signs `invite_link_create` naming the outcome, `link:<group>`;
    the first account to bring the code is admitted and handed the key; the
    second is refused; a link not yet used can be taken back, by its handle,
    and is then refused too. Nothing is registered on the hub: there is no
    account to register until somebody redeems it.
    """
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=_hub_pk_pem(sk_hub), gek=gek,
        roots=one_root(shared_dir), index=indexer.index, stun_servers=[],
    )
    transport._ctx["roster"] = roster
    transport._ctx["has_admin_authority"] = True
    transport._ctx["groups"] = {
        TEST_GROUP: {"gek": gek, "roots": one_root(shared_dir), "index": indexer.index},
    }
    hub = _InviteHub()
    transport._ctx["daemon_state"] = {
        "roster": roster, "groups_ctx": transport._ctx["groups"], "hub": hub,
    }
    sk_admin = Ed25519PrivateKey.generate()
    await roster.pin_identity("user-001", "grenet", pk_to_b64(sk_admin.public_key()),
                              "AA==", "code")
    await roster.set_member("", "user-001", ROLE_OPERATOR, "active", "local-cli")
    pc_admin, ch_admin, q_admin = await _setup_peer(transport, sk_hub, gek, "peer-admin")
    node_pk = pk_to_b64(sk_node.public_key())
    peers = [pc_admin]

    async def signed(request: dict, op: str, subject: str) -> dict:
        ch_admin.send(_pack({"v": MNP_VERSION, **request}))
        challenge = await asyncio.wait_for(q_admin.get(), timeout=5.0)
        assert challenge["type"] == MNP.ADMIN_CHALLENGE, challenge
        assert (challenge["op"], challenge["subject"]) == (op, subject)
        ch_admin.send(_pack({
            "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION, "op_id": challenge["op_id"],
            "signature": base64.b64encode(
                sk_admin.sign(_transcript_from(challenge))).decode(),
        }))
        return await asyncio.wait_for(q_admin.get(), timeout=5.0)

    try:
        link = await signed({"type": MNP.INVITE_LINK_CREATE, "group_id": TEST_GROUP},
                            OP_INVITE_LINK_CREATE, f"link:{TEST_GROUP}")
        assert link["type"] == MNP.INVITE_LINK_RESULT, link
        assert len(link["code"]) == 9 and len(link["invite_id"]) == 32
        assert hub.added == [], "a link registers nobody on the hub"

        pc, result, sk_x_raw, pk_x_raw = await _bearer_join(
            transport, sk_hub, "user-003", "peer-first", link["code"], x25519_keypair,
            node_pk)
        peers.append(pc)
        assert result["ok"] is True and result["gek"] is True, result
        assert unwrap_gek_aes(result, sk_x_raw, pk_x_raw) == gek

        pc, result, _, _ = await _bearer_join(
            transport, sk_hub, "user-004", "peer-second", link["code"], x25519_keypair,
            node_pk)
        peers.append(pc)
        assert result.get("reason") == "code_invalid"

        spare = await signed({"type": MNP.INVITE_LINK_CREATE, "group_id": TEST_GROUP},
                             OP_INVITE_LINK_CREATE, f"link:{TEST_GROUP}")
        done = await signed({"type": MNP.INVITE_CANCEL, "invite_id": spare["invite_id"]},
                            OP_INVITE_CANCEL, spare["invite_id"])
        assert done.get("detail") == "invite_cancelled", done
        pc, result, _, _ = await _bearer_join(
            transport, sk_hub, "user-005", "peer-late", spare["code"], x25519_keypair,
            node_pk)
        peers.append(pc)
        assert result.get("reason") == "code_invalid"
    finally:
        await roster.close()
        for pc in peers:
            await pc.close()
        await transport.close_all()


@pytest.mark.asyncio
async def test_a_link_needs_the_operator(sk_node, sk_hub, gek, shared_dir, tmp_path):
    """A member who is not the operator is refused before any challenge is issued."""
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=_hub_pk_pem(sk_hub), gek=gek,
        roots=one_root(shared_dir), index=indexer.index, stun_servers=[],
    )
    transport._ctx["roster"] = roster
    transport._ctx["has_admin_authority"] = False
    pc, ch, q = await _setup_peer(transport, sk_hub, gek, "peer-member")
    try:
        ch.send(_pack({"type": MNP.INVITE_LINK_CREATE, "v": MNP_VERSION}))
        reply = await asyncio.wait_for(q.get(), timeout=5.0)
        assert reply["type"] == "error" and "operator" in reply["detail"]
        assert await roster.list_invites() == []
    finally:
        await roster.close()
        await pc.close()
        await transport.close_all()

@pytest.mark.asyncio
async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_dir,
                                                  tmp_path, x25519_keypair):
    """Browser fetches GEK bundle from node during the handshake challenge window."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    sk_x_raw, pk_x_raw = x25519_keypair
    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
    await bundle_store.open()

    # Pre-populate a bundle for user-001 in group "g"
    bundle = wrap_gek(gek, pk_x_raw)
    await bundle_store.store("g", "user-001",
                             bundle["pk_eph_b64"], bundle["nonce_b64"],
                             bundle["wrapped_b64"])

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["bundle_store"] = bundle_store

    # Connect manually: handshake → challenge → gek_bundle_fetch → response
    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    buf = bytearray()
    channel = browser_pc.createDataChannel("mnp")
    ready = asyncio.Event()

    @channel.on("open")
    def on_open():
        ready.set()

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        buf.extend(message)
        while len(buf) >= 4:
            length = struct.unpack(">I", buf[:4])[0]
            if len(buf) < 4 + length:
                break
            msg_bytes = bytes(buf[4:4 + length])
            del buf[:4 + length]
            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-fetch")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))
    await asyncio.wait_for(ready.wait(), timeout=5.0)

    # Step 1: Send handshake with group_id so _pending_group is set
    token = _make_jwt(sk_hub, groups=["g"])
    nonce_c = os.urandom(NONCE_LEN)
    channel.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
        "nonce": base64.b64encode(nonce_c).decode(),
    }))
    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == MNP.HANDSHAKE_CHALLENGE

    # Step 2: Fetch GEK bundle from node (during challenge window)
    channel.send(_pack({"type": MNP.GEK_BUNDLE_FETCH, "v": MNP_VERSION}))
    bundle_resp = await asyncio.wait_for(received.get(), timeout=5.0)
    assert bundle_resp["type"] == MNP.GEK_BUNDLE_RESP
    assert bundle_resp["found"] is True

    # Step 3: Unwrap GEK and compute HMAC proof
    recovered_gek = unwrap_gek(bundle_resp, sk_x_raw, pk_x_raw)
    assert recovered_gek == gek

    nonce_s = base64.b64decode(msg["nonce"])
    binding = webrtc_binding(
        _extract_dtls_fp(browser_pc.localDescription.sdp),
        _extract_dtls_fp(browser_pc.remoteDescription.sdp),
    )
    proof = make_proof(recovered_gek, ROLE_CLIENT, "g", nonce_c, nonce_s, binding)

    # Step 4: Complete handshake
    channel.send(_pack({
        "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION,
        "proof": base64.b64encode(proof).decode(),
    }))
    ack = await asyncio.wait_for(received.get(), timeout=5.0)
    assert ack["type"] == MNP.HANDSHAKE_ACK

    await bundle_store.close()
    await browser_pc.close()
    await transport.close_all()


# ── Keypair bundle P2P tests ─────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir, tmp_path):
    """Keypair bundle stored on node, then fetched during handshake window."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
    await bundle_store.open()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["bundle_store"] = bundle_store

    # Connect and store a keypair bundle
    pc1, ch1, q1 = await _setup_peer(transport, sk_hub, gek, "peer-kp-store")
    ch1.send(_pack({
        "type": MNP.KEYPAIR_BUNDLE_STORE,
        "v": MNP_VERSION,
        "bundle_enc": "encrypted-keypair-data-base64",
    }))
    ack = await asyncio.wait_for(q1.get(), timeout=5.0)
    assert ack["type"] == "ack"
    assert ack["detail"] == "keypair_bundle_stored"

    # Verify in DB
    stored = await bundle_store.fetch_keypair("user-001")
    assert stored["bundle_enc"] == "encrypted-keypair-data-base64"
    assert stored["bundle_enc_recovery"] is None

    await pc1.close()

    # New connection: fetch during handshake window
    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    buf = bytearray()
    channel = browser_pc.createDataChannel("mnp")
    ready = asyncio.Event()

    @channel.on("open")
    def on_open():
        ready.set()

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        buf.extend(message)
        while len(buf) >= 4:
            length = struct.unpack(">I", buf[:4])[0]
            if len(buf) < 4 + length:
                break
            msg_bytes = bytes(buf[4:4 + length])
            del buf[:4 + length]
            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-kp-fetch")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))
    await asyncio.wait_for(ready.wait(), timeout=5.0)

    token = _make_jwt(sk_hub, groups=["g"])
    nonce_c = os.urandom(NONCE_LEN)
    channel.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
        "nonce": base64.b64encode(nonce_c).decode(),
    }))
    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == MNP.HANDSHAKE_CHALLENGE

    # Fetch keypair bundle during challenge window
    channel.send(_pack({"type": MNP.KEYPAIR_BUNDLE_FETCH, "v": MNP_VERSION}))
    kp_resp = await asyncio.wait_for(received.get(), timeout=5.0)
    assert kp_resp["type"] == MNP.KEYPAIR_BUNDLE_RESP
    assert kp_resp["found"] is True
    assert kp_resp["bundle_enc"] == "encrypted-keypair-data-base64"

    await bundle_store.close()
    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_path):
    """Keypair bundle fetch returns found=false when no bundle exists."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
    await bundle_store.open()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["bundle_store"] = bundle_store

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    buf = bytearray()
    channel = browser_pc.createDataChannel("mnp")
    ready = asyncio.Event()

    @channel.on("open")
    def on_open():
        ready.set()

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        buf.extend(message)
        while len(buf) >= 4:
            length = struct.unpack(">I", buf[:4])[0]
            if len(buf) < 4 + length:
                break
            msg_bytes = bytes(buf[4:4 + length])
            del buf[:4 + length]
            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-kp-none")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))
    await asyncio.wait_for(ready.wait(), timeout=5.0)

    token = _make_jwt(sk_hub, groups=["g"])
    nonce_c = os.urandom(NONCE_LEN)
    channel.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
        "nonce": base64.b64encode(nonce_c).decode(),
    }))
    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == MNP.HANDSHAKE_CHALLENGE

    channel.send(_pack({"type": MNP.KEYPAIR_BUNDLE_FETCH, "v": MNP_VERSION}))
    resp = await asyncio.wait_for(received.get(), timeout=5.0)
    assert resp["type"] == MNP.KEYPAIR_BUNDLE_RESP
    assert resp["found"] is False

    await bundle_store.close()
    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_gek_not_auto_activated_on_bundle_store(sk_node, sk_hub, gek, shared_dir,
                                                       tmp_path, x25519_keypair):
    """
    A GEK bundle arriving over MNP must NOT become the node's live key (C5b).

    This test previously asserted the opposite: storing a bundle addressed to the
    node operator auto-activated it, with no signature required. Because the
    operator's X25519 public key is public — the node publishes it in handshake_ack
    — any group member could wrap a key of their own choosing for it and take over
    the group, locking every legitimate member out. GEK activation now happens only
    through the node's local admin UI or CLI.
    """
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    sk_x_raw, pk_x_raw = x25519_keypair
    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
    await bundle_store.open()

    attacker_gek = generate_gek()
    assert attacker_gek != gek

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["bundle_store"] = bundle_store
    transport._ctx["node_user_id"] = "node-operator"
    transport._ctx["sk_x25519_raw"] = sk_x_raw
    transport._ctx["pk_x25519_raw"] = pk_x_raw
    transport._ctx["pk_x25519_b64"] = base64.b64encode(pk_x_raw).decode()
    transport._ctx["admin_pk_ed25519"] = Ed25519PrivateKey.generate().public_key()

    pc_admin, ch_admin, q_admin = await _setup_peer(
        transport, sk_hub, gek, "peer-setup-admin")

    # An ordinary member wraps a key of their choosing for the operator's public
    # key and offers it to the node. The message that used to carry this no longer
    # exists (the node wraps the GEK itself now), so it reaches no handler at all —
    # a stronger outcome than the admin challenge this test used to assert.
    node_bundle = wrap_gek_aes(attacker_gek, pk_x_raw)
    ch_admin.send(_pack({
        "type": "gek_bundle_store",
        "v": MNP_VERSION,
        "user_id": "node-operator",
        "group_id": "g",
        "pk_eph_b64": node_bundle["pk_eph_b64"],
        "nonce_b64": node_bundle["nonce_b64"],
        "wrapped_b64": node_bundle["wrapped_b64"],
    }))

    await asyncio.sleep(0.5)
    assert q_admin.empty(), "the retired bundle message still gets a response"

    assert transport._ctx.get("gek") == gek, "group key was seized over MNP (C5b)"
    assert await bundle_store.fetch("g", "node-operator") is None

    await bundle_store.close()
    await pc_admin.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_no_gek_connection_refused(sk_node, sk_hub, shared_dir):
    """WebRTC DataChannel: connection refused when GEK is not initialized."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=None)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=None,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-no-gek")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    await asyncio.sleep(0.5)

    token = _make_jwt(sk_hub)
    channel.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
        "group_id": TEST_GROUP,
        "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(),
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "not initialized" in msg["detail"].lower()

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_path):
    """GEK bundle fetch returns found=false when no bundle exists."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
    await bundle_store.open()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["bundle_store"] = bundle_store

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    buf = bytearray()
    channel = browser_pc.createDataChannel("mnp")
    ready = asyncio.Event()

    @channel.on("open")
    def on_open():
        ready.set()

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        buf.extend(message)
        while len(buf) >= 4:
            length = struct.unpack(">I", buf[:4])[0]
            if len(buf) < 4 + length:
                break
            msg_bytes = bytes(buf[4:4 + length])
            del buf[:4 + length]
            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-nofound")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))
    await asyncio.wait_for(ready.wait(), timeout=5.0)

    token = _make_jwt(sk_hub, groups=["g"])
    nonce_c = os.urandom(NONCE_LEN)
    channel.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
        "nonce": base64.b64encode(nonce_c).decode(),
    }))
    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == MNP.HANDSHAKE_CHALLENGE

    channel.send(_pack({"type": MNP.GEK_BUNDLE_FETCH, "v": MNP_VERSION}))
    resp = await asyncio.wait_for(received.get(), timeout=5.0)
    assert resp["type"] == MNP.GEK_BUNDLE_RESP
    assert resp["found"] is False

    await bundle_store.close()
    await browser_pc.close()
    await transport.close_all()


# ── Per-account blobs, MNP 3.1 (docs/playlists.md §15.2) ─────────────────────
#
# `test_user_blob_mnp.py` drives these six handlers directly, which proves what
# they decide but nothing about how their payloads travel: it never builds a
# frame and never crosses a channel. A playlist body is the largest `bin` value
# this protocol carries after a file chunk, and the failure it would hit is
# silent — a playlist that does not come back reports no error, it is simply
# absent, which is exactly what was seen on a phone once already.


@asynccontextmanager
async def _user_blob_node(sk_node, sk_hub, gek, shared_dir, tmp_path):
    """
    A node with a bundle store, and peers, torn down whatever happens.

    The `finally` is not tidiness. `BundleStore` runs an aiosqlite thread, and a
    test that fails before closing it leaves that thread alive — the process
    then hangs in `threading._shutdown`, *after* pytest has printed the failure
    and the summary. Found by breaking the node on purpose to check these tests
    catch it: they did, in 0.66s, and then the run never ended. A red suite is a
    result; a stuck one is an outage.
    """
    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
                               sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
    await bundle_store.open()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=_hub_pk_pem(sk_hub), gek=gek,
        roots=one_root(shared_dir), index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["bundle_store"] = bundle_store

    peers: list = []
    try:
        yield transport, peers
    finally:
        for pc in peers:
            await pc.close()
        await bundle_store.close()
        await transport.close_all()


@pytest.mark.asyncio
async def test_user_blobs_round_trip_over_a_live_datachannel(
        sk_node, sk_hub, gek, shared_dir, tmp_path):
    """
    Sealed bytes out, the same bytes back, across two connections.

    The blobs are `os.urandom`, deliberately: sealed output is incompressible
    and uses every byte value, so anything that treats this as text — a UTF-8
    decode, msgpack `str` instead of `bin` — corrupts it. Bytes the node could
    have round-tripped by accident would prove nothing.

    The second connection is the case this feature exists for: a device that was
    not the one that wrote.
    """
    async with _user_blob_node(sk_node, sk_hub, gek, shared_dir, tmp_path) as (
            transport, peers):
        manifest = os.urandom(512)
        # Bodies are padded to a multiple of 4 KB before sealing, so this is the
        # smallest one a real client ever sends.
        body = os.urandom(4096)

        pc1, ch1, q1 = await _setup_peer(transport, sk_hub, gek, "peer-blob-write")
        peers.append(pc1)

        ch1.send(_pack({"type": MNP.USER_BLOB_STORE, "v": MNP_VERSION,
                        "kind": "playlists", "rev": 7, "blob_enc": manifest}))
        ack = await asyncio.wait_for(q1.get(), timeout=5.0)
        assert ack["type"] == "ack"
        assert ack["detail"] == "user_blob_stored"

        ch1.send(_pack({"type": MNP.USER_BLOB_STORE, "v": MNP_VERSION,
                        "kind": "playlist:favorites", "rev": 3, "blob_enc": body}))
        ack = await asyncio.wait_for(q1.get(), timeout=5.0)
        assert ack["detail"] == "user_blob_stored"

        await pc1.close()

        # A second device, same account, which has never seen either blob.
        pc2, ch2, q2 = await _setup_peer(transport, sk_hub, gek, "peer-blob-read")
        peers.append(pc2)

        # What is here, and at what revision. This is the call that lets a fresh
        # device discover kinds it cannot guess — they carry client-made ids —
        # and it must not carry payloads.
        ch2.send(_pack({"type": MNP.USER_BLOB_LIST, "v": MNP_VERSION}))
        listing = await asyncio.wait_for(q2.get(), timeout=5.0)
        assert listing["type"] == MNP.USER_BLOB_LIST_RESP
        assert listing["blobs"] == [{"kind": "playlist:favorites", "rev": 3},
                                    {"kind": "playlists", "rev": 7}]

        # `req_id` is what lets the client match a reply to its request rather
        # than to whatever arrives next; over a channel carrying several
        # requests at once, arrival order is not an answer.
        ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION,
                        "kind": "playlists", "req_id": "r-91"}))
        resp = await asyncio.wait_for(q2.get(), timeout=5.0)
        assert resp["type"] == MNP.USER_BLOB_RESP
        assert resp.get("req_id") == "r-91"
        assert resp["kind"] == "playlists"
        assert resp["rev"] == 7
        assert isinstance(resp["blob_enc"], bytes), "came back as something else than bin"
        assert resp["blob_enc"] == manifest

        ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION,
                        "kind": "playlist:favorites"}))
        resp = await asyncio.wait_for(q2.get(), timeout=5.0)
        assert resp["rev"] == 3
        assert resp["blob_enc"] == body

        # A kind never written is an absence, not an error: the ordinary state
        # of a fresh node.
        ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION,
                        "kind": "playlist:never-written"}))
        resp = await asyncio.wait_for(q2.get(), timeout=5.0)
        assert resp["rev"] is None
        assert resp["blob_enc"] is None

        # And a delete reclaims it, which is how a tombstoned playlist stops
        # costing an account its quota.
        ch2.send(_pack({"type": MNP.USER_BLOB_DELETE, "v": MNP_VERSION,
                        "kind": "playlist:favorites"}))
        ack = await asyncio.wait_for(q2.get(), timeout=5.0)
        assert ack["detail"] == "user_blob_deleted"

        ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION,
                        "kind": "playlist:favorites"}))
        resp = await asyncio.wait_for(q2.get(), timeout=5.0)
        assert resp["blob_enc"] is None


@pytest.mark.asyncio
async def test_a_large_user_blob_round_trips_whole(
        sk_node, sk_hub, gek, shared_dir, tmp_path):
    """
    A quarter-megabyte blob, out and back, byte for byte.

    This is the risk §15.2 named, and measuring it corrected how it was
    described: a 256 KB blob arrives as **one** application message of 262213
    bytes, not as fragments the four-byte length prefix reassembles. SCTP
    fragments it and puts it back together underneath. What this pins is that
    nothing in the node's own framing truncates or re-encodes a `bin` value of
    that size.

    Note which direction is proved. A *browser* cannot send a frame this large
    to this node: aiortc advertises `a=max-message-size:65536`, so Chrome
    refuses anything above it — which is why uploads chunk at 48 KB
    (`transport.js:41`). The store below is aiortc talking to aiortc and says
    nothing about that ceiling. The fetch is the direction that matters here: a
    node answers with a whole body, and a long playlist is the largest one.
    """
    async with _user_blob_node(sk_node, sk_hub, gek, shared_dir, tmp_path) as (
            transport, peers):
        # 256 KB: about nine hundred tracks at the 270 bytes a sealed track
        # measures.
        body = os.urandom(256 * 1024)

        pc, ch, q = await _setup_peer(transport, sk_hub, gek, "peer-blob-big")
        peers.append(pc)

        ch.send(_pack({"type": MNP.USER_BLOB_STORE, "v": MNP_VERSION,
                       "kind": "playlist:long", "rev": 1, "blob_enc": body}))
        ack = await asyncio.wait_for(q.get(), timeout=10.0)
        assert ack["detail"] == "user_blob_stored", ack

        ch.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION,
                       "kind": "playlist:long"}))
        resp = await asyncio.wait_for(q.get(), timeout=10.0)
        assert resp["type"] == MNP.USER_BLOB_RESP
        assert len(resp["blob_enc"]) == len(body), "truncated"
        assert resp["blob_enc"] == body