一个简单的 SMTP 服务器(在 Python 中)
Posted
技术标签:
【中文标题】一个简单的 SMTP 服务器(在 Python 中)【英文标题】:A simple SMTP server (in Python) 【发布时间】:2011-02-11 01:38:49 【问题描述】:您能否建议一个简单的 SMTP 服务器,它具有非常基本的 API(我的意思是非常基本的,用于读取、写入、删除电子邮件),可以在 linux 机器上运行? 我只需要把邮件的症结转成XML格式,然后FTP到另一台机器上。
【问题讨论】:
一些关于使用代码发送电子邮件的有趣读物:codinghorror.com/blog/2010/04/… 【参考方案1】:看看这个SMTP sink server:
from __future__ import print_function
from datetime import datetime
import asyncore
from smtpd import SMTPServer
class EmlServer(SMTPServer):
no = 0
def process_message(self, peer, mailfrom, rcpttos, data):
filename = '%s-%d.eml' % (datetime.now().strftime('%Y%m%d%H%M%S'),
self.no)
f = open(filename, 'w')
f.write(data)
f.close
print('%s saved.' % filename)
self.no += 1
def run():
# start the smtp server on localhost:1025
foo = EmlServer(('localhost', 1025), None)
try:
asyncore.loop()
except KeyboardInterrupt:
pass
if __name__ == '__main__':
run()
它使用smtpd.SMTPServer
将电子邮件转储到文件中。
【讨论】:
谢谢,我是 SMTP 接收服务器的作者。我刚刚从@zero 那里了解到该命令对于测试来说更简单:sudo /usr/lib/python2.4/smtpd.py -n -c DebuggingServer localhost:25
嗯,调整它:sudo python -m smtpd -c DebuggingServer -n localhost:25
当 smtp 服务器/接收器和客户端/发送器在同一台机器上时,它运行良好。但我正在尝试从我网络的另一台计算机上的客户端(将服务器设置为我的 smtp 的 IP服务器)但它不起作用..有什么想法吗?我需要以某种方式打开端口?
@hasen 您如何从外部向此电子邮件服务器发送电子邮件?【参考方案2】:
发送电子邮件确实需要两件事:
一个 SMTP 服务器 - 这可以是 Python SMTP Server 或者您可以使用 GMail 或您的 ISP 的服务器。您可能不需要自己运行。 一个 SMTP 库 - 将电子邮件请求发送到 SMTP 服务器的东西。 Python 附带了一个名为 smtplib 的库,它可以为您做到这一点。这里有大量关于如何使用它的信息:http://docs.python.org/library/smtplib.html对于阅读,有两种选择,具体取决于您从哪个服务器阅读电子邮件。
对于 POP 电子邮件服务器 - 您可以使用 poplib python 库:http://docs.python.org/library/poplib.html 对于 IMAP 电子邮件服务器 - 您可以使用 imaplib python 库:http://docs.python.org/library/imaplib.html【讨论】:
【参考方案3】:为了让 Hasen 的脚本在 Python 3 中运行,我不得不稍微调整一下:
from datetime import datetime
import asyncore
from smtpd import SMTPServer
class EmlServer(SMTPServer):
no = 0
def process_message(self, peer, mailfrom, rcpttos, data, **kwargs):
filename = '%s-%d.eml' % (datetime.now().strftime('%Y%m%d%H%M%S'),
self.no)
print(filename)
f = open(filename, 'wb')
f.write(data)
f.close
print('%s saved.' % filename)
self.no += 1
def run():
EmlServer(('localhost', 25), None)
try:
asyncore.loop()
except KeyboardInterrupt:
pass
if __name__ == '__main__':
run()
【讨论】:
【参考方案4】:一种更现代的方法是使用 aiosmtpd 库(提供文档 here)。
你可以在这里找到一个很好的例子:https://aiosmtpd.readthedocs.io/en/latest/aiosmtpd/docs/controller.html。
【讨论】:
【参考方案5】:我成功使用的两个 python smtp 服务器是:
-
Twisted 的 Mail - 一个非常灵活的邮件库,适用于 SMTP、IMAP...
python-slimta - 一个完整的 MTA(smtp 中继/转发服务器)
Twisted 的例子如下所示
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
# You can run this module directly with:
# twistd -ny emailserver.tac
"""
A toy email server.
"""
from __future__ import print_function
from zope.interface import implementer
from twisted.internet import defer
from twisted.mail import smtp
from twisted.mail.imap4 import LOGINCredentials, PLAINCredentials
from twisted.cred.checkers import InMemoryUsernamePasswordDatabaseDontUse
from twisted.cred.portal import IRealm
from twisted.cred.portal import Portal
@implementer(smtp.IMessageDelivery)
class ConsoleMessageDelivery:
def receivedHeader(self, helo, origin, recipients):
return "Received: ConsoleMessageDelivery"
def validateFrom(self, helo, origin):
# All addresses are accepted
return origin
def validateTo(self, user):
# Only messages directed to the "console" user are accepted.
if user.dest.local == "console":
return lambda: ConsoleMessage()
raise smtp.SMTPBadRcpt(user)
@implementer(smtp.IMessage)
class ConsoleMessage:
def __init__(self):
self.lines = []
def lineReceived(self, line):
self.lines.append(line)
def eomReceived(self):
print("New message received:")
print("\n".join(self.lines))
self.lines = None
return defer.succeed(None)
def connectionLost(self):
# There was an error, throw away the stored lines
self.lines = None
class ConsoleSMTPFactory(smtp.SMTPFactory):
protocol = smtp.ESMTP
def __init__(self, *a, **kw):
smtp.SMTPFactory.__init__(self, *a, **kw)
self.delivery = ConsoleMessageDelivery()
def buildProtocol(self, addr):
p = smtp.SMTPFactory.buildProtocol(self, addr)
p.delivery = self.delivery
p.challengers = "LOGIN": LOGINCredentials, "PLAIN": PLAINCredentials
return p
@implementer(IRealm)
class SimpleRealm:
def requestAvatar(self, avatarId, mind, *interfaces):
if smtp.IMessageDelivery in interfaces:
return smtp.IMessageDelivery, ConsoleMessageDelivery(), lambda: None
raise NotImplementedError()
def main():
from twisted.application import internet
from twisted.application import service
portal = Portal(SimpleRealm())
checker = InMemoryUsernamePasswordDatabaseDontUse()
checker.addUser("guest", "password")
portal.registerChecker(checker)
a = service.Application("Console SMTP Server")
internet.TCPServer(2500, ConsoleSMTPFactory(portal)).setServiceParent(a)
return a
application = main()
【讨论】:
【参考方案6】:这些都是很好的开始示例。
smtpd – SMTP 服务器示例
http://pymotw.com/2/smtpd/index.html
smtplib - 简单邮件传输协议客户端
http://pymotw.com/2/smtplib/index.html
【讨论】:
【参考方案7】:我还想用 Python 启动一个 smtp 服务器并用 Python 发送电子邮件。我想在单个进程中在 Flask Web 应用程序中运行所有这些,这意味着 smtp 服务器必须是非阻塞的。这是我最终找到的解决方案 [gist]:
app.py
from flask import Flask, render_template
from smtp_client import send_email
from smtp_server import SMTPServer
app = Flask(__name__)
@app.route('/send_email')
def email():
server = SMTPServer()
server.start()
try:
send_email()
finally:
server.stop()
return 'OK'
@app.route('/')
def index():
return 'Woohoo'
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0')
smtp_server.py
# smtp_server.py
import smtpd
import asyncore
import threading
class CustomSMTPServer(smtpd.SMTPServer):
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))
return
class SMTPServer():
def __init__(self):
self.port = 1025
def start(self):
'''Start listening on self.port'''
# create an instance of the SMTP server, derived from asyncore.dispatcher
self.smtp = CustomSMTPServer(('0.0.0.0', self.port), None)
# start the asyncore loop, listening for SMTP connection, within a thread
# timeout parameter is important, otherwise code will block 30 seconds
# after the smtp channel has been closed
kwargs = 'timeout':1, 'use_poll': True
self.thread = threading.Thread(target=asyncore.loop, kwargs=kwargs)
self.thread.start()
def stop(self):
'''Stop listening to self.port'''
# close the SMTPserver to ensure no channels connect to asyncore
self.smtp.close()
# now it is safe to wait for asyncore.loop() to exit
self.thread.join()
# check for emails in a non-blocking way
def get(self):
'''Return all emails received so far'''
return self.smtp.emails
if __name__ == '__main__':
server = CustomSMTPServer(('0.0.0.0', 1025), None)
asyncore.loop()
smtp_client.py
import smtplib
import email.utils
from email.mime.text import MIMEText
def send_email():
sender='author@example.com'
recipient='6142546977@tmomail.net'
msg = MIMEText('This is the body of the message.')
msg['To'] = email.utils.formataddr(('Recipient', recipient))
msg['From'] = email.utils.formataddr(('Author', 'author@example.com'))
msg['Subject'] = 'Simple test message'
client = smtplib.SMTP('127.0.0.1', 1025)
client.set_debuglevel(True) # show communication with the server
try:
client.sendmail('author@example.com', [recipient], msg.as_string())
finally:
client.quit()
然后使用python app.py
启动服务器,并在另一个请求中使用curl localhost:5000/send_email
模拟对/send_email
的请求。请注意,要实际发送电子邮件(或短信),您需要跳过此处详述的其他环节:https://blog.codinghorror.com/so-youd-like-to-send-some-email-through-code/。
【讨论】:
是否可以从电子邮件帐户接收电子邮件,例如 gmail ?如果是怎么办? @blob 查看上面的 codinghorror 链接——涉及一些步骤,但这是可能的......【参考方案8】:有Python SMTP server。
这个模块提供了几个类来实现 SMTP 服务器。一个是一个 通用的无操作实现,可以被覆盖,而 另外两个提供特定的邮件发送策略。
【讨论】:
是的,我知道。但是,我无法弄清楚如何使用该库阅读电子邮件!也许可以解决这个问题? 虽然此链接可能会回答问题,但最好在此处包含答案的基本部分并提供链接以供参考。如果链接页面发生更改,仅链接的答案可能会失效。【参考方案9】:如果你想用上面的hasen's答案快速测试Django的send_mail:
# Skip lines 3 and 4 if not using virtualenv.
# At command prompt
mkdir django1
cd django1
virtualenv venv
source venv/bin/activate
pip install django==1.11
django-admin startproject django1 .
# run the Django shell
python manage.py shell
# paste into shell following:
from django.core.mail import send_mail
send_mail(
'Subject here',
'Here is the message.',
'from@example.com',
['to@example.com'],
fail_silently=False,
)
# This should write an email like the following:
Content-Type: text/plain; charset="utf-8"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
Subject: Subject here
From: from@example.com
To: to@example.com
Date: Wed, 02 May 2018 02:12:09 -0000
Message-ID: <20180502021209.32641.51865@i1022>
Here is the message.
send_mail 函数中不需要有有效值。以上值适用于 hasen 的示例。
【讨论】:
以上是关于一个简单的 SMTP 服务器(在 Python 中)的主要内容,如果未能解决你的问题,请参考以下文章