[Cherry-pick to branch-1.3] [#14031] fix(flink-connector): Validate credential providers for Iceberg non-REST backends (#14032)

docs/flink-connector/flink-catalog-iceberg.md

@@ -119,6 +119,15 @@ Gravitino catalog property names with the prefix `flink.bypass.` are passed to F
 
 ## Storage
 
+### Credential Vending
+
+Unless the catalog uses a `REST` backend with `data-access` set to `vended-credentials`, the Flink connector fetches catalog-level credentials from Gravitino once, when the catalog is opened, and never refreshes them. Only static credential providers, such as `s3-secret-key`, `oss-secret-key` and `azure-account-key`, are vended to Flink:
+
+- `s3-token`, `gcs-token`, `adls-token` and `oss-token` vend no credential at catalog level. Flink logs a warning and Iceberg uses the credentials available on the Flink nodes, such as an instance profile.
+- `aws-irsa` is rejected when the catalog is loaded, because its credential expires and is never refreshed.
+
+For temporary credentials vended per table and refreshed automatically, use a `REST` backend with `data-access` set to `vended-credentials`.
+
 ### OSS
 
 Additionally, you need download the [Aliyun OSS SDK](https://gosspublic.alicdn.com/sdks/java/aliyun_java_sdk_3.10.2.zip), and copy `aliyun-sdk-oss-3.10.2.jar`, `hamcrest-core-1.1.jar`, `jdom2-2.0.6.jar` to the Flink classpath.

docs/security/credential-vending.md

@@ -27,7 +27,7 @@ Gravitino credential vending is used to generate temporary or static credentials
   - GCS: `GCSTokenCredential`
   - ADLS: `ADLSTokenCredential`, `AzureAccountKeyCredential`
   - OSS: `OSSTokenCredential`, `OSSSecretKeyCredential`
-- Spark, Flink, and Trino connectors automatically consume vended credentials for Hive, Iceberg, Glue, JDBC, and Paimon catalogs.
+- Spark, Flink, and Trino connectors automatically consume vended credentials for Hive, Iceberg, Glue, JDBC, and Paimon catalogs. For Iceberg catalogs without a `REST` backend that vends credentials, the Flink connector receives only static credentials, see [Flink Iceberg credential vending](../flink-connector/flink-catalog-iceberg.md#credential-vending).
 
 ## General Configurations
 
@@ -225,7 +225,7 @@ Clients retrieve vended credentials from:
 GET /api/metalakes/{metalake}/objects/catalog/{catalog}/credentials
 ```
 
-The server returns short-lived or static credentials that the client can use to access the underlying storage directly.
+The server returns short-lived or static credentials that the client can use to access the underlying storage directly. These credentials are catalog-level and are not refreshed: token providers such as `s3-token` vend nothing at catalog level, and `aws-irsa` credentials expire after `s3-token-expire-in-secs`.
 
 ### Spark and Flink Connector Integration
 

flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalog.java

@@ -18,18 +18,37 @@
  */
 package org.apache.gravitino.flink.connector.iceberg;
 
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableSet;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.table.catalog.AbstractCatalog;
 import org.apache.flink.table.catalog.exceptions.CatalogException;
 import org.apache.flink.table.factories.Factory;
 import org.apache.flink.util.Preconditions;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.credential.ADLSTokenCredential;
+import org.apache.gravitino.credential.AwsIrsaCredential;
+import org.apache.gravitino.credential.CredentialConstants;
 import org.apache.gravitino.credential.CredentialPropertyUtils;
+import org.apache.gravitino.credential.GCSTokenCredential;
+import org.apache.gravitino.credential.OSSTokenCredential;
+import org.apache.gravitino.credential.S3SecretKeyCredential;
+import org.apache.gravitino.credential.S3TokenCredential;
 import org.apache.gravitino.exceptions.NoSuchCatalogException;
 import org.apache.gravitino.flink.connector.PartitionConverter;
 import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
 import org.apache.gravitino.flink.connector.catalog.BaseCatalog;
+import org.apache.gravitino.storage.GCSProperties;
 import org.apache.iceberg.flink.FlinkCatalogFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -39,6 +58,13 @@ public class GravitinoIcebergCatalog extends BaseCatalog {
 
   private static final Logger LOG = LoggerFactory.getLogger(GravitinoIcebergCatalog.class);
 
+  private static final Set<String> PATH_SCOPED_TOKEN_PROVIDERS =
+      ImmutableSet.of(
+          S3TokenCredential.S3_TOKEN_CREDENTIAL_TYPE,
+          GCSTokenCredential.GCS_TOKEN_CREDENTIAL_TYPE,
+          ADLSTokenCredential.ADLS_TOKEN_CREDENTIAL_TYPE,
+          OSSTokenCredential.OSS_TOKEN_CREDENTIAL_TYPE);
+
   private final String icebergCatalogName;
   // Mutable copy so credential injection in open() propagates to the inner Iceberg catalog.
   private final Map<String, String> mutableIcebergProperties;
@@ -83,8 +109,20 @@ protected GravitinoIcebergCatalog(
   public void open() throws CatalogException {
     if (icebergCatalog == null) {
       try {
+        Catalog gravitinoCatalog = catalog();
+        List<String> tokenProviders =
+            checkCredentialProviders(
+                catalogName(), mutableIcebergProperties, gravitinoCatalog.properties());
+        if (!tokenProviders.isEmpty()) {
+          LOG.warn(
+              "Catalog '{}' uses credential providers {}, which vend no credential at catalog"
+                  + " level. The Flink connector injects no storage credential for them, so"
+                  + " Iceberg FileIO uses the credentials available on the Flink nodes.",
+              catalogName(),
+              tokenProviders);
+        }
         CredentialPropertyUtils.applyIcebergCredentials(
-            CredentialPropertyUtils.getCredentials(catalog()), mutableIcebergProperties);
+            CredentialPropertyUtils.getCredentials(gravitinoCatalog), mutableIcebergProperties);
       } catch (NoSuchCatalogException e) {
         LOG.warn(
             "Catalog '{}' not found in Gravitino during open(); credential injection skipped."
@@ -119,4 +157,55 @@ protected static AbstractCatalog asAbstractCatalog(Object catalog) {
         catalog.getClass().getName());
     return (AbstractCatalog) catalog;
   }
+
+  // Without per-table vended credentials, the inner catalog only receives the catalog-level
+  // credentials fetched once in open(). A REST backend with vended credentials enabled is exempt
+  // because the inner REST catalog obtains per-table, refreshable credentials itself. aws-irsa is
+  // rejected: its credential is injected but never refreshed, so it overrides the node's own
+  // credentials and then expires. Path-scoped token providers vend nothing at catalog level, which
+  // leaves the node's own credentials in effect, so they are only returned for the caller to warn
+  // about. The REST check uses the effective inner catalog-type, since an explicit catalog-type may
+  // differ from the persisted catalog-backend. When no provider list is set, the server infers
+  // gcs-token from gcs-service-account-file, which is not reflected in the catalog properties seen
+  // by the client, so the same inference is applied here.
+  @VisibleForTesting
+  static List<String> checkCredentialProviders(
+      String catalogName,
+      Map<String, String> icebergCatalogProperties,
+      @Nullable Map<String, String> gravitinoCatalogProperties) {
+    boolean restWithVendedCredentials =
+        IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_REST.equalsIgnoreCase(
+                icebergCatalogProperties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE))
+            && IcebergPropertiesConstants.ICEBERG_DATA_ACCESS_VENDED_CREDENTIALS.equalsIgnoreCase(
+                icebergCatalogProperties.get(IcebergPropertiesConstants.ICEBERG_ACCESS_DELEGATION));
+    if (restWithVendedCredentials || gravitinoCatalogProperties == null) {
+      return Collections.emptyList();
+    }
+    String providers = gravitinoCatalogProperties.get(CredentialConstants.CREDENTIAL_PROVIDERS);
+    if (StringUtils.isBlank(providers)) {
+      return StringUtils.isNotBlank(
+              gravitinoCatalogProperties.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE))
+          ? Collections.singletonList(GCSTokenCredential.GCS_TOKEN_CREDENTIAL_TYPE)
+          : Collections.emptyList();
+    }
+    List<String> providerList =
+        Splitter.on(',').trimResults().omitEmptyStrings().splitToList(providers);
+    if (providerList.stream()
+        .anyMatch(AwsIrsaCredential.AWS_IRSA_CREDENTIAL_TYPE::equalsIgnoreCase)) {
+      throw new CatalogException(
+          String.format(
+              "Catalog '%s' uses credential provider %s, which the Flink connector does not"
+                  + " support for Iceberg catalogs with a non-REST backend: its credential is"
+                  + " never refreshed and expires. Use a static provider such as %s, or use a"
+                  + " REST backend with %s=%s.",
+              catalogName,
+              AwsIrsaCredential.AWS_IRSA_CREDENTIAL_TYPE,
+              S3SecretKeyCredential.S3_SECRET_KEY_CREDENTIAL_TYPE,
+              IcebergPropertiesConstants.GRAVITINO_ICEBERG_DATA_ACCESS,
+              IcebergPropertiesConstants.ICEBERG_DATA_ACCESS_VENDED_CREDENTIALS));
+    }
+    return providerList.stream()
+        .filter(provider -> PATH_SCOPED_TOKEN_PROVIDERS.contains(provider.toLowerCase(Locale.ROOT)))
+        .collect(Collectors.toList());
+  }
 }

flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConstants.java

@@ -52,4 +52,13 @@ public class IcebergPropertiesConstants {
 
   @VisibleForTesting
   public static final String ICEBERG_CATALOG_BACKEND_REST = CatalogUtil.ICEBERG_CATALOG_TYPE_REST;
+
+  /** Gravitino Iceberg catalog property that selects how clients access table data. */
+  public static final String GRAVITINO_ICEBERG_DATA_ACCESS = IcebergConstants.DATA_ACCESS;
+
+  /** Iceberg catalog property that the data access mode is passed to the REST backend as. */
+  public static final String ICEBERG_ACCESS_DELEGATION = IcebergConstants.ICEBERG_ACCESS_DELEGATION;
+
+  /** Data access mode in which the REST backend vends per-table storage credentials. */
+  public static final String ICEBERG_DATA_ACCESS_VENDED_CREDENTIALS = "vended-credentials";
 }

flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestGravitinoIcebergCatalog.java

@@ -19,16 +19,31 @@
 
 package org.apache.gravitino.flink.connector.iceberg;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import java.util.Locale;
+import java.util.Map;
 import org.apache.flink.table.catalog.AbstractCatalog;
 import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.gravitino.credential.CredentialConstants;
+import org.apache.gravitino.credential.GCSTokenCredential;
+import org.apache.gravitino.storage.GCSProperties;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 class TestGravitinoIcebergCatalog {
 
+  private static final Map<String, String> HIVE_CATALOG =
+      ImmutableMap.of(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE, "hive");
+
   @Test
   void testRequireAbstractCatalogAcceptsAbstractCatalog() {
     AbstractCatalog catalog = mock(AbstractCatalog.class);
@@ -43,4 +58,148 @@ void testRequireAbstractCatalogRejectsNonAbstractCatalog() {
     assertThrows(
         IllegalStateException.class, () -> GravitinoIcebergCatalog.asAbstractCatalog(catalog));
   }
+
+  @Test
+  void testCheckCredentialProvidersRejectsAwsIrsaForNonRestBackend() {
+    Map<String, String> properties =
+        ImmutableMap.of(
+            IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+            "hive",
+            CredentialConstants.CREDENTIAL_PROVIDERS,
+            "s3-secret-key, AWS-IRSA");
+
+    CatalogException exception =
+        assertThrows(
+            CatalogException.class,
+            () -> GravitinoIcebergCatalog.checkCredentialProviders("c1", HIVE_CATALOG, properties));
+    assertTrue(exception.getMessage().contains("aws-irsa"));
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"s3-token", "gcs-token", "adls-token", "oss-token"})
+  void testCheckCredentialProvidersReturnsTokenProvidersForNonRestBackend(String provider) {
+    String upperCaseProvider = provider.toUpperCase(Locale.ROOT);
+    Map<String, String> properties =
+        ImmutableMap.of(
+            IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+            "hive",
+            CredentialConstants.CREDENTIAL_PROVIDERS,
+            "s3-secret-key, " + upperCaseProvider);
+
+    assertEquals(
+        ImmutableList.of(upperCaseProvider),
+        GravitinoIcebergCatalog.checkCredentialProviders("c1", HIVE_CATALOG, properties));
+  }
+
+  @Test
+  void testCheckCredentialProvidersIgnoresStaticProviders() {
+    Map<String, String> properties =
+        ImmutableMap.of(
+            IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+            "jdbc",
+            CredentialConstants.CREDENTIAL_PROVIDERS,
+            "s3-secret-key,jdbc-user-password");
+
+    assertTrue(
+        GravitinoIcebergCatalog.checkCredentialProviders("c1", ImmutableMap.of(), properties)
+            .isEmpty());
+  }
+
+  @Test
+  void testCheckCredentialProvidersIgnoresMissingProviders() {
+    Map<String, String> properties =
+        ImmutableMap.of(IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND, "hive");
+
+    assertTrue(
+        GravitinoIcebergCatalog.checkCredentialProviders("c1", HIVE_CATALOG, properties).isEmpty());
+  }
+
+  @Test
+  void testCheckCredentialProvidersInfersGcsTokenFromServiceAccountFile() {
+    Map<String, String> properties =
+        ImmutableMap.of(
+            IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+            "hive",
+            GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE,
+            "/path/to/sa.json");
+
+    assertEquals(
+        ImmutableList.of(GCSTokenCredential.GCS_TOKEN_CREDENTIAL_TYPE),
+        GravitinoIcebergCatalog.checkCredentialProviders("c1", HIVE_CATALOG, properties));
+  }
+
+  @Test
+  void testCheckCredentialProvidersUsesExplicitProvidersOverGcsInference() {
+    Map<String, String> properties =
+        ImmutableMap.of(
+            IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+            "hive",
+            GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE,
+            "/path/to/sa.json",
+            CredentialConstants.CREDENTIAL_PROVIDERS,
+            "s3-secret-key");
+
+    assertTrue(
+        GravitinoIcebergCatalog.checkCredentialProviders("c1", HIVE_CATALOG, properties).isEmpty());
+  }
+
+  @Test
+  void testCheckCredentialProvidersSkipsRestBackendWithVendedCredentials() {
+    Map<String, String> properties =
+        ImmutableMap.of(
+            IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+            "REST",
+            CredentialConstants.CREDENTIAL_PROVIDERS,
+            "s3-token,aws-irsa");
+
+    assertTrue(
+        GravitinoIcebergCatalog.checkCredentialProviders(
+                "c1", restCatalog("REST", true), properties)
+            .isEmpty());
+  }
+
+  @Test
+  void testCheckCredentialProvidersUsesEffectiveCatalogType() {
+    Map<String, String> properties =
+        ImmutableMap.of(
+            IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+            "hive",
+            CredentialConstants.CREDENTIAL_PROVIDERS,
+            "aws-irsa");
+
+    assertTrue(
+        GravitinoIcebergCatalog.checkCredentialProviders(
+                "c1", restCatalog("rest", true), properties)
+            .isEmpty());
+  }
+
+  @Test
+  void testCheckCredentialProvidersRejectsAwsIrsaForRestBackendWithoutVendedCredentials() {
+    Map<String, String> properties =
+        ImmutableMap.of(CredentialConstants.CREDENTIAL_PROVIDERS, "aws-irsa");
+
+    assertThrows(
+        CatalogException.class,
+        () ->
+            GravitinoIcebergCatalog.checkCredentialProviders(
+                "c1", restCatalog("rest", false), properties));
+  }
+
+  @Test
+  void testCheckCredentialProvidersIgnoresNullProperties() {
+    assertTrue(
+        GravitinoIcebergCatalog.checkCredentialProviders("c1", HIVE_CATALOG, null).isEmpty());
+  }
+
+  private static Map<String, String> restCatalog(String catalogType, boolean vendedCredentials) {
+    ImmutableMap.Builder<String, String> builder =
+        ImmutableMap.<String, String>builder()
+            .put(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE, catalogType);
+    if (vendedCredentials) {
+      builder.put(
+          IcebergPropertiesConstants.ICEBERG_ACCESS_DELEGATION,
+          IcebergPropertiesConstants.ICEBERG_DATA_ACCESS_VENDED_CREDENTIALS);
+    }
+    return builder.build();
+  }
 }