fix: throw error if Auto IAM AuthN is unsupported

google/cloud/sql/connector/instance.py

@@ -100,6 +100,15 @@ def __init__(self, *args: Any) -> None:
         super(ExpiredInstanceMetadata, self).__init__(self, *args)
 
 
+class AutoIAMAuthNotSupported(Exception):
+    """
+    Exception to be raised when Automatic IAM Authentication is not
+    supported with database engine version.
+    """
+
+    pass
+
+
 class InstanceMetadata:
     ip_addrs: Dict[str, Any]
     context: ssl.SSLContext
@@ -115,7 +124,6 @@ def __init__(
         enable_iam_auth: bool,
     ) -> None:
         self.ip_addrs = ip_addrs
-
         self.context = ssl.SSLContext(ssl.PROTOCOL_TLS)
 
         # verify OpenSSL version supports TLSv1.3
@@ -366,10 +374,21 @@ async def _perform_refresh(self) -> InstanceMetadata:
                     self._enable_iam_auth,
                 )
             )
-
-            metadata, ephemeral_cert = await asyncio.gather(
-                metadata_task, ephemeral_task
-            )
+            try:
+                metadata = await metadata_task
+                # check if automatic IAM database authn is supported for database engine
+                if self._enable_iam_auth and not metadata[
+                    "database_version"
+                ].startswith("POSTGRES"):
+                    raise AutoIAMAuthNotSupported(
+                        f"'{metadata['database_version']}' does not support automatic IAM authentication. It is only supported with Cloud SQL Postgres instances."
+                    )
+            except Exception:
+                # cancel ephemeral cert task if exception occurs before it is awaited
+                ephemeral_task.cancel()
+                raise
+
+            ephemeral_cert = await ephemeral_task
 
             x509 = load_pem_x509_certificate(
                 ephemeral_cert.encode("UTF-8"), default_backend()

google/cloud/sql/connector/refresh_utils.py

@@ -105,6 +105,7 @@ async def _get_metadata(
     metadata = {
         "ip_addresses": {ip["type"]: ip["ipAddress"] for ip in ret_dict["ipAddresses"]},
         "server_ca_cert": ret_dict["serverCaCert"]["cert"],
+        "database_version": ret_dict["databaseVersion"],
     }
 
     return metadata

tests/system/test_connector_object.py

@@ -15,11 +15,13 @@
 """
 import asyncio
 import os
+import pytest
 import pymysql
 import sqlalchemy
 import logging
 import google.auth
 from google.cloud.sql.connector import Connector
+from google.cloud.sql.connector.instance import AutoIAMAuthNotSupported
 import datetime
 import concurrent.futures
 from threading import Thread
@@ -141,3 +143,34 @@ def test_connector_with_custom_loop() -> None:
         assert result[0] == 1
         # assert that Connector does not start its own thread
         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
+    for SQL Server raises exception.
+    """
+    with pytest.raises(AutoIAMAuthNotSupported):
+        with Connector(enable_iam_auth=True) as connector:
+            connector.connect(
+                os.environ["SQLSERVER_CONNECTION_NAME"],
+                "pytds",
+                user="my-user",
+                password="my-pass",
+                db="my-db",
+            )