Fra*_*ky1 4 python mail-server aiosmtpd
我想用aiosmtpd在 python 中编写我自己的小型邮件服务器应用程序
a) 出于教育目的以更好地理解邮件服务器
b) 实现我自己的功能
所以我的问题是，Mail-Transfer-Agent缺少什么（除了 aiosmtpd），它可以向/从其他完整的 MTA（gmail.com、yahoo.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等等。
只有会员才能使用此邮件服务器，并且只有会员才能收到来自此邮件服务器的电子邮件。不允许其他人向该邮件服务器发送电子邮件，也不会接收来自该服务器的电子邮件。成员电子邮件地址及其组存储在数据库中。
将来我想集成一些其他有用的功能，例如：
我不需要的东西：
打开中继问题：
丢失电子邮件问题：
我想我需要一个“轻量级”的重试机制。但是，如果在多次重试后仍无法发送外发电子邮件，它将被丢弃，并且只会通知管理员，而不是发件人。会员不应被电子邮件传送问题所困扰。是否有任何Python 库可以生成符合 RFC3464 的错误回复电子邮件？
重启问题：
我不确定我是否真的需要持久存储尚未发送的电子邮件？在我的用例中，所有传出的电子邮件通常应该在几秒钟内送达（如果没有发生送达问题）。在（计划中的）重启之前，我可以检查一个空的发送队列。
aiosmtpd 是为电子邮件编写自定义路由和标题重写规则的出色工具。但是，aiosmtpd 不是 MTA，因为它不执行消息队列或 DSN 生成。MTA 的一个流行选择是 postfix，并且由于 postfix 可以配置为将域的所有电子邮件中继到另一个本地 SMTP 服务器（例如 aiosmtpd），自然的选择是使用 postfix 作为面向互联网的前端和 aiosmtpd 作为业务-逻辑后端。
使用 postfix 作为中间人而不是让 aiosmtpd 面对公共互联网的优点：
以下是您如何配置 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使用内容创建：
# Table of special transport method for domains in
# virtual_mailbox_domains. See postmap(5), virtual(5) and
# transport(5).
#
# Remember to run
# postmap /etc/postfix/smtp_transport
# and update relay_domains in main.cf after changing this file!
example.com smtp:127.0.0.1:20381
postmap /etc/postfix/smtp_transport创建该文件后运行（以及每次修改它时）。
在 aiosmtpd 方面，有几件事情需要考虑。
最重要的是您如何处理退回的电子邮件。简而言之，您应该将信封发件人设置为您控制的专门用于接收退回邮件的电子邮件地址，例如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列表取消订阅请求的接收者。这应该是转发给电子邮件管理员（即您）的地址。
下面是执行上述操作的骨架（未经测试）。它将退回的电子邮件存储在一个名为的目录中，bounces并MEMBERS根据组列表 (in GROUPS)转发带有有效 From:-header（出现在 中的）的电子邮件。
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()
您可以考虑以下功能：