WL#16754: The host wildcard no longer applies to localhost

The unittests are updated to reflect on the fact that the host
wildcard match behaviour of schema and object level grants are
not going to be applicable to localhost anymore.

Change-Id: I2cce0a2b7bd266dc0dde84cc03274ad610112476

Author: SubCoder1

CHANGES.txt

@@ -11,6 +11,7 @@ Full release notes:
 v9.3.0
 ======
 
+- WL#16754: The host wildcard no longer applies to localhost
 - WL#16752: Deprecate class methods to access instance data or to know instance internal state
 - WL#16327: Remove Cursors Prepared Raw and Named Tuple
 - BUG#37541353: (Contribution) Fix typing annotation of MySQLConnectionAbstract's close function

mysql-connector-python/tests/otel/context_propagation/sanity_check.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2023, 2024, Oracle and/or its affiliates.
+# Copyright (c) 2023, 2025, 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
@@ -209,12 +209,12 @@ def __init__(self, **config) -> None:
         self._config = config.copy()
         self._tracer = None
         self._mysql_config = {
+            "host": config["mysql_host"],
             "username": config["mysql_user"],
             "password": config["mysql_password"],
             "port": config["mysql_port"],
             "use_pure": config["cpy_use_pure"],
         }
-
         self._init_otel()
 
     def _init_otel(self):
@@ -250,10 +250,10 @@ def test_simple_stmts(
         new_user_password = "s3cr3t"
         new_database = "colors"
         new_user_stmt = (
-            f"CREATE USER '{new_user_name}'@'%' IDENTIFIED BY '{new_user_password}'"
+            f"CREATE USER '{new_user_name}'@'{self._mysql_config['host']}' IDENTIFIED BY '{new_user_password}'"
         )
         grant_stmt = (
-            f"GRANT ALL PRIVILEGES ON *.* TO '{new_user_name}'@'%' WITH GRANT OPTION"
+            f"GRANT ALL PRIVILEGES ON *.* TO '{new_user_name}'@'{self._mysql_config['host']}' WITH GRANT OPTION"
         )
         table_name = "employees"
         create_stmt = f"""CREATE TABLE {table_name} (

mysql-connector-python/tests/otel/test_instrumentation.py

@@ -371,10 +371,10 @@ class PythonWithGlobalInstSpanTests(tests.MySQLConnectorTests):
     new_user_password = "s3cr3t"
     new_database = "colors"
     new_user_stmt = (
-        f"CREATE USER '{new_user_name}'@'%' IDENTIFIED BY '{new_user_password}'"
+        "CREATE USER '{new_user_name}'@'{server_host}' IDENTIFIED BY '{new_user_password}'"
     )
     grant_stmt = (
-        f"GRANT ALL PRIVILEGES ON *.* TO '{new_user_name}'@'%' WITH GRANT OPTION"
+        "GRANT ALL PRIVILEGES ON *.* TO '{new_user_name}'@'{server_host}' WITH GRANT OPTION"
     )
     table_name = "employees"
     create_stmt = f"""CREATE TABLE {table_name} (
@@ -412,6 +412,7 @@ def setUp(self) -> None:
         self.provider.add_span_processor(processor)
         self.tracer = trace.get_tracer(__name__, tracer_provider=self.provider)
         self.cnx_config = self.get_clean_mysql_config()
+        self.server_host = self.cnx_config['host']
         self.cnx_config["use_pure"] = self.pure_python
 
         # Instrumentor
@@ -569,9 +570,20 @@ def _run_client_app1(
                 cur.execute(f"DROP TABLE IF EXISTS {self.table_name}")
 
                 # create a new user
-                cur.execute(f"DROP USER IF EXISTS '{self.new_user_name}'")
-                cur.execute(self.new_user_stmt)
-                cur.execute(self.grant_stmt)
+                cur.execute(f"DROP USER IF EXISTS '{self.new_user_name}'@'{self.server_host}'")
+                cur.execute(
+                    self.new_user_stmt.format(
+                        new_user_name=self.new_user_name,
+                        server_host=self.server_host,
+                        new_user_password=self.new_user_password,
+                    )
+                )
+                cur.execute(
+                    self.grant_stmt.format(
+                        new_user_name=self.new_user_name,
+                        server_host=self.server_host,
+                    )
+                )
                 cur.execute("FLUSH PRIVILEGES")
                 cur.execute(f"CREATE DATABASE IF NOT EXISTS {self.new_database}")
 

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

@@ -1,4 +1,4 @@
-# Copyright (c) 2021, 2024, Oracle and/or its affiliates.
+# Copyright (c) 2021, 2025, 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
@@ -50,14 +50,15 @@ def test_caching_sha2_password_test1(self):
         for use_pure in self.use_pure_options:
             config = self.get_clean_mysql_config()
             config["use_pure"] = use_pure
+            server_host = config["host"]
 
             with mysql.connector.connect(**config) as cnx:
-                cnx.cmd_query("DROP USER IF EXISTS 'sham'@'%'")
+                cnx.cmd_query(f"DROP USER IF EXISTS 'sham'@'{server_host}'")
                 cnx.cmd_query(
-                    "CREATE USER 'sham'@'%' IDENTIFIED "
+                    f"CREATE USER 'sham'@'{server_host}' IDENTIFIED "
                     "WITH caching_sha2_password BY 'shapass'"
                 )
-                cnx.cmd_query("GRANT ALL ON *.* TO 'sham'@'%'")
+                cnx.cmd_query(f"GRANT ALL ON *.* TO 'sham'@'{server_host}'")
 
             config["user"] = "sham"
             config["password"] = "shapass"
@@ -79,22 +80,23 @@ def test_caching_sha2_password_test1(self):
                     cur.execute("SELECT * FROM t1")
                     self.assertEqual(1, len(cur.fetchone()))
                     cur.execute("DROP TABLE IF EXISTS t1")
-                cnx.cmd_query("DROP USER IF EXISTS 'sham'@'%'")
+                cnx.cmd_query(f"DROP USER IF EXISTS 'sham'@'{server_host}'")
 
     def test_caching_sha2_password_test3(self):
         """Test full authentication with SSL after create user,
         flushing privileges, altering user, setting new password."""
         for use_pure in self.use_pure_options:
             config = self.get_clean_mysql_config()
             config["use_pure"] = use_pure
+            server_host = config['host']
 
             with mysql.connector.connect(**config) as cnx:
-                cnx.cmd_query("DROP USER IF EXISTS 'sham'@'%'")
+                cnx.cmd_query(f"DROP USER IF EXISTS 'sham'@'{server_host}'")
                 cnx.cmd_query(
-                    "CREATE USER 'sham'@'%' IDENTIFIED "
+                    f"CREATE USER 'sham'@'{server_host}' IDENTIFIED "
                     "WITH caching_sha2_password BY 'shapass'"
                 )
-                cnx.cmd_query("GRANT ALL ON *.* TO 'sham'@'%'")
+                cnx.cmd_query(f"GRANT ALL ON *.* TO 'sham'@'{server_host}'")
                 cnx.cmd_query("FLUSH PRIVILEGES")
 
             config["user"] = "sham"
@@ -117,7 +119,7 @@ def test_caching_sha2_password_test3(self):
                     cur.execute("SELECT * FROM t2")
                     self.assertEqual(1, len(cur.fetchone()))
                     cur.execute("DROP TABLE IF EXISTS t2")
-                    cur.execute("SET PASSWORD FOR 'sham'@'%'='newshapass'")
+                    cur.execute(f"SET PASSWORD FOR 'sham'@'{server_host}'='newshapass'")
                     config["password"] = "newshapass"
                     mysql.connector.connect(**config)
-                cnx.cmd_query("DROP USER IF EXISTS 'sham'@'%'")
+                cnx.cmd_query(f"DROP USER IF EXISTS 'sham'@'{server_host}'")

mysql-connector-python/tests/test_aio_authentication.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2023, 2024, Oracle and/or its affiliates.
+# Copyright (c) 2023, 2025, 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
@@ -1342,6 +1342,7 @@ class MySQLOpenIDConnectAuthPluginTests(MySQLConnectorAioTestCase):
     @classmethod
     def setUpClass(cls):
         config = tests.get_mysql_config()
+        server_host = config['host']
         plugin_ext = "dll" if os.name == "nt" else "so"
         with mysql.connector.connect(**config) as cnx:
             # Install the auth plugin `authentication_openid_connect`
@@ -1375,18 +1376,18 @@ def setUpClass(cls):
                 f"SET GLOBAL authentication_openid_connect_configuration = '{jwk}'"
             )
             # Create the user configured with OpenID connect
-            cnx.cmd_query("DROP USER IF EXISTS 'openid-test'@'%'")
+            cnx.cmd_query(f"DROP USER IF EXISTS 'openid-test'@'{server_host}'")
             cnx.cmd_query(
-                """CREATE USER 'openid-test'@'%' IDENTIFIED WITH 'authentication_openid_connect' AS
-                '{"identity_provider" : "myissuer", "user" : "mysubj"}'"""
+                f"""CREATE USER 'openid-test'@'{server_host}' IDENTIFIED WITH 'authentication_openid_connect' AS
+                '{{"identity_provider" : "myissuer", "user" : "mysubj"}}'"""
             )
-            cnx.cmd_query("GRANT ALL ON *.* TO 'openid-test'")
+            cnx.cmd_query(f"GRANT ALL ON *.* TO 'openid-test'@'{server_host}'")
 
     @classmethod
     def tearDownClass(cls):
         config = tests.get_mysql_config()
         with mysql.connector.connect(**config) as cnx:
-            cnx.cmd_query("DROP USER IF EXISTS 'openid-test'@'%'")
+            cnx.cmd_query(f"DROP USER IF EXISTS 'openid-test'@'{config['host']}'")
             try:
                 cnx.cmd_query("UNINSTALL PLUGIN authentication_openid_connect")
             except ProgrammingError:

mysql-connector-python/tests/test_authentication.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 
-# Copyright (c) 2014, 2024, Oracle and/or its affiliates.
+# Copyright (c) 2014, 2025, 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
@@ -1373,12 +1373,13 @@ class MySQLOpenIDConnectAuthPluginTests(tests.MySQLConnectorTests):
     """
 
     skip_reason = None
+    config = tests.get_mysql_config()
+    server_host = config['host']
 
     @classmethod
     def setUpClass(cls):
-        config = tests.get_mysql_config()
         plugin_ext = "dll" if os.name == "nt" else "so"
-        with mysql.connector.connect(**config) as cnx:
+        with mysql.connector.connect(**cls.config) as cnx:
             # Install the auth plugin `authentication_openid_connect`
             try:
                 cnx.cmd_query(
@@ -1410,18 +1411,17 @@ def setUpClass(cls):
                 f"SET GLOBAL authentication_openid_connect_configuration = '{jwk}'"
             )
             # Create the user configured with OpenID connect
-            cnx.cmd_query("DROP USER IF EXISTS 'openid-test'@'%'")
+            cnx.cmd_query(f"DROP USER IF EXISTS 'openid-test'@'{cls.server_host}'")
             cnx.cmd_query(
-                """CREATE USER 'openid-test'@'%' IDENTIFIED WITH 'authentication_openid_connect' AS
-                '{"identity_provider" : "myissuer", "user" : "mysubj"}'"""
+                f"""CREATE USER 'openid-test'@'{cls.server_host}' IDENTIFIED WITH 'authentication_openid_connect' AS
+                '{{"identity_provider" : "myissuer", "user" : "mysubj"}}'"""
             )
-            cnx.cmd_query("GRANT ALL ON *.* TO 'openid-test'")
+            cnx.cmd_query(f"GRANT ALL ON *.* TO 'openid-test'@'{cls.server_host}'")
 
     @classmethod
     def tearDownClass(cls):
-        config = tests.get_mysql_config()
-        with mysql.connector.connect(**config) as cnx:
-            cnx.cmd_query("DROP USER IF EXISTS 'openid-test'@'%'")
+        with mysql.connector.connect(**cls.config) as cnx:
+            cnx.cmd_query(f"DROP USER IF EXISTS 'openid-test'@'{cls.server_host}'")
             try:
                 cnx.cmd_query("UNINSTALL PLUGIN authentication_openid_connect")
             except ProgrammingError:

mysqlx-connector-python/tests/test_mysqlx_connection.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 
-# Copyright (c) 2016, 2024, Oracle and/or its affiliates.
+# Copyright (c) 2016, 2025, 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
@@ -503,6 +503,7 @@ def send_auth_ok(self):
 class MySQLxSessionTests(tests.MySQLxTests):
     def setUp(self):
         self.connect_kwargs = tests.get_mysqlx_config()
+        self.server_host = self.connect_kwargs['host']
         self.schema_name = self.connect_kwargs["schema"]
         try:
             self.session = mysqlx.get_session(self.connect_kwargs)
@@ -668,11 +669,11 @@ def test_auth(self):
         sess = mysqlx.get_session(self.connect_kwargs)
 
         try:
-            sess.sql("DROP USER 'native'@'%'").execute()
+            sess.sql(f"DROP USER 'native'@'{self.server_host}'").execute()
         except mysqlx.errors.OperationalError:
             pass
         try:
-            sess.sql("DROP USER 'sha256'@'%'").execute()
+            sess.sql(f"DROP USER 'sha256'@'{self.server_host}'").execute()
         except mysqlx.errors.OperationalError:
             pass
 
@@ -682,14 +683,14 @@ def test_auth(self):
             else "caching_sha2_password"
         )
         sess.sql(
-            f"CREATE USER 'native'@'%' IDENTIFIED WITH {native_auth} BY 'test'"
+            f"CREATE USER 'native'@'{self.server_host}' IDENTIFIED WITH {native_auth} BY 'test'"
         ).execute()
         sess.sql(
-            "CREATE USER 'sha256'@'%' IDENTIFIED WITH sha256_password BY 'sha256'"
+            f"CREATE USER 'sha256'@'{self.server_host}' IDENTIFIED WITH sha256_password BY 'sha256'"
         ).execute()
 
         config = {
-            "host": self.connect_kwargs["host"],
+            "host": self.server_host,
             "port": self.connect_kwargs["port"],
         }
 
@@ -711,8 +712,8 @@ def test_auth(self):
         config["auth"] = "mysql41"
         self.assertRaises(InterfaceError, mysqlx.get_session, config)
 
-        sess.sql("DROP USER 'native'@'%'").execute()
-        sess.sql("DROP USER 'sha256'@'%'").execute()
+        sess.sql(f"DROP USER 'native'@'{self.server_host}'").execute()
+        sess.sql(f"DROP USER 'sha256'@'{self.server_host}'").execute()
         sess.close()
 
     @unittest.skipIf(
@@ -722,7 +723,7 @@ def test_auth(self):
     def test_auth_sha265_memory(self):
         sess = mysqlx.get_session(self.connect_kwargs)
         sess.sql(
-            "CREATE USER 'caching'@'%' IDENTIFIED WITH "
+            f"CREATE USER 'caching'@'{self.server_host}' IDENTIFIED WITH "
             "caching_sha2_password BY 'caching'"
         ).execute()
         config = {
@@ -750,7 +751,7 @@ def test_auth_sha265_memory(self):
         config["auth"] = mysqlx.Auth.SHA256_MEMORY
         mysqlx.get_session(config)
 
-        sess.sql("DROP USER 'caching'@'%'").execute()
+        sess.sql(f"DROP USER 'caching'@'{self.server_host}'").execute()
         sess.close()
 
     @unittest.skipIf(
@@ -1061,9 +1062,9 @@ def test_get_default_schema(self):
 
         # Test BUG#28942938: 'ACCESS DENIED' error for unauthorized user tries
         # to use the default schema if not exists at get_session
-        self.session.sql("DROP USER IF EXISTS 'def_schema'@'%'").execute()
+        self.session.sql(f"DROP USER IF EXISTS 'def_schema'@'{self.server_host}'").execute()
         self.session.sql(
-            "CREATE USER 'def_schema'@'%' IDENTIFIED WITH "
+            f"CREATE USER 'def_schema'@'{self.server_host}' IDENTIFIED WITH "
             "caching_sha2_password BY 'test'"
         ).execute()
         settings = self.connect_kwargs.copy()
@@ -1078,7 +1079,7 @@ def test_get_default_schema(self):
 
         # Grant privilege to one unrelated schema
         self.session.sql(
-            "GRANT ALL PRIVILEGES ON nonexistent.* TO 'def_schema'@'%'"
+            f"GRANT ALL PRIVILEGES ON nonexistent.* TO 'def_schema'@'{self.server_host}'"
         ).execute()
         with self.assertRaises(InterfaceError) as context:
             _ = mysqlx.get_session(settings)
@@ -2604,16 +2605,19 @@ class WL14852(tests.MySQLxTests):
     """WL#14852: Align TLS and SSL options checking and behavior"""
 
     def setUp(self):
-        session = mysqlx.get_session(tests.get_mysqlx_config())
+        config = tests.get_mysqlx_config()
+        session = mysqlx.get_session(config)
+        self.server_host = config['host']
+
         session.sql(
-            "CREATE USER 'native'@'%' IDENTIFIED WITH "
+            f"CREATE USER 'native'@'{self.server_host}' IDENTIFIED WITH "
             "mysql_native_password BY 'test'"
         ).execute()
-        session.sql("grant all privileges on *.* to 'native'@'%'").execute()
+        session.sql(f"grant all privileges on *.* to 'native'@'{self.server_host}'").execute()
 
     def tearDown(self):
         session = mysqlx.get_session(tests.get_mysqlx_config())
-        session.sql("DROP USER IF EXISTS 'native'@'%'").execute()
+        session.sql(f"DROP USER IF EXISTS 'native'@'{self.server_host}'").execute()
 
     def _test_giving_ssl_disable_does_not_raise_error(self, use_uri=False):
         """Verify no error with ssl-mode=disable and other TLS or SSL options."""