from __future__ import annotations import copy import logging import smtplib import ssl from dataclasses import dataclass from email.message import EmailMessage from email.utils import formataddr from govoplan_core.security.outbound_http import ( OutboundHttpError, create_outbound_connection, validate_outbound_host, ) from govoplan_mail.backend.config import SmtpConfig, TransportSecurity from govoplan_mail.backend.dev.mock_mailbox import ( consume_fail_next_smtp, get_failures, is_mock_smtp_host, record_smtp_delivery, ) logger = logging.getLogger(__name__) class _OutboundPolicySMTP(smtplib.SMTP): def _get_socket(self, host: str, port: int, timeout: float | None): # type: ignore[no-untyped-def] return create_outbound_connection( host, port, timeout=timeout, source_address=self.source_address, label="SMTP connector", ) class _OutboundPolicySMTPSSL(smtplib.SMTP_SSL): def _get_socket(self, host: str, port: int, timeout: float | None): # type: ignore[no-untyped-def] sock = create_outbound_connection( host, port, timeout=timeout, source_address=self.source_address, label="SMTP connector", ) try: return self.context.wrap_socket(sock, server_hostname=self._host) except Exception: sock.close() raise class SmtpConfigurationError(ValueError): """Raised when SMTP settings are incomplete or inconsistent.""" class SmtpSendError(RuntimeError): """Raised when an SMTP send attempt fails. ``temporary`` means an explicit response allows a later retry. An ``outcome_unknown`` failure happened after SMTP transmission may have started, so automatic retry is intentionally forbidden. """ def __init__(self, message: str, *, temporary: bool = False, outcome_unknown: bool = False): super().__init__(message) self.temporary = temporary self.outcome_unknown = outcome_unknown @dataclass(frozen=True, slots=True) class SmtpLoginTestResult: host: str port: int security: str authenticated: bool @dataclass(frozen=True, slots=True) class SmtpSendResult: host: str port: int security: str envelope_from: str envelope_recipients: list[str] refused_recipients: dict[str, tuple[int, bytes | str]] @property def accepted_count(self) -> int: return len(self.envelope_recipients) - len(self.refused_recipients) def _log_smtp_cleanup_failure(action: str, exc: BaseException) -> None: logger.debug("SMTP cleanup failed while %s: %s", action, exc, exc_info=True) def _require_smtp_config(config: SmtpConfig) -> tuple[str, int]: if not config.host: raise SmtpConfigurationError("SMTP host is required") if not config.port: raise SmtpConfigurationError("SMTP port is required") if bool(config.username) != bool(config.password): raise SmtpConfigurationError("SMTP username and password must be provided together, or both omitted") return config.host, config.port def _open_smtp(config: SmtpConfig) -> smtplib.SMTP: host, port = _require_smtp_config(config) try: validate_outbound_host(host, port=port, label="SMTP connector") except OutboundHttpError as exc: raise SmtpConfigurationError(str(exc)) from exc context = ssl.create_default_context() try: if config.security == TransportSecurity.TLS: smtp: smtplib.SMTP = _OutboundPolicySMTPSSL( host=host, port=port, timeout=config.timeout_seconds, context=context, ) smtp.ehlo() else: smtp = _OutboundPolicySMTP(host=host, port=port, timeout=config.timeout_seconds) smtp.ehlo() if config.security == TransportSecurity.STARTTLS: smtp.starttls(context=context) smtp.ehlo() if config.username and config.password: smtp.login(config.username, config.password) return smtp except Exception: # If construction/login fails after a socket was created, smtplib usually closes # on GC, but explicit cleanup is safer when the variable exists. try: smtp.quit() # type: ignore[possibly-undefined] except Exception as cleanup_exc: _log_smtp_cleanup_failure("opening connection", cleanup_exc) raise def _decode_refused(refused: dict[str, tuple[int, bytes]]) -> dict[str, tuple[int, bytes | str]]: normalized: dict[str, tuple[int, bytes | str]] = {} for recipient, (code, response) in refused.items(): try: normalized[recipient] = (code, response.decode("utf-8", errors="replace")) except AttributeError: normalized[recipient] = (code, response) return normalized def test_smtp_login(*, smtp_config: SmtpConfig) -> SmtpLoginTestResult: """Open an SMTP connection and authenticate if credentials are configured. This is intentionally side-effect free: it does not send a message and it never receives envelope or recipient data. It is used by the WebUI to check whether the configured transport can be reached before a campaign is built or queued. """ host, port = _require_smtp_config(smtp_config) if is_mock_smtp_host(smtp_config.host): host, port = _require_smtp_config(smtp_config) return SmtpLoginTestResult( host=host, port=port, security=smtp_config.security.value, authenticated=bool(smtp_config.username and smtp_config.password), ) smtp = _open_smtp(smtp_config) try: return SmtpLoginTestResult( host=host, port=port, security=smtp_config.security.value, authenticated=bool(smtp_config.username and smtp_config.password), ) finally: try: smtp.quit() except Exception as quit_exc: _log_smtp_cleanup_failure("testing login quit", quit_exc) try: smtp.close() except Exception as close_exc: _log_smtp_cleanup_failure("testing login close", close_exc) def prepare_test_message( message: EmailMessage, *, test_recipient: str, test_recipient_name: str | None = None, ) -> EmailMessage: """Return a safe copy of a generated campaign message for test delivery. The original recipient headers are removed so a test send cannot accidentally leak the real To/Cc list or deliver to the real recipients. The envelope recipient must also be supplied separately to send_email_message(). """ test_message = copy.deepcopy(message) for header in ["To", "Cc", "Bcc"]: if header in test_message: del test_message[header] # Replace potential previous marker headers if the user test-sends an EML twice. for header in ["X-GovOPlaN-Test-Send"]: if header in test_message: del test_message[header] test_message["To"] = formataddr((test_recipient_name or test_recipient, test_recipient)) test_message["X-GovOPlaN-Test-Send"] = "true" return test_message def _send_smtp_payload( message: EmailMessage | bytes, *, smtp_config: SmtpConfig, envelope_from: str, envelope_recipients: list[str], ) -> SmtpSendResult: host, port, recipients = _prepare_smtp_send( smtp_config=smtp_config, envelope_from=envelope_from, envelope_recipients=envelope_recipients, ) if is_mock_smtp_host(smtp_config.host): _accepted, refused = _send_mock_smtp_payload( message, smtp_config=smtp_config, envelope_from=envelope_from, envelope_recipients=recipients, ) return _smtp_send_result( smtp_config=smtp_config, host=host, port=port, envelope_from=envelope_from, envelope_recipients=recipients, refused=refused, ) refused = _send_network_smtp_payload( message, smtp_config=smtp_config, envelope_from=envelope_from, envelope_recipients=recipients, ) return _smtp_send_result( smtp_config=smtp_config, host=host, port=port, envelope_from=envelope_from, envelope_recipients=recipients, refused=refused, ) def _prepare_smtp_send( *, smtp_config: SmtpConfig, envelope_from: str, envelope_recipients: list[str], ) -> tuple[str, int, list[str]]: host, port = _require_smtp_config(smtp_config) if not envelope_from: raise SmtpConfigurationError("SMTP envelope sender is required") recipients = [recipient for recipient in envelope_recipients if recipient] if not recipients: raise SmtpConfigurationError("at least one SMTP envelope recipient is required") return host, port, recipients def _send_mock_smtp_payload( message: EmailMessage | bytes, *, smtp_config: SmtpConfig, envelope_from: str, envelope_recipients: list[str], ) -> tuple[list[str], dict[str, tuple[int, bytes]]]: if consume_fail_next_smtp(): raise SmtpSendError("Mock SMTP configured to fail the next send") failures = get_failures() reject_text = str(failures.get("smtp_reject_recipients_containing") or "").strip().lower() refused: dict[str, tuple[int, bytes]] = {} accepted = list(envelope_recipients) if reject_text: refused = { recipient: (550, b"mock recipient rejected") for recipient in envelope_recipients if reject_text in recipient.lower() } accepted = [recipient for recipient in envelope_recipients if recipient not in refused] if not accepted: raise SmtpSendError(f"all mock SMTP recipients were refused: {_decode_refused(refused)}") record_smtp_delivery( message, envelope_from=envelope_from, envelope_recipients=accepted, smtp_host=smtp_config.host, ) return accepted, refused def _send_network_smtp_payload( message: EmailMessage | bytes, *, smtp_config: SmtpConfig, envelope_from: str, envelope_recipients: list[str], ) -> dict[str, tuple[int, bytes]]: try: smtp = _open_smtp(smtp_config) except smtplib.SMTPAuthenticationError as exc: raise SmtpSendError( f"SMTP authentication failed: {exc.smtp_code} {exc.smtp_error!r}", temporary=False, ) from exc except smtplib.SMTPResponseException as exc: raise SmtpSendError( f"SMTP connection error: {exc.smtp_code} {exc.smtp_error!r}", temporary=400 <= int(exc.smtp_code) < 500, ) from exc except (OSError, smtplib.SMTPException) as exc: # No message transmission has begun yet; a later explicit retry is safe. raise SmtpSendError(f"SMTP connection failed: {exc}", temporary=True) from exc try: if isinstance(message, bytes): refused = smtp.sendmail(envelope_from, envelope_recipients, message) else: refused = smtp.send_message( message, from_addr=envelope_from, to_addrs=envelope_recipients, ) except smtplib.SMTPRecipientsRefused as exc: raise SmtpSendError( f"all SMTP recipients were refused: {_decode_refused(exc.recipients)}", temporary=False, ) from exc except smtplib.SMTPSenderRefused as exc: raise SmtpSendError( f"SMTP sender was refused: {exc.smtp_code} {exc.smtp_error!r}", temporary=400 <= int(exc.smtp_code) < 500, ) from exc except smtplib.SMTPResponseException as exc: # An explicit SMTP response means the server rejected the transaction; # retryability follows the response class. raise SmtpSendError( f"SMTP error: {exc.smtp_code} {exc.smtp_error!r}", temporary=400 <= int(exc.smtp_code) < 500, ) from exc except (OSError, smtplib.SMTPServerDisconnected, smtplib.SMTPException) as exc: # A connection loss after DATA began can happen after the server accepted # the message but before the client received the final response. raise SmtpSendError( f"SMTP outcome is unknown after transmission started: {exc}", outcome_unknown=True, ) from exc finally: try: smtp.quit() except Exception as quit_exc: _log_smtp_cleanup_failure("sending message quit", quit_exc) try: smtp.close() except Exception as close_exc: _log_smtp_cleanup_failure("sending message close", close_exc) return refused def _smtp_send_result( *, smtp_config: SmtpConfig, host: str, port: int, envelope_from: str, envelope_recipients: list[str], refused: dict[str, tuple[int, bytes]], ) -> SmtpSendResult: return SmtpSendResult( host=host, port=port, security=smtp_config.security.value, envelope_from=envelope_from, envelope_recipients=list(envelope_recipients), refused_recipients=_decode_refused(refused), ) def send_email_bytes( message_bytes: bytes, *, smtp_config: SmtpConfig, envelope_from: str, envelope_recipients: list[str], ) -> SmtpSendResult: """Send exact RFC 5322 bytes through SMTP without reserializing the message.""" return _send_smtp_payload( message_bytes, smtp_config=smtp_config, envelope_from=envelope_from, envelope_recipients=envelope_recipients, ) def send_email_message( message: EmailMessage, *, smtp_config: SmtpConfig, envelope_from: str, envelope_recipients: list[str], ) -> SmtpSendResult: """Send an EmailMessage through SMTP. This low-level function deliberately receives explicit envelope sender and recipients. Headers and SMTP envelope are related but not identical; Bcc and future bounce-address handling depend on keeping them separate. Campaign delivery uses send_email_bytes() so the generated EML is not reserialized. """ return _send_smtp_payload( message, smtp_config=smtp_config, envelope_from=envelope_from, envelope_recipients=envelope_recipients, )