refactor: modularize IMAP server with dual login, STARTTLS, and test suite (#859)

refactor: modularize IMAP server with fixes and E2E tests

- Modularize IMAP server into imap_server, imap_mailbox, imap_message,
  imap_http_client, parse_email, config, models
- Support dual login: JWT token and address+password via backend
- Add STARTTLS support with configurable TLS cert/key
- Fix FETCH/STORE returning UID instead of sequence number (RFC 3501)
- Implement IMessageFile.open() for correct BODY[] raw MIME delivery
- Add UIDNEXT to SELECT response via _cbSelectWork override
- Use per-restart UIDVALIDITY to force client resync
- Pass raw MIME to SimpleMessage for accurate RFC822.SIZE
- Fix SENT mailbox returning empty source
- Handle CREATE command gracefully for Thunderbird compatibility
- Add IMAP E2E tests: auth, LIST, SELECT, STATUS, FETCH, SEARCH,
  STORE, UID FETCH, BODY[] integrity, size, seq numbers, SENT mailbox
- Add SMTP E2E tests using nodemailer: send plain/HTML, auth failure,
  sendbox verification
- Add sendTestMail helper using admin/send_mail

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

smtp_proxy_server/.env.example

@@ -1,2 +1,7 @@
 proxy_url=https://temp-email-api.xxx.xxx
 port=8025
+imap_port=11143
+# imap_tls_cert=/path/to/cert.pem
+# imap_tls_key=/path/to/key.pem
+# imap_cache_size=500
+# imap_http_timeout=30.0

smtp_proxy_server/config.py

@@ -1,5 +1,6 @@
 import logging
-from pydantic_settings import BaseSettings
+from pydantic import field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
 
 logging.basicConfig(
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
@@ -14,9 +15,26 @@ class Settings(BaseSettings):
     port: int = 8025
     imap_port: int = 11143
     basic_password: str = ""
+    imap_tls_cert: str = ""
+    imap_tls_key: str = ""
+    imap_cache_size: int = 500
+    imap_http_timeout: float = 30.0
 
-    class Config:
-        env_file = ".env"
+    model_config = SettingsConfigDict(env_file=".env")
+
+    @field_validator("imap_cache_size")
+    @classmethod
+    def cache_size_positive(cls, v):
+        if v <= 0:
+            raise ValueError("imap_cache_size must be > 0")
+        return v
+
+    @field_validator("imap_http_timeout")
+    @classmethod
+    def timeout_positive(cls, v):
+        if v <= 0:
+            raise ValueError("imap_http_timeout must be > 0")
+        return v
 
 
 settings = Settings()

smtp_proxy_server/imap_http_client.py

@@ -0,0 +1,69 @@
+import logging
+
+import httpx
+from twisted.internet import defer, threads
+
+from config import settings
+
+_logger = logging.getLogger(__name__)
+_logger.setLevel(logging.INFO)
+
+
+class BackendClient:
+    """Async HTTP client for IMAP backend communication.
+
+    All public methods return Deferred via deferToThread to avoid
+    blocking the Twisted reactor with synchronous HTTP calls.
+    """
+
+    def __init__(self, password: str):
+        self.password = password
+        self._client = httpx.Client(
+            base_url=settings.proxy_url,
+            headers={
+                "Authorization": f"Bearer {password}",
+                "x-custom-auth": settings.basic_password,
+                "Content-Type": "application/json",
+            },
+            timeout=settings.imap_http_timeout,
+        )
+
+    def _get_endpoint(self, mailbox_name: str) -> str:
+        if mailbox_name == "INBOX":
+            return "/api/mails"
+        elif mailbox_name == "SENT":
+            return "/api/sendbox"
+        raise ValueError(f"Unknown mailbox: {mailbox_name}")
+
+    def _sync_get_message_count(self, mailbox_name: str) -> int:
+        endpoint = self._get_endpoint(mailbox_name)
+        res = self._client.get(f"{endpoint}?limit=1&offset=0")
+        res.raise_for_status()
+        return res.json()["count"]
+
+    def _sync_get_messages(
+        self, mailbox_name: str, limit: int, offset: int
+    ) -> tuple[list[dict], int | None]:
+        """Fetch messages from backend.
+
+        Returns (results, count) where count is only valid when offset=0.
+        """
+        endpoint = self._get_endpoint(mailbox_name)
+        res = self._client.get(f"{endpoint}?limit={limit}&offset={offset}")
+        res.raise_for_status()
+        data = res.json()
+        count = data.get("count") if offset == 0 else None
+        return data["results"], count
+
+    def get_message_count(self, mailbox_name: str) -> defer.Deferred:
+        return threads.deferToThread(self._sync_get_message_count, mailbox_name)
+
+    def get_messages(
+        self, mailbox_name: str, limit: int, offset: int
+    ) -> defer.Deferred:
+        return threads.deferToThread(
+            self._sync_get_messages, mailbox_name, limit, offset
+        )
+
+    def close(self):
+        self._client.close()

smtp_proxy_server/imap_mailbox.py

@@ -0,0 +1,357 @@
+import bisect
+import logging
+import time
+from collections import OrderedDict
+
+from twisted.internet import defer
+from twisted.mail import imap4
+from zope.interface import implementer
+
+from config import settings
+from imap_http_client import BackendClient
+from imap_message import SimpleMessage
+from parse_email import generate_email_model, parse_email
+
+_logger = logging.getLogger(__name__)
+_logger.setLevel(logging.INFO)
+
+# Use process start time as UIDVALIDITY so clients resync after restart
+_UID_VALIDITY = int(time.time())
+
+
+class MessageCache:
+    """LRU cache for parsed email messages, keyed by backend id (=UID)."""
+
+    def __init__(self, max_size: int = 500):
+        self._cache: OrderedDict[int, SimpleMessage] = OrderedDict()
+        self._max_size = max_size
+
+    def get(self, uid: int):
+        if uid in self._cache:
+            self._cache.move_to_end(uid)
+            return self._cache[uid]
+        return None
+
+    def put(self, uid: int, message: SimpleMessage):
+        if uid in self._cache:
+            self._cache.move_to_end(uid)
+            self._cache[uid] = message
+        else:
+            if len(self._cache) >= self._max_size:
+                self._cache.popitem(last=False)
+            self._cache[uid] = message
+
+    def __contains__(self, uid: int) -> bool:
+        return uid in self._cache
+
+    def __len__(self) -> int:
+        return len(self._cache)
+
+
+@implementer(imap4.IMailboxInfo, imap4.IMailbox, imap4.ISearchableMailbox)
+class SimpleMailbox:
+
+    def __init__(self, name: str, client: BackendClient):
+        self.name = name
+        self._client = client
+        self.listeners = []
+        self.addListener = self.listeners.append
+        self.removeListener = self.listeners.remove
+        self._message_count = 0
+        self._uid_index: list[int] = []
+        self._flags: dict[int, set[str]] = {}
+        self._cache = MessageCache(max_size=settings.imap_cache_size)
+        self._uid_index_built = False
+
+    def getFlags(self):
+        return [r"\Seen", r"\Answered", r"\Flagged", r"\Deleted", r"\Draft"]
+
+    def getUIDValidity(self):
+        return _UID_VALIDITY
+
+    def getMessageCount(self):
+        return self._message_count
+
+    def getRecentCount(self):
+        return 0
+
+    def getUnseenCount(self):
+        return 0
+
+    def isWriteable(self):
+        return 1
+
+    def destroy(self):
+        pass
+
+    def getHierarchicalDelimiter(self):
+        return "/"
+
+    @defer.inlineCallbacks
+    def requestStatus(self, names):
+        if not self._uid_index_built:
+            yield self._build_uid_index()
+        else:
+            count = yield self._refresh_count()
+            if count != self._message_count:
+                self._message_count = count
+                yield self._build_uid_index()
+
+        r = {}
+        if "MESSAGES" in names:
+            r["MESSAGES"] = self._message_count
+        if "RECENT" in names:
+            r["RECENT"] = self.getRecentCount()
+        if "UIDNEXT" in names:
+            r["UIDNEXT"] = self.getUIDNext()
+        if "UIDVALIDITY" in names:
+            r["UIDVALIDITY"] = self.getUIDValidity()
+        if "UNSEEN" in names:
+            r["UNSEEN"] = self.getUnseenCount()
+        return r
+
+    def _refresh_count(self) -> defer.Deferred:
+        return self._client.get_message_count(self.name)
+
+    @defer.inlineCallbacks
+    def _build_uid_index(self):
+        """Build UID index by fetching all message IDs from backend."""
+        count = yield self._client.get_message_count(self.name)
+        self._message_count = count
+        _logger.info("Building UID index for %s: count=%d", self.name, count)
+
+        if count == 0:
+            self._uid_index = []
+            self._uid_index_built = True
+            return
+
+        uid_set = set()
+        batch_size = 100
+        offset = 0
+
+        while offset < count:
+            limit = min(batch_size, count - offset)
+            results, _ = yield self._client.get_messages(
+                self.name, limit, offset
+            )
+            for item in results:
+                item_id = item.get("id")
+                if item_id is not None and item_id not in uid_set:
+                    uid_set.add(item_id)
+            _logger.info(
+                "UID index batch: offset=%d limit=%d got=%d total_uids=%d",
+                offset, limit, len(results), len(uid_set),
+            )
+            offset += limit
+
+        self._uid_index = sorted(uid_set)
+        self._uid_index_built = True
+        _logger.info(
+            "UID index built for %s: %d UIDs, range=%s..%s",
+            self.name, len(self._uid_index),
+            self._uid_index[0] if self._uid_index else "N/A",
+            self._uid_index[-1] if self._uid_index else "N/A",
+        )
+
+    def _seq_to_uid(self, seq: int) -> int | None:
+        """Convert 1-based sequence number to UID."""
+        if 1 <= seq <= len(self._uid_index):
+            return self._uid_index[seq - 1]
+        return None
+
+    def _uid_to_seq(self, uid: int) -> int | None:
+        """Convert UID to 1-based sequence number."""
+        idx = bisect.bisect_left(self._uid_index, uid)
+        if idx < len(self._uid_index) and self._uid_index[idx] == uid:
+            return idx + 1
+        return None
+
+    def _resolve_message_set(self, messages, uid: bool) -> list[int]:
+        """Resolve an IMAP MessageSet to a list of UIDs."""
+        result_uids = []
+        if not self._uid_index:
+            return result_uids
+
+        max_uid = self._uid_index[-1]
+        max_seq = len(self._uid_index)
+
+        _logger.info(
+            "Resolving message_set: uid=%s ranges=%s max_uid=%d max_seq=%d",
+            uid, list(messages.ranges), max_uid, max_seq,
+        )
+
+        for start, end in messages.ranges:
+            if uid:
+                actual_end = end if end is not None else max_uid
+                for u in self._uid_index:
+                    if start <= u <= actual_end:
+                        result_uids.append(u)
+            else:
+                actual_end = end if end is not None else max_seq
+                actual_start = max(start, 1)
+                actual_end = min(actual_end, max_seq)
+                for seq in range(actual_start, actual_end + 1):
+                    u = self._seq_to_uid(seq)
+                    if u is not None:
+                        result_uids.append(u)
+
+        return result_uids
+
+    @defer.inlineCallbacks
+    def _fetch_and_cache_messages(self, uids: list[int]):
+        """Fetch uncached messages from backend in batches."""
+        uncached = [u for u in uids if u not in self._cache]
+        if not uncached:
+            return
+
+        uncached_set = set(uncached)
+        id_to_data = {}
+        batch_size = 50
+        total = self._message_count
+
+        _logger.info(
+            "Fetching %d uncached messages (total=%d) for %s",
+            len(uncached), total, self.name,
+        )
+
+        if total == 0:
+            return
+
+        fetched_ids = set()
+        offset = 0
+
+        while offset < total and len(fetched_ids) < len(uncached):
+            limit = min(batch_size, total - offset)
+            results, _ = yield self._client.get_messages(
+                self.name, limit, offset
+            )
+            for item in results:
+                item_id = item.get("id")
+                if item_id in uncached_set and item_id not in fetched_ids:
+                    id_to_data[item_id] = item
+                    fetched_ids.add(item_id)
+
+            if len(fetched_ids) >= len(uncached):
+                break
+            offset += limit
+
+        _logger.info(
+            "Fetched %d/%d messages for %s",
+            len(id_to_data), len(uncached), self.name,
+        )
+
+        for uid_val in uncached:
+            if uid_val in id_to_data:
+                item = id_to_data[uid_val]
+                try:
+                    if self.name == "INBOX":
+                        raw = item.get("raw", "")
+                        email_model = parse_email(raw)
+                    elif self.name == "SENT":
+                        email_model, raw = generate_email_model(item)
+                    else:
+                        continue
+
+                    if uid_val not in self._flags:
+                        self._flags[uid_val] = {r"\Seen"}
+                    flags = self._flags[uid_val]
+                    msg = SimpleMessage(
+                        uid_val, email_model, flags=flags, raw=raw
+                    )
+                    self._cache.put(uid_val, msg)
+                except Exception as e:
+                    _logger.error(f"Failed to parse message uid={uid_val}: {e}")
+
+    @defer.inlineCallbacks
+    def fetch(self, messages, uid):
+        if not self._uid_index_built:
+            yield self._build_uid_index()
+        else:
+            count = yield self._refresh_count()
+            if count != self._message_count:
+                self._message_count = count
+                yield self._build_uid_index()
+
+        target_uids = self._resolve_message_set(messages, uid)
+        _logger.info(
+            "FETCH: uid=%s target_uids=%d message_set=%s",
+            uid, len(target_uids),
+            target_uids[:5] if len(target_uids) > 5 else target_uids,
+        )
+        if not target_uids:
+            return []
+
+        yield self._fetch_and_cache_messages(target_uids)
+
+        result = []
+        for u in target_uids:
+            cached = self._cache.get(u)
+            if cached is not None:
+                flags = self._flags.get(u, set())
+                cached._flags = flags
+                seq = self._uid_to_seq(u)
+                if seq is not None:
+                    result.append((seq, cached))
+
+        return result
+
+    def getUID(self, message):
+        return message
+
+    @defer.inlineCallbacks
+    def store(self, messages, flags, mode, uid):
+        if not self._uid_index_built:
+            yield self._build_uid_index()
+        if not self._uid_index:
+            return {}
+
+        target_uids = self._resolve_message_set(messages, uid)
+        result = {}
+
+        for u in target_uids:
+            current_flags = self._flags.get(u, set())
+
+            if mode == 1:    # +FLAGS
+                current_flags = current_flags | set(flags)
+            elif mode == -1:  # -FLAGS
+                current_flags = current_flags - set(flags)
+            elif mode == 0:   # FLAGS (replace)
+                current_flags = set(flags)
+
+            self._flags[u] = current_flags
+            seq = self._uid_to_seq(u)
+            if seq is not None:
+                result[seq] = current_flags
+
+        return result
+
+    @defer.inlineCallbacks
+    def search(self, query, uid):
+        if not self._uid_index_built:
+            yield self._build_uid_index()
+
+        results = []
+
+        for term in query:
+            if isinstance(term, str) and term.upper() == "ALL":
+                if uid:
+                    results = list(self._uid_index)
+                else:
+                    results = list(range(1, len(self._uid_index) + 1))
+                break
+
+        if not results:
+            if uid:
+                results = list(self._uid_index)
+            else:
+                results = list(range(1, len(self._uid_index) + 1))
+
+        return results
+
+    def getUIDNext(self):
+        if self._uid_index:
+            return self._uid_index[-1] + 1
+        return 1
+
+    def expunge(self):
+        return defer.succeed([])

smtp_proxy_server/imap_message.py

@@ -0,0 +1,71 @@
+from io import BytesIO
+
+from twisted.mail import imap4
+from zope.interface import implementer
+
+from models import EmailModel
+
+
+@implementer(imap4.IMessage, imap4.IMessageFile)
+class SimpleMessage:
+
+    def __init__(self, uid: int, email_model: EmailModel,
+                 flags: set[str] = None, raw: str = None):
+        self.uid = uid
+        self.email = email_model
+        self.subparts = self.email.subparts
+        self._flags = flags if flags is not None else set()
+        self._raw = raw
+
+    def getUID(self):
+        return self.uid
+
+    def getHeaders(self, negate, *names):
+        # Twisted passes header names as bytes (e.g. b"SUBJECT");
+        # normalize to lowercase str for comparison.
+        names_lower = set()
+        for n in names:
+            if isinstance(n, bytes):
+                names_lower.add(n.decode("ascii", errors="replace").lower())
+            else:
+                names_lower.add(n.lower())
+        if not names_lower:
+            return {k.lower(): v for k, v in self.email.headers.items()}
+        if negate:
+            return {
+                k.lower(): v
+                for k, v in self.email.headers.items()
+                if k.lower() not in names_lower
+            }
+        return {
+            k.lower(): v
+            for k, v in self.email.headers.items()
+            if k.lower() in names_lower
+        }
+
+    def isMultipart(self):
+        return len(self.subparts) > 0
+
+    def getSubPart(self, part):
+        return SimpleMessage(self.uid, self.subparts[part], flags=self._flags)
+
+    def getBodyFile(self):
+        return BytesIO(self.email.body.encode("utf-8"))
+
+    def getSize(self):
+        if self._raw is not None:
+            return len(self._raw.encode("utf-8"))
+        return self.email.size
+
+    def getFlags(self):
+        return list(self._flags)
+
+    def getInternalDate(self):
+        return self.email.headers.get("Date", "Mon, 1 Jan 1900 00:00:00 +0000")
+
+    # IMessageFile
+    def open(self):
+        """Return complete raw MIME message for BODY[] requests."""
+        if self._raw is not None:
+            return BytesIO(self._raw.encode("utf-8"))
+        return BytesIO(self.email.body.encode("utf-8"))

smtp_proxy_server/imap_server.py

@@ -1,292 +1,128 @@
 import json
 import logging
-import httpx
 
-from io import BytesIO
+import httpx
 from twisted.mail import imap4
 from zope.interface import implementer
 from twisted.cred.portal import Portal, IRealm
-from twisted.internet import protocol, reactor, defer
+from twisted.internet import protocol, reactor, defer, ssl, threads
+from twisted.cred import error as cred_error
 from twisted.cred.checkers import ICredentialsChecker, IUsernamePassword
 
 from config import settings
-from parse_email import generate_email_model, parse_email
-from models import EmailModel
+from imap_http_client import BackendClient
+from imap_mailbox import SimpleMailbox
 
 _logger = logging.getLogger(__name__)
-_logger.setLevel(logging.INFO)
-
-
-@implementer(imap4.IMessage)
-class SimpleMessage:
-
-    def __init__(self, uid=None, email_model: EmailModel = None):
-        self.uid = uid
-        self.email = email_model
-        self.subparts = self.email.subparts
-
-    def getUID(self):
-        return self.uid
-
-    def getHeaders(self, negate, *names):
-        self.got_headers = negate, names
-        return {
-            k.lower(): v
-            for k, v in self.email.headers.items()
-        }
-
-    def isMultipart(self):
-        return len(self.subparts) > 0
-
-    def getSubPart(self, part):
-        self.got_subpart = part
-        return SimpleMessage(email_model=self.subparts[part])
-
-    def getBodyFile(self):
-        return BytesIO(self.email.body.encode("utf-8"))
-
-    def getSize(self):
-        return self.email.size
-
-    def getFlags(self):
-        return ["\\Seen"]
-
-    def getInternalDate(self):
-        return self.email.headers.get("Date", "Mon, 1 Jan 1900 00:00:00 +0000")
-
-
-@implementer(imap4.IMailboxInfo, imap4.IMailbox)
-class SimpleMailbox:
-
-    def __init__(self, name, password):
-        self.name = name
-        self.password = password
-        self.listeners = []
-        self.addListener = self.listeners.append
-        self.removeListener = self.listeners.remove
-        self.message_count = 0
-        self._update_message_count()
-
-    def _update_message_count(self):
-        """主动获取邮件总数"""
-        try:
-            if self.name == "INBOX":
-                endpoint = "/api/mails"
-            elif self.name == "SENT":
-                endpoint = "/api/sendbox"
-            else:
-                return
-
-            res = httpx.get(
-                f"{settings.proxy_url}{endpoint}?limit=1&offset=0",
-                headers={
-                    "Authorization": f"Bearer {self.password}",
-                    "x-custom-auth": f"{settings.basic_password}",
-                    "Content-Type": "application/json"
-                }
-            )
-            if res.status_code == 200:
-                self.message_count = res.json()["count"]
-                # _logger.info(f"Updated {self.name} message count: {self.message_count}")
-        except Exception as e:
-            _logger.error(f"Failed to update message count for {self.name}: {e}")
-
-    def getFlags(self):
-        return ["\\Seen"]
-
-    def getUIDValidity(self):
-        return 0
-
-    def getMessageCount(self):
-        # 每次请求时更新邮件总数
-        self._update_message_count()
-        return self.message_count
-
-    def getRecentCount(self):
-        return 0
-
-    def getUnseenCount(self):
-        return 0
-
-    def isWriteable(self):
-        return 0
-
-    def destroy(self):
-        pass
-
-    def getHierarchicalDelimiter(self):
-        return "/"
-
-    def requestStatus(self, names):
-        # 在状态请求时也更新邮件总数
-        self._update_message_count()
-        r = {}
-        if "MESSAGES" in names:
-            r["MESSAGES"] = self.getMessageCount()
-        if "RECENT" in names:
-            r["RECENT"] = self.getRecentCount()
-        if "UIDNEXT" in names:
-            r["UIDNEXT"] = self.getMessageCount() + 1
-        if "UIDVALIDITY" in names:
-            r["UIDVALIDITY"] = self.getUIDValidity()
-        if "UNSEEN" in names:
-            r["UNSEEN"] = self.getUnseenCount()
-        return defer.succeed(r)
-
-    def fetch(self, messages, uid):
-        """边查边返回邮件"""
-        result = []
-        for range_item in messages.ranges:
-            start, end = range_item
-            _logger.info(f"Fetching messages: {self.name}, range: {start}-{end}")
-
-            for email_data in self.fetchGenerator(start, end):
-                result.append(email_data)
-
-        # 返回列表而不是生成器，以支持 IMAP SEARCH 等需要索引访问的操作
-        return result
-
-    def fetchGenerator(self, start, end):
-        """通用的邮件获取生成器，边查边返回"""
-        start = max(start, 1)
-
-        # 根据邮箱类型确定API端点
-        if self.name == "INBOX":
-            endpoint = "/api/mails"
-        elif self.name == "SENT":
-            endpoint = "/api/sendbox"
-        else:
-            return
-
-        # 首先获取服务端邮件总数
-        count_res = httpx.get(
-            f"{settings.proxy_url}{endpoint}?limit=1&offset=0",
-            headers={
-                "Authorization": f"Bearer {self.password}",
-                "x-custom-auth": f"{settings.basic_password}",
-                "Content-Type": "application/json"
-            }
-        )
-        if count_res.status_code != 200:
-            _logger.error(
-                f"Failed to get {self.name} email count: "
-                f"code=[{count_res.status_code}] text=[{count_res.text}]"
-            )
-            return
-
-        total_count = count_res.json()["count"]
-        self.message_count = total_count
-
-        if total_count == 0 or start > total_count:
-            return
-
-        # 分批处理，每次获取一小批就立即返回
-        batch_size = 20
-        current_start = start
-        current_end = min(end or total_count, total_count)
-
-        while current_start <= current_end:
-            batch_end = min(current_start + batch_size - 1, current_end)
-
-            # 计算这一批的参数
-            limit = batch_end - current_start + 1
-            server_offset = total_count - batch_end
-            server_offset = max(0, server_offset)
-
-            _logger.info(
-                f"Fetching batch: start={current_start}, end={batch_end}, "
-                f"total_count={total_count}, limit={limit}, "
-                f"server_offset={server_offset}"
-            )
-
-            res = httpx.get(
-                f"{settings.proxy_url}{endpoint}?limit={limit}&offset={server_offset}",
-                headers={
-                    "Authorization": f"Bearer {self.password}",
-                    "x-custom-auth": f"{settings.basic_password}",
-                    "Content-Type": "application/json"
-                }
-            )
-            if res.status_code != 200:
-                _logger.error(
-                    f"Failed to fetch {self.name} emails: "
-                    f"code=[{res.status_code}] text=[{res.text}]"
-                )
-                break
-
-            emails = res.json()["results"]
-            for i, item in enumerate(reversed(emails)):
-                uid = total_count - server_offset - len(emails) + i + 1
-                if current_start <= uid <= batch_end:
-                    if self.name == "INBOX":
-                        email_model = parse_email(item["raw"])
-                    elif self.name == "SENT":
-                        email_model = generate_email_model(item)
-
-                    # 立即返回这封邮件
-                    yield (uid, SimpleMessage(uid, email_model))
-
-            current_start = batch_end + 1
-
-    def getUID(self, message):
-        return message.uid
-
-    def store(self, messages, flags, mode, uid):
-        # IMailboxIMAP.store
-        raise NotImplementedError
-
-
-class Account(imap4.MemoryAccount):
-
-    def __init__(self, user, password):
-        self.password = password
-        super().__init__(user)
-
-    def isSubscribed(self, name):
-        return name.upper() in ["INBOX", "SENT"]
-
-    def _emptyMailbox(self, name, id):
-        _logger.info(f"New mailbox: {name}, {id}")
-        if name == "INBOX":
-            return SimpleMailbox(name, self.password)
-        if name == "SENT":
-            return SimpleMailbox(name, self.password)
-        raise imap4.NoSuchMailbox(name.encode("utf-8"))
-
-    def select(self, name, rw=1):
-        return imap4.MemoryAccount.select(self, name)
+_logger.setLevel(logging.DEBUG)
+logging.basicConfig(level=logging.DEBUG)
 
 
 class SimpleIMAPServer(imap4.IMAP4Server):
-    def __init__(self, factory):
-        imap4.IMAP4Server.__init__(self)
-        self.factory = factory
+    def __init__(self, context_factory=None):
+        chal = {
+            b"LOGIN": imap4.LOGINCredentials,
+            b"PLAIN": imap4.PLAINCredentials,
+        }
+        imap4.IMAP4Server.__init__(
+            self, chal=chal, contextFactory=context_factory
+        )
 
     def lineReceived(self, line):
-        # _logger.info(f"Received: {line}")
-        super().lineReceived(line)
+        _logger.debug("C: %s", line)
+        return imap4.IMAP4Server.lineReceived(self, line)
 
     def sendLine(self, line):
-        # _logger.info(f"Sent: {line}")
-        super().sendLine(line)
+        _logger.debug("S: %s", line)
+        return imap4.IMAP4Server.sendLine(self, line)
+
+    def connectionMade(self):
+        """Wrap transport to log raw data sent to client."""
+        imap4.IMAP4Server.connectionMade(self)
+        real_write_seq = self.transport.writeSequence
+        def logging_write_seq(data):
+            joined = b''.join(data)
+            for line in joined.split(b'\r\n'):
+                if line:
+                    _logger.debug("S-RAW: %s", line[:300])
+            return real_write_seq(data)
+        self.transport.writeSequence = logging_write_seq
+
+    def _cbSelectWork(self, mbox, cmdName, tag):
+        """Override to add UIDNEXT in SELECT response (RFC 3501)."""
+        if mbox is None:
+            self.sendNegativeResponse(tag, b"No such mailbox")
+            return
+        if "\\noselect" in [s.lower() for s in mbox.getFlags()]:
+            self.sendNegativeResponse(tag, "Mailbox cannot be selected")
+            return
+
+        flags = [imap4.networkString(flag) for flag in mbox.getFlags()]
+        self.sendUntaggedResponse(b"%d EXISTS" % (mbox.getMessageCount(),))
+        self.sendUntaggedResponse(b"%d RECENT" % (mbox.getRecentCount(),))
+        self.sendUntaggedResponse(b"FLAGS (" + b" ".join(flags) + b")")
+        self.sendPositiveResponse(
+            None, b"[UIDVALIDITY %d]" % (mbox.getUIDValidity(),)
+        )
+        self.sendPositiveResponse(
+            None, b"[UIDNEXT %d]" % (mbox.getUIDNext(),)
+        )
+
+        s = mbox.isWriteable() and b"READ-WRITE" or b"READ-ONLY"
+        mbox.addListener(self)
+        self.sendPositiveResponse(
+            tag, b"[" + s + b"] " + cmdName + b" successful"
+        )
+        self.state = "select"
+        self.mbox = mbox
+
+
+class Account(imap4.MemoryAccount):
+    """Custom account that initializes mailbox UID index on select."""
+
+    def _emptyMailbox(self, name, id):
+        """Ignore CREATE for unknown mailboxes instead of crashing."""
+        return None
+
+    def create(self, pathspec):
+        """Silently ignore mailbox creation requests from clients."""
+        _logger.debug("Ignoring CREATE request for %s", pathspec)
+        return False
+
+    @defer.inlineCallbacks
+    def select(self, name, readwrite=1):
+        mbox = self.mailboxes.get(imap4._parseMbox(name.upper()))
+        if mbox is not None:
+            yield mbox._build_uid_index()
+        return mbox
 
 
 @implementer(IRealm)
 class SimpleRealm:
     def requestAvatar(self, avatarId, mind, *interfaces):
         res = json.loads(avatarId)
-        account = Account(res["username"], res["password"])
-        account.addMailbox("INBOX")
-        account.addMailbox("SENT")
-        return imap4.IAccount, account, lambda: None
+        username = res["username"]
+        password = res["password"]
+
+        client = BackendClient(password)
+
+        inbox = SimpleMailbox("INBOX", client)
+        sent = SimpleMailbox("SENT", client)
+
+        account = Account(username)
+        account.mailboxes = {"INBOX": inbox, "SENT": sent}
+        account.subscriptions = ["INBOX", "SENT"]
+
+        return imap4.IAccount, account, lambda: client.close()
 
 
 class IMAPFactory(protocol.Factory):
-    def __init__(self, portal):
+    def __init__(self, portal, context_factory=None):
         self.portal = portal
+        self._context_factory = context_factory
 
     def buildProtocol(self, addr):
-        p = SimpleIMAPServer(self)
+        p = SimpleIMAPServer(context_factory=self._context_factory)
         p.portal = self.portal
         return p
 
@@ -295,20 +131,77 @@ class IMAPFactory(protocol.Factory):
 class CustomChecker:
     credentialInterfaces = (IUsernamePassword,)
 
+    @staticmethod
+    def _is_jwt(token: str) -> bool:
+        """Check if token looks like a JWT (eyJ... with 3 dot-separated parts)."""
+        parts = token.split(".")
+        return len(parts) == 3 and parts[0].startswith("eyJ")
+
     def requestAvatarId(self, credentials):
-        return defer.succeed(json.dumps({
-            "username": credentials.username.decode(),
-            "password": credentials.password.decode(),
-        }))
+        username = credentials.username.decode()
+        password = credentials.password.decode()
+
+        if self._is_jwt(password):
+            _logger.info("Login via JWT token")
+            return defer.succeed(json.dumps({
+                "username": username,
+                "password": password,
+            }))
+
+        # Not a JWT — try address+password login via backend
+        _logger.info("Login via address+password")
+        d = threads.deferToThread(self._login_with_password, username, password)
+        return d
+
+    @staticmethod
+    def _login_with_password(username: str, password: str) -> str:
+        """Exchange address+password for a JWT via backend."""
+        res = httpx.post(
+            f"{settings.proxy_url}/api/address_login",
+            json={"email": username, "password": password},
+            headers={
+                "x-custom-auth": settings.basic_password,
+                "Content-Type": "application/json",
+            },
+            timeout=settings.imap_http_timeout,
+        )
+        if res.status_code == 200:
+            jwt = res.json().get("jwt")
+            if jwt:
+                return json.dumps({
+                    "username": username,
+                    "password": jwt,
+                })
+        raise cred_error.UnauthorizedLogin(f"address_login failed: {res.status_code}")
 
 
 def start_imap_server():
-    _logger.info(f"Starting IMAP server on port {settings.imap_port}")
+    _logger.info("Starting IMAP server on port %s", settings.imap_port)
+
+    context_factory = None
+    has_cert = bool(settings.imap_tls_cert)
+    has_key = bool(settings.imap_tls_key)
+    if has_cert != has_key:
+        raise ValueError(
+            "Both imap_tls_cert and imap_tls_key must be set together"
+        )
+    if has_cert and has_key:
+        _logger.info("TLS enabled for IMAP (STARTTLS)")
+        context_factory = ssl.DefaultOpenSSLContextFactory(
+            settings.imap_tls_key,
+            settings.imap_tls_cert,
+        )
+
     portal = Portal(SimpleRealm(), [CustomChecker()])
-    reactor.listenTCP(settings.imap_port, IMAPFactory(portal))
+    factory = IMAPFactory(portal, context_factory=context_factory)
+    reactor.listenTCP(settings.imap_port, factory)
     reactor.run()
 
 
 if __name__ == "__main__":
-    _logger.info(f"Starting server settings[{settings}]")
+    _logger.info(
+        "Starting IMAP server proxy_url=%s port=%s tls=%s",
+        settings.proxy_url, settings.imap_port,
+        bool(settings.imap_tls_cert and settings.imap_tls_key),
+    )
     start_imap_server()

smtp_proxy_server/main.py

@@ -9,7 +9,10 @@ _logger = logging.getLogger(__name__)
 _logger.setLevel(logging.INFO)
 
 if __name__ == '__main__':
-    _logger.info(f"Starting server settings[{settings}]")
+    _logger.info(
+        "Starting server proxy_url=%s smtp_port=%s imap_port=%s",
+        settings.proxy_url, settings.port, settings.imap_port,
+    )
     process_list = [
         multiprocessing.Process(target=start_smtp_server, args=()),
         multiprocessing.Process(target=start_imap_server, args=()),

smtp_proxy_server/parse_email.py

@@ -19,7 +19,15 @@ def get_email_model(msg: Message):
         get_email_model(subpart)
         for subpart in msg.get_payload()
     ] if msg.is_multipart() else []
-    body = "" if msg.is_multipart() else msg._payload
+    if msg.is_multipart():
+        body = ""
+    else:
+        raw_body = msg.get_payload(decode=True) or b""
+        charset = msg.get_content_charset() or "utf-8"
+        try:
+            body = raw_body.decode(charset, errors="replace")
+        except LookupError:
+            body = raw_body.decode("utf-8", errors="replace")
     return EmailModel(
         headers={k: v for k, v in msg.items()},
         body=body,
@@ -44,7 +52,12 @@ def parse_email(raw: str) -> EmailModel:
         )
 
 
-def generate_email_model(item: dict) -> EmailModel:
+def generate_email_model(item: dict) -> tuple[EmailModel, str]:
+    """Build an EmailModel from a sendbox item.
+
+    Returns (EmailModel, raw_mime_string) so callers can pass the
+    synthesised MIME to SimpleMessage for correct BODY[] responses.
+    """
     email_json = json.loads(item["raw"])
     message = MIMEMultipart()
     if email_json.get("version") == "v2":
@@ -66,4 +79,5 @@ def generate_email_model(item: dict) -> EmailModel:
     message["Date"] = datetime.datetime.strptime(
         item["created_at"], "%Y-%m-%d %H:%M:%S"
     ).strftime("%a, %d %b %Y %H:%M:%S +0000")
-    return parse_email(message.as_string())
+    raw_mime = message.as_string()
+    return parse_email(raw_mime), raw_mime

smtp_proxy_server/requirements.txt

@@ -1,5 +1,6 @@
 aiosmtpd==1.4.6
-pydantic-settings==2.9.1
-requests==2.32.4
+pydantic-settings==2.13.1
 Twisted==25.5.0
 httpx==0.28.1
+pyOpenSSL==25.3.0
+service-identity==24.2.0

smtp_proxy_server/smtp_server.py

@@ -12,6 +12,15 @@ _logger = logging.getLogger(__name__)
 _logger.setLevel(logging.INFO)
 
 
+def _safe_decode_payload(payload, charset):
+    if payload is None:
+        return ""
+    try:
+        return payload.decode(charset or "utf-8", errors="replace")
+    except LookupError:
+        return payload.decode("utf-8", errors="replace")
+
+
 class CustomSMTPHandler:
 
     def authenticator(self, server, session, envelope, mechanism, auth_data):
@@ -49,7 +58,7 @@ class CustomSMTPHandler:
                     value = part.get_payload(decode=False)
                 else:
                     payload = part.get_payload(decode=True)
-                    value = payload.decode(charset) if charset else payload
+                    value = _safe_decode_payload(payload, charset)
                 if not value:
                     continue
                 content_list.append({
@@ -63,8 +72,8 @@ class CustomSMTPHandler:
                 value = msg.get_payload(decode=False)
             else:
                 payload = msg.get_payload(decode=True)
-                value = payload.decode(charset) if charset else payload
-            _logger.info(f"Payload {msg._payload} charset {charset}")
+                value = _safe_decode_payload(payload, charset)
+            _logger.debug("Parsed content charset=%s", charset)
             content_list.append({
                 "type": msg.get_content_type(),
                 "value": value
@@ -121,27 +130,23 @@ class CustomSMTPHandler:
         return '250 OK'
 
 
-handler = CustomSMTPHandler()
-server = Controller(
-    handler,
-    hostname="",
-    port=settings.port,
-    auth_require_tls=False,
-    decode_data=True,
-    authenticator=handler.authenticator,
-    auth_exclude_mechanism=["DONT"]
-)
+def start_smtp_server():
+    handler = CustomSMTPHandler()
+    server = Controller(
+        handler,
+        hostname="",
+        port=settings.port,
+        auth_require_tls=False,
+        decode_data=True,
+        authenticator=handler.authenticator,
+        auth_exclude_mechanism=["DONT"]
+    )
 
-
-async def start():
-    _logger.info(f"Starting server on port {settings.port}")
+    _logger.info("Starting SMTP server on port %s", settings.port)
     server.start()
 
-
-def start_smtp_server():
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
-    task = loop.create_task(start())
     try:
         loop.run_forever()
     except KeyboardInterrupt:
@@ -150,5 +155,8 @@ def start_smtp_server():
 
 
 if __name__ == "__main__":
-    _logger.info(f"Starting server settings[{settings}]")
+    _logger.info(
+        "Starting SMTP server proxy_url=%s port=%s",
+        settings.proxy_url, settings.port,
+    )
     start_smtp_server()