Support GCS vended credentials in Iceberg REST catalog

lib/trino-filesystem-gcs/src/main/java/io/trino/filesystem/gcs/GcsStorageFactory.java

@@ -14,6 +14,9 @@
 package io.trino.filesystem.gcs;
 
 import com.google.api.gax.retrying.RetrySettings;
+import com.google.auth.oauth2.AccessToken;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.cloud.NoCredentials;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageOptions;
 import com.google.inject.Inject;
@@ -22,11 +25,18 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.time.Duration;
+import java.util.Date;
 import java.util.Map;
 import java.util.Optional;
 
 import static com.google.cloud.storage.StorageRetryStrategy.getUniformStorageRetryStrategy;
 import static com.google.common.net.HttpHeaders.USER_AGENT;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_NO_AUTH_PROPERTY;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_EXPIRES_AT_PROPERTY;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_PROPERTY;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_PROJECT_ID_PROPERTY;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_SERVICE_HOST_PROPERTY;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_USER_PROJECT_PROPERTY;
 import static java.util.Objects.requireNonNull;
 
 public class GcsStorageFactory
@@ -59,14 +69,45 @@ public GcsStorageFactory(GcsFileSystemConfig config, GcsAuth gcsAuth)
     public Storage create(ConnectorIdentity identity)
     {
         try {
+            Map<String, String> extraCredentials = identity.getExtraCredentials();
+            boolean noAuth = Boolean.parseBoolean(extraCredentials.getOrDefault(EXTRA_CREDENTIALS_GCS_NO_AUTH_PROPERTY, "false"));
+            String vendedOAuthToken = extraCredentials.get(EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_PROPERTY);
+
             StorageOptions.Builder storageOptionsBuilder = StorageOptions.newBuilder();
-            if (projectId != null) {
-                storageOptionsBuilder.setProjectId(projectId);
+
+            String effectiveProjectId = extraCredentials.getOrDefault(EXTRA_CREDENTIALS_GCS_PROJECT_ID_PROPERTY, projectId);
+            if (effectiveProjectId != null) {
+                storageOptionsBuilder.setProjectId(effectiveProjectId);
             }
 
-            gcsAuth.setAuth(storageOptionsBuilder, identity);
+            if (noAuth) {
+                storageOptionsBuilder.setCredentials(NoCredentials.getInstance());
+            }
+            else if (vendedOAuthToken != null) {
+                Date expirationTime = null;
+                String expiresAt = extraCredentials.get(EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_EXPIRES_AT_PROPERTY);
+                if (expiresAt != null) {
+                    expirationTime = new Date(Long.parseLong(expiresAt));
+                }
+                AccessToken accessToken = new AccessToken(vendedOAuthToken, expirationTime);
+                storageOptionsBuilder.setCredentials(GoogleCredentials.create(accessToken));
+            }
+            else {
+                gcsAuth.setAuth(storageOptionsBuilder, identity);
+            }
 
-            endpoint.ifPresent(storageOptionsBuilder::setHost);
+            String vendedServiceHost = extraCredentials.get(EXTRA_CREDENTIALS_GCS_SERVICE_HOST_PROPERTY);
+            if (vendedServiceHost != null) {
+                storageOptionsBuilder.setHost(vendedServiceHost);
+            }
+            else {
+                endpoint.ifPresent(storageOptionsBuilder::setHost);
+            }
+
+            String vendedUserProject = extraCredentials.get(EXTRA_CREDENTIALS_GCS_USER_PROJECT_PROPERTY);
+            if (vendedUserProject != null) {
+                storageOptionsBuilder.setQuotaProjectId(vendedUserProject);
+            }
 
             // Note: without uniform strategy we cannot retry idempotent operations.
             // The trino-filesystem api does not violate the conditions for idempotency, see https://cloud.google.com/storage/docs/retry-strategy#java for details.

lib/trino-filesystem-gcs/src/test/java/io/trino/filesystem/gcs/TestGcsStorageFactory.java

@@ -14,12 +14,21 @@
 package io.trino.filesystem.gcs;
 
 import com.google.auth.Credentials;
+import com.google.auth.oauth2.AccessToken;
+import com.google.auth.oauth2.GoogleCredentials;
 import com.google.cloud.NoCredentials;
 import com.google.cloud.storage.Storage;
+import com.google.common.collect.ImmutableMap;
 import io.trino.spi.security.ConnectorIdentity;
 import org.junit.jupiter.api.Test;
 
 import static io.trino.filesystem.gcs.GcsFileSystemConfig.AuthType;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_NO_AUTH_PROPERTY;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_EXPIRES_AT_PROPERTY;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_PROPERTY;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_PROJECT_ID_PROPERTY;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_SERVICE_HOST_PROPERTY;
+import static io.trino.filesystem.gcs.GcsFileSystemConstants.EXTRA_CREDENTIALS_GCS_USER_PROJECT_PROPERTY;
 import static org.assertj.core.api.Assertions.assertThat;
 
 final class TestGcsStorageFactory
@@ -38,4 +47,192 @@ void testApplicationDefaultCredentials()
 
         assertThat(actualCredentials).isEqualTo(NoCredentials.getInstance());
     }
+
+    @Test
+    void testVendedOAuthToken()
+            throws Exception
+    {
+        GcsFileSystemConfig config = new GcsFileSystemConfig().setAuthType(AuthType.APPLICATION_DEFAULT);
+        GcsStorageFactory storageFactory = new GcsStorageFactory(config, new ApplicationDefaultAuth());
+
+        ConnectorIdentity identity = ConnectorIdentity.forUser("test")
+                .withExtraCredentials(ImmutableMap.of(
+                        EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_PROPERTY, "ya29.test-token"))
+                .build();
+
+        try (Storage storage = storageFactory.create(identity)) {
+            Credentials credentials = storage.getOptions().getCredentials();
+            assertThat(credentials).isInstanceOf(GoogleCredentials.class);
+            GoogleCredentials googleCredentials = (GoogleCredentials) credentials;
+            AccessToken accessToken = googleCredentials.getAccessToken();
+            assertThat(accessToken).isNotNull();
+            assertThat(accessToken.getTokenValue()).isEqualTo("ya29.test-token");
+        }
+    }
+
+    @Test
+    void testVendedOAuthTokenWithExpiration()
+            throws Exception
+    {
+        GcsFileSystemConfig config = new GcsFileSystemConfig().setAuthType(AuthType.APPLICATION_DEFAULT);
+        GcsStorageFactory storageFactory = new GcsStorageFactory(config, new ApplicationDefaultAuth());
+
+        ConnectorIdentity identity = ConnectorIdentity.forUser("test")
+                .withExtraCredentials(ImmutableMap.of(
+                        EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_PROPERTY, "ya29.test-token",
+                        EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_EXPIRES_AT_PROPERTY, "1700000000000"))
+                .build();
+
+        try (Storage storage = storageFactory.create(identity)) {
+            Credentials credentials = storage.getOptions().getCredentials();
+            assertThat(credentials).isInstanceOf(GoogleCredentials.class);
+            GoogleCredentials googleCredentials = (GoogleCredentials) credentials;
+            AccessToken accessToken = googleCredentials.getAccessToken();
+            assertThat(accessToken).isNotNull();
+            assertThat(accessToken.getTokenValue()).isEqualTo("ya29.test-token");
+            assertThat(accessToken.getExpirationTime()).isNotNull();
+            assertThat(accessToken.getExpirationTime().getTime()).isEqualTo(1700000000000L);
+        }
+    }
+
+    @Test
+    void testVendedProjectId()
+            throws Exception
+    {
+        GcsFileSystemConfig config = new GcsFileSystemConfig()
+                .setAuthType(AuthType.APPLICATION_DEFAULT)
+                .setProjectId("static-project");
+        GcsStorageFactory storageFactory = new GcsStorageFactory(config, new ApplicationDefaultAuth());
+
+        ConnectorIdentity identity = ConnectorIdentity.forUser("test")
+                .withExtraCredentials(ImmutableMap.of(
+                        EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_PROPERTY, "ya29.test-token",
+                        EXTRA_CREDENTIALS_GCS_PROJECT_ID_PROPERTY, "vended-project"))
+                .build();
+
+        try (Storage storage = storageFactory.create(identity)) {
+            assertThat(storage.getOptions().getProjectId()).isEqualTo("vended-project");
+        }
+    }
+
+    @Test
+    void testVendedServiceHost()
+            throws Exception
+    {
+        GcsFileSystemConfig config = new GcsFileSystemConfig()
+                .setAuthType(AuthType.APPLICATION_DEFAULT);
+        GcsStorageFactory storageFactory = new GcsStorageFactory(config, new ApplicationDefaultAuth());
+
+        ConnectorIdentity identity = ConnectorIdentity.forUser("test")
+                .withExtraCredentials(ImmutableMap.of(
+                        EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_PROPERTY, "ya29.test-token",
+                        EXTRA_CREDENTIALS_GCS_SERVICE_HOST_PROPERTY, "https://custom-storage.googleapis.com"))
+                .build();
+
+        try (Storage storage = storageFactory.create(identity)) {
+            assertThat(storage.getOptions().getHost()).isEqualTo("https://custom-storage.googleapis.com");
+        }
+    }
+
+    @Test
+    void testVendedNoAuth()
+            throws Exception
+    {
+        GcsFileSystemConfig config = new GcsFileSystemConfig().setAuthType(AuthType.APPLICATION_DEFAULT);
+        GcsStorageFactory storageFactory = new GcsStorageFactory(config, new ApplicationDefaultAuth());
+
+        ConnectorIdentity identity = ConnectorIdentity.forUser("test")
+                .withExtraCredentials(ImmutableMap.of(
+                        EXTRA_CREDENTIALS_GCS_NO_AUTH_PROPERTY, "true"))
+                .build();
+
+        try (Storage storage = storageFactory.create(identity)) {
+            assertThat(storage.getOptions().getCredentials()).isEqualTo(NoCredentials.getInstance());
+        }
+    }
+
+    @Test
+    void testNoAuthTakesPriorityOverOAuthToken()
+            throws Exception
+    {
+        GcsFileSystemConfig config = new GcsFileSystemConfig().setAuthType(AuthType.APPLICATION_DEFAULT);
+        GcsStorageFactory storageFactory = new GcsStorageFactory(config, new ApplicationDefaultAuth());
+
+        ConnectorIdentity identity = ConnectorIdentity.forUser("test")
+                .withExtraCredentials(ImmutableMap.of(
+                        EXTRA_CREDENTIALS_GCS_NO_AUTH_PROPERTY, "true",
+                        EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_PROPERTY, "ya29.test-token"))
+                .build();
+
+        try (Storage storage = storageFactory.create(identity)) {
+            assertThat(storage.getOptions().getCredentials()).isEqualTo(NoCredentials.getInstance());
+        }
+    }
+
+    @Test
+    void testVendedUserProject()
+            throws Exception
+    {
+        GcsFileSystemConfig config = new GcsFileSystemConfig()
+                .setAuthType(AuthType.APPLICATION_DEFAULT);
+        GcsStorageFactory storageFactory = new GcsStorageFactory(config, new ApplicationDefaultAuth());
+
+        ConnectorIdentity identity = ConnectorIdentity.forUser("test")
+                .withExtraCredentials(ImmutableMap.of(
+                        EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_PROPERTY, "ya29.test-token",
+                        EXTRA_CREDENTIALS_GCS_USER_PROJECT_PROPERTY, "billing-project"))
+                .build();
+
+        try (Storage storage = storageFactory.create(identity)) {
+            assertThat(storage.getOptions().getQuotaProjectId()).isEqualTo("billing-project");
+        }
+    }
+
+    @Test
+    void testNoAuthFalseDoesNotSkipAuth()
+            throws Exception
+    {
+        GcsFileSystemConfig config = new GcsFileSystemConfig().setAuthType(AuthType.APPLICATION_DEFAULT);
+        GcsStorageFactory storageFactory = new GcsStorageFactory(config, new ApplicationDefaultAuth());
+
+        ConnectorIdentity identity = ConnectorIdentity.forUser("test")
+                .withExtraCredentials(ImmutableMap.of(
+                        EXTRA_CREDENTIALS_GCS_NO_AUTH_PROPERTY, "false",
+                        EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_PROPERTY, "ya29.test-token"))
+                .build();
+
+        try (Storage storage = storageFactory.create(identity)) {
+            Credentials credentials = storage.getOptions().getCredentials();
+            assertThat(credentials).isInstanceOf(GoogleCredentials.class);
+            assertThat(((GoogleCredentials) credentials).getAccessToken().getTokenValue()).isEqualTo("ya29.test-token");
+        }
+    }
+
+    @Test
+    void testUserProjectNotSetWithoutVendedCredentials()
+            throws Exception
+    {
+        GcsFileSystemConfig config = new GcsFileSystemConfig()
+                .setAuthType(AuthType.APPLICATION_DEFAULT);
+        GcsStorageFactory storageFactory = new GcsStorageFactory(config, new ApplicationDefaultAuth());
+
+        try (Storage storage = storageFactory.create(ConnectorIdentity.ofUser("test"))) {
+            assertThat(storage.getOptions().getQuotaProjectId()).isNull();
+        }
+    }
+
+    @Test
+    void testStaticConfigUsedWithoutVendedCredentials()
+            throws Exception
+    {
+        GcsFileSystemConfig config = new GcsFileSystemConfig()
+                .setAuthType(AuthType.APPLICATION_DEFAULT)
+                .setProjectId("static-project");
+        GcsStorageFactory storageFactory = new GcsStorageFactory(config, new ApplicationDefaultAuth());
+
+        try (Storage storage = storageFactory.create(ConnectorIdentity.ofUser("test"))) {
+            assertThat(storage.getOptions().getProjectId()).isEqualTo("static-project");
+            assertThat(storage.getOptions().getCredentials()).isEqualTo(NoCredentials.getInstance());
+        }
+    }
 }

lib/trino-filesystem/src/main/java/io/trino/filesystem/gcs/GcsFileSystemConstants.java

@@ -0,0 +1,26 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.filesystem.gcs;
+
+public final class GcsFileSystemConstants
+{
+    public static final String EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_PROPERTY = "internal$gcs_oauth2_token";
+    public static final String EXTRA_CREDENTIALS_GCS_OAUTH_TOKEN_EXPIRES_AT_PROPERTY = "internal$gcs_oauth2_token_expires_at";
+    public static final String EXTRA_CREDENTIALS_GCS_PROJECT_ID_PROPERTY = "internal$gcs_project_id";
+    public static final String EXTRA_CREDENTIALS_GCS_SERVICE_HOST_PROPERTY = "internal$gcs_service_host";
+    public static final String EXTRA_CREDENTIALS_GCS_NO_AUTH_PROPERTY = "internal$gcs_no_auth";
+    public static final String EXTRA_CREDENTIALS_GCS_USER_PROJECT_PROPERTY = "internal$gcs_user_project";
+
+    private GcsFileSystemConstants() {}
+}

plugin/trino-iceberg/pom.xml

@@ -778,6 +778,7 @@
                                 <exclude>**/TestIcebergS3TablesConnectorSmokeTest.java</exclude>
                                 <exclude>**/TestIcebergBigLakeMetastoreConnectorSmokeTest.java</exclude>
                                 <exclude>**/TestIcebergGcsConnectorSmokeTest.java</exclude>
+                                <exclude>**/TestIcebergGcsVendingRestCatalogConnectorSmokeTest.java</exclude>
                                 <exclude>**/TestIcebergAbfsConnectorSmokeTest.java</exclude>
                                 <exclude>**/Test*FailureRecoveryTest.java</exclude>
                                 <exclude>**/TestIcebergSnowflakeCatalogConnectorSmokeTest.java</exclude>
@@ -845,6 +846,7 @@
                                 <include>**/TestIcebergS3TablesConnectorSmokeTest.java</include>
                                 <include>**/TestIcebergBigLakeMetastoreConnectorSmokeTest.java</include>
                                 <include>**/TestIcebergGcsConnectorSmokeTest.java</include>
+                                <include>**/TestIcebergGcsVendingRestCatalogConnectorSmokeTest.java</include>
                                 <include>**/TestIcebergAbfsConnectorSmokeTest.java</include>
                                 <include>**/TestIcebergSnowflakeCatalogConnectorSmokeTest.java</include>
                                 <include>**/TestTrinoSnowflakeCatalog.java</include>