Merge branch 'main' into issue-14113-final-final

src/integrations/prefect-email/prefect_email/credentials.py

@@ -9,6 +9,9 @@
 from pydantic import Field, SecretStr, field_validator
 
 from prefect.blocks.core import Block
+from prefect.logging.loggers import get_logger, get_run_logger
+
+internal_logger = get_logger(__name__)
 
 
 class SMTPType(Enum):
@@ -82,6 +85,7 @@ class EmailServerCredentials(Block):
             keys from the built-in SMTPServer Enum members, like "gmail".
         smtp_type: Either "SSL", "STARTTLS", or "INSECURE".
         smtp_port: If provided, overrides the smtp_type's default port number.
+        verify: If `False`, SSL certificates will not be verified. Default to `True`.
 
     Example:
         Load stored email server credentials:
@@ -128,6 +132,13 @@ class EmailServerCredentials(Block):
         title="SMTP Port",
     )
 
+    verify: Optional[bool] = Field(
+        default=True,
+        description=(
+            "If `False`, SSL certificates will not be verified. Default to `True`."
+        ),
+    )
+
     @field_validator("smtp_server", mode="before")
     def _cast_smtp_server(cls, value: str):
         """
@@ -182,13 +193,27 @@ def example_get_server_flow():
         if smtp_type == SMTPType.INSECURE:
             server = SMTP(smtp_server, smtp_port)
         else:
-            context = ssl.create_default_context()
+            context = (
+                ssl.create_default_context()
+                if self.verify
+                else ssl._create_unverified_context(protocol=ssl.PROTOCOL_TLS_CLIENT)
+            )
             if smtp_type == SMTPType.SSL:
                 server = SMTP_SSL(smtp_server, smtp_port, context=context)
             elif smtp_type == SMTPType.STARTTLS:
                 server = SMTP(smtp_server, smtp_port)
                 server.starttls(context=context)
-            if self.username is not None:
-                server.login(self.username, self.password.get_secret_value())
+        if self.username is not None:
+            if not self.verify or smtp_type == SMTPType.INSECURE:
+                try:
+                    logger = get_run_logger()
+                except Exception:
+                    logger = internal_logger
+                logger.warning(
+                    """SMTP login is not secure without a verified SSL/TLS or SECURE connection. 
+                    Without such a connection, the password may be sent in plain text, 
+                    making it vulnerable to interception."""
+                )
+            server.login(self.username, self.password.get_secret_value())
 
         return server

src/integrations/prefect-email/tests/test_credentials.py

@@ -91,3 +91,21 @@ def test_email_service_credentials_roundtrip_smtp_type_enum(smtp, smtp_type):
     assert credentials.smtp_type == SMTPType.STARTTLS
     server = credentials.get_server()
     assert server.port == 587
+
+
+@pytest.mark.parametrize("smtp_type", [SMTPType.STARTTLS, "STARTTLS", 587])
+@pytest.mark.parametrize("verify", [False])
+def test_email_service_credentials_unverified_context(smtp, smtp_type, verify):
+    email_server_credentials = EmailServerCredentials(
+        smtp_server="us-smtp-outbound-1.mimecast.com",
+        smtp_type=smtp_type,
+        username="username",
+        password="password",
+        verify=verify,
+    )
+    email_server_credentials.save("email-credentials", overwrite=True)
+    credentials = EmailServerCredentials.load("email-credentials")
+    assert credentials.smtp_type == SMTPType.STARTTLS
+    assert credentials.verify is False
+    server = credentials.get_server()
+    assert server.port == 587