<s> from twisted . internet . protocol import Factory <EOL> from twisted . protocols import basic <EOL> from twisted . internet import reactor <EOL> import sys , time <EOL> USER = "<STR_LIT:test>" <EOL> PASS = "<STR_LIT>" <EOL> PORT = <NUM_LIT> <EOL> SSL_SUPPORT = True <EOL> UIDL_SUPPORT = True <EOL> INVALID_SERVER_RESPONSE = False <EOL> INVALID_CAPABILITY_RESPONSE = False <EOL> INVALID_LOGIN_RESPONSE = False <EOL> DENY_CONNECTION = False <EOL> DROP_CONNECTION = False <EOL> BAD_TLS_RESPONSE = False <EOL> TIMEOUT_RESPONSE = False <EOL> TIMEOUT_DEFERRED = False <EOL> SLOW_GREETING = False <EOL> """<STR_LIT>""" <EOL> CONNECTION_MADE = "<STR_LIT>" <EOL> CAPABILITIES = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> ] <EOL> CAPABILITIES_SSL = "<STR_LIT>" <EOL> CAPABILITIES_UIDL = "<STR_LIT>" <EOL> INVALID_RESPONSE = "<STR_LIT>" <EOL> VALID_RESPONSE = "<STR_LIT>" <EOL> AUTH_DECLINED = "<STR_LIT>" <EOL> AUTH_ACCEPTED = "<STR_LIT>" <EOL> TLS_ERROR = "<STR_LIT>" <EOL> LOGOUT_COMPLETE = "<STR_LIT>" <EOL> NOT_LOGGED_IN = "<STR_LIT>" <EOL> STAT = "<STR_LIT>" <EOL> UIDL = "<STR_LIT>" <EOL> LIST = "<STR_LIT>" <EOL> CAP_START = "<STR_LIT>" <EOL> class POP3TestServer ( basic . LineReceiver ) : <EOL> def __init__ ( self , contextFactory = None ) : <EOL> self . loggedIn = False <EOL> self . caps = None <EOL> self . tmpUser = None <EOL> self . ctx = contextFactory <EOL> def sendSTATResp ( self , req ) : <EOL> self . sendLine ( STAT ) <EOL> def sendUIDLResp ( self , req ) : <EOL> self . sendLine ( UIDL ) <EOL> def sendLISTResp ( self , req ) : <EOL> self . sendLine ( LIST ) <EOL> def sendCapabilities ( self ) : <EOL> if self . caps is None : <EOL> self . caps = [ CAP_START ] <EOL> if UIDL_SUPPORT : <EOL> self . caps . append ( CAPABILITIES_UIDL ) <EOL> if SSL_SUPPORT : <EOL> self . caps . append ( CAPABILITIES_SSL ) <EOL> for cap in CAPABILITIES : <EOL> self . caps . append ( cap ) <EOL> resp = '<STR_LIT:\r\n>' . join ( self . caps ) <EOL> resp += "<STR_LIT>" <EOL> self . sendLine ( resp ) <EOL> def connectionMade ( self ) : <EOL> if DENY_CONNECTION : <EOL> self . disconnect ( ) <EOL> return <EOL> if SLOW_GREETING : <EOL> reactor . callLater ( <NUM_LIT:20> , self . sendGreeting ) <EOL> else : <EOL> self . sendGreeting ( ) <EOL> def sendGreeting ( self ) : <EOL> self . sendLine ( CONNECTION_MADE ) <EOL> def lineReceived ( self , line ) : <EOL> """<STR_LIT>""" <EOL> uline = line . upper ( ) <EOL> find = lambda s : uline . find ( s ) != - <NUM_LIT:1> <EOL> if TIMEOUT_RESPONSE : <EOL> return <EOL> if DROP_CONNECTION : <EOL> self . disconnect ( ) <EOL> return <EOL> elif find ( "<STR_LIT>" ) : <EOL> if INVALID_CAPABILITY_RESPONSE : <EOL> self . sendLine ( INVALID_RESPONSE ) <EOL> else : <EOL> self . sendCapabilities ( ) <EOL> elif find ( "<STR_LIT>" ) and SSL_SUPPORT : <EOL> self . startTLS ( ) <EOL> elif find ( "<STR_LIT>" ) : <EOL> if INVALID_LOGIN_RESPONSE : <EOL> self . sendLine ( INVALID_RESPONSE ) <EOL> return <EOL> resp = None <EOL> try : <EOL> self . tmpUser = line . split ( "<STR_LIT:U+0020>" ) [ <NUM_LIT:1> ] <EOL> resp = VALID_RESPONSE <EOL> except : <EOL> resp = AUTH_DECLINED <EOL> self . sendLine ( resp ) <EOL> elif find ( "<STR_LIT>" ) : <EOL> resp = None <EOL> try : <EOL> pwd = line . split ( "<STR_LIT:U+0020>" ) [ <NUM_LIT:1> ] <EOL> if self . tmpUser is None or pwd is None : <EOL> resp = AUTH_DECLINED <EOL> elif self . tmpUser == USER and pwd == PASS : <EOL> resp = AUTH_ACCEPTED <EOL> self . loggedIn = True <EOL> else : <EOL> resp = AUTH_DECLINED <EOL> except : <EOL> resp = AUTH_DECLINED <EOL> self . sendLine ( resp ) <EOL> elif find ( "<STR_LIT>" ) : <EOL> self . loggedIn = False <EOL> self . sendLine ( LOGOUT_COMPLETE ) <EOL> self . disconnect ( ) <EOL> elif INVALID_SERVER_RESPONSE : <EOL> self . sendLine ( INVALID_RESPONSE ) <EOL> elif not self . loggedIn : <EOL> self . sendLine ( NOT_LOGGED_IN ) <EOL> elif find ( "<STR_LIT>" ) : <EOL> self . sendLine ( VALID_RESPONSE ) <EOL> elif find ( "<STR_LIT>" ) : <EOL> if TIMEOUT_DEFERRED : <EOL> return <EOL> self . sendLine ( STAT ) <EOL> elif find ( "<STR_LIT>" ) : <EOL> if TIMEOUT_DEFERRED : <EOL> return <EOL> self . sendLine ( LIST ) <EOL> elif find ( "<STR_LIT>" ) : <EOL> if TIMEOUT_DEFERRED : <EOL> return <EOL> elif not UIDL_SUPPORT : <EOL> self . sendLine ( INVALID_RESPONSE ) <EOL> return <EOL> self . sendLine ( UIDL ) <EOL> def startTLS ( self ) : <EOL> if self . ctx is None : <EOL> self . getContext ( ) <EOL> if SSL_SUPPORT and self . ctx is not None : <EOL> self . sendLine ( '<STR_LIT>' ) <EOL> self . transport . startTLS ( self . ctx ) <EOL> else : <EOL> self . sendLine ( '<STR_LIT>' ) <EOL> def disconnect ( self ) : <EOL> self . transport . loseConnection ( ) <EOL> def getContext ( self ) : <EOL> try : <EOL> from twisted . internet import ssl <EOL> except ImportError : <EOL> self . ctx = None <EOL> else : <EOL> self . ctx = ssl . ClientContextFactory ( ) <EOL> self . ctx . method = ssl . SSL . TLSv1_METHOD <EOL> usage = """<STR_LIT>""" <EOL> def printMessage ( msg ) : <EOL> print "<STR_LIT>" % msg <EOL> def processArg ( arg ) : <EOL> if arg . lower ( ) == '<STR_LIT>' : <EOL> global SSL_SUPPORT <EOL> SSL_SUPPORT = False <EOL> printMessage ( "<STR_LIT>" ) <EOL> elif arg . lower ( ) == '<STR_LIT>' : <EOL> global UIDL_SUPPORT <EOL> UIDL_SUPPORT = False <EOL> printMessage ( "<STR_LIT>" ) <EOL> elif arg . lower ( ) == '<STR_LIT>' : <EOL> global INVALID_SERVER_RESPONSE <EOL> INVALID_SERVER_RESPONSE = True <EOL> printMessage ( "<STR_LIT>" ) <EOL> elif arg . lower ( ) == '<STR_LIT>' : <EOL> global INVALID_CAPABILITY_RESPONSE <EOL> INVALID_CAPABILITY_RESPONSE = True <EOL> printMessage ( "<STR_LIT>" ) <EOL> elif arg . lower ( ) == '<STR_LIT>' : <EOL> global INVALID_LOGIN_RESPONSE <EOL> INVALID_LOGIN_RESPONSE = True <EOL> printMessage ( "<STR_LIT>" ) <EOL> elif arg . lower ( ) == '<STR_LIT>' : <EOL> global DENY_CONNECTION <EOL> DENY_CONNECTION = True <EOL> printMessage ( "<STR_LIT>" ) <EOL> elif arg . lower ( ) == '<STR_LIT>' : <EOL> global DROP_CONNECTION <EOL> DROP_CONNECTION = True <EOL> printMessage ( "<STR_LIT>" ) <EOL> elif arg . lower ( ) == '<STR_LIT>' : <EOL> global BAD_TLS_RESPONSE <EOL> BAD_TLS_RESPONSE = True <EOL> printMessage ( "<STR_LIT>" ) <EOL> elif arg . lower ( ) == '<STR_LIT>' : <EOL> global TIMEOUT_RESPONSE <EOL> TIMEOUT_RESPONSE = True <EOL> printMessage ( "<STR_LIT>" ) <EOL> elif arg . lower ( ) == '<STR_LIT>' : <EOL> global TIMEOUT_DEFERRED <EOL> TIMEOUT_DEFERRED = True <EOL> printMessage ( "<STR_LIT>" ) <EOL> elif arg . lower ( ) == '<STR_LIT>' : <EOL> global SLOW_GREETING <EOL> SLOW_GREETING = True <EOL> printMessage ( "<STR_LIT>" ) <EOL> elif arg . lower ( ) == '<STR_LIT>' : <EOL> print usage <EOL> sys . exit ( ) <EOL> else : <EOL> print usage <EOL> sys . exit ( ) <EOL> def main ( ) : <EOL> if len ( sys . argv ) < <NUM_LIT:2> : <EOL> printMessage ( "<STR_LIT>" ) <EOL> else : <EOL> args = sys . argv [ <NUM_LIT:1> : ] <EOL> for arg in args : <EOL> processArg ( arg ) <EOL> f = Factory ( ) <EOL> f . protocol = POP3TestServer <EOL> reactor . listenTCP ( PORT , f ) <EOL> reactor . run ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( ) </s>
