"""
Tier 2: a member verifies another member's device for themselves.

`docs/MESHBAY_DESIGN.md` §3.3, which records why it could not ship with the
encryption: **the evidence was not being kept.** `_do_device_add` verified the
countersignature and stored only
`added_by_pk` — *which* key approved, never the proof — and the transcript binds
`nonce_node`, the approving connection's handshake nonce, so even a stored
signature was unverifiable by anyone who was not on that connection.

So the node half is two things: keep `(sig, nonce, ts)` beside the pin, and
relay them to any member of the group who asks. The node deliberately decides
nothing here — it hands over evidence, and the client walks the chain. A node
that lies is caught by a client that has seen the account before, which is the
property, and it is why trust is not something the node is asked to assert.

What this does **not** claim, per the convention: nothing is gained at first
sight. A member who has never seen Alice has nothing to compare against.
"""

import base64
import time

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from conftest import one_root
from meshbay_common.crypto import generate_gek, pk_to_b64
from meshbay_common.device import device_add_transcript
from meshbay_common.groupbox import PURPOSE_ROSTER, unseal
from meshbay_common.join import ROLE_MEMBER
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

GROUP = "g" * 32
NONCE = b"\x11" * 32


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


def _keys():
    sk_ed = Ed25519PrivateKey.generate()
    sk_x = Ed25519PrivateKey.generate()          # stand-in; only its b64 is used
    return sk_ed, pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key())


def _session(tmp_path, roster, gek, user_id="alice"):
    shared = tmp_path / "shared"
    shared.mkdir(exist_ok=True)
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {
        "roster": roster, "sk_node": index.sk_node,
        "groups": {GROUP: {"gek": gek, "index": index,
                           "roots": one_root(shared)}},
    }
    session._group_id = GROUP
    session._user_id = user_id
    session._username = user_id
    session._nonce_node = NONCE
    session._pinned_pk = ""
    session._device_confirmed = False
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    return session


async def _add_device(session, roster, approver_sk, approver_pk, new_pk, new_px,
                      user_id="alice"):
    """Run the real device-add path, so the evidence is stored the real way."""
    ts = int(time.time())
    transcript = device_add_transcript(
        node_pk_b64=session._node_pk_b64(), user_id=user_id,
        pk_ed25519_b64=new_pk, pk_x25519_b64=new_px, nonce_node=NONCE, ts=ts)
    session._user_id = user_id
    await session._do_device_add({
        "pk_ed25519": new_pk, "pk_x25519": new_px, "ts": ts,
        "sig": base64.b64encode(approver_sk.sign(transcript)).decode(),
    })
    return ts


# ── the evidence is kept ─────────────────────────────────────────────────────

async def test_the_countersignature_is_stored_not_discarded(tmp_path, roster):
    """
    The finding that blocked Tier 2. Before this, `add_sig` did not exist and
    `added_by_pk` was all that survived — which proves nothing to a third party.
    """
    sk_a, pk_a, px_a = _keys()
    _sk_b, pk_b, px_b = _keys()
    await roster.pin_identity("alice", "alice", pk_a, px_a, via="code")
    await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER,
                            status="active", approved_by="op")

    session = _session(tmp_path, roster, generate_gek())
    ts = await _add_device(session, roster, sk_a, pk_a, pk_b, px_b)

    devices = {d["pk_ed25519"]: d for d in await roster.group_devices(GROUP)}
    added = devices[pk_b]
    assert added["added_by_pk"] == pk_a
    assert added["add_sig"], "the countersignature was thrown away again"
    assert added["add_ts"] == ts
    assert base64.b64decode(added["add_nonce"]) == NONCE, (
        "without the nonce the stored signature is unverifiable — the "
        "transcript binds it")


async def test_the_stored_evidence_actually_verifies(tmp_path, roster):
    """
    The point of storing it. A third party rebuilds the transcript from the
    roster alone and checks the signature — no access to the connection that
    approved it, which is the whole difficulty.
    """
    sk_a, pk_a, px_a = _keys()
    _sk_b, pk_b, px_b = _keys()
    await roster.pin_identity("alice", "alice", pk_a, px_a, via="code")
    await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER,
                            status="active", approved_by="op")
    session = _session(tmp_path, roster, generate_gek())
    await _add_device(session, roster, sk_a, pk_a, pk_b, px_b)

    devices = {d["pk_ed25519"]: d for d in await roster.group_devices(GROUP)}
    d = devices[pk_b]
    transcript = device_add_transcript(
        node_pk_b64=session._node_pk_b64(), user_id="alice",
        pk_ed25519_b64=d["pk_ed25519"], pk_x25519_b64=d["pk_x25519"],
        nonce_node=base64.b64decode(d["add_nonce"]), ts=d["add_ts"])
    from cryptography.hazmat.primitives.asymmetric.ed25519 import (
        Ed25519PublicKey,
    )
    pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(d["added_by_pk"]))
    pk.verify(base64.b64decode(d["add_sig"]), transcript)   # raises if wrong


async def test_a_first_device_has_no_evidence_and_says_so(tmp_path, roster):
    """
    An operator code admitted it; there is no countersignature and there cannot
    be. It must read as trust-on-first-use rather than as verified — a client
    that treated an absent signature as a valid one would verify anything.
    """
    _sk_a, pk_a, px_a = _keys()
    await roster.pin_identity("alice", "alice", pk_a, px_a, via="code")
    await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER,
                            status="active", approved_by="op")

    (d,) = await roster.group_devices(GROUP)
    assert d["added_by_pk"] == "" and d["add_sig"] == ""


# ── the relay ────────────────────────────────────────────────────────────────

async def test_a_member_is_served_the_roster_sealed(tmp_path, roster):
    """
    Any member, not only the operator — that is the point. Sealed under a
    GEK-derived subkey for the same reason the index is.
    """
    _sk_a, pk_a, px_a = _keys()
    await roster.pin_identity("alice", "alice", pk_a, px_a, via="code")
    await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER,
                            status="active", approved_by="op")
    await roster.set_member(group_id=GROUP, user_id="bob", role=ROLE_MEMBER,
                            status="active", approved_by="op")
    _sk_b, pk_b, px_b = _keys()
    await roster.pin_identity("bob", "bob", pk_b, px_b, via="code")

    gek = generate_gek()
    session = _session(tmp_path, roster, gek, user_id="bob")
    await session._do_group_roster_req({})

    resp = session.sent[-1]
    assert resp["type"] == MNP.GROUP_ROSTER_RESP
    assert "devices" not in resp, "the roster must not travel in clear"
    payload = unseal(gek, PURPOSE_ROSTER, MNP.GROUP_ROSTER_RESP, GROUP, resp)
    assert {d["user_id"] for d in payload["devices"]} == {"alice", "bob"}
    assert payload["node_pk"], "the transcript needs the node key to rebuild"


async def test_a_revoked_device_is_not_relayed(tmp_path, roster):
    """A retired laptop must stop being offered as one of the account's keys."""
    sk_a, pk_a, px_a = _keys()
    _sk_b, pk_b, px_b = _keys()
    await roster.pin_identity("alice", "alice", pk_a, px_a, via="code")
    await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER,
                            status="active", approved_by="op")
    session = _session(tmp_path, roster, generate_gek())
    await _add_device(session, roster, sk_a, pk_a, pk_b, px_b)
    await roster.revoke_device("alice", pk_b)

    keys = {d["pk_ed25519"] for d in await roster.group_devices(GROUP)}
    assert keys == {pk_a}


async def test_another_groups_members_are_not_disclosed(tmp_path, roster):
    """
    Scoped to this group. A person in two groups on one node is not revealed to
    the second by being in the first — the roster is member-visible, so its
    scope *is* the privacy boundary.
    """
    _sk_a, pk_a, px_a = _keys()
    _sk_c, pk_c, px_c = _keys()
    await roster.pin_identity("alice", "alice", pk_a, px_a, via="code")
    await roster.pin_identity("carol", "carol", pk_c, px_c, via="code")
    await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER,
                            status="active", approved_by="op")
    await roster.set_member(group_id="h" * 32, user_id="carol",
                            role=ROLE_MEMBER, status="active", approved_by="op")

    users = {d["user_id"] for d in await roster.group_devices(GROUP)}
    assert users == {"alice"}


async def test_a_substituted_key_carries_no_evidence(tmp_path, roster):
    """
    The attack Tier 2 exists to detect, from the node's side of it.

    A node that invents a device for an account can put it in the roster — it
    writes the roster. What it cannot do is produce a countersignature from a
    key it does not hold, so the fabricated device arrives with `add_sig` empty
    and no chain reaches it. The client is what refuses to walk to it; this
    asserts the node cannot manufacture the evidence.
    """
    _sk_a, pk_a, px_a = _keys()
    _sk_evil, pk_evil, px_evil = _keys()
    await roster.pin_identity("alice", "alice", pk_a, px_a, via="code")
    await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER,
                            status="active", approved_by="op")
    # The node simply writes a second device for Alice, as a malicious one would.
    await roster.pin_identity("alice", "alice", pk_evil, px_evil, via="device")

    devices = {d["pk_ed25519"]: d for d in await roster.group_devices(GROUP)}
    assert devices[pk_evil]["add_sig"] == "", (
        "a fabricated device cannot come with a countersignature — if this ever "
        "holds evidence, the node has been handed a way to mint trust")


# ── the operator is a member of every group this node hosts ──────────────────

async def test_the_operator_is_in_the_roster_of_a_group_they_host(tmp_path,
                                                                  roster):
    """
    Found on two live machines, and it is the shape this file exists to stop.

    An operator's authority is node-wide and is stored with an **empty**
    group_id (`is_authorized` says so, and has always said so). `group_devices`
    wrote that rule out a second time as `WHERE m.group_id = ?`, which excludes
    them — so the person running the node was absent from the roster relayed to
    everyone else, their device key could be vouched for by nobody, and every
    single message they sent arrived under "this account is using a key you
    have not seen before".

    A notice that fires on the most ordinary event there is — the operator
    talking in their own group — is worse than no notice, because it is the one
    people learn to dismiss. Both queries now share `_MEMBER_OF_GROUP`.
    """
    _sk_op, pk_op, px_op = _keys()
    _sk_m, pk_m, px_m = _keys()
    # The operator pairs node-wide: group_id is empty, exactly as
    # `pairOperator` sends it and `_pin_and_admit` records it.
    await roster.pin_identity("toto", "toto", pk_op, px_op, via="code")
    await roster.set_member(group_id="", user_id="toto", role="operator",
                            status="active", approved_by="self")
    # An invited member of one group this node hosts.
    await roster.pin_identity("cbesson", "cbesson", pk_m, px_m, via="invite")
    await roster.set_member(group_id=GROUP, user_id="cbesson",
                            role=ROLE_MEMBER, status="active",
                            approved_by="toto")

    users = {d["user_id"] for d in await roster.group_devices(GROUP)}
    assert users == {"toto", "cbesson"}, (
        "the operator must appear in the roster of a group they host — "
        "otherwise every message they send reads as an unknown key")

    # And the two rules genuinely agree, rather than happening to agree here.
    assert await roster.is_authorized(GROUP, "toto") is True


async def test_an_operator_who_is_also_a_member_appears_once(tmp_path, roster):
    """
    Both halves of the clause match such a person. Listed twice, a client would
    see the same key arrive as two devices — harmless today, and exactly the
    kind of thing that grows teeth later.
    """
    _sk_op, pk_op, px_op = _keys()
    await roster.pin_identity("toto", "toto", pk_op, px_op, via="code")
    await roster.set_member(group_id="", user_id="toto", role="operator",
                            status="active", approved_by="self")
    await roster.set_member(group_id=GROUP, user_id="toto", role=ROLE_MEMBER,
                            status="active", approved_by="self")

    devices = await roster.group_devices(GROUP)
    assert [d["pk_ed25519"] for d in devices] == [pk_op]


async def test_an_operator_of_another_node_is_not_invented(tmp_path, roster):
    """
    The clause admits an operator, not anyone with an empty group_id. A
    revoked or suspended one must not come back through it.
    """
    _sk_op, pk_op, px_op = _keys()
    await roster.pin_identity("toto", "toto", pk_op, px_op, via="code")
    await roster.set_member(group_id="", user_id="toto", role="operator",
                            status="revoked", approved_by="self")

    assert await roster.group_devices(GROUP) == []
    assert await roster.is_authorized(GROUP, "toto") is False