"""
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 conftest import one_root
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 = {
        "roots": one_root(shared_root),
        "index": index,
        "sk_node": index.sk_node,
        "roster": roster,
    }
    if group_id:
        session._ctx["groups"] = {
            group_id: {
                "gek": gek,
                "roots": one_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
    session._ctx["daemon_state"] = {
        "roster": roster,
        "groups_ctx": session._ctx.get("groups", {}),
    }
    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_a_key_this_node_never_pinned_is_refused(tmp_path, roster):
    """
    11.5.8's rule, applied to people: an unrecognised key does not get in, and
    a code cannot talk its way past that.

    What changed with device linking (2026-08-18) is the way back, not the
    refusal. This used to be `key_changed` and needed an operator to unpin; now
    it is `unknown_device` and the person approves the new key from a device
    already paired here. Nothing is pinned either way, which is the part that
    matters.
    """
    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") == "unknown_device"
    assert await roster.find_device("grenet", new_pk_ed) is None
    assert [d["pk_ed25519"] for d in await roster.list_devices("grenet")] == \
        [old_pk_ed]


async def test_a_pinned_key_arriving_with_a_different_x25519_is_refused(
        tmp_path, roster):
    """
    The join transcript signs both keys together, so a pinned Ed25519 key
    presenting a different encryption key is either a client that regenerated
    half its identity or two messages spliced. Either way the pair is not the
    one admitted, and the group key must not be wrapped for it.
    """
    session = _session(tmp_path, roster)
    sk_ed, pk_ed, pk_x = _keypair()
    await roster.pin_identity("grenet", "grenet", pk_ed, pk_x, "code")

    _, _, other_pk_x = _keypair()
    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed, other_pk_x, code="ANY-CODE"))

    assert _last(session).get("reason") == "key_changed"
    assert (await roster.find_device("grenet", pk_ed))["pk_x25519"] == pk_x


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 rotate" 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")


# ── Removing a directory ─────────────────────────────────────────────────────

async def _dir_session(tmp_path, roster):
    """A session with a shared root and an operator paired, ready for admin ops."""
    session = _session(tmp_path, roster, group_id="g1", gek=generate_gek())
    session._admin_ops = {}
    session._ctx["has_admin_authority"] = True
    return session


async def test_a_directory_with_anything_in_it_is_refused(tmp_path, roster):
    session = await _dir_session(tmp_path, roster)
    full = tmp_path / "shared" / "full"
    full.mkdir()
    (full / "keep.txt").write_text("still here")

    session._do_dir_delete({"dir": "shared/full"})

    assert _last(session).get("detail") == "Directory is not empty"
    assert full.exists() and (full / "keep.txt").exists()


async def test_no_challenge_is_issued_without_an_operator(tmp_path, roster):
    """Fails closed, and says so, rather than asking for a signature nobody can give."""
    session = await _dir_session(tmp_path, roster)
    session._ctx["has_admin_authority"] = False
    (tmp_path / "shared" / "empty").mkdir()

    session._do_dir_delete({"dir": "shared/empty"})

    assert _last(session).get("detail") == "No authorized key for deletion"
    assert (tmp_path / "shared" / "empty").exists()


async def test_a_root_itself_is_not_a_target(tmp_path, roster):
    """
    Neither the virtual root nor a root directory can be removed this way.

    Removing a root is a configuration change: doing it through a file operation
    would leave the group config naming a directory nobody can reach. And the
    virtual root is not a directory on anyone's disk at all — it belongs to no
    volume.
    """
    session = await _dir_session(tmp_path, roster)
    for attempt in ("", ".", "/", "../shared", "shared", "shared/", "SHARED"):
        session._do_dir_delete({"dir": attempt})
        assert _last(session).get("type") == "error", f"{attempt!r} was accepted"
    assert (tmp_path / "shared").is_dir()


async def test_escaping_the_shared_root_is_refused(tmp_path, roster):
    session = await _dir_session(tmp_path, roster)
    outside = tmp_path / "outside"
    outside.mkdir()

    # Both shapes: a path that names no root at all, and one that starts inside
    # a real root and then climbs out of it.
    for attempt in ("../outside", "../../outside", "sub/../../outside",
                    "shared/../outside", "shared/../../outside",
                    "shared/sub/../../outside"):
        session._do_dir_delete({"dir": attempt})
        assert _last(session).get("type") == "error", f"{attempt!r} was accepted"
    assert outside.is_dir(), "a path leaving the shared root removed a directory"


async def test_an_empty_directory_needs_a_signature_and_then_goes(tmp_path, roster):
    """The whole round trip: challenge, operator signature, removal."""
    from meshbay_common.adminop import admin_transcript

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

    session = await _dir_session(tmp_path, roster)
    (tmp_path / "shared" / "gone").mkdir()

    session._do_dir_delete({"dir": "shared/gone"})
    challenge = _last(session)
    assert challenge["type"] == "admin_challenge"
    assert challenge["op"] == "dir_delete"
    assert challenge["subject"] == "shared/gone"

    transcript = admin_transcript(
        op="dir_delete", node_pk_b64=session._node_pk_b64(), group_id="g1",
        subject="shared/gone", nonce=base64.b64decode(challenge["nonce"]),
        ts=challenge["ts"])
    await session._admin_exec_dir_delete(
        session._admin_ops.pop(challenge["op_id"]) if session._admin_ops
        else {"op": "dir_delete", "subject": "shared/gone"},
        transcript, sk_ed.sign(transcript))

    assert _last(session)["type"] == "dir_delete_ack"
    assert not (tmp_path / "shared" / "gone").exists()


async def test_someone_elses_signature_does_not_remove_it(tmp_path, roster):
    from meshbay_common.adminop import admin_transcript

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

    sk_member, pk_member, pk_x_m = _keypair()
    await roster.pin_identity("mallory", "mallory", pk_member, pk_x_m, "code")
    await roster.set_member("g1", "mallory", ROLE_MEMBER, "active", "grenet")

    session = await _dir_session(tmp_path, roster)
    (tmp_path / "shared" / "theirs").mkdir()

    transcript = admin_transcript(
        op="dir_delete", node_pk_b64=session._node_pk_b64(), group_id="g1",
        subject="shared/theirs", nonce=b"\x22" * 32, ts=int(time.time()))
    await session._admin_exec_dir_delete(
        {"op": "dir_delete", "subject": "shared/theirs"},
        transcript, sk_member.sign(transcript))

    assert _last(session).get("detail") == "Signature verification failed"
    assert (tmp_path / "shared" / "theirs").is_dir(), (
        "a member's signature removed a directory — only the operator may")


# ── Removing a member ────────────────────────────────────────────────────────

async def test_revoking_needs_an_operator_signature(tmp_path, roster):
    from meshbay_common.adminop import admin_transcript

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

    sk_m, pk_m, pk_x_m = _keypair()
    await roster.pin_identity("victim", "victim", pk_m, pk_x_m, "code")
    await roster.set_member("g1", "victim", ROLE_MEMBER, "active", "grenet")

    session = _session(tmp_path, roster, group_id="g1", gek=generate_gek())
    session._admin_ops = {}
    session._ctx["has_admin_authority"] = True
    session._ctx["peers"] = {}

    transcript = admin_transcript(
        op="member_revoke", node_pk_b64=session._node_pk_b64(), group_id="g1",
        subject="victim", nonce=b"\x33" * 32, ts=int(time.time()))

    # A member's own signature is not enough.
    await session._admin_exec_member_revoke(
        {"op": "member_revoke", "subject": "victim"}, transcript,
        sk_m.sign(transcript))
    assert _last(session).get("detail") == "Signature verification failed"
    assert (await roster.get_member("g1", "victim"))["status"] == "active"

    # The operator's is.
    await session._admin_exec_member_revoke(
        {"op": "member_revoke", "subject": "victim"}, transcript,
        sk_op.sign(transcript))
    assert _last(session)["type"] == "member_revoke_ack"
    assert (await roster.get_member("g1", "victim"))["status"] == "revoked"


async def test_revoking_is_confined_to_the_group_it_was_asked_for(tmp_path, roster):
    """
    A node hosting two groups must not lose someone from both. Their pinned
    identity survives as well — forgetting a key is `member unpin`, and saying
    "remove them" should not silently do it.
    """
    from meshbay_common.adminop import admin_transcript

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

    _, pk_m, pk_x_m = _keypair()
    await roster.pin_identity("both", "both", pk_m, pk_x_m, "code")
    await roster.set_member("g1", "both", ROLE_MEMBER, "active", "grenet")
    await roster.set_member("g2", "both", ROLE_MEMBER, "active", "grenet")

    session = _session(tmp_path, roster, group_id="g1", gek=generate_gek())
    session._admin_ops = {}
    session._ctx["has_admin_authority"] = True
    session._ctx["peers"] = {}

    transcript = admin_transcript(
        op="member_revoke", node_pk_b64=session._node_pk_b64(), group_id="g1",
        subject="both", nonce=b"\x44" * 32, ts=int(time.time()))
    await session._admin_exec_member_revoke(
        {"op": "member_revoke", "subject": "both"}, transcript,
        sk_op.sign(transcript))

    assert (await roster.get_member("g1", "both"))["status"] == "revoked"
    assert (await roster.get_member("g2", "both"))["status"] == "active"
    assert await roster.get_identity("both") is not None, (
        "the pinned identity was dropped; that is `member unpin`, not this")


async def test_an_operator_cannot_revoke_themselves(tmp_path, roster):
    """It would leave the group with nobody able to invite or remove."""
    session = _session(tmp_path, roster, group_id="g1", gek=generate_gek())
    session._admin_ops = {}
    session._ctx["has_admin_authority"] = True

    session._do_member_revoke({"user_id": session._user_id})
    assert _last(session).get("detail") == "Cannot revoke yourself"