"""
`gek_rotate` and `member_unpin` over MNP.

Both are destructive and both are new, so the tests are negative assertions:
nobody without the operator's pinned key can reach them, a signature over the
wrong transcript does not count, and the operation cannot be triggered by the
request message alone.

The rule these live under is worth restating, because it is easy to read
draft-v5 §5.1 as forbidding them: **"nothing arriving over MNP can activate a
GEK" is about key material arriving from outside** (C5b — a member handing the
node a key of their choosing). An operator-signed instruction where the node
generates the key with its own CSPRNG is a different shape, and it is the only
thing that finishes a revocation: the ex-member still holds the current key.
"""

import base64
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.adminop import (
    OP_GEK_ROTATE,
    OP_MEMBER_UNPIN,
    admin_transcript,
)
from meshbay_common.crypto import pk_to_b64
from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
from meshbay_common.protocol import MNP
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roster import open_roster
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

from conftest import one_root

GROUP = "g" * 32


@pytest.fixture
async def roster(tmp_path):
    r = await open_roster(tmp_path)
    yield r
    await r.close()


def _keypair():
    sk = Ed25519PrivateKey.generate()
    return sk, pk_to_b64(sk.public_key())


async def _session(tmp_path: Path, roster, *, operator: bool) -> WebRTCPeerSession:
    """A peer session with an operator pinned, or deliberately without one."""
    shared = tmp_path / "shared"
    shared.mkdir(exist_ok=True)
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roots = one_root(shared)

    sk_op, pk_op = _keypair()
    if operator:
        await roster.pin_identity("grenet", "grenet", pk_op, pk_op, "code")
        await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")

    group_ctx = {"gek": b"\x01" * 32, "roots": roots, "index": index,
                 "join_policy": "invite"}
    state = {
        "groups_ctx": {GROUP: group_ctx},
        "roster": roster,
        "indexes": {GROUP: index},
        "bundle_store": _FakeBundleStore(),
        "pk_x25519_raw": b"\x02" * 32,
        "hub": _FakeHub(),
        "node_user_id": "node-user",
    }

    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {
        "roots": roots, "index": index, "sk_node": index.sk_node,
        "roster": roster, "groups": {GROUP: group_ctx},
        "has_admin_authority": operator,
        "daemon_state": state,
    }
    session._group_id = GROUP
    session._user_id = "grenet" if operator else "mallory"
    session._username = session._user_id
    session._pk_user = ""
    session._uploads = {}
    session._admin_ops = {}
    session._remote_ip = ""
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    session.state = state
    session.sk_op = sk_op
    session.spawned = []
    session._spawn = session.spawned.append
    return session


class _FakeBundleStore:
    def __init__(self):
        self.stored = []
        self.deleted_keypairs = []

    async def store(self, *args):
        self.stored.append(args)

    async def delete_keypair(self, user_id):
        self.deleted_keypairs.append(user_id)
        return True


class _FakeHub:
    class _S:
        user_id = "node-user"
    _session = _S()


def _last(session):
    return session.sent[-1] if session.sent else {}


async def _drain(session):
    """Await whatever `_spawn` started. The real session holds its tasks; a
    hand-built one collects them here so the assertion sees the result."""
    for coro in session.spawned:
        await coro
    session.spawned.clear()


async def _sign_and_exec(session, op: str, subject: str, sk, exec_fn):
    challenge = _last(session)
    assert challenge["type"] == "admin_challenge", challenge
    transcript = admin_transcript(
        op=op, node_pk_b64=session._node_pk_b64(), group_id=GROUP,
        subject=subject, nonce=base64.b64decode(challenge["nonce"]),
        ts=challenge["ts"])
    pending = session._admin_ops.get(challenge["op_id"]) or {
        "op": op, "subject": subject}
    await exec_fn(pending, transcript, sk.sign(transcript))


# ── gek_rotate ───────────────────────────────────────────────────────────────

async def test_rotation_needs_an_operator(tmp_path, roster):
    """Without a paired operator there is nobody who could sign, so the node
    fails closed and says why rather than issuing a challenge nobody can meet."""
    session = await _session(tmp_path, roster, operator=False)

    session._do_gek_rotate({})

    assert _last(session)["type"] == "error"
    assert "authorized key" in _last(session)["detail"]
    assert not session._admin_ops


async def test_the_request_alone_rotates_nothing(tmp_path, roster):
    """The message asks; only a signature acts. A node that rotated here would
    let any member lock the group out."""
    session = await _session(tmp_path, roster, operator=True)
    before = session._ctx["groups"][GROUP]["gek"]

    session._do_gek_rotate({})

    assert _last(session)["type"] == "admin_challenge"
    assert session._ctx["groups"][GROUP]["gek"] == before


async def test_a_members_signature_does_not_rotate(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True)
    sk_mallory, pk_mallory = _keypair()
    await roster.pin_identity("mallory", "mallory", pk_mallory, pk_mallory, "code")
    await roster.set_member(GROUP, "mallory", ROLE_MEMBER, "active", "grenet")
    before = session._ctx["groups"][GROUP]["gek"]

    session._do_gek_rotate({})
    await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, sk_mallory,
                         session._admin_exec_gek_rotate)

    assert _last(session)["type"] == "error"
    assert session._ctx["groups"][GROUP]["gek"] == before


async def test_a_signature_over_another_operation_does_not_count(tmp_path, roster):
    """
    H5's rule: the node rebuilds the transcript from the operation it is holding
    and verifies against *that*, so a signature collected for one act cannot be
    presented as another.

    Driven through `_do_admin_response`, deliberately. Handing a transcript
    straight to `_admin_exec_*` would skip the reconstruction that is the
    control, and the test would pass while proving nothing.
    """
    session = await _session(tmp_path, roster, operator=True)
    before = session._ctx["groups"][GROUP]["gek"]

    session._do_gek_rotate({})
    challenge = _last(session)

    # Signed over member_unpin, presented against the pending gek_rotate.
    wrong = admin_transcript(
        op=OP_MEMBER_UNPIN, node_pk_b64=session._node_pk_b64(), group_id=GROUP,
        subject=GROUP, nonce=base64.b64decode(challenge["nonce"]),
        ts=challenge["ts"])
    session._do_admin_response({
        "op_id": challenge["op_id"],
        "signature": base64.b64encode(session.sk_op.sign(wrong)).decode(),
    })
    await _drain(session)

    assert _last(session)["type"] == "error"
    assert session.state["groups_ctx"][GROUP]["gek"] == before


async def test_the_operator_rotates_and_the_node_makes_the_key(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True)
    before = session._ctx["groups"][GROUP]["gek"]

    session._do_gek_rotate({})
    await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, session.sk_op,
                         session._admin_exec_gek_rotate)

    ack = _last(session)
    assert ack["type"] == MNP.GEK_ROTATE_ACK, ack
    after = session.state["groups_ctx"][GROUP]["gek"]
    assert after != before, "the key did not change"
    assert len(after) == 32
    # Produced here, not received: no key material crossed the wire (C5b).
    assert session.state["bundle_store"].stored, (
        "the node's own copy was not stored — the daemon could not reload it")


async def test_rotation_reaches_the_index(tmp_path, roster):
    """The index is encrypted under the GEK. Leaving the old key on it would
    serve members a listing they cannot open."""
    session = await _session(tmp_path, roster, operator=True)

    session._do_gek_rotate({})
    await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, session.sk_op,
                         session._admin_exec_gek_rotate)

    assert session.state["indexes"][GROUP].gek == \
        session.state["groups_ctx"][GROUP]["gek"]


# ── member_unpin ─────────────────────────────────────────────────────────────

async def test_unpinning_needs_an_operator(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=False)
    session._do_member_unpin({"user_id": "bob"})
    assert _last(session)["type"] == "error"


async def test_unpinning_yourself_is_refused(tmp_path, roster):
    """It would end the authority of the connection performing the operation,
    halfway through it."""
    session = await _session(tmp_path, roster, operator=True)
    session._do_member_unpin({"user_id": "grenet"})
    assert _last(session)["detail"] == "Cannot unpin yourself"


async def test_a_members_signature_does_not_unpin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True)
    sk_bob, pk_bob = _keypair()
    await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code")

    session._do_member_unpin({"user_id": "bob"})
    await _sign_and_exec(session, OP_MEMBER_UNPIN, "bob", sk_bob,
                         session._admin_exec_member_unpin)

    assert _last(session)["type"] == "error"
    assert await roster.get_identity("bob") is not None, (
        "a member removed their own pin — only the operator may")


async def test_the_operator_unpins(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True)
    _, pk_bob = _keypair()
    await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code")

    session._do_member_unpin({"user_id": "bob"})
    await _sign_and_exec(session, OP_MEMBER_UNPIN, "bob", session.sk_op,
                         session._admin_exec_member_unpin)

    assert _last(session)["type"] == MNP.MEMBER_UNPIN_ACK
    assert await roster.get_identity("bob") is None
    # The stored keypair bundle goes too — left behind it blocks the re-join
    # the unpin exists to enable.
    assert "bob" in session.state["bundle_store"].deleted_keypairs


async def test_unpinning_someone_unknown_says_so(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True)
    session._do_member_unpin({"user_id": "nobody"})
    await _sign_and_exec(session, OP_MEMBER_UNPIN, "nobody", session.sk_op,
                         session._admin_exec_member_unpin)
    assert _last(session)["type"] == "error"
    assert "No such pinned identity" in _last(session)["detail"]