""" app/mail.py ─────────── Adaptive SMTP delivery engine. Three mutually-exclusive modes are selected at runtime based on environment: Mode A (Authenticated TLS relay) All three MAIL_RELAYHOST + MAIL_RELAYHOST_USERNAME + MAIL_RELAYHOST_PASSWORD must be set. The backend connects with STARTTLS and authenticates. Mode B (Simple container/host proxy relay) Only MAIL_RELAY_HOST is set (hostname, no credentials). Backend opens an unauthenticated SMTP session on port 587 to that host. Designed for a companion boky/postfix container in the same Docker network. Direct SMTP (fallback) No relay variables present. Backend attempts delivery via localhost:25 (or any MTA listening locally). """ from __future__ import annotations import logging import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from enum import Enum from app.config import settings from app.models import STATUS_DELIVERED, STATUS_FAILED, STATUS_RELAY logger = logging.getLogger(__name__) # ───────────────────────────────────────────────────────────────────────────── # Mode detection # ───────────────────────────────────────────────────────────────────────────── class MailMode(str, Enum): MODE_A = "authenticated_relay" MODE_B = "proxy_relay" DIRECT = "direct_smtp" def _detect_mode() -> MailMode: if ( settings.MAIL_RELAYHOST and settings.MAIL_RELAYHOST_USERNAME and settings.MAIL_RELAYHOST_PASSWORD ): return MailMode.MODE_A if settings.MAIL_RELAY_HOST: return MailMode.MODE_B return MailMode.DIRECT def _build_message(to_email: str, subject: str, body: str) -> MIMEMultipart: msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = f"{settings.MAIL_FROM_NAME} <{settings.MAIL_FROM}>" msg["To"] = to_email msg.attach(MIMEText(body, "plain", "utf-8")) return msg # ───────────────────────────────────────────────────────────────────────────── # Public send function # ───────────────────────────────────────────────────────────────────────────── def send_email(to_email: str, subject: str, body: str) -> str: """ Deliver an email to *to_email* using the configured strategy. Returns one of the STATUS_* constants on success. Raises on failure so the caller can log STATUS_FAILED. """ mode = _detect_mode() msg = _build_message(to_email, subject, body) raw = msg.as_string() try: if mode == MailMode.MODE_A: # ── Authenticated TLS relay ────────────────────────────────────── relay = settings.MAIL_RELAYHOST if ":" in relay: host, port_str = relay.rsplit(":", 1) port = int(port_str) else: host, port = relay, 587 with smtplib.SMTP(host, port, timeout=20) as srv: srv.ehlo() srv.starttls() srv.ehlo() srv.login(settings.MAIL_RELAYHOST_USERNAME, settings.MAIL_RELAYHOST_PASSWORD) srv.sendmail(settings.MAIL_FROM, [to_email], raw) logger.info("[MAIL MODE-A] → %s via %s:%d", to_email, host, port) return STATUS_DELIVERED elif mode == MailMode.MODE_B: # ── Unauthenticated container proxy relay ──────────────────────── host = settings.MAIL_RELAY_HOST with smtplib.SMTP(host, 587, timeout=20) as srv: srv.ehlo() srv.sendmail(settings.MAIL_FROM, [to_email], raw) logger.info("[MAIL MODE-B] → %s via %s:587", to_email, host) return STATUS_RELAY else: # ── Direct localhost:25 delivery ───────────────────────────────── with smtplib.SMTP("localhost", 25, timeout=20) as srv: srv.sendmail(settings.MAIL_FROM, [to_email], raw) logger.info("[MAIL DIRECT] → %s via localhost:25", to_email) return STATUS_DELIVERED except Exception as exc: logger.error( "[MAIL ERROR] Failed to deliver to %s (mode=%s): %s", to_email, mode.value, exc, exc_info=True, ) raise