"""
Rotating the group key and forgetting a pinned identity — `ops.set_gek` and
`ops.unpin_member`, which the Node page and the CLI reach over loopback — and
the H5 rule every signed MNP operation lives under.

`gek_rotate` and `member_unpin` were MNP messages until 6.0. No client sent
them, so they went; what they did is still tested here, at the door that is
used. **"Nothing arriving over MNP can activate a GEK" is about key material
arriving from outside** (C5b): the node generates the new key with its own
CSPRNG, which is the only thing that finishes a revocation — the ex-member
still holds the current key.
"""

import base64
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.adminop import OP_CHAT_EPOCH, admin_transcript
from meshbay_common.crypto import pk_to_b64
from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
from meshbay_node import ops
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roster import open_roster
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

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


# ── H5: a signature is for one operation ─────────────────────────────────────

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)
    _, pk_bob = _keypair()
    await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code")
    await roster.set_member(GROUP, "bob", ROLE_MEMBER, "active", "code")

    session._do_member_revoke({"user_id": "bob"})
    challenge = _last(session)
    assert challenge["type"] == "admin_challenge", challenge

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

    assert _last(session)["type"] == "error"
    assert (await roster.get_member(GROUP, "bob"))["status"] == "active"


# ── Rotating the group key ───────────────────────────────────────────────────

async def test_rotating_makes_a_new_key_on_the_node(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True)
    before = session.state["groups_ctx"][GROUP]["gek"]

    result = await ops.set_gek(session.state, GROUP, rotate=True)

    assert result["rotated"] is True
    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)

    await ops.set_gek(session.state, GROUP, rotate=True)

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


# ── Forgetting a pinned identity ─────────────────────────────────────────────

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

    await ops.unpin_member(session.state, "bob")

    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)
    with pytest.raises(ops.OpError, match="No such pinned identity"):
        await ops.unpin_member(session.state, "nobody")


# ── What MNP no longer carries ───────────────────────────────────────────────

@pytest.mark.parametrize("op", ["gek_rotate", "member_unpin", "transfer_limits",
                                "group_detach"])
def test_operations_no_client_sent_are_not_signed_ops_any_more(op):
    """Gone with MNP 6.0: a door nobody uses is an untested way in. The Node
    page and the CLI do the same work over loopback."""
    from meshbay_common import adminop
    from meshbay_node.transport.webrtc.admin import _ADMIN_EXECUTORS
    from meshbay_node.transport.webrtc.dispatch import _HANDLERS

    assert op not in _HANDLERS
    assert op not in _ADMIN_EXECUTORS
    assert op not in vars(adminop).values()