feat: add IAM AuthN to Cloud SQL Sources (#414)

Add IAM support for Cloud SQL source connection using Go language
connector:
https://pkg.go.dev/cloud.google.com/go/cloudsqlconn#section-readme

docs/en/resources/sources/cloud-sql-pg.md

@@ -42,6 +42,11 @@ scope](https://cloud.google.com/compute/docs/access/service-accounts#accesscopes
 to connect using the Cloud SQL Admin API.
 {{< /notice >}}
 
+To connect to your Cloud SQL Source using IAM authentication:
+
+1. Specify your IAM email as the `user` or leave it blank for Toolbox to fetch from ADC.
+2. Leave the `password` field blank.
+
 [csql-go-conn]: https://github.com/GoogleCloudPlatform/cloud-sql-go-connector
 [adc]: https://cloud.google.com/docs/authentication#adc
 [set-adc]: https://cloud.google.com/docs/authentication/provide-credentials-adc
@@ -93,6 +98,6 @@ sources:
 | region    |  string  |     true     | Name of the GCP region that the cluster was created in (e.g. "us-central1").                |
 | instance  |  string  |     true     | Name of the Cloud SQL instance within the cluster (e.g. "my-instance").                     |
 | database  |  string  |     true     | Name of the Postgres database to connect to (e.g. "my_db").                                 |
-| user      |  string  |     true     | Name of the Postgres user to connect as (e.g. "my-pg-user").                                |
-| password  |  string  |     true     | Password of the Postgres user (e.g. "my-password").                                         |
+| user      |  string  |     false     | Name of the Postgres user to connect as (e.g. "my-pg-user"). Defaults to IAM auth using [ADC][adc] email if unspecified.                               |
+| password  |  string  |     false     | Password of the Postgres user (e.g. "my-password"). Defaults to attempting IAM authentication if unspecified.                                        |
 | ipType    |  string  |    false     | IP Type of the Cloud SQL instance; must be one of `public` or `private`. Default: `public`. |

internal/sources/cloudsqlmssql/cloud_sql_mssql.go

@@ -103,7 +103,7 @@ func initCloudSQLMssqlConnection(ctx context.Context, tracer trace.Tracer, name,
 	if err != nil {
 		return nil, err
 	}
-	opts, err := sources.GetCloudSQLOpts(ipType, userAgent)
+	opts, err := sources.GetCloudSQLOpts(ipType, userAgent, false)
 	if err != nil {
 		return nil, err
 	}

internal/sources/cloudsqlmysql/cloud_sql_mysql.go

@@ -92,7 +92,7 @@ func initCloudSQLMySQLConnectionPool(ctx context.Context, tracer trace.Tracer, n
 	if err != nil {
 		return nil, err
 	}
-	opts, err := sources.GetCloudSQLOpts(ipType, userAgent)
+	opts, err := sources.GetCloudSQLOpts(ipType, userAgent, false)
 	if err != nil {
 		return nil, err
 	}

internal/sources/cloudsqlpg/cloud_sql_pg.go

@@ -38,9 +38,9 @@ type Config struct {
 	Region   string         `yaml:"region" validate:"required"`
 	Instance string         `yaml:"instance" validate:"required"`
 	IPType   sources.IPType `yaml:"ipType" validate:"required"`
-	User     string         `yaml:"user" validate:"required"`
-	Password string         `yaml:"password" validate:"required"`
 	Database string         `yaml:"database" validate:"required"`
+	User     string         `yaml:"user"`
+	Password string         `yaml:"password"`
 }
 
 func (r Config) SourceConfigKind() string {
@@ -82,13 +82,46 @@ func (s *Source) PostgresPool() *pgxpool.Pool {
 	return s.Pool
 }
 
+func getConnectionConfig(ctx context.Context, user, pass, dbname string) (string, bool, error) {
+	useIAM := true
+
+	// If username and password both provided, use password authentication
+	if user != "" && pass != "" {
+		dsn := fmt.Sprintf("user=%s password=%s dbname=%s sslmode=disable", user, pass, dbname)
+		useIAM = false
+		return dsn, useIAM, nil
+	}
+
+	// If username is empty, fetch email from ADC
+	// otherwise, use username as IAM email
+	if user == "" {
+		if pass != "" {
+			// If password is provided without an username, raise an error
+			return "", useIAM, fmt.Errorf("password is provided without a username. Please provide both a username and password, or leave both fields empty")
+		}
+		email, err := sources.GetIAMPrincipalEmailFromADC(ctx)
+		if err != nil {
+			return "", useIAM, fmt.Errorf("error getting email from ADC: %v", err)
+		}
+		user = email
+	}
+
+	// Construct IAM connection string with username
+	dsn := fmt.Sprintf("user=%s dbname=%s sslmode=disable", user, dbname)
+	return dsn, useIAM, nil
+}
+
 func initCloudSQLPgConnectionPool(ctx context.Context, tracer trace.Tracer, name, project, region, instance, ipType, user, pass, dbname string) (*pgxpool.Pool, error) {
 	//nolint:all // Reassigned ctx
 	ctx, span := sources.InitConnectionSpan(ctx, tracer, SourceKind, name)
 	defer span.End()
 
 	// Configure the driver to connect to the database
-	dsn := fmt.Sprintf("user=%s password=%s dbname=%s sslmode=disable", user, pass, dbname)
+	dsn, useIAM, err := getConnectionConfig(ctx, user, pass, dbname)
+	if err != nil {
+		return nil, fmt.Errorf("unable to get Cloud SQL connection config: %w", err)
+	}
+
 	config, err := pgxpool.ParseConfig(dsn)
 	if err != nil {
 		return nil, fmt.Errorf("unable to parse connection uri: %w", err)
@@ -99,7 +132,7 @@ func initCloudSQLPgConnectionPool(ctx context.Context, tracer trace.Tracer, name
 	if err != nil {
 		return nil, err
 	}
-	opts, err := sources.GetCloudSQLOpts(ipType, userAgent)
+	opts, err := sources.GetCloudSQLOpts(ipType, userAgent, useIAM)
 	if err != nil {
 		return nil, err
 	}

internal/sources/util.go

@@ -28,7 +28,7 @@ import (
 
 // GetCloudSQLDialOpts retrieve dial options with the right ip type and user agent for cloud sql
 // databases.
-func GetCloudSQLOpts(ipType, userAgent string) ([]cloudsqlconn.Option, error) {
+func GetCloudSQLOpts(ipType, userAgent string, useIAM bool) ([]cloudsqlconn.Option, error) {
 	opts := []cloudsqlconn.Option{cloudsqlconn.WithUserAgent(userAgent)}
 	switch strings.ToLower(ipType) {
 	case "private":
@@ -38,6 +38,10 @@ func GetCloudSQLOpts(ipType, userAgent string) ([]cloudsqlconn.Option, error) {
 	default:
 		return nil, fmt.Errorf("invalid ipType %s", ipType)
 	}
+
+	if useIAM {
+		opts = append(opts, cloudsqlconn.WithIAMAuthN())
+	}
 	return opts, nil
 }
 

tests/alloydb_pg_integration_test.go

@@ -255,9 +255,13 @@ func TestAlloyDBIAMConnection(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			err := RunSourceConnectionTest(t, tc.sourceConfig, ALLOYDB_POSTGRES_TOOL_KIND)
 			if err != nil {
-				if !tc.isErr {
-					t.Fatalf("Connection test failure: %s", err)
+				if tc.isErr {
+					return
 				}
+				t.Fatalf("Connection test failure: %s", err)
+			}
+			if tc.isErr {
+				t.Fatalf("Expected error but test passed.")
 			}
 		})
 	}

tests/cloud_sql_pg_integration_test.go

@@ -181,3 +181,70 @@ func TestCloudSQLPgIpConnection(t *testing.T) {
 		})
 	}
 }
+
+func TestCloudSQLIAMConnection(t *testing.T) {
+	getCloudSQLPgVars(t)
+	// service account email used for IAM should trim the suffix
+	serviceAccountEmail := strings.TrimSuffix(SERVICE_ACCOUNT_EMAIL, ".gserviceaccount.com")
+
+	noPassSourceConfig := map[string]any{
+		"kind":     CLOUD_SQL_POSTGRES_SOURCE_KIND,
+		"project":  CLOUD_SQL_POSTGRES_PROJECT,
+		"instance": CLOUD_SQL_POSTGRES_INSTANCE,
+		"region":   CLOUD_SQL_POSTGRES_REGION,
+		"database": CLOUD_SQL_POSTGRES_DATABASE,
+		"user":     serviceAccountEmail,
+	}
+
+	noUserSourceConfig := map[string]any{
+		"kind":     CLOUD_SQL_POSTGRES_SOURCE_KIND,
+		"project":  CLOUD_SQL_POSTGRES_PROJECT,
+		"instance": CLOUD_SQL_POSTGRES_INSTANCE,
+		"region":   CLOUD_SQL_POSTGRES_REGION,
+		"database": CLOUD_SQL_POSTGRES_DATABASE,
+		"password": "random",
+	}
+
+	noUserNoPassSourceConfig := map[string]any{
+		"kind":     CLOUD_SQL_POSTGRES_SOURCE_KIND,
+		"project":  CLOUD_SQL_POSTGRES_PROJECT,
+		"instance": CLOUD_SQL_POSTGRES_INSTANCE,
+		"region":   CLOUD_SQL_POSTGRES_REGION,
+		"database": CLOUD_SQL_POSTGRES_DATABASE,
+	}
+	tcs := []struct {
+		name         string
+		sourceConfig map[string]any
+		isErr        bool
+	}{
+		{
+			name:         "no user no pass",
+			sourceConfig: noUserNoPassSourceConfig,
+			isErr:        false,
+		},
+		{
+			name:         "no password",
+			sourceConfig: noPassSourceConfig,
+			isErr:        false,
+		},
+		{
+			name:         "no user",
+			sourceConfig: noUserSourceConfig,
+			isErr:        true,
+		},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			err := RunSourceConnectionTest(t, tc.sourceConfig, CLOUD_SQL_POSTGRES_TOOL_KIND)
+			if err != nil {
+				if tc.isErr {
+					return
+				}
+				t.Fatalf("Connection test failure: %s", err)
+			}
+			if tc.isErr {
+				t.Fatalf("Expected error but test passed.")
+			}
+		})
+	}
+}