From 3f04c7a5f60bd622ae1659b04a5a5faafd75e657 Mon Sep 17 00:00:00 2001 From: Petra Date: Thu, 14 May 2026 12:11:21 +1000 Subject: [PATCH] fix(email): use implicit TLS for SMTPS port 465 --- tests/gateway/test_email.py | 70 +++++++++++++++++++++++++++++++++++++ tools/send_message_tool.py | 21 ++++++++--- 2 files changed, 86 insertions(+), 5 deletions(-) diff --git a/tests/gateway/test_email.py b/tests/gateway/test_email.py index 8cfaa22c5d32..826fd9b6d75c 100644 --- a/tests/gateway/test_email.py +++ b/tests/gateway/test_email.py @@ -22,6 +22,11 @@ from gateway.platforms.base import SendResult +# Keep this test module hermetic when run from a live Hermes environment whose +# .env may set EMAIL_SMTP_PORT=465 for AgentMail. Individual tests opt into +# the port they need via patch.dict. +os.environ.pop("EMAIL_SMTP_PORT", None) + class TestConfigEnvOverrides(unittest.TestCase): """Verify email config is loaded from environment variables.""" @@ -849,6 +854,41 @@ def test_connect_smtp_failure(self): result = asyncio.run(adapter.connect()) self.assertFalse(result) + def test_connect_uses_smtp_ssl_for_port_465(self): + """Port 465 is implicit TLS and must not use STARTTLS.""" + import asyncio + import ssl + from gateway.config import PlatformConfig + with patch.dict(os.environ, { + "EMAIL_ADDRESS": "hermes@test.com", + "EMAIL_PASSWORD": "secret", + "EMAIL_IMAP_HOST": "imap.test.com", + "EMAIL_SMTP_HOST": "smtp.test.com", + "EMAIL_SMTP_PORT": "465", + }): + from gateway.platforms.email import EmailAdapter + adapter = EmailAdapter(PlatformConfig(enabled=True)) + + mock_imap = MagicMock() + mock_imap.uid.return_value = ("OK", [b""]) + mock_server = MagicMock() + + with patch("imaplib.IMAP4_SSL", return_value=mock_imap), \ + patch("smtplib.SMTP") as mock_starttls_smtp, \ + patch("smtplib.SMTP_SSL", return_value=mock_server) as mock_smtp_ssl: + result = asyncio.run(adapter.connect()) + + self.assertTrue(result) + mock_starttls_smtp.assert_not_called() + _, _, kwargs = mock_smtp_ssl.mock_calls[0] + self.assertEqual(mock_smtp_ssl.call_args.args[:2], ("smtp.test.com", 465)) + self.assertIsInstance(kwargs["context"], ssl.SSLContext) + mock_server.login.assert_called_once_with("hermes@test.com", "secret") + mock_server.quit.assert_called_once() + adapter._running = False + if adapter._poll_task: + adapter._poll_task.cancel() + def test_disconnect_cancels_poll(self): """disconnect() should cancel the polling task.""" import asyncio @@ -1041,6 +1081,36 @@ def test_send_email_tool_success(self): self.assertEqual(send_call["To"], "user@test.com") self.assertEqual(send_call["From"], "hermes@test.com") + @patch.dict(os.environ, { + "EMAIL_ADDRESS": "hermes@test.com", + "EMAIL_PASSWORD": "secret", + "EMAIL_SMTP_HOST": "smtp.test.com", + "EMAIL_SMTP_PORT": "465", + }) + def test_send_email_tool_uses_smtp_ssl_for_port_465(self): + """_send_email should use implicit TLS for SMTPS port 465.""" + import asyncio + import ssl + from tools.send_message_tool import _send_email + + with patch("smtplib.SMTP") as mock_starttls_smtp, \ + patch("smtplib.SMTP_SSL") as mock_smtp_ssl: + mock_server = MagicMock() + mock_smtp_ssl.return_value = mock_server + + result = asyncio.run( + _send_email({"address": "hermes@test.com", "smtp_host": "smtp.test.com"}, "user@test.com", "Hello") + ) + + self.assertTrue(result["success"]) + mock_starttls_smtp.assert_not_called() + self.assertEqual(mock_smtp_ssl.call_args.args[:2], ("smtp.test.com", 465)) + self.assertIsInstance(mock_smtp_ssl.call_args.kwargs["context"], ssl.SSLContext) + mock_server.starttls.assert_not_called() + mock_server.login.assert_called_once_with("hermes@test.com", "secret") + mock_server.send_message.assert_called_once() + mock_server.quit.assert_called_once() + @patch.dict(os.environ, { "EMAIL_ADDRESS": "hermes@test.com", "EMAIL_PASSWORD": "secret", diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index 72311f87c41b..524d1d08035a 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -1438,11 +1438,22 @@ async def _send_email(extra, chat_id, message): msg["Subject"] = "Hermes Agent" msg["Date"] = formatdate(localtime=True) - server = smtplib.SMTP(smtp_host, smtp_port) - server.starttls(context=ssl.create_default_context()) - server.login(address, password) - server.send_message(msg) - server.quit() + context = ssl.create_default_context() + if smtp_port == 465: + server = smtplib.SMTP_SSL(smtp_host, smtp_port, timeout=30, context=context) + try: + server.login(address, password) + server.send_message(msg) + finally: + server.quit() + else: + server = smtplib.SMTP(smtp_host, smtp_port, timeout=30) + try: + server.starttls(context=context) + server.login(address, password) + server.send_message(msg) + finally: + server.quit() return {"success": True, "platform": "email", "chat_id": chat_id} except Exception as e: return _error(f"Email send failed: {e}")