303 lines
11 KiB
Python
303 lines
11 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Outbound mail, behind an interface.
|
|
|
|
There is no SMTP server configured on this host, and inventing one is not this
|
|
module's job. So the transport is pluggable and the default for a non-production
|
|
environment is explicit about what it is: :class:`LoggingMailer` does not send
|
|
anything. It writes the message to the application log, which makes the
|
|
verification flow fully exercisable end to end during development, and it says
|
|
so in its own output so nobody mistakes a logged link for a delivered one.
|
|
|
|
:class:`SmtpMailer` is the real transport. It is implemented, not stubbed, but
|
|
it is unexercised here because no SMTP configuration exists to exercise it
|
|
against, and the README records that.
|
|
|
|
Nothing in this module ever logs a message body at INFO in production, and no
|
|
transport is allowed to raise with the recipient list in the message.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import smtplib
|
|
import ssl
|
|
from dataclasses import dataclass
|
|
from email.message import EmailMessage
|
|
from email.utils import formataddr, parseaddr
|
|
from typing import Protocol
|
|
|
|
LOGGER = logging.getLogger("umbratrans_web.mailer")
|
|
|
|
#: Verification mail is transactional and small. A cap keeps a bug from
|
|
#: generating megabyte messages.
|
|
MAX_SUBJECT_CHARS = 200
|
|
MAX_BODY_CHARS = 20_000
|
|
|
|
|
|
class MailerError(RuntimeError):
|
|
"""Delivery failed. Safe to show a caller; never contains a token."""
|
|
|
|
|
|
class Mailer(Protocol):
|
|
"""One method, so a transport can be swapped or faked in tests."""
|
|
|
|
def send(self, *, to: str, subject: str, body: str) -> None: ...
|
|
|
|
def verify(self) -> bool:
|
|
"""Whether this transport can actually deliver right now.
|
|
|
|
Configuration is not evidence. A host and port and a password can all be
|
|
present and still be refused by the server, and that is the ordinary case
|
|
for SMTP: Gmail rejects an account password outright and wants an app
|
|
password instead, without saying so in a way a person would notice.
|
|
|
|
So a deployment checks rather than assumes. Without this, a deployment
|
|
with a rejected credential reports that mail works, the control panel
|
|
allows account creation, and every account created is stranded in
|
|
``invited`` with a confirmation link that was never delivered and cannot
|
|
be re-sent. That is a data problem created by a configuration problem, and
|
|
it is invisible until somebody notices the accounts never activate.
|
|
|
|
Returns a bool rather than raising: a transport being briefly unreachable
|
|
should not stop the service from serving pages or accepting sign-in.
|
|
"""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SentMessage:
|
|
"""What a transport reports back. Used by the development transport and tests."""
|
|
|
|
to: str
|
|
subject: str
|
|
body: str
|
|
|
|
|
|
def validate_recipient(address: str) -> str:
|
|
from .models import normalize_email
|
|
|
|
return normalize_email(address)
|
|
|
|
|
|
def _check_lengths(subject: str, body: str) -> None:
|
|
if len(subject) > MAX_SUBJECT_CHARS:
|
|
raise MailerError("Mail subject exceeds the permitted length.")
|
|
if len(body) > MAX_BODY_CHARS:
|
|
raise MailerError("Mail body exceeds the permitted length.")
|
|
|
|
|
|
class LoggingMailer:
|
|
"""Writes the message to the log instead of sending it.
|
|
|
|
The single most important property is that it says what it is. A log line
|
|
that looks like a delivered mail would let someone believe an address was
|
|
verified when nothing left the machine.
|
|
"""
|
|
|
|
transport_name = "log"
|
|
|
|
def send(self, *, to: str, subject: str, body: str) -> None:
|
|
_check_lengths(subject, body)
|
|
recipient = validate_recipient(to)
|
|
LOGGER.warning(
|
|
"DEV TRANSPORT: mail was NOT sent. "
|
|
"transport=log recipient=%s subject=%s\n--- body ---\n%s\n--- end body ---",
|
|
recipient,
|
|
subject,
|
|
body,
|
|
)
|
|
|
|
def verify(self) -> bool:
|
|
"""Always False: this transport delivers nothing by design."""
|
|
|
|
return False
|
|
|
|
def last_message(self) -> SentMessage | None: # pragma: no cover - convenience
|
|
return None
|
|
|
|
|
|
class CapturingMailer:
|
|
"""Keeps messages in memory. For tests only; never selected by configuration."""
|
|
|
|
transport_name = "capture"
|
|
|
|
def __init__(self) -> None:
|
|
self.sent: list[SentMessage] = []
|
|
|
|
def send(self, *, to: str, subject: str, body: str) -> None:
|
|
_check_lengths(subject, body)
|
|
self.sent.append(SentMessage(to=validate_recipient(to), subject=subject, body=body))
|
|
|
|
def verify(self) -> bool:
|
|
return True
|
|
|
|
|
|
class SmtpMailer:
|
|
"""Real delivery over SMTP.
|
|
|
|
Implemented but unexercised in this environment: no SMTP server is
|
|
configured here. ``starttls`` is used unless the deployment explicitly sets
|
|
``starttls=False`` for a relay that is already inside a tunnel. Credentials
|
|
come from configuration and are never logged, never included in an
|
|
exception, and never written to a development data file.
|
|
"""
|
|
|
|
transport_name = "smtp"
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
host: str,
|
|
port: int = 587,
|
|
username: str | None = None,
|
|
password: str | None = None,
|
|
sender: str = "umbratrans@localhost",
|
|
starttls: bool = True,
|
|
timeout: int = 20,
|
|
) -> None:
|
|
if not host:
|
|
raise MailerError("An SMTP host is required for the smtp transport.")
|
|
self._host = host
|
|
self._port = port
|
|
self._username = username
|
|
self._password = password
|
|
self._sender = sender
|
|
self._starttls = starttls
|
|
self._timeout = timeout
|
|
|
|
def _session(self) -> "smtplib.SMTP":
|
|
"""Open a connection, upgrade it, and authenticate.
|
|
|
|
Shared by :meth:`send` and :meth:`verify` so both exercise the same path.
|
|
A probe that took a different route than real delivery would be able to
|
|
report success for a configuration that cannot actually send.
|
|
"""
|
|
|
|
client = smtplib.SMTP(self._host, self._port, timeout=self._timeout)
|
|
try:
|
|
if self._starttls:
|
|
client.starttls(context=ssl.create_default_context())
|
|
if self._username:
|
|
# Credentials go to the client, never into the log and never into
|
|
# the exception text below.
|
|
client.login(self._username, self._password or "")
|
|
except BaseException:
|
|
client.close()
|
|
raise
|
|
return client
|
|
|
|
def send(self, *, to: str, subject: str, body: str) -> None:
|
|
_check_lengths(subject, body)
|
|
recipient = validate_recipient(to)
|
|
message = EmailMessage()
|
|
message["Subject"] = subject
|
|
message["From"] = formataddr(("UmbraTrans", self._sender))
|
|
message["To"] = recipient
|
|
message.set_content(body)
|
|
|
|
try:
|
|
with self._session() as client:
|
|
client.send_message(message)
|
|
except smtplib.SMTPException as error:
|
|
# smtplib exception text can quote the server response, which can
|
|
# include an address. Report the class only.
|
|
LOGGER.error("smtp delivery failed: %s", type(error).__name__)
|
|
raise MailerError("The message could not be delivered.") from None
|
|
except (OSError, ssl.SSLError) as error:
|
|
LOGGER.error("smtp transport failed: %s", type(error).__name__)
|
|
raise MailerError("The mail server could not be reached.") from None
|
|
LOGGER.info("verification mail sent transport=smtp recipient=%s", _mask(recipient))
|
|
|
|
def verify(self) -> bool:
|
|
"""Authenticate and disconnect, without sending anything.
|
|
|
|
Reports the server's own rejection code, because the difference between
|
|
"wrong password" and "server unreachable" is exactly what an operator
|
|
needs and what a bare False throws away. The reply text is safe to log
|
|
here: it comes from the server and contains no credential, and Google's
|
|
rejection names the reason in plain words.
|
|
"""
|
|
|
|
try:
|
|
with self._session() as client:
|
|
client.noop()
|
|
except smtplib.SMTPAuthenticationError as error:
|
|
LOGGER.error(
|
|
"smtp transport rejected the configured credentials: code=%s reply=%s",
|
|
error.smtp_code,
|
|
_one_line(error.smtp_error),
|
|
)
|
|
return False
|
|
except smtplib.SMTPException as error:
|
|
LOGGER.error("smtp transport handshake failed: %s", type(error).__name__)
|
|
return False
|
|
except (OSError, ssl.SSLError) as error:
|
|
LOGGER.error("smtp transport unreachable: %s", type(error).__name__)
|
|
return False
|
|
LOGGER.info("smtp transport verified host=%s port=%d", self._host, self._port)
|
|
return True
|
|
|
|
|
|
def _one_line(value: object) -> str:
|
|
"""Collapse a server reply to a single line for logging."""
|
|
|
|
return " ".join(str(value).split())[:200]
|
|
|
|
|
|
def _mask(address: str) -> str:
|
|
"""Log a partially masked address.
|
|
|
|
A log line is a place a full address tends to end up copied out of, so the
|
|
domain is kept for diagnosis and the local part is not.
|
|
"""
|
|
|
|
_, domain = parseaddr(address)
|
|
name = domain or address.split("@")[-1]
|
|
return f"***@{name}" if name else "***"
|
|
|
|
|
|
def build_mailer(settings: object) -> Mailer:
|
|
"""Select a transport from configuration.
|
|
|
|
Production refuses the logging transport. Otherwise a production deployment
|
|
would appear to send verification mail that never left the machine, and
|
|
accounts would sit unverified with nobody noticing.
|
|
"""
|
|
|
|
environment = str(getattr(settings, "environment", "development"))
|
|
kind = str(getattr(settings, "mailer_transport", "log"))
|
|
is_production = environment in {"production", "prod"}
|
|
|
|
if kind == "smtp":
|
|
host = getattr(settings, "smtp_host", "") or ""
|
|
if not host:
|
|
raise MailerError("The smtp transport requires a host.")
|
|
return SmtpMailer(
|
|
host=host,
|
|
port=int(getattr(settings, "smtp_port", 587) or 587),
|
|
username=getattr(settings, "smtp_username", None),
|
|
password=getattr(settings, "smtp_password", None),
|
|
sender=getattr(settings, "smtp_sender", "umbratrans@localhost"),
|
|
starttls=bool(getattr(settings, "smtp_starttls", True)),
|
|
)
|
|
|
|
if is_production:
|
|
raise MailerError(
|
|
"Refusing to start in production with the logging mail transport: "
|
|
"verification links would be written to the log and never delivered."
|
|
)
|
|
if kind != "log":
|
|
raise MailerError(f"Unknown mail transport: {kind!r}.")
|
|
return LoggingMailer()
|
|
|
|
|
|
__all__ = [
|
|
"CapturingMailer",
|
|
"LoggingMailer",
|
|
"Mailer",
|
|
"MailerError",
|
|
"SmtpMailer",
|
|
"build_mailer",
|
|
]
|