"""
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

import jwt
import msgpack
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey, Ed25519PublicKey,
)
from aiortc import RTCPeerConnection, RTCSessionDescription

from meshbay_common import MNP_VERSION
from meshbay_common.crypto import (
    generate_gek,
    pk_to_b64,
    wrap_gek,
    wrap_gek_aes,
    unwrap_gek,
    unwrap_gek_aes,
)
from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
from meshbay_common.protocol import MNP
TEST_GROUP = "g"

from meshbay_common.handshake import (
    NONCE_LEN, ROLE_CLIENT, ROLE_NODE, handshake_transcript,
    make_proof, verify_proof, webrtc_binding,
)
from meshbay_common.adminop import (
    OP_FILE_DELETE,
    OP_INVITE_CREATE,
    admin_transcript,
)
from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript
from meshbay_node.bundle_store import BundleStore
from meshbay_node.roster import Roster
from meshbay_node.indexer import DirectoryIndexer
from meshbay_node.transport.webrtc_server import WebRTCTransport


@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),
    )
    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


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(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,
        shared_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(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,
        shared_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

    # 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" in idx_msg
    assert len(idx_msg["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(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,
        shared_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_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(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,
        shared_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(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,
        shared_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({
        "type": MNP.CHAT_MESSAGE,
        "v": MNP_VERSION,
        "payload": "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
    assert hist["messages"][0]["payload"] == "hello from browser"
    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(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,
        shared_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(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,
        shared_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")

    ch_a.send(_pack({
        "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "payload": "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
    assert broadcast["sender_id"] == "user-A"
    assert broadcast["payload"] == "hi from A"

    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(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,
        shared_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(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,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )

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

    assert "user-001" in transport._ctx["_peers"]
    assert transport.active_peers == 1

    await transport.close_peer("peer-cleanup")

    assert "user-001" not in transport._ctx["_peers"]
    assert transport.active_peers == 0

    await browser_pc.close()


@pytest.mark.asyncio
async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: stream_segment for non-existent file returns error."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(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,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )

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

    channel.send(_pack({
        "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION,
        "file_id": "nonexistent-file-id",
        "segment_index": 0, "segment_duration": 4,
    }))

    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_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(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,
        shared_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(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,
        shared_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_webrtc_admin_challenge_response(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: admin file delete requires Ed25519 challenge-response."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(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,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["admin_pk_ed25519"] = sk_admin.public_key()
    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):
    """WebRTC DataChannel: wrong Ed25519 signature is rejected — hub can't fake admin."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(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,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["admin_pk_ed25519"] = sk_admin.public_key()
    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(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,
        shared_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(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,
        shared_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(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,
        shared_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


@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(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,
        shared_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, "shared_root": shared_dir, "index": indexer.index},
    }

    # 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"])

    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=pk_to_b64(sk_node.public_key()),
        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()


@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(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,
        shared_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(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,
        shared_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 == "encrypted-keypair-data-base64"

    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(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,
        shared_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(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,
        shared_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(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,
        shared_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(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,
        shared_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()