connection: require cert validation for hostname checks

Author: dkropachev

cassandra/connection.py

@@ -56,7 +56,6 @@
 
 log = logging.getLogger(__name__)
 
-
 segment_codec_no_compression = SegmentCodec()
 segment_codec_lz4 = None
 
@@ -1208,14 +1207,17 @@ def _build_ssl_context_from_options(self):
         # being explicit
         ssl_version = opts.get('ssl_version', None) or ssl.PROTOCOL_TLS_CLIENT
         cert_reqs = opts.get('cert_reqs', None)
+        check_hostname = bool(opts.get('check_hostname', False))
         if cert_reqs is None:
             cert_reqs = (ssl.CERT_REQUIRED
                          if self.ssl_options
                          else ssl.CERT_NONE)
+        elif check_hostname and cert_reqs == ssl.CERT_NONE:
+            cert_reqs = ssl.CERT_REQUIRED
         rv = ssl.SSLContext(protocol=int(ssl_version))
         rv.check_hostname = False
         rv.verify_mode = cert_reqs
-        rv.check_hostname = bool(opts.get('check_hostname', False))
+        rv.check_hostname = check_hostname
 
         certfile = opts.get('certfile', None)
         keyfile = opts.get('keyfile', None)

tests/unit/test_cloud.py

@@ -1,4 +1,4 @@
-# Copyright DataStax, Inc.
+# Copyright 2026 ScyllaDB, Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

tests/unit/test_connection.py

@@ -257,6 +257,16 @@ def test_ssl_options_check_hostname_requires_validation(self):
         assert c.ssl_context.check_hostname
         assert c._check_hostname
 
+    def test_ssl_options_check_hostname_promotes_cert_none_to_cert_required(self):
+        c = Connection(
+            DefaultEndPoint('1.2.3.4'),
+            ssl_options={'cert_reqs': ssl.CERT_NONE, 'check_hostname': True})
+
+        assert isinstance(c.ssl_context, ssl.SSLContext)
+        assert c.ssl_context.verify_mode == ssl.CERT_REQUIRED
+        assert c.ssl_context.check_hostname
+        assert c._check_hostname
+
     def test_bad_protocol_version(self, *args):
         c = self.make_connection()
         c._requests = Mock()