Require TLS for MySQL ssl_mode required

docs/mysql.md

@@ -170,6 +170,8 @@ The `ssl_mode` setting controls how the connection to MySQL is secured:
 | `verify-ca` | Require SSL and verify the server certificate against the CA |
 | `verify-identity` | Require SSL, verify the certificate, and verify the hostname |
 
+With `required`, `verify-ca` or `verify-identity`, the connection fails when the server offers no TLS. With `preferred`, the connection continues without TLS. The tool needs PyMySQL 1.2.0 or later, which stops before it sends the login.
+
 For PlanetScale and most cloud-hosted MySQL instances, use `required` or higher.
 
 ## What Data Is Collected

planetscale_discovery/database/mysql_discovery.py

@@ -4,6 +4,7 @@
 """
 
 import logging
+import ssl
 from datetime import datetime, timezone
 from typing import Dict, Any
 
@@ -14,6 +15,23 @@
 from .mysql_analyzers.feature_analyzer import MySQLFeatureAnalyzer
 from .mysql_analyzers.security_analyzer import MySQLSecurityAnalyzer
 
+# The ssl_mode values that must not connect without TLS.
+TLS_REQUIRED_MODES = ("required", "verify-ca", "verify-identity")
+
+# The first PyMySQL that stops before the login when TLS is required and the
+# server offers none.
+MIN_PYMYSQL = (1, 2, 0)
+
+
+def _uses_tls(connection) -> bool:
+    """Whether PyMySQL encrypted this connection.
+
+    PyMySQL has no public attribute for this. ``_sock`` is the socket it reads
+    and writes, and PyMySQL replaces it with an ``ssl.SSLSocket`` when TLS
+    starts.
+    """
+    return isinstance(getattr(connection, "_sock", None), ssl.SSLSocket)
+
 
 class MySQLDiscoveryTool:
     """MySQL discovery tool wrapper for unified CLI."""
@@ -89,19 +107,28 @@ def connect(self) -> bool:
             }
 
             # SSL handling — map MySQL-style ssl_mode to PyMySQL params.
-            #   disabled              -> ssl_disabled=True, no TLS
-            #   preferred / required  -> ssl={}, TLS on, no cert chain verification
-            #   verify-ca             -> ssl_ca required, verify chain only
-            #   verify-identity       -> ssl_ca required, verify chain + hostname
+            #   disabled         -> ssl_disabled=True, no TLS
+            #   preferred        -> no TLS option, TLS when the server offers it
+            #   required         -> TLS required, no cert chain verification
+            #   verify-ca        -> ssl_ca required, verify chain only
+            #   verify-identity  -> ssl_ca required, verify chain + hostname
             ssl_mode = (self.connection_params.get("ssl_mode", "") or "").lower()
             ssl_ca = self.connection_params.get("ssl_ca")
             ssl_cert = self.connection_params.get("ssl_cert")
             ssl_key = self.connection_params.get("ssl_key")
 
             if ssl_mode in ("", "disabled"):
                 params["ssl_disabled"] = True
-            elif ssl_mode in ("preferred", "required"):
-                params["ssl"] = {}
+            elif ssl_mode == "preferred":
+                # Set no TLS option. PyMySQL then tries TLS without verifying
+                # the certificate, and continues without TLS when the server
+                # offers none.
+                pass
+            elif ssl_mode == "required":
+                # PyMySQL ignores an empty dict. A dict with a key turns TLS
+                # on, and PyMySQL 1.2.0 and later then fail when the server
+                # offers no TLS. With no CA, PyMySQL skips verification.
+                params["ssl"] = {"check_hostname": False, "verify_mode": False}
             elif ssl_mode in ("verify-ca", "verify-identity"):
                 if not ssl_ca:
                     self.logger.error(
@@ -124,11 +151,40 @@ def connect(self) -> bool:
                 )
                 params["ssl_disabled"] = True
 
+            # The floor is 1.2.0, but an existing install can be older, and
+            # nothing checks the version at install time. pymysql.__version__
+            # is the MySQLdb compatibility number, so read VERSION instead.
+            version = getattr(pymysql, "VERSION", None)
+            if (
+                ssl_mode in ("preferred",) + TLS_REQUIRED_MODES
+                and isinstance(version, tuple)
+                and tuple(version[:3]) < MIN_PYMYSQL
+            ):
+                self.logger.warning(
+                    f"PyMySQL {'.'.join(str(v) for v in version[:3])} is older "
+                    "than 1.2.0. With it, ssl_mode='preferred' never tries TLS, "
+                    "and a mode that requires TLS sends the login before this "
+                    "tool can refuse a connection without TLS. Upgrade PyMySQL."
+                )
+
             # Connect to specific database, or information_schema as fallback
             db = self.connection_params.get("database", "")
             params["database"] = db if db else "information_schema"
 
-            self.connection = pymysql.connect(**params)
+            connection = pymysql.connect(**params)
+
+            # PyMySQL before 1.2.0 does not stop when the server offers no
+            # TLS, even with an ssl dict. So a mode that requires TLS checks
+            # the socket, and refuses a connection that is not encrypted.
+            if ssl_mode in TLS_REQUIRED_MODES and not _uses_tls(connection):
+                connection.close()
+                self.logger.error(
+                    f"ssl_mode={ssl_mode!r} requires TLS, but the connection "
+                    "is not encrypted. Make sure the server offers TLS, and "
+                    "use PyMySQL 1.2.0 or later."
+                )
+                return False
+            self.connection = connection
 
             # Get connection info
             cursor = self.connection.cursor()

pyproject.toml

@@ -34,7 +34,7 @@ postgres = [
     "psycopg2-binary>=2.9",
 ]
 mysql = [
-    "PyMySQL>=1.1.0",
+    "PyMySQL>=1.2.0",
 ]
 aws = [
     "boto3>=1.42.78",
@@ -75,7 +75,7 @@ workload = [
 ]
 all = [
     "psycopg2-binary>=2.9",
-    "PyMySQL>=1.1.0",
+    "PyMySQL>=1.2.0",
     "pglast>=7,<9",
     "boto3>=1.42.78",
     "botocore>=1.42.78",

requirements.txt

@@ -37,4 +37,4 @@ azure-mgmt-network>=32.0.0
 requests>=2.32.5
 
 # MySQL database discovery dependencies
-PyMySQL>=1.1.0
+PyMySQL>=1.2.0

setup.py

@@ -36,7 +36,7 @@
             "psycopg2-binary>=2.9",
         ],
         "mysql": [
-            "PyMySQL>=1.1.0",
+            "PyMySQL>=1.2.0",
         ],
         "aws": [
             "boto3>=1.42.52",
@@ -76,7 +76,7 @@
             "pglast>=7,<9",
         ],
         "all": [
-            "PyMySQL>=1.1.0",
+            "PyMySQL>=1.2.0",
             "pglast>=7,<9",
             "boto3>=1.42.52",
             "botocore>=1.42.52",

setup.sh

@@ -225,7 +225,7 @@ else
 fi
 
 if [ "$engine_choice" = "mysql" ]; then
-    DEPS+=("PyMySQL>=1.1.0")
+    DEPS+=("PyMySQL>=1.2.0")
     INSTALLED_ENGINES="mysql"
 else
     DEPS+=("psycopg2-binary>=2.9.11")

tests/unit/test_mysql_discovery.py

@@ -4,6 +4,8 @@
 """
 
 import pytest
+import socket
+from ssl import CERT_NONE, SSLSocket
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
@@ -30,6 +32,7 @@ def _capture_connect_params(self, params):
         def fake_connect(**kwargs):
             captured.update(kwargs)
             conn = MagicMock()
+            conn._sock = MagicMock(spec=SSLSocket)
             cursor = MagicMock()
             cursor.fetchone.return_value = ("8.0.34", "ps@%", "app")
             conn.cursor.return_value = cursor
@@ -54,16 +57,98 @@ def test_ssl_empty_defaults_to_disabled(self, base_params):
         captured = self._capture_connect_params(base_params)
         assert captured.get("ssl_disabled") is True
 
-    def test_ssl_required_enables_tls_without_verification(self, base_params):
+    def _pymysql_connection(self, params):
+        """What PyMySQL itself makes of the params, with no server.
+
+        Asserting on the params alone passed while ``required`` connected
+        without TLS: PyMySQL ignores an empty ssl dict.
+        """
+        pymysql = pytest.importorskip("pymysql")
+        captured = self._capture_connect_params(params)
+        return pymysql.connections.Connection(defer_connect=True, **captured)
+
+    def test_ssl_required_requires_tls_without_verification(self, base_params):
+        # Catches ssl={}, which PyMySQL ignores, so the connection was plaintext.
         base_params["ssl_mode"] = "required"
-        captured = self._capture_connect_params(base_params)
-        assert captured.get("ssl") == {}
-        assert "ssl_disabled" not in captured
+        conn = self._pymysql_connection(base_params)
+        assert conn.ssl is True
+        assert conn._ssl_required is True
+        assert conn.ctx.verify_mode == CERT_NONE
+        assert conn.ctx.check_hostname is False
+
+    def test_ssl_preferred_tries_tls_and_may_fall_back(self, base_params):
+        # Catches a preferred mode that never tries TLS, or that refuses a
+        # server with no TLS, which is the behavior of required.
+        base_params["ssl_mode"] = "preferred"
+        conn = self._pymysql_connection(base_params)
+        assert conn.ssl is True
+        assert conn._ssl_required is False
+        assert conn.ctx.verify_mode == CERT_NONE
+
+    def _connect_without_tls(self, params):
+        """connect() when PyMySQL returns a connection without TLS. PyMySQL
+        before 1.2.0 can do that even with an ssl dict."""
+        conn = MagicMock()
+        conn._sock = MagicMock(spec=socket.socket)
+        conn.cursor.return_value.fetchone.return_value = ("8.0.34", "ps@%", "app")
+        fake_pymysql = MagicMock()
+        fake_pymysql.connect.return_value = conn
+        with patch.dict("sys.modules", {"pymysql": fake_pymysql}):
+            discovery = MySQLDiscovery(params)
+            return discovery.connect(), discovery, conn
+
+    # A literal list, not the production constant: narrowing the constant must
+    # fail this test, not shrink it.
+    @pytest.mark.parametrize("mode", ["required", "verify-ca", "verify-identity"])
+    def test_mode_that_requires_tls_refuses_a_plaintext_connection(
+        self, base_params, mode
+    ):
+        # Catches an old PyMySQL that drops TLS without an error.
+        base_params["ssl_mode"] = mode
+        base_params["ssl_ca"] = "/etc/ssl/ca.pem"
+        connected, discovery, conn = self._connect_without_tls(base_params)
+        assert connected is False
+        assert discovery.connection is None
+        conn.close.assert_called_once()
 
-    def test_ssl_preferred_enables_tls(self, base_params):
+    def test_preferred_accepts_a_plaintext_connection(self, base_params):
         base_params["ssl_mode"] = "preferred"
-        captured = self._capture_connect_params(base_params)
-        assert captured.get("ssl") == {}
+        connected, discovery, conn = self._connect_without_tls(base_params)
+        assert connected is True
+        assert discovery.connection is conn
+
+    @pytest.mark.parametrize("mode", [None, "disabled", "garbage"])
+    def test_a_mode_without_tls_accepts_a_plaintext_connection(self, base_params, mode):
+        # Catches a socket check that reaches past the modes that require TLS.
+        # The other tests of these modes always hand back a TLS socket.
+        if mode is not None:
+            base_params["ssl_mode"] = mode
+        connected, discovery, conn = self._connect_without_tls(base_params)
+        assert connected is True
+        assert discovery.connection is conn
+
+    @pytest.mark.parametrize(
+        "version, mode, warned",
+        [
+            ((1, 1, 2, "final"), "required", True),
+            ((1, 1, 2, "final"), "preferred", True),
+            ((1, 1, 2, "final"), "disabled", False),
+            ((1, 2, 0, "final"), "required", False),
+        ],
+    )
+    def test_an_old_pymysql_is_named_when_tls_matters(
+        self, base_params, caplog, version, mode, warned
+    ):
+        # The floor constrains a fresh install only. An old install still runs.
+        base_params["ssl_mode"] = mode
+        fake_pymysql = MagicMock()
+        fake_pymysql.VERSION = version
+        conn = fake_pymysql.connect.return_value
+        conn._sock = MagicMock(spec=SSLSocket)
+        conn.cursor.return_value.fetchone.return_value = ("8.0.34", "ps@%", "app")
+        with patch.dict("sys.modules", {"pymysql": fake_pymysql}):
+            MySQLDiscovery(base_params).connect()
+        assert ("older than 1.2.0" in caplog.text) is warned
 
     def test_verify_ca_requires_ssl_ca(self, base_params):
         base_params["ssl_mode"] = "verify-ca"