如何使用 aiosmtpd 正确支持 STARTTLS？
How do I properly support STARTTLS with aiosmtpd?
我几乎直接从 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 wayne@example.com --from "something@example.org" --server localhost --port 8026 -tls
30秒后超时。如果我从服务器删除 ssl_context=context 并从客户端删除 -tls 然后它发送邮件正常。
此外，当我尝试通过 telnet 连接并发送 EHLO whatever 时，服务器实际上关闭了连接。
实现支持 tls 的 aiosmtpd 服务器的正确方法是什么？
我很接近。我从我可以通过 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 wayne@example.com --from "something@example.com" --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 method. It turned out that this was exactly what I needed. Following these instructions 我生成了一个完全不安全的自签名证书：
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')
现在我可以发送电子邮件了。对于 懒惰好奇的人，这里是整个服务器代码：
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 someone@example.com --from "someone_else@example.org" --server localhost --port 8026 -tlsc
基于 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
... 或在发出初始 STARTTLS 命令后使用 openssl s_client 与服务器对话：
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()
我几乎直接从 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 wayne@example.com --from "something@example.org" --server localhost --port 8026 -tls
30秒后超时。如果我从服务器删除 ssl_context=context 并从客户端删除 -tls 然后它发送邮件正常。
此外，当我尝试通过 telnet 连接并发送 EHLO whatever 时，服务器实际上关闭了连接。
实现支持 tls 的 aiosmtpd 服务器的正确方法是什么？
我很接近。我从我可以通过 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 wayne@example.com --from "something@example.com" --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 method. It turned out that this was exactly what I needed. Following these instructions 我生成了一个完全不安全的自签名证书：
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')
现在我可以发送电子邮件了。对于 懒惰好奇的人，这里是整个服务器代码：
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 someone@example.com --from "someone_else@example.org" --server localhost --port 8026 -tlsc
基于 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
... 或在发出初始 STARTTLS 命令后使用 openssl s_client 与服务器对话：
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()