"""
Simple IMAP4 client which displays the subjects of all messages in a
particular mailbox.
"""
import sys
from twisted.internet import protocol
from twisted.internet import ssl
from twisted.internet import defer
from twisted.internet import stdio
from twisted.protocols import imap4
from twisted.protocols import basic
from twisted.python import util
from twisted.python import log
class TrivialPrompter(basic.LineReceiver):
from os import linesep as delimiter
promptDeferred = None
def prompt(self, msg):
assert self.promptDeferred is None
self.display(msg)
self.promptDeferred = defer.Deferred()
return self.promptDeferred
def display(self, msg):
self.transport.write(msg)
def lineReceived(self, line):
if self.promptDeferred is None:
return
d, self.promptDeferred = self.promptDeferred, None
d.callback(line)
class SimpleIMAP4Client(imap4.IMAP4Client):
greetDeferred = None
def serverGreeting(self, caps):
self.serverCapabilities = caps
if self.greetDeferred is not None:
d, self.greetDeferred = self.greetDeferred, None
d.callback(self)
class SimpleIMAP4ClientFactory(protocol.ClientFactory):
usedUp = False
protocol = SimpleIMAP4Client
def __init__(self, username, onConn):
self.ctx = ssl.ClientContextFactory()
self.username = username
self.onConn = onConn
def buildProtocol(self, addr):
assert not self.usedUp
self.usedUp = True
p = self.protocol(self.ctx)
p.factory = self
p.greetDeferred = self.onConn
auth = imap4.CramMD5ClientAuthenticator(self.username)
p.registerAuthenticator(auth)
return p
def clientConnectionFailed(self, connector, reason):
d, self.onConn = self.onConn, None
d.errback(reason)
def cbServerGreeting(proto, username, password):
tp = TrivialPrompter()
stdio.StandardIO(tp)
proto.prompt = tp.prompt
proto.display = tp.display
return proto.authenticate(password
).addCallback(cbAuthentication, proto
).addErrback(ebAuthentication, proto, username, password
)
def ebConnection(reason):
log.startLogging(sys.stdout)
log.err(reason)
from twisted.internet import reactor
reactor.stop()
def cbAuthentication(result, proto):
return proto.list("", ""
).addCallback(cbMailboxList, proto
)
def ebAuthentication(failure, proto, username, password):
failure.trap(imap4.NoSupportedAuthentication)
return proto.prompt("No secure authentication available. Login insecurely? (y/N) "
).addCallback(cbInsecureLogin, proto, username, password
)
def cbInsecureLogin(result, proto, username, password):
if result.lower() == "y":
return proto.login(username, password
).addCallback(cbAuthentication, proto
)
return defer.fail(Exception("Login failed for security reasons."))
def cbMailboxList(result, proto):
result = [e[2] for e in result]
s = '\n'.join(['%d. %s' % (n + 1, m) for (n, m) in zip(range(len(result)), result)])
if not s:
return defer.fail(Exception("No mailboxes exist on server!"))
return proto.prompt(s + "\nWhich mailbox? [1] "
).addCallback(cbPickMailbox, proto, result
)
def cbPickMailbox(result, proto, mboxes):
mbox = mboxes[int(result or '1') - 1]
return proto.examine(mbox
).addCallback(cbExamineMbox, proto
)
def cbExamineMbox(result, proto):
return proto.fetchSpecific('1:*',
headerType='HEADER.FIELDS',
headerArgs=['SUBJECT']
).addCallback(cbFetch, proto
)
def cbFetch(result, proto):
keys = result.keys()
keys.sort()
for k in keys:
proto.display('%s %s' % (k, result[k][0][2]))
return proto.logout()
PORT = 143
def main():
hostname = raw_input('IMAP4 Server Hostname: ')
username = raw_input('IMAP4 Username: ')
password = util.getPassword('IMAP4 Password: ')
onConn = defer.Deferred(
).addCallback(cbServerGreeting, username, password
).addErrback(ebConnection
)
factory = SimpleIMAP4ClientFactory(username, onConn)
from twisted.internet import reactor
conn = reactor.connectTCP(hostname, PORT, factory)
reactor.run()
if __name__ == '__main__':
main()