import logging import smtplib import ssl import time from email.headerregistry import Address from email.message import EmailMessage from email.utils import format_datetime, make_msgid, parseaddr from app.config import settings from app.security import utcnow log = logging.getLogger(__name__) # Fehler, bei denen ein erneuter Versuch sinnlos ist: das Relay hat die # Nachricht endgueltig abgelehnt (4xx waere temporaer, 5xx ist permanent). _PERMANENT = (smtplib.SMTPRecipientsRefused, smtplib.SMTPSenderRefused, smtplib.SMTPAuthenticationError) def _connect() -> smtplib.SMTP: """Baut die Verbindung zum Relay auf. local_hostname landet im EHLO; ohne diesen Wert schickt Python den Container-Hostnamen, was bei strengen Filtern Punkte kostet.""" helo = settings.smtp_helo_hostname or None context = ssl.create_default_context() if settings.smtp_security == "ssl": server = smtplib.SMTP_SSL( settings.smtp_host, settings.smtp_port, local_hostname=helo, context=context, timeout=settings.smtp_timeout, ) else: server = smtplib.SMTP( settings.smtp_host, settings.smtp_port, local_hostname=helo, timeout=settings.smtp_timeout, ) if settings.smtp_security == "starttls": server.ehlo() server.starttls(context=context) server.ehlo() if settings.smtp_user: server.login(settings.smtp_user, settings.smtp_password) return server def build_message(to: str, subject: str, body: str) -> EmailMessage: """Setzt die Kopfzeilen, die Spamfilter erwarten. Fehlendes Date oder eine Message-ID mit fremder Domain sind zwei der haeufigsten Gruende, warum sonst harmlose Transaktionsmails im Spam landen.""" msg = EmailMessage() local, _, domain = parseaddr(settings.smtp_from)[1].partition("@") msg["From"] = Address(settings.from_name, local, domain) msg["To"] = to msg["Subject"] = subject msg["Date"] = format_datetime(utcnow().astimezone()) # Domain der Message-ID sollte zur DKIM-Signatur passen. msg["Message-ID"] = make_msgid(domain=settings.mail_domain) if settings.smtp_reply_to: msg["Reply-To"] = settings.smtp_reply_to # RFC 3834: automatisch erzeugte Nachricht. Verhindert Abwesenheits- # Autoresponder und Ping-Pong-Schleifen. msg["Auto-Submitted"] = "auto-generated" msg["X-Auto-Response-Suppress"] = "All" msg.set_content(body, charset="utf-8") return msg def send_mail(to: str, subject: str, body: str) -> bool: """Blockierender Versand - immer ueber BackgroundTasks aufrufen, nie direkt im Request-Pfad. Fehler werden geloggt, aber nicht nach oben gereicht: aus einem 500er koennte man sonst ableiten, ob ein Konto existiert. """ msg = build_message(to, subject, body) envelope = settings.envelope_from last: Exception | None = None for attempt in range(1, settings.smtp_max_retries + 1): try: with _connect() as server: # from_addr getrennt vom From-Header: hierueber laeuft die # SPF-Pruefung und hierhin gehen Bounces (Return-Path). server.send_message(msg, from_addr=envelope, to_addrs=[to]) log.info("Mail an %s versendet (Versuch %d).", to, attempt) return True except _PERMANENT as exc: log.error("Relay hat die Mail an %s endgueltig abgelehnt: %s", to, exc) return False except Exception as exc: last = exc if attempt < settings.smtp_max_retries: delay = 2 ** attempt log.warning( "Versand an %s fehlgeschlagen (Versuch %d/%d): %s - " "neuer Versuch in %ds", to, attempt, settings.smtp_max_retries, exc, delay, ) time.sleep(delay) log.error("Versand an %s endgueltig fehlgeschlagen: %s", to, last) return False def check_connection() -> tuple[bool, str]: """Verbindungstest ohne Versand - fuer /api/admin/mail/check.""" try: with _connect() as server: code, _ = server.noop() return True, f"Verbindung zu {settings.smtp_host}:{settings.smtp_port} ok (NOOP {code})" except Exception as exc: return False, f"{type(exc).__name__}: {exc}" def send_verification(to: str, token: str) -> None: url = f"{settings.public_base_url.rstrip('/')}/api/auth/verify?token={token}" send_mail( to, f"{settings.app_name}: E-Mail-Adresse bestätigen", "Hallo,\n\n" "bitte bestätige deine E-Mail-Adresse über den folgenden Link:\n\n" f"{url}\n\n" "Der Link ist 24 Stunden gültig.\n\n" "Wenn du dich nicht registriert hast, ignoriere diese Nachricht.\n", ) def send_password_reset(to: str, token: str) -> None: url = f"{settings.public_base_url.rstrip('/')}/reset?token={token}" send_mail( to, f"{settings.app_name}: Passwort zurücksetzen", "Hallo,\n\n" "über den folgenden Link kannst du ein neues Passwort setzen:\n\n" f"{url}\n\n" "Der Link ist 1 Stunde gültig.\n\n" "Wenn du das nicht angefordert hast, ignoriere diese Nachricht -\n" "dein Passwort bleibt dann unverändert.\n", ) def send_invitation( to: str, token: str, list_name: str, inviter: str, valid_days: int ) -> None: url = f"{settings.public_base_url.rstrip('/')}/invite?token={token}" send_mail( to, f"{settings.app_name}: {inviter} teilt die Liste „{list_name}“ mit dir", "Hallo,\n\n" f"{inviter} möchte die Einkaufsliste „{list_name}“ mit dir teilen.\n\n" "Über den folgenden Link kannst du die Einladung annehmen:\n\n" f"{url}\n\n" f"Der Link ist {valid_days} Tage gültig und gilt nur für diese " "E-Mail-Adresse.\n" "Falls du noch kein Konto hast, kannst du dir beim Öffnen des Links " "eines anlegen.\n\n" "Wenn du damit nichts anfangen kannst, ignoriere diese Nachricht -\n" "ohne den Link passiert nichts.\n", ) def send_welcome(to: str, token: str, inviter: str, valid_days: int) -> None: """Willkommensnachricht mit Link zum Setzen des Passworts. Bewusst kein vom Administrator vergebenes Passwort: Das wäre ihm bekannt und ginge zudem im Klartext per Mail. """ url = f"{settings.public_base_url.rstrip('/')}/willkommen?token={token}" send_mail( to, f"{settings.app_name}: Dein Zugang steht bereit", "Hallo,\n\n" f"{inviter} hat für dich einen Zugang zu {settings.app_name} " "eingerichtet.\n\n" "Über den folgenden Link legst du dein Passwort fest und schaltest " "den Zugang frei:\n\n" f"{url}\n\n" f"Der Link ist {valid_days} Tage gültig.\n\n" "Wenn du damit nichts anfangen kannst, ignoriere diese Nachricht - " "ohne den Link\npassiert nichts.\n", ) def send_email_change_verify(to: str, token: str, old_email: str, hours: int) -> None: """An die NEUE Adresse: Bestätigung, dass sie erreichbar ist.""" url = f"{settings.public_base_url.rstrip('/')}/adresswechsel?token={token}" send_mail( to, f"{settings.app_name}: Neue E-Mail-Adresse bestätigen", "Hallo,\n\n" f"für das Konto {old_email} soll künftig diese Adresse verwendet " "werden.\n\n" "Bitte bestätige das über den folgenden Link:\n\n" f"{url}\n\n" f"Der Link ist {hours} Stunden gültig.\n\n" "Wenn du das nicht veranlasst hast, ignoriere diese Nachricht.\n", ) def send_email_change_notice(to: str, new_email: str, hours: int) -> None: """An die ALTE Adresse: Hinweis, damit eine untergeschobene Änderung auffällt.""" send_mail( to, f"{settings.app_name}: Änderung deiner E-Mail-Adresse angefordert", "Hallo,\n\n" "für dein Konto wurde eine neue E-Mail-Adresse angefordert:\n\n" f" {new_email}\n\n" f"Sobald die neue Adresse bestätigt ist (innerhalb von {hours} " "Stunden), gilt sie\nfür die Anmeldung. Diese Adresse hier " "funktioniert dann nicht mehr.\n\n" "Wenn du das nicht veranlasst hast, wende dich bitte umgehend an " "die Administration -\ndann versucht jemand, dir den Zugang zu " "entziehen.\n", ) def send_email_change_verify_old( to: str, token: str, new_email: str, hours: int ) -> None: """An die ALTE Adresse eines Administratorkontos: hier reicht der Hinweis nicht, es wird eine ausdrückliche Zustimmung verlangt.""" url = f"{settings.public_base_url.rstrip('/')}/adresswechsel?token={token}" send_mail( to, f"{settings.app_name}: Adressänderung deines Administratorkontos bestätigen", "Hallo,\n\n" "für dein Administratorkonto wurde eine neue E-Mail-Adresse " "angefordert:\n\n" f" {new_email}\n\n" "Bei Administratorkonten muss die Änderung von BEIDEN Adressen " "bestätigt werden.\n" "Bitte bestätige hier von deiner bisherigen Adresse aus:\n\n" f"{url}\n\n" f"Der Link ist {hours} Stunden gültig.\n\n" "Wenn du das nicht veranlasst hast, klicke NICHT auf den Link. Ohne " "deine\nBestätigung bleibt die Adresse unverändert - und du " "solltest dein Passwort ändern.\n", ) def send_deactivation_notice(to: str, months: int) -> None: send_mail( to, f"{settings.app_name}: Dein Zugang wurde deaktiviert", "Hallo,\n\n" f"dein Zugang zu {settings.app_name} wurde deaktiviert, weil er " f"länger als {months} Monate\nnicht genutzt wurde.\n\n" "Deine Listen bleiben erhalten. Wende dich an die Administration, " "wenn du den\nZugang wieder brauchst.\n", ) def send_list_deletion_warning( to: str, list_name: str, days: int, archived_since: str, item_count: int ) -> None: """Vorwarnung an den Eigentümer, bevor eine archivierte Liste automatisch gelöscht wird.""" url = settings.public_base_url.rstrip("/") frist = "in 7 Tagen" if days == 7 else "morgen" send_mail( to, f"{settings.app_name}: Liste „{list_name}“ wird {frist} gelöscht", "Hallo,\n\n" f"deine Liste „{list_name}“ liegt seit {archived_since} in deinem " "Archiv und wird\n" f"{frist} automatisch gelöscht. Sie enthält derzeit {item_count} " "Eintrag/Einträge.\n\n" "Wenn du sie behalten möchtest, hol sie einfach aus dem Archiv " "zurück:\n\n" f"{url}\n\n" "Dort findest du das Archiv in der Listenübersicht. Ein einziger " "Klick auf\n„Aus dem Archiv holen“ genügt - die Frist beginnt " "dann von vorn.\n\n" "Wenn du nichts unternimmst, wird die Liste mit allen Einträgen, " "Märkten und\nWarengruppen entfernt. Die erfassten Preise bleiben " "davon unberührt.\n", ) def send_test_mail(to: str) -> None: send_mail( to, f"{settings.app_name}: Testnachricht", "Diese Nachricht bestätigt, dass der SMTP-Versand funktioniert.\n\n" f"Relay: {settings.smtp_host}:{settings.smtp_port}\n" f"Verschlüsselung: {settings.smtp_security}\n" f"From-Header: {settings.smtp_from}\n" f"Envelope-From: {settings.envelope_from}\n" f"EHLO-Hostname: {settings.smtp_helo_hostname or '(automatisch)'}\n", )