A Simple SMTP Server in Python #
Could you please suggest a simple SMTP server with the very basic APIs (by very basic I mean, to read, write, delete email), that could be run on a linux box? I just need to convert the crux of the email into XML format and FTP it to another machine.
Here is Some Solutions :
Solution 1: #
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()
It uses smtpd.SMTPServer to dump emails to files.
Solution 2: #
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()
Solution 3: #
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()
Solution 4: #
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()
