Update smptlib and test_smtpnet.py from 3.13.11 (#6435)

* Update `test_smtpnet.py` from 3.13.11

* Update `test_smtplib.py` from 3.13.11

* Update `smtplib.py` from 3.13.11

* Catch AttributeError

Lib/smtplib.py

@@ -171,7 +171,7 @@ def quotedata(data):
     internet CRLF end-of-line.
     """
     return re.sub(r'(?m)^\.', '..',
-                  re.sub(r'(?:\r\n|\n|\r(?!\n))', CRLF, data))
+        re.sub(r'(?:\r\n|\n|\r(?!\n))', CRLF, data))
 
 def _quote_periods(bindata):
     return re.sub(br'(?m)^\.', b'..', bindata)
@@ -179,6 +179,16 @@ def _quote_periods(bindata):
 def _fix_eols(data):
     return  re.sub(r'(?:\r\n|\n|\r(?!\n))', CRLF, data)
 
+
+try:
+    hmac.digest(b'', b'', 'md5')
+# except ValueError:
+except (ValueError, AttributeError): # TODO: RUSTPYTHON
+    _have_cram_md5_support = False
+else:
+    _have_cram_md5_support = True
+
+
 try:
     import ssl
 except ImportError:
@@ -475,7 +485,7 @@ class SMTP:
             if auth_match:
                 # This doesn't remove duplicates, but that's no problem
                 self.esmtp_features["auth"] = self.esmtp_features.get("auth", "") \
-                                              + " " + auth_match.groups(0)[0]
+                        + " " + auth_match.groups(0)[0]
                 continue
 
             # RFC 1869 requires a space between ehlo keyword and parameters.
@@ -488,7 +498,7 @@ class SMTP:
                 params = m.string[m.end("feature"):].strip()
                 if feature == "auth":
                     self.esmtp_features[feature] = self.esmtp_features.get(feature, "") \
-                                                   + " " + params
+                            + " " + params
                 else:
                     self.esmtp_features[feature] = params
         return (code, msg)
@@ -542,7 +552,7 @@ class SMTP:
                     raise SMTPNotSupportedError(
                         'SMTPUTF8 not supported by server')
             optionlist = ' ' + ' '.join(options)
-        self.putcmd("mail", "FROM:%s%s" % (quoteaddr(sender), optionlist))
+        self.putcmd("mail", "from:%s%s" % (quoteaddr(sender), optionlist))
         return self.getreply()
 
     def rcpt(self, recip, options=()):
@@ -550,7 +560,7 @@ class SMTP:
         optionlist = ''
         if options and self.does_esmtp:
             optionlist = ' ' + ' '.join(options)
-        self.putcmd("rcpt", "TO:%s%s" % (quoteaddr(recip), optionlist))
+        self.putcmd("rcpt", "to:%s%s" % (quoteaddr(recip), optionlist))
         return self.getreply()
 
     def data(self, msg):
@@ -667,8 +677,11 @@ class SMTP:
         # 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()
+        if not _have_cram_md5_support:
+            raise SMTPException("CRAM-MD5 is not supported")
+        password = self.password.encode('ascii')
+        authcode = hmac.HMAC(password, challenge, 'md5')
+        return f"{self.user} {authcode.hexdigest()}"
 
     def auth_plain(self, challenge=None):
         """ Authobject to use with PLAIN authentication. Requires self.user and
@@ -720,8 +733,10 @@ class SMTP:
         advertised_authlist = self.esmtp_features["auth"].split()
 
         # Authentication methods we can handle in our preferred order:
-        preferred_auths = ['CRAM-MD5', 'PLAIN', 'LOGIN']
-
+        if _have_cram_md5_support:
+            preferred_auths = ['CRAM-MD5', 'PLAIN', 'LOGIN']
+        else:
+            preferred_auths = ['PLAIN', 'LOGIN']
         # We try the supported authentications in our preferred order, if
         # the server supports them.
         authlist = [auth for auth in preferred_auths
@@ -905,7 +920,7 @@ class SMTP:
         The arguments are as for sendmail, except that msg is an
         email.message.Message object.  If from_addr is None or to_addrs is
         None, these arguments are taken from the headers of the Message as
-        described in RFC 2822 (a ValueError is raised if there is more than
+        described in RFC 5322 (a ValueError is raised if there is more than
         one set of 'Resent-' headers).  Regardless of the values of from_addr and
         to_addr, any Bcc field (or Resent-Bcc field, when the Message is a
         resent) of the Message object won't be transmitted.  The Message
@@ -919,7 +934,7 @@ class SMTP:
         policy.
 
         """
-        # 'Resent-Date' is a mandatory field if the Message is resent (RFC 2822
+        # 'Resent-Date' is a mandatory field if the Message is resent (RFC 5322
         # Section 3.6.6). In such a case, we use the 'Resent-*' fields.  However,
         # if there is more than one 'Resent-' block there's no way to
         # unambiguously determine which one is the most recent in all cases,
@@ -938,10 +953,10 @@ class SMTP:
         else:
             raise ValueError("message has more than one 'Resent-' header block")
         if from_addr is None:
-            # Prefer the sender field per RFC 2822:3.6.2.
+            # Prefer the sender field per RFC 5322 section 3.6.2.
             from_addr = (msg[header_prefix + 'Sender']
-                         if (header_prefix + 'Sender') in msg
-                         else msg[header_prefix + 'From'])
+                           if (header_prefix + 'Sender') in msg
+                           else msg[header_prefix + 'From'])
             from_addr = email.utils.getaddresses([from_addr])[0][1]
         if to_addrs is None:
             addr_fields = [f for f in (msg[header_prefix + 'To'],

Lib/test/test_smtplib.py

@@ -17,6 +17,7 @@ import textwrap
 import threading
 
 import unittest
+import unittest.mock as mock
 from test import support, mock_socket
 from test.support import hashlib_helper
 from test.support import socket_helper
@@ -350,7 +351,7 @@ class DebuggingServerTests(unittest.TestCase):
                             timeout=support.LOOPBACK_TIMEOUT)
         self.addCleanup(smtp.close)
         expected = (252, b'Cannot VRFY user, but will accept message ' + \
-                    b'and attempt delivery')
+                         b'and attempt delivery')
         self.assertEqual(smtp.vrfy('nobody@nowhere.com'), expected)
         self.assertEqual(smtp.verify('nobody@nowhere.com'), expected)
         smtp.quit()
@@ -371,7 +372,7 @@ class DebuggingServerTests(unittest.TestCase):
                             timeout=support.LOOPBACK_TIMEOUT)
         self.addCleanup(smtp.close)
         self.assertEqual(smtp.help(), b'Supported commands: EHLO HELO MAIL ' + \
-                         b'RCPT DATA RSET NOOP QUIT VRFY')
+                                      b'RCPT DATA RSET NOOP QUIT VRFY')
         smtp.quit()
 
     def testSend(self):
@@ -527,7 +528,7 @@ class DebuggingServerTests(unittest.TestCase):
         smtp.quit()
         # make sure the Bcc header is still in the message.
         self.assertEqual(m['Bcc'], 'John Root <root@localhost>, "Dinsdale" '
-                                   '<warped@silly.walks.com>')
+                                    '<warped@silly.walks.com>')
 
         self.client_evt.set()
         self.serv_evt.wait()
@@ -766,7 +767,7 @@ class BadHELOServerTests(unittest.TestCase):
 
     def testFailingHELO(self):
         self.assertRaises(smtplib.SMTPConnectError, smtplib.SMTP,
-                          HOST, self.port, 'localhost', 3)
+                            HOST, self.port, 'localhost', 3)
 
 
 class TooLongLineTests(unittest.TestCase):
@@ -804,14 +805,14 @@ class TooLongLineTests(unittest.TestCase):
 sim_users = {'Mr.A@somewhere.com':'John A',
              'Ms.B@xn--fo-fka.com':'Sally B',
              'Mrs.C@somewhereesle.com':'Ruth C',
-             }
+            }
 
 sim_auth = ('Mr.A@somewhere.com', 'somepassword')
 sim_cram_md5_challenge = ('PENCeUxFREJoU0NnbmhNWitOMjNGNn'
                           'dAZWx3b29kLmlubm9zb2Z0LmNvbT4=')
 sim_lists = {'list-1':['Mr.A@somewhere.com','Mrs.C@somewhereesle.com'],
              'list-2':['Ms.B@xn--fo-fka.com',],
-             }
+            }
 
 # Simulated SMTP channel & server
 class ResponseException(Exception): pass
@@ -830,6 +831,7 @@ class SimSMTPChannel(smtpd.SMTPChannel):
     def __init__(self, extra_features, *args, **kw):
         self._extrafeatures = ''.join(
             [ "250-{0}\r\n".format(x) for x in extra_features ])
+        self.all_received_lines = []
         super(SimSMTPChannel, self).__init__(*args, **kw)
 
     # AUTH related stuff.  It would be nice if support for this were in smtpd.
@@ -844,6 +846,7 @@ class SimSMTPChannel(smtpd.SMTPChannel):
                 self.smtp_state = self.COMMAND
                 self.push('%s %s' % (e.smtp_code, e.smtp_error))
             return
+        self.all_received_lines.append(self.received_lines)
         super().found_terminator()
 
 
@@ -924,11 +927,14 @@ class SimSMTPChannel(smtpd.SMTPChannel):
             except ValueError as e:
                 self.push('535 Splitting response {!r} into user and password '
                           'failed: {}'.format(logpass, e))
-                return False
-            valid_hashed_pass = hmac.HMAC(
-                sim_auth[1].encode('ascii'),
-                self._decode_base64(sim_cram_md5_challenge).encode('ascii'),
-                'md5').hexdigest()
+                return
+            pwd = sim_auth[1].encode('ascii')
+            msg = self._decode_base64(sim_cram_md5_challenge).encode('ascii')
+            try:
+                valid_hashed_pass = hmac.HMAC(pwd, msg, 'md5').hexdigest()
+            except ValueError:
+                self.push('504 CRAM-MD5 is not supported')
+                return
             self._authenticated(user, hashed_pass == valid_hashed_pass)
     # end AUTH related stuff.
 
@@ -1170,8 +1176,7 @@ class SMTPSimTests(unittest.TestCase):
         finally:
             smtp.close()
 
-    # TODO: RUSTPYTHON
-    @unittest.expectedFailure
+    @unittest.expectedFailure # TODO: RUSTPYTHON
     @hashlib_helper.requires_hashdigest('md5', openssl=True)
     def testAUTH_CRAM_MD5(self):
         self.serv.add_feature("AUTH CRAM-MD5")
@@ -1181,8 +1186,39 @@ class SMTPSimTests(unittest.TestCase):
         self.assertEqual(resp, (235, b'Authentication Succeeded'))
         smtp.close()
 
-    # TODO: RUSTPYTHON
-    @unittest.expectedFailure
+    @mock.patch("hmac.HMAC")
+    @mock.patch("smtplib._have_cram_md5_support", False)
+    def testAUTH_CRAM_MD5_blocked(self, hmac_constructor):
+        # CRAM-MD5 is the only "known" method by the server,
+        # but it is not supported by the client. In particular,
+        # no challenge will ever be sent.
+        self.serv.add_feature("AUTH CRAM-MD5")
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
+        self.addCleanup(smtp.close)
+        msg = re.escape("No suitable authentication method found.")
+        with self.assertRaisesRegex(smtplib.SMTPException, msg):
+            smtp.login(sim_auth[0], sim_auth[1])
+        hmac_constructor.assert_not_called()  # call has been bypassed
+
+    @mock.patch("smtplib._have_cram_md5_support", False)
+    def testAUTH_CRAM_MD5_blocked_and_fallback(self):
+        # Test that PLAIN is tried after CRAM-MD5 failed
+        self.serv.add_feature("AUTH CRAM-MD5 PLAIN")
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
+        self.addCleanup(smtp.close)
+        with (
+            mock.patch.object(smtp, "auth_cram_md5") as smtp_auth_cram_md5,
+            mock.patch.object(
+                smtp, "auth_plain", wraps=smtp.auth_plain
+            ) as smtp_auth_plain
+        ):
+            resp = smtp.login(sim_auth[0], sim_auth[1])
+        smtp_auth_plain.assert_called_once()
+        smtp_auth_cram_md5.assert_not_called()  # no call to HMAC constructor
+        self.assertEqual(resp, (235, b'Authentication Succeeded'))
+
     @hashlib_helper.requires_hashdigest('md5', openssl=True)
     def testAUTH_multiple(self):
         # Test that multiple authentication methods are tried.
@@ -1193,8 +1229,7 @@ class SMTPSimTests(unittest.TestCase):
         self.assertEqual(resp, (235, b'Authentication Succeeded'))
         smtp.close()
 
-    # TODO: RUSTPYTHON
-    @unittest.expectedFailure
+    @unittest.expectedFailure # TODO: RUSTPYTHON
     def test_auth_function(self):
         supported = {'PLAIN', 'LOGIN'}
         try:
@@ -1354,6 +1389,18 @@ class SMTPSimTests(unittest.TestCase):
         self.assertEqual(self.serv._addresses['from'], 'michael@example.com')
         self.assertEqual(self.serv._addresses['tos'], ['rene@example.com'])
 
+    def test_lowercase_mail_from_rcpt_to(self):
+        m = 'A test message'
+        smtp = smtplib.SMTP(
+            HOST, self.port, local_hostname='localhost',
+            timeout=support.LOOPBACK_TIMEOUT)
+        self.addCleanup(smtp.close)
+
+        smtp.sendmail('John', 'Sally', m)
+
+        self.assertIn(['mail from:<John> size=14'], self.serv._SMTPchannel.all_received_lines)
+        self.assertIn(['rcpt to:<Sally>'], self.serv._SMTPchannel.all_received_lines)
+
 
 class SimSMTPUTF8Server(SimSMTPServer):
 
@@ -1372,7 +1419,7 @@ class SimSMTPUTF8Server(SimSMTPServer):
         )
 
     def process_message(self, peer, mailfrom, rcpttos, data, mail_options=None,
-                        rcpt_options=None):
+                                                             rcpt_options=None):
         self.last_peer = peer
         self.last_mailfrom = mailfrom
         self.last_rcpttos = rcpttos

Lib/test/test_smtpnet.py

@@ -2,6 +2,7 @@ import unittest
 from test import support
 from test.support import import_helper
 from test.support import socket_helper
+import os
 import smtplib
 import socket
 
@@ -9,6 +10,8 @@ ssl = import_helper.import_module("ssl")
 
 support.requires("network")
 
+SMTP_TEST_SERVER = os.getenv('CPYTHON_TEST_SMTP_SERVER', 'smtp.gmail.com')
+
 def check_ssl_verifiy(host, port):
     context = ssl.create_default_context()
     with socket.create_connection((host, port)) as sock:
@@ -22,7 +25,7 @@ def check_ssl_verifiy(host, port):
 
 
 class SmtpTest(unittest.TestCase):
-    testServer = 'smtp.gmail.com'
+    testServer = SMTP_TEST_SERVER
     remotePort = 587
 
     def test_connect_starttls(self):
@@ -44,7 +47,7 @@ class SmtpTest(unittest.TestCase):
 
 
 class SmtpSSLTest(unittest.TestCase):
-    testServer = 'smtp.gmail.com'
+    testServer = SMTP_TEST_SERVER
     remotePort = 465
 
     def test_connect(self):
@@ -87,4 +90,4 @@ class SmtpSSLTest(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()