path: root/packages/meshbay-hub/tests/test_availability_between_members.py

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

    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_a_captured_relay_registration_is_not_replayable(client):
    """Same reason /v1/nodes/announce bounds its timestamp."""
    from meshbay_hub.api import relay as relay_mod

    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']}"}

    r = await client.patch("/v1/users/me", headers=headers,
                           json={"email": "a-stranger@example.test"})
    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"})
    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