Wishful Coding

Didn't you ever wish your
computer understood you?

Twisted pop3 example server

Remember last time when I showed you the Twisted SMTP server? This time it's a POP3 server. I think it is more common to write SMTP servers as a part of a project(to send notifications and stuff), but I needed a POP3 server, so I think there might be others who need a POP3 server as well.

The process is rather similar to SMTP, only you need to implement the IMailbox interface. This is usually done with some sort of file or db based solution, but I used a simple list in this case. You might want to have a look at the mailbox module for a serious implementation.

Below you'll find a bare bones POP3 server, which can be run with `twistd -ny pop3.tac` and below that a sample telnet session, by running `telnet localhost 1230` in another terminal.

Connected to localhost.
Escape character is '^]'.
+OK <20100818095058.3878.894125537.0@pepijn-de-voss-imac.local>
user guest
+OK USER accepted, send PASS
pass password
+OK Authentication succeeded
+OK 20 1020
retr 1
+OK 51
From: me
To: you
Subject: A test mail

Hello world!
retr 40
-ERR Bad message number argument
Connection closed by foreign host.
An example pop3 server

from twisted.application import internet, service
from twisted.cred.portal import Portal, IRealm
from twisted.internet.protocol import ServerFactory
from twisted.mail import pop3
from twisted.mail.pop3 import IMailbox
from twisted.cred.checkers import InMemoryUsernamePasswordDatabaseDontUse
from zope.interface import implements
from itertools import repeat
from hashlib import md5
from StringIO import StringIO

class SimpleMailbox:

    def __init__(self):
        message = """From: me
To: you
Subject: A test mail

Hello world!"""
        self.messages = [m for m in repeat(message, 20)]

    def listMessages(self, index=None):
        if index != None:
            return len(self.messages[index])
            return [len(m) for m in self.messages]

    def getMessage(self, index):
        return StringIO(self.messages[index])

    def getUidl(self, index):
        return md5(self.messages[index]).hexdigest()

    def deleteMessage(self, index):

    def undeleteMessages(self):

    def sync(self):

class SimpleRealm:

    def requestAvatar(self, avatarId, mind, *interfaces):
        if IMailbox in interfaces:
            return IMailbox, SimpleMailbox(), lambda: None
            raise NotImplementedError()

portal = Portal(SimpleRealm())

checker = InMemoryUsernamePasswordDatabaseDontUse()
checker.addUser("guest", "password")

application = service.Application("example pop3 server")

f = ServerFactory()
f.protocol = pop3.POP3
f.protocol.portal = portal
internet.TCPServer(1230, f).setServiceParent(application)
Published on