Python aiosmtpd - 邮件传输代理 (MTA) 缺少什么?

2023-11-24

我想用 python 编写我自己的小型邮件服务器应用程序aiosmtpd

a) 出于教育目的,更好地理解邮件服务器
b) 实现我自己的特点

所以我的问题是,缺少什么(除了 aiosmtpd )邮件传输代理,可以向/从其他完整 MTA(gmail.com、yahoo.com ...)发送和接收电子邮件吗?

我正在猜测:

1.) 当然domain和静态ip
2.) 有效证书对于这个域
...应该可以通过 Lets Encrypt 实现
3.) 加密
...应该可以使用 SSL/Context/Starttls...使用 aiosmtpd 本身
4.) 解决外发电子邮件的 MX DNS 条目!?
...应该可以使用 python 库 dnspython
5.) Error处理 SMTP 通信错误、来自其他 MTA 的错误回复、退回!?
6.) Queue用于处理入站和待处理的出站电子邮件!?

还有其他的吗“基本的”缺少功能?

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

感谢所有提示!


EDIT:

让我澄清一下我的想法:
我想为俱乐部编写一个邮件服务器。其主要用途是邮件列表服务器。俱乐部的不同群体会有不同的名单。 可以说我的域名是myclub.org那么就会有例如[电子邮件受保护], [电子邮件受保护]等等。
仅限会员将被允许​​使用此邮件服务器,并且只有成员才能从该邮件服务器接收电子邮件。其他人将不允许向该邮件服务器发送电子邮件,也不会从该服务器接收电子邮件。成员的电子邮件地址及其组存储在数据库中。

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

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

我不需要什么:

  • 用户邮箱
  • POP/IMAP 访问
  • 网页界面

开放中继问题:

  • 我想在 SMTP 协商期间拒绝任何不在成员数据库中的 [FROM] 电子邮件地址。
  • 我想检查发送邮件服务器的有效证书。
  • 电子邮件/会员/天的数量将受到限制。
  • 我不确定我是否真的需要对传入电子邮件进行垃圾邮件检测?

丢失电子邮件问题:

我想我需要一个“轻量级”重试机制。但是,如果重试后仍无法发送外发电子邮件,该电子邮件将被丢弃,并且只有管理员会收到通知,而不是发件人。会员不应被电子邮件传送问题所困扰。有没有Python库这样可以生成符合 RFC3464 标准的回复邮件出错?

重启问题:

我不确定我是否真的需要持久存储尚未发送的电子邮件?在我的用例中,所有外发电子邮件通常应在几秒钟内送达(如果没有发生送达问题)。在(计划的)重新启动之前,我可以检查是否有空的发送队列。


aiosmtpd 是一个优秀的工具,用于编写电子邮件的自定义路由和标头重写规则。但是,aiosmtpd 不是 MTA,因为它不进行消息排队或 DSN 生成。 MTA 的一个流行选择是 postfix,并且由于 postfix 可以配置为将某个域的所有电子邮件中继到另一个本地 SMTP 服务器(例如 aiosmtpd),因此自然的选择是使用 postfix 作为面向互联网的前端,使用 aiosmtpd 作为业务-逻辑后端。

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

  • 无需在 aiosmtpd 中处理 DNS MX 查找——只需通过 postfix 进行中继 (localhost:25)
  • 不用担心 aiosmtpd 中不兼容的 SMTP 客户端
  • 不用担心 aiosmtpd 中的 STARTTLS——而是在 postfix 中配置它(更简单、更久经沙场)
  • 不用担心重试失败的电子邮件发送和发送发送状态通知
  • aiosmtpd 可以配置为在编程错误时响应“暂时失败”(SMTP 4xx 代码),因此只要在 4 天内修复编程错误,就不会丢失电子邮件

以下是您如何配置 postfix 以与由例如支持的本地 SMTP 服务器一起使用。 aiosmtpd。

我们将在端口 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

Run postmap /etc/postfix/smtp_transport创建该文件后(以及每次修改它时)。


在 aiosmtpd 方面,有一些事情需要考虑。

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

另一件需要考虑的重要事情是如何告诉会员的电子邮件提供商您正在进行邮件列表转发。将电子邮件转发至时,您可能需要添加以下标头[email protected]:

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

在这里,我用了[email protected]作为列表取消订阅请求的收件人。这应该是转发给电子邮件管理员(即您)的地址。

下面是执行上述操作的骨架(未经测试)。它将退回电子邮件存储在名为的目录中bounces并转发带有有效 From:-标头的电子邮件(出现在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 = ['[email protected]', '[email protected]',
           '[email protected]']

GROUPS = {
    'group1': ['fo[email protected]', '[email protected]'],
    POSTMASTER: ['[email protected]'],
}


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()
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

Python aiosmtpd - 邮件传输代理 (MTA) 缺少什么? 的相关文章

  • 如何计算Numpy数组中特定范围内的值?

    我有一个 NumPy 值数组 我想计算有多少个值在特定范围内 例如 x25 我已阅读有关计数器的信息 但它似乎仅对特定值有效 对值范围无效 我已经搜索过 但没有找到任何关于我的具体问题的信息 如果有人能指出我正确的文档 我将不胜感激 谢谢
  • 异常处理的范围规则是什么? [复制]

    这个问题在这里已经有答案了 我偶然发现了一个有趣的场景这个问题 https stackoverflow com q 69464430 6045800 考虑以下简单示例 try 1 0 error error except Exception
  • 特定代码行的类似装饰器的语法

    链接主题 但不重复 装饰器对代码的特定行而不是整个方法进行计时 https stackoverflow com questions 30433910 decorator to time specific lines of the code
  • XGBoost 产生预测结果和概率

    我可能正在文档中查看它 但我想知道 XGBoost 是否有办法生成结果的预测和概率 就我而言 我正在尝试预测多类分类器 如果我能返回Medium 88 那就太好了 分类器 中 预测概率 88 参数 params max depth 3 ob
  • Python daysBetweenDate

    我想我可能有一个无限循环 因为每当我运行代码时 我都会收到一条错误消息 它说 程序因使用 13 CPU 秒而关闭 整个代码 应该以日期作为输入并输出第二天 此代码假设所有月份都是 30 天 除了daysBetweenDates功能正常 其他
  • PyPDF2 复制后返回空白 PDF

    def EncryptPDFFiles password directory pdfFiles success 0 Get all PDF files from a directory for folderName subFolders f
  • 使用 Pandas 解析时避免 Excel 的科学记数法舍入

    我有一个自动生成的 Excel 文件 其中偶尔包含非常大的数字 例如135061808695 在 Excel 文件中 当您单击单元格时 它会显示完整的数字135061808695然而 在视觉上 使用自动 常规 格式 数字显示为1 35063
  • 加速 Numpy 数组上的循环

    在我的代码中 我有一个 for 循环 它对多维 numpy 数组进行索引 并使用每次迭代时获得的子数组进行一些操作 看起来像这样 for sub in Arr do stuff using sub 现在使用完成的东西sub是完全矢量化的 所
  • Python 字典不按顺序排列

    我创建了一个字母表字典 其值从0开始 并根据单词文件增加一定的量 我对最初的字典进行了硬编码 我希望它保持按字母顺序排列 但事实并非如此 我希望它按字母顺序返回字典 基本上与初始字典保持相同 我怎样才能保持秩序 from wordData
  • 如何在 pygame 中水平翻转图像?

    这是在 pygame 如何翻转图像 假设一个图像 猪向右看 时向左看 我按向左箭头键 然后保持这样 即使我不按任何键或者按向上和向下箭头键 那么 当我按向右箭头键时 如何再次将其切换回向右看 并使其保持这种状态 即使我不按任何键或按向上和向
  • Django 视图中的原始 SQL 查询

    我将如何使用原始 SQL 执行以下操作views py from app models import Picture def results request all Picture objects all yes Picture objec
  • 如何从 google place api for python 中的地点 id 获取地点详细信息

    我正在使用 Google Places API 和 Python 来构建一个食品集体智能应用程序 例如周围有哪些餐馆 他们的评级如何 营业时间是什么 等等 我正在Python中执行以下操作 from googleplaces import
  • 在 pandas 中展开列表列时,是否有一种Python式的方法来添加枚举列?

    考虑以下DataFrame gt gt gt df pd DataFrame A 1 2 3 B abc def ghi apply A int B list gt gt gt df A B 0 1 a b c 1 2 d e f 2 3
  • 在Python中寻找坐标系中某些点之间的最短路径

    我编写了一个代码 可以在坐标系中的特定宽度和长度范围内生成所需数量的点 它计算并列出我使用欧几里德方法生成的这些点的距离矩阵 我的代码在这里 import pandas as pd from scipy spatial import dis
  • 使用 ABCMeta 和 EnumMeta 的抽象枚举类[重复]

    这个问题在这里已经有答案了 简单的例子 目标是通过从两者派生的元类创建一个抽象枚举类abc ABCMeta and enum EnumMeta 例如 import abc import enum class ABCEnumMeta abc
  • Python - 函数无法在新线程中运行

    我正试图杀死notepad exe使用此函数在 Windows 上进行处理 import thread wmi os print CMD Kill command called def kill c wmi WMI Commands not
  • 如何使用 google.oauth2 python 库?

    我试图对谷歌机器学习项目的安全预测端点进行简单的休息调用 但它找不到 google oauth2 模块 这是我的代码 import urllib2 from google oauth2 import service account Cons
  • 在ActivePython-2.6中安装pyCurl?

    我过去曾使用过 pyCurl 并让它与我的系统默认 python 安装一起使用 但是 我有一个项目需要 python 更具可移植性 并且我正在使用 ActivePython 2 6 到目前为止 我安装任何其他模块都没有问题 但安装 pyCu
  • 选择 matplotlib xticks 频率

    我正在用字符串作为 x 标签绘制数据 我想控制标签频率 以免文本使轴过载 在下面的示例中 我只想每 3 个刻度看到一个标签 a d g j 我可以做到这一点的一种方法是每 n 个元素用 2 个空字符串替换 my xticks 元素 但我确信
  • 为什么 Pytest 对夹具参数执行嵌套循环

    使用 Pytest 我想编写一个测试函数 该函数接受多个装置作为参数 每个灯具都有几个参数 例如 test demo py 中是一个函数test squared is less than 10需要固定装置 negative integer

随机推荐