Fix ordering of context.check_hostname and context.verify_mode so as to not raise ValueError if cert_reqs=ssl.CERT_NONE.

4 files changed, 66 insertions, 4 deletions

diff --git a/AUTHORS b/AUTHORS
index d6628d4..cc03f59 100644
--- a/AUTHORS
+++ b/AUTHORS
@@ -24,3 +24,4 @@ ChangBo Guo(gcb) <eric.guo@easystack.cn>
 Alan Justino <alan.justino@yahoo.com.br>
 Jelte Fennema <github-tech@jeltef.nl>
 Jon Dufresne <jon.dufresne@gmail.com>
+Colton Hicks <github@coltonhicks.com>
diff --git a/Changelog b/Changelog
index 9d64cc4..148a880 100644
--- a/Changelog
+++ b/Changelog
@@ -5,6 +5,17 @@ py-amqp is fork of amqplib used by Kombu containing additional features and impr
 The previous amqplib changelog is here:
 http://code.google.com/p/py-amqplib/source/browse/CHANGES
 
+.. _version-5.0.6:
+
+5.0.6
+=====
+- Change the order in which context.check_hostname and context.verify_mode get set
+  in SSLTransport._wrap_socket_sni. Fixes bug introduced in 5.0.3 where setting
+  context.verify_mode = ssl.CERT_NONE would raise
+  "ValueError: Cannot set verify_mode to CERT_NONE when check_hostname is enabled."
+  Setting context.check_hostname prior to setting context.verify_mode resolves the
+  issue.
+
 .. _version-5.0.5:
 
 5.0.5
diff --git a/amqp/transport.py b/amqp/transport.py
index 4130681..7056be8 100644
--- a/amqp/transport.py
+++ b/amqp/transport.py
@@ -525,9 +525,13 @@ class SSLTransport(_AbstractTransport):
             context.load_verify_locations(ca_certs)
         if ciphers is not None:
             context.set_ciphers(ciphers)
-        if cert_reqs is not None:
-            context.verify_mode = cert_reqs
-        # Set SNI headers if supported
+        # Set SNI headers if supported. 
+        # Must set context.check_hostname before setting context.verify_mode 
+        # to avoid setting context.verify_mode=ssl.CERT_NONE while
+        # context.check_hostname is still True (the default value in context
+        # if client-side) which results in the following exception:
+        # ValueError: Cannot set verify_mode to CERT_NONE when check_hostname
+        # is enabled.
         try:
             context.check_hostname = (
                 ssl.HAS_SNI and server_hostname is not None
@@ -535,6 +539,11 @@ class SSLTransport(_AbstractTransport):
         except AttributeError:
             pass  # ask forgiveness not permission
 
+        # See note above re: ordering for context.check_hostname and 
+        # context.verify_mode assignments.
+        if cert_reqs is not None:
+            context.verify_mode = cert_reqs
+
         if ca_certs is None and context.verify_mode != ssl.CERT_NONE:
             purpose = (
                 ssl.Purpose.CLIENT_AUTH
diff --git a/t/unit/test_transport.py b/t/unit/test_transport.py
index f217fb6..d93116a 100644
--- a/t/unit/test_transport.py
+++ b/t/unit/test_transport.py
@@ -700,7 +700,6 @@ class test_SSLTransport:
             set_ciphers_method_mock.assert_called_with(sentinel.CIPHERS)
 
     def test_wrap_socket_sni_cert_reqs(self):
-        # testing _wrap_socket_sni() with parameter cert_reqs == ssl.CERT_NONE
         with patch('ssl.SSLContext') as mock_ssl_context_class:
             sock = Mock()
             context = mock_ssl_context_class()
@@ -720,6 +719,48 @@ class test_SSLTransport:
             )
             assert context.verify_mode == sentinel.CERT_REQS
 
+        # testing context creation inside _wrap_socket_sni() with parameter 
+        # cert_reqs == ssl.CERT_NONE. Previously raised ValueError because
+        # code path attempted to set context.verify_mode=ssl.CERT_NONE before
+        # setting context.check_hostname = False which raised a ValueError
+        with patch('ssl.SSLContext.wrap_socket') as mock_wrap_socket:
+            with patch('ssl.SSLContext.load_default_certs') as mock_load_default_certs:
+                sock = Mock()
+                self.t._wrap_socket_sni(
+                    sock, server_side=True, cert_reqs=ssl.CERT_NONE
+                )
+                mock_load_default_certs.assert_not_called()
+                mock_wrap_socket.assert_called_once()
+
+        with patch('ssl.SSLContext.wrap_socket') as mock_wrap_socket:
+            with patch('ssl.SSLContext.load_default_certs') as mock_load_default_certs:
+                sock = Mock()
+                self.t._wrap_socket_sni(
+                    sock, server_side=False, cert_reqs=ssl.CERT_NONE
+                )
+                mock_load_default_certs.assert_not_called()
+                mock_wrap_socket.assert_called_once()
+        
+        with patch('ssl.SSLContext.wrap_socket') as mock_wrap_socket:
+            with patch('ssl.SSLContext.load_default_certs') as mock_load_default_certs:
+                sock = Mock()
+                self.t._wrap_socket_sni(
+                    sock, server_side=True, cert_reqs=ssl.CERT_REQUIRED
+                )
+                mock_load_default_certs.assert_called_with(ssl.Purpose.CLIENT_AUTH)
+                mock_wrap_socket.assert_called_once()
+
+        with patch('ssl.SSLContext.wrap_socket') as mock_wrap_socket:
+            with patch('ssl.SSLContext.load_default_certs') as mock_load_default_certs:
+                sock = Mock()
+                self.t._wrap_socket_sni(
+                    sock, server_side=False, cert_reqs=ssl.CERT_REQUIRED
+                )
+                mock_load_default_certs.assert_called_once_with(
+                    ssl.Purpose.SERVER_AUTH
+                )
+                mock_wrap_socket.assert_called_once()
+
     def test_wrap_socket_sni_setting_sni_header(self):
         # testing _wrap_socket_sni() without parameter server_hostname