feat(hub): invitation-link tickets bound to a verified address

group_invite_links holds sha256(ticket) and the invitee's address blind
index; redeeming grants membership to that account only. Owner-only
create/list/cancel (a node token may create, never mail), 20 outstanding
per group, optional mail written by the hub itself and capped at 10 per
sender per day (mail.invite_link_daily_cap). MESHBAY_DESIGN.md §3.4 now
carries the whole link design.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

2 files changed, 340 insertions, 1 deletions

diff --git a/packages/meshbay-hub/tests/test_group_purge.py b/packages/meshbay-hub/tests/test_group_purge.py
index e729297..9c6a664 100644
--- a/packages/meshbay-hub/tests/test_group_purge.py
+++ b/packages/meshbay-hub/tests/test_group_purge.py
@@ -25,6 +25,7 @@ from meshbay_hub.db.models import (
     ContentReport,
     EmailVerification,
     Group,
+    GroupInviteLink,
     GroupMember,
     IPLog,
     Notification,
@@ -34,7 +35,8 @@ from meshbay_hub.db.purge import _referencing
 from sqlalchemy import delete, func, select, text
 from sqlalchemy.exc import IntegrityError
 
-SEEDED = {"group_members", "notifications", "email_verifications", "content_reports"}
+SEEDED = {"group_members", "notifications", "email_verifications", "content_reports",
+          "group_invite_links"}
 
 
 def _auth_key(password: str, username: str) -> str:
@@ -80,6 +82,11 @@ async def _group_with_everything(client, db, owner_token: str, member: str, name
                              expires_at=datetime.now(UTC) + timedelta(days=1)))
     db.add(ContentReport(reporter_id=member_id, content_hash="ab" * 32, group_id=gid,
                          ip_address="192.0.2.1"))
+    owner_id = (await db.execute(select(Group.admin_id).where(Group.id == gid))).scalar_one()
+    db.add(GroupInviteLink(group_id=gid, created_by=owner_id, ticket_hash=gid[:8] * 8,
+                           email_hash="1" * 64, email_masked="m***@e***.com",
+                           expires_at=datetime.now(UTC) + timedelta(days=1),
+                           redeemed_by=member_id, redeemed_at=datetime.now(UTC)))
     await db.commit()
     return gid
 
diff --git a/packages/meshbay-hub/tests/test_invite_links.py b/packages/meshbay-hub/tests/test_invite_links.py
new file mode 100644
index 0000000..1b60dc1
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_invite_links.py
@@ -0,0 +1,332 @@
+"""
+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
+    r = await client.get(f"/v1/groups/{gid}/invite-links", headers=owner["h"])
+    assert r.json()["links"][0]["redeemed_by"] == "used_invitee"
+
+
+# ── 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