"""
Roster and operator pairing (M3, and the mechanism that will close H3).

Negative assertions, per the posture set in Phase 11.5: each test states an attack
or a mistake that must not work. The one to keep an eye on is
`test_daemon_does_not_auto_pin_keystore_key` — the auto-pin is what made node
sovereignty inert as shipped, and it fails closed, so nothing else in the suite
notices if it comes back.

See `docs/invite-pairing-v1.md`.
"""

import base64
import time
from pathlib import Path

import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey

from meshbay_common.crypto import generate_gek, pk_to_b64, unwrap_gek_aes
from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roster import Roster, hash_code, normalize_code
from meshbay_node.transport.webrtc_server import WebRTCPeerSession


# ── Fixtures ──────────────────────────────────────────────────────────────────

@pytest.fixture
async def roster(tmp_path):
    r = Roster(db_path=tmp_path / "roster.db")
    await r.open()
    yield r
    await r.close()


def _keypair_full():
    """(sk_ed, pk_ed_b64, pk_x_b64, sk_x) — the X25519 secret is needed to unwrap."""
    sk_ed = Ed25519PrivateKey.generate()
    sk_x = X25519PrivateKey.generate()
    pk_ed_b64 = pk_to_b64(sk_ed.public_key())
    pk_x_b64 = base64.b64encode(
        sk_x.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )
    ).decode()
    return sk_ed, pk_ed_b64, pk_x_b64, sk_x


def _keypair():
    sk_ed, pk_ed_b64, pk_x_b64, _ = _keypair_full()
    return sk_ed, pk_ed_b64, pk_x_b64


def _session(tmp_path: Path, roster, user_id: str = "grenet",
             group_id: str | None = None, gek: bytes | None = None,
             join_policy: str = "invite") -> WebRTCPeerSession:
    """A peer session with the join path wired and sending stubbed out."""
    shared_root = tmp_path / "shared"
    shared_root.mkdir(exist_ok=True)
    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())

    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {
        "shared_root": shared_root,
        "index": index,
        "sk_node": index.sk_node,
        "roster": roster,
    }
    if group_id:
        session._ctx["groups"] = {
            group_id: {
                "gek": gek,
                "shared_root": shared_root,
                "index": index,
                "join_policy": join_policy,
            },
        }
    session._group_id = group_id
    session._user_id = user_id
    session._username = user_id
    session._pk_user = ""
    session._uploads = {}
    session._join_attempts = 0
    session._nonce_node = b"\x11" * 32
    session._remote_ip = ""
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    return session


def _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="", user_id="grenet",
              group_id="", nonce=None, ts=None):
    ts = int(time.time()) if ts is None else ts
    transcript = join_transcript(
        node_pk_b64=session._node_pk_b64(),
        group_id=group_id,
        user_id=user_id,
        pk_ed25519_b64=pk_ed_b64,
        pk_x25519_b64=pk_x_b64,
        nonce_node=nonce if nonce is not None else session._nonce_node,
        ts=ts,
    )
    return {
        "type": "join_request",
        "group_id": group_id,
        "pk_ed25519": pk_ed_b64,
        "pk_x25519": pk_x_b64,
        "code": code,
        "ts": ts,
        "sig": base64.b64encode(sk_ed.sign(transcript)).decode(),
    }


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


# ── Roster ────────────────────────────────────────────────────────────────────

async def test_invite_is_single_use(roster):
    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
    assert await roster.consume_invite(code, "grenet") is not None
    assert await roster.consume_invite(code, "grenet") is None, (
        "a pairing code must not be redeemable twice")


async def test_invite_is_bound_to_one_account(roster):
    """A leaked code must be useless to whoever finds it."""
    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
    assert await roster.consume_invite(code, "eve") is None
    assert await roster.consume_invite(code, "grenet") is not None


async def test_expired_invite_is_refused(roster):
    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli", ttl=-1)
    assert await roster.consume_invite(code, "grenet") is None


async def test_reinvite_supersedes_the_previous_code(roster):
    first = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
    second = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
    assert await roster.consume_invite(first, "grenet") is None
    assert await roster.consume_invite(second, "grenet") is not None


async def test_codes_are_not_stored_in_the_clear(roster, tmp_path):
    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
    rows = await roster.list_invites()
    assert rows and rows[0]["code_hash"] != normalize_code(code)
    assert rows[0]["code_hash"] == hash_code(code)


def test_code_normalization_absorbs_human_error():
    """Someone reading a code aloud must not be able to get it wrong."""
    assert normalize_code("k7m2-qx4p") == normalize_code("K7M2QX4P")
    assert normalize_code("O1IL") == "0111"
    assert normalize_code(" k7m2 qx4p ") == "K7M2QX4P"


async def test_operator_pks_reflect_unpinning(roster):
    _, pk_ed_b64, pk_x_b64 = _keypair()
    await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code")
    await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")
    assert await roster.operator_pks() == [pk_ed_b64]

    await roster.unpin("grenet")
    assert await roster.operator_pks() == [], (
        "authority must disappear with the pin, without a daemon restart")


# ── Join / pairing over MNP ───────────────────────────────────────────────────

async def test_pairing_with_a_valid_code_pins_the_identity(tmp_path, roster):
    session = _session(tmp_path, roster)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")

    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code))

    assert _last(session).get("ok") is True
    pinned = await roster.get_identity("grenet")
    assert pinned["pk_ed25519"] == pk_ed_b64
    assert await roster.operator_pks() == [pk_ed_b64]


async def test_pairing_without_a_code_is_refused(tmp_path, roster):
    """Fails closed: an unknown identity gets nothing until someone authorizes it."""
    session = _session(tmp_path, roster)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()

    await session._do_join_request(_join_msg(session, sk_ed, pk_ed_b64, pk_x_b64))

    assert _last(session).get("ok") is False
    assert _last(session).get("reason") == "code_required"
    assert await roster.get_identity("grenet") is None


async def test_wrong_code_pins_nothing(tmp_path, roster):
    session = _session(tmp_path, roster)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")

    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="ZZZZ-ZZZZ"))

    assert _last(session).get("reason") == "code_invalid"
    assert await roster.get_identity("grenet") is None


async def test_signature_must_cover_the_presented_keys(tmp_path, roster):
    """
    The heart of it: the X25519 key is only trustworthy because the Ed25519
    identity signed it. Swapping in another encryption key after signing must fail.
    """
    session = _session(tmp_path, roster)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")

    msg = _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code)
    _, _, attacker_pk_x = _keypair()
    msg["pk_x25519"] = attacker_pk_x

    await session._do_join_request(msg)

    assert _last(session).get("reason") == "signature_invalid"
    assert await roster.get_identity("grenet") is None


async def test_join_cannot_be_replayed_onto_another_connection(tmp_path, roster):
    session = _session(tmp_path, roster)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")

    # Signed against a nonce this connection never issued.
    msg = _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code,
                    nonce=b"\x99" * 32)
    await session._do_join_request(msg)

    assert _last(session).get("reason") == "signature_invalid"
    assert await roster.get_identity("grenet") is None


async def test_pinned_identity_presenting_a_new_key_is_refused(tmp_path, roster):
    """
    11.5.8's rule, applied to people: a changed key is refused outright rather
    than warned about, and clearing it is a deliberate operator action.
    """
    session = _session(tmp_path, roster)
    _, old_pk_ed, old_pk_x = _keypair()
    await roster.pin_identity("grenet", "grenet", old_pk_ed, old_pk_x, "code")

    sk_ed2, new_pk_ed, new_pk_x = _keypair()
    await session._do_join_request(
        _join_msg(session, sk_ed2, new_pk_ed, new_pk_x, code="ANY-CODE"))

    assert _last(session).get("reason") == "key_changed"
    assert (await roster.get_identity("grenet"))["pk_ed25519"] == old_pk_ed


async def test_attempts_are_bounded(tmp_path, roster):
    session = _session(tmp_path, roster)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")

    for _ in range(6):
        await session._do_join_request(
            _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="AAAA-AAAA"))

    assert any(m.get("detail") == "Too many attempts" for m in session.sent), (
        "a connection must not be able to sit there guessing codes")


async def test_failures_are_counted_across_connections(tmp_path, roster):
    """
    The adversary who can mint a token for any account is the hub, and it can
    reconnect at will — so a per-connection budget alone would bound nothing.
    """
    shared_ctx = None
    for _ in range(6):
        session = _session(tmp_path, roster)
        if shared_ctx is None:
            shared_ctx = session._ctx
        else:
            session._ctx = shared_ctx          # same node, new connection
        sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
        for _ in range(4):
            await session._do_join_request(
                _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="AAAA-AAAA"))

    assert any(m.get("detail") == "Pairing temporarily locked"
               for m in session.sent), (
        "reconnecting must not reset the pairing budget")


async def test_group_id_cannot_name_another_group(tmp_path, roster):
    session = _session(tmp_path, roster)
    session._group_id = "a" * 32
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()

    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, group_id="b" * 32))

    assert _last(session).get("reason") == "group_mismatch"


# ── H3: the node wraps the group key, and only for people it admitted ─────────

GROUP = "g" * 32


async def test_node_wraps_the_gek_for_the_key_the_member_proved(tmp_path, roster):
    """
    The H3 fix. Nobody fetches a public key from the hub: the node encrypts the
    group key for the X25519 key the joiner signed with their pinned identity, so
    a hub substituting a key of its own has nothing to substitute into.
    """
    gek = generate_gek()
    session = _session(tmp_path, roster, user_id="bob", group_id=GROUP, gek=gek)
    sk_ed, pk_ed_b64, pk_x_b64, sk_x = _keypair_full()

    code = await roster.create_invite(GROUP, "bob", ROLE_MEMBER, "grenet")
    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code,
                  user_id="bob", group_id=GROUP))

    reply = _last(session)
    assert reply["ok"] is True and reply["gek"] is True

    pk_x_raw = base64.b64decode(pk_x_b64)
    sk_x_raw = sk_x.private_bytes(
        encoding=serialization.Encoding.Raw,
        format=serialization.PrivateFormat.Raw,
        encryption_algorithm=serialization.NoEncryption(),
    )
    assert unwrap_gek_aes(reply, sk_x_raw, pk_x_raw) == gek


async def test_hub_membership_alone_yields_no_key(tmp_path, roster):
    """
    A hub can invent an account, add it to a group and mint it a token. What it
    cannot do is put it on the node's roster — so the key never leaves.
    """
    gek = generate_gek()
    session = _session(tmp_path, roster, user_id="eve", group_id=GROUP, gek=gek)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()

    # Pinned on this node (say, for another group) but never admitted to this one.
    await roster.pin_identity("eve", "eve", pk_ed_b64, pk_x_b64, "code")

    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64,
                  user_id="eve", group_id=GROUP))

    reply = _last(session)
    assert reply.get("gek") is False
    assert reply.get("reason") == "not_authorized_for_group"
    assert "wrapped_b64" not in reply


async def test_open_join_group_admits_without_a_code(tmp_path, roster):
    """§3.4: where anyone may join, a code protects nothing and is not required."""
    gek = generate_gek()
    session = _session(tmp_path, roster, user_id="newcomer", group_id=GROUP,
                       gek=gek, join_policy="open")
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()

    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64,
                  user_id="newcomer", group_id=GROUP))

    reply = _last(session)
    assert reply["ok"] is True and reply["gek"] is True
    pinned = await roster.get_identity("newcomer")
    assert pinned["pinned_via"] == "tofu"


async def test_invite_only_group_still_demands_a_code(tmp_path, roster):
    """Being public (discoverable) is not being open (admitting anyone)."""
    gek = generate_gek()
    session = _session(tmp_path, roster, user_id="newcomer", group_id=GROUP,
                       gek=gek, join_policy="invite")
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()

    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64,
                  user_id="newcomer", group_id=GROUP))

    assert _last(session).get("reason") == "code_required"
    assert await roster.get_identity("newcomer") is None


async def test_unknown_group_is_invite_only(tmp_path, roster):
    """
    Fail closed: a group whose policy the node cannot read is treated as
    invite-only, never as open.
    """
    session = _session(tmp_path, roster, user_id="newcomer")
    session._group_id = "unconfigured-group"
    assert session._group_join_policy("unconfigured-group") == "invite"
    assert session._group_join_policy("") == "invite"


def test_join_policy_is_carried_from_node_config():
    """
    The policy reaches the transport from node.toml. If it ever came from the hub
    instead, a hub could declare any group open and be handed its key.
    """
    daemon_src = (Path(__file__).parent.parent
                  / "src" / "meshbay_node" / "daemon.py").read_text()
    assert '"join_policy": group_cfg.join_policy' in daemon_src

    config_src = (Path(__file__).parent.parent
                  / "src" / "meshbay_node" / "config.py").read_text()
    assert "join_policy" in config_src, "GroupConfig must carry the admission policy"


async def test_revoked_member_stops_receiving_the_key(tmp_path, roster):
    """
    Wrapping on demand is what makes revocation work. A stored bundle survived
    revocation; this does not. (Rotating the GEK is still required — the
    ex-member has the old one.)
    """
    gek = generate_gek()
    session = _session(tmp_path, roster, user_id="bob", group_id=GROUP, gek=gek)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    await roster.pin_identity("bob", "bob", pk_ed_b64, pk_x_b64, "code")
    await roster.set_member(GROUP, "bob", ROLE_MEMBER, "active", "grenet")

    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64,
                  user_id="bob", group_id=GROUP))
    assert _last(session)["gek"] is True

    await roster.set_status(GROUP, "bob", "revoked")
    session.sent.clear()
    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64,
                  user_id="bob", group_id=GROUP))
    assert _last(session).get("gek") is False


# ── What a first-time joiner can know ─────────────────────────────────────────

def test_challenge_carries_node_pk_in_source():
    """
    Belt and braces for the above: the field must be in the message the node
    builds, whatever the surrounding handshake does.
    """
    source = (Path(__file__).parent.parent
              / "src" / "meshbay_node" / "transport" / "webrtc_server.py").read_text()
    challenge = source[source.find("MNP.HANDSHAKE_CHALLENGE,"):]
    challenge = challenge[:challenge.find("})")]
    assert "node_pk" in challenge, (
        "the challenge must announce the node key — a first-time joiner cannot "
        "learn it any other way, and join_request signs it")


async def test_a_key_pinned_by_one_node_is_worthless_at_another(tmp_path, roster):
    """
    The whole point of per-node identity: node A's operator who cracks the bundle
    on their own disk holds a key node B has never seen. Presenting it there is a
    first contact like any other — it needs a code from B's operator.
    """
    gek = generate_gek()
    node_b = _session(tmp_path, roster, user_id="bob", group_id=GROUP, gek=gek)

    # The key bob uses at node A. Node B's roster knows nothing about it.
    sk_ed_a, pk_ed_a, pk_x_a = _keypair()

    await node_b._do_join_request(
        _join_msg(node_b, sk_ed_a, pk_ed_a, pk_x_a,
                  user_id="bob", group_id=GROUP))

    assert _last(node_b).get("reason") == "code_required"
    assert await roster.get_identity("bob") is None


async def test_the_stolen_key_cannot_be_forced_in_with_someone_elses_code(
        tmp_path, roster):
    """And a code issued for another account does not help either."""
    gek = generate_gek()
    session = _session(tmp_path, roster, user_id="eve", group_id=GROUP, gek=gek)
    sk_ed, pk_ed, pk_x = _keypair()
    code = await roster.create_invite(GROUP, "bob", ROLE_MEMBER, "grenet")

    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed, pk_x, code=code,
                  user_id="eve", group_id=GROUP))

    assert _last(session).get("reason") == "code_invalid"
    assert await roster.get_identity("eve") is None


# ── Code lifetimes ────────────────────────────────────────────────────────────

async def test_invitations_outlive_pairing_codes(roster):
    """
    An invitation crosses a human conversation; a pairing code crosses an SSH
    session. A day was long enough for the second and not for the first — a code
    that dies over a weekend means someone has to be at a browser to reissue it.
    """
    from meshbay_node.roster import DEFAULT_INVITE_TTL, DEFAULT_PAIR_TTL

    assert DEFAULT_INVITE_TTL == 7 * 24 * 3600
    assert DEFAULT_PAIR_TTL == 24 * 3600
    assert DEFAULT_INVITE_TTL > DEFAULT_PAIR_TTL


def test_code_lifetimes_are_configurable(tmp_path):
    """The operator decides, not the default."""
    from meshbay_node.config import load_config

    path = tmp_path / "node.toml"
    path.write_text(
        '[hub]\nurl = "https://example.org"\nusername = "grenet"\n'
        "[node]\ninvite_ttl_hours = 72\npair_ttl_hours = 2\n"
    )
    cfg = load_config(path)
    assert cfg.node.invite_ttl_hours == 72
    assert cfg.node.pair_ttl_hours == 2

    default = load_config(tmp_path / "missing.toml")
    assert default.node.invite_ttl_hours == 168
    assert default.node.pair_ttl_hours == 24


async def test_expiry_is_enforced_at_redemption(tmp_path, roster):
    """Purging is housekeeping; the check that matters happens on use."""
    session = _session(tmp_path, roster)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli", ttl=-1)

    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code))

    assert _last(session).get("reason") == "code_invalid"
    assert await roster.get_identity("grenet") is None


# ── M3: where node authority comes from ───────────────────────────────────────

async def test_admin_signature_verified_against_the_paired_key(tmp_path, roster):
    session = _session(tmp_path, roster)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code")
    await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")

    transcript = b"meshbay:admin:v1 whatever"
    assert await session._verify_admin_sig(transcript, sk_ed.sign(transcript))

    stranger = Ed25519PrivateKey.generate()
    assert not await session._verify_admin_sig(
        transcript, stranger.sign(transcript))


async def test_unpinned_operator_loses_authority_immediately(tmp_path, roster):
    """No caching: revoking a paired browser must not need a daemon restart."""
    session = _session(tmp_path, roster)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code")
    await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")

    transcript = b"meshbay:admin:v1 whatever"
    assert await session._verify_admin_sig(transcript, sk_ed.sign(transcript))

    await roster.unpin("grenet")
    assert not await session._verify_admin_sig(transcript, sk_ed.sign(transcript))


# ── Operator surface (slice 3) ────────────────────────────────────────────────

def _ui_client(tmp_path, roster, **extra):
    from fastapi.testclient import TestClient

    from meshbay_node.config import Config
    from meshbay_node.ui.app import create_ui_app

    state = {
        "status": "running", "groups_ctx": {GROUP: {"gek": b"k" * 32}},
        "indexes": {}, "ui_token": "tok", "roster": roster,
        "node_user_id": "grenet", "config": Config(),
    }
    state.update(extra)
    return TestClient(create_ui_app(state)), state


async def test_revoke_endpoint_stops_authorization(tmp_path, roster):
    client, _ = _ui_client(tmp_path, roster)
    _, pk_ed_b64, pk_x_b64 = _keypair()
    await roster.pin_identity("bob", "bob", pk_ed_b64, pk_x_b64, "code")
    await roster.set_member(GROUP, "bob", ROLE_MEMBER, "active", "grenet")
    assert await roster.is_authorized(GROUP, "bob")

    resp = client.post(f"/api/members/bob/revoke?group_id={GROUP}&t=tok")
    assert resp.status_code == 200
    assert "gek-init" in resp.json()["reminder"], (
        "revocation must remind the operator to rotate the key they still hold")
    assert not await roster.is_authorized(GROUP, "bob")


async def test_unpin_endpoint_allows_repairing(tmp_path, roster):
    client, _ = _ui_client(tmp_path, roster)
    _, pk_ed_b64, pk_x_b64 = _keypair()
    await roster.pin_identity("bob", "bob", pk_ed_b64, pk_x_b64, "code")

    assert client.post("/api/members/bob/unpin?t=tok").status_code == 200
    assert await roster.get_identity("bob") is None
    assert client.post("/api/members/bob/unpin?t=tok").status_code == 404


async def test_operator_surface_needs_the_session_token(tmp_path, roster):
    """11.5.3 applies to every one of these: they change who may hold the key."""
    client, _ = _ui_client(tmp_path, roster)
    for path in ("/api/roster",
                 "/api/operator/pair",
                 f"/api/members/bob/revoke?group_id={GROUP}",
                 "/api/members/bob/unpin",
                 f"/api/groups/{GROUP}/invites?username=bob"):
        method = client.get if path == "/api/roster" else client.post
        assert method(path).status_code == 403, f"{path} reachable without a token"


async def test_cli_invite_asks_the_hub_for_an_account_never_a_key(tmp_path, roster):
    """
    The CLI resolves a username to an account id through the hub, and stops there.
    A key fetched from the hub is what H3 was; an account id is not a secret and
    a wrong one produces an invite whose code the hub never learns.
    """
    class _Hub:
        _session = object()

        async def get_user_pubkeys(self, username):
            return {"user_id": f"id-of-{username}",
                    "pk_x25519": "SHOULD-NOT-BE-USED",
                    "pk_ed25519": "SHOULD-NOT-BE-USED"}

    client, _ = _ui_client(tmp_path, roster, hub=_Hub())
    resp = client.post(f"/api/groups/{GROUP}/invites?username=bob&t=tok")
    assert resp.status_code == 200
    body = resp.json()
    assert body["user_id"] == "id-of-bob"

    invites = await roster.list_invites()
    assert [i["user_id"] for i in invites] == ["id-of-bob"]
    # Whatever the hub said about keys was never stored anywhere.
    assert "SHOULD-NOT-BE-USED" not in str(invites)
    assert await roster.get_identity("id-of-bob") is None


def _run_cli(monkeypatch, tmp_path, argv, responses):
    """Drive the real CLI with the daemon API stubbed, capturing the calls."""
    import sys as _sys

    from meshbay_node import daemon as _daemon

    calls = []

    def fake_api(cfg, path, method="GET", timeout=30):
        calls.append((method, path))
        for key, value in responses.items():
            if key in path:
                return value
        return {}

    monkeypatch.setattr(_daemon, "_daemon_api", fake_api)

    conf = tmp_path / "node.toml"
    conf.write_text(
        f'data_dir = "{tmp_path}"\n'
        '[hub]\nurl = "https://example.org"\nusername = "grenet"\n'
        f'[[groups]]\nid = "{GROUP}"\nname = "demo"\n'
        f'shared_dir = "{tmp_path}"\n'
    )
    monkeypatch.setattr(_sys, "argv",
                        ["meshbay-node", *argv, "--config", str(conf)])
    try:
        _daemon.main()
    except SystemExit as e:
        calls.append(("exit", e.code))
    return calls


def test_cli_member_commands_reach_the_right_endpoints(monkeypatch, tmp_path, capsys):
    resolved = {"user_id": "u-bob", "source": "roster"}

    calls = _run_cli(monkeypatch, tmp_path, ["member", "revoke", "bob"],
                     {"/api/resolve": resolved,
                      "revoke": {"status": "revoked", "reminder": "gek-init"}})
    assert ("POST", f"/api/members/u-bob/revoke?group_id={GROUP}") in calls
    # The operator is told the revocation does not take back the key they hold.
    assert "rotate" in capsys.readouterr().out.lower()

    calls = _run_cli(monkeypatch, tmp_path, ["member", "unpin", "bob"],
                     {"/api/resolve": resolved, "unpin": {"status": "unpinned"}})
    assert ("POST", "/api/members/u-bob/unpin") in calls


def test_cli_resolves_a_name_before_acting(monkeypatch, tmp_path):
    """
    The name has to be turned into an account first, and the node's own roster is
    asked before the hub. A JWT carries no username, so an identity pinned without
    an invitation has none — the hub fallback is what keeps it manageable.
    """
    calls = _run_cli(monkeypatch, tmp_path, ["member", "revoke", "bob"],
                     {"/api/resolve": {"user_id": "u-bob", "source": "hub"},
                      "revoke": {"status": "revoked", "reminder": "gek-init"}})

    assert ("GET", "/api/resolve?username=bob") == calls[0], (
        "the CLI must resolve the name before acting on anyone")
    assert ("POST", f"/api/members/u-bob/revoke?group_id={GROUP}") in calls

def test_daemon_does_not_auto_pin_keystore_key():
    """
    M3: the daemon used to auto-pin its own keystore key as the admin key, while
    the browser signs with the user's identity key. Different keys, so every
    privileged operation failed closed with a signature error that looked like a
    bug elsewhere — and the demo only worked because a deploy script overwrote it.

    Authority now comes from the roster, or from an explicit node.toml value.
    """
    source = (Path(__file__).parent.parent
              / "src" / "meshbay_node" / "daemon.py").read_text()
    assert "Auto-pinning admin key" not in source
    assert "_resolve_admin_pk" not in source, (
        "the auto-pin resolver is back — node authority must be established "
        "locally by pairing, never inferred from the node's own keystore (M3)")


def test_admin_authority_is_never_fetched_from_the_hub():
    """
    The fix M3 invites: ask the hub which key belongs to the operator. That would
    hand a malicious hub the node — the same substitution as H3, one level deeper.
    """
    src = Path(__file__).parent.parent / "src" / "meshbay_node"

    verifier = (src / "transport" / "webrtc_server.py").read_text()
    body = verifier[verifier.index("async def _verify_admin_sig"):]
    body = body[:body.index("\n    def ", 1)]
    assert "operator_pks" in body, "the roster is where authority comes from"
    # Past the docstring: it names what was removed on purpose, so a reader knows
    # not to put it back. What must not reappear is code.
    code = body[body.index('"""', body.index('"""') + 3):]
    for forbidden in ("hub", "pubkeys", "admin_pk_ed25519"):
        assert forbidden not in code, (
            f"_verify_admin_sig mentions {forbidden!r} — authority must come from "
            "the local roster and nothing else")

    daemon = (src / "daemon.py").read_text()
    assert "has_operator()" in daemon, "the daemon reads authority from the roster"
    assert "admin_pk_ed25519" not in daemon, (
        "the node.toml operator key is gone; it must not come back as a second "
        "source of authority")


# ── Hosting another group ────────────────────────────────────────────────────

async def test_group_add_appends_without_rewriting_the_file(tmp_path):
    """
    node.toml is hand-written and full of comments explaining decisions. The
    block is appended as text for that reason: a round trip through a TOML
    writer would silently throw all of it away.
    """
    from meshbay_node.config import load_config

    conf = tmp_path / "node.toml"
    conf.write_text(
        '# keep me\n[hub]\nurl = "https://meshbay.org"\nusername = "grenet"\n\n'
        '[[groups]]\nid = "aaaa"\nname = "first"\nshared_dir = "/tmp/a"\n')

    block = ('\n[[groups]]\n'
             'id          = "bbbb"\n'
             'name        = "second"\n'
             'shared_dir  = "/tmp/b"\n'
             'visibility  = "private"\n')
    with conf.open("a") as f:
        f.write(block)

    assert "# keep me" in conf.read_text(), "comments must survive"
    cfg = load_config(conf)
    assert [g.name for g in cfg.groups] == ["first", "second"]
    assert [g.shared_dir for g in cfg.groups] == ["/tmp/a", "/tmp/b"]


async def test_each_group_gets_its_own_key(tmp_path, roster):
    """
    Two groups on one node are two separate memberships and two separate keys:
    being admitted to one must say nothing about the other. This is the property
    that makes hosting a second group meaningful rather than cosmetic.
    """
    from meshbay_common.crypto import generate_gek

    gek_a, gek_b = generate_gek(), generate_gek()
    assert gek_a != gek_b

    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    await roster.pin_identity("member", "member", pk_ed_b64, pk_x_b64, "code")
    await roster.set_member("group-a", "member", ROLE_MEMBER, "active", "op")

    assert await roster.is_authorized("group-a", "member") is True
    assert await roster.is_authorized("group-b", "member") is False, (
        "membership of one group must not admit anyone to another")