diff options
author | Barry Warsaw <barry@python.org> | 2015-07-09 14:39:55 (GMT) |
---|---|---|
committer | Barry Warsaw <barry@python.org> | 2015-07-09 14:39:55 (GMT) |
commit | c5ea754e484d73f04b1a361d82d0eed1b51dfdc8 (patch) | |
tree | 9b311e7a3649ea679a47a81ea085efe3c6ace849 /Lib | |
parent | b85b427507a8034878d08c2abe2d6b1b7eab97d0 (diff) | |
download | cpython-c5ea754e484d73f04b1a361d82d0eed1b51dfdc8.zip cpython-c5ea754e484d73f04b1a361d82d0eed1b51dfdc8.tar.gz cpython-c5ea754e484d73f04b1a361d82d0eed1b51dfdc8.tar.bz2 |
- Issue #15014: SMTP.auth() and SMTP.login() now support RFC 4954's optional
initial-response argument to the SMTP AUTH command.
Diffstat (limited to 'Lib')
-rwxr-xr-x | Lib/smtplib.py | 52 | ||||
-rw-r--r-- | Lib/test/test_smtplib.py | 92 |
2 files changed, 119 insertions, 25 deletions
diff --git a/Lib/smtplib.py b/Lib/smtplib.py index 71ccd2a..4fe6aaf 100755 --- a/Lib/smtplib.py +++ b/Lib/smtplib.py @@ -601,7 +601,7 @@ class SMTP: if not (200 <= code <= 299): raise SMTPHeloError(code, resp) - def auth(self, mechanism, authobject): + def auth(self, mechanism, authobject, *, initial_response_ok=True): """Authentication command - requires response processing. 'mechanism' specifies which authentication mechanism is to @@ -615,32 +615,46 @@ class SMTP: It will be called to process the server's challenge response; the challenge argument it is passed will be a bytes. It should return bytes data that will be base64 encoded and sent to the server. - """ + Keyword arguments: + - initial_response_ok: Allow sending the RFC 4954 initial-response + to the AUTH command, if the authentication methods supports it. + """ + # RFC 4954 allows auth methods to provide an initial response. Not all + # methods support it. By definition, if they return something other + # than None when challenge is None, then they do. See issue #15014. mechanism = mechanism.upper() - (code, resp) = self.docmd("AUTH", mechanism) - # Server replies with 334 (challenge) or 535 (not supported) - if code == 334: - challenge = base64.decodebytes(resp) - response = encode_base64( - authobject(challenge).encode('ascii'), eol='') - (code, resp) = self.docmd(response) - if code in (235, 503): - return (code, resp) + initial_response = (authobject() if initial_response_ok else None) + if initial_response is not None: + response = encode_base64(initial_response.encode('ascii'), eol='') + (code, resp) = self.docmd("AUTH", mechanism + " " + response) + else: + (code, resp) = self.docmd("AUTH", mechanism) + # Server replies with 334 (challenge) or 535 (not supported) + if code == 334: + challenge = base64.decodebytes(resp) + response = encode_base64( + authobject(challenge).encode('ascii'), eol='') + (code, resp) = self.docmd(response) + if code in (235, 503): + return (code, resp) raise SMTPAuthenticationError(code, resp) - def auth_cram_md5(self, challenge): + def auth_cram_md5(self, challenge=None): """ Authobject to use with CRAM-MD5 authentication. Requires self.user and self.password to be set.""" + # CRAM-MD5 does not support initial-response. + if challenge is None: + return None return self.user + " " + hmac.HMAC( self.password.encode('ascii'), challenge, 'md5').hexdigest() - def auth_plain(self, challenge): + def auth_plain(self, challenge=None): """ Authobject to use with PLAIN authentication. Requires self.user and self.password to be set.""" return "\0%s\0%s" % (self.user, self.password) - def auth_login(self, challenge): + def auth_login(self, challenge=None): """ Authobject to use with LOGIN authentication. Requires self.user and self.password to be set.""" (code, resp) = self.docmd( @@ -649,13 +663,17 @@ class SMTP: return self.password raise SMTPAuthenticationError(code, resp) - def login(self, user, password): + def login(self, user, password, *, initial_response_ok=True): """Log in on an SMTP server that requires authentication. The arguments are: - user: The user name to authenticate with. - password: The password for the authentication. + Keyword arguments: + - initial_response_ok: Allow sending the RFC 4954 initial-response + to the AUTH command, if the authentication methods supports it. + If there has been no previous EHLO or HELO command this session, this method tries ESMTP EHLO first. @@ -698,7 +716,9 @@ class SMTP: for authmethod in authlist: method_name = 'auth_' + authmethod.lower().replace('-', '_') try: - (code, resp) = self.auth(authmethod, getattr(self, method_name)) + (code, resp) = self.auth( + authmethod, getattr(self, method_name), + initial_response_ok=initial_response_ok) # 235 == 'Authentication successful' # 503 == 'Error: already authenticated' if code in (235, 503): diff --git a/Lib/test/test_smtplib.py b/Lib/test/test_smtplib.py index e66ae9b..8e362414 100644 --- a/Lib/test/test_smtplib.py +++ b/Lib/test/test_smtplib.py @@ -1,6 +1,7 @@ import asyncore import email.mime.text from email.message import EmailMessage +from email.base64mime import body_encode as encode_base64 import email.utils import socket import smtpd @@ -814,11 +815,11 @@ class SMTPSimTests(unittest.TestCase): def testVRFY(self): smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) - for email, name in sim_users.items(): + for addr_spec, name in sim_users.items(): expected_known = (250, bytes('%s %s' % - (name, smtplib.quoteaddr(email)), + (name, smtplib.quoteaddr(addr_spec)), "ascii")) - self.assertEqual(smtp.vrfy(email), expected_known) + self.assertEqual(smtp.vrfy(addr_spec), expected_known) u = 'nobody@nowhere.com' expected_unknown = (550, ('No such user: %s' % u).encode('ascii')) @@ -851,7 +852,7 @@ class SMTPSimTests(unittest.TestCase): def testAUTH_PLAIN(self): self.serv.add_feature("AUTH PLAIN") smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) - try: smtp.login(sim_auth[0], sim_auth[1]) + try: smtp.login(sim_auth[0], sim_auth[1], initial_response_ok=False) except smtplib.SMTPAuthenticationError as err: self.assertIn(sim_auth_plain, str(err)) smtp.close() @@ -892,7 +893,7 @@ class SMTPSimTests(unittest.TestCase): 'LOGIN': smtp.auth_login, } for mechanism, method in supported.items(): - try: smtp.auth(mechanism, method) + try: smtp.auth(mechanism, method, initial_response_ok=False) except smtplib.SMTPAuthenticationError as err: self.assertIn(sim_auth_credentials[mechanism.lower()].upper(), str(err)) @@ -1142,12 +1143,85 @@ class SMTPUTF8SimTests(unittest.TestCase): smtp.send_message(msg)) +EXPECTED_RESPONSE = encode_base64(b'\0psu\0doesnotexist', eol='') + +class SimSMTPAUTHInitialResponseChannel(SimSMTPChannel): + def smtp_AUTH(self, arg): + # RFC 4954's AUTH command allows for an optional initial-response. + # Not all AUTH methods support this; some require a challenge. AUTH + # PLAIN does those, so test that here. See issue #15014. + args = arg.split() + if args[0].lower() == 'plain': + if len(args) == 2: + # AUTH PLAIN <initial-response> with the response base 64 + # encoded. Hard code the expected response for the test. + if args[1] == EXPECTED_RESPONSE: + self.push('235 Ok') + return + self.push('571 Bad authentication') + +class SimSMTPAUTHInitialResponseServer(SimSMTPServer): + channel_class = SimSMTPAUTHInitialResponseChannel + + +@unittest.skipUnless(threading, 'Threading required for this test.') +class SMTPAUTHInitialResponseSimTests(unittest.TestCase): + def setUp(self): + self.real_getfqdn = socket.getfqdn + socket.getfqdn = mock_socket.getfqdn + self.serv_evt = threading.Event() + self.client_evt = threading.Event() + # Pick a random unused port by passing 0 for the port number + self.serv = SimSMTPAUTHInitialResponseServer( + (HOST, 0), ('nowhere', -1), decode_data=True) + # Keep a note of what port was assigned + self.port = self.serv.socket.getsockname()[1] + serv_args = (self.serv, self.serv_evt, self.client_evt) + self.thread = threading.Thread(target=debugging_server, args=serv_args) + self.thread.start() + + # wait until server thread has assigned a port number + self.serv_evt.wait() + self.serv_evt.clear() + + def tearDown(self): + socket.getfqdn = self.real_getfqdn + # indicate that the client is finished + self.client_evt.set() + # wait for the server thread to terminate + self.serv_evt.wait() + self.thread.join() + + def testAUTH_PLAIN_initial_response_login(self): + self.serv.add_feature('AUTH PLAIN') + smtp = smtplib.SMTP(HOST, self.port, + local_hostname='localhost', timeout=15) + smtp.login('psu', 'doesnotexist') + smtp.close() + + def testAUTH_PLAIN_initial_response_auth(self): + self.serv.add_feature('AUTH PLAIN') + smtp = smtplib.SMTP(HOST, self.port, + local_hostname='localhost', timeout=15) + smtp.user = 'psu' + smtp.password = 'doesnotexist' + code, response = smtp.auth('plain', smtp.auth_plain) + smtp.close() + self.assertEqual(code, 235) + + @support.reap_threads def test_main(verbose=None): - support.run_unittest(GeneralTests, DebuggingServerTests, - NonConnectingTests, - BadHELOServerTests, SMTPSimTests, - TooLongLineTests) + support.run_unittest( + BadHELOServerTests, + DebuggingServerTests, + GeneralTests, + NonConnectingTests, + SMTPAUTHInitialResponseSimTests, + SMTPSimTests, + TooLongLineTests, + ) + if __name__ == '__main__': test_main() |