[SS-69] Add a GCP Connection to hold Google service account creds

Cargo.toml

@@ -347,6 +347,7 @@ futures = "0.3.32"
 futures-core = "0.3.32"
 futures-task = "0.3.32"
 futures-util = "0.3.32"
+gcp_auth = "0.12.6"
 glob = "0.3.3"
 globset = "0.4.18"
 governor = "0.10.1"

deny.toml

@@ -130,6 +130,19 @@ skip = [
     { name = "hashbrown", version = "0.16.1" },
     # Used by dynfmt; iceberg/typetag pulls in v0.4.
     { name = "erased-serde", version = "0.3.26" },
+    # gcp_auth → hyper-rustls → rustls-native-certs pulls newer versions
+    # while native-tls still pulls older versions.
+    { name = "core-foundation", version = "0.10.1" },
+    { name = "security-framework", version = "3.7.0" },
+    { name = "openssl-probe", version = "0.2.1" },
+    # reqsign (via iceberg-storage-opendal / opendal) pins older deps
+    # than the workspace.
+    { name = "jsonwebtoken", version = "9.3.1" },
+    { name = "quick-xml", version = "0.37.5" },
+    # aws-lc-rs (via jsonwebtoken 10) and ring pull different `untrusted`.
+    { name = "untrusted", version = "0.7.1" },
+    # Held back by lazy_static 1.4.0 (used by num-bigint-dig).
+    { name = "spin", version = "0.5.2" },
 ]
 
 [[bans.deny]]
@@ -206,9 +219,11 @@ wrappers = [
 ]
 
 # We prefer the system's native TLS or OpenSSL to Rustls, since they are more
-# mature and more widely used.
+# mature and more widely used. `gcp_auth` only ships with rustls-based TLS,
+# so allow it through.
 [[bans.deny]]
 name = "rustls"
+wrappers = ["hyper-rustls", "tokio-rustls"]
 
 # once_cell is going to be added to std, and doesn't use macros
 # Unfortunately, its heavily used, so we have lots of exceptions.
@@ -219,6 +234,7 @@ wrappers = [
     "findshlibs",
     "launchdarkly-server-sdk",
     "launchdarkly-server-sdk-evaluation",
+    "num-bigint-dig",
     "prometheus",
     "rayon-core",
     "sharded-slab",

src/adapter/src/catalog/builtin_table_updates.rs

@@ -937,6 +937,7 @@ impl CatalogState {
             }
             ConnectionDetails::Csr(_)
             | ConnectionDetails::GlueSchemaRegistry(_)
+            | ConnectionDetails::Gcp(_)
             | ConnectionDetails::Postgres(_)
             | ConnectionDetails::MySql(_)
             | ConnectionDetails::SqlServer(_)

src/adapter/src/catalog/state.rs

@@ -2776,6 +2776,7 @@ impl ConnectionResolver for CatalogState {
             Ssh(conn) => Ssh(conn),
             Aws(conn) => Aws(conn),
             AwsPrivatelink(conn) => AwsPrivatelink(conn),
+            Gcp(conn) => Gcp(conn),
             MySql(conn) => MySql(conn.into_inline_connection(self)),
             SqlServer(conn) => SqlServer(conn.into_inline_connection(self)),
             IcebergCatalog(conn) => IcebergCatalog(conn.into_inline_connection(self)),

src/adapter/src/coord/ddl.rs

@@ -1140,6 +1140,7 @@ impl Coordinator {
                             | ConnectionDetails::GlueSchemaRegistry(_)
                             | ConnectionDetails::Ssh { .. }
                             | ConnectionDetails::Aws(_)
+                            | ConnectionDetails::Gcp(_)
                             | ConnectionDetails::IcebergCatalog(_) => {}
                         },
                         CatalogItem::Table(_) => {
@@ -1316,6 +1317,7 @@ impl Coordinator {
                 | ConnectionDetails::GlueSchemaRegistry(_)
                 | ConnectionDetails::Ssh { .. }
                 | ConnectionDetails::Aws(_)
+                | ConnectionDetails::Gcp(_)
                 | ConnectionDetails::IcebergCatalog(_) => {}
             }
         }

src/adapter/src/coord/sequencer/inner.rs

@@ -43,6 +43,7 @@ use mz_repr::{
     CatalogItemId, Datum, Diff, GlobalId, RelationVersion, RelationVersionSelector, Row, RowArena,
     RowIterator, Timestamp,
 };
+use mz_secrets::SecretsReader;
 use mz_sql::ast::{
     AlterSourceAddSubsourceOption, CreateSinkOption, CreateSinkOptionName, CreateSourceOptionName,
     CreateSubsourceOption, CreateSubsourceOptionName, SqlServerConfigOption,
@@ -63,6 +64,7 @@ use mz_sql::plan::{
     StatementContext,
 };
 use mz_sql::pure::{PurifiedSourceExport, generate_subsource_statements};
+use mz_storage_types::connections::gcp::GcpServiceAccountKeyTokenUri;
 use mz_storage_types::sinks::StorageSinkDesc;
 // Import `plan` module, but only import select elements to avoid merge conflicts on use statements.
 use mz_sql::plan::{
@@ -662,6 +664,19 @@ impl Coordinator {
                 }
                 self.caching_secrets_reader.invalidate(connection_id);
             }
+            ConnectionDetails::Gcp(gcp) => {
+                // A service account key defines its own OAuth2 token URI.
+                // We only want to send requests to the actual Google OAuth2 token API,
+                // so we inspect the key as early as we can.
+                if let Err(err) = self
+                    .caching_secrets_reader
+                    .read_string(gcp.credentials_json)
+                    .await
+                    .and_then(|json| GcpServiceAccountKeyTokenUri::validate_json(&json))
+                {
+                    return ctx.retire(Err(err.into()));
+                }
+            }
             _ => (),
         };
 

src/sql-lexer/src/keywords.txt

@@ -25,6 +25,7 @@
 
 Abort
 Access
+Account
 Action
 Add
 Added
@@ -200,6 +201,7 @@ Full
 Fullname
 Function
 Fusion
+Gcp
 Generator
 Glue
 Grant

src/sql-parser/src/ast/defs/ddl.rs

@@ -792,6 +792,7 @@ pub enum ConnectionOptionName {
     Scope,
     SecretAccessKey,
     SecurityProtocol,
+    ServiceAccountKey,
     ServiceName,
     SshTunnel,
     SslCertificate,
@@ -836,6 +837,7 @@ impl AstDisplay for ConnectionOptionName {
             ConnectionOptionName::Scope => "SCOPE",
             ConnectionOptionName::SecurityProtocol => "SECURITY PROTOCOL",
             ConnectionOptionName::SecretAccessKey => "SECRET ACCESS KEY",
+            ConnectionOptionName::ServiceAccountKey => "SERVICE ACCOUNT KEY",
             ConnectionOptionName::ServiceName => "SERVICE NAME",
             ConnectionOptionName::SshTunnel => "SSH TUNNEL",
             ConnectionOptionName::SslCertificate => "SSL CERTIFICATE",
@@ -886,6 +888,7 @@ impl WithOptionName for ConnectionOptionName {
             | ConnectionOptionName::Scope
             | ConnectionOptionName::SecurityProtocol
             | ConnectionOptionName::SecretAccessKey
+            | ConnectionOptionName::ServiceAccountKey
             | ConnectionOptionName::ServiceName
             | ConnectionOptionName::SshTunnel
             | ConnectionOptionName::SslCertificate
@@ -915,6 +918,7 @@ pub enum CreateConnectionType {
     Aws,
     AwsPrivatelink,
     GlueSchemaRegistry,
+    Gcp,
     Kafka,
     Csr,
     Postgres,
@@ -933,6 +937,7 @@ impl CreateConnectionType {
             Self::Aws => "aws",
             Self::AwsPrivatelink => "aws-privatelink",
             Self::GlueSchemaRegistry => "glue-schema-registry",
+            Self::Gcp => "gcp",
             Self::Ssh => "ssh-tunnel",
             Self::MySql => "mysql",
             Self::SqlServer => "sql-server",
@@ -962,6 +967,9 @@ impl AstDisplay for CreateConnectionType {
             Self::GlueSchemaRegistry => {
                 f.write_str("AWS GLUE SCHEMA REGISTRY");
             }
+            Self::Gcp => {
+                f.write_str("GCP");
+            }
             Self::Ssh => {
                 f.write_str("SSH TUNNEL");
             }