BUG32162928: Change user command fails on pure python implementation

The cmd_change_user() command failed with pure-python due to the missing
connection attributes not being sent along the authentication data.

Author: isrgomez

CHANGES.txt

@@ -16,6 +16,7 @@ v8.0.24
 - WL#14212: Support connection close notification
 - WL#14027: Add support for Python 3.9
 - BUG#32435181: Add support for Django 3.2
+- BUG#32162928: Change user command fails on pure python implementation
 - BUG#32029891: Add context manager support for pooled connections
 - BUG#31490101: Fix wrong cast of Python unicode to std::string
 - BUG#30416704: Binary columns returned as strings

lib/mysql/connector/connection.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2009, 2020, Oracle and/or its affiliates.
+# Copyright (c) 2009, 2021, Oracle and/or its affiliates.
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License, version 2.0, as
@@ -842,13 +842,13 @@ def cmd_change_user(self, username='', password='', database='',
         if self._compress:
             raise errors.NotSupportedError("Change user is not supported with "
                                            "compression.")
-
         packet = self._protocol.make_change_user(
             handshake=self._handshake,
             username=username, password=password, database=database,
             charset=charset, client_flags=self._client_flags,
             ssl_enabled=self._ssl_active,
-            auth_plugin=self._auth_plugin)
+            auth_plugin=self._auth_plugin,
+            conn_attrs=self._conn_attrs)
         self._socket.send(packet, 0, 0)
 
         ok_packet = self._auth_switch_request(username, password)

lib/mysql/connector/protocol.py

@@ -156,7 +156,7 @@ def make_stmt_fetch(self, statement_id, rows=1):
 
     def make_change_user(self, handshake, username=None, password=None,
                          database=None, charset=45, client_flags=0,
-                         ssl_enabled=False, auth_plugin=None):
+                         ssl_enabled=False, auth_plugin=None, conn_attrs=None):
         """Make a MySQL packet with the Change User command"""
 
         try:
@@ -188,6 +188,9 @@ def make_change_user(self, handshake, username=None, password=None,
         if client_flags & ClientFlag.PLUGIN_AUTH:
             packet += auth_plugin.encode('utf8') + b'\x00'
 
+        if (client_flags & ClientFlag.CONNECT_ARGS) and conn_attrs is not None:
+            packet += self.make_conn_attrs(conn_attrs)
+
         return packet
 
     def parse_handshake(self, packet):

tests/test_bugs.py

@@ -60,8 +60,8 @@
 
 from tests import foreach_cnx, cnx_config
 from . import check_tls_versions_support
-from mysql.connector import (connection, cursor, conversion, protocol,
-                             errors, constants, pooling)
+from mysql.connector import (connection, constants, conversion, cursor,
+                             errors, MySQLConnection, pooling, protocol,)
 from mysql.connector.optionfiles import read_option_files
 from mysql.connector.pooling import PooledMySQLConnection
 import mysql.connector
@@ -5943,3 +5943,121 @@ def test_without_use_unicode_cursor_prepared(self):
             cursor=cursor.MySQLCursorPrepared,
             use_binary_charset=True,
         )
+
+
+class Bug32162928(tests.MySQLConnectorTests):
+    """BUG#32162928: change user command fails with pure python implementation.
+
+    The cmd_change_user() command fails with pure-python.
+    """
+    def setUp(self):
+        self.connect_kwargs = tests.get_mysql_config()
+        cnx = MySQLConnection(**self.connect_kwargs)
+        self.users = [
+            {'user': 'user_native1',
+             'host': self.connect_kwargs['host'],
+             'port': self.connect_kwargs['port'],
+             'database': self.connect_kwargs['database'],
+             'password': 'native1_pass',
+             'auth_plugin': 'mysql_native_password'},
+            {'user': 'user_native2',
+             'host': self.connect_kwargs['host'],
+             'port': self.connect_kwargs['port'],
+             'database': self.connect_kwargs['database'],
+             'password': 'native2_pass',
+             'auth_plugin': 'mysql_native_password'},
+            {'user': 'user_sha2561',
+             'host': self.connect_kwargs['host'],
+             'port': self.connect_kwargs['port'],
+             'database': self.connect_kwargs['database'],
+             'password': 'sha2561_pass',
+             'auth_plugin': 'sha256_password'},
+            {'user': 'user_sha2562',
+             'host': self.connect_kwargs['host'],
+             'port': self.connect_kwargs['port'],
+             'database': self.connect_kwargs['database'],
+             'password': 'sha2562_pass',
+             'auth_plugin': 'sha256_password'},
+            {'user': 'user_caching1',
+             'host': self.connect_kwargs['host'],
+             'port': self.connect_kwargs['port'],
+             'database': self.connect_kwargs['database'],
+             'password': 'caching1_pass',
+             'auth_plugin': 'caching_sha2_password'},
+            {'user': 'user_caching2',
+             'host': self.connect_kwargs['host'],
+             'port': self.connect_kwargs['port'],
+             'database': self.connect_kwargs['database'],
+             'password': 'caching2_pass',
+             'auth_plugin': 'caching_sha2_password'}]
+
+        # create users
+        if tests.MYSQL_VERSION < (8, 0, 0):
+            self.new_users = self.users[0:4]
+        else:
+            self.new_users = self.users
+
+        for new_user in self.new_users:
+            cnx.cmd_query("DROP USER IF EXISTS '{user}'@'{host}'".format(**new_user))
+
+            stmt = ("CREATE USER IF NOT EXISTS '{user}'@'{host}' IDENTIFIED "
+                    "WITH {auth_plugin} BY '{password}'").format(**new_user)
+            cnx.cmd_query(stmt)
+
+            cnx.cmd_query("GRANT ALL PRIVILEGES ON {database}.* TO "
+                          "'{user}'@'{host}'".format(**new_user))
+
+    @foreach_cnx()
+    def test_change_user(self):
+        # test users can connect
+        for user in self.new_users:
+            conn_args = user.copy()
+            try:
+                self.connect_kwargs.pop('auth_plugin')
+            except:
+                pass
+            cnx_test = self.cnx.__class__(**conn_args)
+            cnx_test.cmd_query("SELECT USER()")
+            logged_user = cnx_test.get_rows()[0][0][0]
+            self.assertEqual(u"{user}@{host}".format(**user), logged_user)
+            cnx_test.close()
+
+        # tests change user
+        if tests.MYSQL_VERSION < (8, 0, 0):
+            # 5.6 does not support caching_sha2_password auth plugin
+            test_cases = [(0, 1), (1, 2), (2,3), (3, 0)]
+        else:
+            test_cases = [(0, 1), (1, 2), (2,3), (3, 0), (3, 4), (4, 5),
+                          (5, 3), (5, 0)]
+        for user1, user2 in test_cases:
+            conn_args_user1 = self.users[user1].copy()
+            try:
+                conn_args_user1.pop('auth_plugin')
+            except:
+                pass
+            if tests.MYSQL_VERSION < (8, 0, 0):
+                # change user does not work in 5.x with charset utf8mb4
+                conn_args_user1['charset'] = 'utf8'
+
+            cnx_test = self.cnx.__class__(**conn_args_user1)
+            cnx_test.cmd_query("SELECT USER()")
+            first_user = cnx_test.get_rows()[0][0][0]
+            self.assertEqual(
+                u"{user}@{host}".format(**self.users[user1]), first_user)
+
+            cnx_test.cmd_change_user(self.users[user2]['user'],
+                                     self.users[user2]['password'],
+                                     self.users[user2]['database'])
+            cnx_test.cmd_query("SELECT USER()")
+            rows = cnx_test.get_rows()
+            current_user = rows[0][0][0]
+            self.assertNotEqual(first_user, current_user)
+            self.assertEqual(
+                u"{user}@{host}".format(**self.users[user2]), current_user)
+            cnx_test.close()
+
+    def tearDown(self):
+        cnx = MySQLConnection(**self.connect_kwargs)
+        # cleanup users
+        for new_user in self.users:
+            cnx.cmd_query("DROP USER IF EXISTS '{user}'@'{host}'".format(**new_user))