"""
Availability: what one participant can do to the others.

The three security reviews asked who can *read* what, who can impersonate whom,
and what a hostile node can forge. None of them asked what a legitimate but
misconfigured or careless member costs everyone else — and that is the question
a group platform lives or dies on, because every member is invited by someone
who trusted them and none of them is an attacker.

Written after 2026-09-11, where a member's unconfigured node was registered by
the hub as a host for a group it could not serve, and — being answered first —
made that group unopenable for everyone in it. Nothing was compromised and
nothing was forged. The group was simply gone.

Each test below is two accounts, because that is the shape the single-node
suite could not express: `_make_user` twice, a group owned by one, the other
holding whatever the defect needs. A one-member test proves a one-member
property, and every finding here needed a second person to exist at all.
"""

import base64
import time

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from meshbay_common.crypto import pk_to_b64


async def _make_user(client, username: str) -> dict:
    sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate()
    pk_ed, pk_x = pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key())
    r = await client.post("/v1/users/register", json={
        "username": username,
        "email": f"{username}@example.test",
        "auth_key": base64.b64encode(b"k" * 32).decode(),
        "pk_user_ed25519": pk_ed,
        "pk_user_x25519": pk_x,
    })
    assert r.status_code == 201, r.text
    user_id = r.json()["user_id"]
    r = await client.post("/v1/users/login", json={
        "username": username,
        "auth_key": base64.b64encode(b"k" * 32).decode(),
    })
    assert r.status_code == 200, r.text
    return {"user_id": user_id, "username": username,
            "token": r.json()["access_token"], "pk_ed": pk_ed, "sk_ed": sk_ed}


async def _announce_node(client, user: dict) -> str:
    ts = int(time.time())
    msg = f"meshbay:node_announce:{user['user_id']}:{user['pk_ed']}:{ts}".encode()
    r = await client.post("/v1/nodes/announce", json={
        "pk_node": user["pk_ed"], "endpoint_hint": "test", "timestamp": ts,
        "signature": base64.b64encode(user["sk_ed"].sign(msg)).decode(),
    }, headers={"Authorization": f"Bearer {user['token']}"})
    assert r.status_code == 201, r.text
    return r.json()["node_id"]


async def _make_group(client, owner: dict, name: str) -> str:
    r = await client.post(
        "/v1/groups",
        json={"name": name, "visibility": "private", "join_policy": "invite"},
        headers={"Authorization": f"Bearer {owner['token']}"})
    assert r.status_code == 201, r.text
    return r.json()["group_id"]


async def _add_member(client, owner: dict, group_id: str, member: dict) -> None:
    r = await client.post(
        f"/v1/groups/{group_id}/members/{member['username']}",
        headers={"Authorization": f"Bearer {owner['token']}"})
    assert r.status_code == 201, r.text


# ── A member's node must not speak for a group it does not host ──────────────

@pytest.mark.asyncio
async def test_a_members_node_cannot_notify_a_group_it_does_not_host(client):
    """
    `chat_notify` carried a `group_id` the hub believed, so any connected node
    could write a notification to every member of any group on the hub, with a
    display string of its own choosing. The node's account needed no relation
    to the group whatsoever — this is the same defect as the group claim, one
    message further along the same socket.
    """
    from meshbay_hub.api import revocation as rev

    owner = await _make_user(client, "av_owner1")
    outsider = await _make_user(client, "av_outsider1")
    owner_node = await _announce_node(client, owner)
    outsider_node = await _announce_node(client, outsider)
    group_id = await _make_group(client, owner, "not-yours")

    rev._node_groups[owner_node] = [group_id]
    rev._node_groups[outsider_node] = []          # hosts nothing, as in the incident
    try:
        await rev._handle_chat_notify(
            group_id, "Someone you do not know", outsider["user_id"],
            node_id=outsider_node)

        r = await client.get("/v1/notifications",
                             headers={"Authorization": f"Bearer {owner['token']}"})
        assert r.status_code == 200, r.text
        assert r.json()["notifications"] == [], (
            "an unrelated node wrote into this account's notifications")

        # And the node that does host it is still able to.
        await rev._handle_chat_notify(
            group_id, "A real member", outsider["user_id"], node_id=owner_node)
        r = await client.get("/v1/notifications",
                             headers={"Authorization": f"Bearer {owner['token']}"})
        assert len(r.json()["notifications"]) == 1
    finally:
        rev._node_groups.pop(owner_node, None)
        rev._node_groups.pop(outsider_node, None)


@pytest.mark.asyncio
async def test_a_member_whose_node_hosts_nothing_still_cannot_notify(client):
    """
    The incident's own shape, and the one a check on *membership* would have
    missed: the second account is a real member of the group, invited by its
    owner. Their node is simply not the one holding the files. Being entitled
    to be in a group is not being entitled to speak for it.
    """
    from meshbay_hub.api import revocation as rev

    owner = await _make_user(client, "av_owner2")
    member = await _make_user(client, "av_member2")
    owner_node = await _announce_node(client, owner)
    member_node = await _announce_node(client, member)
    group_id = await _make_group(client, owner, "shared-room")
    await _add_member(client, owner, group_id, member)

    rev._node_groups[owner_node] = [group_id]
    rev._node_groups[member_node] = []
    try:
        await rev._handle_chat_notify(
            group_id, "Sounds like a member", member["user_id"],
            node_id=member_node)
        r = await client.get("/v1/notifications",
                             headers={"Authorization": f"Bearer {owner['token']}"})
        assert r.json()["notifications"] == [], (
            "a member's empty node wrote into the group owner's notifications")
    finally:
        rev._node_groups.pop(owner_node, None)
        rev._node_groups.pop(member_node, None)


@pytest.mark.asyncio
async def test_one_node_cannot_spend_the_hubs_database_on_notifications():
    """
    Each chat_notify is a query and a write per member of the group, spawned
    without backpressure. The budget is what stops one group's node from
    costing every other group on the instance.
    """
    from meshbay_hub.api.revocation import NOTIFY_BURST, _notify_budget, _notify_window

    node = "budget-node"
    _notify_window.pop(node, None)
    try:
        assert all(_notify_budget(node) for _ in range(NOTIFY_BURST))
        assert not _notify_budget(node), "the burst was not bounded"

        # A second node is unaffected — the budget is per node, not global.
        assert _notify_budget("another-node")
    finally:
        _notify_window.pop(node, None)
        _notify_window.pop("another-node", None)


@pytest.mark.asyncio
async def test_the_notify_budget_is_not_refilled_by_reconnecting(client):
    """
    The obvious place to clear this is the socket's `finally`, beside the other
    two registries — and that would make reconnecting the way around it. The
    same node token is valid for an hour.
    """
    from meshbay_hub.api import revocation as rev

    user = await _make_user(client, "av_reconnect")
    node_id = await _announce_node(client, user)
    rev._notify_window.pop(node_id, None)
    try:
        for _ in range(rev.NOTIFY_BURST):
            rev._notify_budget(node_id)
        assert not rev._notify_budget(node_id)

        # The disconnect path itself, not a re-enactment of it: `forget_node`
        # is what the socket's `finally` calls, so adding a line there is
        # caught here.
        rev.forget_node(node_id)
        assert node_id not in rev._connected_nodes
        assert node_id not in rev._node_groups

        assert not rev._notify_budget(node_id), (
            "disconnecting refilled the budget, so reconnecting defeats it")
    finally:
        rev._notify_window.pop(node_id, None)


# ── A member must not aim other people's traffic ─────────────────────────────

@pytest.mark.asyncio
async def test_a_swarm_source_cannot_name_someone_elses_address(client):
    """
    `endpoint` was free text documented as "ip:port", so an account could
    publish a third party's address as a source for any content. Nothing dials
    a swarm source today, which is the only reason this was not already the
    reflection primitive that `notify_incoming` was fixed for (H6). A port is
    all a reader needs: where the node is comes from the node record, which is
    stamped with the address its announce arrived from.
    """
    user = await _make_user(client, "av_swarm1")
    headers = {"Authorization": f"Bearer {user['token']}"}

    for bad in ("192.0.2.7:4433", "evil.example:53", "webrtc:0", "webrtc:70000",
                "webrtc:4433 ", "http://example.test"):
        r = await client.post("/v1/swarm/register", headers=headers,
                              json={"content_hash": "ab" * 32, "endpoint": bad})
        assert r.status_code == 422, f"{bad!r} was accepted: {r.text}"

    r = await client.post("/v1/swarm/register", headers=headers,
                          json={"content_hash": "ab" * 32, "endpoint": "webrtc:19010"})
    assert r.status_code == 201, r.text


@pytest.mark.asyncio
async def test_one_account_cannot_fill_the_swarm_table(client, monkeypatch):
    """Rows are keyed (hash, account) with no cap — an invented hash each time."""
    import meshbay_hub.api.groups as groups_api
    monkeypatch.setattr(groups_api, "MAX_SWARM_HASHES_PER_ACCOUNT", 3)

    user = await _make_user(client, "av_swarm2")
    headers = {"Authorization": f"Bearer {user['token']}"}
    for i in range(3):
        r = await client.post("/v1/swarm/register", headers=headers,
                              json={"content_hash": f"{i:064x}",
                                    "endpoint": "webrtc:19010"})
        assert r.status_code == 201, r.text

    r = await client.post("/v1/swarm/register", headers=headers,
                          json={"content_hash": f"{99:064x}", "endpoint": "webrtc:19010"})
    assert r.status_code == 429, r.text

    # Refreshing one already held is not a new claim and must still work.
    r = await client.post("/v1/swarm/register", headers=headers,
                          json={"content_hash": f"{0:064x}", "endpoint": "webrtc:19011"})
    assert r.status_code == 201, r.text


# ── A member's node must not answer for another's ────────────────────────────

def test_a_node_cannot_answer_an_offer_it_was_never_sent():
    """
    `handle_webrtc_answer` resolved any pending `peer_id` from any node's
    socket. The answer is the SDP the browser then connects to, so the check is
    what keeps one node from standing in for the node a client asked for. That
    it had not happened rested on a uuid4 being unguessable.
    """
    import asyncio

    from meshbay_hub.api.signaling import _answer_owner, _webrtc_answers, handle_webrtc_answer

    loop = asyncio.new_event_loop()
    try:
        future = loop.create_future()
        _webrtc_answers["peer-1"] = future
        _answer_owner["peer-1"] = "the-node-asked-for"

        handle_webrtc_answer({"peer_id": "peer-1", "sdp": "v=0 impostor"},
                             "a-different-node")
        assert not future.done(), "another node resolved this offer"

        handle_webrtc_answer({"peer_id": "peer-1", "sdp": "v=0 genuine"},
                             "the-node-asked-for")
        assert future.done() and future.result()["sdp"] == "v=0 genuine"
    finally:
        _webrtc_answers.pop("peer-1", None)
        _answer_owner.pop("peer-1", None)
        loop.close()


# ── One account must not be able to mail another at will ─────────────────────

@pytest.mark.asyncio
async def test_an_invite_email_says_what_the_hub_knows_not_what_it_is_told(
        client, monkeypatch):
    """
    `group_name` went from the request body into the subject line of an email
    the hub sends under its own domain, and the invitee need not be a member of
    anything — as they cannot be, being invited. So any account, having made a
    group, could mail any other account arbitrary text. The name now comes from
    the group row, and the code has to look like one.
    """
    sent: list = []
    import meshbay_hub.mail as mail_mod
    monkeypatch.setattr(mail_mod, "send_invite_notification",
                        lambda *a, **kw: sent.append(a))

    owner = await _make_user(client, "av_inviter")
    await _make_user(client, "av_invitee")
    group_id = await _make_group(client, owner, "real-group-name")
    headers = {"Authorization": f"Bearer {owner['token']}"}

    r = await client.post(f"/v1/groups/{group_id}/invite-notify", headers=headers,
                          json={"username": "av_invitee", "code": "not a code",
                                "group_name": "Your account is suspended"})
    assert r.status_code == 422, r.text
    assert not sent

    r = await client.post(f"/v1/groups/{group_id}/invite-notify", headers=headers,
                          json={"username": "av_invitee", "code": "AB12-CD34",
                                "group_name": "Your account is suspended"})
    assert r.status_code == 200, r.text
    assert sent, "the legitimate path stopped working"
    assert sent[0][3] == "real-group-name", (
        "the sender chose the subject line of a message the hub signs")


# ── A relay is not authenticated by the key it publishes ─────────────────────

@pytest.mark.asyncio
async def test_a_relay_must_prove_it_holds_the_approved_key(client, monkeypatch):
    """
    `relay_register` had no `Depends` and verified nothing: it compared
    `pk_relay` against the approved value, which is a **public** key. Anyone
    who could read it could rewrite where the hub tells nodes to send relayed
    traffic — an unauthenticated write to state other people's machines act
    on. The module docstring said the relay "signs keepalive JWTs"; `jwt` was
    imported and never used.
    """
    from meshbay_hub.api import relay as relay_mod

    # The registry ships closed (`relay.RELAYS_ENABLED`); the proof it demands
    # is still what will be wanted the day it opens.
    monkeypatch.setattr(relay_mod, "RELAYS_ENABLED", True)
    sk = Ed25519PrivateKey.generate()
    pk = pk_to_b64(sk.public_key())
    relay_mod._relays["r1"] = {"pk": pk, "active": False}
    try:
        # The public key alone, which used to be enough.
        r = await client.post("/v1/relays/register", json={
            "relay_id": "r1", "endpoint": "198.51.100.9:9999",
            "pk_relay": pk, "capacity": 100})
        assert r.status_code == 400, r.text
        assert relay_mod._relays["r1"].get("endpoint") is None

        # A signature over someone else's endpoint does not carry either: the
        # endpoint is inside the signed message.
        ts = int(time.time())
        sig = sk.sign(f"meshbay:relay_register:r1:10.0.0.1:4433:{ts}".encode())
        r = await client.post("/v1/relays/register", json={
            "relay_id": "r1", "endpoint": "198.51.100.9:9999", "pk_relay": pk,
            "timestamp": ts, "signature": base64.b64encode(sig).decode()})
        assert r.status_code == 401, r.text

        endpoint = "203.0.113.4:4433"
        sig = sk.sign(f"meshbay:relay_register:r1:{endpoint}:{ts}".encode())
        r = await client.post("/v1/relays/register", json={
            "relay_id": "r1", "endpoint": endpoint, "pk_relay": pk,
            "timestamp": ts, "signature": base64.b64encode(sig).decode()})
        assert r.status_code == 201, r.text
        assert relay_mod._relays["r1"]["endpoint"] == endpoint
    finally:
        relay_mod._relays.pop("r1", None)


@pytest.mark.asyncio
async def test_every_relay_route_is_closed_as_the_hub_ships(client):
    """Nothing in the tree uses the registry, and two of its routes take no account.

    A dependency on the router, so a route added later is closed too. The flag
    is read as shipped, not set here — a test that closes the gate itself
    would keep passing the day somebody opens it.
    """
    admin = await _make_user(client, "relayadmin")
    from meshbay_hub.api.deps import set_admin_usernames
    set_admin_usernames(["relayadmin"])
    auth = {"Authorization": f"Bearer {admin['token']}"}

    for method, path in (("get", "/v1/relays"),
                         ("post", "/v1/relays/register"),
                         ("post", "/v1/relays/approve")):
        kwargs = {"headers": auth} if method == "get" else {"json": {}, "headers": auth}
        r = await getattr(client, method)(path, **kwargs)
        assert r.status_code == 503, (path, r.status_code, r.text)


@pytest.mark.asyncio
async def test_a_stranger_who_locks_your_name_does_not_sign_you_out(client, db_session):
    """AV26. The sign-in lockout is keyed by username, and usernames are public.

    So anyone can spend your attempts, and the design has to make that cost as
    little as possible: a lockout refuses *passphrase* sign-in and nothing else.
    The session you already have keeps working and keeps renewing, and a reset
    code sent to your own address ends the lockout at once.
    """
    victim_key = base64.b64encode(b"k" * 32).decode()
    r = await client.post("/v1/users/register", json={
        "username": "victim26", "email": "victim26@example.test", "auth_key": victim_key})
    assert r.status_code == 201, r.text
    session = (await client.post("/v1/users/login", json={
        "username": "victim26", "auth_key": victim_key})).json()

    # The stranger needs no account at all — only the name.
    for _ in range(4):
        r = await client.post("/v1/users/login", json={
            "username": "victim26", "auth_key": "guess" + "x" * 39})
        assert r.status_code == 401
    r = await client.post("/v1/users/login", json={
        "username": "victim26", "auth_key": victim_key})
    assert r.status_code == 429, r.text

    # Still signed in, and still able to stay signed in.
    auth = {"Authorization": f"Bearer {session['access_token']}"}
    assert (await client.get("/v1/users/me", headers=auth)).status_code == 200
    r = await client.post("/v1/users/token/refresh",
                          json={"refresh_token": session["refresh_token"]})
    assert r.status_code == 200, r.text

    # The way out that needs nobody's help: a code to the address on file.
    from meshbay_hub.db.models import EmailVerification, User
    from sqlalchemy import select
    r = await client.post("/v1/users/password/reset-request", json={
        "username": "victim26", "email": "victim26@example.test"})
    assert r.status_code == 200, r.text
    uid = (await db_session.execute(
        select(User.id).where(User.username == "victim26"))).scalar_one()
    code = (await db_session.execute(select(EmailVerification.code).where(
        EmailVerification.user_id == uid,
        EmailVerification.purpose == "password_reset"))).scalar_one()
    new_key = base64.b64encode(b"n" * 32).decode()
    r = await client.post("/v1/users/password/reset", json={
        "username": "victim26", "code": code, "new_auth_key": new_key})
    assert r.status_code == 200, r.text
    r = await client.post("/v1/users/login", json={
        "username": "victim26", "auth_key": new_key})
    assert r.status_code == 200, r.text


@pytest.mark.asyncio
async def test_a_captured_relay_registration_is_not_replayable(client, monkeypatch):
    """Same reason /v1/nodes/announce bounds its timestamp."""
    from meshbay_hub.api import relay as relay_mod

    monkeypatch.setattr(relay_mod, "RELAYS_ENABLED", True)
    sk = Ed25519PrivateKey.generate()
    pk = pk_to_b64(sk.public_key())
    relay_mod._relays["r2"] = {"pk": pk, "active": False}
    try:
        ts = int(time.time()) - 3600
        sig = sk.sign(f"meshbay:relay_register:r2:203.0.113.5:4433:{ts}".encode())
        r = await client.post("/v1/relays/register", json={
            "relay_id": "r2", "endpoint": "203.0.113.5:4433", "pk_relay": pk,
            "timestamp": ts, "signature": base64.b64encode(sig).decode()})
        assert r.status_code == 401, r.text
    finally:
        relay_mod._relays.pop("r2", None)


# ── Mail: three paths out of the hub, one of them unmetered ──────────────────

@pytest.mark.asyncio
async def test_changing_your_address_cannot_mail_strangers_at_will(
        client, monkeypatch):
    """
    `PATCH /v1/users/me` is the third path that makes the hub send mail, and
    it was the one with no rate limit and no captcha — while `register` and
    `password/reset-request` have both. The address is any string the caller
    types, and the duplicate check only rejects one already held by an account
    here, so every address *not* registered on this hub was a valid target.
    """
    sent: list = []
    import meshbay_hub.mail as mail_mod
    monkeypatch.setattr(mail_mod, "send_email_change_code",
                        lambda *a, **kw: sent.append(a))

    user = await _make_user(client, "av_mailer")
    headers = {"Authorization": f"Bearer {user['token']}"}

    ak = base64.b64encode(b"k" * 32).decode()   # the auth_key _make_user signs up with
    r = await client.patch("/v1/users/me", headers=headers,
                           json={"email": "a-stranger@example.test", "auth_key": ak})
    assert r.status_code == 200, r.text
    assert len(sent) == 1

    r = await client.patch("/v1/users/me", headers=headers,
                           json={"email": "another-stranger@example.test", "auth_key": ak})
    assert r.status_code == 429, r.text
    assert len(sent) == 1, "the hub mailed a second stranger on demand"


@pytest.mark.asyncio
async def test_a_reset_mail_lands_once_per_account_per_window(client, monkeypatch):
    """Knowing the username/email pair is the hard part, and this endpoint is
    careful about it. Once someone does, the cost of repeating lands in a
    mailbox that is not theirs — and the rate limit above counts by IP."""
    sent: list = []
    import meshbay_hub.mail as mail_mod
    monkeypatch.setattr(mail_mod, "send_password_reset_code",
                        lambda *a, **kw: sent.append(a))

    user = await _make_user(client, "av_resettee")
    body = {"username": user["username"], "email": "av_resettee@example.test"}

    for _ in range(3):
        r = await client.post("/v1/users/password/reset-request", json=body)
        assert r.status_code == 200, r.text
    assert len(sent) == 1, f"{len(sent)} reset mails for one account in one window"


def test_no_mail_is_sent_from_the_event_loop():
    """
    `smtplib` is synchronous and waits up to ten seconds. Called straight from
    an async handler — which is what all four call sites did — that wait is not
    one request's, it is the whole hub's: nothing else is served, no node
    socket is read, no offer relayed, until the MTA answers.

    Read from the source because the failure has no symptom a test can catch:
    everything works, slowly, for everyone, whenever the mail server is having
    a bad day.
    """
    import pathlib
    import re as _re

    root = pathlib.Path(__file__).resolve().parents[1] / "src" / "meshbay_hub"
    # A direct *call* — `mail.send_x(`. A bare `mail.send_x` with no paren is
    # the function being handed to send_off_loop, which is the point. The
    # first version of this matched those continuation lines and so failed on
    # the fixed code: look for the call, not for the name.
    direct_call = _re.compile(r"\bmail\.send_(?!off_loop)\w+\s*\(")
    offenders = []
    for path in root.rglob("*.py"):
        if path.name == "mail.py":
            continue
        for n, line in enumerate(path.read_text().splitlines(), 1):
            if direct_call.search(line):
                offenders.append(f"{path.name}:{n}: {line.strip()}")
    assert not offenders, (
        "these call a blocking SMTP send directly; use mail.send_off_loop:\n"
        + "\n".join(offenders))


# ── One account's rows are not the whole table ───────────────────────────────

@pytest.mark.asyncio
async def test_the_preference_namespace_is_not_open(client):
    """
    `default_tab:` accepted any suffix, on a `{key:path}` route, with an
    unbounded Text value: one account could write unbounded rows into a table
    shared with everyone. The suffix is a group id — that is what the SPA
    writes — so it is checked as one.
    """
    user = await _make_user(client, "av_prefs")
    headers = {"Authorization": f"Bearer {user['token']}"}
    gid = await _make_group(client, user, "prefs-group")

    r = await client.put(f"/v1/users/me/preferences/default_tab:{gid}",
                         headers=headers, json={"value": "files"})
    assert r.status_code == 200, r.text

    for bad in ("default_tab:" + "x" * 300, "default_tab:not-a-uuid",
                "default_tab:", "default_tab:../../etc"):
        r = await client.put(f"/v1/users/me/preferences/{bad}",
                             headers=headers, json={"value": "files"})
        assert r.status_code == 400, f"{bad!r} was accepted: {r.text}"

    r = await client.put(f"/v1/users/me/preferences/default_tab:{gid}",
                         headers=headers, json={"value": "f" * 5000})
    assert r.status_code == 422, r.text


@pytest.mark.asyncio
async def test_a_list_cannot_be_asked_for_the_whole_table(client):
    """Every list in admin.py carries `le=200`. These two did not — and the
    public group directory takes no authentication at all."""
    user = await _make_user(client, "av_lister")
    headers = {"Authorization": f"Bearer {user['token']}"}

    r = await client.get("/v1/notifications?limit=1000000", headers=headers)
    assert r.status_code == 422, r.text
    r = await client.get("/v1/notifications?limit=-1", headers=headers)
    assert r.status_code == 422, r.text

    r = await client.get("/v1/groups?limit=1000000")
    assert r.status_code == 422, r.text
    r = await client.get("/v1/groups?offset=-5")
    assert r.status_code == 422, r.text

    r = await client.get("/v1/notifications?limit=20", headers=headers)
    assert r.status_code == 200, r.text


# ── A node that hosts nothing is not a free target ──────────────────────────

@pytest.mark.asyncio
async def test_a_node_hosting_nothing_is_not_brokered_to_a_stranger(client):
    """
    Signaling read its membership check as `if node_group_ids:` — so when the
    set was empty, the membership check, the group-status check and the
    public-group gate were all skipped and the offer was relayed.

    Since AV1, an empty claim is the *normal* registration of a node that hosts
    nothing: the unconfigured node left running, which is the machine in this
    register's founding incident. Each offer makes it allocate an
    `RTCPeerConnection` and gather ICE, which is finding H6 restored in exactly
    the case AV1 made common — and the stranger paying nothing for it.
    """
    from meshbay_hub.api import revocation as rev

    owner = await _make_user(client, "av_sig_owner")
    stranger = await _make_user(client, "av_sig_stranger")
    node_id = await _announce_node(client, owner)

    class _FakeWS:
        def __init__(self):
            self.sent = []

        async def send_text(self, text):
            self.sent.append(text)

    ws = _FakeWS()
    rev._connected_nodes[node_id] = ws
    rev._node_groups[node_id] = []          # hosts nothing, as in the incident
    try:
        r = await client.post(f"/v1/nodes/{node_id}/webrtc/offer",
                              json={"sdp": "v=0\r\noffer", "ice_candidates": []},
                              headers={"Authorization": f"Bearer {stranger['token']}"})
        assert r.status_code == 403, r.text
        assert not ws.sent, (
            "the node was made to negotiate for someone with no group on it")
    finally:
        rev._connected_nodes.pop(node_id, None)
        rev._node_groups.pop(node_id, None)


@pytest.mark.asyncio
async def test_a_member_still_reaches_the_node_they_share_a_group_with(client):
    """The other half, so the test above is about the claim and not about
    refusing everyone."""
    from meshbay_hub.api import revocation as rev

    owner = await _make_user(client, "av_sig_owner2")
    member = await _make_user(client, "av_sig_member2")
    group_id = await _make_group(client, owner, "shared-one")
    await _add_member(client, owner, group_id, member)
    node_id = await _announce_node(client, owner)

    answered = []

    class _AnsweringWS:
        async def send_text(self, text):
            import json as _json

            from meshbay_hub.api.signaling import handle_webrtc_answer
            msg = _json.loads(text)
            answered.append(msg)
            handle_webrtc_answer({"peer_id": msg["peer_id"], "sdp": "v=0\r\nanswer",
                                  "ice_candidates": []}, node_id)

    rev._connected_nodes[node_id] = _AnsweringWS()
    rev._node_groups[node_id] = [group_id]
    try:
        r = await client.post(f"/v1/nodes/{node_id}/webrtc/offer",
                              json={"sdp": "v=0\r\noffer", "ice_candidates": []},
                              headers={"Authorization": f"Bearer {member['token']}"})
        assert r.status_code == 200, r.text
        assert answered
    finally:
        rev._connected_nodes.pop(node_id, None)
        rev._node_groups.pop(node_id, None)


# ── A private group's hosts are not public knowledge ────────────────────────

@pytest.mark.asyncio
async def test_a_private_groups_node_list_is_for_its_members(client):
    """
    `GET /v1/groups/{id}/nodes` checked membership only for a public group with
    public groups switched off. A private group answered any authenticated
    account that knew the id — which an ex-member knows for ever — with the ids
    and public keys of the machines hosting it.

    §7.4 already states the property for the public case: a non-member is handed
    no node to connect to. This is that sentence, for the groups the whole
    design optimises for.
    """
    from meshbay_hub.api import revocation as rev

    owner = await _make_user(client, "av_nodes_owner")
    member = await _make_user(client, "av_nodes_member")
    stranger = await _make_user(client, "av_nodes_stranger")
    group_id = await _make_group(client, owner, "private-hosts")
    await _add_member(client, owner, group_id, member)
    node_id = await _announce_node(client, owner)

    rev._node_groups[node_id] = [group_id]
    rev._connected_nodes[node_id] = object()
    try:
        for who in (owner, member):
            r = await client.get(
                f"/v1/groups/{group_id}/nodes",
                headers={"Authorization": f"Bearer {who['token']}"})
            assert r.status_code == 200, r.text
            assert [n["node_id"] for n in r.json()["nodes"]] == [node_id]

        r = await client.get(
            f"/v1/groups/{group_id}/nodes",
            headers={"Authorization": f"Bearer {stranger['token']}"})
        assert r.status_code == 403, (
            "a stranger who knows the group id learned which machines host it: "
            + r.text)
    finally:
        rev._connected_nodes.pop(node_id, None)
        rev._node_groups.pop(node_id, None)


async def _announce_key(client, user: dict, sk) -> int:
    """Announce a *distinct* node key, and return the status code."""
    from meshbay_common.crypto import pk_to_b64

    pk = pk_to_b64(sk.public_key())
    ts = int(time.time())
    msg = f"meshbay:node_announce:{user['user_id']}:{pk}:{ts}".encode()
    r = await client.post("/v1/nodes/announce", json={
        "pk_node": pk, "endpoint_hint": "test", "timestamp": ts,
        "signature": base64.b64encode(sk.sign(msg)).decode(),
    }, headers={"Authorization": f"Bearer {user['token']}"})
    return r.status_code


async def test_one_account_cannot_announce_unlimited_nodes(client, monkeypatch):
    """
    Each new node key is a row in `nodes` and a row in the IP log, and the IP log
    is kept for a year. Proof of possession (M8) settles *whose* key it is and
    says nothing about how many: an account in a loop wrote a year of storage on
    the operator's disk having paid only for signatures.

    Two accounts, because the ceiling has to be per account. One that is shared
    would let a single member deny every other member the ability to bring a
    machine online, which is the same defect with better manners.
    """
    from meshbay_hub.api import nodes as nodes_api

    monkeypatch.setattr(nodes_api, "MAX_NODES_PER_ACCOUNT", 3)
    alice = await _make_user(client, "av_nodecap_alice")
    bob = await _make_user(client, "av_nodecap_bob")

    keys = [Ed25519PrivateKey.generate() for _ in range(4)]
    for sk in keys[:3]:
        assert await _announce_key(client, alice, sk) == 201

    assert await _announce_key(client, alice, keys[3]) == 409, (
        "an account announced past the ceiling")

    # Bob has announced nothing and must be unaffected.
    assert await _announce_key(client, bob, Ed25519PrivateKey.generate()) == 201, (
        "one account's ceiling was charged to another's"
    )


async def test_a_node_at_the_ceiling_can_still_refresh_its_address(client, monkeypatch):
    """
    The ceiling counts rows, so it must be checked only where a row is added.
    Applied to every announce, it would freeze the address of every node an
    account already runs the moment it reached the limit — and a node that
    cannot re-announce is a node nobody can reach after their ISP renumbers
    them, which is an outage caused by the protection.
    """
    from meshbay_hub.api import nodes as nodes_api

    monkeypatch.setattr(nodes_api, "MAX_NODES_PER_ACCOUNT", 2)
    alice = await _make_user(client, "av_nodecap_refresh")

    keys = [Ed25519PrivateKey.generate() for _ in range(2)]
    for sk in keys:
        assert await _announce_key(client, alice, sk) == 201
    assert await _announce_key(client, alice, Ed25519PrivateKey.generate()) == 409

    for sk in keys:
        assert await _announce_key(client, alice, sk) == 201, (
            "a node already known could not re-announce at the ceiling")