Fix authenticate bug and add unverified context

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

@@ -5,6 +5,7 @@
 from functools import partial
 from smtplib import SMTP, SMTP_SSL
 from typing import Optional, Union
+from prefect.logging import get_run_logger
 
 from pydantic import Field, SecretStr, field_validator
 
@@ -82,6 +83,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:
@@ -94,6 +96,7 @@ class EmailServerCredentials(Block):
     _block_type_name = "Email Server Credentials"
     _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/82bc6ed16ca42a2252a5512c72233a253b8a58eb-250x250.png"  # noqa
     _documentation_url = "https://prefecthq.github.io/prefect-email/credentials/#prefect_email.credentials.EmailServerCredentials"  # noqa
+    _logger = get_run_logger()
 
     username: Optional[str] = Field(
         default=None,
@@ -128,6 +131,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 +192,22 @@ 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:
+                self._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,23 @@ 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