feat: enforce IAM authentication for Cloud SQL connections by removing password handling

Author: jirhiker

alembic/env.py

@@ -120,7 +120,6 @@ 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)
         ip_type = os.environ.get("CLOUD_SQL_IP_TYPE", "public")
@@ -147,10 +146,11 @@ def getconn():
                 "ip_type": ip_type,
                 "enable_iam_auth": use_iam_auth,
             }
-            if use_iam_auth:
-                connect_kwargs["password"] = get_iam_login_token()
-            else:
-                connect_kwargs["password"] = password
+            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()
             return connector.connect(
                 instance_name,
                 "pg8000",

db/engine.py

@@ -69,7 +69,6 @@ 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)
         ip_type = os.environ.get("CLOUD_SQL_IP_TYPE", "public")
@@ -80,10 +79,11 @@ def asyncify_connection():
             "enable_iam_auth": use_iam_auth,
             "ip_type": ip_type,
         }
-        if use_iam_auth:
-            connect_kwargs["password"] = get_iam_login_token()
-        else:
-            connect_kwargs["password"] = password
+        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()
 
         connection = connector.connect_async(instance_name, "asyncpg", **connect_kwargs)
 
@@ -106,7 +106,6 @@ 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)
         ip_type = os.environ.get("CLOUD_SQL_IP_TYPE", "public")
@@ -118,10 +117,11 @@ def getconn():
                 "ip_type": ip_type,
                 "enable_iam_auth": use_iam_auth,
             }
-            if use_iam_auth:
-                connect_kwargs["password"] = get_iam_login_token()
-            else:
-                connect_kwargs["password"] = password
+            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()
 
             conn = connector.connect(
                 instance_name,  # The Cloud SQL instance name