如何使用 aiosmtpd 正确支持 STARTTLS？

问题描述 投票：0回答：3

我几乎直接从 aiosmtpd 文档获取以下服务器：

import asyncio
import ssl
from aiosmtpd.controller import Controller


class ExampleHandler:
    async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
        if not address.endswith('@example.com'):
            return '550 not relaying to that domain'
        envelope.rcpt_tos.append(address)
        return '250 OK'

    async def handle_DATA(self, server, session, envelope):
        print(f'Message from {envelope.mail_from}')
        print(f'Message for {envelope.rcpt_tos}')
        print(f'Message data:\n{envelope.content.decode("utf8", errors="replace")}')
        print('End of message')
        return '250 Message accepted for delivery'

context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
controller = Controller(ExampleHandler(), port=8026, ssl_context=context)
controller.start()

input('Press enter to stop')
controller.stop()

但是，当我启动此服务器并尝试使用 swaks 向其发送电子邮件时：

echo 'Testing' | swaks --to [email protected] --from "[email protected]" --server localhost --port 8026 -tls

30秒后超时。如果我从服务器中删除 

ssl_context=context
并从客户端中删除 
-tls
，那么它会正常发送邮件。

此外，当我尝试通过 telnet 连接并发送 

EHLO whatever
时，服务器实际上会关闭连接。

实现支持tls的aiosmtpd服务器的正确方法是什么？

python python-3.x aiosmtpd
3个回答
12
投票

基于 Wayne 自己的答案，以下是如何使用 aiosmtpd 创建 STARTTLS 服务器。

1.创建 SSL 上下文

为了进行测试，请使用以下命令为 

localhost
生成自签名证书：

openssl req -x509 -newkey rsa:4096 -keyout key.pem -out cert.pem -days 365 -nodes -subj '/CN=localhost'

使用 

ssl
模块将其加载到 Python 中：

import ssl
context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
context.load_cert_chain('cert.pem', 'key.pem')

2.将 SSL 上下文传递给 aiosmtpd

创建 aiosmtpd 控制器的子类，将此上下文作为 

tls_context
传递到 
SMTP
:

from aiosmtpd.smtp import SMTP
from aiosmtpd.controller import Controller

class ControllerTls(Controller):
    def factory(self):
        return SMTP(self.handler, require_starttls=True, tls_context=context)

3.运行它

使用处理程序实例化该控制器并启动它。这里，我使用 aiosmtpd 自己的 

Debugging
处理程序：

from aiosmtpd.handlers import Debugging
controller = ControllerTls(Debugging(), port=1025)
controller.start()
input('Press enter to stop')
controller.stop()

4.测试一下

配置本地邮件客户端发送至 

localhost:1025
，或使用 
swaks
:

swaks -tls -t test --server localhost:1025

...或在发出初始 

openssl s_client
命令后使用 
STARTTLS
与服务器对话：

openssl s_client -crlf -CAfile cert.pem -connect localhost:1025 -starttls smtp

完整代码

下面的代码还使用 swaks 测试服务器，并且还展示了如何创建 TLS-on-connect 服务器（如 Wayne 的回答）。

import os
import ssl
import subprocess
from aiosmtpd.smtp import SMTP
from aiosmtpd.controller import Controller
from aiosmtpd.handlers import Debugging

# Create cert and key if they don't exist
if not os.path.exists('cert.pem') and not os.path.exists('key.pem'):
    subprocess.call('openssl req -x509 -newkey rsa:4096 -keyout key.pem -out cert.pem ' +
                    '-days 365 -nodes -subj "/CN=localhost"', shell=True)

# Load SSL context
context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
context.load_cert_chain('cert.pem', 'key.pem')

# Pass SSL context to aiosmtpd
class ControllerStarttls(Controller):
    def factory(self):
        return SMTP(self.handler, require_starttls=True, tls_context=context)

# Start server
controller = ControllerStarttls(Debugging(), port=1025)
controller.start()
# Test using swaks (if available)
subprocess.call('swaks -tls -t test --server localhost:1025', shell=True)
input('Running STARTTLS server. Press enter to stop.\n')
controller.stop()

# Alternatively: Use TLS-on-connect
controller = Controller(Debugging(), port=1025, ssl_context=context)
controller.start()
# Test using swaks (if available)
subprocess.call('swaks -tlsc -t test --server localhost:1025', shell=True)
input('Running TLSC server. Press enter to stop.\n')
controller.stop()
3
投票

我很接近。我认为我可以通过 telnet 连接，但是 

EHLO hostname
会断开连接，表明服务器正在尝试提前要求 TLS 连接。

当我检查

swaks --help
时，我发现有一个稍微不同的选项可能会达到我想要的效果：

--tlsc, --tls-on-connect
    Initiate a TLS connection immediately on connection.  Following common convention,
    if this option is specified the default port changes from 25 to 465, though this can
    still be overridden with the --port option.

当我尝试时，我仍然收到错误：

$ echo 'Testing' | swaks --to [email protected] --from "[email protected]" --server localhost --port 8026 -tlsc
=== Trying localhost:8026...
=== Connected to localhost.
*** TLS startup failed (connect(): error:00000000:lib(0):func(0):reason(0))

通过仔细阅读 Python ssl 文档，我注意到了 

load_cert_chain
方法。事实证明这正是我所需要的。按照这些说明我生成了一个完全不安全的自签名证书：

openssl req -x509 -newkey rsa:4096 -keyout key.pem -out cert.pem -days 365 -nodes -subj '/CN=localhost'

然后我添加了这一行：

context.load_cert_chain('cert.pem', 'key.pem')

现在我可以发送电子邮件了。对于 lazy 好奇的人，这里是整个服务器代码：

import asyncio
import ssl
from aiosmtpd.controller import Controller


class ExampleHandler:
    async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
        if not address.endswith('@example.com'):
            return '550 not relaying to that domain'
        envelope.rcpt_tos.append(address)
        return '250 OK'

    async def handle_DATA(self, server, session, envelope):
        print(f'Message from {envelope.mail_from}')
        print(f'Message for {envelope.rcpt_tos}')
        print(f'Message data:\n{envelope.content.decode("utf8", errors="replace")}')
        print('End of message')
        return '250 Message accepted for delivery'

context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
context.load_cert_chain('cert.pem', 'key.pem')
controller = Controller(ExampleHandler(), port=8026, ssl_context=context)
controller.start()

input('Press enter to stop')
controller.stop()

可以通过以下方式验证：

echo 'Testing' | swaks --to [email protected] --from "[email protected]" --server localhost --port 8026 -tlsc
0
投票

如果 Mathias 的答案不适合您，请尝试以下操作：

import asyncio
import ssl
from aiosmtpd.controller import Controller
from aiosmtpd.handlers import Debugging


ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.load_cert_chain('keys/fullchain.pem', 'keys/privkey.pem')

PORT = 8025

controller = Controller(
    Debugging(),
    hostname='0.0.0.0',
    port=PORT,
    server_kwargs={
        'tls_context': ssl_context,
        'require_starttls': False  # if True, starttls will be a requirment
    }
)
controller.start()

try:
    input(f"SMTP server is running on port {PORT}. Press Enter to stop...")
finally:
    controller.stop()
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.