[v3-2-test] Validate SMTP server certificate on STARTTLS upgrade (apache#65346)

* Validate SMTP server certificate on STARTTLS upgrade

smtplib.SMTP.starttls() does not validate the server certificate
unless an SSL context is passed. airflow.utils.email.send_mime_email
and the SMTP provider's SmtpHook (both sync get_conn and async
aget_conn) were calling starttls() without a context, so the STARTTLS
upgrade accepted any certificate and the subsequent login() call
could send credentials over a connection terminated by a MITM.

Pass the existing SSL-context machinery (the email.ssl_context
config in core and the ssl_context connection extra in the provider)
to starttls() at all three call sites. The default becomes
ssl.create_default_context(), which validates against the system's
trusted CAs. Users who intentionally use self-signed certificates
can still opt out by setting the value to "none".

Generated-by: Claude Opus 4.6 (1M context) following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions

* Add newsfragment and SMTP provider changelog for STARTTLS cert default

Document the default behaviour change introduced by passing an SSL
context to the STARTTLS upgrade: system-default CA validation now
applies to both airflow.utils.email.send_email (via
email.ssl_context) and the SMTP provider's SmtpHook (via the
ssl_context connection extra). Users who intentionally run against
self-signed SMTP servers can preserve the old behaviour by setting
the value to "none".
(cherry picked from commit 06981d4)

Co-authored-by: Jarek Potiuk <jarek@potiuk.com>
Generated-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: potiuk

airflow-core/newsfragments/65346.significant.rst

@@ -0,0 +1,3 @@
+The SMTP STARTTLS upgrade performed by ``airflow.utils.email.send_email`` now validates the SMTP server's certificate against the system's trusted CA bundle by default. Previously the ``starttls()`` call was made without an SSL context, so any certificate was accepted.
+
+Deployments that intentionally point Airflow at an SMTP server with a self-signed or otherwise non-validating certificate and need to preserve the previous behaviour must set ``email.ssl_context = "none"`` in ``airflow.cfg``. The ``"default"`` value (now also the default when the option is unset) uses :func:`ssl.create_default_context`. Previously this option applied only to the ``SMTP_SSL`` path; it now applies to the STARTTLS path as well.

airflow-core/src/airflow/utils/email.py

@@ -266,7 +266,7 @@ def send_mime_email(
                     raise
             else:
                 if smtp_starttls:
-                    smtp_conn.starttls()
+                    smtp_conn.starttls(context=_get_ssl_context())
                 if smtp_user and smtp_password:
                     smtp_conn.login(smtp_user, smtp_password)
                 log.info("Sent an alert email to %s", e_to)
@@ -292,6 +292,26 @@ def get_email_address_list(addresses: str | Iterable[str]) -> list[str]:
     raise TypeError(f"Unexpected argument type: Received '{type(addresses).__name__}'.")
 
 
+def _get_ssl_context() -> ssl.SSLContext | None:
+    """
+    Return the SSL context configured via the ``email.ssl_context`` option.
+
+    ``"default"`` produces :func:`ssl.create_default_context`; ``"none"``
+    returns ``None`` so callers that explicitly want to skip certificate
+    validation (for example, against a self-signed SMTP server in a
+    lab environment) can still do so.
+    """
+    ssl_context_string = conf.get("email", "SSL_CONTEXT")
+    if ssl_context_string == "default":
+        return ssl.create_default_context()
+    if ssl_context_string == "none":
+        return None
+    raise RuntimeError(
+        f"The email.ssl_context configuration variable must "
+        f"be set to 'default' or 'none' and is '{ssl_context_string}."
+    )
+
+
 def _get_smtp_connection(host: str, port: int, timeout: int, with_ssl: bool) -> smtplib.SMTP:
     """
     Return an SMTP connection to the specified host and port, with optional SSL encryption.
@@ -304,17 +324,7 @@ def _get_smtp_connection(host: str, port: int, timeout: int, with_ssl: bool) ->
     """
     if not with_ssl:
         return smtplib.SMTP(host=host, port=port, timeout=timeout)
-    ssl_context_string = conf.get("email", "SSL_CONTEXT")
-    if ssl_context_string == "default":
-        ssl_context = ssl.create_default_context()
-    elif ssl_context_string == "none":
-        ssl_context = None
-    else:
-        raise RuntimeError(
-            f"The email.ssl_context configuration variable must "
-            f"be set to 'default' or 'none' and is '{ssl_context_string}."
-        )
-    return smtplib.SMTP_SSL(host=host, port=port, timeout=timeout, context=ssl_context)
+    return smtplib.SMTP_SSL(host=host, port=port, timeout=timeout, context=_get_ssl_context())
 
 
 def _get_email_list_from_str(addresses: str) -> list[str]:

providers/smtp/docs/changelog.rst

@@ -27,6 +27,19 @@
 Changelog
 ---------
 
+Breaking changes
+~~~~~~~~~~~~~~~~
+
+The SMTP STARTTLS upgrade performed by ``SmtpHook.get_conn`` and ``SmtpHook.aget_conn`` now
+validates the SMTP server's certificate against the system's trusted CA bundle by default.
+Previously the ``starttls()`` call was made without an SSL context, so any certificate was
+accepted.
+
+Deployments that intentionally point ``SmtpHook`` at an SMTP server with a self-signed or
+otherwise non-validating certificate and need to preserve the previous behaviour must set the
+``ssl_context`` field in the SMTP connection extras to ``"none"``. Leaving the field unset (or
+setting it to ``"default"``) now applies ``ssl.create_default_context()`` to the STARTTLS
+upgrade as well as to the existing ``SMTP_SSL`` path.
 
 2.4.3
 .....

providers/smtp/src/airflow/providers/smtp/hooks/smtp.py

@@ -126,7 +126,7 @@ def get_conn(self) -> SmtpHook:
                         raise AirflowException("Unable to connect to smtp server")
                 else:
                     if self.smtp_starttls:
-                        self._smtp_client.starttls()
+                        self._smtp_client.starttls(context=self._build_ssl_context())
                         self._smtp_client.ehlo()
 
                     # choose auth
@@ -172,7 +172,7 @@ async def aget_conn(self) -> SmtpHook:
                         raise AirflowException("Unable to connect to smtp server")
                 else:
                     if self.smtp_starttls:
-                        await async_client.starttls()
+                        await async_client.starttls(tls_context=self._build_ssl_context())
                         await async_client.ehlo()
 
                     # choose auth
@@ -191,10 +191,27 @@ async def aget_conn(self) -> SmtpHook:
 
         return self
 
-    def _build_client_kwargs(self, is_async: bool) -> dict[str, Any]:
-        """Build kwargs appropriate for sync or async SMTP client."""
+    def _build_ssl_context(self) -> ssl.SSLContext | None:
+        """
+        Return the SSL context configured via the ``ssl_context`` connection extra.
+
+        The default (unset or ``"default"``) returns
+        :func:`ssl.create_default_context`, which validates the server
+        certificate against the system's trusted CAs. ``"none"`` returns
+        ``None`` so callers that explicitly want to skip validation (for
+        example, against a self-signed SMTP server in a lab environment)
+        can opt out.
+        """
         valid_contexts = (None, "default", "none")  # Values accepted for ssl_context configuration
+        if self.ssl_context not in valid_contexts:
+            raise RuntimeError(
+                f"The connection extra field `ssl_context` must "
+                f"be set to 'default' or 'none' but it is set to '{self.ssl_context}'."
+            )
+        return None if self.ssl_context == "none" else ssl.create_default_context()
 
+    def _build_client_kwargs(self, is_async: bool) -> dict[str, Any]:
+        """Build kwargs appropriate for sync or async SMTP client."""
         kwargs: dict[str, Any] = {"timeout": self.timeout}
 
         if self.port:
@@ -204,15 +221,11 @@ def _build_client_kwargs(self, is_async: bool) -> dict[str, Any]:
             kwargs["hostname"] = self.host
             kwargs["use_tls"] = self.use_ssl
             kwargs["start_tls"] = self.smtp_starttls if not self.use_ssl else None
+            kwargs["tls_context"] = self._build_ssl_context()
         else:
             kwargs["host"] = self.host
             if self.use_ssl:
-                if self.ssl_context not in valid_contexts:
-                    raise RuntimeError(
-                        f"The connection extra field `ssl_context` must "
-                        f"be set to 'default' or 'none' but it is set to '{self.ssl_context}'."
-                    )
-                kwargs["context"] = None if self.ssl_context == "none" else ssl.create_default_context()
+                kwargs["context"] = self._build_ssl_context()
 
         return kwargs
 

providers/smtp/tests/unit/smtp/hooks/test_smtp.py

@@ -20,6 +20,7 @@
 import json
 import os
 import smtplib
+import ssl
 import tempfile
 from email.mime.application import MIMEApplication
 from unittest import mock
@@ -535,9 +536,15 @@ def test_ehlo_called_after_starttls(self, mock_smtplib):
         with SmtpHook(smtp_conn_id=CONN_ID_NONSSL):
             pass
 
-        # Verify ehlo is called after starttls and before login
-        expected_calls = [call.starttls(), call.ehlo(), call.login(SMTP_LOGIN, SMTP_PASSWORD)]
-        assert manager.mock_calls == expected_calls
+        # Verify ehlo is called after starttls and before login,
+        # and starttls is invoked with an SSL context so certificate validation
+        # happens on the TLS upgrade.
+        assert len(manager.mock_calls) == 3
+        starttls_call, ehlo_call, login_call = manager.mock_calls
+        assert starttls_call[0] == "starttls"
+        assert isinstance(starttls_call.kwargs.get("context"), ssl.SSLContext)
+        assert ehlo_call == call.ehlo()
+        assert login_call == call.login(SMTP_LOGIN, SMTP_PASSWORD)
 
 
 @pytest.mark.asyncio
@@ -626,13 +633,14 @@ async def test_async_connection(
         async with SmtpHook(smtp_conn_id=conn_id) as hook:
             assert hook is not None
 
-        mock_smtp.assert_called_once_with(
-            hostname=SMTP_HOST,
-            port=expected_port,
-            timeout=DEFAULT_TIMEOUT,
-            use_tls=expected_ssl,
-            start_tls=None if expected_ssl else True,
-        )
+        mock_smtp.assert_called_once()
+        call_kwargs = mock_smtp.call_args.kwargs
+        assert call_kwargs["hostname"] == SMTP_HOST
+        assert call_kwargs["port"] == expected_port
+        assert call_kwargs["timeout"] == DEFAULT_TIMEOUT
+        assert call_kwargs["use_tls"] == expected_ssl
+        assert call_kwargs["start_tls"] == (None if expected_ssl else True)
+        assert isinstance(call_kwargs["tls_context"], ssl.SSLContext)
 
         if expected_ssl:
             assert mock_smtp_client.starttls.await_count == 1