kundenplattform/mail.py

import asyncio
import logging
import os
import smtplib
import ssl
import sys
import time
import uuid
from dataclasses import dataclass
from email.message import EmailMessage
from email.utils import formatdate, make_msgid, parseaddr
from pathlib import Path

from email_validator import EmailNotValidError, validate_email

from db import get_database_connection


# Mailversand der Plattform (ADR-0041, 12.10.2026). Aufrufer tragen eine
# Mail mit einreihen() in die Tabelle mail_ausgang ein, die Versandschleife
# (in app.py beim Start angestoßen) liefert sie per SMTP mit Anmeldung beim
# Mailanbieter der Domain ein. anode verschickt nie direkt an Empfänger -
# Mails von einer Heimanschluss-Adresse landen im Spam.
#
# Bewusst nur Standardbibliothek (smtplib), keine neue Abhängigkeit. Der
# Dienst läuft als einzelner uvicorn-Prozess (siehe auth.py), es gibt also
# genau eine Versandschleife; FOR UPDATE SKIP LOCKED schützt trotzdem
# gegen einen parallel laufenden scripts/testmail.py.

log = logging.getLogger("kundenplattform.mail")
if not log.handlers:
    # Die Anwendung konfiguriert sonst kein Logging; ohne eigenen Handler
    # gingen INFO-Meldungen (Trockenlauf!) verloren. stderr landet im Journal.
    log.addHandler(logging.StreamHandler(sys.stderr))
    log.setLevel(logging.INFO)

# Wartezeiten vor dem 2., 3. und 4. Versuch. Danach gilt die Mail als
# fehlgeschlagen. Zusammen gut 20 Minuten - passt in die Stundenfrist der
# Links für "Passwort vergessen?".
WIEDERHOLUNG_NACH_SEKUNDEN = (60, 300, 900)

SCHLEIFEN_INTERVALL_SEKUNDEN = 10
FEHLER_PAUSE_SEKUNDEN = 300
SMTP_TIMEOUT_SEKUNDEN = 30

# Die Tabelle enthält Empfängeradressen (personenbezogen) - alte Einträge
# werden gelöscht.
AUFBEWAHRUNG_TAGE = 90
AUFRAEUMEN_INTERVALL_SEKUNDEN = 60 * 60


@dataclass
class Konfiguration:
    modus: str
    host: str
    port: int
    benutzer: str
    absender: str
    antwort_an: str
    passwort_datei: str


def _konfiguration() -> Konfiguration:
    """
    Bei jedem Aufruf frisch aus der Umgebung gelesen (nicht beim Import wie
    in db.py/encryption.py), damit Tests und scripts/testmail.py den Modus
    umschalten können.

    KUNDENPLATTFORM_MAIL_MODUS: "trockenlauf" (Default - schreibt die Mail
    ins Protokoll statt sie zu verschicken) oder "smtp".
    Das Passwort steht nie in der Umgebung, sondern in einer Datei:
    systemd-Credential "mail_passwort" (siehe kundenplattform.service) oder
    ausdrücklich KUNDENPLATTFORM_MAIL_PASSWORT_DATEI.
    """

    passwort_datei = os.environ.get("KUNDENPLATTFORM_MAIL_PASSWORT_DATEI", "")
    if not passwort_datei and os.environ.get("CREDENTIALS_DIRECTORY"):
        passwort_datei = os.path.join(os.environ["CREDENTIALS_DIRECTORY"], "mail_passwort")

    return Konfiguration(
        modus=os.environ.get("KUNDENPLATTFORM_MAIL_MODUS", "trockenlauf"),
        host=os.environ.get("KUNDENPLATTFORM_MAIL_HOST", ""),
        port=int(os.environ.get("KUNDENPLATTFORM_MAIL_PORT", "587")),
        benutzer=os.environ.get("KUNDENPLATTFORM_MAIL_BENUTZER", ""),
        absender=os.environ.get("KUNDENPLATTFORM_MAIL_ABSENDER", ""),
        antwort_an=os.environ.get("KUNDENPLATTFORM_MAIL_ANTWORT_AN", ""),
        passwort_datei=passwort_datei,
    )


def _als_text(wert):
    # psycopg liefert TEXT-Spalten hier teils als bytes (siehe auth.py).
    return wert.decode("utf-8") if isinstance(wert, bytes) else wert


def einreihen(empfaenger: str, betreff: str, text: str, zweck: str, bezug: str | None = None, conn=None) -> str:
    """
    Trägt eine Mail zum Versand ein und gibt ihre ID zurück (für
    mail_status()). Mit conn läuft der Eintrag in der Transaktion des
    Aufrufers - die Mail geht dann nur raus, wenn auch dessen übrige
    Änderungen festgeschrieben werden.

    Wirft ValueError bei ungültiger Adresse oder Zeilenumbruch im Betreff.
    """

    try:
        adresse = validate_email(empfaenger.strip(), check_deliverability=False).normalized
    except EmailNotValidError as fehler:
        raise ValueError(f"Ungültige E-Mail-Adresse: {fehler}") from fehler

    if "\n" in betreff or "\r" in betreff:
        raise ValueError("Der Betreff darf keinen Zeilenumbruch enthalten.")

    mail_id = uuid.uuid4()
    sql = "INSERT INTO mail_ausgang (id, empfaenger, betreff, text, zweck, bezug) VALUES (%s, %s, %s, %s, %s, %s)"
    werte = (mail_id, adresse, betreff, text, zweck, bezug)

    if conn is not None:
        with conn.cursor() as cur:
            cur.execute(sql, werte)
    else:
        with get_database_connection() as eigene_verbindung:
            with eigene_verbindung.cursor() as cur:
                cur.execute(sql, werte)

    return str(mail_id)


def mail_status(mail_id: str) -> dict | None:
    with get_database_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT status, versuche, letzter_fehler, created_at, gesendet_at
                FROM mail_ausgang WHERE id = %s
                """,
                (mail_id,),
            )
            row = cur.fetchone()

    if row is None:
        return None

    status, versuche, letzter_fehler, created_at, gesendet_at = row

    return {
        "status": _als_text(status),
        "versuche": versuche,
        "letzter_fehler": _als_text(letzter_fehler),
        "created_at": created_at,
        "gesendet_at": gesendet_at,
    }


def _nachricht(konfiguration: Konfiguration, empfaenger: str, betreff: str, text: str) -> EmailMessage:
    nachricht = EmailMessage()
    nachricht["From"] = konfiguration.absender
    nachricht["To"] = empfaenger
    nachricht["Subject"] = betreff
    nachricht["Date"] = formatdate(localtime=True)
    nachricht["Message-ID"] = make_msgid(domain=parseaddr(konfiguration.absender)[1].rpartition("@")[2] or None)
    # Kennzeichnet die Mail als automatisch erzeugt, damit
    # Abwesenheitsnotizen nicht darauf antworten.
    nachricht["Auto-Submitted"] = "auto-generated"
    if konfiguration.antwort_an:
        nachricht["Reply-To"] = konfiguration.antwort_an
    nachricht.set_content(text)

    return nachricht


def _per_smtp_senden(konfiguration: Konfiguration, nachricht: EmailMessage) -> None:
    fehlend = [
        name for name, wert in (
            ("KUNDENPLATTFORM_MAIL_HOST", konfiguration.host),
            ("KUNDENPLATTFORM_MAIL_BENUTZER", konfiguration.benutzer),
            ("KUNDENPLATTFORM_MAIL_ABSENDER", konfiguration.absender),
            ("Passwortdatei (Credential mail_passwort)", konfiguration.passwort_datei),
        ) if not wert
    ]
    if fehlend:
        raise RuntimeError("Mailversand nicht konfiguriert, es fehlt: " + ", ".join(fehlend))

    passwort = Path(konfiguration.passwort_datei).read_text(encoding="utf-8").strip()

    with smtplib.SMTP(konfiguration.host, konfiguration.port, timeout=SMTP_TIMEOUT_SEKUNDEN) as smtp:
        smtp.starttls(context=ssl.create_default_context())
        smtp.login(konfiguration.benutzer, passwort)
        smtp.send_message(nachricht)


def ausgang_abarbeiten(limit: int = 20) -> int:
    """
    Verschickt die fälligen Mails (höchstens limit) und gibt zurück, wie
    viele bearbeitet wurden. Jede Mail läuft in einer eigenen Transaktion,
    die Zeile bleibt während des Versands gesperrt.
    """

    konfiguration = _konfiguration()
    bearbeitet = 0

    with get_database_connection() as conn:
        while bearbeitet < limit:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    SELECT id, empfaenger, betreff, text, zweck, versuche
                    FROM mail_ausgang
                    WHERE status = 'wartet' AND naechster_versuch_at <= CURRENT_TIMESTAMP
                    ORDER BY created_at
                    LIMIT 1
                    FOR UPDATE SKIP LOCKED
                    """
                )
                row = cur.fetchone()
                if row is None:
                    break

                mail_id, empfaenger, betreff, text, zweck, versuche = row
                empfaenger, betreff, text, zweck = (_als_text(w) for w in (empfaenger, betreff, text, zweck))
                versuche += 1

                status, fehler, wartezeit = _zustellen(konfiguration, empfaenger, betreff, text, zweck, versuche)

                if status == "wartet":
                    cur.execute(
                        """
                        UPDATE mail_ausgang
                        SET versuche = %s, letzter_fehler = %s,
                            naechster_versuch_at = CURRENT_TIMESTAMP + make_interval(secs => %s)
                        WHERE id = %s
                        """,
                        (versuche, fehler, wartezeit, mail_id),
                    )
                else:
                    cur.execute(
                        """
                        UPDATE mail_ausgang
                        SET status = %s, versuche = %s, letzter_fehler = %s, text = NULL,
                            gesendet_at = CASE WHEN %s = 'gesendet' THEN CURRENT_TIMESTAMP END
                        WHERE id = %s
                        """,
                        (status, versuche, fehler, status, mail_id),
                    )

            conn.commit()
            bearbeitet += 1

    return bearbeitet


def _zustellen(konfiguration, empfaenger, betreff, text, zweck, versuche) -> tuple[str, str | None, int]:
    """Ein Zustellversuch. Gibt (neuer Status, Fehlertext, Wartezeit bis zum nächsten Versuch) zurück."""

    if konfiguration.modus != "smtp":
        log.info("Mail (Trockenlauf) an %s\nBetreff: %s\n%s\n---", empfaenger, betreff, text)
        return "trockenlauf", None, 0

    try:
        _per_smtp_senden(konfiguration, _nachricht(konfiguration, empfaenger, betreff, text))
    except smtplib.SMTPRecipientsRefused as fehler:
        # Der Mailserver lehnt die Adresse ab - ein weiterer Versuch ändert
        # daran nichts.
        log.warning("Mail (%s) an %s abgelehnt: %s", zweck, empfaenger, fehler)
        return "fehlgeschlagen", f"Empfänger abgelehnt: {fehler}", 0
    except Exception as fehler:  # noqa: BLE001 - jeder Fehler führt zur Wiederholung
        if versuche > len(WIEDERHOLUNG_NACH_SEKUNDEN):
            log.error("Mail (%s) an %s endgültig fehlgeschlagen nach %s Versuchen: %s", zweck, empfaenger, versuche, fehler)
            return "fehlgeschlagen", str(fehler), 0

        wartezeit = WIEDERHOLUNG_NACH_SEKUNDEN[versuche - 1]
        log.warning("Mail (%s) an %s: Versuch %s fehlgeschlagen (%s), nächster in %s s", zweck, empfaenger, versuche, fehler, wartezeit)
        return "wartet", str(fehler), wartezeit

    log.info("Mail (%s) an %s verschickt.", zweck, empfaenger)
    return "gesendet", None, 0


def aufraeumen() -> int:
    with get_database_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "DELETE FROM mail_ausgang WHERE created_at < CURRENT_TIMESTAMP - make_interval(days => %s)",
                (AUFBEWAHRUNG_TAGE,),
            )
            return cur.rowcount


async def versand_schleife() -> None:
    """Läuft für die Lebensdauer des Dienstes (siehe lifespan in app.py)."""

    zuletzt_aufgeraeumt = 0.0

    while True:
        try:
            await asyncio.to_thread(ausgang_abarbeiten)

            if time.monotonic() - zuletzt_aufgeraeumt > AUFRAEUMEN_INTERVALL_SEKUNDEN:
                await asyncio.to_thread(aufraeumen)
                zuletzt_aufgeraeumt = time.monotonic()
        except Exception:  # noqa: BLE001 - die Schleife darf nie enden
            # Typisch: Datenbank nicht erreichbar oder Migration 0009 fehlt.
            # Lange Pause, damit das Journal nicht vollläuft.
            log.exception("Versandschleife: Durchlauf fehlgeschlagen, nächster in %s s.", FEHLER_PAUSE_SEKUNDEN)
            await asyncio.sleep(FEHLER_PAUSE_SEKUNDEN)
            continue

        await asyncio.sleep(SCHLEIFEN_INTERVALL_SEKUNDEN)