87 lines
3.4 KiB
Python
87 lines
3.4 KiB
Python
 
|
"""Send one message through the relay. STARTTLS verified against SMTP_CA_BUNDLE, never a plain-text fallback.
|
||
|
|
|
||
|
|
To: is the From address; the real recipients go in the envelope only, so nobody sees the list. No credentials.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import smtplib
|
||
|
|
import ssl
|
||
|
|
from collections.abc import Sequence
|
||
|
|
from email.message import EmailMessage
|
||
|
|
from email.utils import formatdate, make_msgid
|
||
|
|
|
||
|
|
from ..settings import Settings
|
||
|
|
|
||
|
|
TIMEOUT_SECONDS = 15
|
||
|
|
MAIL_OFF = "Mail is off: SMTP_HOST is empty."
|
||
|
|
|
||
|
|
|
||
|
|
class MailError(Exception):
|
||
|
|
"""The relay could not take the message. The text is safe to store and show (no secrets)."""
|
||
|
|
|
||
|
|
|
||
|
|
def tls_context(settings: Settings) -> ssl.SSLContext:
|
||
|
|
"""The same verified context the doctor uses: system trust plus SMTP_CA_BUNDLE when set."""
|
||
|
|
return ssl.create_default_context(cafile=str(settings.smtp_ca_bundle) if settings.smtp_ca_bundle else None)
|
||
|
|
|
||
|
|
|
||
|
|
def subject_prefix(settings: Settings) -> str:
|
||
|
|
return "[Packet Station]" if settings.stack == "prod" else "[Packet Station TEST]"
|
||
|
|
|
||
|
|
|
||
|
|
def full_subject(settings: Settings, title: str) -> str:
|
||
|
|
return f"{subject_prefix(settings)} {title}".strip()[:250]
|
||
|
|
|
||
|
|
|
||
|
|
def build_message(settings: Settings, subject: str, body: str) -> EmailMessage:
|
||
|
|
msg = EmailMessage()
|
||
|
|
msg["From"] = settings.smtp_from
|
||
|
|
msg["To"] = settings.smtp_from
|
||
|
|
msg["Subject"] = subject
|
||
|
|
msg["Date"] = formatdate(localtime=False)
|
||
|
|
domain = settings.smtp_from.rpartition("@")[2] or None
|
||
|
|
msg["Message-ID"] = make_msgid(domain=domain)
|
||
|
|
msg["Auto-Submitted"] = "auto-generated"
|
||
|
|
msg.set_content(body or subject)
|
||
|
|
return msg
|
||
|
|
|
||
|
|
|
||
|
|
def send(settings: Settings, subject: str, body: str, to: Sequence[str]) -> str | None:
|
||
|
|
"""Deliver one message to the envelope addresses `to`. Raises MailError when nothing was delivered.
|
||
|
|
|
||
|
|
Returns None, or a note naming the addresses the relay refused when it took the message for the rest."""
|
||
|
|
host = (settings.smtp_host or "").strip()
|
||
|
|
if not host:
|
||
|
|
raise MailError(MAIL_OFF)
|
||
|
|
if not to:
|
||
|
|
raise MailError("No recipients.")
|
||
|
|
msg = build_message(settings, subject, body)
|
||
|
|
try:
|
||
|
|
with smtplib.SMTP(host, settings.smtp_port, timeout=TIMEOUT_SECONDS) as smtp:
|
||
|
|
smtp.ehlo()
|
||
|
|
if settings.smtp_starttls:
|
||
|
|
# starttls raises SMTPNotSupportedError when the relay does not offer it: no fallback
|
||
|
|
smtp.starttls(context=tls_context(settings))
|
||
|
|
smtp.ehlo()
|
||
|
|
refused = smtp.send_message(msg, from_addr=settings.smtp_from, to_addrs=list(to))
|
||
|
|
except ssl.SSLError as e:
|
||
|
|
raise MailError(f"TLS: {e.reason or e}") from None
|
||
|
|
except ssl.CertificateError as e:
|
||
|
|
raise MailError(f"TLS: {e}") from None
|
||
|
|
except smtplib.SMTPNotSupportedError:
|
||
|
|
raise MailError("The relay does not offer STARTTLS; not sent.") from None
|
||
|
|
except smtplib.SMTPRecipientsRefused as e:
|
||
|
|
raise MailError(f"Every recipient was refused ({len(e.recipients)}).") from None
|
||
|
|
except smtplib.SMTPResponseException as e:
|
||
|
|
raise MailError(f"SMTP {e.smtp_code}: {_text(e.smtp_error)}") from None
|
||
|
|
except (OSError, smtplib.SMTPException) as e:
|
||
|
|
raise MailError(f"{type(e).__name__}: {e}") from None
|
||
|
|
if refused:
|
||
|
|
return f"Refused for {len(refused)} of {len(to)} recipients: {', '.join(sorted(refused))[:300]}"
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def _text(value: bytes | str) -> str:
|
||
|
|
return (value.decode("utf-8", "replace") if isinstance(value, bytes) else str(value))[:300]
|