"""
Which device is on this connection, and what depends on knowing.

The MNP handshake authenticates a *group membership* (the GEK-HMAC) and an
*account* (the hub's token). It has never authenticated a device. While one
person meant one key on a node those were the same statement; device linking
(2026-08-18) ended that, and two things were left resolving "the account's
oldest live device" and calling it the answer:

  * `_load_pinned_pk`, whose result is recorded as `entry.uploader_pk` on every
    upload — so a phone's uploads were attributed to a laptop;
  * `_admin_exec_file_delete`, which authorized deletion against **that exact
    key** — so a person could not delete their own file from their other device,
    and the only symptom was "Signature verification failed" on their own upload
    (`docs/MESHBAY_DESIGN.md` §3.3).

`device_hello` closes the first: additive, signed, refused unless the key is a
live device *of this account in the node's own roster*. The second is closed by
authorizing against the account rather than the key.
"""

import base64
import time

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import pk_to_b64
from meshbay_common.device import device_hello_transcript
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
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, 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 = {
        "roots": one_root(shared), "index": index, "sk_node": index.sk_node,
        "roster": roster,
        "groups": {GROUP: {"gek": b"\x01" * 32, "index": index,
                           "roots": one_root(shared)}},
    }
    session._group_id = GROUP
    session._user_id = user_id
    session._username = user_id
    session._pinned_pk = ""
    session._device_confirmed = False
    session._nonce_node = NONCE
    session._remote_ip = ""
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    return session


async def _hello(session, sk, pk_ed, *, ts=None, user_id=None):
    ts = int(time.time()) if ts is None else ts
    transcript = device_hello_transcript(
        node_pk_b64=session._node_pk_b64(), group_id=session._group_id,
        user_id=user_id or session._user_id, pk_ed25519_b64=pk_ed,
        nonce_node=NONCE, ts=ts)
    await session._do_device_hello({
        "pk_ed25519": pk_ed, "ts": ts,
        "sig": base64.b64encode(sk.sign(transcript)).decode(),
    })


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


# ── device_hello ─────────────────────────────────────────────────────────────

async def test_a_pinned_device_identifies_itself(tmp_path, roster):
    sk_a, pk_ed_a, pk_x_a = _keys()
    sk_b, pk_ed_b, pk_x_b = _keys()
    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
    await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device")

    session = _session(tmp_path, roster)
    await _hello(session, sk_b, pk_ed_b)

    assert _last(session)["type"] == MNP.DEVICE_HELLO_ACK
    assert session._pinned_pk == pk_ed_b, (
        "the connection must be the device that signed, not the account's "
        "oldest key")
    assert session._device_confirmed is True


async def test_a_key_this_node_never_pinned_is_refused(tmp_path, roster):
    sk_a, pk_ed_a, pk_x_a = _keys()
    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
    sk_x, pk_ed_x, _ = _keys()

    session = _session(tmp_path, roster)
    await _hello(session, sk_x, pk_ed_x)

    assert _last(session)["type"] == "error"
    assert session._device_confirmed is False


async def test_a_revoked_device_cannot_identify_itself(tmp_path, roster):
    sk_a, pk_ed_a, pk_x_a = _keys()
    sk_b, pk_ed_b, pk_x_b = _keys()
    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
    await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device")
    await roster.revoke_device("alice", pk_ed_b)

    session = _session(tmp_path, roster)
    await _hello(session, sk_b, pk_ed_b)

    assert _last(session)["type"] == "error", (
        "a revoked key must stay refused — that is why revocation marks the "
        "row instead of deleting it")


async def test_another_accounts_device_cannot_identify_here(tmp_path, roster):
    """The roster lookup is scoped to *this* account, never to the key alone."""
    sk_a, pk_ed_a, pk_x_a = _keys()
    sk_m, pk_ed_m, pk_x_m = _keys()
    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
    await roster.pin_identity("mallory", "mallory", pk_ed_m, pk_x_m, via="code")

    session = _session(tmp_path, roster, user_id="alice")
    await _hello(session, sk_m, pk_ed_m)

    assert _last(session)["type"] == "error"


async def test_a_signature_for_another_connection_does_not_transfer(
        tmp_path, roster):
    """`nonce_node` binds the statement to one connection (L4)."""
    sk_a, pk_ed_a, pk_x_a = _keys()
    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")

    session = _session(tmp_path, roster)
    ts = int(time.time())
    transcript = device_hello_transcript(
        node_pk_b64=session._node_pk_b64(), group_id=GROUP,
        user_id="alice", pk_ed25519_b64=pk_ed_a,
        nonce_node=b"\x99" * 32, ts=ts)          # another connection's nonce
    await session._do_device_hello({
        "pk_ed25519": pk_ed_a, "ts": ts,
        "sig": base64.b64encode(sk_a.sign(transcript)).decode(),
    })

    assert _last(session)["type"] == "error"


async def test_a_stale_hello_is_refused(tmp_path, roster):
    sk_a, pk_ed_a, pk_x_a = _keys()
    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
    session = _session(tmp_path, roster)
    await _hello(session, sk_a, pk_ed_a, ts=int(time.time()) - 3600)
    assert _last(session)["type"] == "error"


async def test_a_connection_cannot_become_a_second_device(tmp_path, roster):
    """
    Both keys are legitimately this account's, and it is still refused: one
    connection's uploads must be attributable to one device.
    """
    sk_a, pk_ed_a, pk_x_a = _keys()
    sk_b, pk_ed_b, pk_x_b = _keys()
    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
    await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device")

    session = _session(tmp_path, roster)
    await _hello(session, sk_a, pk_ed_a)
    assert _last(session)["type"] == MNP.DEVICE_HELLO_ACK
    await _hello(session, sk_b, pk_ed_b)
    assert _last(session)["type"] == "error"
    assert session._pinned_pk == pk_ed_a


async def test_the_late_roster_load_does_not_undo_a_confirmed_device(
        tmp_path, roster):
    """
    `_load_pinned_pk` is spawned at handshake and can finish *after* a fast
    client has identified itself. It must not put the account's oldest key back
    — a race that would have been intermittent and attributed to nothing.
    """
    sk_a, pk_ed_a, pk_x_a = _keys()
    sk_b, pk_ed_b, pk_x_b = _keys()
    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
    await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device")

    session = _session(tmp_path, roster)
    await _hello(session, sk_b, pk_ed_b)
    await session._load_pinned_pk()               # arrives late

    assert session._pinned_pk == pk_ed_b


# ── deletion is authorized by account, not by the exact device ───────────────

class _Entry:
    def __init__(self, uploader_id="", uploader_pk=""):
        self.uploader_id = uploader_id
        self.uploader_pk = uploader_pk


async def test_a_second_device_can_delete_the_first_devices_upload(
        tmp_path, roster):
    """
    docs/MESHBAY_DESIGN.md §3.3. Alice uploads from her phone and deletes
    from her desktop. Before the fix this failed with "Signature verification
    failed" on her own file.
    """
    sk_phone, pk_phone, pk_x_phone = _keys()
    sk_desk, pk_desk, pk_x_desk = _keys()
    await roster.pin_identity("alice", "alice", pk_phone, pk_x_phone, via="code")
    await roster.pin_identity("alice", "alice", pk_desk, pk_x_desk, via="device")

    session = _session(tmp_path, roster)
    entry = _Entry(uploader_id="alice", uploader_pk=pk_phone)
    transcript = b"delete-this-file"

    assert await session._verify_uploader_sig(
        entry, transcript, sk_desk.sign(transcript)) is True


async def test_a_stranger_still_cannot_delete_someone_elses_upload(
        tmp_path, roster):
    sk_alice, pk_alice, pk_x_alice = _keys()
    sk_mallory, pk_mallory, pk_x_mallory = _keys()
    await roster.pin_identity("alice", "alice", pk_alice, pk_x_alice, via="code")
    await roster.pin_identity("mallory", "mallory", pk_mallory, pk_x_mallory,
                              via="code")

    session = _session(tmp_path, roster)
    entry = _Entry(uploader_id="alice", uploader_pk=pk_alice)
    transcript = b"delete-this-file"

    assert await session._verify_uploader_sig(
        entry, transcript, sk_mallory.sign(transcript)) is False


async def test_a_revoked_device_can_no_longer_delete(tmp_path, roster):
    """Ownership survives revocation; the revoked *key* stops being able to act."""
    sk_old, pk_old, pk_x_old = _keys()
    sk_new, pk_new, pk_x_new = _keys()
    await roster.pin_identity("alice", "alice", pk_old, pk_x_old, via="code")
    await roster.pin_identity("alice", "alice", pk_new, pk_x_new, via="device")
    await roster.revoke_device("alice", pk_old)

    session = _session(tmp_path, roster)
    entry = _Entry(uploader_id="alice", uploader_pk=pk_old)
    transcript = b"delete-this-file"

    assert await session._verify_uploader_sig(
        entry, transcript, sk_old.sign(transcript)) is False
    assert await session._verify_uploader_sig(
        entry, transcript, sk_new.sign(transcript)) is True, (
        "the file is still Alice's — a retired laptop does not orphan its uploads")


async def test_an_entry_with_no_uploader_id_falls_back_to_the_recorded_key(
        tmp_path, roster):
    """An index written before `uploader_id` existed must not become undeletable."""
    sk_a, pk_a, pk_x_a = _keys()
    session = _session(tmp_path, roster)
    entry = _Entry(uploader_id="", uploader_pk=pk_a)
    transcript = b"delete-this-file"

    assert await session._verify_uploader_sig(
        entry, transcript, sk_a.sign(transcript)) is True