1 import asyncore 2 import email.utils 3 import socket 4 import smtpd 5 import smtplib 6 import StringIO 7 import sys 8 import time 9 import select 10 11 import unittest 12 from test import test_support 13 14 try: 15 import threading 16 except ImportError: 17 threading = None 18 19 HOST = test_support.HOST 20 21 def server(evt, buf, serv): 22 serv.listen(5) 23 evt.set() 24 try: 25 conn, addr = serv.accept() 26 except socket.timeout: 27 pass 28 else: 29 n = 500 30 while buf and n > 0: 31 r, w, e = select.select([], [conn], []) 32 if w: 33 sent = conn.send(buf) 34 buf = buf[sent:] 35 36 n -= 1 37 38 conn.close() 39 finally: 40 serv.close() 41 evt.set() 42 43 @unittest.skipUnless(threading, 'Threading required for this test.') 44 class GeneralTests(unittest.TestCase): 45 46 def setUp(self): 47 self._threads = test_support.threading_setup() 48 self.evt = threading.Event() 49 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 50 self.sock.settimeout(15) 51 self.port = test_support.bind_port(self.sock) 52 servargs = (self.evt, "220 Hola mundo\n", self.sock) 53 self.thread = threading.Thread(target=server, args=servargs) 54 self.thread.start() 55 self.evt.wait() 56 self.evt.clear() 57 58 def tearDown(self): 59 self.evt.wait() 60 self.thread.join() 61 test_support.threading_cleanup(*self._threads) 62 63 def testBasic1(self): 64 # connects 65 smtp = smtplib.SMTP(HOST, self.port) 66 smtp.close() 67 68 def testBasic2(self): 69 # connects, include port in host name 70 smtp = smtplib.SMTP("%s:%s" % (HOST, self.port)) 71 smtp.close() 72 73 def testLocalHostName(self): 74 # check that supplied local_hostname is used 75 smtp = smtplib.SMTP(HOST, self.port, local_hostname="testhost") 76 self.assertEqual(smtp.local_hostname, "testhost") 77 smtp.close() 78 79 def testTimeoutDefault(self): 80 self.assertTrue(socket.getdefaulttimeout() is None) 81 socket.setdefaulttimeout(30) 82 try: 83 smtp = smtplib.SMTP(HOST, self.port) 84 finally: 85 socket.setdefaulttimeout(None) 86 self.assertEqual(smtp.sock.gettimeout(), 30) 87 smtp.close() 88 89 def testTimeoutNone(self): 90 self.assertTrue(socket.getdefaulttimeout() is None) 91 socket.setdefaulttimeout(30) 92 try: 93 smtp = smtplib.SMTP(HOST, self.port, timeout=None) 94 finally: 95 socket.setdefaulttimeout(None) 96 self.assertTrue(smtp.sock.gettimeout() is None) 97 smtp.close() 98 99 def testTimeoutValue(self): 100 smtp = smtplib.SMTP(HOST, self.port, timeout=30) 101 self.assertEqual(smtp.sock.gettimeout(), 30) 102 smtp.close() 103 104 105 # Test server thread using the specified SMTP server class 106 def debugging_server(serv, serv_evt, client_evt): 107 serv_evt.set() 108 109 try: 110 if hasattr(select, 'poll'): 111 poll_fun = asyncore.poll2 112 else: 113 poll_fun = asyncore.poll 114 115 n = 1000 116 while asyncore.socket_map and n > 0: 117 poll_fun(0.01, asyncore.socket_map) 118 119 # when the client conversation is finished, it will 120 # set client_evt, and it's then ok to kill the server 121 if client_evt.is_set(): 122 serv.close() 123 break 124 125 n -= 1 126 127 except socket.timeout: 128 pass 129 finally: 130 if not client_evt.is_set(): 131 # allow some time for the client to read the result 132 time.sleep(0.5) 133 serv.close() 134 asyncore.close_all() 135 serv_evt.set() 136 137 MSG_BEGIN = '---------- MESSAGE FOLLOWS ----------\n' 138 MSG_END = '------------ END MESSAGE ------------\n' 139 140 # NOTE: Some SMTP objects in the tests below are created with a non-default 141 # local_hostname argument to the constructor, since (on some systems) the FQDN 142 # lookup caused by the default local_hostname sometimes takes so long that the 143 # test server times out, causing the test to fail. 144 145 # Test behavior of smtpd.DebuggingServer 146 @unittest.skipUnless(threading, 'Threading required for this test.') 147 class DebuggingServerTests(unittest.TestCase): 148 149 def setUp(self): 150 # temporarily replace sys.stdout to capture DebuggingServer output 151 self.old_stdout = sys.stdout 152 self.output = StringIO.StringIO() 153 sys.stdout = self.output 154 155 self._threads = test_support.threading_setup() 156 self.serv_evt = threading.Event() 157 self.client_evt = threading.Event() 158 # Pick a random unused port by passing 0 for the port number 159 self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1)) 160 # Keep a note of what port was assigned 161 self.port = self.serv.socket.getsockname()[1] 162 serv_args = (self.serv, self.serv_evt, self.client_evt) 163 self.thread = threading.Thread(target=debugging_server, args=serv_args) 164 self.thread.start() 165 166 # wait until server thread has assigned a port number 167 self.serv_evt.wait() 168 self.serv_evt.clear() 169 170 def tearDown(self): 171 # indicate that the client is finished 172 self.client_evt.set() 173 # wait for the server thread to terminate 174 self.serv_evt.wait() 175 self.thread.join() 176 test_support.threading_cleanup(*self._threads) 177 # restore sys.stdout 178 sys.stdout = self.old_stdout 179 180 def testBasic(self): 181 # connect 182 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3) 183 smtp.quit() 184 185 def testNOOP(self): 186 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3) 187 expected = (250, 'Ok') 188 self.assertEqual(smtp.noop(), expected) 189 smtp.quit() 190 191 def testRSET(self): 192 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3) 193 expected = (250, 'Ok') 194 self.assertEqual(smtp.rset(), expected) 195 smtp.quit() 196 197 def testNotImplemented(self): 198 # EHLO isn't implemented in DebuggingServer 199 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3) 200 expected = (502, 'Error: command "EHLO" not implemented') 201 self.assertEqual(smtp.ehlo(), expected) 202 smtp.quit() 203 204 def testVRFY(self): 205 # VRFY isn't implemented in DebuggingServer 206 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3) 207 expected = (502, 'Error: command "VRFY" not implemented') 208 self.assertEqual(smtp.vrfy('nobody (at] nowhere.com'), expected) 209 self.assertEqual(smtp.verify('nobody (at] nowhere.com'), expected) 210 smtp.quit() 211 212 def testSecondHELO(self): 213 # check that a second HELO returns a message that it's a duplicate 214 # (this behavior is specific to smtpd.SMTPChannel) 215 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3) 216 smtp.helo() 217 expected = (503, 'Duplicate HELO/EHLO') 218 self.assertEqual(smtp.helo(), expected) 219 smtp.quit() 220 221 def testHELP(self): 222 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3) 223 self.assertEqual(smtp.help(), 'Error: command "HELP" not implemented') 224 smtp.quit() 225 226 def testSend(self): 227 # connect and send mail 228 m = 'A test message' 229 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3) 230 smtp.sendmail('John', 'Sally', m) 231 # XXX(nnorwitz): this test is flaky and dies with a bad file descriptor 232 # in asyncore. This sleep might help, but should really be fixed 233 # properly by using an Event variable. 234 time.sleep(0.01) 235 smtp.quit() 236 237 self.client_evt.set() 238 self.serv_evt.wait() 239 self.output.flush() 240 mexpect = '%s%s\n%s' % (MSG_BEGIN, m, MSG_END) 241 self.assertEqual(self.output.getvalue(), mexpect) 242 243 244 class NonConnectingTests(unittest.TestCase): 245 246 def testNotConnected(self): 247 # Test various operations on an unconnected SMTP object that 248 # should raise exceptions (at present the attempt in SMTP.send 249 # to reference the nonexistent 'sock' attribute of the SMTP object 250 # causes an AttributeError) 251 smtp = smtplib.SMTP() 252 self.assertRaises(smtplib.SMTPServerDisconnected, smtp.ehlo) 253 self.assertRaises(smtplib.SMTPServerDisconnected, 254 smtp.send, 'test msg') 255 256 def testNonnumericPort(self): 257 # check that non-numeric port raises socket.error 258 self.assertRaises(socket.error, smtplib.SMTP, 259 "localhost", "bogus") 260 self.assertRaises(socket.error, smtplib.SMTP, 261 "localhost:bogus") 262 263 264 # test response of client to a non-successful HELO message 265 @unittest.skipUnless(threading, 'Threading required for this test.') 266 class BadHELOServerTests(unittest.TestCase): 267 268 def setUp(self): 269 self.old_stdout = sys.stdout 270 self.output = StringIO.StringIO() 271 sys.stdout = self.output 272 273 self._threads = test_support.threading_setup() 274 self.evt = threading.Event() 275 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 276 self.sock.settimeout(15) 277 self.port = test_support.bind_port(self.sock) 278 servargs = (self.evt, "199 no hello for you!\n", self.sock) 279 self.thread = threading.Thread(target=server, args=servargs) 280 self.thread.start() 281 self.evt.wait() 282 self.evt.clear() 283 284 def tearDown(self): 285 self.evt.wait() 286 self.thread.join() 287 test_support.threading_cleanup(*self._threads) 288 sys.stdout = self.old_stdout 289 290 def testFailingHELO(self): 291 self.assertRaises(smtplib.SMTPConnectError, smtplib.SMTP, 292 HOST, self.port, 'localhost', 3) 293 294 295 sim_users = {'Mr.A (at] somewhere.com':'John A', 296 'Ms.B (at] somewhere.com':'Sally B', 297 'Mrs.C (at] somewhereesle.com':'Ruth C', 298 } 299 300 sim_auth = ('Mr.A (at] somewhere.com', 'somepassword') 301 sim_cram_md5_challenge = ('PENCeUxFREJoU0NnbmhNWitOMjNGNn' 302 'dAZWx3b29kLmlubm9zb2Z0LmNvbT4=') 303 sim_auth_credentials = { 304 'login': 'TXIuQUBzb21ld2hlcmUuY29t', 305 'plain': 'AE1yLkFAc29tZXdoZXJlLmNvbQBzb21lcGFzc3dvcmQ=', 306 'cram-md5': ('TXIUQUBZB21LD2HLCMUUY29TIDG4OWQ0MJ' 307 'KWZGQ4ODNMNDA4NTGXMDRLZWMYZJDMODG1'), 308 } 309 sim_auth_login_password = 'C29TZXBHC3N3B3JK' 310 311 sim_lists = {'list-1':['Mr.A (at] somewhere.com','Mrs.C (at] somewhereesle.com'], 312 'list-2':['Ms.B (at] somewhere.com',], 313 } 314 315 # Simulated SMTP channel & server 316 class SimSMTPChannel(smtpd.SMTPChannel): 317 318 def __init__(self, extra_features, *args, **kw): 319 self._extrafeatures = ''.join( 320 [ "250-{0}\r\n".format(x) for x in extra_features ]) 321 smtpd.SMTPChannel.__init__(self, *args, **kw) 322 323 def smtp_EHLO(self, arg): 324 resp = ('250-testhost\r\n' 325 '250-EXPN\r\n' 326 '250-SIZE 20000000\r\n' 327 '250-STARTTLS\r\n' 328 '250-DELIVERBY\r\n') 329 resp = resp + self._extrafeatures + '250 HELP' 330 self.push(resp) 331 332 def smtp_VRFY(self, arg): 333 raw_addr = email.utils.parseaddr(arg)[1] 334 quoted_addr = smtplib.quoteaddr(arg) 335 if raw_addr in sim_users: 336 self.push('250 %s %s' % (sim_users[raw_addr], quoted_addr)) 337 else: 338 self.push('550 No such user: %s' % arg) 339 340 def smtp_EXPN(self, arg): 341 list_name = email.utils.parseaddr(arg)[1].lower() 342 if list_name in sim_lists: 343 user_list = sim_lists[list_name] 344 for n, user_email in enumerate(user_list): 345 quoted_addr = smtplib.quoteaddr(user_email) 346 if n < len(user_list) - 1: 347 self.push('250-%s %s' % (sim_users[user_email], quoted_addr)) 348 else: 349 self.push('250 %s %s' % (sim_users[user_email], quoted_addr)) 350 else: 351 self.push('550 No access for you!') 352 353 def smtp_AUTH(self, arg): 354 if arg.strip().lower()=='cram-md5': 355 self.push('334 {0}'.format(sim_cram_md5_challenge)) 356 return 357 mech, auth = arg.split() 358 mech = mech.lower() 359 if mech not in sim_auth_credentials: 360 self.push('504 auth type unimplemented') 361 return 362 if mech == 'plain' and auth==sim_auth_credentials['plain']: 363 self.push('235 plain auth ok') 364 elif mech=='login' and auth==sim_auth_credentials['login']: 365 self.push('334 Password:') 366 else: 367 self.push('550 No access for you!') 368 369 def handle_error(self): 370 raise 371 372 373 class SimSMTPServer(smtpd.SMTPServer): 374 375 def __init__(self, *args, **kw): 376 self._extra_features = [] 377 smtpd.SMTPServer.__init__(self, *args, **kw) 378 379 def handle_accept(self): 380 conn, addr = self.accept() 381 self._SMTPchannel = SimSMTPChannel(self._extra_features, 382 self, conn, addr) 383 384 def process_message(self, peer, mailfrom, rcpttos, data): 385 pass 386 387 def add_feature(self, feature): 388 self._extra_features.append(feature) 389 390 def handle_error(self): 391 raise 392 393 394 # Test various SMTP & ESMTP commands/behaviors that require a simulated server 395 # (i.e., something with more features than DebuggingServer) 396 @unittest.skipUnless(threading, 'Threading required for this test.') 397 class SMTPSimTests(unittest.TestCase): 398 399 def setUp(self): 400 self._threads = test_support.threading_setup() 401 self.serv_evt = threading.Event() 402 self.client_evt = threading.Event() 403 # Pick a random unused port by passing 0 for the port number 404 self.serv = SimSMTPServer((HOST, 0), ('nowhere', -1)) 405 # Keep a note of what port was assigned 406 self.port = self.serv.socket.getsockname()[1] 407 serv_args = (self.serv, self.serv_evt, self.client_evt) 408 self.thread = threading.Thread(target=debugging_server, args=serv_args) 409 self.thread.start() 410 411 # wait until server thread has assigned a port number 412 self.serv_evt.wait() 413 self.serv_evt.clear() 414 415 def tearDown(self): 416 # indicate that the client is finished 417 self.client_evt.set() 418 # wait for the server thread to terminate 419 self.serv_evt.wait() 420 self.thread.join() 421 test_support.threading_cleanup(*self._threads) 422 423 def testBasic(self): 424 # smoke test 425 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) 426 smtp.quit() 427 428 def testEHLO(self): 429 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) 430 431 # no features should be present before the EHLO 432 self.assertEqual(smtp.esmtp_features, {}) 433 434 # features expected from the test server 435 expected_features = {'expn':'', 436 'size': '20000000', 437 'starttls': '', 438 'deliverby': '', 439 'help': '', 440 } 441 442 smtp.ehlo() 443 self.assertEqual(smtp.esmtp_features, expected_features) 444 for k in expected_features: 445 self.assertTrue(smtp.has_extn(k)) 446 self.assertFalse(smtp.has_extn('unsupported-feature')) 447 smtp.quit() 448 449 def testVRFY(self): 450 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) 451 452 for email, name in sim_users.items(): 453 expected_known = (250, '%s %s' % (name, smtplib.quoteaddr(email))) 454 self.assertEqual(smtp.vrfy(email), expected_known) 455 456 u = 'nobody (at] nowhere.com' 457 expected_unknown = (550, 'No such user: %s' % smtplib.quoteaddr(u)) 458 self.assertEqual(smtp.vrfy(u), expected_unknown) 459 smtp.quit() 460 461 def testEXPN(self): 462 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) 463 464 for listname, members in sim_lists.items(): 465 users = [] 466 for m in members: 467 users.append('%s %s' % (sim_users[m], smtplib.quoteaddr(m))) 468 expected_known = (250, '\n'.join(users)) 469 self.assertEqual(smtp.expn(listname), expected_known) 470 471 u = 'PSU-Members-List' 472 expected_unknown = (550, 'No access for you!') 473 self.assertEqual(smtp.expn(u), expected_unknown) 474 smtp.quit() 475 476 def testAUTH_PLAIN(self): 477 self.serv.add_feature("AUTH PLAIN") 478 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) 479 480 expected_auth_ok = (235, b'plain auth ok') 481 self.assertEqual(smtp.login(sim_auth[0], sim_auth[1]), expected_auth_ok) 482 483 # SimSMTPChannel doesn't fully support LOGIN or CRAM-MD5 auth because they 484 # require a synchronous read to obtain the credentials...so instead smtpd 485 # sees the credential sent by smtplib's login method as an unknown command, 486 # which results in smtplib raising an auth error. Fortunately the error 487 # message contains the encoded credential, so we can partially check that it 488 # was generated correctly (partially, because the 'word' is uppercased in 489 # the error message). 490 491 def testAUTH_LOGIN(self): 492 self.serv.add_feature("AUTH LOGIN") 493 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) 494 try: smtp.login(sim_auth[0], sim_auth[1]) 495 except smtplib.SMTPAuthenticationError as err: 496 if sim_auth_login_password not in str(err): 497 raise "expected encoded password not found in error message" 498 499 def testAUTH_CRAM_MD5(self): 500 self.serv.add_feature("AUTH CRAM-MD5") 501 smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) 502 503 try: smtp.login(sim_auth[0], sim_auth[1]) 504 except smtplib.SMTPAuthenticationError as err: 505 if sim_auth_credentials['cram-md5'] not in str(err): 506 raise "expected encoded credentials not found in error message" 507 508 #TODO: add tests for correct AUTH method fallback now that the 509 #test infrastructure can support it. 510 511 512 def test_main(verbose=None): 513 test_support.run_unittest(GeneralTests, DebuggingServerTests, 514 NonConnectingTests, 515 BadHELOServerTests, SMTPSimTests) 516 517 if __name__ == '__main__': 518 test_main() 519