ssl: move pyopenssl hostname validation to follow-up

dkropachev · dkropachev · commit f8c3ca44584c · 2026-07-21T16:09:58.000-04:00
diff --git a/cassandra/connection.py b/cassandra/connection.py
@@ -18,7 +18,6 @@
 from functools import wraps, partial, total_ordering
 from heapq import heappush, heappop
 import io
-import ipaddress
 import logging
 import socket
 import struct
@@ -57,103 +56,6 @@
 log = logging.getLogger(__name__)
 
 
-def _pyopenssl_cert_to_ssl_cert(peer_cert):
-    from cryptography import x509
-    from cryptography.x509.oid import NameOID
-
-    cert = peer_cert.to_cryptography()
-    ssl_cert = {}
-
-    subject = tuple((('commonName', attr.value),)
-                    for attr in cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME))
-    if subject:
-        ssl_cert['subject'] = subject
-
-    subject_alt_names = []
-    try:
-        san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value
-    except x509.ExtensionNotFound:
-        pass
-    else:
-        subject_alt_names.extend(('DNS', name) for name in san.get_values_for_type(x509.DNSName))
-        subject_alt_names.extend(('IP Address', str(address))
-                                 for address in san.get_values_for_type(x509.IPAddress))
-
-    if subject_alt_names:
-        ssl_cert['subjectAltName'] = tuple(subject_alt_names)
-
-    return ssl_cert
-
-
-def _validate_pyopenssl_hostname(peer_cert, hostname):
-    cert = _pyopenssl_cert_to_ssl_cert(peer_cert)
-    try:
-        host_ip = ipaddress.ip_address(hostname)
-    except ValueError:
-        host_ip = None
-
-    names = []
-    for key, value in cert.get('subjectAltName', ()):
-        if key == 'DNS':
-            if host_ip is None and _dnsname_match(value, hostname):
-                return
-            names.append(value)
-        elif key == 'IP Address':
-            if host_ip is not None and _ipaddress_match(value, host_ip):
-                return
-            names.append(value)
-
-    if not names:
-        for subject in cert.get('subject', ()):
-            for key, value in subject:
-                if key == 'commonName':
-                    if _dnsname_match(value, hostname):
-                        return
-                    names.append(value)
-
-    if len(names) > 1:
-        raise ssl.CertificateError("hostname {!r} doesn't match either of {}".format(
-            hostname, ', '.join(map(repr, names))))
-    if len(names) == 1:
-        raise ssl.CertificateError("hostname {!r} doesn't match {!r}".format(hostname, names[0]))
-    raise ssl.CertificateError("no appropriate commonName or subjectAltName fields were found")
-
-
-def _dnsname_match(pattern, hostname):
-    if not pattern:
-        return False
-
-    wildcards = pattern.count('*')
-    if not wildcards:
-        return pattern.lower() == hostname.lower()
-    if wildcards > 1:
-        raise ssl.CertificateError(
-            "too many wildcards in certificate DNS name: {!r}.".format(pattern))
-
-    leftmost, sep, remainder = pattern.partition('.')
-    if '*' in remainder:
-        raise ssl.CertificateError(
-            "wildcard can only be present in the leftmost label: {!r}.".format(pattern))
-    if not sep:
-        raise ssl.CertificateError(
-            "sole wildcard without additional labels is not supported: {!r}.".format(pattern))
-    if leftmost != '*':
-        raise ssl.CertificateError(
-            "partial wildcards in leftmost label are not supported: {!r}.".format(pattern))
-
-    hostname_leftmost, sep, hostname_remainder = hostname.partition('.')
-    if not hostname_leftmost or not sep:
-        return False
-    return remainder.lower() == hostname_remainder.lower()
-
-
-def _ipaddress_match(pattern, host_ip):
-    try:
-        return ipaddress.ip_address(pattern) == host_ip
-    except ValueError:
-        return False
-
-
 segment_codec_no_compression = SegmentCodec()
 segment_codec_lz4 = None
 
diff --git a/cassandra/io/eventletreactor.py b/cassandra/io/eventletreactor.py
@@ -23,8 +23,7 @@
 from threading import Event
 import time
 
-from cassandra.connection import (Connection, ConnectionShutdown, Timer, TimerManager,
-                                  _validate_pyopenssl_hostname)
+from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager
 try:
     from eventlet.green.OpenSSL import SSL
     _PYOPENSSL = True
@@ -153,8 +152,10 @@ def _validate_hostname(self):
         if self.uses_legacy_ssl_options:
             super(EventletConnection, self)._validate_hostname()
         else:
-            expected_name = (self.ssl_options or {}).get('server_hostname') or self.endpoint.address
-            _validate_pyopenssl_hostname(self._socket.get_peer_certificate(), expected_name)
+            cert_name = self._socket.get_peer_certificate().get_subject().commonName
+            if cert_name != self.endpoint.address:
+                raise Exception("Hostname verification failed! Certificate name '{}' "
+                                "doesn't endpoint '{}'".format(cert_name, self.endpoint.address))
 
     def _build_ssl_context_from_options(self):
         _check_pyopenssl()
diff --git a/cassandra/io/twistedreactor.py b/cassandra/io/twistedreactor.py
@@ -28,8 +28,7 @@
 from twisted.python.failure import Failure
 from zope.interface import implementer
 
-from cassandra.connection import (Connection, ConnectionShutdown, Timer, TimerManager, ConnectionException,
-                                  _validate_pyopenssl_hostname)
+from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager, ConnectionException
 
 try:
     from OpenSSL import SSL
@@ -193,17 +192,11 @@ def __init__(self, endpoint, ssl_context, ssl_options, check_hostname, timeout):
     def verify_callback(self, connection, x509, errnum, errdepth, ok):
         return ok
 
-    def _hostname_to_verify(self):
-        return self.ssl_options.get("server_hostname") or self.endpoint.address
-
     def info_callback(self, connection, where, ret):
         if where & SSL.SSL_CB_HANDSHAKE_DONE:
-            if self.check_hostname:
-                try:
-                    _validate_pyopenssl_hostname(connection.get_peer_certificate(), self._hostname_to_verify())
-                except Exception as exc:
-                    transport = connection.get_app_data()
-                    transport.failVerification(Failure(ConnectionException(str(exc), self.endpoint)))
+            if self.check_hostname and self.endpoint.address != connection.get_peer_certificate().get_subject().commonName:
+                transport = connection.get_app_data()
+                transport.failVerification(Failure(ConnectionException("Hostname verification failed", self.endpoint)))
 
     def clientConnectionForTLS(self, tlsProtocol):
         connection = SSL.Connection(self.context, None)
diff --git a/tests/unit/io/test_eventletreactor.py b/tests/unit/io/test_eventletreactor.py
@@ -13,22 +13,11 @@
 # limitations under the License.
 
 import os
-import ssl
 import unittest
-from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 
 from unittest.mock import Mock, patch
 
-try:
-    from cryptography import x509
-    from cryptography.hazmat.primitives import hashes
-    from cryptography.hazmat.primitives.asymmetric import rsa
-    from cryptography.x509.oid import NameOID
-    from OpenSSL import crypto
-except ImportError:
-    crypto = None
-
 from tests.unit.io.utils import TimerTestMixin
 from tests import notpypy, EVENT_LOOP_MANAGER
 
@@ -47,27 +36,6 @@
     os.path.dirname(__file__), '..', '..', 'integration', 'long', 'ssl', 'rootCa.crt'))
 
 
-def _make_certificate(common_name, san_dns_names=None):
-    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
-    subject = issuer = x509.Name([
-        x509.NameAttribute(NameOID.COMMON_NAME, common_name),
-    ])
-    now = datetime.now(timezone.utc)
-    builder = (x509.CertificateBuilder()
-               .subject_name(subject)
-               .issuer_name(issuer)
-               .public_key(key.public_key())
-               .serial_number(x509.random_serial_number())
-               .not_valid_before(now - timedelta(days=1))
-               .not_valid_after(now + timedelta(days=1)))
-    if san_dns_names:
-        builder = builder.add_extension(
-            x509.SubjectAlternativeName([x509.DNSName(name) for name in san_dns_names]),
-            critical=False)
-
-    return crypto.X509.from_cryptography(builder.sign(key, hashes.SHA256()))
-
-
 @unittest.skipIf(EventletConnection is None, "eventlet is not available")
 @unittest.skipIf(not getattr(eventletreactor, '_PYOPENSSL', False), "pyOpenSSL is not available")
 class EventletSSLContextTest(unittest.TestCase):
@@ -120,48 +88,6 @@ def test_check_hostname_option_enables_hostname_validation(self):
 
         assert conn._check_hostname
 
-    def test_validate_hostname_rejects_mismatch(self):
-        conn = EventletConnection.__new__(EventletConnection)
-        conn.uses_legacy_ssl_options = False
-        conn.endpoint = DefaultEndPoint('1.2.3.4')
-        conn.ssl_options = {}
-        conn._socket = Mock()
-        conn._socket.get_peer_certificate.return_value = _make_certificate('wrong.host')
-
-        with self.assertRaises(ssl.CertificateError):
-            conn._validate_hostname()
-
-    def test_validate_hostname_uses_server_hostname(self):
-        conn = EventletConnection.__new__(EventletConnection)
-        conn.uses_legacy_ssl_options = False
-        conn.endpoint = DefaultEndPoint('proxy.host')
-        conn.ssl_options = {'server_hostname': 'sni.host'}
-        conn._socket = Mock()
-        conn._socket.get_peer_certificate.return_value = _make_certificate('proxy.host', ['sni.host'])
-
-        conn._validate_hostname()
-
-    def test_validate_hostname_prefers_san_over_common_name(self):
-        conn = EventletConnection.__new__(EventletConnection)
-        conn.uses_legacy_ssl_options = False
-        conn.endpoint = DefaultEndPoint('sni.host')
-        conn.ssl_options = {}
-        conn._socket = Mock()
-        conn._socket.get_peer_certificate.return_value = _make_certificate('sni.host', ['other.host'])
-
-        with self.assertRaises(ssl.CertificateError):
-            conn._validate_hostname()
-
-    def test_validate_hostname_matches_wildcard_san(self):
-        conn = EventletConnection.__new__(EventletConnection)
-        conn.uses_legacy_ssl_options = False
-        conn.endpoint = DefaultEndPoint('node.example.com')
-        conn.ssl_options = {}
-        conn._socket = Mock()
-        conn._socket.get_peer_certificate.return_value = _make_certificate('other.host', ['*.example.com'])
-
-        conn._validate_hostname()
-
     def test_wrap_socket_from_context_returns_wrapped_socket(self):
         conn = EventletConnection.__new__(EventletConnection)
         conn.ssl_context = object()
diff --git a/tests/unit/io/test_twistedreactor.py b/tests/unit/io/test_twistedreactor.py
@@ -14,19 +14,9 @@
 
 import os
 import unittest
-from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 from unittest.mock import Mock, patch
 
-try:
-    from cryptography import x509
-    from cryptography.hazmat.primitives import hashes
-    from cryptography.hazmat.primitives.asymmetric import rsa
-    from cryptography.x509.oid import NameOID
-    from OpenSSL import crypto
-except ImportError:
-    crypto = None
-
 from cassandra.connection import Connection, DefaultEndPoint
 
 try:
@@ -45,27 +35,6 @@
     os.path.dirname(__file__), '..', '..', 'integration', 'long', 'ssl', 'rootCa.crt'))
 
 
-def _make_certificate(common_name, san_dns_names=None):
-    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
-    subject = issuer = x509.Name([
-        x509.NameAttribute(NameOID.COMMON_NAME, common_name),
-    ])
-    now = datetime.now(timezone.utc)
-    builder = (x509.CertificateBuilder()
-               .subject_name(subject)
-               .issuer_name(issuer)
-               .public_key(key.public_key())
-               .serial_number(x509.random_serial_number())
-               .not_valid_before(now - timedelta(days=1))
-               .not_valid_after(now + timedelta(days=1)))
-    if san_dns_names:
-        builder = builder.add_extension(
-            x509.SubjectAlternativeName([x509.DNSName(name) for name in san_dns_names]),
-            critical=False)
-
-    return crypto.X509.from_cryptography(builder.sign(key, hashes.SHA256()))
-
-
 @unittest.skipIf(TwistedConnection is None, "Twisted libraries are not available")
 @unittest.skipIf(not getattr(twistedreactor, '_HAS_SSL', False), "pyOpenSSL is not available")
 class TwistedSSLContextTest(unittest.TestCase):
@@ -108,39 +77,6 @@ def test_check_hostname_option_enables_hostname_validation(self):
 
         assert conn._check_hostname
 
-    def test_hostname_verification_uses_server_hostname(self):
-        context = Mock()
-        creator = twistedreactor._SSLCreator(DefaultEndPoint('proxy.host'), context,
-                                             {'server_hostname': 'sni.host'}, True, None)
-        connection = Mock()
-        connection.get_peer_certificate.return_value = _make_certificate('proxy.host', ['sni.host'])
-
-        creator.info_callback(connection, twistedreactor.SSL.SSL_CB_HANDSHAKE_DONE, None)
-
-        connection.get_app_data.assert_not_called()
-
-    def test_hostname_verification_prefers_san_over_common_name(self):
-        context = Mock()
-        creator = twistedreactor._SSLCreator(DefaultEndPoint('sni.host'), context, {}, True, None)
-        connection = Mock()
-        transport = Mock()
-        connection.get_app_data.return_value = transport
-        connection.get_peer_certificate.return_value = _make_certificate('sni.host', ['other.host'])
-
-        creator.info_callback(connection, twistedreactor.SSL.SSL_CB_HANDSHAKE_DONE, None)
-
-        transport.failVerification.assert_called_once()
-
-    def test_hostname_verification_matches_wildcard_san(self):
-        context = Mock()
-        creator = twistedreactor._SSLCreator(DefaultEndPoint('node.example.com'), context, {}, True, None)
-        connection = Mock()
-        connection.get_peer_certificate.return_value = _make_certificate('other.host', ['*.example.com'])
-
-        creator.info_callback(connection, twistedreactor.SSL.SSL_CB_HANDSHAKE_DONE, None)
-
-        connection.get_app_data.assert_not_called()
-
 
 class TestTwistedTimer(TimerTestMixin, unittest.TestCase):
     """
