298 lines
11 KiB
Python
298 lines
11 KiB
Python
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",
|
|
)
|