diff -r 3ae2cd85a908 Lib/smtplib.py --- a/Lib/smtplib.py Sun Mar 09 11:18:16 2014 +0100 +++ b/Lib/smtplib.py Thu May 29 00:23:05 2014 +0200 @@ -558,12 +558,57 @@ if not (200 <= code <= 299): raise SMTPHeloError(code, resp) + def auth(self, mechanism, authobject): + """Authentication command - requires response processing. + + 'mechanism' specifies which authentication mechanism is to + be used - it must appear in .capabilities in the + form AUTH=. + + 'authobject' must be a callable object: + + data = authobject(challenge) + + It will be called to process the servers 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. + """ + + 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) + raise SMTPAuthenticationError(code, resp) + + def _CRAM_MD5_AUTH(self, challenge): + """ Authobject to use with CRAM-MD5 authentication. """ + return self.user + " " + hmac.HMAC( + self.password.encode('ascii'), challenge, 'md5').hexdigest() + + def _PLAIN_AUTH(self, challenge): + """ Authobject to use with PLAIN authentication. """ + return "\0%s\0%s" % (self.user, self.password) + + def _LOGIN_AUTH(self, challenge): + """ Authobject to use with LOGIN authentication. """ + (code, resp) = self.docmd( + encode_base64(self.user.encode('ascii'), eol='')) + if code == 334: + return self.password + raise SMTPAuthenticationError(code, resp) + def login(self, user, password): """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. + - user: The user name to authenticate with. + - password: The password for the authentication. If there has been no previous EHLO or HELO command this session, this method tries ESMTP EHLO first. @@ -580,63 +625,52 @@ found. """ - def encode_cram_md5(challenge, user, password): - challenge = base64.decodebytes(challenge) - response = user + " " + hmac.HMAC(password.encode('ascii'), - challenge, 'md5').hexdigest() - return encode_base64(response.encode('ascii'), eol='') - - def encode_plain(user, password): - s = "\0%s\0%s" % (user, password) - return encode_base64(s.encode('ascii'), eol='') - - AUTH_PLAIN = "PLAIN" - AUTH_CRAM_MD5 = "CRAM-MD5" - AUTH_LOGIN = "LOGIN" - self.ehlo_or_helo_if_needed() - if not self.has_extn("auth"): raise SMTPException("SMTP AUTH extension not supported by server.") # Authentication methods the server claims to support advertised_authlist = self.esmtp_features["auth"].split() - # List of authentication methods we support: from preferred to - # less preferred methods. Except for the purpose of testing the weaker - # ones, we prefer stronger methods like CRAM-MD5: - preferred_auths = [AUTH_CRAM_MD5, AUTH_PLAIN, AUTH_LOGIN] + # Authentication methods we can handle in our preferred order: + supported_auths = ['CRAM-MD5', 'PLAIN', 'LOGIN'] - # We try the authentication methods the server advertises, but only the - # ones *we* support. And in our preferred order. - authlist = [auth for auth in preferred_auths if auth in advertised_authlist] + # We try the supported authentication methods the server advertises + # in our preferred order. + authlist = [ + auth for auth in supported_auths if auth in advertised_authlist + ] if not authlist: raise SMTPException("No suitable authentication method found.") # Some servers advertise authentication methods they don't really # support, so if authentication fails, we continue until we've tried # all methods. + self.user, self.password = user, password for authmethod in authlist: - if authmethod == AUTH_CRAM_MD5: - (code, resp) = self.docmd("AUTH", AUTH_CRAM_MD5) - if code == 334: - (code, resp) = self.docmd(encode_cram_md5(resp, user, password)) - elif authmethod == AUTH_PLAIN: - (code, resp) = self.docmd("AUTH", - AUTH_PLAIN + " " + encode_plain(user, password)) - elif authmethod == AUTH_LOGIN: - (code, resp) = self.docmd("AUTH", - "%s %s" % (AUTH_LOGIN, encode_base64(user.encode('ascii'), eol=''))) - if code == 334: - (code, resp) = self.docmd(encode_base64(password.encode('ascii'), eol='')) - - # 235 == 'Authentication successful' - # 503 == 'Error: already authenticated' - if code in (235, 503): - return (code, resp) + try: + if authmethod == 'PLAIN': + (code, resp) = self.auth( + authmethod, self._PLAIN_AUTH) + elif authmethod == 'CRAM-MD5': + (code, resp) = self.auth( + authmethod, self._CRAM_MD5_AUTH) + elif authmethod == 'LOGIN': + (code, resp) = self.auth( + authmethod, self._LOGIN_AUTH) + else: + raise SMTPException( + "Authentication method '%s' is not implemented" + % authmethod) + # 235 == 'Authentication successful' + # 503 == 'Error: already authenticated' + if code in (235, 503): + return (code, resp) + except SMTPAuthenticationError as e: + last_exception = e # We could not login sucessfully. Return result of last attempt. - raise SMTPAuthenticationError(code, resp) + raise last_exception def starttls(self, keyfile=None, certfile=None, context=None): """Puts the connection to the SMTP server into TLS mode. diff -r 3ae2cd85a908 Lib/test/test_smtplib.py --- a/Lib/test/test_smtplib.py Sun Mar 09 11:18:16 2014 +0100 +++ b/Lib/test/test_smtplib.py Thu May 29 00:23:05 2014 +0200 @@ -604,7 +604,8 @@ 'cram-md5': ('TXIUQUBZB21LD2HLCMUUY29TIDG4OWQ0MJ' 'KWZGQ4ODNMNDA4NTGXMDRLZWMYZJDMODG1'), } -sim_auth_login_password = 'C29TZXBHC3N3B3JK' +sim_auth_login_user = 'TXIUQUBZB21LD2HLCMUUY29T' +sim_auth_plain = 'AE1YLKFAC29TZXDOZXJLLMNVBQBZB21LCGFZC3DVCMQ=' sim_lists = {'list-1':['Mr.A@somewhere.com','Mrs.C@somewhereesle.com'], 'list-2':['Ms.B@xn--fo-fka.com',], @@ -657,18 +658,16 @@ self.push('550 No access for you!') def smtp_AUTH(self, arg): - if arg.strip().lower()=='cram-md5': + mech = arg.strip().lower() + if mech=='cram-md5': self.push('334 {}'.format(sim_cram_md5_challenge)) - return - mech, auth = arg.split() - mech = mech.lower() - if mech not in sim_auth_credentials: + elif mech not in sim_auth_credentials: self.push('504 auth type unimplemented') return - if mech == 'plain' and auth==sim_auth_credentials['plain']: - self.push('235 plain auth ok') - elif mech=='login' and auth==sim_auth_credentials['login']: - self.push('334 Password:') + elif mech=='plain': + self.push('334 ') + elif mech=='login': + self.push('334 ') else: self.push('550 No access for you!') @@ -816,9 +815,9 @@ def testAUTH_PLAIN(self): self.serv.add_feature("AUTH PLAIN") smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) - - expected_auth_ok = (235, b'plain auth ok') - self.assertEqual(smtp.login(sim_auth[0], sim_auth[1]), expected_auth_ok) + try: smtp.login(sim_auth[0], sim_auth[1]) + except smtplib.SMTPAuthenticationError as err: + self.assertIn(sim_auth_plain, str(err)) smtp.close() # SimSMTPChannel doesn't fully support LOGIN or CRAM-MD5 auth because they @@ -834,7 +833,7 @@ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) try: smtp.login(sim_auth[0], sim_auth[1]) except smtplib.SMTPAuthenticationError as err: - self.assertIn(sim_auth_login_password, str(err)) + self.assertIn(sim_auth_login_user, str(err)) smtp.close() def testAUTH_CRAM_MD5(self): @@ -852,7 +851,17 @@ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) try: smtp.login(sim_auth[0], sim_auth[1]) except smtplib.SMTPAuthenticationError as err: - self.assertIn(sim_auth_login_password, str(err)) + self.assertIn(sim_auth_login_user, str(err)) + smtp.close() + + def test_auth_function(self): + smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15) + self.serv.add_feature("AUTH CRAM-MD5") + smtp.user, smtp.password = sim_auth[0], sim_auth[1] + + try: smtp.auth('CRAM-MD5', smtp._CRAM_MD5_AUTH) + except smtplib.SMTPAuthenticationError as err: + self.assertIn(sim_auth_credentials['cram-md5'], str(err)) smtp.close() def test_with_statement(self):