feat: add support for MySQL auto IAM AuthN (#466)

.github/workflows/tests.yml

@@ -83,7 +83,9 @@ jobs:
         with:
           secrets: |-
             MYSQL_CONNECTION_NAME:${{ secrets.GOOGLE_CLOUD_PROJECT }}/MYSQL_CONNECTION_NAME
+            MYSQL_IAM_CONNECTION_NAME:${{ secrets.GOOGLE_CLOUD_PROJECT }}/MYSQL_IAM_CONNECTION_NAME
             MYSQL_USER:${{ secrets.GOOGLE_CLOUD_PROJECT }}/MYSQL_USER
+            MYSQL_IAM_USER:${{ secrets.GOOGLE_CLOUD_PROJECT }}/MYSQL_USER_IAM_PYTHON
             MYSQL_PASS:${{ secrets.GOOGLE_CLOUD_PROJECT }}/MYSQL_PASS
             MYSQL_DB:${{ secrets.GOOGLE_CLOUD_PROJECT }}/MYSQL_DB
             POSTGRES_CONNECTION_NAME:${{ secrets.GOOGLE_CLOUD_PROJECT }}/POSTGRES_CONNECTION_NAME
@@ -100,7 +102,9 @@ jobs:
       - name: Run tests
         env:
           MYSQL_CONNECTION_NAME: '${{ steps.secrets.outputs.MYSQL_CONNECTION_NAME }}'
+          MYSQL_IAM_CONNECTION_NAME: '${{ steps.secrets.outputs.MYSQL_IAM_CONNECTION_NAME }}'
           MYSQL_USER: '${{ steps.secrets.outputs.MYSQL_USER }}'
+          MYSQL_IAM_USER: '${{ steps.secrets.outputs.MYSQL_IAM_USER }}'
           MYSQL_PASS: '${{ steps.secrets.outputs.MYSQL_PASS }}'
           MYSQL_DB: '${{ steps.secrets.outputs.MYSQL_DB }}'
           POSTGRES_CONNECTION_NAME: '${{ steps.secrets.outputs.POSTGRES_CONNECTION_NAME }}'

README.md

@@ -242,10 +242,16 @@ connector.connect(
 Note: If specifying Private IP, your application must already be in the same VPC network as your Cloud SQL Instance.
 
 ### IAM Authentication
-Connections using [Automatic IAM database authentication](https://cloud.google.com/sql/docs/postgres/authentication#automatic) are supported when using the Postgres driver. This feature is unsupported for other drivers. If automatic IAM authentication is not supported for your driver, you can use [Manual IAM database authentication](https://cloud.google.com/sql/docs/postgres/authentication#manual) to connect.
-First, make sure to [configure your Cloud SQL Instance to allow IAM authentication](https://cloud.google.com/sql/docs/postgres/create-edit-iam-instances#configure-iam-db-instance) and [add an IAM database user](https://cloud.google.com/sql/docs/postgres/create-manage-iam-users#creating-a-database-user).
+Connections using [Automatic IAM database authentication](https://cloud.google.com/sql/docs/postgres/authentication#automatic) are supported when using Postgres or MySQL drivers.
+First, make sure to [configure your Cloud SQL Instance to allow IAM authentication](https://cloud.google.com/sql/docs/postgres/create-edit-iam-instances#configure-iam-db-instance)
+and [add an IAM database user](https://cloud.google.com/sql/docs/postgres/create-manage-iam-users#creating-a-database-user).
+
 Now, you can connect using user or service account credentials instead of a password.
-In the call to connect, set the `enable_iam_auth` keyword argument to true and `user` to the email address associated with your IAM user.
+In the call to connect, set the `enable_iam_auth` keyword argument to true and the `user` argument to the appropriately formatted IAM principal.
+> Postgres: For an IAM user account, this is the user's email address. For a service account, it is the service account's email without the `.gserviceaccount.com` domain suffix.
+
+> MySQL: For an IAM user account, this is the user's email address, without the @ or domain name. For example, for `test-user@gmail.com`, set the `user` argument to `test-user`. For a service account, this is the service account's email address without the `@project-id.iam.gserviceaccount.com` suffix.
+
 Example:
 ```python
 connector.connect(

google/cloud/sql/connector/connector.py

@@ -46,7 +46,8 @@ class Connector:
 
     :type enable_iam_auth: bool
     :param enable_iam_auth
-        Enables IAM based authentication (Postgres only).
+        Enables automatic IAM database authentication for Postgres or MySQL
+        instances.
 
     :type timeout: int
     :param timeout
@@ -319,7 +320,8 @@ async def create_async_connector(
 
     :type enable_iam_auth: bool
     :param enable_iam_auth
-        Enables IAM based authentication (Postgres only).
+        Enables automatic IAM database authentication for Postgres or MySQL
+        instances.
 
     :type timeout: int
     :param timeout

google/cloud/sql/connector/exceptions.py

@@ -59,7 +59,7 @@ class CredentialsTypeError(Exception):
 class AutoIAMAuthNotSupported(Exception):
     """
     Exception to be raised when Automatic IAM Authentication is not
-    supported with database engine version.f
+    supported with database engine version.
     """
 
     pass

google/cloud/sql/connector/instance.py

@@ -143,7 +143,8 @@ class Instance:
         If not specified, Application Default Credentials are used.
 
     :param enable_iam_auth
-        Enables IAM based authentication for Postgres instances.
+        Enables automatic IAM database authentication for Postgres or MySQL
+        instances.
     :type enable_iam_auth: bool
 
     :param loop:
@@ -328,9 +329,9 @@ async def _perform_refresh(self) -> InstanceMetadata:
                 # check if automatic IAM database authn is supported for database engine
                 if self._enable_iam_auth and not metadata[
                     "database_version"
-                ].startswith("POSTGRES"):
+                ].startswith(("POSTGRES", "MYSQL")):
                     raise AutoIAMAuthNotSupported(
-                        f"'{metadata['database_version']}' does not support automatic IAM authentication. It is only supported with Cloud SQL Postgres instances."
+                        f"'{metadata['database_version']}' does not support automatic IAM authentication. It is only supported with Cloud SQL Postgres or MySQL instances."
                     )
             except Exception:
                 # cancel ephemeral cert task if exception occurs before it is awaited

google/cloud/sql/connector/pymysql.py

@@ -46,6 +46,9 @@ def connect(
             'Unable to import module "pymysql." Please install and try again.'
         )
 
+    # allow automatic IAM database authentication to not require password
+    kwargs["password"] = kwargs["password"] if "password" in kwargs else None
+
     # Create socket and wrap with context.
     sock = ctx.wrap_socket(
         socket.create_connection((ip_address, SERVER_PROXY_PORT)),

google/cloud/sql/connector/refresh_utils.py

@@ -144,7 +144,8 @@ async def _get_ephemeral(
 
     :type enable_iam_auth: bool
     :param enable_iam_auth
-        Enables IAM based authentication for Postgres instances.
+        Enables automatic IAM database authentication for Postgres or MySQL
+        instances.
 
     :rtype: str
     :returns: An ephemeral certificate from the Cloud SQL instance that allows

tests/system/test_connector_object.py

@@ -145,21 +145,6 @@ def test_connector_with_custom_loop() -> None:
         assert connector._thread is None
 
 
-def test_connector_mysql_iam_auth_error() -> None:
-    """
-    Test that connecting with enable_iam_auth set to True
-    for MySQL raises exception.
-    """
-    with pytest.raises(AutoIAMAuthNotSupported):
-        with Connector(enable_iam_auth=True) as connector:
-            connector.connect(
-                os.environ["MYSQL_CONNECTION_NAME"],
-                "pymysql",
-                user="my-user",
-                db="my-db",
-            )
-
-
 def test_connector_sqlserver_iam_auth_error() -> None:
     """
     Test that connecting with enable_iam_auth set to True

tests/system/test_pymysql_iam_auth.py

@@ -0,0 +1,86 @@
+"""
+Copyright 2022 Google LLC
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+  https://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+import os
+import uuid
+from typing import Generator
+
+import pytest
+import pymysql
+import sqlalchemy
+from google.cloud.sql.connector import Connector
+
+table_name = f"books_{uuid.uuid4().hex}"
+
+
+# [START cloud_sql_connector_mysql_pymysql_iam_auth]
+# The Cloud SQL Python Connector can be used along with SQLAlchemy using the
+# 'creator' argument to 'create_engine'
+def init_connection_engine() -> sqlalchemy.engine.Engine:
+    def getconn() -> pymysql.connections.Connection:
+        # initialize Connector object for connections to Cloud SQL
+        with Connector() as connector:
+            conn: pymysql.connections.Connection = connector.connect(
+                os.environ["MYSQL_IAM_CONNECTION_NAME"],
+                "pymysql",
+                user=os.environ["MYSQL_IAM_USER"],
+                db=os.environ["MYSQL_DB"],
+                enable_iam_auth=True,
+            )
+            return conn
+
+    # create SQLAlchemy connection pool
+    pool = sqlalchemy.create_engine(
+        "mysql+pymysql://",
+        creator=getconn,
+    )
+    return pool
+
+
+# [END cloud_sql_connector_mysql_pymysql_iam_auth]
+
+
+@pytest.fixture(name="pool")
+def setup() -> Generator:
+    pool = init_connection_engine()
+
+    with pool.connect() as conn:
+        conn.execute(
+            f"CREATE TABLE IF NOT EXISTS {table_name}"
+            " ( id CHAR(20) NOT NULL, title TEXT NOT NULL );"
+        )
+
+    yield pool
+
+    with pool.connect() as conn:
+        conn.execute(f"DROP TABLE IF EXISTS {table_name}")
+
+
+def test_pooled_connection_with_pymysql_iam_auth(
+    pool: sqlalchemy.engine.Engine,
+) -> None:
+    insert_stmt = sqlalchemy.text(
+        f"INSERT INTO {table_name} (id, title) VALUES (:id, :title)",
+    )
+    with pool.connect() as conn:
+        conn.execute(insert_stmt, id="book1", title="Book One")
+        conn.execute(insert_stmt, id="book2", title="Book Two")
+
+    select_stmt = sqlalchemy.text(f"SELECT title FROM {table_name} ORDER BY ID;")
+    with pool.connect() as conn:
+        rows = conn.execute(select_stmt).fetchall()
+        titles = [row[0] for row in rows]
+
+    assert titles == ["Book One", "Book Two"]

tests/unit/test_instance.py

@@ -414,7 +414,6 @@ async def test_ClientResponseError(
     "mock_instance",
     [
         mocks.FakeCSQLInstance(db_version="SQLSERVER_2019_STANDARD"),
-        mocks.FakeCSQLInstance(db_version="MYSQL_8_0"),
     ],
 )
 async def test_AutoIAMAuthNotSupportedError(instance: Instance) -> None: