feat: disable default IAM authentication for Cloud SQL connections and allow password handling

Author: jirhiker

alembic/env.py

@@ -59,7 +59,7 @@ def build_database_url():
         user = os.environ.get("CLOUD_SQL_USER", "")
         password = os.environ.get("CLOUD_SQL_PASSWORD", "")
         database = os.environ.get("CLOUD_SQL_DATABASE", "")
-        use_iam_auth = get_bool_env("CLOUD_SQL_IAM_AUTH", True)
+        use_iam_auth = get_bool_env("CLOUD_SQL_IAM_AUTH", False)
         # Host is provided by connector, so leave blank.
         if use_iam_auth:
             return f"postgresql+pg8000://{user}@/{database}"
@@ -120,8 +120,9 @@ def run_migrations_online() -> None:
 
         instance_name = os.environ.get("CLOUD_SQL_INSTANCE_NAME")
         user = os.environ.get("CLOUD_SQL_USER")
+        password = os.environ.get("CLOUD_SQL_PASSWORD")
         database = os.environ.get("CLOUD_SQL_DATABASE")
-        use_iam_auth = get_bool_env("CLOUD_SQL_IAM_AUTH", True)
+        use_iam_auth = get_bool_env("CLOUD_SQL_IAM_AUTH", False)
         ip_type = os.environ.get("CLOUD_SQL_IP_TYPE", "public")
 
         connector = Connector()
@@ -146,11 +147,10 @@ def getconn():
                 "ip_type": ip_type,
                 "enable_iam_auth": use_iam_auth,
             }
-            if not use_iam_auth:
-                raise RuntimeError(
-                    "CLOUD_SQL_IAM_AUTH must be true when DB_DRIVER=cloudsql."
-                )
-            connect_kwargs["password"] = get_iam_login_token()
+            if use_iam_auth:
+                connect_kwargs["password"] = get_iam_login_token()
+            else:
+                connect_kwargs["password"] = password
             return connector.connect(
                 instance_name,
                 "pg8000",

db/engine.py

@@ -69,8 +69,9 @@ def asyncify_connection():
 
         instance_name = os.environ.get("CLOUD_SQL_INSTANCE_NAME")
         user = os.environ.get("CLOUD_SQL_USER")
+        password = os.environ.get("CLOUD_SQL_PASSWORD")
         database = os.environ.get("CLOUD_SQL_DATABASE")
-        use_iam_auth = get_bool_env("CLOUD_SQL_IAM_AUTH", True)
+        use_iam_auth = get_bool_env("CLOUD_SQL_IAM_AUTH", False)
         ip_type = os.environ.get("CLOUD_SQL_IP_TYPE", "public")
 
         connect_kwargs = {
@@ -79,11 +80,10 @@ def asyncify_connection():
             "enable_iam_auth": use_iam_auth,
             "ip_type": ip_type,
         }
-        if not use_iam_auth:
-            raise RuntimeError(
-                "CLOUD_SQL_IAM_AUTH must be true when DB_DRIVER=cloudsql."
-            )
-        connect_kwargs["password"] = get_iam_login_token()
+        if use_iam_auth:
+            connect_kwargs["password"] = get_iam_login_token()
+        else:
+            connect_kwargs["password"] = password
 
         connection = connector.connect_async(instance_name, "asyncpg", **connect_kwargs)
 
@@ -106,8 +106,9 @@ def asyncify_connection():
     def init_connection_pool(connector):
         instance_name = os.environ.get("CLOUD_SQL_INSTANCE_NAME")
         user = os.environ.get("CLOUD_SQL_USER")
+        password = os.environ.get("CLOUD_SQL_PASSWORD")
         database = os.environ.get("CLOUD_SQL_DATABASE")
-        use_iam_auth = get_bool_env("CLOUD_SQL_IAM_AUTH", True)
+        use_iam_auth = get_bool_env("CLOUD_SQL_IAM_AUTH", False)
         ip_type = os.environ.get("CLOUD_SQL_IP_TYPE", "public")
 
         def getconn():
@@ -117,11 +118,10 @@ def getconn():
                 "ip_type": ip_type,
                 "enable_iam_auth": use_iam_auth,
             }
-            if not use_iam_auth:
-                raise RuntimeError(
-                    "CLOUD_SQL_IAM_AUTH must be true when DB_DRIVER=cloudsql."
-                )
-            connect_kwargs["password"] = get_iam_login_token()
+            if use_iam_auth:
+                connect_kwargs["password"] = get_iam_login_token()
+            else:
+                connect_kwargs["password"] = password
 
             conn = connector.connect(
                 instance_name,  # The Cloud SQL instance name

transfers/transfer.py

@@ -47,10 +47,6 @@
 ):
     os.environ["DB_DRIVER"] = "cloudsql"
 
-# Cloud SQL should use IAM auth by default unless explicitly disabled.
-if (os.getenv("DB_DRIVER") or "").strip().lower() == "cloudsql":
-    os.environ.setdefault("CLOUD_SQL_IAM_AUTH", "true")
-
 from alembic import command
 from alembic.config import Config