diff --git a/gateway/platforms/email.py b/gateway/platforms/email.py index 7717494de52..3cb6974c457 100644 --- a/gateway/platforms/email.py +++ b/gateway/platforms/email.py @@ -65,6 +65,25 @@ MAX_MESSAGE_LENGTH = 50_000 # Supported image extensions for inline detection _IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".webp"} +def _send_imap_id(imap: "imaplib.IMAP4") -> None: + """Send RFC 2971 IMAP ID command identifying this client. + + Required by 163/NetEase mailbox after LOGIN: without it, every UID + SEARCH/FETCH returns ``BYE Unsafe Login`` and disconnects. Other + IMAP servers either honor it silently or reject the unknown command; + we swallow failures so non-supporting servers keep working. + """ + try: + imap.xatom( + "ID", + '("name" "hermes-agent" "version" "1.0" ' + '"vendor" "NousResearch" ' + '"support-email" "noreply@nousresearch.com")', + ) + except Exception as e: # noqa: BLE001 — best-effort, never fatal + logger.debug("[Email] IMAP ID command not accepted: %s", e) + + def _is_automated_sender(address: str, headers: dict) -> bool: """Return True if this email is from an automated/noreply source.""" addr = address.lower() @@ -276,6 +295,7 @@ class EmailAdapter(BasePlatformAdapter): # Test IMAP connection imap = imaplib.IMAP4_SSL(self._imap_host, self._imap_port, timeout=30) imap.login(self._address, self._password) + _send_imap_id(imap) # Mark all existing messages as seen so we only process new ones imap.select("INBOX") status, data = imap.uid("search", None, "ALL") @@ -344,6 +364,7 @@ class EmailAdapter(BasePlatformAdapter): imap = imaplib.IMAP4_SSL(self._imap_host, self._imap_port, timeout=30) try: imap.login(self._address, self._password) + _send_imap_id(imap) imap.select("INBOX") status, data = imap.uid("search", None, "UNSEEN") diff --git a/tests/gateway/test_email.py b/tests/gateway/test_email.py index d378eecea7c..78034fe8075 100644 --- a/tests/gateway/test_email.py +++ b/tests/gateway/test_email.py @@ -1131,5 +1131,80 @@ class TestImapConnectionCleanup(unittest.TestCase): mock_imap.logout.assert_called_once() +class TestImapIdExtensionForNetEase(unittest.TestCase): + """Regression for #22271: 163/NetEase mailbox requires the RFC 2971 + IMAP ID command after LOGIN, otherwise it returns ``BYE Unsafe Login`` + on every UID SEARCH. We send ID best-effort after every login so that + 163 works while non-supporting servers stay unaffected. + """ + + def _make_adapter(self): + from gateway.config import PlatformConfig + with patch.dict(os.environ, { + "EMAIL_ADDRESS": "hermes@163.com", + "EMAIL_PASSWORD": "secret", + "EMAIL_IMAP_HOST": "imap.163.com", + "EMAIL_SMTP_HOST": "smtp.163.com", + }): + from gateway.platforms.email import EmailAdapter + adapter = EmailAdapter(PlatformConfig(enabled=True)) + return adapter + + def test_connect_sends_imap_id_after_login(self): + """connect() must call xatom('ID', ...) after LOGIN for 163 support.""" + import asyncio + adapter = self._make_adapter() + + mock_imap = MagicMock() + mock_imap.uid.return_value = ("OK", [b""]) + + with patch("imaplib.IMAP4_SSL", return_value=mock_imap), \ + patch("smtplib.SMTP") as mock_smtp: + mock_smtp.return_value = MagicMock() + asyncio.run(adapter.connect()) + adapter._running = False + if adapter._poll_task: + adapter._poll_task.cancel() + + id_calls = [c for c in mock_imap.xatom.call_args_list if c.args and c.args[0] == "ID"] + self.assertTrue( + id_calls, + "EmailAdapter.connect() must call imap.xatom('ID', ...) after " + "LOGIN so 163/NetEase mailbox does not return 'Unsafe Login'.", + ) + payload = id_calls[0].args[1] + self.assertIn("hermes-agent", payload) + + names = [c[0] for c in mock_imap.method_calls] + self.assertIn("login", names) + self.assertLess(names.index("login"), names.index("xatom")) + + def test_fetch_new_messages_sends_imap_id_after_login(self): + """_fetch_new_messages must also send ID — it opens its own IMAP session.""" + adapter = self._make_adapter() + mock_imap = MagicMock() + mock_imap.uid.return_value = ("OK", [b""]) + + with patch("imaplib.IMAP4_SSL", return_value=mock_imap): + adapter._fetch_new_messages() + + id_calls = [c for c in mock_imap.xatom.call_args_list if c.args and c.args[0] == "ID"] + self.assertTrue( + id_calls, + "_fetch_new_messages() must call imap.xatom('ID', ...) after " + "LOGIN — the polling path opens a fresh IMAP connection.", + ) + + def test_send_imap_id_swallows_errors_for_non_supporting_servers(self): + """Servers that reject ID must not break the connection.""" + from gateway.platforms.email import _send_imap_id + + mock_imap = MagicMock() + mock_imap.xatom.side_effect = Exception("BAD command unknown: ID") + + _send_imap_id(mock_imap) + mock_imap.xatom.assert_called_once() + + if __name__ == "__main__": unittest.main()