Python aiosmtpd mailtransfagent（MTA）缺少什么？

Question

我想用python编写我自己的小型mailserver应用程序

a）为了更好地了解邮件服务器
b） 认识到自己的特点

所以我的问题是，邮件传输代理缺少什么（除了aiosmtpd），它可以向其他完整的mta发送和接收电子邮件(gmail.com网站, 雅虎...)? 在

我猜：

1）当然是一个域和静态ip
2.）此域的有效证书
…应该可以使用Lets Encrypt
3.）加密
…应该可以使用SSL/Context/Starttls。。。使用aiosmtpd本身
4.）正在解析传出电子邮件的MX DNS条目！？
…应该可以使用python库dnspython
5.）错误处理SMTP通信错误、来自其他MTA的错误回复、跳转！？
6.）排队处理入站和挂起的突发电子邮件！？在

是否缺少任何其他“基本”功能？在

当然我知道，邮件服务器还有很多“高级”功能，比如垃圾邮件检查、恶意软件检查、证书验证、黑名单、规则、邮箱等等。。。在

谢谢你的提示！在

---

编辑：

让我澄清一下我的想法：
我想为俱乐部写一个邮件服务器。它的主要用途是邮件列表服务器。俱乐部的不同团体会有不同的名单。 假设我的域名是myclub.org网站然后会有youth@myclub.org，trainer@myclub.org等等。
只有成员才能使用此邮件服务器，并且只有成员才能接收来自此邮件服务器的电子邮件。其他人将不被允许向此邮件服务器发送电子邮件，也不会从该服务器接收电子邮件。成员的电子邮件地址及其组存储在数据库中。在

将来我想集成一些其他有用的功能，例如：

• 自动提醒
• 聊天机器人，会员可以通过电子邮件控制服务和请求信息

我不需要的是：

• 用户邮箱
• POP/IMAP访问
• 网络接口

开放式继电器问题：

• 在SMTP协商期间，我要拒绝任何不在成员数据库中的[发件人]电子邮件地址。在
• 我想检查发送邮件服务器的有效证书。在
• 电子邮件/会员/天的数量将受到限制。在
• 我不确定，如果我真的需要垃圾邮件检测收到的电子邮件？在

电子邮件丢失问题：

我想我需要一个“轻量级”的重试机制。但是，如果传出的电子邮件在重试后仍无法传递，它将被丢弃，并且只会通知管理员，而不会通知发件人。会员不应该被电子邮件传递问题所困扰。是否有任何Python库可以生成RFC3464兼容的错误回复电子邮件？在

重新启动问题：

我不确定我是否真的需要持久存储的电子邮件，那些尚未发送？在我的用例中，所有发出的电子邮件通常应该在几秒钟内送达（如果没有传递问题发生）。在（计划的）重新启动之前，我可以检查空的发送队列。在

Answer 1

您可以考虑以下特点：

• 消息线程
• 支持交付状态
• 支持POP和IMAP协议
• 支持RFC 2821 SMTP和RFC 2033 LMTP电子邮件传输等协议
• 支持多消息标记
• 支持PGP/MIME（RFC2015）
• 支持列表回复
• 允许每个用户管理自己的邮件列表支持
• 合成期间的消息头控制
• 支持地址组
• 防止邮件列表循环
• 垃圾邮件控制

Answer 2

aiosmtpd是为电子邮件编写自定义路由和头重写规则的优秀工具。但是，aiosmtpd不是MTA，因为它不执行消息队列或DSN生成。MTA的一个流行选择是postfix，由于postfix可以配置为将一个域的所有电子邮件转发到另一个本地SMTP服务器（如aiosmtpd），一个自然的选择是使用postfix作为面向互联网的前端，aiosmtpd作为业务逻辑后端。在

使用postfix作为中间人而不是让aiosmtpd面对公共互联网的优势：

• 不需要在aiosmtpd中处理dnsmx查找，只需通过后缀进行中继(本地主机：25)在
• 不用担心aiosmtpd中不兼容的SMTP客户端
• 不用担心aiosmtpd中的STARTTLS，而是在postfix中配置它（更简单，更坚固）
• 不用担心重新尝试失败的电子邮件传递和发送传递状态通知
• aiosmtpd可以配置为在编程错误时以“瞬时故障”（smtp4xx代码）响应，因此只要在4天内修复编程错误，就不会丢失电子邮件

---

下面是如何将postfix配置为使用aiosmtpd等支持的本地SMTP服务器。在

我们将在端口25上运行postfix，在端口20381上运行aiosmtpd。在

要指定postfix应将example.com的电子邮件中继到运行在端口20381上的SMTP服务器，请将以下内容添加到/etc/postfix/main.cf：

transport_maps = hash:/etc/postfix/smtp_transport
relay_domains = example.com

并使用以下内容创建/etc/postfix/smtp_transport：

^{pr2}$

创建该文件后（以及每次修改它时）运行postmap /etc/postfix/smtp_transport。在

---

在AIOMTPD方面，有一些事情需要考虑。在

最重要的是你如何处理回邮邮件。简而言之，您应该将信封发件人设置为您控制的一个电子邮件地址，该地址专用于接收回执，例如bounce@example.com。当电子邮件到达此地址时，应将其存储在某个地方，以便您可以处理退回，例如从数据库中删除成员电子邮件地址。在

另一个需要考虑的重要事项是如何告诉您的成员的电子邮件提供商您正在进行邮件列表转发。在将电子邮件转发到GROUP@example.com时，可能需要添加以下标题：

Sender: bounce@example.com
List-Name: GROUP
List-Id: GROUP.example.com
List-Unsubscribe: <mailto:postmaster@example.com?subject=unsubscribe%20GROUP>
List-Help: <mailto:postmaster@example.com?subject=list-help>
List-Subscribe: <mailto:postmaster@example.com?subject=subscribe%20GROUP>
Precedence: bulk
X-Auto-Response-Suppress: OOF

在这里，我使用postmaster@example.com作为列表取消订阅请求的收件人。这应该是转发给电子邮件管理员（即您）的地址。在

下面是一个骨架（未经测试）可以完成上述操作。它将bounce电子邮件存储在一个名为bounces的目录中，并通过发件人：-页眉（出现在MEMBERS中的一个）根据组列表（在GROUPS中）。在

import os
import email
import email.utils
import mailbox
import smtplib
import aiosmtpd.controller

LISTEN_HOST = '127.0.0.1'
LISTEN_PORT = 20381
DOMAIN = 'example.com'
BOUNCE_ADDRESS = 'bounce'
POSTMASTER = 'postmaster'
BOUNCE_DIRECTORY = os.path.join(
    os.path.dirname(__file__), 'bounces')


def get_extra_headers(list_name, is_group=True, skip=()):
    list_id = '%s.%s' % (list_name, DOMAIN)
    bounce = '%s@%s' % (BOUNCE_ADDRESS, DOMAIN)
    postmaster = '%s@%s' % (POSTMASTER, DOMAIN)
    unsub = '<mailto:%s?subject=unsubscribe%%20%s>' % (postmaster, list_name)
    help = '<mailto:%s?subject=list-help>' % (postmaster,)
    sub = '<mailto:%s?subject=subscribe%%20%s>' % (postmaster, list_name)
    headers = [
        ('Sender', bounce),
        ('List-Name', list_name),
        ('List-Id', list_id),
        ('List-Unsubscribe', unsub),
        ('List-Help', help),
        ('List-Subscribe', sub),
    ]
    if is_group:
        headers.extend([
            ('Precedence', 'bulk'),
            ('X-Auto-Response-Suppress', 'OOF'),
        ])
    headers = [(k, v) for k, v in headers if k.lower() not in skip]
    return headers


def store_bounce_message(message):
    mbox = mailbox.Maildir(BOUNCE_DIRECTORY)
    mbox.add(message)


MEMBERS = ['foo@example.net', 'bar@example.org',
           'clubadmin@example.org']

GROUPS = {
    'group1': ['foo@example.net', 'bar@example.org'],
    POSTMASTER: ['clubadmin@example.org'],
}


class ClubHandler:
    def validate_sender(self, message):
        from_ = message.get('From')
        if not from_:
            return False
        realname, address = email.utils.parseaddr(from_)
        if address not in MEMBERS:
            return False
        return True

    def translate_recipient(self, local_part):
        try:
            return GROUPS[local_part]
        except KeyError:
            return None

    async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
        local, domain = address.split('@')
        if domain.lower() != DOMAIN:
            return '550 wrong domain'
        if local.lower() == BOUNCE:
            envelope.is_bounce = True
            return '250 OK'
        translated = self.translate_recipient(local.lower())
        if translated is None:
            return '550 no such user'
        envelope.rcpt_tos.extend(translated)
        return '250 OK'

    async def handle_DATA(self, server, session, envelope):
        if getattr(envelope, 'is_bounce', False):
            if len(envelope.rcpt_tos) > 0:
                return '500 Cannot send bounce message to multiple recipients'
            store_bounce_message(envelope.original_content)
            return '250 OK'

        message = email.message_from_bytes(envelope.original_content)
        if not self.validate_sender(message):
            return '500 I do not know you'

        for header_key, header_value in get_extra_headers('club'):
            message[header_key] = header_value

        bounce = '%s@%s' % (BOUNCE_ADDRESS, DOMAIN)
        with smtplib.SMTP('localhost', 25) as smtp:
            smtp.sendmail(bounce, envelope.rcpt_tos, message.as_bytes())

        return '250 OK'


if __name__ == '__main__':
    controller = aiosmtpd.controller.Controller(ClubHandler, hostname=LISTEN_HOST, port=LISTEN_PORT)
    controller.start()
    print("Controller started")
    try:
        while True:
            input()
    except (EOFError, KeyboardInterrupt):
        controller.stop()

Answer 3

运行自己的SMTP服务器最重要的一点是，不能是一个开放中继。这意味着你不能接受陌生人的邮件并将其转发到互联网上的任何目的地，因为这将使垃圾邮件发送者能够通过你的SMTP服务器发送垃圾邮件，这会很快阻止你。在

因此，您的服务器应该

• 从经过身份验证的用户/发送者中继到远程目标，或
• 从陌生人传递到你自己的领域。在

既然你的问题涉及到如何解决发送电子邮件的MX记录，我假设你希望你的服务器接受来自经过身份验证的用户的电子邮件。因此，您需要考虑用户如何向服务器进行身份验证。aiosmtpd当前有一个提供基本smtpauth实现的open pull request；您可以使用它，也可以实现自己的（通过子类化aiosmtpd.smtp.SMTP并实现smtp_AUTH()方法）。在

---

运行自己的SMTP服务器的第二个最重要的事情是，必须在不通知发件人的情况下丢失电子邮件。当您接受来自经过身份验证的用户的电子邮件要转发到外部目标时，您应该让用户知道（通过电子邮件发送一个RFC 3464 Delivery Status Notification）消息是否被延迟或根本没有被传递。在

如果远程目的地接收不到电子邮件，您不应立即丢弃它；您应该稍后再试，并反复尝试，直到您认为您已经尝试了足够长的时间。例如，Postfix在第一次尝试传递邮件失败后等待10分钟，如果第二次尝试失败，则等待20分钟，以此类推，直到几天内尝试传递邮件。在

您还应该注意允许运行邮件服务器的主机重新启动，这意味着您应该将排队的邮件存储在磁盘上。为此，您可以使用mailbox module。在

---

当然，我没有把每一个细节都讲清楚，但我认为以上两点是最重要的，而你似乎没有在你的问题中涵盖它们。在