rfc-app/backend/app/email_otc.py

"""Outbound OTC email — a thin wrapper over the existing SMTP layer.

The §15.4 notification mailer in `email.py` is purpose-built for
inbox-driven mail (unsubscribe footers, quiet-hours holds, bundling).
OTC mail is structurally different: it carries a credential, has no
inbox row behind it, and ignores user-preferences (a contributor
who's opted out of every notification still needs to receive the
code they explicitly requested).

So this module reuses `EmailConfig.from_env()` for the SMTP plumbing
and the From identity, but writes its own envelope. In dev (no
SMTP_HOST set), the envelope is logged at INFO level and pushed to
the same `_SENT` buffer the notification mailer uses, so the
integration tests can assert on the outbound shape without standing
up an SMTP server.

The send is synchronous. The `/auth/otc/request` endpoint always
returns 202 regardless of send outcome — the user-facing surface
doesn't know whether the SMTP relay was reachable, since revealing
that would let an attacker probe for valid emails on a tight loop.
"""
from __future__ import annotations

import logging
import smtplib
from email.utils import formataddr

from .email import EmailConfig, _SENT, record_outbound
from .email_envelope import build_envelope

log = logging.getLogger(__name__)


def send_otc_email(to_address: str, code: str) -> bool:
    """Compose and send the one-time-code email. Returns True on the
    happy path; False on SMTP failure. The notifier-side buffer
    `_SENT` is appended either way so tests can assert on content.

    The subject and body intentionally avoid branding strings that
    belong to a deployment — only `EMAIL_FROM_NAME` (operator-supplied
    via env) lands in the From line. The body names the code, the
    TTL, and a single instruction line. No tracking pixel, no
    deep-link query, no embedded JS — plain text only."""
    cfg = EmailConfig.from_env()
    subject = f"Your sign-in code for {cfg.from_name}"
    body = _body(code, cfg)
    # v0.18.0: OTC mail carries NO List-Unsubscribe — the recipient
    # explicitly requested the code; advertising an unsubscribe
    # header would imply OHM has them on a list, which it doesn't.
    # See the proposal's "Tradeoff discussion" for the binding
    # rationale.
    msg = build_envelope(
        to_address=to_address,
        from_address=cfg.from_address,
        from_name=cfg.from_name,
        subject=subject,
        body_plain=body,
    )
    envelope = {
        "to": to_address,
        "from": formataddr((cfg.from_name, cfg.from_address)),
        "subject": subject,
        "body": body,
        "kind": "otc",
        "message": msg,
    }
    _SENT.append(envelope)

    message_id = msg["Message-ID"]
    if not cfg.enabled:
        log.info("otc email disabled (EMAIL_ENABLED=0): to=%s", to_address)
        record_outbound(
            to_address=to_address, from_address=cfg.from_address,
            subject=subject, kind="otc", status="deferred", message_id=message_id,
        )
        return True
    if not cfg.smtp_host:
        # Dev fallback: surface the code at INFO so the operator can
        # complete a sign-in flow without an SMTP relay. In production
        # SMTP_HOST is always set per OHM's overlay.
        log.info("otc email (stdout fallback): to=%s code=%s", to_address, code)
        record_outbound(
            to_address=to_address, from_address=cfg.from_address,
            subject=subject, kind="otc", status="deferred", message_id=message_id,
        )
        return True

    try:
        smtp = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30)
        try:
            if cfg.smtp_starttls:
                smtp.starttls()
            if cfg.smtp_user:
                smtp.login(cfg.smtp_user, cfg.smtp_password)
            smtp.send_message(msg)
        finally:
            smtp.quit()
        record_outbound(
            to_address=to_address, from_address=cfg.from_address,
            subject=subject, kind="otc", status="sent", message_id=message_id,
        )
        return True
    except Exception as exc:
        log.exception("otc email send failed: to=%s", to_address)
        record_outbound(
            to_address=to_address, from_address=cfg.from_address,
            subject=subject, kind="otc", status="failed",
            error=f"{type(exc).__name__}: {exc}", message_id=message_id,
        )
        return False


def _body(code: str, cfg: EmailConfig) -> str:
    return (
        f"Your sign-in code is:\n\n"
        f"    {code}\n\n"
        f"Enter this code in the sign-in screen to finish signing in.\n"
        f"The code expires in 10 minutes. If you did not request this,\n"
        f"you can safely ignore this email — no account was created.\n\n"
        f"---\n"
        f"{cfg.from_name} · {cfg.app_url}\n"
    )