将 SMTP AUTH 支持添加到 Python smtpd 库...无法覆盖该方法吗?

发布于 2024-07-29 02:56:10 字数 3350 浏览 6 评论 0原文

因此,我想扩展 Python smtpd SMTPServer 类,以便它可以处理 SMTP AUTH 连接。 看起来很简单......

所以,看起来我可以这样开始:

def smtp_EHLO(self, arg):
    print 'got in arg: ', arg
    # do stuff here...

但由于某种原因,它永远不会被调用。 Python smtpd 库调用其他类似的方法,如下所示:

        method = None
        i = line.find(' ')
        if i < 0:
            command = line.upper()
            arg = None
        else:
            command = line[:i].upper()
            arg = line[i+1:].strip()
        method = getattr(self, 'smtp_' + command, None)

为什么它不调用我的方法?

之后,我想我可能可以重写整个found_terminator(self):方法,但这似乎也不起作用。

 def found_terminator(self):
     # I add this to my child class and it never gets called... 

我是在做一些愚蠢的事情还是……? 也许我今天还没有完全醒来......

import smtpd
import asyncore

class CustomSMTPServer(smtpd.SMTPServer):

    def smtp_EHLO(self, arg):

        print 'got in arg: ', arg

    def process_message(self, peer, mailfrom, rcpttos, data):
        print 'Receiving message from:', peer
        print 'Message addressed from:', mailfrom
        print 'Message addressed to  :', rcpttos
        print 'Message length        :', len(data)
        print 'HERE WE ARE MAN!'
        return

    # Implementation of base class abstract method
    def found_terminator(self):
        print 'THIS GOT CALLED RIGHT HERE!'

        line = EMPTYSTRING.join(self.__line)
        print >> DEBUGSTREAM, 'Data:', repr(line)
        self.__line = []
        if self.__state == self.COMMAND:
            if not line:
                self.push('500 Error: bad syntax')
                return
            method = None
            i = line.find(' ')
            if i < 0:
                command = line.upper()
                arg = None
            else:
                command = line[:i].upper()
                arg = line[i+1:].strip()
            method = getattr(self, 'smtp_' + command, None)

            print 'looking for: ', command
            print 'method is: ', method

            if not method:
                self.push('502 Error: command "%s" not implemented' % command)
                return
            method(arg)
            return
        else:
            if self.__state != self.DATA:
                self.push('451 Internal confusion')
                return
            # Remove extraneous carriage returns and de-transparency according
            # to RFC 821, Section 4.5.2.
            data = []
            for text in line.split('\r\n'):
                if text and text[0] == '.':
                    data.append(text[1:])
                else:
                    data.append(text)
            self.__data = NEWLINE.join(data)
            status = self.__server.process_message(self.__peer,
                                                   self.__mailfrom,
                                                   self.__rcpttos,
                                                   self.__data)
            self.__rcpttos = []
            self.__mailfrom = None
            self.__state = self.COMMAND
            self.set_terminator('\r\n')
            if not status:
                self.push('250 Ok')
            else:
                self.push(status)

server = CustomSMTPServer(('127.0.0.1', 1025), None)

asyncore.loop()

So, I wanted to extend the Python smtpd SMTPServer class so that it could handle SMTP AUTH connections. Seemed simple enough...

So, it looked like I could just start like this:

def smtp_EHLO(self, arg):
    print 'got in arg: ', arg
    # do stuff here...

But for some reason, that never gets called. The Python smtpd library calls other similar methods like this:

        method = None
        i = line.find(' ')
        if i < 0:
            command = line.upper()
            arg = None
        else:
            command = line[:i].upper()
            arg = line[i+1:].strip()
        method = getattr(self, 'smtp_' + command, None)

Why won't it call my method?

After that, I thought that I could probably just override the entire found_terminator(self): method, but that doesn't seem to work either.

 def found_terminator(self):
     # I add this to my child class and it never gets called... 

Am I doing something stupid or...? Maybe I just haven't woken up fully yet today...

import smtpd
import asyncore

class CustomSMTPServer(smtpd.SMTPServer):

    def smtp_EHLO(self, arg):

        print 'got in arg: ', arg

    def process_message(self, peer, mailfrom, rcpttos, data):
        print 'Receiving message from:', peer
        print 'Message addressed from:', mailfrom
        print 'Message addressed to  :', rcpttos
        print 'Message length        :', len(data)
        print 'HERE WE ARE MAN!'
        return

    # Implementation of base class abstract method
    def found_terminator(self):
        print 'THIS GOT CALLED RIGHT HERE!'

        line = EMPTYSTRING.join(self.__line)
        print >> DEBUGSTREAM, 'Data:', repr(line)
        self.__line = []
        if self.__state == self.COMMAND:
            if not line:
                self.push('500 Error: bad syntax')
                return
            method = None
            i = line.find(' ')
            if i < 0:
                command = line.upper()
                arg = None
            else:
                command = line[:i].upper()
                arg = line[i+1:].strip()
            method = getattr(self, 'smtp_' + command, None)

            print 'looking for: ', command
            print 'method is: ', method

            if not method:
                self.push('502 Error: command "%s" not implemented' % command)
                return
            method(arg)
            return
        else:
            if self.__state != self.DATA:
                self.push('451 Internal confusion')
                return
            # Remove extraneous carriage returns and de-transparency according
            # to RFC 821, Section 4.5.2.
            data = []
            for text in line.split('\r\n'):
                if text and text[0] == '.':
                    data.append(text[1:])
                else:
                    data.append(text)
            self.__data = NEWLINE.join(data)
            status = self.__server.process_message(self.__peer,
                                                   self.__mailfrom,
                                                   self.__rcpttos,
                                                   self.__data)
            self.__rcpttos = []
            self.__mailfrom = None
            self.__state = self.COMMAND
            self.set_terminator('\r\n')
            if not status:
                self.push('250 Ok')
            else:
                self.push(status)

server = CustomSMTPServer(('127.0.0.1', 1025), None)

asyncore.loop()

如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论, 你可以免费 注册 一个本站的账号。

评论(2

我三岁 2024-08-05 02:56:10

您需要扩展 SMTPChannel —— 这是实现 smtp_verb 方法的地方; 您的 SMTPServer 扩展只需要返回您自己的通道子类。

You need to extend SMTPChannel -- that's where the smtp_verb methods are implemented; your extension of SMTPServer just needs to return your own subclass of the channel.

一桥轻雨一伞开 2024-08-05 02:56:10

TL&DR: 要向 SMTPChannel 添加附加功能,您只需声明一个函数,然后将其直接添加到 smtpd.SMTPChannel

说明:

SMTPChannel 类旨在响应用户在开放端口(通常是端口 25)上输入的命令。 它搜索可以回答哪些命令的方式基于“内省”,它检查函数的所有可用属性。

请注意,SMTPChannel 中的函数需要以“smtp_”开头。 例如,如果您想响应帮助,您可以创建 smtpd.SMTPChannel.smtp_HELP。

下面的函数来自详细介绍自省

class SMTPChannel(asynchat.async_chat):
  method = getattr(self, 'smtp_' + command, None)

CodeThatWorks

源代码步骤 1:声明一个将被调用的函数

def smtp_HELP(self,arg):
  self.push("[8675309] GPT Answers to HELP")

步骤 2:将以下函数添加到 smtpd.SMTPChannel

class FakeSMTPServer(smtpd.SMTPServer):

"""A Fake smtp server"""
        smtpd.SMTPChannel.smtp_HELP = smtp_HELP

步骤 3:Telnet 到 localhost 25 并检验出

Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
220 vics-imac.fios-router.home ESMTP Sendmail 6.7.4 Sunday 17 March 2019
HELP
[8675309] GPT Answers to HELP

TL&DR: To add additional functionality to SMTPChannel you just need to declare a function, and then add it directly to smtpd.SMTPChannel

Explanation:

The SMTPChannel class is designed to respond to the commands that are entered by the user on the open port (typically port 25). The way it searches for which commands it can answer is based off 'Introspection' where it examines all the available attributes of the function.

Take note that the functions within SMTPChannel need to start with the "smtp_". For Example, if you wanted to respond to HELP you would create smtpd.SMTPChannel.smtp_HELP.

The Function below is from the source-code that details the introspection

class SMTPChannel(asynchat.async_chat):
  method = getattr(self, 'smtp_' + command, None)

CodeThatWorks

Step 1: Declare a FUNCTION that will be called

def smtp_HELP(self,arg):
  self.push("[8675309] GPT Answers to HELP")

Step 2: Add the following function to smtpd.SMTPChannel

class FakeSMTPServer(smtpd.SMTPServer):

"""A Fake smtp server"""
        smtpd.SMTPChannel.smtp_HELP = smtp_HELP

Step 3: Telnet to localhost 25 and test out

Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
220 vics-imac.fios-router.home ESMTP Sendmail 6.7.4 Sunday 17 March 2019
HELP
[8675309] GPT Answers to HELP
~没有更多了~
我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站,即表示您同意使用 Cookies 和您的相关数据。
原文