SMTP port 465 implicit TLS support in email adapter

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #46084 fix(email): use implicit TLS for port 465

  merged to main

  files: gateway/platforms/email.py, tests/gateway/test_email.py

  Show the diff

  • gateway/platforms/email.py

    @@ -22,6 +22,7 @@
     import os
     import re
     import smtplib
    +import socket
     import ssl
     import uuid
     from email.header import decode_header
    @@ -62,6 +63,63 @@
     # Gmail-safe max length per email body
     MAX_MESSAGE_LENGTH = 50_000
     
    +SMTP_CONNECT_TIMEOUT = 30
    +
    +
    +def _create_ipv4_connection(
    +    host: str,
    +    port: int,
    +    timeout: float,
    +    source_address: Any = None,
    +) -> socket.socket:
    +    """Create a TCP connection using only IPv4 addresses.
    +
    +    This mirrors ``socket.create_connection`` but constrains DNS resolution to
    +    ``AF_INET``.  It avoids mutating process-global socket functions, which
    +    matters because email sends run in executor threads.
    +    """
    +    last_error: OSError | None = None
    +    for family, socktype, proto, _canonname, sockaddr in socket.getaddrinfo(
    +        host, port, socket.AF_INET, socket.SOCK_STREAM
    +    ):
    +        sock = socket.socket(family, socktype, proto)
    +        sock.settimeout(timeout)
    +        try:
    +            if source_address:
    +                sock.bind(source_address)
    +            sock.connect(sockaddr)
    +            return sock
    +        except OSError as exc:
    +            last_error = exc
    +            sock.close()
    +    if last_error is not None:
    +        raise last_error
    +    raise OSError(f"No IPv4 address found for {host}:{port}")
    +
    +
    +class _IPv4SMTP(smtplib.SMTP):
    +    def _get_socket(self, host, port, timeout):  # type: ignore[override]
    +        return _create_ipv4_connection(
    +            host,
    +            port,
    +            timeout,
    +            source_address=self.source_address,
    +        )
    +
    +
    +class _IPv4SMTP_SSL(smtplib.SMTP_SSL):
    +    def _get_socket(self, host, port, timeout):  # type: ignore[override]
    +        raw_sock = _create_ipv4_connection(
    +            host,
    +            port,
    +            timeout,
    +            source_address=self.source_address,
    +        )
    +        return self.context.wrap_socket(
    +            raw_sock,
    +            server_hostname=getattr(self, "_host", host),
    +        )
    +
     # Supported image extensions for inline detection
     _IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
     
    @@ -293,6 +351,48 @@ def _trim_seen_uids(self) -> None:
                 # Fallback: just clear old entries if sort fails
                 self._seen_uids = set(list(self._seen_uids)[-self._seen_uids_max // 2:])
     
    +    def _connect_smtp(self) -> smtplib.SMTP:
    +        """Create an SMTP connection, selecting the correct protocol for the port.
    +
    +        Port 465 uses implicit TLS (``SMTP_SSL``).  All other ports use
    +        ``SMTP`` + ``STARTTLS``.
    +
    +        When the host resolves to an IPv6 address that is unreachable
    +        (common on networks without IPv6 routing), the default connection can
    +        hang until the socket timeout expires.  We retry connection-level
    +        failures through an IPv4-only socket path, without mutating global
    +        resolver state.  TLS verification errors are not retried.
    +
    +        Returns a connected SMTP object with TLS established — callers
    +        can proceed directly to ``login()``.
    +        """
    +        ctx = ssl.create_default_context()
    +        host = self._smtp_host
    +        port = self._smtp_port
    +
    +        def _connect(*, ipv4_only: bool = False) -> smtplib.SMTP:
    +            """Attempt one SMTP connection."""
    +            smtp_cls = _IPv4SMTP if ipv4_only else smtplib.SMTP
    +            smtp_ssl_cls = _IPv4SMTP_SSL if ipv4_only else smtplib.SMTP_SSL
    +            if port == 465:
    +                return smtp_ssl_cls(host, port, timeout=SMTP_CONNECT_TIMEOUT, context=ctx)
    +            smtp = smtp_cls(host, port, timeout=SMTP_CONNECT_TIMEOUT)
    +            try:
    +                smtp.starttls(context=ctx)
    +            except Exception:
    +                smtp.close()
    +                raise
    +            return smtp
    +
    +        try:
    +            return _connect()
    +        except (socket.timeout, TimeoutError, ConnectionError, OSError) as exc:
    +            if isinstance(exc, ssl.SSLError):
    +                raise
    +            # Connection-level failure (may be unreachable IPv6).
    +            # Retry with IPv4 only.
    +            return _connect(ipv4_only=True)
    +
         async def connect(self) -> bool:
             """Connect to the IMAP server and start polling for new messages."""
             try:
    @@ -316,10 +416,11 @@ async def connect(self) -> bool:
     
             try:
                 # Test SMTP connection
    -            smtp = smtplib.SMTP(self._smtp_host, self._smtp_port, timeout=30)
    -            smtp.starttls(context=ssl.create_default_context())
    -            smtp.login(self._address, self._password)
    -            smtp.quit()
    +            smtp = self._connect_smtp()
    +            try:
    +                smtp.login(self._address, self._password)
    +            finally:
    +                smtp.quit()
                 logger.info("[Email] SMTP connection test passed.")
             except Exception as e:
                 logger.error("[Email] SMTP connection failed: %s", e)
    @@ -555,9 +656,8 @@ def _send_email(
     
             msg.attach(MIMEText(body, "plain", "utf-8"))
     
    -        smtp = smtplib.SMTP(self._smtp_host, self._smtp_port, timeout=30)
    +        smtp = self._connect_smtp()
             try:
    -            smtp.starttls(context=ssl.create_default_context())
                 smtp.login(self._address, self._password)
                 smtp.send_message(msg)
             finally:
    @@ -677,9 +777,8 @@ def _send_email_with_attachments(
                 except Exception as e:
                     logger.warning("[Email] Failed to attach %s: %s", file_path, e)
     
    -        smtp = smtplib.SMTP(self._smtp_host, self._smtp_port, timeout=30)
    +        smtp = self._connect_smtp()
             try:
    -            smtp.starttls(context=ssl.create_default_context())
                 smtp.login(self._address, self._password)
                 smtp.send_message(msg)
             finally:
    @@ -756,9 +855,8 @@ def _send_email_with_attachment(
                 part.add_header("Content-Disposition", f"attachment; filename={fname}")
                 msg.attach(part)
     
    -        smtp = smtplib.SMTP(self._smtp_host, self._smtp_port, timeout=30)
    +        smtp = self._connect_smtp()
             try:
    -            smtp.starttls(context=ssl.create_default_context())
                 smtp.login(self._address, self._password)
                 smtp.send_message(msg)
             finally:

  • tests/gateway/test_email.py

    @@ -18,7 +18,7 @@
     from email.mime.multipart import MIMEMultipart
     from email.mime.base import MIMEBase
     from email import encoders
    -from unittest.mock import patch, MagicMock, AsyncMock
    +from unittest.mock import patch, MagicMock, AsyncMock, ANY
     
     from gateway.platforms.base import SendResult
     
    @@ -1265,5 +1265,123 @@ def test_send_imap_id_swallows_errors_for_non_supporting_servers(self):
             mock_imap.xatom.assert_called_once()
     
     
    +class TestConnectSmtp(unittest.TestCase):
    +    """Test _connect_smtp() helper: protocol selection and IPv6 fallback."""
    +
    +    def _make_adapter(self, port="587"):
    +        from gateway.config import PlatformConfig
    +        with patch.dict(os.environ, {
    +            "EMAIL_ADDRESS": "[redacted:email]",
    +            "EMAIL_PASSWORD": "secret",
    +            "EMAIL_IMAP_HOST": "imap.test.com",
    +            "EMAIL_SMTP_HOST": "smtp.test.com",
    +            "EMAIL_SMTP_PORT": port,
    +        }):
    +            from gateway.platforms.email import EmailAdapter
    +            return EmailAdapter(PlatformConfig(enabled=True))
    +
    +    def test_port_587_uses_smtp_with_starttls(self):
    +        """Port 587 should use smtplib.SMTP + STARTTLS."""
    +        adapter = self._make_adapter("587")
    +
    +        with patch("smtplib.SMTP") as mock_smtp, \
    +             patch("smtplib.SMTP_SSL") as mock_smtp_ssl:
    +            mock_server = MagicMock()
    +            mock_smtp.return_value = mock_server
    +
    +            result = adapter._connect_smtp()
    +
    +            mock_smtp.assert_called_once()
    +            mock_smtp_ssl.assert_not_called()
    +            mock_server.starttls.assert_called_once()
    +            self.assertIs(result, mock_server)
    +
    +    def test_port_465_uses_smtp_ssl(self):
    +        """Port 465 should use smtplib.SMTP_SSL (implicit TLS)."""
    +        adapter = self._make_adapter("465")
    +
    +        with patch("smtplib.SMTP") as mock_smtp, \
    +             patch("smtplib.SMTP_SSL") as mock_smtp_ssl:
    +            mock_server = MagicMock()
    +            mock_smtp_ssl.return_value = mock_server
    +
    +            result = adapter._connect_smtp()
    +
    +            mock_smtp_ssl.assert_called_once()
    +            mock_smtp.assert_not_called()
    +            self.assertIs(result, mock_server)
    +
    +    def test_ipv6_timeout_falls_back_to_ipv4(self):
    +        """When default connection times out, retry with an IPv4-only SMTP path."""
    +        import socket as _socket
    +        from gateway.platforms import email as email_mod
    +
    +        adapter = self._make_adapter("587")
    +
    +        with patch("smtplib.SMTP", side_effect=_socket.timeout("timed out")), \
    +             patch.object(email_mod, "_IPv4SMTP") as mock_ipv4_smtp:
    +            mock_server = MagicMock()
    +            mock_ipv4_smtp.return_value = mock_server
    +
    +            result = adapter._connect_smtp()
    +
    +            self.assertIs(result, mock_server)
    +            mock_ipv4_smtp.assert_called_once_with("smtp.test.com", 587, timeout=30)
    +            mock_server.starttls.assert_called_once()
    +
    +    def test_port_465_ipv6_fallback(self):
    +        """Port 465 IPv6 timeout falls back to IPv4 with SMTP_SSL."""
    +        import socket as _socket
    +        from gateway.platforms import email as email_mod
    +
    +        adapter = self._make_adapter("465")
    +
    +        with patch("smtplib.SMTP_SSL", side_effect=_socket.timeout("timed out")), \
    +             patch.object(email_mod, "_IPv4SMTP_SSL") as mock_ipv4_smtp_ssl:
    +            mock_server = MagicMock()
    +            mock_ipv4_smtp_ssl.return_value = mock_server
    +
    +            result = adapter._connect_smtp()
    +
    +            self.assertIs(result, mock_server)
    +            mock_ipv4_smtp_ssl.assert_called_once_with(
    +                "smtp.test.com", 465, timeout=30, context=ANY,
    +            )
    +
    +    def test_tls_verification_error_does_not_retry_ipv4(self):
    +        """Certificate failures are security errors, not IPv6 reachability failures."""
    +        import ssl as _ssl
    +        from gateway.platforms import email as email_mod
    +
    +        adapter = self._make_adapter("465")
    +
    +        with patch("smtplib.SMTP_SSL", side_effect=_ssl.SSLError("cert verify failed")), \
    +             patch.object(email_mod, "_IPv4SMTP_SSL") as mock_ipv4_smtp_ssl:
    +            with self.assertRaises(_ssl.SSLError):
    +                adapter._connect_smtp()
    +
    +            mock_ipv4_smtp_ssl.assert_not_called()
    +
    +    def test_ipv4_connection_does_not_mutate_global_resolver(self):
    +        """IPv4 fallback must not monkeypatch process-global socket state."""
    +        import socket as _socket
    +        from gateway.platforms.email import _create_ipv4_connection
    +
    +        original_getaddrinfo = _socket.getaddrinfo
    +        fake_sock = MagicMock()
    +
    +        with patch(
    +            "socket.getaddrinfo",
    +            return_value=[(_socket.AF_INET, _socket.SOCK_STREAM, 6, "", ("192.0.2.1", 587))],
    +        ) as mock_getaddrinfo, patch("socket.socket", return_value=fake_sock):
    +            result = _create_ipv4_connection("smtp.test.com", 587, 30)
    +
    +        self.assertIs(result, fake_sock)
    +        mock_getaddrinfo.assert_called_once_with(
    +            "smtp.test.com", 587, _socket.AF_INET, _socket.SOCK_STREAM,
    +        )
    +        self.assertIs(_socket.getaddrinfo, original_getaddrinfo)
    +
    +
     if __name__ == "__main__":
         unittest.main()