BUG#37055435: Connection fails during the TLS negotiation when specifying TLSv1.3 ciphers

With this patch, the SSL/TLS setup stage on the
connector side no longer fails for Pure-Python
(including the asynchronous one) and C-extension
implementations when specifying TLSv1.3 ciphers
as part of the connection options.

Change-Id: Iecaa52082010d85dee0948eed844da9c57e9abd5

Author: oscpache

CHANGES.txt

@@ -18,6 +18,7 @@ v9.1.0
 - WL#16341: OpenID Connect (Oauth2 - JWT) Authentication Support
 - WL#16307: Remove Python 3.8 support
 - WL#16306: Add support for Python 3.13
+- BUG#37055435: Connection fails during the TLS negotiation when specifying TLSv1.3 ciphers
 - BUG#37013057: mysql-connector-python Parameterized query SQL injection
 - BUG#36765200: python mysql connector 8.3.0 raise %-.100s:%u when input a wrong host
 - BUG#36577957: Update charset/collation description indicate this is 16 bits

mysql-connector-python/lib/mysql/connector/aio/network.py

@@ -562,6 +562,8 @@ def build_ssl_context(
         tls_cipher_suites: Optional[List[str]] = [],
     ) -> ssl.SSLContext:
         """Build a SSLContext."""
+        tls_version: Optional[str] = None
+
         if not self._reader:
             raise InterfaceError(errno=2048)
 
@@ -607,7 +609,9 @@ def build_ssl_context(
                 except (IOError, ssl.SSLError) as err:
                     raise InterfaceError(f"Invalid Certificate/Key: {err}") from err
 
-            if tls_cipher_suites:
+            # TLSv1.3 ciphers cannot be disabled with `SSLContext.set_ciphers(...)`,
+            # see https://docs.python.org/3/library/ssl.html#ssl.SSLContext.set_ciphers.
+            if tls_cipher_suites and tls_version == "TLSv1.2":
                 context.set_ciphers(":".join(tls_cipher_suites))
 
             return context

mysql-connector-python/lib/mysql/connector/connection_cext.py

@@ -298,9 +298,9 @@ def _open_connection(self) -> None:
             tls_versions.sort(reverse=True)  # type: ignore[union-attr]
             tls_versions = ",".join(tls_versions)
         if self._ssl.get("tls_ciphersuites") is not None:
-            ssl_ciphersuites = self._ssl.get("tls_ciphersuites")[  # type: ignore[index]
-                0
-            ]
+            ssl_ciphersuites = (
+                self._ssl.get("tls_ciphersuites")[0] or None  # type: ignore[index]
+            )  # if it's the empty string, then use `None` instead
             tls_ciphersuites = self._ssl.get("tls_ciphersuites")[  # type: ignore[index]
                 1
             ]

mysql-connector-python/lib/mysql/connector/network.py

@@ -543,6 +543,8 @@ def build_ssl_context(
             NotSupportedError: Python installation has no SSL support.
             InterfaceError: Socket undefined or invalid ssl data.
         """
+        tls_version: Optional[str] = None
+
         if not self.sock:
             raise InterfaceError(errno=2048)
 
@@ -595,7 +597,9 @@ def build_ssl_context(
                 except (IOError, ssl.SSLError) as err:
                     raise InterfaceError(f"Invalid Certificate/Key: {err}") from err
 
-            if tls_cipher_suites:
+            # TLSv1.3 ciphers cannot be disabled with `SSLContext.set_ciphers(...)`,
+            # see https://docs.python.org/3/library/ssl.html#ssl.SSLContext.set_ciphers.
+            if tls_cipher_suites and tls_version == "TLSv1.2":
                 context.set_ciphers(":".join(tls_cipher_suites))
 
             return context

mysql-connector-python/tests/qa/test_qa_ciphers.py

@@ -31,6 +31,7 @@
 import asyncio
 import os
 import platform
+import ssl
 import unittest
 
 from contextlib import nullcontext
@@ -41,7 +42,7 @@
 import mysql.connector
 import mysql.connector.aio
 
-from mysql.connector.errors import NotSupportedError
+from mysql.connector.errors import InterfaceError, NotSupportedError
 from mysql.connector.tls_ciphers import (
     APPROVED_TLS_CIPHERSUITES,
     DEPRECATED_TLS_CIPHERSUITES,
@@ -82,7 +83,22 @@ def setUpModule() -> None:
     "No SSL support.",
 )
 class CipherTests(tests.MySQLConnectorTests):
-    """Testing cipher and cipher-suite lists for synchronous/asynchronous connection."""
+    """Testing cipher and cipher-suite lists for synchronous/asynchronous connection.
+
+    These tests verify the following about tls versions and ciphers:
+
+        Ciphers
+        -------
+        * Mandatory and approved ciphers are allowed.
+        * Deprecated ciphers are allowed, however a warning is raised.
+        * Unacceptable ciphers are forbidden; an error is raised.
+
+        TLS Versions
+        ------------
+        * Approved TLS versions are allowed.
+        * Deprecated TLS versions are allowed, however a warning is raised.
+        * Unacceptable TLS versions are forbidden; an error is raised.
+    """
 
     # when more than one approved TLS version is defined,
     # the latest available version is enforced.
@@ -368,3 +384,185 @@ def test_tls_ciphersuites_7(self):
     @tests.foreach_cnx()
     def test_tls_ciphersuites_8(self):
         self._test_tls_ciphersuites(test_case_id="8")
+
+
+@unittest.skipIf(
+    tests.MYSQL_EXTERNAL_SERVER,
+    "Test not available for external MySQL servers",
+)
+@unittest.skipIf(
+    not tests.SSL_AVAILABLE,
+    "No SSL support.",
+)
+@unittest.skipIf(
+    tests.MYSQL_VERSION < (9, 0, 0),
+    "MySQL Server should be 9.0 or newer.",
+)
+class CiphersAndTlsTests(tests.MySQLConnectorTests):
+    """Testing cipher and cipher-suite lists for synchronous/asynchronous connection.
+
+    These tests verify Connector/Python does support the TLS versions v1.2 and v1.3
+    when specifying valid ciphers.
+
+    When only caring about a specific TLS version, users can use the connection option
+    `tls_versions`. The ciphers to be used will be determined by the MySQL Server
+    during TLS negotiation.
+
+    On the other hand, when caring about specific ciphers, you should specify the
+    connection option `tls_ciphersuites`. Additionally, we recommend to also specify
+    the option `tls_versions`. If this latter option is skipped, Connector/Python will
+    use the latest supported TLS version. This might lead to cipher discrepancies;
+    assuming you provided TLSv1.2-related ciphers but didn't specify the TLS version,
+    these ciphers will be ignored as TLSv1.3 will be used, and the actual cipher in
+    use will be determined by the MySQL Server.
+
+    NOTE: the pure-python implementation does not support cipher selection
+    for TLSv1.3. The ultimate cipher to be used will be determined by the MySQL Server
+    during TLS negotiation. This limitation is because TLSv1.3 ciphers cannot be
+    disabled with `SSLContext.set_ciphers(...)`.
+    See https://docs.python.org/3/library/ssl.html#ssl.SSLContext.set_ciphers.
+    """
+
+    conf_ssl = {
+        "ssl_ca": os.path.abspath(os.path.join(tests.SSL_DIR, "tests_CA_cert.pem")),
+        "ssl_cert": os.path.abspath(
+            os.path.join(tests.SSL_DIR, "tests_client_cert.pem")
+        ),
+        "ssl_key": os.path.abspath(os.path.join(tests.SSL_DIR, "tests_client_key.pem")),
+    }
+
+    # SHOW STATUS LIKE 'ssl_cipher_list';
+    ssl_v13_ciphers_server = (
+        "TLS_AES_128_GCM_SHA256:TLS_AES_256_GCM_SHA384:TLS_CHACHA20_POLY1305_SHA256:"
+        "TLS_AES_128_CCM_SHA256"
+    )
+    ssl_v12_ciphers_server = (
+        "ECDHE-RSA-AES128-GCM-SHA256:"
+        "ECDHE-RSA-AES256-GCM-SHA384:"
+        "ECDHE-RSA-CHACHA20-POLY1305:"
+        "DHE-RSA-AES128-GCM-SHA256:DHE-RSA-AES256-GCM-SHA384:DHE-RSA-AES256-CCM:"
+        "DHE-RSA-AES128-CCM:DHE-RSA-CHACHA20-POLY1305"
+    )
+
+    def setUp(self) -> None:
+        # tls_cases[i] is a 2-tuple: (expected error/warning if any, cipher to be used)
+        self.tls_v12_cases = [
+            (None, [cipher]) for cipher in self.ssl_v12_ciphers_server.split(":")
+        ]
+        self.tls_v13_cases = [
+            (None, [cipher]) for cipher in self.ssl_v13_ciphers_server.split(":")
+        ]
+
+    async def _check_async_tls_ciphersuites(
+        self, settings, exp_ssl_version: Optional[str]
+    ) -> Tuple[Optional[str], Optional[str]]:
+        ssl_version, ssl_cipher = None, None
+        async with await mysql.connector.aio.connect(**settings) as cnx:
+            async with await cnx.cursor() as cur:
+                if exp_ssl_version:
+                    await cur.execute("SHOW STATUS LIKE 'Ssl_version'")
+                    res = await cur.fetchone()
+                    ssl_version = res[-1]
+
+                await cur.execute("SHOW STATUS LIKE 'Ssl_cipher'")
+                res = await cur.fetchone()
+                ssl_cipher = res[-1]
+
+        return ssl_version, ssl_cipher
+
+    def _check_sync_tls_ciphersuites(
+        self, settings, exp_ssl_version: Optional[str]
+    ) -> Tuple[Optional[str], Optional[str]]:
+        ssl_version, ssl_cipher = None, None
+        with mysql.connector.connect(**settings) as cnx:
+            with cnx.cursor() as cur:
+                if exp_ssl_version:
+                    cur.execute("SHOW STATUS LIKE 'Ssl_version'")
+                    res = cur.fetchone()
+                    ssl_version = res[-1]
+
+                cur.execute("SHOW STATUS LIKE 'Ssl_cipher'")
+                res = cur.fetchone()
+                ssl_cipher = res[-1]
+
+        return ssl_version, ssl_cipher
+
+    def _test_tls_ciphersuites(
+        self, tls_versions: Optional[list[str]], test_case: tuple, verify: bool = False
+    ):
+        exp_event, tls_ciphersuites = test_case
+
+        conf = {**tests.get_mysql_config(), **self.conf_ssl}
+        conf["use_pure"] = isinstance(self.cnx, mysql.connector.MySQLConnection)
+        conf["unix_socket"] = None
+        conf["tls_ciphersuites"] = tls_ciphersuites
+
+        if tls_versions is not None:
+            conf["tls_versions"] = tls_versions
+
+        event_handler = (
+            self.assertWarns if exp_event == DeprecationWarning else self.assertRaises
+        )
+
+        exp_ssl_version = tls_versions[0] if tls_versions else "TLSv1.3"
+        with nullcontext() if exp_event is None else event_handler(exp_event):
+            ssl_version, ssl_cipher = self._check_sync_tls_ciphersuites(
+                conf, exp_ssl_version
+            )
+            self.assertEqual(ssl_version, exp_ssl_version)
+            if verify:
+                self.assertEqual(ssl_cipher, tls_ciphersuites[0])
+
+        # C-ext implementation isn't supported yet for aio.
+        if (CEXT_SUPPORT_FOR_AIO and not conf["use_pure"]) or conf["use_pure"]:
+            with nullcontext() if exp_event is None else event_handler(exp_event):
+                ssl_version, ssl_cipher = asyncio.run(
+                    self._check_async_tls_ciphersuites(conf, exp_ssl_version)
+                )
+                self.assertEqual(ssl_version, exp_ssl_version)
+                if verify:
+                    self.assertEqual(ssl_cipher, tls_ciphersuites[0])
+
+    @tests.foreach_cnx(mysql.connector.MySQLConnection)
+    def test_tls_v12_ciphers(self):
+        # verify=True means the test checks the selected cipher matches
+        # with the one returned by the server.
+        for test_case in self.tls_v12_cases:
+            self._test_tls_ciphersuites(
+                tls_versions=["TLSv1.2"], test_case=test_case, verify=True
+            )
+
+    @tests.foreach_cnx(mysql.connector.MySQLConnection)
+    def test_tls_v13_ciphers(self):
+        # verify=True means the test checks the selected cipher matches
+        # with the one returned by the server.
+        for test_case in self.tls_v13_cases:
+            # verification should be False since cipher selection
+            # for TLSv1.3 isn't supported.
+            self._test_tls_ciphersuites(
+                tls_versions=["TLSv1.3"], test_case=test_case, verify=False
+            )
+            self._test_tls_ciphersuites(
+                tls_versions=None, test_case=test_case, verify=False
+            )
+
+    @tests.foreach_cnx(mysql.connector.CMySQLConnection)
+    def test_tls_v12_ciphers_cext(self):
+        # verify=True means the test checks the selected cipher matches
+        # with the one returned by the server.
+        for test_case in self.tls_v12_cases:
+            self._test_tls_ciphersuites(
+                tls_versions=["TLSv1.2"], test_case=test_case, verify=True
+            )
+
+    @tests.foreach_cnx(mysql.connector.CMySQLConnection)
+    def test_tls_v13_ciphers_cext(self):
+        # verify=True means the test checks the selected cipher matches
+        # with the one returned by the server.
+        for test_case in self.tls_v13_cases:
+            self._test_tls_ciphersuites(
+                tls_versions=["TLSv1.3"], test_case=test_case, verify=True
+            )
+            self._test_tls_ciphersuites(
+                tls_versions=None, test_case=test_case, verify=True
+            )