写了个邮件代理：在任意的邮件客户端中，使用来自 Resend 的域名邮箱

最近把 Codium 配置好了，接下来肯定得写点小项目来找下手感。Resend 不支持 IMAP 服务，QQ 邮箱需要 VIP 才能使用自定义域名。我目前能够使用 Cloudflare Routing 收邮件，也可以使用 Resend 发邮件，但还需要自己拼起来。因此我就有了一个十分明确的需求：我要编写一个支持 IMAP 和 SMTP 的邮件代理，用苹果的邮箱登录它。IMAP 不需要真的有收件箱的功能，只需要有发件箱证明自己发送过即可，收件箱可以来自 Cloudflare Routing。按照这个思路，我很快就想到了代理的实现原理：在本机拉起两个端口，分别代表 SMTP 和 IMAP，从 SMTP 发的邮件统一把请求转发到 Resend 服务，本地 ~/.sent 留有一份发件箱邮件，让苹果的客户端能够联系上下文：

import asyncio
import os
import smtplib
import time
import uuid
from email.message import EmailMessage
from email.parser import BytesParser
from email.policy import default as defaultPolicy
from email.utils import formataddr, parseaddr
from pathlib import Path

from aiosmtpd.controller import Controller
from aiosmtpd.smtp import SMTP, AuthResult, LoginPassword

SMTP_LISTEN_HOST = "0.0.0.0"
SMTP_LISTEN_PORT = 2525
IMAP_LISTEN_HOST = "0.0.0.0"
IMAP_LISTEN_PORT = 2143

RESEND_SMTP_HOST = "smtp.resend.com"
RESEND_SMTP_PORT = 465
RESEND_SMTP_USER = "resend"
RESEND_SMTP_PASSWORD = "" # 写你自己的 Resend API Key
RESEND_FROM_ADDRESS = "" # 写你自己的域名收件人地址

SENT_CACHE_DIR = Path.home() / ".sent"
SENT_MAILBOX_NAME = "Sent"
SENT_MAILBOX_ALIASES = ["Sent", "Sent Messages", "Sent Items"]
INBOX_MAILBOX_NAME = "INBOX"


def cacheSentMessage(rawMessage):
  SENT_CACHE_DIR.mkdir(parents=True, exist_ok=True)
  fileName = f"{int(time.time() * 1000)}-{uuid.uuid4().hex}.eml"
  filePath = SENT_CACHE_DIR / fileName
  filePath.write_bytes(rawMessage)
  return filePath


def relayToResend(rawMessage):
  message = BytesParser(policy=defaultPolicy).parsebytes(rawMessage)
  originalName, _ = parseaddr(message.get("From", ""))
  del message["From"]
  message["From"] = formataddr((originalName, RESEND_FROM_ADDRESS)) if originalName else RESEND_FROM_ADDRESS
  with smtplib.SMTP_SSL(RESEND_SMTP_HOST, RESEND_SMTP_PORT) as client:
    client.login(RESEND_SMTP_USER, RESEND_SMTP_PASSWORD)
    client.send_message(message)
  cacheSentMessage(rawMessage)
  return message


class SmtpHandler:
  async def handle_DATA(self, server, session, envelope):
    try:
      message = await asyncio.to_thread(relayToResend, envelope.original_content)
      print(f"Relayed message to Resend: {message.get('Subject', '(no subject)')}")
      return "250 Message accepted for delivery"
    except Exception as error:
      print(f"Failed to relay message to Resend: {error}")
      return "451 Relay failed, try again later"


class ProxySmtp(SMTP):
  async def auth_PLAIN(self, server, args):
    return AuthResult(success=True)

  async def auth_LOGIN(self, server, args):
    return AuthResult(success=True)

  async def auth_CRAM_MD5(self, server, args):
    return AuthResult(success=True)


class ProxyController(Controller):
  def factory(self):
    return ProxySmtp(self.handler, hostname=self.hostname, enable_SMTPUTF8=True)


def sendImapResponse(writer, line):
  writer.write((line + "\r\n").encode("utf-8"))


def isSentMailbox(mailboxName):
  return mailboxName.strip().strip('"').upper() in [name.upper() for name in SENT_MAILBOX_ALIASES]


def listCachedMessages():
  if not SENT_CACHE_DIR.exists():
    return []
  return sorted(SENT_CACHE_DIR.glob("*.eml"))


def readCachedMessage(filePath):
  try:
    return filePath.read_bytes()
  except OSError:
    return b""


def buildFetchResponse(sequenceNumber, filePath, items):
  rawMessage = readCachedMessage(filePath)
  message = BytesParser(policy=defaultPolicy).parsebytes(rawMessage)
  upperItems = items.upper()
  parts = []
  if "UID" in upperItems:
    parts.append(f"UID {sequenceNumber}")
  if "FLAGS" in upperItems:
    parts.append("FLAGS (\\Seen)")
  if "RFC822.SIZE" in upperItems:
    parts.append(f"RFC822.SIZE {len(rawMessage)}")
  if "BODY.PEEK[]" in upperItems or "BODY[]" in upperItems or "RFC822" in upperItems:
    parts.append(f"BODY[] {{{len(rawMessage)}}}")
    header = f"* {sequenceNumber} FETCH (" + " ".join(parts) + ")\r\n"
    return header.encode("utf-8") + rawMessage + b")\r\n"
  if "BODY.PEEK[HEADER]" in upperItems or "BODY[HEADER]" in upperItems or "RFC822.HEADER" in upperItems:
    headerBytes = message.as_bytes()
    headerEnd = headerBytes.find(b"\r\n\r\n")
    headerOnly = headerBytes[:headerEnd + 4] if headerEnd >= 0 else headerBytes
    parts.append(f"BODY[HEADER] {{{len(headerOnly)}}}")
    header = f"* {sequenceNumber} FETCH (" + " ".join(parts) + ")\r\n"
    return header.encode("utf-8") + headerOnly + b")\r\n"
  if "BODY.PEEK[TEXT]" in upperItems or "BODY[TEXT]" in upperItems or "RFC822.TEXT" in upperItems:
    bodyBytes = message.get_payload(decode=True) or b""
    parts.append(f"BODY[TEXT] {{{len(bodyBytes)}}}")
    header = f"* {sequenceNumber} FETCH (" + " ".join(parts) + ")\r\n"
    return header.encode("utf-8") + bodyBytes + b")\r\n"
  if not parts:
    parts.append("FLAGS (\\Seen)")
  return (f"* {sequenceNumber} FETCH (" + " ".join(parts) + ")\r\n").encode("utf-8")


def parseFetchItems(text):
  start = text.find("(")
  end = text.rfind(")")
  if start >= 0 and end > start:
    return text[start + 1:end]
  return text


async def handleImapConnection(reader, writer):
  sendImapResponse(writer, "* OK IMAP4rev1 mail-proxy ready")
  await writer.drain()
  while True:
    line = await reader.readline()
    if not line:
      break
    text = line.decode("utf-8", errors="replace").rstrip("\r\n")
    if not text:
      continue
    parts = text.split(" ", 2)
    if len(parts) < 2:
      sendImapResponse(writer, "* BAD malformed command")
      await writer.drain()
      continue
    tag = parts[0]
    command = parts[1].upper()
    shouldClose = False
    if command == "CAPABILITY":
      sendImapResponse(writer, "* CAPABILITY IMAP4rev1 AUTH=PLAIN SPECIAL-USE")
      sendImapResponse(writer, f"{tag} OK CAPABILITY completed")
    elif command == "LOGIN":
      sendImapResponse(writer, f"{tag} OK LOGIN completed")
    elif command == "AUTHENTICATE":
      sendImapResponse(writer, f"{tag} OK AUTHENTICATE completed")
    elif command in ("SELECT", "EXAMINE"):
      mailboxName = parts[2].strip().strip('"') if len(parts) > 2 else INBOX_MAILBOX_NAME
      if isSentMailbox(mailboxName):
        cachedMessages = listCachedMessages()
        sendImapResponse(writer, f"* {len(cachedMessages)} EXISTS")
        sendImapResponse(writer, "* 0 RECENT")
        sendImapResponse(writer, "* OK [UIDVALIDITY 1] UIDs valid")
        sendImapResponse(writer, f"* OK [UIDNEXT {len(cachedMessages) + 1}] Predicted next UID")
        sendImapResponse(writer, "* FLAGS (\\Seen \\Answered \\Flagged \\Deleted \\Draft)")
        sendImapResponse(writer, f"{tag} OK [READ-WRITE] SELECT completed")
      else:
        sendImapResponse(writer, "* 0 EXISTS")
        sendImapResponse(writer, "* 0 RECENT")
        sendImapResponse(writer, "* OK [UIDVALIDITY 1] UIDs valid")
        sendImapResponse(writer, "* OK [UIDNEXT 1] Predicted next UID")
        sendImapResponse(writer, "* FLAGS (\\Seen \\Answered \\Flagged \\Deleted \\Draft)")
        sendImapResponse(writer, f"{tag} OK [READ-WRITE] SELECT completed")
    elif command in ("LIST", "LSUB"):
      sendImapResponse(writer, '* LIST (\\HasNoChildren) "/" "INBOX"')
      sendImapResponse(writer, f'* LIST (\\HasNoChildren \\Sent) "/" "{SENT_MAILBOX_NAME}"')
      sendImapResponse(writer, f"{tag} OK LIST completed")
    elif command == "STATUS":
      mailboxName = parts[2].strip().strip('"') if len(parts) > 2 else INBOX_MAILBOX_NAME
      if isSentMailbox(mailboxName):
        cachedMessages = listCachedMessages()
        sendImapResponse(writer, f"* STATUS {mailboxName} (MESSAGES {len(cachedMessages)} RECENT 0 UIDNEXT {len(cachedMessages) + 1} UIDVALIDITY 1 UNSEEN 0)")
      else:
        sendImapResponse(writer, "* STATUS INBOX (MESSAGES 0 RECENT 0 UIDNEXT 1 UIDVALIDITY 1 UNSEEN 0)")
      sendImapResponse(writer, f"{tag} OK STATUS completed")
    elif command == "FETCH":
      cachedMessages = listCachedMessages()
      sequenceSpec = parts[2].split(" ", 1)[0] if len(parts) > 2 else "1:*"
      items = parseFetchItems(parts[2]) if len(parts) > 2 else "FLAGS"
      if sequenceSpec in ("1:*", "*"):
        targetIndexes = list(range(len(cachedMessages)))
      else:
        targetIndexes = []
        for token in sequenceSpec.split(","):
          if ":" in token:
            startText, endText = token.split(":", 1)
            startIndex = int(startText) - 1 if startText != "*" else len(cachedMessages) - 1
            endIndex = int(endText) - 1 if endText != "*" else len(cachedMessages) - 1
            targetIndexes.extend(range(startIndex, endIndex + 1))
          else:
            targetIndexes.append(int(token) - 1)
      for index in targetIndexes:
        if 0 <= index < len(cachedMessages):
          writer.write(buildFetchResponse(index + 1, cachedMessages[index], items))
      sendImapResponse(writer, f"{tag} OK FETCH completed")
    elif command == "UID":
      subCommand = parts[2].split(" ", 1)[0].upper() if len(parts) > 2 else ""
      if subCommand == "FETCH":
        cachedMessages = listCachedMessages()
        rest = parts[2].split(" ", 2)
        sequenceSpec = rest[1] if len(rest) > 1 else "1:*"
        items = parseFetchItems(rest[2]) if len(rest) > 2 else "FLAGS"
        if sequenceSpec in ("1:*", "*"):
          targetIndexes = list(range(len(cachedMessages)))
        else:
          targetIndexes = []
          for token in sequenceSpec.split(","):
            if ":" in token:
              startText, endText = token.split(":", 1)
              startIndex = int(startText) - 1 if startText != "*" else len(cachedMessages) - 1
              endIndex = int(endText) - 1 if endText != "*" else len(cachedMessages) - 1
              targetIndexes.extend(range(startIndex, endIndex + 1))
            else:
              targetIndexes.append(int(token) - 1)
        for index in targetIndexes:
          if 0 <= index < len(cachedMessages):
            writer.write(buildFetchResponse(index + 1, cachedMessages[index], items))
        sendImapResponse(writer, f"{tag} OK UID FETCH completed")
      else:
        sendImapResponse(writer, "* SEARCH")
        sendImapResponse(writer, f"{tag} OK UID completed")
    elif command == "SEARCH":
      cachedMessages = listCachedMessages()
      sequenceNumbers = " ".join(str(index + 1) for index in range(len(cachedMessages)))
      sendImapResponse(writer, f"* SEARCH {sequenceNumbers}".rstrip())
      sendImapResponse(writer, f"{tag} OK SEARCH completed")
    elif command == "NOOP":
      sendImapResponse(writer, f"{tag} OK NOOP completed")
    elif command == "LOGOUT":
      sendImapResponse(writer, "* BYE logging out")
      sendImapResponse(writer, f"{tag} OK LOGOUT completed")
      shouldClose = True
    elif command == "CLOSE":
      sendImapResponse(writer, f"{tag} OK CLOSE completed")
    elif command == "EXPUNGE":
      sendImapResponse(writer, f"{tag} OK EXPUNGE completed")
    elif command in ("STORE", "COPY", "MOVE"):
      sendImapResponse(writer, f"{tag} OK {command} completed")
    elif command == "APPEND":
      sendImapResponse(writer, f"{tag} OK APPEND completed")
    else:
      sendImapResponse(writer, f"{tag} BAD unknown command")
    await writer.drain()
    if shouldClose:
      break
  writer.close()
  await writer.wait_closed()


async def main():
  smtpController = ProxyController(SmtpHandler(), hostname=SMTP_LISTEN_HOST, port=SMTP_LISTEN_PORT)
  smtpController.start()
  imapServer = await asyncio.start_server(handleImapConnection, IMAP_LISTEN_HOST, IMAP_LISTEN_PORT)
  print(f"SMTP proxy listening on {SMTP_LISTEN_HOST}:{SMTP_LISTEN_PORT}")
  print(f"IMAP proxy listening on {IMAP_LISTEN_HOST}:{IMAP_LISTEN_PORT}")
  try:
    async with imapServer:
      await imapServer.serve_forever()
  except (KeyboardInterrupt, asyncio.CancelledError):
    print("Shutting down mail-proxy")
  finally:
    smtpController.stop()
    print("mail-proxy stopped")


if __name__ == "__main__":
  try:
    asyncio.run(main())
  except KeyboardInterrupt:
    print("Interrupted, exiting")

整个脚本使用 Codium 和 Aider 进行设计，我后续还用 Nuitka 对代码进行了编译，放在了自己家目录的 ~/.local/bin/ 下面，方便自己平时使用不需要进入项目文件夹。整套代码的分发同自己的文章一样，都是使用木兰协议分发代码。整体采用小驼峰命名规范，变量使用全大写加下划线。除了开头定义变量那几个地方外，其它的地方都没有注释，所有输出都使用了纯英文，考虑到 Ctrl 加 C 会导致程序报错所以关闭了报错问题，适合想研究代码的强迫症读者。目前我把这个程序运行在自己的 macOS 终端里，隐藏窗口后一直挂在后台也啥影响。Nuitka 编译过一遍的程序基本等同 C 写的，因此我不需要担心啥流畅度问题。顺便分享下代码依赖环境，方便各位读者部署使用：

aiosmtpd==1.4.6
atpublic==9.0.0
attrs==26.1.0
Nuitka==4.3.1