Misc bugfixes

CHANGELOG.rst

@@ -1,6 +1,14 @@
 Changelog
 =========
 
+5.1.1 (unreleased)
+------------------
+
+- Bugfix: Timeout ignored during CRAM-MD5 verification
+- Bugfix: Only parse EHLO response after validating success
+- Bugfix: return None from extract_sender when address list is empty
+
+
 5.1.0
 -----
 

src/aiosmtplib/__init__.py

@@ -34,7 +34,7 @@
 
 
 __title__ = "aiosmtplib"
-__version__ = "5.1.0"
+__version__ = "5.1.0dev0"
 __author__ = "Cole Maclean"
 __license__ = "MIT"
 __copyright__ = "Copyright 2022 Cole Maclean"

src/aiosmtplib/email.py

@@ -141,7 +141,11 @@ def extract_sender(
     if sender_header is None:
         return None
 
-    return extract_addresses(sender_header)[0]
+    addresses = extract_addresses(sender_header)
+    if not addresses:
+        return None
+
+    return addresses[0]
 
 
 def extract_recipients(

src/aiosmtplib/smtp.py

@@ -967,11 +967,12 @@ async def ehlo(
         response = await self.execute_command(
             b"EHLO", hostname.encode("ascii"), timeout=timeout
         )
-        self.last_ehlo_response = response
 
         if response.code != SMTPStatus.completed:
             raise SMTPHeloError(response.code, response.message)
 
+        self.last_ehlo_response = response
+
         return response
 
     def supports_extension(self, extension: str, /) -> bool:
@@ -1165,7 +1166,7 @@ async def auth_crammd5(
         verification_bytes = auth_crammd5_verify(
             username, password, initial_response.message
         )
-        response = await self.execute_command(verification_bytes)
+        response = await self.execute_command(verification_bytes, timeout=timeout)
 
         if response.code != SMTPStatus.auth_successful:
             raise SMTPAuthenticationError(response.code, response.message)

tests/auth.py

@@ -12,13 +12,15 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
 
         self.received_commands: list[bytes] = []
+        self.received_kwargs: list[dict[str, Any]] = []
         self.responses: deque[tuple[int, str]] = deque()
         self.esmtp_extensions = {"auth": ""}
         self.server_auth_methods = ["cram-md5", "login", "plain"]
         self.supports_esmtp = True
 
     async def execute_command(self, *args: Any, **kwargs: Any) -> SMTPResponse:
         self.received_commands.append(b" ".join(args))
+        self.received_kwargs.append(kwargs)
 
         response = self.responses.popleft()
 

tests/test_auth_methods.py

@@ -304,3 +304,22 @@ async def get_token() -> str:
 
     encoded = auth_xoauth2_encode("[email protected]", "test_oauth_token")
     assert mock_auth.received_commands == [b"AUTH XOAUTH2 " + encoded]
+
+
+async def test_auth_crammd5_passes_timeout(mock_auth: DummySMTPAuth) -> None:
+    """
+    Test that auth_crammd5 passes timeout to the verification command.
+
+    Both execute_command calls in auth_crammd5 should receive the timeout parameter.
+    """
+    continue_response = (
+        SMTPStatus.auth_continue,
+        base64.b64encode(b"challenge").decode("utf-8"),
+    )
+    mock_auth.responses.extend([continue_response, SUCCESS_RESPONSE])
+    await mock_auth.auth_crammd5("user", "pass", timeout=5.0)
+
+    # Both commands should have received the timeout
+    assert len(mock_auth.received_kwargs) == 2
+    assert mock_auth.received_kwargs[0].get("timeout") == 5.0
+    assert mock_auth.received_kwargs[1].get("timeout") == 5.0

tests/test_commands.py

@@ -91,6 +91,25 @@ async def test_ehlo_error(smtp_client: SMTP) -> None:
         assert exception_info.value.code == SMTPStatus.unrecognized_command
 
 
+@pytest.mark.smtpd_mocks(smtp_EHLO=mock_response_unrecognized_command)
+async def test_ehlo_error_does_not_set_supports_esmtp(smtp_client: SMTP) -> None:
+    """
+    Test that a failed EHLO response does not set supports_esmtp to True.
+
+    The EHLO response should only be parsed for extensions after validating
+    that the response code indicates success.
+    """
+    async with smtp_client:
+        assert smtp_client.supports_esmtp is False
+
+        with pytest.raises(SMTPHeloError):
+            await smtp_client.ehlo()
+
+        assert smtp_client.supports_esmtp is False
+        assert smtp_client.last_ehlo_response is None
+        assert smtp_client.esmtp_extensions == {}
+
+
 @pytest.mark.smtpd_mocks(smtp_EHLO=mock_response_ehlo_full)
 async def test_ehlo_parses_esmtp_extensions(smtp_client: SMTP) -> None:
     async with smtp_client:

tests/test_email_utils.py

@@ -362,3 +362,18 @@ def test_extract_sender_valueerror_on_multiple_resent_message() -> None:
 
     with pytest.raises(ValueError):
         extract_sender(message)
+
+
+def test_extract_sender_empty_address_returns_none() -> None:
+    """
+    Test that extract_sender returns None when the From header contains no valid addresses.
+
+    If extract_addresses() returns an empty list (e.g., malformed From header),
+    extract_sender should return None instead of raising IndexError.
+    """
+    message = EmailMessage()
+    # Use undisclosed-recipients which parses to an empty list
+    message["From"] = "undisclosed-recipients:;"
+
+    sender = extract_sender(message)
+    assert sender is None