emailserver.tac.py

from __future__ import print_function

from zope.interface import implementer

from twisted.internet import ssl, defer
from twisted.mail import smtp
from twisted.cred.portal import IRealm
from twisted.cred.portal import Portal



@implementer(smtp.IMessageDelivery)
class MyMessageDelivery:
    def receivedHeader(self, helo, origin, recipients):
        return "Received: ConsoleMessageDelivery"

    
    def validateFrom(self, helo, origin):
        # All addresses are accepted
        return origin

    
    def validateTo(self, user):
        dest = user.dest.local #TODO use this?
        passed = True
        if passed:
            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 MySMTPFactory(smtp.SMTPFactory):
    protocol = smtp.ESMTP

    def __init__(self, portal=None, cert=None):
        smtp.SMTPFactory.__init__(self, portal)
        self.delivery = MyMessageDelivery()
        self.cert = cert
    

    def buildProtocol(self, addr):
        p = self.protocol(contextFactory=self.cert.options())
        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, MyMessageDelivery(), 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")

    with open('server.pem') as f:
        certData = f.read()
    cert = ssl.PrivateCertificate.loadPEM(certData)

    factory = MySMTPFactory(portal=portal, cert=cert)
    internet.TCPServer(2500, factory).setServiceParent(a)
    return a

application = main()