- Issue #15014: SMTP.auth() and SMTP.login() now support RFC 4954's optional
initial-response argument to the SMTP AUTH command.
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
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 @@
'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 @@
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()