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

"""
Invitation links, the hub's half (docs/MESHBAY_DESIGN.md §3.4, §7.3).

The ticket is what lets somebody reach the node, so each test here is a way it
could let in someone other than the one account it was meant for, tell the
inviter something the hub should not, or make the hub mail somebody it should
not. Every test has at least two accounts: a one-account test proves a
one-account property (CLAUDE.md, "ask who pays").
"""

import base64
import time
from datetime import UTC, datetime, timedelta

import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_hub import mail as mail_mod
from meshbay_hub.api import invite_links
from meshbay_hub.db.models import GroupInviteLink, GroupMember
from sqlalchemy import select

NODE_INVITE_ID = "ab" * 16
NODE_PK = "A" * 43
CODE = "K7P2-9WQX"


async def _account(client, username: str, email: str | None = None) -> dict:
    auth_key = base64.b64encode(b"k" * 32).decode()
    r = await client.post("/v1/users/register", json={
        "username": username, "email": email or f"{username}@example.test",
        "auth_key": auth_key})
    assert r.status_code == 201, r.text
    r = await client.post("/v1/users/login", json={"username": username, "auth_key": auth_key})
    assert r.status_code == 200, r.text
    return {"username": username,
            "h": {"Authorization": f"Bearer {r.json()['access_token']}"}}


async def _group(client, owner: dict, name: str = "family-photos", **extra) -> str:
    r = await client.post("/v1/groups", json={"name": name, **extra}, headers=owner["h"])
    assert r.status_code in (200, 201), r.text
    return r.json()["group_id"]


def _expires(days: float = 7) -> str:
    return (datetime.now(UTC) + timedelta(days=days)).isoformat(timespec="seconds")


async def _link(client, owner, gid, email="invitee@example.test", **extra):
    body = {"email": email, "expires_at": _expires(), "node_invite_id": NODE_INVITE_ID,
            **extra}
    return await client.post(f"/v1/groups/{gid}/invite-links", json=body, headers=owner["h"])


@pytest.fixture
def sent(monkeypatch):
    out = []
    monkeypatch.setattr(mail_mod, "_send", lambda msg, **_: out.append(msg) or True)
    return out


# ── Who gets in ──────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_the_addressed_account_joins_and_nobody_else(client, db_session):
    owner = await _account(client, "link_owner")
    gid = await _group(client, owner)
    r = await _link(client, owner, gid, email="Invitee@Example.test")
    assert r.status_code == 201, r.text
    ticket = r.json()["ticket"]

    mallory = await _account(client, "link_mallory")
    for route in ("preview", "redeem"):
        r = await client.post(f"/v1/invite-links/{route}", json={"ticket": ticket},
                              headers=mallory["h"])
        assert r.status_code == 403 and r.json()["detail"] == "invite_other_account"
        assert "invitee" not in r.text.lower(), "the refusal must not name the address"

    # Registered with the address the owner typed, case aside.
    invitee = await _account(client, "link_invitee", email="invitee@example.test")
    r = await client.post("/v1/invite-links/preview", json={"ticket": ticket},
                          headers=invitee["h"])
    assert r.status_code == 200, r.text
    assert r.json()["group_name"] == "family-photos" and r.json()["inviter"] == "link_owner"
    assert r.json()["already_member"] is False

    r = await client.post("/v1/invite-links/redeem", json={"ticket": ticket},
                          headers=invitee["h"])
    assert r.status_code == 200 and r.json()["group_id"] == gid
    # A member now — which the members list answers to members only. (Not
    # `/groups/mine`: a group no node has hosted yet is shown to its owner alone.)
    r = await client.get(f"/v1/groups/{gid}/members", headers=invitee["h"])
    assert r.status_code == 200
    assert "link_invitee" in {m["username"] for m in r.json()["members"]}

    # A second tab, or a reload, is the same person: same answer, no second row.
    r = await client.post("/v1/invite-links/redeem", json={"ticket": ticket},
                          headers=invitee["h"])
    assert r.status_code == 200
    rows = (await db_session.execute(select(GroupMember).where(
        GroupMember.group_id == gid))).scalars().all()
    assert len(rows) == 2

    # And the one who comes after, even with the right address, gets nothing:
    # a twin account cannot exist (addresses are unique), so try the other.
    r = await client.post("/v1/invite-links/redeem", json={"ticket": ticket},
                          headers=mallory["h"])
    assert r.status_code == 404


@pytest.mark.asyncio
async def test_a_ticket_is_stored_only_as_a_hash(client, db_session):
    owner = await _account(client, "hash_owner")
    gid = await _group(client, owner)
    ticket = (await _link(client, owner, gid)).json()["ticket"]
    row = (await db_session.execute(select(GroupInviteLink))).scalar_one()
    assert ticket not in (row.ticket_hash, row.email_masked, row.email_hash)
    assert row.ticket_hash == invite_links.ticket_hash(ticket)
    assert "invitee@" not in row.email_masked


@pytest.mark.asyncio
async def test_expired_cancelled_and_unknown_tickets_are_one_answer(client):
    owner = await _account(client, "dead_owner")
    invitee = await _account(client, "dead_invitee", email="invitee@example.test")
    gid = await _group(client, owner)

    cancelled = (await _link(client, owner, gid)).json()
    r = await client.delete(f"/v1/groups/{gid}/invite-links/{cancelled['link_id']}",
                            headers=owner["h"])
    assert r.status_code == 200

    for ticket in (cancelled["ticket"], "x" * 22, "not-a-ticket", ""):
        r = await client.post("/v1/invite-links/redeem", json={"ticket": ticket},
                              headers=invitee["h"])
        assert (r.status_code, r.json()["detail"]) == (404, "invite_not_valid")


@pytest.mark.asyncio
async def test_a_suspended_group_admits_nobody_by_link(client, db_session):
    from meshbay_hub.db.models import Group
    owner = await _account(client, "susp_owner")
    invitee = await _account(client, "susp_invitee", email="invitee@example.test")
    gid = await _group(client, owner)
    ticket = (await _link(client, owner, gid)).json()["ticket"]
    (await db_session.get(Group, gid)).status = "suspended"
    await db_session.commit()
    r = await client.post("/v1/invite-links/redeem", json={"ticket": ticket},
                          headers=invitee["h"])
    assert r.status_code == 404


# ── Who may issue ────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_only_the_owner_issues_lists_and_cancels(client):
    owner = await _account(client, "own_owner")
    other = await _account(client, "own_other")
    gid = await _group(client, owner)
    link = (await _link(client, owner, gid)).json()

    assert (await _link(client, other, gid)).status_code == 403
    assert (await client.get(f"/v1/groups/{gid}/invite-links",
                             headers=other["h"])).status_code == 403
    assert (await client.delete(f"/v1/groups/{gid}/invite-links/{link['link_id']}",
                                headers=other["h"])).status_code == 403

    r = await client.get(f"/v1/groups/{gid}/invite-links", headers=owner["h"])
    [row] = r.json()["links"]
    assert row["status"] == "pending" and row["email"] == "in***@ex***.test"
    assert row["node_invite_id"] == NODE_INVITE_ID


@pytest.mark.asyncio
async def test_creating_a_link_says_nothing_about_whether_the_address_has_an_account(client):
    """M1: the same answer for a registered address and a stranger's."""
    owner = await _account(client, "m1_owner")
    await _account(client, "m1_known", email="known@example.test")
    gid = await _group(client, owner)
    a = (await _link(client, owner, gid, email="known@example.test")).json()
    b = (await _link(client, owner, gid, email="nobody@example.test")).json()
    assert a.keys() == b.keys()
    assert a["email_status"] == b["email_status"] == "not_requested"


@pytest.mark.asyncio
async def test_links_are_capped_per_group(client):
    owner = await _account(client, "cap_owner")
    gid = await _group(client, owner)
    for _ in range(invite_links.MAX_OUTSTANDING_PER_GROUP):
        assert (await _link(client, owner, gid)).status_code == 201
    assert (await _link(client, owner, gid)).status_code == 429
    other = await _group(client, owner, name="another-group")
    assert (await _link(client, owner, other)).status_code == 201


@pytest.mark.asyncio
async def test_an_open_group_needs_no_link(client):
    owner = await _account(client, "open_owner")
    gid = await _group(client, owner, visibility="public", join_policy="open")
    assert (await _link(client, owner, gid)).status_code == 409


@pytest.mark.asyncio
async def test_a_lifetime_is_clamped_and_a_past_one_refused(client, db_session):
    owner = await _account(client, "ttl_owner")
    gid = await _group(client, owner)
    r = await _link(client, owner, gid, expires_at=_expires(days=-1))
    assert r.status_code == 422
    r = await _link(client, owner, gid, expires_at=_expires(days=400))
    expires = datetime.fromisoformat(r.json()["expires_at"])
    assert expires <= datetime.now(UTC) + invite_links.MAX_LIFETIME


@pytest.mark.asyncio
async def test_a_used_link_cannot_be_cancelled_here(client):
    owner = await _account(client, "used_owner")
    invitee = await _account(client, "used_invitee", email="invitee@example.test")
    gid = await _group(client, owner)
    link = (await _link(client, owner, gid)).json()
    await client.post("/v1/invite-links/redeem", json={"ticket": link["ticket"]},
                      headers=invitee["h"])
    r = await client.delete(f"/v1/groups/{gid}/invite-links/{link['link_id']}",
                            headers=owner["h"])
    assert r.status_code == 409


@pytest.mark.asyncio
async def test_a_used_link_leaves_the_owners_list(client, db_session):
    """The invitee is a member now; the link saying so as well is clutter.

    The row itself stays for `KEEP_REDEEMED`, which is what lets a reload of
    the invitation page answer the account that used it instead of refusing.
    """
    owner = await _account(client, "gone_owner")
    invitee = await _account(client, "gone_invitee", email="invitee@example.test")
    gid = await _group(client, owner)
    waiting = (await _link(client, owner, gid, email="other@example.test")).json()
    link = (await _link(client, owner, gid)).json()

    r = await client.get(f"/v1/groups/{gid}/invite-links", headers=owner["h"])
    assert {row["link_id"] for row in r.json()["links"]} == {waiting["link_id"],
                                                            link["link_id"]}

    r = await client.post("/v1/invite-links/redeem", json={"ticket": link["ticket"]},
                          headers=invitee["h"])
    assert r.status_code == 200, r.text

    r = await client.get(f"/v1/groups/{gid}/invite-links", headers=owner["h"])
    rows = r.json()["links"]
    assert [row["link_id"] for row in rows] == [waiting["link_id"]]
    assert "redeemed" not in r.text
    # Still on the hub, so the invitee's second tab is answered, not refused.
    assert (await db_session.get(GroupInviteLink, link["link_id"])) is not None
    r = await client.post("/v1/invite-links/preview", json={"ticket": link["ticket"]},
                          headers=invitee["h"])
    assert r.status_code == 200 and r.json()["already_member"] is True


# ── What the hub mails ───────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_the_mail_carries_the_hubs_own_link_and_nothing_chosen(client, sent):
    owner = await _account(client, "mail_owner")
    gid = await _group(client, owner)
    r = await _link(client, owner, gid, send_email=True, node_pk=NODE_PK, code=CODE)
    assert r.json()["email_status"] == "sent"
    [msg] = sent
    body = msg.get_content()
    assert msg["To"] == "invitee@example.test"
    expected = invite_links.invite_url(gid, r.json()["ticket"], NODE_PK, CODE)
    assert expected in body and expected.startswith(mail_mod.hub_url() + "/#/invite?")
    assert "family-photos" in msg["Subject"] and "mail_owner" in msg["Subject"]


@pytest.mark.asyncio
async def test_no_box_no_mail(client, sent):
    owner = await _account(client, "nomail_owner")
    gid = await _group(client, owner)
    r = await _link(client, owner, gid, node_pk=NODE_PK, code=CODE)
    assert r.json()["email_status"] == "not_requested" and sent == []


@pytest.mark.asyncio
async def test_a_mail_needs_a_well_formed_code_and_key(client, sent):
    owner = await _account(client, "shape_owner")
    gid = await _group(client, owner)
    for bad in ({"node_pk": NODE_PK, "code": "https://elsewhere.example/"},
                {"node_pk": "../../x", "code": CODE}, {}):
        r = await _link(client, owner, gid, send_email=True, **bad)
        assert r.status_code == 422, bad
    assert sent == []


@pytest.mark.asyncio
async def test_link_mail_is_capped_per_sender_and_leaves_other_mail_alone(client, sent):
    """
    AV10's shape: a link reaches addresses the hub has no relationship with, so
    the account asking is counted. The eleventh link of the day is still made
    and shown — only the mail is refused — and nobody else's mail is touched.
    """
    from meshbay_hub import hub_settings
    owner = await _account(client, "cap_mailer")
    cap = hub_settings.mail_default("invite_link_daily_cap")
    assert cap == 10
    groups = [await _group(client, owner, name=f"g-{i}") for i in range(2)]
    statuses = []
    for i in range(cap + 1):
        r = await _link(client, owner, groups[i // 10], email=f"friend{i}@example.test",
                        send_email=True, node_pk=NODE_PK, code=CODE)
        assert r.status_code == 201, r.text
        statuses.append(r.json()["email_status"])
    assert statuses == ["sent"] * cap + ["refused"]

    other = await _account(client, "cap_other")
    gid = await _group(client, other, name="theirs")
    r = await _link(client, other, gid, email="friend0b@example.test",
                    send_email=True, node_pk=NODE_PK, code=CODE)
    assert r.json()["email_status"] == "sent"


def test_invite_link_is_its_own_mail_purpose_and_not_a_recovery_one():
    assert "invite_link" in mail_mod.ALLOWED_PURPOSES
    assert "invite_link" not in mail_mod.RECOVERY_PURPOSES


# ── The CLI's door ───────────────────────────────────────────────────────────

async def _node_token(client, owner: dict) -> dict:
    sk = Ed25519PrivateKey.generate()
    pk = base64.b64encode(sk.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw)).decode()
    r = await client.put("/v1/users/me/node_key", json={"pk_node_ed25519": pk},
                         headers=owner["h"])
    assert r.status_code == 200, r.text
    ts = int(time.time())
    r = await client.post("/v1/nodes/auth", json={
        "username": owner["username"], "timestamp": ts,
        "signature": base64.b64encode(
            sk.sign(f"meshbay:node_auth:{owner['username']}:{ts}".encode())).decode()})
    assert r.status_code == 200, r.text
    return {"h": {"Authorization": f"Bearer {r.json()['access_token']}"}}


@pytest.mark.asyncio
async def test_a_node_may_issue_for_its_operator_but_never_mail(client, sent):
    owner = await _account(client, "cli_owner")
    gid = await _group(client, owner)
    node = await _node_token(client, owner)
    r = await _link(client, node, gid)
    assert r.status_code == 201, r.text
    r = await _link(client, node, gid, send_email=True, node_pk=NODE_PK, code=CODE)
    assert r.status_code == 403 and sent == []

    other = await _account(client, "cli_other")
    theirs = await _group(client, other, name="not-yours")
    assert (await _link(client, node, theirs)).status_code == 403
    # Redeeming is a person's act, never a node's.
    r = await client.post("/v1/invite-links/preview", json={"ticket": "x" * 22},
                          headers=node["h"])
    assert r.status_code == 403