BUG21656282: Connection fails using unicode passwords with C extension

The connector fails when using a password with unicode passwords with
the C extension. This issue only occurs using Python 2.

This patch fixes this issue by encoding the password when using Python 2.

A test was added for regression.

Author: nmariz

src/mysql_capi.c

@@ -1025,10 +1025,9 @@ MySQL_commit(MySQL *self)
 PyObject*
 MySQL_connect(MySQL *self, PyObject *args, PyObject *kwds)
 {
-	char *host= NULL, *user= NULL, *password= NULL, *database= NULL,
-	     *unix_socket= NULL;
+	char *host= NULL, *user= NULL, *database= NULL, *unix_socket= NULL;
 	char *ssl_ca= NULL, *ssl_cert= NULL, *ssl_key= NULL;
-	PyObject *charset_name, *compress, *ssl_verify_cert;
+	PyObject *charset_name, *compress, *ssl_verify_cert, *password;
 	const char* auth_plugin;
 	unsigned long client_flags= 0;
 	unsigned int port= 3306, tmp_uint;
@@ -1046,7 +1045,11 @@ MySQL_connect(MySQL *self, PyObject *args, PyObject *kwds)
 		NULL
     };
 
+#ifdef PY3
     if (!PyArg_ParseTupleAndKeywords(args, kwds, "|zzzzkzkzzzO!O!", kwlist,
+#else
+    if (!PyArg_ParseTupleAndKeywords(args, kwds, "|zzOzkzkzzzO!O!", kwlist,
+#endif
                                      &host, &user, &password, &database,
                                      &port, &unix_socket,
                                      &client_flags,
@@ -1179,9 +1182,24 @@ MySQL_connect(MySQL *self, PyObject *args, PyObject *kwds)
         mysql_options(&self->session, MYSQL_OPT_LOCAL_INFILE, (unsigned int*)&abool);
     }
 
-    res= mysql_real_connect(&self->session,
-                            host, user, password, database,
-    						port, unix_socket, client_flags);
+#ifdef PY3
+    res= mysql_real_connect(&self->session, host, user, password, database,
+                            port, unix_socket, client_flags);
+#else
+    char* c_password;
+    if (PyUnicode_Check(password))
+    {
+        PyObject* u_password= PyUnicode_AsUTF8String(password);
+        c_password= PyString_AsString(u_password);
+        Py_DECREF(u_password);
+    }
+    else
+    {
+        c_password= PyString_AsString(password);
+    }
+    res= mysql_real_connect(&self->session, host, user, c_password, database,
+                            port, unix_socket, client_flags);
+#endif
 
     Py_END_ALLOW_THREADS
 

tests/test_bugs.py

@@ -4143,3 +4143,56 @@ def test_set(self):
             results.append(result.fetchall())
         self.assertEqual(exp, results)
         cur.close()
+
+
+@unittest.skipIf(not CMySQLConnection, ERR_NO_CEXT)
+class BugOra21656282(tests.MySQLConnectorTests):
+    """BUG#21656282: CONNECT FAILURE WITH C-EXT WHEN PASSWORD CONTAINS UNICODE
+    CHARACTER
+    """
+    def setUp(self):
+        config = tests.get_mysql_config()
+        self.cnx = connection.MySQLConnection(**config)
+        self.host = 'localhost' if config['unix_socket'] and os.name != 'nt' \
+                    else config['host']
+        self.user = 'unicode_user'
+        self.password = u'步'
+
+        # Use utf8mb4 character set
+        self.cnx.cmd_query("SET character_set_server='utf8mb4'")
+
+        # Drop user if exists
+        self._drop_user(self.host, self.user)
+
+        # Create the user with unicode password
+        create_user = (u"CREATE USER '{user}'@'{host}' IDENTIFIED BY "
+                       u"'{password}'")
+        self.cnx.cmd_query(create_user.format(user=self.user, host=self.host,
+                                              password=self.password))
+
+        # Grant all to new user on database
+        grant = "GRANT ALL ON {database}.* TO '{user}'@'{host}'"
+        self.cnx.cmd_query(grant.format(database=config['database'],
+                                        user=self.user, host=self.host))
+
+    def tearDown(self):
+        self._drop_user(self.host, self.user)
+
+    def _drop_user(self, host, user):
+        try:
+            drop_user = "DROP USER '{user}'@'{host}'"
+            self.cnx.cmd_query(drop_user.format(user=user, host=host))
+        except errors.DatabaseError:
+            # It's OK when drop user fails
+            pass
+
+    def test_unicode_password(self):
+        config = tests.get_mysql_config()
+        config['user'] = self.user
+        config['password'] = self.password
+        try:
+            cnx = CMySQLConnection(**config)
+        except:
+            self.fail('Failed using password with unicode characters')
+        else:
+            cnx.close()