| Index: third_party/twisted_8_1/twisted/mail/test/pop3testserver.py
|
| diff --git a/third_party/twisted_8_1/twisted/mail/test/pop3testserver.py b/third_party/twisted_8_1/twisted/mail/test/pop3testserver.py
|
| deleted file mode 100644
|
| index 360aae0dcc0312196af8cf254c59e702f7222a79..0000000000000000000000000000000000000000
|
| --- a/third_party/twisted_8_1/twisted/mail/test/pop3testserver.py
|
| +++ /dev/null
|
| @@ -1,311 +0,0 @@
|
| -#!/usr/bin/python
|
| -# -*- test-case-name: twisted.mail.test.test_pop3client -*-
|
| -
|
| -from twisted.internet.protocol import Factory
|
| -from twisted.protocols import basic
|
| -from twisted.internet import reactor
|
| -import sys, time
|
| -
|
| -USER = "test"
|
| -PASS = "twisted"
|
| -
|
| -PORT = 1100
|
| -
|
| -SSL_SUPPORT = True
|
| -UIDL_SUPPORT = True
|
| -INVALID_SERVER_RESPONSE = False
|
| -INVALID_CAPABILITY_RESPONSE = False
|
| -INVALID_LOGIN_RESPONSE = False
|
| -DENY_CONNECTION = False
|
| -DROP_CONNECTION = False
|
| -BAD_TLS_RESPONSE = False
|
| -TIMEOUT_RESPONSE = False
|
| -TIMEOUT_DEFERRED = False
|
| -SLOW_GREETING = False
|
| -
|
| -"""Commands"""
|
| -CONNECTION_MADE = "+OK POP3 localhost v2003.83 server ready"
|
| -
|
| -CAPABILITIES = [
|
| -"TOP",
|
| -"LOGIN-DELAY 180",
|
| -"USER",
|
| -"SASL LOGIN"
|
| -]
|
| -
|
| -CAPABILITIES_SSL = "STLS"
|
| -CAPABILITIES_UIDL = "UIDL"
|
| -
|
| -
|
| -INVALID_RESPONSE = "-ERR Unknown request"
|
| -VALID_RESPONSE = "+OK Command Completed"
|
| -AUTH_DECLINED = "-ERR LOGIN failed"
|
| -AUTH_ACCEPTED = "+OK Mailbox open, 0 messages"
|
| -TLS_ERROR = "-ERR server side error start TLS handshake"
|
| -LOGOUT_COMPLETE = "+OK quit completed"
|
| -NOT_LOGGED_IN = "-ERR Unknown AUHORIZATION state command"
|
| -STAT = "+OK 0 0"
|
| -UIDL = "+OK Unique-ID listing follows\r\n."
|
| -LIST = "+OK Mailbox scan listing follows\r\n."
|
| -CAP_START = "+OK Capability list follows:"
|
| -
|
| -
|
| -class POP3TestServer(basic.LineReceiver):
|
| - def __init__(self, contextFactory = None):
|
| - self.loggedIn = False
|
| - self.caps = None
|
| - self.tmpUser = None
|
| - self.ctx = contextFactory
|
| -
|
| - def sendSTATResp(self, req):
|
| - self.sendLine(STAT)
|
| -
|
| - def sendUIDLResp(self, req):
|
| - self.sendLine(UIDL)
|
| -
|
| - def sendLISTResp(self, req):
|
| - self.sendLine(LIST)
|
| -
|
| - def sendCapabilities(self):
|
| - if self.caps is None:
|
| - self.caps = [CAP_START]
|
| -
|
| - if UIDL_SUPPORT:
|
| - self.caps.append(CAPABILITIES_UIDL)
|
| -
|
| - if SSL_SUPPORT:
|
| - self.caps.append(CAPABILITIES_SSL)
|
| -
|
| - for cap in CAPABILITIES:
|
| - self.caps.append(cap)
|
| - resp = '\r\n'.join(self.caps)
|
| - resp += "\r\n."
|
| -
|
| - self.sendLine(resp)
|
| -
|
| -
|
| - def connectionMade(self):
|
| - if DENY_CONNECTION:
|
| - self.disconnect()
|
| - return
|
| -
|
| - if SLOW_GREETING:
|
| - reactor.callLater(20, self.sendGreeting)
|
| -
|
| - else:
|
| - self.sendGreeting()
|
| -
|
| - def sendGreeting(self):
|
| - self.sendLine(CONNECTION_MADE)
|
| -
|
| - def lineReceived(self, line):
|
| - """Error Conditions"""
|
| -
|
| - uline = line.upper()
|
| - find = lambda s: uline.find(s) != -1
|
| -
|
| - if TIMEOUT_RESPONSE:
|
| - # Do not respond to clients request
|
| - return
|
| -
|
| - if DROP_CONNECTION:
|
| - self.disconnect()
|
| - return
|
| -
|
| - elif find("CAPA"):
|
| - if INVALID_CAPABILITY_RESPONSE:
|
| - self.sendLine(INVALID_RESPONSE)
|
| - else:
|
| - self.sendCapabilities()
|
| -
|
| - elif find("STLS") and SSL_SUPPORT:
|
| - self.startTLS()
|
| -
|
| - elif find("USER"):
|
| - if INVALID_LOGIN_RESPONSE:
|
| - self.sendLine(INVALID_RESPONSE)
|
| - return
|
| -
|
| - resp = None
|
| - try:
|
| - self.tmpUser = line.split(" ")[1]
|
| - resp = VALID_RESPONSE
|
| - except:
|
| - resp = AUTH_DECLINED
|
| -
|
| - self.sendLine(resp)
|
| -
|
| - elif find("PASS"):
|
| - resp = None
|
| - try:
|
| - pwd = line.split(" ")[1]
|
| -
|
| - if self.tmpUser is None or pwd is None:
|
| - resp = AUTH_DECLINED
|
| - elif self.tmpUser == USER and pwd == PASS:
|
| - resp = AUTH_ACCEPTED
|
| - self.loggedIn = True
|
| - else:
|
| - resp = AUTH_DECLINED
|
| - except:
|
| - resp = AUTH_DECLINED
|
| -
|
| - self.sendLine(resp)
|
| -
|
| - elif find("QUIT"):
|
| - self.loggedIn = False
|
| - self.sendLine(LOGOUT_COMPLETE)
|
| - self.disconnect()
|
| -
|
| - elif INVALID_SERVER_RESPONSE:
|
| - self.sendLine(INVALID_RESPONSE)
|
| -
|
| - elif not self.loggedIn:
|
| - self.sendLine(NOT_LOGGED_IN)
|
| -
|
| - elif find("NOOP"):
|
| - self.sendLine(VALID_RESPONSE)
|
| -
|
| - elif find("STAT"):
|
| - if TIMEOUT_DEFERRED:
|
| - return
|
| - self.sendLine(STAT)
|
| -
|
| - elif find("LIST"):
|
| - if TIMEOUT_DEFERRED:
|
| - return
|
| - self.sendLine(LIST)
|
| -
|
| - elif find("UIDL"):
|
| - if TIMEOUT_DEFERRED:
|
| - return
|
| - elif not UIDL_SUPPORT:
|
| - self.sendLine(INVALID_RESPONSE)
|
| - return
|
| -
|
| - self.sendLine(UIDL)
|
| -
|
| - def startTLS(self):
|
| - if self.ctx is None:
|
| - self.getContext()
|
| -
|
| - if SSL_SUPPORT and self.ctx is not None:
|
| - self.sendLine('+OK Begin TLS negotiation now')
|
| - self.transport.startTLS(self.ctx)
|
| - else:
|
| - self.sendLine('-ERR TLS not available')
|
| -
|
| - def disconnect(self):
|
| - self.transport.loseConnection()
|
| -
|
| - def getContext(self):
|
| - try:
|
| - from twisted.internet import ssl
|
| - except ImportError:
|
| - self.ctx = None
|
| - else:
|
| - self.ctx = ssl.ClientContextFactory()
|
| - self.ctx.method = ssl.SSL.TLSv1_METHOD
|
| -
|
| -
|
| -usage = """popServer.py [arg] (default is Standard POP Server with no messages)
|
| -no_ssl - Start with no SSL support
|
| -no_uidl - Start with no UIDL support
|
| -bad_resp - Send a non-RFC compliant response to the Client
|
| -bad_cap_resp - send a non-RFC compliant response when the Client sends a 'CAPABILITY' request
|
| -bad_login_resp - send a non-RFC compliant response when the Client sends a 'LOGIN' request
|
| -deny - Deny the connection
|
| -drop - Drop the connection after sending the greeting
|
| -bad_tls - Send a bad response to a STARTTLS
|
| -timeout - Do not return a response to a Client request
|
| -to_deferred - Do not return a response on a 'Select' request. This
|
| - will test Deferred callback handling
|
| -slow - Wait 20 seconds after the connection is made to return a Server Greeting
|
| -"""
|
| -
|
| -def printMessage(msg):
|
| - print "Server Starting in %s mode" % msg
|
| -
|
| -def processArg(arg):
|
| -
|
| - if arg.lower() == 'no_ssl':
|
| - global SSL_SUPPORT
|
| - SSL_SUPPORT = False
|
| - printMessage("NON-SSL")
|
| -
|
| - elif arg.lower() == 'no_uidl':
|
| - global UIDL_SUPPORT
|
| - UIDL_SUPPORT = False
|
| - printMessage("NON-UIDL")
|
| -
|
| - elif arg.lower() == 'bad_resp':
|
| - global INVALID_SERVER_RESPONSE
|
| - INVALID_SERVER_RESPONSE = True
|
| - printMessage("Invalid Server Response")
|
| -
|
| - elif arg.lower() == 'bad_cap_resp':
|
| - global INVALID_CAPABILITY_RESPONSE
|
| - INVALID_CAPABILITY_RESPONSE = True
|
| - printMessage("Invalid Capability Response")
|
| -
|
| - elif arg.lower() == 'bad_login_resp':
|
| - global INVALID_LOGIN_RESPONSE
|
| - INVALID_LOGIN_RESPONSE = True
|
| - printMessage("Invalid Capability Response")
|
| -
|
| - elif arg.lower() == 'deny':
|
| - global DENY_CONNECTION
|
| - DENY_CONNECTION = True
|
| - printMessage("Deny Connection")
|
| -
|
| - elif arg.lower() == 'drop':
|
| - global DROP_CONNECTION
|
| - DROP_CONNECTION = True
|
| - printMessage("Drop Connection")
|
| -
|
| -
|
| - elif arg.lower() == 'bad_tls':
|
| - global BAD_TLS_RESPONSE
|
| - BAD_TLS_RESPONSE = True
|
| - printMessage("Bad TLS Response")
|
| -
|
| - elif arg.lower() == 'timeout':
|
| - global TIMEOUT_RESPONSE
|
| - TIMEOUT_RESPONSE = True
|
| - printMessage("Timeout Response")
|
| -
|
| - elif arg.lower() == 'to_deferred':
|
| - global TIMEOUT_DEFERRED
|
| - TIMEOUT_DEFERRED = True
|
| - printMessage("Timeout Deferred Response")
|
| -
|
| - elif arg.lower() == 'slow':
|
| - global SLOW_GREETING
|
| - SLOW_GREETING = True
|
| - printMessage("Slow Greeting")
|
| -
|
| - elif arg.lower() == '--help':
|
| - print usage
|
| - sys.exit()
|
| -
|
| - else:
|
| - print usage
|
| - sys.exit()
|
| -
|
| -def main():
|
| -
|
| - if len(sys.argv) < 2:
|
| - printMessage("POP3 with no messages")
|
| - else:
|
| - args = sys.argv[1:]
|
| -
|
| - for arg in args:
|
| - processArg(arg)
|
| -
|
| - f = Factory()
|
| - f.protocol = POP3TestServer
|
| - reactor.listenTCP(PORT, f)
|
| - reactor.run()
|
| -
|
| -if __name__ == '__main__':
|
| - main()
|
|
|