lukasandClaude Opus 5.5
463f96ad57
- Invitations: each user can invite two friends by email (sign-up works through the link even while registration is closed); admin-editable template, subject and sender. - Invitation sender defaults to the system-mail From address, which the SMTP account is allowed to send as (was postmaster@APP_DOMAIN, refused with 553 when the two differ). - Sign-in, sign-up and forgot-password fields show the fixed @domain after the typed name; the registration status reports it. - System mail gets a Date header; the invitation's plain-text part keeps its paragraphs and line breaks. - SMTP EHLO uses SMTP_HELO_NAME / APP_DOMAIN, not the container name. - DNS recommendations for the mail domains. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
350 lines
14 KiB
Python
350 lines
14 KiB
Python
"""Invite a friend: each user sends up to two single-use sign-up links, which
|
|
work even while registration is closed (app/identity/invitations.py).
|
|
|
|
Runs against the dev database; the settings row is put back as it was and every
|
|
user made here is removed on the way out. Mail is captured, never sent.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import datetime as dt
|
|
import re
|
|
import uuid
|
|
from collections.abc import AsyncIterator
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import ASGITransport
|
|
from sqlalchemy import delete, select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
from app.api import invitations as invitations_api
|
|
from app.config import get_settings
|
|
from app.identity import invitations
|
|
from app.identity.mailer import MailerError
|
|
from app.identity.passwords import hash_password
|
|
from app.identity.sessions import create_session
|
|
from app.main import create_app
|
|
from app.models import Account, AppSettings, Invitation, User
|
|
|
|
CSRF = "csrf-token"
|
|
_SETTINGS = (
|
|
"registration_open",
|
|
"invitations_enabled",
|
|
"smtp_host",
|
|
"smtp_from",
|
|
"public_url",
|
|
"invite_from",
|
|
"invite_subject",
|
|
"invite_template",
|
|
)
|
|
|
|
|
|
class World:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
self.created: list[uuid.UUID] = []
|
|
self.handles: list[str] = []
|
|
self.sent: list[dict[str, Any]] = []
|
|
|
|
async def user(self, *, admin: bool = False, name: str = "") -> User:
|
|
handle = f"inv{uuid.uuid4().hex[:8]}@test.local"
|
|
user = User(handle=handle, password_hash=hash_password("pw"), is_admin=admin)
|
|
self.session.add(user)
|
|
await self.session.flush()
|
|
if name:
|
|
self.session.add(
|
|
Account(
|
|
user_id=user.id,
|
|
email=handle,
|
|
display_name=name,
|
|
imap_config={},
|
|
smtp_config={},
|
|
secret_ct=b"x",
|
|
discovery_src="manual",
|
|
)
|
|
)
|
|
await self.session.commit()
|
|
self.created.append(user.id)
|
|
return user
|
|
|
|
async def client(self, user: User | None = None) -> httpx.AsyncClient:
|
|
cookies = {"csrf": CSRF}
|
|
if user is not None:
|
|
cookies["session"] = await create_session(user.id, "pytest")
|
|
return httpx.AsyncClient(
|
|
transport=ASGITransport(app=create_app()),
|
|
base_url="http://testserver",
|
|
cookies=cookies,
|
|
headers={"X-CSRF": CSRF},
|
|
)
|
|
|
|
async def settings(self, **values: Any) -> None:
|
|
await self.session.execute(update(AppSettings).where(AppSettings.id == 1).values(**values))
|
|
await self.session.commit()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def world(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[World]:
|
|
engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
|
|
session = async_sessionmaker(engine, expire_on_commit=False)()
|
|
row = await session.scalar(select(AppSettings).where(AppSettings.id == 1))
|
|
assert row is not None
|
|
saved = {name: getattr(row, name) for name in _SETTINGS}
|
|
w = World(session)
|
|
|
|
async def capture(cfg, to, subject, text, html, **kw): # type: ignore[no-untyped-def]
|
|
w.sent.append({"to": to, "subject": subject, "text": text, "html": html, **kw})
|
|
|
|
monkeypatch.setattr(invitations_api, "send_mail", capture)
|
|
await w.settings(
|
|
registration_open=False,
|
|
invitations_enabled=True,
|
|
smtp_host="smtp.example.com",
|
|
smtp_from="system@example.com",
|
|
public_url="https://mail.example.com",
|
|
invite_from="",
|
|
invite_subject="",
|
|
invite_template="",
|
|
)
|
|
try:
|
|
yield w
|
|
finally:
|
|
await session.rollback()
|
|
await w.settings(**saved)
|
|
await session.execute(delete(User).where(User.id.in_(w.created)))
|
|
if w.handles:
|
|
await session.execute(delete(User).where(User.handle.in_(w.handles)))
|
|
await session.commit()
|
|
await session.close()
|
|
await engine.dispose()
|
|
|
|
|
|
def _token(mail: dict[str, Any]) -> str:
|
|
m = re.search(r"#invite=([A-Za-z0-9_-]+)", mail["text"])
|
|
assert m is not None, mail["text"]
|
|
return m.group(1)
|
|
|
|
|
|
async def _register(w: World, token: str | None) -> httpx.Response:
|
|
localpart = f"new{uuid.uuid4().hex[:8]}"
|
|
w.handles.append(f"{localpart}@{get_settings().app_domain}")
|
|
body: dict[str, Any] = {"localpart": localpart, "password": "correct horse battery"}
|
|
if token is not None:
|
|
body["invite"] = token
|
|
async with await w.client() as anon:
|
|
return await anon.post("/api/auth/register", json=body)
|
|
|
|
|
|
async def test_two_invitations_then_none(world: World) -> None:
|
|
inviter = await world.user(name="Jan Kowalski")
|
|
async with await world.client(inviter) as c:
|
|
state = (await c.get("/api/invitations")).json()
|
|
assert state["available"] is True and state["remaining"] == 2
|
|
|
|
for i, name in enumerate(["Anna", "Piotr"]):
|
|
resp = await c.post(
|
|
"/api/invitations", json={"name": name, "email": f"f{i}@example.org"}
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["remaining"] == 0
|
|
assert [s["invitee_name"] for s in resp.json()["sent"]] == ["Piotr", "Anna"]
|
|
|
|
third = await c.post("/api/invitations", json={"name": "Ola", "email": "o@example.org"})
|
|
assert third.status_code == 409
|
|
assert third.json()["error"]["code"] == "no_invitations_left"
|
|
|
|
mail = world.sent[0]
|
|
assert mail["to"] == "f0@example.org"
|
|
assert mail["sender"] == "system@example.com"
|
|
assert mail["reply_to"] == inviter.handle
|
|
assert mail["subject"] == "Jan Kowalski invites you to Ultimo Email"
|
|
assert "Hi Anna" in mail["html"] and "https://mail.example.com/register#invite=" in mail["html"]
|
|
|
|
|
|
async def test_an_invitation_signs_up_once_while_registration_is_closed(world: World) -> None:
|
|
inviter = await world.user()
|
|
async with await world.client(inviter) as c:
|
|
await c.post("/api/invitations", json={"name": "Anna", "email": "anna@example.org"})
|
|
token = _token(world.sent[0])
|
|
|
|
closed = await _register(world, None)
|
|
assert closed.status_code == 403 # registration itself stays closed
|
|
|
|
async with await world.client() as anon:
|
|
check = (await anon.post("/api/auth/invitation/check", json={"token": token})).json()
|
|
assert check == {"valid": True, "invitee_name": "Anna", "inviter_name": inviter.handle}
|
|
|
|
first = await _register(world, token)
|
|
assert first.status_code == 200, first.text
|
|
again = await _register(world, token)
|
|
assert again.status_code == 400
|
|
assert again.json()["error"]["code"] == "invalid_invitation"
|
|
|
|
row = await world.session.scalar(select(Invitation).where(Invitation.inviter_id == inviter.id))
|
|
assert row is not None and row.used_at is not None and row.used_by is not None
|
|
|
|
|
|
async def test_a_failed_sign_up_keeps_the_invitation(world: World) -> None:
|
|
inviter = await world.user()
|
|
async with await world.client(inviter) as c:
|
|
await c.post("/api/invitations", json={"name": "Anna", "email": "anna@example.org"})
|
|
token = _token(world.sent[0])
|
|
weak = await world.client()
|
|
async with weak:
|
|
resp = await weak.post(
|
|
"/api/auth/register",
|
|
json={"localpart": "taken" + uuid.uuid4().hex[:6], "password": "password123"},
|
|
)
|
|
assert resp.status_code == 400 # refused password: nothing used
|
|
taken = await _register(world, token)
|
|
assert taken.status_code == 200
|
|
# Already used now; but a sign-up that failed on a taken address must not use it:
|
|
second_inviter = await world.user()
|
|
async with await world.client(second_inviter) as c:
|
|
await c.post("/api/invitations", json={"name": "B", "email": "b@example.org"})
|
|
token2 = _token(world.sent[-1])
|
|
async with await world.client() as anon:
|
|
dup = await anon.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"localpart": world.handles[-1].split("@")[0],
|
|
"password": "correct horse battery",
|
|
"invite": token2,
|
|
},
|
|
)
|
|
assert dup.status_code == 409
|
|
assert (await invitations.find_valid(world.session, token2)) is not None
|
|
|
|
|
|
async def test_switched_off_means_no_sending_and_no_signing_up(world: World) -> None:
|
|
inviter = await world.user()
|
|
async with await world.client(inviter) as c:
|
|
await c.post("/api/invitations", json={"name": "Anna", "email": "anna@example.org"})
|
|
token = _token(world.sent[0])
|
|
await world.settings(invitations_enabled=False)
|
|
assert (await c.get("/api/invitations")).json()["available"] is False
|
|
refused = await c.post("/api/invitations", json={"name": "B", "email": "b@example.org"})
|
|
assert refused.status_code == 403
|
|
assert (await _register(world, token)).status_code == 400
|
|
|
|
|
|
async def test_a_mail_that_could_not_go_out_is_given_back(
|
|
world: World, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
async def failing(*args, **kwargs): # type: ignore[no-untyped-def]
|
|
raise MailerError("relay refused")
|
|
|
|
monkeypatch.setattr(invitations_api, "send_mail", failing)
|
|
inviter = await world.user()
|
|
async with await world.client(inviter) as c:
|
|
resp = await c.post("/api/invitations", json={"name": "Anna", "email": "a@example.org"})
|
|
assert resp.status_code == 502
|
|
assert (await c.get("/api/invitations")).json()["remaining"] == 2
|
|
|
|
|
|
async def test_an_expired_unused_invitation_is_given_back(world: World) -> None:
|
|
inviter = await world.user()
|
|
async with await world.client(inviter) as c:
|
|
await c.post("/api/invitations", json={"name": "Anna", "email": "a@example.org"})
|
|
await world.session.execute(
|
|
update(Invitation)
|
|
.where(Invitation.inviter_id == inviter.id)
|
|
.values(expires_at=dt.datetime.now(dt.UTC) - dt.timedelta(minutes=1))
|
|
)
|
|
await world.session.commit()
|
|
state = (await c.get("/api/invitations")).json()
|
|
assert state["remaining"] == 2 and state["sent"][0]["expired"] is True
|
|
assert (await _register(world, _token(world.sent[0]))).status_code == 400
|
|
|
|
|
|
async def test_bad_input_is_refused(world: World) -> None:
|
|
inviter = await world.user()
|
|
async with await world.client(inviter) as c:
|
|
bad = await c.post("/api/invitations", json={"name": "A", "email": "not-an-address"})
|
|
assert bad.status_code == 400
|
|
blank = await c.post("/api/invitations", json={"name": " ", "email": "a@example.org"})
|
|
assert blank.status_code == 400
|
|
|
|
|
|
# --- the email ------------------------------------------------------------------------
|
|
|
|
|
|
def test_names_are_escaped_and_placeholders_filled() -> None:
|
|
row = AppSettings(
|
|
invite_subject="{{inviter_name}} → {{invitee_name}}",
|
|
invite_template="<p>Hi {{ invitee_name }}</p><p>{{inviter_name}}</p>{{invite_button}}",
|
|
)
|
|
mail = invitations.render(
|
|
row, invitee_name="<b>Eve</b>", inviter_name='Bob "x"', link="https://m.example/r#invite=t"
|
|
)
|
|
assert "<b>Eve</b>" in mail.html and "<b>Eve" not in mail.html
|
|
assert 'href="https://m.example/r#invite=t"' in mail.html
|
|
assert mail.subject == 'Bob "x" → <b>Eve</b>' # a header, not HTML
|
|
assert "https://m.example/r#invite=t" in mail.text and "<p>" not in mail.text
|
|
|
|
|
|
def test_template_images_travel_inside_the_mail() -> None:
|
|
png = base64.b64encode(b"\x89PNG\r\n\x1a\nfake").decode()
|
|
row = AppSettings(invite_template=f'<img src="data:image/png;base64,{png}">{{{{invite_link}}}}')
|
|
mail = invitations.render(row, invitee_name="A", inviter_name="B", link="https://x/y")
|
|
assert len(mail.inline) == 1 and "data:image" not in mail.html
|
|
assert f'src="cid:{mail.inline[0].cid}"' in mail.html
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"html,ok",
|
|
[
|
|
("<p>{{invite_link}}</p>", True),
|
|
("<p>{{invite_button}}</p>", True),
|
|
("<p>Hello {{invitee_name}}</p>", False), # no way to accept
|
|
("<p>{{invite_link}} {{password}}</p>", False), # unknown placeholder
|
|
],
|
|
)
|
|
def test_template_checks(html: str, ok: bool) -> None:
|
|
assert (invitations.template_problem(html) is None) is ok
|
|
|
|
|
|
async def test_admin_uploads_previews_and_resets_the_template(world: World) -> None:
|
|
admin = await world.user(admin=True)
|
|
async with await world.client(admin) as c:
|
|
bad = await c.put("/api/admin/invitations/template", json={"html": "<p>hi</p>"})
|
|
assert bad.status_code == 400
|
|
saved = await c.put(
|
|
"/api/admin/invitations/template",
|
|
json={
|
|
"html": "<p>Dear {{invitee_name}}, {{invite_button}}</p>",
|
|
"subject": "Join {{inviter_name}}",
|
|
"from_address": "postmaster@example.org",
|
|
},
|
|
)
|
|
assert saved.status_code == 200
|
|
assert saved.json()["from_address"] == "postmaster@example.org"
|
|
preview = await c.post(
|
|
"/api/admin/invitations/preview",
|
|
json={"subject": saved.json()["subject"], "html": saved.json()["html"]},
|
|
)
|
|
assert preview.json()["subject"] == "Join Jan Kowalski"
|
|
assert "Dear Anna Nowak" in preview.json()["html"]
|
|
reset = await c.put(
|
|
"/api/admin/invitations/template",
|
|
json={"html": "", "subject": "", "from_address": ""},
|
|
)
|
|
assert reset.json()["html"] == "" and reset.json()["default_html"]
|
|
user = await world.user()
|
|
async with await world.client(user) as c:
|
|
assert (await c.get("/api/admin/invitations/template")).status_code == 403
|
|
|
|
|
|
async def test_admin_switch(world: World) -> None:
|
|
admin = await world.user(admin=True)
|
|
async with await world.client(admin) as c:
|
|
off = await c.put("/api/admin/settings", json={"invitations_enabled": False})
|
|
assert off.json()["invitations_enabled"] is False
|
|
on = await c.put("/api/admin/settings", json={"invitations_enabled": True})
|
|
assert on.json()["invitations_enabled"] is True
|