invites: an admin lets one person register while registration stays closed

Until now adding a second person meant re-opening registration to the
whole internet while they signed up (#2939 §1). An admin now makes an
invite in Settings: a link that works once, expires (7 days by default,
1 to 30), and can be pinned to one email address. Only the token's hash
is stored, so the link is shown once.

POST /api/auth/register takes `invite`. Redemption is one conditional
UPDATE inside the transaction that creates the account, so two people
racing one link can't both get in, and a taken email leaves the invite
unused. Every refusal says "invalid or expired invite". The register
page reads ?invite= and opens even while registration is closed.

Refs #5172

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

tests/test_integration.py

@@ -15,18 +15,20 @@ deployment has ever seen.
 """
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import uuid
 from datetime import datetime, timedelta, timezone
 
 import pytest
 import pytest_asyncio
-from sqlalchemy import func, select, text
+from sqlalchemy import func, select, text, update
 
 from inkwell import ratelimit
 from inkwell.app import create_app
 from inkwell.config import Config
 from inkwell.db import dispose_engine, session_scope
+from inkwell.models.invite import Invite
 from inkwell.models.label import NoteLabel
 from inkwell.models.note import Note
 from inkwell.models.note_attachment import NoteAttachment
@@ -44,7 +46,7 @@ pytestmark = pytest.mark.integration
 
 # Every table the tests touch, child-first so FKs never block the truncate.
 # RESTART IDENTITY + CASCADE keeps this honest if a table gains children later.
-_TABLES = "notes, note_revisions, note_labels, note_link_previews, labels, users"
+_TABLES = "notes, note_revisions, note_labels, note_link_previews, labels, invites, users"
 
 
 @pytest_asyncio.fixture
@@ -390,8 +392,8 @@ async def test_registration_closes_itself_once_an_admin_exists(app_client, db):
     )
     assert second.status_code == 403
 
-    # Re-opening it deliberately still works — that is how a second person gets in
-    # until invites exist.
+    # Re-opening it deliberately still works, for an admin who would rather open the
+    # door than send an invite.
     async with session_scope() as fresh:
         await set_settings(fresh, {"allow_registration": True})
         await fresh.commit()
@@ -932,3 +934,133 @@ async def test_a_keep_note_that_is_only_a_photo_imports(app_client, db):
     [note] = (await db.scalars(select(Note))).all()
     [att] = (await db.scalars(select(NoteAttachment).where(NoteAttachment.note_id == note.id))).all()
     assert att.mime == "image/png"
+
+
+# ---- invites (#5172) ---------------------------------------------------------------
+
+_PASSWORD = "a-long-enough-password"
+
+
+async def _admin_with_invite(app_client, **body) -> str:
+    """Register the instance's first account (the admin, signed in on `app_client`)
+    and have it issue an invite. Returns the token."""
+    created = await app_client.post("/api/auth/register", json={"email": "owner@example.test", "password": _PASSWORD})
+    assert created.status_code == 201
+    resp = await app_client.post("/api/invites", json=body)
+    assert resp.status_code == 201, await resp.get_data(as_text=True)
+    return (await resp.get_json())["token"]
+
+
+async def _register(email: str, invite: str | None = None):
+    """Register from a separate client, so the admin's session stays where it is."""
+    body = {"email": email, "password": _PASSWORD}
+    if invite is not None:
+        body["invite"] = invite
+    return await create_app().test_client().post("/api/auth/register", json=body)
+
+
+async def test_an_invite_lets_one_person_in_while_registration_stays_closed(app_client, db):
+    token = await _admin_with_invite(app_client)
+
+    # Registration closed itself behind the admin; a stranger with no invite is out.
+    stranger = await _register("stranger@example.test")
+    assert stranger.status_code == 403
+
+    invited = await _register("guest@example.test", token)
+    assert invited.status_code == 201, await invited.get_data(as_text=True)
+    assert (await invited.get_json())["is_admin"] is False
+
+    # Single use: the same link again, for anyone, is refused with the generic answer.
+    again = await _register("someone-else@example.test", token)
+    assert again.status_code == 403
+    assert (await again.get_json())["error"] == "invalid or expired invite"
+
+    # The admin's list says who used it.
+    listed = (await (await app_client.get("/api/invites")).get_json())["invites"]
+    assert [(i["status"], i["redeemed_by_email"]) for i in listed] == [("redeemed", "guest@example.test")]
+    async with session_scope() as fresh:
+        assert await get_setting(fresh, "allow_registration") is False
+
+
+async def test_only_an_admin_handles_invites(app_client, db):
+    token = await _admin_with_invite(app_client)
+    guest = create_app().test_client()
+    joined = await guest.post(
+        "/api/auth/register", json={"email": "guest@example.test", "password": _PASSWORD, "invite": token}
+    )
+    assert joined.status_code == 201
+    # Signed in as the guest now, who is not an admin.
+    assert (await guest.post("/api/invites", json={})).status_code == 403
+    assert (await guest.get("/api/invites")).status_code == 403
+
+
+async def test_an_invite_pinned_to_an_email_admits_only_that_email(app_client, db):
+    token = await _admin_with_invite(app_client, email="Pinned@Example.test")
+
+    wrong = await _register("other@example.test", token)
+    assert wrong.status_code == 403
+    assert (await wrong.get_json())["error"] == "invalid or expired invite"
+
+    # Any capitalisation of the pinned address, since emails compare case-insensitively.
+    right = await _register("PINNED@example.test", token)
+    assert right.status_code == 201
+
+
+async def test_revoked_and_expired_invites_are_refused(app_client, db):
+    revoked = await _admin_with_invite(app_client)
+    expiring = (await (await app_client.post("/api/invites", json={"days": 1})).get_json())["token"]
+
+    listed = (await (await app_client.get("/api/invites")).get_json())["invites"]
+    by_status = {i["status"] for i in listed}
+    assert by_status == {"pending"}
+    # The invite made first is the last in the list (newest first).
+    revoked_id = listed[-1]["id"]
+    resp = await app_client.delete(f"/api/invites/{revoked_id}")
+    assert resp.status_code == 200
+    assert (await resp.get_json())["status"] == "revoked"
+
+    async with session_scope() as fresh:
+        await fresh.execute(
+            update(Invite)
+            .where(Invite.id != uuid.UUID(revoked_id))
+            .values(expires_at=datetime.now(timezone.utc) - timedelta(minutes=1))
+        )
+        await fresh.commit()
+
+    assert (await _register("a@example.test", revoked)).status_code == 403
+    assert (await _register("b@example.test", expiring)).status_code == 403
+    statuses = sorted(i["status"] for i in (await (await app_client.get("/api/invites")).get_json())["invites"])
+    assert statuses == ["expired", "revoked"]
+
+
+async def test_an_invite_lifetime_outside_the_bounds_is_refused(app_client, db):
+    await _admin_with_invite(app_client)
+    for days in (0, 31, "a week", True):
+        resp = await app_client.post("/api/invites", json={"days": days})
+        assert resp.status_code == 400, days
+    resp = await app_client.post("/api/invites", json={"email": "not-an-address"})
+    assert resp.status_code == 400
+
+
+async def test_a_taken_email_leaves_the_invite_unused(app_client, db):
+    """The redemption and the new account are one transaction: an account that can't
+    be created must not use up the link."""
+    token = await _admin_with_invite(app_client)
+    taken = await _register("owner@example.test", token)
+    assert taken.status_code == 409
+    listed = (await (await app_client.get("/api/invites")).get_json())["invites"]
+    assert listed[0]["status"] == "pending"
+    assert (await _register("guest@example.test", token)).status_code == 201
+
+
+async def test_two_people_racing_one_invite_cannot_both_get_in(app_client, db):
+    token = await _admin_with_invite(app_client)
+    results = await asyncio.gather(
+        _register("first@example.test", token),
+        _register("second@example.test", token),
+    )
+    assert sorted(r.status_code for r in results) == [201, 403]
+    async with session_scope() as fresh:
+        count = await fresh.scalar(select(func.count()).select_from(User))
+    assert count == 2, "the admin and exactly one of the two"
+