230 lines
8.7 KiB
Python
230 lines
8.7 KiB
Python
import asyncio
|
||
import logging
|
||
import email
|
||
import platform
|
||
from email.policy import default
|
||
from aiosmtpd.controller import Controller
|
||
from aiosmtpd.smtp import SMTP as SMTPProtocol
|
||
from aiosmtpd.handlers import Message
|
||
import os
|
||
import sys
|
||
import threading
|
||
import traceback
|
||
|
||
from ..models.domain import Domain
|
||
from ..models.mailbox import Mailbox
|
||
from ..utils import email_parser
|
||
from ..models import Email
|
||
|
||
from aiosmtpd.smtp import SMTP, Session, Envelope
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 检测是否Windows环境
|
||
IS_WINDOWS = platform.system().lower() == 'windows'
|
||
|
||
class EmailHandler(Message):
|
||
"""处理接收的电子邮件"""
|
||
|
||
def __init__(self, mail_store):
|
||
super().__init__()
|
||
self.mail_store = mail_store
|
||
|
||
def handle_message(self, message):
|
||
"""处理邮件消息，这是Message类的抽象方法，必须实现"""
|
||
# 这个方法在异步DATA处理完成后被调用，但我们的邮件处理逻辑已经在handle_DATA中实现
|
||
# 所以这里只是一个空实现
|
||
return
|
||
|
||
async def handle_DATA(self, server, session, envelope):
|
||
"""处理邮件数据"""
|
||
try:
|
||
logging.info(f"收到邮件: 发件人={envelope.mail_from}, 收件人={envelope.rcpt_tos}")
|
||
logging.debug(f"邮件内容: {envelope.content.decode('utf-8', errors='replace')}")
|
||
|
||
# 检查收件人列表是否有效
|
||
if not envelope.rcpt_tos:
|
||
logging.error("收件人列表无效，无法保存邮件")
|
||
return '550 收件人列表无效'
|
||
elif not isinstance(envelope.rcpt_tos, (list, str)):
|
||
logging.error("收件人列表格式不正确，无法保存邮件")
|
||
return '550 收件人列表格式不正确'
|
||
|
||
# 保存原始邮件数据
|
||
raw_data = envelope.content.decode('utf-8', errors='replace')
|
||
|
||
# 解析邮件数据
|
||
message = email_parser.Parser(policy=default).parsestr(raw_data)
|
||
subject = message.get('Subject', '')
|
||
logging.info(f"邮件主题: {subject}")
|
||
|
||
# 记录邮件结构和内容
|
||
logging.debug(f"邮件结构: is_multipart={message.is_multipart()}")
|
||
if message.is_multipart():
|
||
logging.debug(f"多部分邮件: 部分数量={len(list(message.walk()))}")
|
||
for i, part in enumerate(message.walk()):
|
||
content_type = part.get_content_type()
|
||
logging.debug(f"部分 {i+1}: 内容类型={content_type}")
|
||
|
||
# 使用邮件存储服务保存邮件
|
||
success, error_msg = await self.mail_store.save_email(
|
||
message,
|
||
envelope.mail_from,
|
||
envelope.rcpt_tos,
|
||
raw_data=raw_data
|
||
)
|
||
|
||
if success:
|
||
logging.info(f"邮件保存成功: 来自 {envelope.mail_from} 发送给 {envelope.rcpt_tos}")
|
||
return '250 消息接收完成'
|
||
else:
|
||
logging.error(f"邮件保存失败#server: {error_msg}")
|
||
# 即使保存失败，也返回成功状态码，避免邮件服务器重试
|
||
return '250 消息已收到'
|
||
|
||
except Exception as e:
|
||
logging.error(f"处理邮件时出错: {str(e)}")
|
||
traceback.print_exc()
|
||
return '451 处理邮件时出现错误，请稍后重试'
|
||
|
||
|
||
# 为Windows环境自定义SMTP控制器
|
||
if IS_WINDOWS:
|
||
class WindowsSafeController(Controller):
|
||
"""Windows环境安全的Controller，跳过连接测试"""
|
||
def _trigger_server(self):
|
||
"""Windows环境下跳过SMTP服务器自检连接测试"""
|
||
# 在Windows环境下，我们跳过自检连接测试
|
||
logger.info("Windows环境: 跳过SMTP服务器连接自检")
|
||
return
|
||
|
||
|
||
class SMTPServer:
|
||
"""SMTP服务器实现"""
|
||
|
||
def __init__(self, host='0.0.0.0', port=25, mail_store=None):
|
||
self.host = host
|
||
self.port = port
|
||
self.mail_store = mail_store
|
||
self.controller = None
|
||
self.server_thread = None
|
||
|
||
def start(self):
|
||
"""启动SMTP服务器"""
|
||
if self.controller:
|
||
logger.warning("SMTP服务器已经在运行")
|
||
return
|
||
|
||
try:
|
||
handler = EmailHandler(self.mail_store)
|
||
|
||
# 根据环境选择适当的Controller
|
||
if IS_WINDOWS:
|
||
# Windows环境使用自定义Controller
|
||
logger.info(f"Windows环境: 使用自定义Controller启动SMTP服务器 {self.host}:{self.port}")
|
||
self.controller = WindowsSafeController(
|
||
handler,
|
||
hostname=self.host,
|
||
port=self.port
|
||
)
|
||
else:
|
||
# 非Windows环境使用标准Controller
|
||
self.controller = Controller(
|
||
handler,
|
||
hostname=self.host,
|
||
port=self.port
|
||
)
|
||
|
||
# 在单独的线程中启动服务器
|
||
self.server_thread = threading.Thread(
|
||
target=self.controller.start,
|
||
daemon=True
|
||
)
|
||
self.server_thread.start()
|
||
|
||
logger.info(f"SMTP服务器已启动在 {self.host}:{self.port}")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"启动SMTP服务器失败: {str(e)}")
|
||
return False
|
||
|
||
def stop(self):
|
||
"""停止SMTP服务器"""
|
||
if not self.controller:
|
||
logger.warning("SMTP服务器没有运行")
|
||
return
|
||
|
||
try:
|
||
self.controller.stop()
|
||
self.controller = None
|
||
self.server_thread = None
|
||
logger.info("SMTP服务器已停止")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"停止SMTP服务器失败: {str(e)}")
|
||
return False
|
||
|
||
class MailHandler:
|
||
"""邮件处理器，用于处理接收的SMTP邮件"""
|
||
|
||
def __init__(self, mail_store):
|
||
self.mail_store = mail_store
|
||
|
||
async def handle_EHLO(self, server, session, envelope, hostname):
|
||
session.host_name = hostname
|
||
return '250-AUTH PLAIN\n250-SIZE 52428800\n250 SMTPUTF8'
|
||
|
||
async def handle_MAIL(self, server, session, envelope, address, mail_options=None):
|
||
if not mail_options:
|
||
mail_options = []
|
||
envelope.mail_from = address
|
||
envelope.mail_options.extend(mail_options)
|
||
return '250 OK'
|
||
|
||
async def handle_RCPT(self, server, session, envelope, address, rcpt_options=None):
|
||
if not rcpt_options:
|
||
rcpt_options = []
|
||
envelope.rcpt_tos.append(address)
|
||
envelope.rcpt_options.extend(rcpt_options)
|
||
return '250 OK'
|
||
|
||
async def handle_DATA(self, server, session, envelope):
|
||
"""处理接收到的邮件数据"""
|
||
try:
|
||
logging.info(f"收到邮件: 发件人={envelope.mail_from}, 收件人={envelope.rcpt_tos}")
|
||
|
||
# 保存原始邮件数据
|
||
raw_data = envelope.content.decode('utf-8', errors='replace')
|
||
|
||
# 解析邮件数据
|
||
message = email_parser.parsestr(raw_data)
|
||
subject = message.get('Subject', '')
|
||
logging.info(f"邮件主题: {subject}")
|
||
|
||
# 记录邮件结构和内容
|
||
logging.debug(f"邮件结构: is_multipart={message.is_multipart()}")
|
||
if message.is_multipart():
|
||
logging.debug(f"多部分邮件: 部分数量={len(list(message.walk()))}")
|
||
for i, part in enumerate(message.walk()):
|
||
content_type = part.get_content_type()
|
||
logging.debug(f"部分 {i+1}: 内容类型={content_type}")
|
||
|
||
# 使用邮件存储服务保存邮件
|
||
success, error_msg = await self.mail_store.save_email(
|
||
message,
|
||
envelope.mail_from,
|
||
envelope.rcpt_tos,
|
||
raw_data=raw_data
|
||
)
|
||
|
||
if success:
|
||
logging.info(f"邮件保存成功: 来自 {envelope.mail_from} 发送给 {envelope.rcpt_tos}")
|
||
return '250 消息接收完成'
|
||
else:
|
||
logging.error(f"邮件保存失败: {error_msg}")
|
||
return '451 处理邮件时出现错误，请稍后重试'
|
||
|
||
except Exception as e:
|
||
logging.error(f"处理邮件时出错: {str(e)}")
|
||
traceback.print_exc()
|
||
return '451 处理邮件时出现错误，请稍后重试' |