From 75f4bdc68d4b5745c518472e8bc2b02cec0f81e6 Mon Sep 17 00:00:00 2001
From: Jerry Zhao <jerryzhao@fortinet.com>
Date: Thu, 10 Dec 2020 17:43:54 -0800
Subject: [PATCH] add ssl_check_hostname option in mysqldb

provide option to disable check_hostname for ssl connections to
the server

Fixes: #5397
---
 lib/sqlalchemy/dialects/mysql/mysqldb.py |  8 ++--
 test/dialect/mysql/test_dialect.py       | 48 ++++++++++++------------
 2 files changed, 30 insertions(+), 26 deletions(-)

diff --git a/lib/sqlalchemy/dialects/mysql/mysqldb.py b/lib/sqlalchemy/dialects/mysql/mysqldb.py
index b20e061fb5..5c1e964115 100644
--- a/lib/sqlalchemy/dialects/mysql/mysqldb.py
+++ b/lib/sqlalchemy/dialects/mysql/mysqldb.py
@@ -199,11 +199,13 @@ class MySQLDialect_mysqldb(MySQLDialect):
         # query string.
 
         ssl = {}
-        keys = ["ssl_ca", "ssl_key", "ssl_cert", "ssl_capath", "ssl_cipher"]
-        for key in keys:
+        keys = [("ssl_ca", str), ("ssl_key", str), ("ssl_cert", str),
+                ("ssl_capath", str), ("ssl_cipher", str),
+                ("ssl_check_hostname", bool)]
+        for key, kw_type in keys:
             if key in opts:
                 ssl[key[4:]] = opts[key]
-                util.coerce_kw_type(ssl, key[4:], str)
+                util.coerce_kw_type(ssl, key[4:], kw_type)
                 del opts[key]
         if ssl:
             opts["ssl"] = ssl
diff --git a/test/dialect/mysql/test_dialect.py b/test/dialect/mysql/test_dialect.py
index abd3a491ff..148449281c 100644
--- a/test/dialect/mysql/test_dialect.py
+++ b/test/dialect/mysql/test_dialect.py
@@ -128,7 +128,7 @@ class DialectTest(fixtures.TestBase):
         from sqlalchemy.dialects.mysql import mysqldb
 
         dialect = mysqldb.dialect()
-        self._test_ssl_arguments(dialect)
+        self._test_ssl_arguments(dialect, sql_type=mysqldb)
 
     def test_ssl_arguments_oursql(self):
         from sqlalchemy.dialects.mysql import oursql
@@ -136,31 +136,33 @@ class DialectTest(fixtures.TestBase):
         dialect = oursql.dialect()
         self._test_ssl_arguments(dialect)
 
-    def _test_ssl_arguments(self, dialect):
-        kwarg = dialect.create_connect_args(
-            make_url(
-                "mysql://scott:tiger@localhost:3306/test"
-                "?ssl_ca=/ca.pem&ssl_cert=/cert.pem&ssl_key=/key.pem"
-            )
-        )[1]
+    def _test_ssl_arguments(self, dialect, sql_type="oursql"):
+        url = (
+            "mysql://scott:tiger@localhost:3306/test"
+            "?ssl_ca=/ca.pem&ssl_cert=/cert.pem&ssl_key=/key.pem"
+        )
+        expected = {
+            "passwd": "tiger",
+            "db": "test",
+            "ssl": {
+                "ca": "/ca.pem",
+                "cert": "/cert.pem",
+                "key": "/key.pem"
+            },
+            "host": "localhost",
+            "user": "scott",
+            "port": 3306
+        }
+        # add check_hostname check for mysqldb
+        if sql_type == "mysqldb":
+            url = url + "&ssl_check_hostname=false"
+            expected['ssl']['check_hostname'] = False
+
+        kwarg = dialect.create_connect_args(make_url(url))[1]
         # args that differ among mysqldb and oursql
         for k in ("use_unicode", "found_rows", "client_flag"):
             kwarg.pop(k, None)
-        eq_(
-            kwarg,
-            {
-                "passwd": "tiger",
-                "db": "test",
-                "ssl": {
-                    "ca": "/ca.pem",
-                    "cert": "/cert.pem",
-                    "key": "/key.pem",
-                },
-                "host": "localhost",
-                "user": "scott",
-                "port": 3306,
-            },
-        )
+        eq_(kwarg, expected)
 
     @testing.combinations(
         ("compress", True),
-- 
2.47.3