path: root/packages/meshbay-hub/src/meshbay_hub/mail.py

"""
MeshBay Hub — email sending via localhost Postfix.

Postfix listens on loopback only (inet_interfaces = loopback-only), so no
authentication is needed. See docs/MAIL-SERVER.md for the full setup.

**No authentication is needed** is exactly why everything below exists. The
hub can emit mail from its own domain to anywhere, and three API paths reach
that ability — two of them at an address the caller types. Unbounded, that is
an open relay wearing the instance's reputation, so the bounds are here, in
the one function every send passes through, rather than at the call sites
where the next one added would forget them (`AV9`–`AV10`, §13.5b).
"""

import asyncio
import hashlib
import logging
import smtplib
from datetime import UTC, datetime, timedelta
from email.message import EmailMessage

log = logging.getLogger(__name__)


_hub_domain: str = "meshbay.org"
_hub_url: str = "https://meshbay.org"


def configure(hub_id: str) -> None:
    global _hub_domain, _hub_url
    _hub_domain = hub_id
    _hub_url = f"https://{hub_id}"


class MailRefused(Exception):
    """The gate below declined to send. Never carries the address."""


# The complete list of reasons this hub will ever send mail. A `send_*`
# function that names anything else does not send — and one that names nothing
# is a TypeError, because `purpose` is keyword-required.
#
#   registration   confirm an address at sign-up
#   password_reset a code to the address already on file for that account
#   invite         tell a registered member they were invited to a group
#   email_change   confirm a new address before it replaces the old one
#
# Only `registration` and `email_change` can reach an address this hub has no
# prior relationship with. Both are necessary; both are why the bound that
# matters is keyed on the recipient rather than on who asked.
ALLOWED_PURPOSES = frozenset({
    "registration", "password_reset", "invite", "email_change"})

# The two a person is actively waiting on. They may spend the whole hourly
# budget; the other two may not spend the reserved share of it, so a flood of
# sign-ups cannot lock out someone trying to recover their passphrase.
RECOVERY_PURPOSES = frozenset({"password_reset", "invite"})


def destination_key(address: str) -> str:
    """A stable handle for one recipient that is not the address itself.

    Hashed because this table would otherwise be the one place in the hub
    holding a list of plaintext addresses — the rest of the codebase goes to
    the trouble of encrypting them at rest (S2).
    """
    return "dest:" + hashlib.sha256(
        address.strip().lower().encode()).hexdigest()[:32]


async def _take(db, key: str, window: timedelta, ceiling: int,
                cooldown: timedelta | None) -> None:
    """Charge one send against a counter, or raise MailRefused.

    Rows live in `mail_quota` rather than in a module dict, so the allowance
    survives the restart that a deploy is. The caller owns the commit — which
    means a send that then fails to commit is not charged, and that is the
    right way round: the alternative charges for mail nobody received.
    """
    from meshbay_hub.db.models import MailQuota

    now = datetime.now(UTC)
    row = await db.get(MailQuota, key)
    if row is None:
        row = MailQuota(key=key, window_start=now, count=0, last_sent=None)
        db.add(row)

    started = row.window_start
    if started.tzinfo is None:
        started = started.replace(tzinfo=UTC)
    if now - started >= window:
        row.window_start, row.count = now, 0

    if cooldown is not None and row.last_sent is not None:
        last = row.last_sent
        if last.tzinfo is None:
            last = last.replace(tzinfo=UTC)
        if now - last < cooldown:
            raise MailRefused("too soon since the last message to this recipient")

    if row.count >= ceiling:
        raise MailRefused("allowance spent for this window")

    row.count += 1
    row.last_sent = now


# The two global ceilings, and what each one means to somebody waiting.
EXHAUSTED_GENERAL = "general"      # sign-ups and address changes refused
EXHAUSTED_ALL = "all"              # even a passphrase reset is refused


async def _announce_exhaustion(scope: str) -> None:
    """Tell the administrators, once per hour, that the hub stopped sending.

    In its own session, and never raising: this runs while a request is being
    refused, and an alert that fails must not turn a refusal into a 500. Same
    reason `_mark_hosted` opens its own session next door.

    Once per window, keyed on a row rather than a flag in memory, because the
    thing being reported is exactly the kind of event a restart would erase —
    and a hub that is refusing mail is a hub somebody is probably restarting.
    """
    from sqlalchemy import select

    from meshbay_hub.api.deps import user_is_admin
    from meshbay_hub.api.notifications import create_notification
    from meshbay_hub.db.engine import get_session_factory
    from meshbay_hub.db.models import MailQuota, User

    try:
        async with get_session_factory()() as db:
            key = f"alert:{scope}"
            now = datetime.now(UTC)
            row = await db.get(MailQuota, key)
            if row is not None and row.last_sent is not None:
                last = row.last_sent
                if last.tzinfo is None:
                    last = last.replace(tzinfo=UTC)
                if now - last < timedelta(hours=1):
                    return
            if row is None:
                row = MailQuota(key=key, window_start=now, count=0)
                db.add(row)
            row.last_sent = now
            row.count += 1

            admins = [u for u in (await db.execute(select(User).where(
                User.status == "active"))).scalars().all() if user_is_admin(u)]
            for admin in admins:
                await create_notification(
                    db, admin.id, f"mail_budget_{scope}",
                    "This hub has stopped sending mail for this hour"
                    if scope == EXHAUSTED_ALL
                    else "This hub has stopped sending sign-up mail for this hour",
                    detail=("Passphrase resets and invitations are refused too."
                            if scope == EXHAUSTED_ALL
                            else "Passphrase resets and invitations still go out."),
                    link="#/admin",
                    aggregate=False,
                )
            await db.commit()
            log.warning("Mail budget exhausted (%s) — %d administrator(s) told",
                        scope, len(admins))
    except Exception as e:
        log.warning("Could not announce the mail budget: %s", e)


async def reserve(db, purpose: str, address: str) -> None:
    """Charge one send, or raise MailRefused. The caller owns the commit."""
    from meshbay_hub import hub_settings

    if purpose not in ALLOWED_PURPOSES:
        raise MailRefused(f"not a purpose this hub sends mail for: {purpose!r}")

    limits = await hub_settings.mail_limits(db)

    # The instance ceiling first, and the recovery share is subtracted for the
    # purposes that are not one. Registration is open, so a limit counted per
    # account or per IP is one an attacker buys more of; this one is not.
    budget = limits["hourly_budget"]
    if purpose not in RECOVERY_PURPOSES:
        budget = max(0, budget - limits["hourly_reserved_for_recovery"])
    try:
        await _take(db, "hour", timedelta(hours=1), budget, None)
    except MailRefused:
        # An operator finds out here or not at all: a refusal is otherwise a
        # line in the journal, and a hub that has stopped sending sign-up
        # codes looks exactly like one nobody is signing up to.
        await _announce_exhaustion(
            EXHAUSTED_ALL if purpose in RECOVERY_PURPOSES else EXHAUSTED_GENERAL)
        raise

    # Then the recipient: across every purpose, account and endpoint. This is
    # what a person being mail-bombed actually experiences, and the only bound
    # that describes it.
    await _take(
        db, destination_key(address), timedelta(days=1),
        limits["destination_daily_cap"],
        timedelta(seconds=limits["destination_cooldown_seconds"]))


async def status(db) -> dict:
    """What the operator sees in the panel: is the hub still sending?"""
    from sqlalchemy import func, select

    from meshbay_hub import hub_settings
    from meshbay_hub.db.models import MailQuota

    limits = await hub_settings.mail_limits(db)
    row = await db.get(MailQuota, "hour")
    used = 0
    window_start = None
    if row is not None:
        started = row.window_start
        if started.tzinfo is None:
            started = started.replace(tzinfo=UTC)
        if datetime.now(UTC) - started < timedelta(hours=1):
            used, window_start = row.count, started.isoformat()

    recipients = await db.scalar(
        select(func.count()).select_from(MailQuota)
        .where(MailQuota.key.like("dest:%"))) or 0

    general = max(0, limits["hourly_budget"] - limits["hourly_reserved_for_recovery"])
    # Named rather than left for the reader to compute from two numbers: the
    # panel draws a warning off these, and "is it still sending" is the
    # question an operator opens this page to answer.
    general_exhausted = used >= general
    all_exhausted = used >= limits["hourly_budget"]
    return {
        "general_exhausted":  general_exhausted,
        "all_exhausted":      all_exhausted,
        "hourly_budget":      limits["hourly_budget"],
        "hourly_used":        used,
        "hour_started_at":    window_start,
        "reserved_for_recovery": limits["hourly_reserved_for_recovery"],
        # What a sign-up would meet right now, which is the number an operator
        # is actually asking about when they open this panel.
        "general_remaining":  max(0, general - used),
        "recovery_remaining": max(0, limits["hourly_budget"] - used),
        "recipients_tracked": recipients,
    }


async def purge_expired_quota(db) -> int:
    """Drop counters whose window has passed. Returns how many went."""
    from sqlalchemy import delete

    from meshbay_hub.db.models import MailQuota

    cutoff = datetime.now(UTC) - timedelta(days=1)
    result = await db.execute(
        delete(MailQuota).where(MailQuota.window_start < cutoff))
    await db.commit()
    return result.rowcount or 0


def _send(msg: EmailMessage, *, purpose: str) -> bool:
    """Blocking. Every caller in an async handler must use `send_off_loop`.

    The gate is here rather than in `send_off_loop` so that it cannot be
    stepped around: a new `send_*` helper, a script, a test — everything that
    puts a message on the wire comes through this function, and `purpose` is
    keyword-required so forgetting it is a TypeError rather than an
    unrestricted send.
    """
    if purpose not in ALLOWED_PURPOSES:
        # The structural half of the gate, and the half that needs no state:
        # it is here rather than only in `reserve` so that nothing which puts a
        # message on the wire — a new helper, a script, a test — can name a
        # reason this hub does not send mail for. The counting half is
        # `reserve`, which needs a database session and so cannot live here.
        log.warning("Mail refused: %r is not a purpose this hub sends for",
                    purpose)
        return False
    try:
        with smtplib.SMTP("localhost", 25, timeout=10) as s:
            s.send_message(msg)
        return True
    except Exception:
        log.exception("Failed to send email to %s", _mask_email(msg["To"] or ""))
        return False


async def send_off_loop(db, fn, *args, purpose: str, **kwargs) -> bool:
    """Run one of the `send_*` functions below in a worker thread.

    `smtplib` is synchronous and this one waits up to ten seconds. Called
    directly from an async handler — which is what all four call sites did —
    that ten seconds is not one request's, it is **the whole hub's**: no other
    request is served, no node socket is read, no WebRTC offer is relayed,
    for as long as the MTA takes to answer. An unreachable mail server made
    the instance stop responding to everyone, and one of the three paths that
    reaches it (`PATCH /v1/users/me`) had no rate limit at all.

    So the cost of a slow MTA is one request now, not the instance.

    It is also the one door: `reserve` charges the send against the recipient's
    allowance and the instance's before anything is handed to the thread, and
    the recipient is `args[0]` because that is the first parameter of every
    `send_*` function below. Returns whether the message went. The caller owns
    the commit, so a request that fails afterwards is not charged for mail
    nobody received.
    """
    try:
        await reserve(db, purpose, args[0])
    except MailRefused as refusal:
        # Never the address: this line goes to the journal.
        log.warning("Mail refused (%s): %s", purpose, refusal)
        return False
    await asyncio.to_thread(fn, *args, **kwargs)
    return True


def send_verification_code(to: str, code: str, recovery_key: str | None = None) -> None:
    """
    Registration verification e-mail. When `recovery_key` is given it is
    appended to the body so the recipient's mailbox becomes the backup for it
    (docs/MESHBAY_DESIGN.md §3.6).

    `recovery_key` is a **pass-through**: it is generated on the client, never
    stored anywhere on the hub, and never logged — only whether one was present.
    """
    body = (
        f"Your verification code is: {code}\n"
        "\n"
        "Enter this code to verify your email address.\n"
        "This code expires in 24 hours.\n"
    )
    if recovery_key:
        body += (
            "\n"
            "---- Account recovery key ----\n"
            "\n"
            "Keep this message. If you ever forget your passphrase, this key is\n"
            "what restores your access to your groups. It is not stored on the\n"
            f"server and nobody at {_hub_domain} can recover it for you.\n"
            "\n"
            f"    {recovery_key}\n"
        )
    body += (
        "\n"
        "If you did not create a MeshBay account, ignore this email.\n"
        "\n"
        f"{_hub_url}\n"
    )
    msg = EmailMessage()
    msg["From"] = f"noreply@{_hub_domain}"
    msg["To"] = to
    msg["Subject"] = f"MeshBay — Your verification code: {code}"
    msg.set_content(body)
    _send(msg, purpose="registration")
    log.info("Verification code sent to %s (recovery_key=%s)",
             _mask_email(to), bool(recovery_key))


def send_email_change_code(to: str, code: str) -> None:
    msg = EmailMessage()
    msg["From"] = f"noreply@{_hub_domain}"
    msg["To"] = to
    msg["Subject"] = f"MeshBay — Confirm your new email: {code}"
    msg.set_content(
        f"Your verification code is: {code}\n"
        "\n"
        "Enter this code to confirm your new email address.\n"
        "This code expires in 24 hours.\n"
        "\n"
        "If you did not request this change, ignore this email.\n"
        "\n"
        f"{_hub_url}\n"
    )
    _send(msg, purpose="email_change")
    log.info("Email change code sent to %s", _mask_email(to))


def send_password_reset_code(to: str, code: str) -> None:
    """
    Passphrase-reset code (docs/MESHBAY_DESIGN.md §3.6). This only re-opens hub
    login; it recovers no group content — that needs the recovery key.
    """
    msg = EmailMessage()
    msg["From"] = f"noreply@{_hub_domain}"
    msg["To"] = to
    msg["Subject"] = f"MeshBay — Passphrase reset code: {code}"
    msg.set_content(
        f"Your passphrase reset code is: {code}\n"
        "\n"
        "Enter it to set a new passphrase. This code expires in 1 hour.\n"
        "\n"
        "This restores your sign-in only. If you also have your recovery key,\n"
        "you can restore access to your groups in the same step.\n"
        "\n"
        "If you did not request this, ignore this email — your account is\n"
        "unchanged.\n"
        "\n"
        f"{_hub_url}\n"
    )
    _send(msg, purpose="password_reset")
    log.info("Passphrase reset code sent to %s", _mask_email(to))


def send_invite_notification(
    to: str, code: str, inviter: str, group_name: str,
) -> None:
    msg = EmailMessage()
    msg["From"] = f"noreply@{_hub_domain}"
    msg["To"] = to
    msg["Subject"] = f"MeshBay — {inviter} invited you to {group_name}"
    msg.set_content(
        f"{inviter} invited you to the group \"{group_name}\" on MeshBay.\n"
        "\n"
        f"Your one-time code is: {code}\n"
        "\n"
        "Open the group and enter this code when prompted.\n"
        "The code works once and expires in 7 days.\n"
        "\n"
        f"{_hub_url}\n"
    )
    _send(msg, purpose="invite")
    log.info("Invite notification sent to %s", _mask_email(to))


def _mask_email(email: str) -> str:
    local, _, domain = email.partition("@")
    if len(local) <= 2:
        return f"{'*' * len(local)}@{domain}"
    return f"{local[0]}{'*' * (len(local) - 2)}{local[-1]}@{domain}"