Allows to choose SSL context for SMTP connection (#33070)

This change add two options to choose from when SSL SMTP connection
is created:

* default - for balance between compatibility and security
* none - in case compatibility with existing infrastructure is
  preferred

(cherry picked from commit 120efc1)

airflow/config_templates/config.yml

@@ -1862,6 +1862,22 @@ email:
       type: string
       example: "Airflow <airflow@example.com>"
       default: ~
+    ssl_context:
+      description: |
+        ssl context to use when using SMTP and IMAP SSL connections. By default, the context is "default"
+        which sets it to ``ssl.create_default_context()`` which provides the right balance between
+        compatibility and security, it however requires that certificates in your operating system are
+        updated and that SMTP/IMAP servers of yours have valid certificates that have corresponding public
+        keys installed on your machines. You can switch it to "none" if you want to disable checking
+        of the certificates, but it is not recommended as it allows MITM (man-in-the-middle) attacks
+        if your infrastructure is not sufficiently secured. It should only be set temporarily while you
+        are fixing your certificate configuration. This can be typically done by upgrading to newer
+        version of the operating system you run Airflow components on,by upgrading/refreshing proper
+        certificates in the OS or by updating certificates for your mail servers.
+      type: string
+      version_added: 2.7.0
+      example: "default"
+      default: "default"
 smtp:
   description: |
     If you want airflow to send emails on retries, failure, and you want to use

airflow/utils/email.py

@@ -21,6 +21,7 @@
 import logging
 import os
 import smtplib
+import ssl
 import warnings
 from email.mime.application import MIMEApplication
 from email.mime.multipart import MIMEMultipart
@@ -312,11 +313,20 @@ def _get_smtp_connection(host: str, port: int, timeout: int, with_ssl: bool) ->
     :param with_ssl: Whether to use SSL encryption for the connection.
     :return: An SMTP connection to the specified host and port.
     """
-    return (
-        smtplib.SMTP_SSL(host=host, port=port, timeout=timeout)
-        if with_ssl
-        else smtplib.SMTP(host=host, port=port, timeout=timeout)
-    )
+    if not with_ssl:
+        return smtplib.SMTP(host=host, port=port, timeout=timeout)
+    else:
+        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)
 
 
 def _get_email_list_from_str(addresses: str) -> list[str]:

newsfragments/33070.significant.rst

@@ -0,0 +1,8 @@
+In case of SMTP SSL connection, the default context now uses "default" context
+
+The "default" context is Python's ``default_ssl_contest`` instead of previously used "none". The
+``default_ssl_context`` provides a balance between security and compatibility but in some cases,
+when certificates are old, self-signed or misconfigured, it might not work. This can be configured
+by setting "ssl_context" in "email" configuration of Airflow. Setting it to "none" brings back
+the "none" setting that was used in Airflow 2.6 and before, but it is not recommended due to security
+reasons ad this setting disables validation of certificates and allows MITM attacks.

tests/utils/test_email.py

@@ -240,15 +240,50 @@ def test_send_mime_conn_id(self, mock_hook, mock_smtp):
 
     @mock.patch("smtplib.SMTP_SSL")
     @mock.patch("smtplib.SMTP")
-    def test_send_mime_ssl(self, mock_smtp, mock_smtp_ssl):
+    def test_send_mime_ssl_none_context(self, mock_smtp, mock_smtp_ssl):
+        mock_smtp_ssl.return_value = mock.Mock()
+        with conf_vars({("smtp", "smtp_ssl"): "True", ("email", "ssl_context"): "none"}):
+            email.send_mime_email("from", "to", MIMEMultipart(), dryrun=False)
+        assert not mock_smtp.called
+        mock_smtp_ssl.assert_called_once_with(
+            host=conf.get("smtp", "SMTP_HOST"),
+            port=conf.getint("smtp", "SMTP_PORT"),
+            timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+            context=None,
+        )
+
+    @mock.patch("smtplib.SMTP_SSL")
+    @mock.patch("smtplib.SMTP")
+    @mock.patch("ssl.create_default_context")
+    def test_send_mime_ssl_default_context_if_not_set(self, create_default_context, mock_smtp, mock_smtp_ssl):
         mock_smtp_ssl.return_value = mock.Mock()
         with conf_vars({("smtp", "smtp_ssl"): "True"}):
             email.send_mime_email("from", "to", MIMEMultipart(), dryrun=False)
         assert not mock_smtp.called
+        assert create_default_context.called
         mock_smtp_ssl.assert_called_once_with(
             host=conf.get("smtp", "SMTP_HOST"),
             port=conf.getint("smtp", "SMTP_PORT"),
             timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+            context=create_default_context.return_value,
+        )
+
+    @mock.patch("smtplib.SMTP_SSL")
+    @mock.patch("smtplib.SMTP")
+    @mock.patch("ssl.create_default_context")
+    def test_send_mime_ssl_default_context_with_value_set_to_default(
+        self, create_default_context, mock_smtp, mock_smtp_ssl
+    ):
+        mock_smtp_ssl.return_value = mock.Mock()
+        with conf_vars({("smtp", "smtp_ssl"): "True", ("email", "ssl_context"): "default"}):
+            email.send_mime_email("from", "to", MIMEMultipart(), dryrun=False)
+        assert not mock_smtp.called
+        assert create_default_context.called
+        mock_smtp_ssl.assert_called_once_with(
+            host=conf.get("smtp", "SMTP_HOST"),
+            port=conf.getint("smtp", "SMTP_PORT"),
+            timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+            context=create_default_context.return_value,
         )
 
     @mock.patch("smtplib.SMTP_SSL")
@@ -284,7 +319,6 @@ def test_send_mime_complete_failure(self, mock_smtp: mock.Mock, mock_smtp_ssl: m
         msg = MIMEMultipart()
         with pytest.raises(SMTPServerDisconnected):
             email.send_mime_email("from", "to", msg, dryrun=False)
-
         mock_smtp.assert_any_call(
             host=conf.get("smtp", "SMTP_HOST"),
             port=conf.getint("smtp", "SMTP_PORT"),
@@ -299,7 +333,8 @@ def test_send_mime_complete_failure(self, mock_smtp: mock.Mock, mock_smtp_ssl: m
 
     @mock.patch("smtplib.SMTP_SSL")
     @mock.patch("smtplib.SMTP")
-    def test_send_mime_ssl_complete_failure(self, mock_smtp, mock_smtp_ssl):
+    @mock.patch("ssl.create_default_context")
+    def test_send_mime_ssl_complete_failure(self, create_default_context, mock_smtp, mock_smtp_ssl):
         mock_smtp_ssl.side_effect = SMTPServerDisconnected()
         msg = MIMEMultipart()
         with conf_vars({("smtp", "smtp_ssl"): "True"}):
@@ -310,7 +345,9 @@ def test_send_mime_ssl_complete_failure(self, mock_smtp, mock_smtp_ssl):
             host=conf.get("smtp", "SMTP_HOST"),
             port=conf.getint("smtp", "SMTP_PORT"),
             timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+            context=create_default_context.return_value,
         )
+        assert create_default_context.called
         assert mock_smtp_ssl.call_count == conf.getint("smtp", "SMTP_RETRY_LIMIT")
         assert not mock_smtp.called
         assert not mock_smtp_ssl.return_value.starttls.called