-1

首先，我正在使用aiosmtpd，并且正在尝试编写一个围绕它的类，以使用 StartTLS 以编程方式启动 SMTP 服务器。现在，直到最近，此代码与您可能传递给它的任何处理程序都按预期工作，例如我编写的用于调整消息参数等的基本消息处理程序，并将其作为消息头的一部分传递。

import asyncio

import aiosmtpd
import aiosmtpd.controller
import aiosmtpd.handlers
import aiosmtpd.smtp

import email

import regex
import logging

import ssl


EMPTYBYTES = b''
COMMASPACE = ', '
CRLF = b'\r\n'
NLCRE = regex.compile(br'\r\n|\r|\n')


class StartTLSServer(aiosmtpd.controller.Controller):
    def __init__(self, handler, ssl_cert_file, ssl_key_file, loop=None, hostname=None,
                 port=8025, *, ready_timeout=1.0, enable_SMTPUTF8=True, decode_data=False,
                 require_starttls=True, smtp_ident=None, data_size_limit=10485760,
                 smtp_timeout=300):
        context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        context.load_cert_chain(ssl_cert_file, ssl_key_file)
        self.tls_context = context
        self.require_starttls = require_starttls
        self.enable_SMTPUTF8 = enable_SMTPUTF8
        self.decode_data = decode_data
        self.smtp_ident = smtp_ident
        self.data_size_limit = data_size_limit
        self.smtp_timeout = smtp_timeout
        super().__init__(handler, loop=loop, hostname=hostname, port=port,
                         ready_timeout=ready_timeout, enable_SMTPUTF8=enable_SMTPUTF8)

    def factory(self):
        return aiosmtpd.smtp.SMTP(self.handler, data_size_limit=self.data_size_limit,
                                  enable_SMTPUTF8=self.enable_SMTPUTF8,
                                  decode_data=self.decode_data,
                                  require_starttls=self.require_starttls,
                                  hostname=self.smtp_ident,
                                  ident=self.smtp_ident,
                                  tls_context=self.tls_context,
                                  timeout=self.smtp_timeout)
                                 
                                  
class MessageHandler(aiosmtpd.handlers.Message):
    def __init__(self, message_class=None, *, loop=None):
        super().__init__(message_class)
        self.loop = loop or asyncio.get_event_loop()

    async def handle_DATA(self, server, session, envelope):
        message = self.prepare_message(session, envelope)
        await self.handle_message(message)
        return '250 OK'

    def prepare_message(self, session, envelope):
        # If the server was created with decode_data True, then data will be a
        # str, otherwise it will be bytes.
        data = envelope.content
        if isinstance(data, bytes):
            message = email.message_from_bytes(data, self.message_class)
        else:
            assert isinstance(data, str), (
              'Expected str or bytes, got {}'.format(type(data)))
            message = email.message_from_string(data, self.message_class)
        message['X-Peer'] = str(session.peer)
        message['X-Envelope-MailFrom'] = envelope.mail_from
        message['X-Envelope-RcptTo'] = COMMASPACE.join(envelope.rcpt_tos)
        return message  # This is handed off to handle_message directly.

    async def handle_message(self, message):
        print(message.as_string())
        return

这驻留在custom_handlers.py其中，然后通过 Python 控制台在测试中调用，如下所示：

>>> from custom_handlers import StartTLSServer, MessageHandler
>>> server = StartTLSServer(MessageHandler, ssl_cert_file="valid cert path", ssl_key_file="valid key path", hostname="0.0.0.0", port=25, require_starttls=True, smtp_ident="StartTLSSMTPServer01")
>>> server.start()

当我想停止测试服务器时，我会简单地做一个server.stop()但是在处理任何消息期间，我们会被这个邪恶的错误阻止：

Traceback (most recent call last):
  File "/home/sysadmin/.local/lib/python3.8/site-packages/aiosmtpd/smtp.py", line 728, in _handle_client
    await method(arg)
  File "/home/sysadmin/.local/lib/python3.8/site-packages/aiosmtpd/smtp.py", line 1438, in smtp_DATA
    status = await self._call_handler_hook('DATA')
  File "/home/sysadmin/.local/lib/python3.8/site-packages/aiosmtpd/smtp.py", line 465, in _call_handler_hook
    status = await hook(self, self.session, self.envelope, *args)
TypeError: handle_DATA() missing 1 required positional argument: 'envelope'

现在，我可以使用传入 SMTP 工厂的任何处理程序来复制它。

但是，我无法使用带有调试处理程序的普通 aiosmtpd 来复制它，就像文档中定义的那样：

aiosmtpd -c aiosmtpd.handlers.Debugging stdout -l 0.0.0.0:8025

...效果很好。将调试处理程序传递到 StartTLSServer 会导致与自定义 MessageHandler 类相同的错误，即使使用调试处理程序也是如此。

我是否遗漏了一些关于我的课程的明显内容，这些课程在这里以一种不同于 aiosmtpd 预期的编程用法的方式爆炸？

4

1 回答 1

2

您缺少()为了实例化 MessageHandler 类的对象：

>>> server = StartTLSServer(MessageHandler(), ...)

当您只是简单地传递MessageHandler时()，aiosmtpd 将尝试调用常规函数MessageHandler.handle_DATA(...)（而不是绑定方法 function MessageHandler().handle_DATA(...)）。

这个常规函数有四个参数：一个 MessageHandler 实例作为它的第一个参数，然后是通常的服务器、会话和信封参数。这解释了为什么错误消息抱怨缺少位置参数。

PS，请注意您的handle_DATA实现是多余的，因为它与基类中的实现相同aiosmtpd.handlers.Message- 所以您可以删除它，它应该仍然可以正常工作。

于 2021-08-04T20:19:45.040 回答