Repository: PrefectHQ/prefect
Pull request: Fix authenticate bug and add unverified context
<!-- 
Thanks for opening a pull request to Prefect! We've got a few requests to help us review contributions:

- Make sure that your title neatly summarizes the proposed changes.
- Provide a short overview of the change and the value it adds.
- Share an example to help us understand the change in user experience.
- Confirm that you've done common tasks so we can give a timely review.
- Review our contribution guidelines: https://docs.prefect.io/latest/contributing/overview/

Happy engineering!
-->

<!-- Include an overview here -->
### Description
This PR introduces a new configuration parameter "verify", which determines whether an unverified SSL context should be used. By default, this parameter is set to True.

#### Purpose
When using custom certificates for the email server, email functionality may fail due to certificate verification issues. To address this, the new "verify" parameter allows users to initialize an unverified SSL context, facilitating successful email transmission in such cases.

#### Bug Fix
Additionally, this PR fixes a bug related to the handling of insecure email servers, ensuring we will authenticate in the case of having username and password.

#### Related PRs and issues in prefect_email repository
PR [81](https://github.com/PrefectHQ/prefect-email/pull/81)
Issue [13089](https://github.com/PrefectHQ/prefect/issues/13089)
PR [70](https://github.com/PrefectHQ/prefect-email/pull/70)
Issue [13091](https://github.com/PrefectHQ/prefect/issues/13091)
### Example
<!-- 
Share an example of the change in action.

A code blurb is best. Changes to features should include an example that is executable by a new user.
If changing documentation, a link to a preview of the page is great.
 -->
```python
from prefect import flow
from prefect.context import get_run_context
from prefect_email import EmailServerCredentials, email_send_message

credentials = EmailServerCredentials(
    username="EMAIL-ADDRESS-PLACEHOLDER",
    password="PASSWORD-PLACEHOLDER",
    smtp_type="STARTTLS",
    verify=False
)
credentials.save("email-server-credentials")

def notify_exc_by_email(exc):
    context = get_run_context()
    flow_run_name = context.flow_run.name
    email_server_credentials = EmailServerCredentials.load("email-server-credentials")
    email_send_message(
        email_server_credentials=email_server_credentials,
        subject=f"Flow run {flow_run_name!r} failed",
        msg=f"Flow run {flow_run_name!r} failed due to {exc}.",
        email_to=email_server_credentials.username,
    )

@flow
def example_flow():
    try:
        1 / 0
    except Exception as exc:
        notify_exc_by_email(exc)
        raise
```
### Checklist
<!-- These boxes may be checked after opening the pull request. -->

- [x] This pull request references any related issue by including "closes `<link to issue>`"
	- closes [13091](https://github.com/PrefectHQ/prefect/issues/13091)
	- closes [13089](https://github.com/PrefectHQ/prefect/issues/13089)
- [x] If this pull request adds new functionality, it includes unit tests that cover the changes
- [ ] This pull request includes a label categorizing the change e.g. `maintenance`, `fix`, `feature`, `enhancement`, `docs`.
  <!-- If you do not have permission to add a label, a maintainer will add one for you -->

For documentation changes:

- [ ] This pull request includes redirect settings in `netlify.toml` for files that are removed or renamed.

For new functions or classes in the Python SDK:

- [x] This pull request includes helpful docstrings.
- [ ] If a new Python file was added, this pull request contains a stub page in the Python SDK docs and an entry in `mkdocs.yml` navigation.


Problem statement:
Add authentication to INSECURE server
<!-- Thanks for submitting an issue! Please describe the issue. -->

# Expectation / Proposal

We don't have authentication in INSECURE servers so we are not able to connect to those mail servers with prefect-email.



Support of unverified smtp server
<!-- Thanks for submitting an issue! Please describe the issue. -->

# Expectation / Proposal

Self-signed certificates are often used in private or isolated network environments where obtaining a trusted certificate authority-signed certificate is not feasible or necessary. As a result, Prefect Email users who operate in such environments are unable to use the Email feature due to the current lack of support for self-signed SMTP servers. We should add an attribute to credentials which indicates the server is self-signed or not.

Hints from issue discussion:

Commit message: comments applied

Patch to review:
diff --git a/src/integrations/prefect-email/prefect_email/credentials.py b/src/integrations/prefect-email/prefect_email/credentials.py
index 1fdc8bc061f4..9f8d766fc925 100644
--- a/src/integrations/prefect-email/prefect_email/credentials.py
+++ b/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
diff --git a/src/integrations/prefect-email/tests/test_credentials.py b/src/integrations/prefect-email/tests/test_credentials.py
index b49fb6899b92..7a2b732efc31 100644
--- a/src/integrations/prefect-email/tests/test_credentials.py
+++ b/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


Code review comments:
diff_hunk: @@ -95,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()
path: src/integrations/prefect-email/prefect_email/credentials.py
text: @user1:
```suggestion
```

diff_hunk: @@ -191,16 +193,21 @@ def example_get_server_flow():
             server = SMTP(smtp_server, smtp_port)
         else:
             context = (
+                ssl.create_default_context() if self.verify else
                 ssl._create_unverified_context(protocol=ssl.PROTOCOL_TLS_CLIENT)
-                if self.unverified_context
-                else ssl.create_default_context()
             )
             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:
+            if not self.verify or smtp_type == SMTPType.INSECURE:
+                self._logger.warning(
path: src/integrations/prefect-email/prefect_email/credentials.py
text: @user1:
We'll want to get the run logger here to ensure it is populated with details of the current run.
```suggestion
                get_run_logger().warning(
```