IMAP ID extension support for 163/NetEase mailboxes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #22796 fix(email): send IMAP ID extension to support 163/NetEase mailbox (salvage #22528)

  merged to main

  files: gateway/platforms/email.py, tests/gateway/test_email.py

  Show the diff

  • gateway/platforms/email.py

    @@ -65,6 +65,29 @@
     # 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:
    +        try:
    +            from hermes_cli import __version__ as _hermes_version
    +        except Exception:  # noqa: BLE001 — keep ID best-effort if import fails
    +            _hermes_version = "0"
    +        imap.xatom(
    +            "ID",
    +            f'("name" "hermes-agent" "version" "{_hermes_version}" '
    +            '"vendor" "NousResearch" '
    +            '"support-email" "[redacted:email]")',
    +        )
    +    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 +299,7 @@ async def connect(self) -> bool:
                 # 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 +368,7 @@ def _fetch_new_messages(self) -> List[Dict[str, Any]]:
                 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")

  • tests/gateway/test_email.py

    @@ -1131,5 +1131,80 @@ def test_imap_logout_called_on_early_return(self):
             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": "[redacted:email]",
    +            "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()