webmail-client/backend/app/identity/invitations.py

"""Invitations: a user sends a friend a single-use sign-up link.

While the administrator has invitations switched on, every user may send
``PER_USER`` of them. A link lets exactly one account be created — even with
public registration closed — and works for ``VALID_FOR``; one that ran out
unused is handed back to its sender. Only a hash of the link's token is kept.

The email is built from the administrator's template (or the built-in one)
with these placeholders, written ``{{name}}``:

    invitee_name   the friend's name, as the sender typed it
    inviter_name   the sender's name (their mail account's display name,
                   else their address)
    invite_link    the sign-up address itself
    invite_button  a ready-made button linking to it
    logo           the Ultimo Email logo

Names are HTML-escaped where they go into the HTML: they come from users and
end up in mail to strangers.
"""

from __future__ import annotations

import datetime as dt
import hashlib
import html as html_lib
import re
import secrets
import uuid
from dataclasses import dataclass

import nh3
from sqlalchemy import func, or_, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import get_settings
from app.identity.mailer import LOGO_HTML
from app.models import Account, AppSettings, Invitation, User
from app.outbox.inline_images import InlinePart, data_image

PER_USER = 2
VALID_FOR = dt.timedelta(days=14)
MAX_TEMPLATE_BYTES = 500 * 1024

PLACEHOLDERS = ("invitee_name", "inviter_name", "invite_link", "invite_button", "logo")
_PLACEHOLDER = re.compile(r"\{\{\s*([a-z_]+)\s*\}\}")
_DATA_SRC = re.compile(r"""(\bsrc\s*=\s*)(["'])(data:image/[^"']+)\2""", re.IGNORECASE)

DEFAULT_SUBJECT = "{{inviter_name}} invites you to Ultimo Email"
_BODY = "font-family:Arial,Helvetica,sans-serif;font-size:15px;color:#1f2937;line-height:1.5"
_MUTED = "font-size:13px;color:#6b7280"
_BUTTON = (
    "display:inline-block;background:#2563eb;color:#ffffff;text-decoration:none;"
    "padding:10px 18px;border-radius:8px;font-weight:bold"
)
DEFAULT_TEMPLATE = f"""<!doctype html>
<html><body style="{_BODY}">
{{{{logo}}}}
<p>Hi {{{{invitee_name}}}},</p>
<p><b>{{{{inviter_name}}}}</b> invites you to Ultimo Email — private email,
calendar and contacts in one place.</p>
<p>{{{{invite_button}}}}</p>
<p style="{_MUTED}">Or paste this address into your browser:<br>{{{{invite_link}}}}</p>
<p style="{_MUTED}">The invitation works once and expires in 14 days. If you
weren't expecting it, you can ignore this email.</p>
</body></html>
"""


class InvitationError(Exception):
    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message


@dataclass(frozen=True)
class RenderedMail:
    subject: str
    text: str
    html: str
    inline: list[InlinePart]


def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


def sender_address(row: AppSettings) -> str:
    """The invitation's From. Empty: the system-mail From address, the one the
    SMTP account is known to be allowed to send as."""
    return row.invite_from.strip() or row.smtp_from


def template_problem(html: str) -> str | None:
    """Why ``html`` cannot be used as the invitation template, or None."""
    if len(html.encode()) > MAX_TEMPLATE_BYTES:
        return f"the template is larger than {MAX_TEMPLATE_BYTES // 1024} KB"
    used = set(_PLACEHOLDER.findall(html))
    unknown = sorted(used - set(PLACEHOLDERS))
    if unknown:
        return "unknown placeholder: " + ", ".join("{{" + u + "}}" for u in unknown)
    if not used & {"invite_link", "invite_button"}:
        return "the template must contain {{invite_link}} or {{invite_button}}"
    return None


def render(row: AppSettings, *, invitee_name: str, inviter_name: str, link: str) -> RenderedMail:
    """The invitation email for one friend, from the configured template."""
    template = row.invite_template or DEFAULT_TEMPLATE
    subject_template = row.invite_subject or DEFAULT_SUBJECT
    esc_link = html_lib.escape(link, quote=True)
    html_values = {
        "invitee_name": html_lib.escape(invitee_name),
        "inviter_name": html_lib.escape(inviter_name),
        "invite_link": f'<a href="{esc_link}">{esc_link}</a>',
        "invite_button": f'<a href="{esc_link}" style="{_BUTTON}">Accept the invitation</a>',
        "logo": LOGO_HTML,
    }
    text_values = {
        "invitee_name": invitee_name,
        "inviter_name": inviter_name,
        "invite_link": link,
        "invite_button": link,
        "logo": "",
    }

    def fill(source: str, values: dict[str, str]) -> str:
        return _PLACEHOLDER.sub(lambda m: values.get(m.group(1), m.group(0)), source)

    # An image written into the template itself (data:) goes as an inline part:
    # Gmail does not show data: images.
    inline: list[InlinePart] = []

    def to_cid(match: re.Match[str]) -> str:
        part = data_image(match.group(3), get_settings().app_domain)
        if part is None:
            return match.group(0)
        inline.append(part)
        return f"{match.group(1)}{match.group(2)}cid:{part.cid}{match.group(2)}"

    html = _DATA_SRC.sub(to_cid, fill(template, html_values))
    text = fill(template, text_values)
    # Keep the paragraphs and line breaks the HTML shows once its tags are gone.
    text = re.sub(r"<br\s*/?>", "\n", text, flags=re.IGNORECASE)
    text = re.sub(r"</(?:p|div|h[1-6]|li|tr|table)\s*>", "\n\n", text, flags=re.IGNORECASE)
    text = nh3.clean(text, tags=set(), attributes={}, strip_comments=True)
    text = html_lib.unescape(re.sub(r"\n\s*\n\s*", "\n\n", text)).strip() + "\n"
    subject = " ".join(fill(subject_template, text_values).split())
    return RenderedMail(subject=subject, text=text, html=html, inline=inline)


async def inviter_name(db: AsyncSession, user: User) -> str:
    """How the friend is told who invited them."""
    names = (
        await db.execute(
            select(Account.email, Account.display_name).where(Account.user_id == user.id)
        )
    ).all()
    own = next((n for e, n in names if e.lower() == user.handle.lower() and n), None)
    return own or next((n for _, n in names if n), None) or user.handle


def _counts(now: dt.datetime):  # type: ignore[no-untyped-def]
    # Used, or still valid: an invitation that expired unused is given back.
    return or_(Invitation.used_at.isnot(None), Invitation.expires_at > now)


async def remaining(db: AsyncSession, user_id: uuid.UUID) -> int:
    now = dt.datetime.now(dt.UTC)
    taken = await db.scalar(
        select(func.count())
        .select_from(Invitation)
        .where(Invitation.inviter_id == user_id, _counts(now))
    )
    return max(0, PER_USER - int(taken or 0))


async def sent(db: AsyncSession, user_id: uuid.UUID) -> list[Invitation]:
    return list(
        await db.scalars(
            select(Invitation)
            .where(Invitation.inviter_id == user_id)
            .order_by(Invitation.created_at.desc())
        )
    )


async def create(db: AsyncSession, user: User, *, name: str, email: str) -> tuple[Invitation, str]:
    """A new invitation from ``user`` (not committed): the row and the token.
    Raises :class:`InvitationError` when the user has none left."""
    # One at a time per user, so two quick clicks cannot both take the last one.
    await db.execute(select(User.id).where(User.id == user.id).with_for_update())
    if await remaining(db, user.id) <= 0:
        raise InvitationError("no_invitations_left", "you have no invitations left")
    token = secrets.token_urlsafe(32)
    invitation = Invitation(
        inviter_id=user.id,
        token_hash=hash_token(token),
        invitee_name=name,
        invitee_email=email,
        expires_at=dt.datetime.now(dt.UTC) + VALID_FOR,
    )
    db.add(invitation)
    await db.flush()
    return invitation, token


async def find_valid(db: AsyncSession, token: str) -> Invitation | None:
    """The unused, unexpired invitation behind ``token``, without using it."""
    invitation: Invitation | None = await db.scalar(
        select(Invitation).where(
            Invitation.token_hash == hash_token(token),
            Invitation.used_at.is_(None),
            Invitation.expires_at > dt.datetime.now(dt.UTC),
        )
    )
    return invitation


async def claim(db: AsyncSession, token: str) -> uuid.UUID | None:
    """Use the invitation up (in the caller's transaction): its id, or None if
    it is unknown, used or expired. Atomic — two sign-ups with one link cannot
    both get it."""
    return await db.scalar(
        update(Invitation)
        .where(
            Invitation.token_hash == hash_token(token),
            Invitation.used_at.is_(None),
            Invitation.expires_at > dt.datetime.now(dt.UTC),
        )
        .values(used_at=func.now())
        .returning(Invitation.id)
    )


async def record_user(db: AsyncSession, invitation_id: uuid.UUID, user_id: uuid.UUID) -> None:
    await db.execute(
        update(Invitation).where(Invitation.id == invitation_id).values(used_by=user_id)
    )