2 # -*- test-case-name: twisted.mail.test.test_pop3client -*-
4 # Copyright (c) Twisted Matrix Laboratories.
5 # See LICENSE for details.
7 from twisted.internet.protocol import Factory
8 from twisted.protocols import basic
9 from twisted.internet import reactor
19 INVALID_SERVER_RESPONSE = False
20 INVALID_CAPABILITY_RESPONSE = False
21 INVALID_LOGIN_RESPONSE = False
22 DENY_CONNECTION = False
23 DROP_CONNECTION = False
24 BAD_TLS_RESPONSE = False
25 TIMEOUT_RESPONSE = False
26 TIMEOUT_DEFERRED = False
30 CONNECTION_MADE = "+OK POP3 localhost v2003.83 server ready"
39 CAPABILITIES_SSL = "STLS"
40 CAPABILITIES_UIDL = "UIDL"
43 INVALID_RESPONSE = "-ERR Unknown request"
44 VALID_RESPONSE = "+OK Command Completed"
45 AUTH_DECLINED = "-ERR LOGIN failed"
46 AUTH_ACCEPTED = "+OK Mailbox open, 0 messages"
47 TLS_ERROR = "-ERR server side error start TLS handshake"
48 LOGOUT_COMPLETE = "+OK quit completed"
49 NOT_LOGGED_IN = "-ERR Unknown AUHORIZATION state command"
51 UIDL = "+OK Unique-ID listing follows\r\n."
52 LIST = "+OK Mailbox scan listing follows\r\n."
53 CAP_START = "+OK Capability list follows:"
56 class POP3TestServer(basic.LineReceiver):
57 def __init__(self, contextFactory = None):
61 self.ctx = contextFactory
63 def sendSTATResp(self, req):
66 def sendUIDLResp(self, req):
69 def sendLISTResp(self, req):
72 def sendCapabilities(self):
74 self.caps = [CAP_START]
77 self.caps.append(CAPABILITIES_UIDL)
80 self.caps.append(CAPABILITIES_SSL)
82 for cap in CAPABILITIES:
84 resp = '\r\n'.join(self.caps)
90 def connectionMade(self):
96 reactor.callLater(20, self.sendGreeting)
101 def sendGreeting(self):
102 self.sendLine(CONNECTION_MADE)
104 def lineReceived(self, line):
105 """Error Conditions"""
108 find = lambda s: uline.find(s) != -1
111 # Do not respond to clients request
119 if INVALID_CAPABILITY_RESPONSE:
120 self.sendLine(INVALID_RESPONSE)
122 self.sendCapabilities()
124 elif find("STLS") and SSL_SUPPORT:
128 if INVALID_LOGIN_RESPONSE:
129 self.sendLine(INVALID_RESPONSE)
134 self.tmpUser = line.split(" ")[1]
135 resp = VALID_RESPONSE
144 pwd = line.split(" ")[1]
146 if self.tmpUser is None or pwd is None:
148 elif self.tmpUser == USER and pwd == PASS:
159 self.loggedIn = False
160 self.sendLine(LOGOUT_COMPLETE)
163 elif INVALID_SERVER_RESPONSE:
164 self.sendLine(INVALID_RESPONSE)
166 elif not self.loggedIn:
167 self.sendLine(NOT_LOGGED_IN)
170 self.sendLine(VALID_RESPONSE)
185 elif not UIDL_SUPPORT:
186 self.sendLine(INVALID_RESPONSE)
195 if SSL_SUPPORT and self.ctx is not None:
196 self.sendLine('+OK Begin TLS negotiation now')
197 self.transport.startTLS(self.ctx)
199 self.sendLine('-ERR TLS not available')
201 def disconnect(self):
202 self.transport.loseConnection()
204 def getContext(self):
206 from twisted.internet import ssl
210 self.ctx = ssl.ClientContextFactory()
211 self.ctx.method = ssl.SSL.TLSv1_METHOD
214 usage = """popServer.py [arg] (default is Standard POP Server with no messages)
215 no_ssl - Start with no SSL support
216 no_uidl - Start with no UIDL support
217 bad_resp - Send a non-RFC compliant response to the Client
218 bad_cap_resp - send a non-RFC compliant response when the Client sends a 'CAPABILITY' request
219 bad_login_resp - send a non-RFC compliant response when the Client sends a 'LOGIN' request
220 deny - Deny the connection
221 drop - Drop the connection after sending the greeting
222 bad_tls - Send a bad response to a STARTTLS
223 timeout - Do not return a response to a Client request
224 to_deferred - Do not return a response on a 'Select' request. This
225 will test Deferred callback handling
226 slow - Wait 20 seconds after the connection is made to return a Server Greeting
229 def printMessage(msg):
230 print "Server Starting in %s mode" % msg
234 if arg.lower() == 'no_ssl':
237 printMessage("NON-SSL")
239 elif arg.lower() == 'no_uidl':
242 printMessage("NON-UIDL")
244 elif arg.lower() == 'bad_resp':
245 global INVALID_SERVER_RESPONSE
246 INVALID_SERVER_RESPONSE = True
247 printMessage("Invalid Server Response")
249 elif arg.lower() == 'bad_cap_resp':
250 global INVALID_CAPABILITY_RESPONSE
251 INVALID_CAPABILITY_RESPONSE = True
252 printMessage("Invalid Capability Response")
254 elif arg.lower() == 'bad_login_resp':
255 global INVALID_LOGIN_RESPONSE
256 INVALID_LOGIN_RESPONSE = True
257 printMessage("Invalid Capability Response")
259 elif arg.lower() == 'deny':
260 global DENY_CONNECTION
261 DENY_CONNECTION = True
262 printMessage("Deny Connection")
264 elif arg.lower() == 'drop':
265 global DROP_CONNECTION
266 DROP_CONNECTION = True
267 printMessage("Drop Connection")
270 elif arg.lower() == 'bad_tls':
271 global BAD_TLS_RESPONSE
272 BAD_TLS_RESPONSE = True
273 printMessage("Bad TLS Response")
275 elif arg.lower() == 'timeout':
276 global TIMEOUT_RESPONSE
277 TIMEOUT_RESPONSE = True
278 printMessage("Timeout Response")
280 elif arg.lower() == 'to_deferred':
281 global TIMEOUT_DEFERRED
282 TIMEOUT_DEFERRED = True
283 printMessage("Timeout Deferred Response")
285 elif arg.lower() == 'slow':
288 printMessage("Slow Greeting")
290 elif arg.lower() == '--help':
300 if len(sys.argv) < 2:
301 printMessage("POP3 with no messages")
309 f.protocol = POP3TestServer
310 reactor.listenTCP(PORT, f)
313 if __name__ == '__main__':