fix: throw error if Auto IAM AuthN is unsupported

google/cloud/sql/connector/connector.py

@@ -234,6 +234,7 @@ async def connect_async(
         # helper function to wrap in timeout
         async def get_connection() -> Any:
             instance_data, ip_address = await instance.connect_info(ip_type)
+
             # async drivers are unblocking and can be awaited directly
             if driver in ASYNC_DRIVERS:
                 return await connector(ip_address, instance_data.context, **kwargs)

google/cloud/sql/connector/instance.py

@@ -22,7 +22,7 @@
     _seconds_until_refresh,
     _is_valid,
 )
-from google.cloud.sql.connector.utils import write_to_file
+from google.cloud.sql.connector.utils import verify_iam_auth, write_to_file
 from google.cloud.sql.connector.version import __version__ as version
 
 # Importing libraries
@@ -115,7 +115,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
@@ -356,10 +355,20 @@ async def _perform_refresh(self) -> InstanceMetadata:
                 )
             )
 
+            # gather Admin API tasks, return_exceptions=True to check auto iam authn
             metadata, ephemeral_cert = await asyncio.gather(
-                metadata_task, ephemeral_task
+                metadata_task, ephemeral_task, return_exceptions=True
             )
 
+            if isinstance(metadata, BaseException):
+                raise metadata
+
+            # check if automatic IAM database authn is supported for database engine
+            verify_iam_auth(metadata["database_version"], self._enable_iam_auth)
+
+            if isinstance(ephemeral_cert, BaseException):
+                raise ephemeral_cert
+
             x509 = load_pem_x509_certificate(
                 ephemeral_cert.encode("UTF-8"), default_backend()
             )

google/cloud/sql/connector/refresh_utils.py

@@ -101,6 +101,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

google/cloud/sql/connector/utils.py

@@ -77,3 +77,35 @@ def write_to_file(
         priv_out.write(priv_key)
 
     return (ca_filename, cert_filename, key_filename)
+
+
+class AutoIAMAuthNotSupported(Exception):
+    """
+    Exception to be raised when Automatic IAM Authentication is not
+    supported with database engine version.
+    """
+
+    pass
+
+
+def verify_iam_auth(database_version: str, enable_iam_auth: bool) -> None:
+    """
+    Check if database engine supports automatic IAM authentication. (Postgres only)
+
+    :type database_version: str
+    :param database_version
+        Cloud SQL database version. (i.e. POSTGRES_14, MYSQL_8_0, etc.)
+
+    :type enable_iam_auth: bool
+    :param enable_iam_auth
+        Enables automatic IAM database authentication.
+    """
+    # don't check engine version if IAM AuthN isn't enabled
+    if not enable_iam_auth:
+        return
+    if database_version.startswith("POSTGRES"):
+        pass
+    else:
+        raise AutoIAMAuthNotSupported(
+            f"'{database_version}' does not support automatic IAM authentication. It is only supported with Cloud SQL Postgres instances."
+        )

tests/system/test_connector_object.py

@@ -15,6 +15,7 @@
 """
 import asyncio
 import os
+import pytest
 import pymysql
 import sqlalchemy
 import logging
@@ -24,6 +25,8 @@
 import concurrent.futures
 from threading import Thread
 
+from google.cloud.sql.connector.utils import AutoIAMAuthNotSupported
+
 
 def init_connection_engine(
     custom_connector: Connector,
@@ -141,3 +144,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",
+            )

tests/unit/test_utils.py

@@ -39,3 +39,46 @@ async def test_generate_keys_returns_bytes_and_str() -> None:
 
     res1, res2 = await utils.generate_keys()
     assert isinstance(res1, bytes) and (isinstance(res2, str))
+
+
+def test_verify_iam_auth_mysql_AutoIAMAuthNotSupported() -> None:
+    """
+    Test that verify_iam_auth throws exception for MySQL
+    with enable_iam_auth set to True.
+    """
+    with pytest.raises(utils.AutoIAMAuthNotSupported):
+        utils.verify_iam_auth("MYSQL_8_0", True)
+
+
+def test_verify_iam_auth_mysql_pass() -> None:
+    """
+    Test that verify_iam_auth passes for MySQL
+    with enable_iam_auth set to False.
+    """
+    utils.verify_iam_auth("MYSQL_8_0", False)
+
+
+def test_verify_iam_auth_sqlserver_AutoIAMAuthNotSupported() -> None:
+    """
+    Test that verify_iam_auth throws exception for SQL Server
+    with enable_iam_auth set to True.
+    """
+    with pytest.raises(utils.AutoIAMAuthNotSupported):
+        utils.verify_iam_auth("SQLSERVER_2019_STANDARD", True)
+
+
+def test_verify_iam_auth_sqlserver_pass() -> None:
+    """
+    Test that verify_iam_auth passes for SQL Server
+    with enable_iam_auth set to False.
+    """
+    utils.verify_iam_auth("SQLSERVER_2019_STANDARD", False)
+
+
+def test_verify_iam_auth_postgres_pass() -> None:
+    """
+    Test that verify_iam_auth passes for Postgres
+    with enable_iam_auth set to False or True.
+    """
+    utils.verify_iam_auth("POSTGRES_14", False)
+    utils.verify_iam_auth("POSTGRES_14", True)