邮件代理第二版：优化了邮箱发件人逻辑，现在能自定义地址前缀了

之前不是给大伙分享了一个邮件代理项目吗？我一开始光想着实现一个邮件代理，后来发现初版的错误有点多了。初版只能使用一个地址发邮件，但问题是我都注册域名了，用的都是这个域名发邮件了，为什么要删掉发信人？最新版修复了这个基础问题。取而代之的是，现在可以在客户端自定义地址，你只需要确保这地址能用就行了。反正你只需要知道，如果你的域名没法用，那么邮件代理的日志就会给报错。我不想解释太多，之前博客写过了。这次的依赖列表依旧是之前那些，想要找那篇文章可以看看本文末尾的推荐信息，或者直接在本站搜索栏里输入“代理”。这篇文章只是为了展示代码，相关代码我贴在下面了，感兴趣的可以自己部署一个玩玩：

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 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

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)
  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")