This is an automated email from the ASF dual-hosted git repository.
mchades pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 8ea1e3a90f [Cherry-pick to branch-1.3] [#10990] feat(iceberg-rest):
Vended credential refresh for S3, GCS, OSS, and ADLS (#11294) (#11449)
8ea1e3a90f is described below
commit 8ea1e3a90fb583828689f95d8a0a3e01c4a17a7a
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Jun 5 15:59:11 2026 +0800
[Cherry-pick to branch-1.3] [#10990] feat(iceberg-rest): Vended credential
refresh for S3, GCS, OSS, and ADLS (#11294) (#11449)
**Cherry-pick Information:**
- Original commit: 91ed7a2e750537276d8ac2f579b81dec162ff307
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Co-authored-by: MaSai <[email protected]>
---
.../gravitino/credential/ADLSTokenCredential.java | 2 +-
.../gravitino/credential/AwsIrsaCredential.java | 9 +-
.../gravitino/credential/GCSTokenCredential.java | 2 +-
.../gravitino/credential/JdbcCredential.java | 2 +-
.../credential/OSSSecretKeyCredential.java | 2 +-
.../gravitino/credential/OSSTokenCredential.java | 2 +-
.../credential/S3SecretKeyCredential.java | 2 +-
.../gravitino/credential/S3TokenCredential.java | 2 +-
.../s3/credential/AwsIrsaCredentialGenerator.java | 10 +-
.../credential/CredentialPropertyUtils.java | 134 +++++--
.../credential/TestCredentialFactory.java | 24 ++
.../credential/TestCredentialPropertiesUtils.java | 75 +++-
.../credential/CatalogCredentialManager.java | 9 +
.../iceberg/service/CatalogWrapperForREST.java | 159 ++++++--
.../iceberg/service/IcebergRESTUtils.java | 124 ++++++
.../iceberg/service/TestCatalogWrapperForREST.java | 416 ++++++++++++++++++++-
.../iceberg/service/TestIcebergRESTUtils.java | 231 ++++++++++++
17 files changed, 1121 insertions(+), 84 deletions(-)
diff --git
a/api/src/main/java/org/apache/gravitino/credential/ADLSTokenCredential.java
b/api/src/main/java/org/apache/gravitino/credential/ADLSTokenCredential.java
index 6f1c463033..d18aa7f115 100644
--- a/api/src/main/java/org/apache/gravitino/credential/ADLSTokenCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/ADLSTokenCredential.java
@@ -112,6 +112,6 @@ public class ADLSTokenCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(sasToken), "ADLS SAS token should not be
empty.");
Preconditions.checkArgument(
- expireTimeInMS > 0, "The expire time of ADLSTokenCredential should
great than 0");
+ expireTimeInMS > 0, "The expiration time of ADLSTokenCredential should
be greater than 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/AwsIrsaCredential.java
b/api/src/main/java/org/apache/gravitino/credential/AwsIrsaCredential.java
index 0c1888cffc..4fecf9ada1 100644
--- a/api/src/main/java/org/apache/gravitino/credential/AwsIrsaCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/AwsIrsaCredential.java
@@ -50,7 +50,7 @@ public class AwsIrsaCredential implements Credential {
*/
public AwsIrsaCredential(
String accessKeyId, String secretAccessKey, String sessionToken, long
expireTimeInMs) {
- validate(accessKeyId, secretAccessKey, sessionToken);
+ validate(accessKeyId, secretAccessKey, sessionToken, expireTimeInMs);
this.accessKeyId = accessKeyId;
this.secretAccessKey = secretAccessKey;
this.sessionToken = sessionToken;
@@ -84,7 +84,7 @@ public class AwsIrsaCredential implements Credential {
String accessKeyId = credentialInfo.get(ACCESS_KEY_ID);
String secretAccessKey = credentialInfo.get(SECRET_ACCESS_KEY);
String sessionToken = credentialInfo.get(SESSION_TOKEN);
- validate(accessKeyId, secretAccessKey, sessionToken);
+ validate(accessKeyId, secretAccessKey, sessionToken, expireTimeInMs);
this.accessKeyId = accessKeyId;
this.secretAccessKey = secretAccessKey;
this.sessionToken = sessionToken;
@@ -118,12 +118,15 @@ public class AwsIrsaCredential implements Credential {
return sessionToken;
}
- private void validate(String accessKeyId, String secretAccessKey, String
sessionToken) {
+ private void validate(
+ String accessKeyId, String secretAccessKey, String sessionToken, long
expireTimeInMs) {
Preconditions.checkArgument(
StringUtils.isNotBlank(accessKeyId), "Access key Id should not be
empty");
Preconditions.checkArgument(
StringUtils.isNotBlank(secretAccessKey), "Secret access key should not
be empty");
Preconditions.checkArgument(
StringUtils.isNotBlank(sessionToken), "Session token should not be
empty");
+ Preconditions.checkArgument(
+ expireTimeInMs > 0, "The expiration time of AwsIrsaCredential should
be greater than 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/GCSTokenCredential.java
b/api/src/main/java/org/apache/gravitino/credential/GCSTokenCredential.java
index d55799c52a..a3230645fe 100644
--- a/api/src/main/java/org/apache/gravitino/credential/GCSTokenCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/GCSTokenCredential.java
@@ -88,6 +88,6 @@ public class GCSTokenCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(token), "GCS session token should not be
empty");
Preconditions.checkArgument(
- expireTimeInMs > 0, "The expire time of GcsTokenCredential should
great than 0");
+ expireTimeInMs > 0, "The expiration time of GcsTokenCredential should
be greater than 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/JdbcCredential.java
b/api/src/main/java/org/apache/gravitino/credential/JdbcCredential.java
index eeaccc2620..af19110d7e 100644
--- a/api/src/main/java/org/apache/gravitino/credential/JdbcCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/JdbcCredential.java
@@ -111,6 +111,6 @@ public class JdbcCredential implements Credential {
StringUtils.isNotBlank(jdbcPassword), "JDBC password should not be
empty");
// JDBC credentials are static (no server-issued expiry). expireTimeInMs
must always be 0.
Preconditions.checkArgument(
- expireTimeInMs == 0, "The expire time of JdbcCredential should be 0");
+ expireTimeInMs == 0, "The expiration time of JdbcCredential should be
0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/OSSSecretKeyCredential.java
b/api/src/main/java/org/apache/gravitino/credential/OSSSecretKeyCredential.java
index fd98a3bfd3..34baa51707 100644
---
a/api/src/main/java/org/apache/gravitino/credential/OSSSecretKeyCredential.java
+++
b/api/src/main/java/org/apache/gravitino/credential/OSSSecretKeyCredential.java
@@ -115,6 +115,6 @@ public class OSSSecretKeyCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(secretAccessKey), "OSS secret access key should
not empty");
Preconditions.checkArgument(
- expireTimeInMs == 0, "The expire time of OSSSecretKeyCredential is not
0");
+ expireTimeInMs == 0, "The expiration time of OSSSecretKeyCredential is
not 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/OSSTokenCredential.java
b/api/src/main/java/org/apache/gravitino/credential/OSSTokenCredential.java
index 70e8839489..4f1a871501 100644
--- a/api/src/main/java/org/apache/gravitino/credential/OSSTokenCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/OSSTokenCredential.java
@@ -133,6 +133,6 @@ public class OSSTokenCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(sessionToken), "OSS session token should not be
empty");
Preconditions.checkArgument(
- expireTimeInMs > 0, "The expire time of OSSTokenCredential should
great than 0");
+ expireTimeInMs > 0, "The expiration time of OSSTokenCredential should
be greater than 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/S3SecretKeyCredential.java
b/api/src/main/java/org/apache/gravitino/credential/S3SecretKeyCredential.java
index d9b2b60925..db6de05854 100644
---
a/api/src/main/java/org/apache/gravitino/credential/S3SecretKeyCredential.java
+++
b/api/src/main/java/org/apache/gravitino/credential/S3SecretKeyCredential.java
@@ -107,6 +107,6 @@ public class S3SecretKeyCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(secretAccessKey), "S3 secret access key should
not empty");
Preconditions.checkArgument(
- expireTimeInMs == 0, "The expire time of S3SecretKeyCredential is not
0");
+ expireTimeInMs == 0, "The expiration time of S3SecretKeyCredential is
not 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/S3TokenCredential.java
b/api/src/main/java/org/apache/gravitino/credential/S3TokenCredential.java
index 89d7de760f..ce69ab3b45 100644
--- a/api/src/main/java/org/apache/gravitino/credential/S3TokenCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/S3TokenCredential.java
@@ -131,6 +131,6 @@ public class S3TokenCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(sessionToken), "S3 session token should not be
empty");
Preconditions.checkArgument(
- expireTimeInMs > 0, "The expire time of S3TokenCredential should great
than 0");
+ expireTimeInMs > 0, "The expiration time of S3TokenCredential should
be greater than 0");
}
}
diff --git
a/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/AwsIrsaCredentialGenerator.java
b/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/AwsIrsaCredentialGenerator.java
index cd5ccad8c5..3c1641506e 100644
---
a/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/AwsIrsaCredentialGenerator.java
+++
b/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/AwsIrsaCredentialGenerator.java
@@ -81,10 +81,12 @@ public class AwsIrsaCredentialGenerator implements
CredentialGenerator<AwsIrsaCr
AwsCredentials creds = baseCredentialsProvider.resolveCredentials();
if (creds instanceof AwsSessionCredentials) {
AwsSessionCredentials sessionCreds = (AwsSessionCredentials) creds;
- long expiration =
- sessionCreds.expirationTime().isPresent()
- ? sessionCreds.expirationTime().get().toEpochMilli()
- : 0L;
+ if (!sessionCreds.expirationTime().isPresent()) {
+ throw new IllegalStateException(
+ "AWS IRSA session credentials must include an expiration time
for vended credential"
+ + " refresh");
+ }
+ long expiration = sessionCreds.expirationTime().get().toEpochMilli();
return new AwsIrsaCredential(
sessionCreds.accessKeyId(),
sessionCreds.secretAccessKey(),
diff --git
a/common/src/main/java/org/apache/gravitino/credential/CredentialPropertyUtils.java
b/common/src/main/java/org/apache/gravitino/credential/CredentialPropertyUtils.java
index b66b650178..d9543b93d3 100644
---
a/common/src/main/java/org/apache/gravitino/credential/CredentialPropertyUtils.java
+++
b/common/src/main/java/org/apache/gravitino/credential/CredentialPropertyUtils.java
@@ -23,6 +23,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
@@ -35,12 +36,22 @@ public class CredentialPropertyUtils {
@VisibleForTesting static final String ICEBERG_S3_ACCESS_KEY_ID =
"s3.access-key-id";
@VisibleForTesting static final String ICEBERG_S3_SECRET_ACCESS_KEY =
"s3.secret-access-key";
@VisibleForTesting static final String ICEBERG_S3_TOKEN = "s3.session-token";
- @VisibleForTesting static final String ICEBERG_GCS_TOKEN =
"gcs.oauth2.token";
+
+ @VisibleForTesting
+ static final String ICEBERG_S3_TOKEN_EXPIRES_AT_MS =
"s3.session-token-expires-at-ms";
+
+ @VisibleForTesting
+ static final String ICEBERG_CLIENT_REFRESH_CREDENTIALS_ENDPOINT =
+ "client.refresh-credentials-endpoint";
@VisibleForTesting static final String ICEBERG_OSS_ACCESS_KEY_ID =
"client.access-key-id";
@VisibleForTesting static final String ICEBERG_OSS_ACCESS_KEY_SECRET =
"client.access-key-secret";
@VisibleForTesting static final String ICEBERG_OSS_SECURITY_TOKEN =
"client.security-token";
+ @VisibleForTesting
+ static final String ICEBERG_OSS_SECURITY_TOKEN_EXPIRES_AT_MS =
+ "client.security-token-expires-at-ms";
+
@VisibleForTesting static final String ICEBERG_ADLS_TOKEN = "adls.sas-token";
@VisibleForTesting
@@ -49,7 +60,21 @@ public class CredentialPropertyUtils {
@VisibleForTesting
static final String ICEBERG_ADLS_ACCOUNT_KEY =
"adls.auth.shared-key.account.key";
- private static final String GCS_OAUTH_2_TOKEN_EXPIRES_AT =
"gcs.oauth2.token-expires-at";
+ @VisibleForTesting
+ static final String ICEBERG_ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX =
"adls.sas-token-expires-at-ms.";
+
+ @VisibleForTesting
+ static final String ICEBERG_ADLS_REFRESH_CREDENTIALS_ENDPOINT =
+ "adls.refresh-credentials-endpoint";
+
+ @VisibleForTesting static final String ICEBERG_GCS_TOKEN =
"gcs.oauth2.token";
+
+ @VisibleForTesting
+ static final String ICEBERG_GCS_TOKEN_EXPIRES_AT =
"gcs.oauth2.token-expires-at";
+
+ @VisibleForTesting
+ static final String ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT =
+ "gcs.oauth2.refresh-credentials-endpoint";
private static Map<String, String> icebergCredentialPropertyMap =
ImmutableMap.<String, String>builder()
@@ -82,36 +107,40 @@ public class CredentialPropertyUtils {
* @return a map of Iceberg properties derived from the credential
*/
public static Map<String, String> toIcebergProperties(Credential credential)
{
- if (credential instanceof S3TokenCredential
- || credential instanceof S3SecretKeyCredential
- || credential instanceof OSSTokenCredential
- || credential instanceof OSSSecretKeyCredential
- || credential instanceof AzureAccountKeyCredential
- || credential instanceof AwsIrsaCredential) {
- return transformProperties(credential.credentialInfo(),
icebergCredentialPropertyMap);
- }
-
- if (credential instanceof GCSTokenCredential) {
- Map<String, String> icebergGCSCredentialProperties =
- transformProperties(credential.credentialInfo(),
icebergCredentialPropertyMap);
- icebergGCSCredentialProperties.put(
- GCS_OAUTH_2_TOKEN_EXPIRES_AT,
String.valueOf(credential.expireTimeInMs()));
- return icebergGCSCredentialProperties;
- }
-
if (credential instanceof ADLSTokenCredential) {
ADLSTokenCredential adlsCredential = (ADLSTokenCredential) credential;
- String sasTokenKey =
- String.format(
- "%s.%s.%s",
- ICEBERG_ADLS_TOKEN, adlsCredential.accountName(),
ADLSTokenCredential.ADLS_DOMAIN);
+ String adlsHost = adlsCredential.accountName() + "." +
ADLSTokenCredential.ADLS_DOMAIN;
Map<String, String> icebergADLSCredentialProperties = new HashMap<>();
- icebergADLSCredentialProperties.put(sasTokenKey,
adlsCredential.sasToken());
+ icebergADLSCredentialProperties.put(
+ ICEBERG_ADLS_TOKEN + "." + adlsHost, adlsCredential.sasToken());
+ icebergADLSCredentialProperties.put(
+ ICEBERG_ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX + adlsHost,
+ String.valueOf(adlsCredential.expireTimeInMs()));
return icebergADLSCredentialProperties;
}
- return credential.toProperties();
+ Map<String, String> icebergProperties =
+ transformProperties(credential.credentialInfo(),
icebergCredentialPropertyMap);
+ if (credential instanceof S3TokenCredential || credential instanceof
AwsIrsaCredential) {
+ icebergProperties.put(
+ ICEBERG_S3_TOKEN_EXPIRES_AT_MS,
String.valueOf(credential.expireTimeInMs()));
+ return icebergProperties;
+ } else if (credential instanceof OSSTokenCredential) {
+ icebergProperties.put(
+ ICEBERG_OSS_SECURITY_TOKEN_EXPIRES_AT_MS,
String.valueOf(credential.expireTimeInMs()));
+ return icebergProperties;
+ } else if (credential instanceof GCSTokenCredential) {
+ icebergProperties.put(
+ ICEBERG_GCS_TOKEN_EXPIRES_AT,
String.valueOf(credential.expireTimeInMs()));
+ return icebergProperties;
+ } else if (credential instanceof S3SecretKeyCredential
+ || credential instanceof OSSSecretKeyCredential
+ || credential instanceof AzureAccountKeyCredential) {
+ return icebergProperties;
+ } else {
+ return credential.toProperties();
+ }
}
/**
@@ -126,20 +155,69 @@ public class CredentialPropertyUtils {
*/
public static Map<String, String> filterCredentialProperties(Map<String,
String> properties) {
Set<String> credentialPropertyKeys =
Sets.newHashSet(icebergCredentialPropertyMap.values());
- credentialPropertyKeys.add(GCS_OAUTH_2_TOKEN_EXPIRES_AT);
+ credentialPropertyKeys.add(ICEBERG_S3_TOKEN_EXPIRES_AT_MS);
+ credentialPropertyKeys.add(ICEBERG_OSS_SECURITY_TOKEN_EXPIRES_AT_MS);
+ credentialPropertyKeys.add(ICEBERG_GCS_TOKEN_EXPIRES_AT);
Map<String, String> filteredProperties = Maps.newHashMap(properties);
filteredProperties
.entrySet()
.removeIf(
entry ->
!credentialPropertyKeys.contains(entry.getKey())
- && !entry.getKey().startsWith(ICEBERG_ADLS_TOKEN));
+ && !entry.getKey().startsWith(ICEBERG_ADLS_TOKEN)
+ &&
!entry.getKey().startsWith(ICEBERG_ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX));
return filteredProperties;
}
+ /**
+ * Builds refresh credential endpoint properties for Iceberg credential
properties.
+ *
+ * @param encodedCatalogName Iceberg REST encoded catalog name
+ * @param encodedNamespace Iceberg REST encoded namespace
+ * @param encodedTableName Iceberg REST encoded table name
+ * @param credentialProperties Iceberg credential properties used to
determine refresh keys
+ * @return refresh endpoint properties keyed by Iceberg client config names
+ */
+ public static Map<String, String> buildRefreshProps(
+ String encodedCatalogName,
+ String encodedNamespace,
+ String encodedTableName,
+ Map<String, String> credentialProperties) {
+ if (credentialProperties == null || credentialProperties.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ String credentialsRefreshPath =
+ String.format(
+ "v1/%s/namespaces/%s/tables/%s/credentials",
+ encodedCatalogName, encodedNamespace, encodedTableName);
+ Map<String, String> refreshEndpointProperties = Maps.newHashMap();
+ if (credentialProperties.containsKey(ICEBERG_S3_TOKEN)) {
+ refreshEndpointProperties.put(
+ ICEBERG_CLIENT_REFRESH_CREDENTIALS_ENDPOINT, credentialsRefreshPath);
+ }
+ if (credentialProperties.containsKey(ICEBERG_GCS_TOKEN)) {
+ refreshEndpointProperties.put(
+ ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT,
credentialsRefreshPath);
+ }
+ if (credentialProperties.keySet().stream()
+ .anyMatch(key -> key.startsWith(ICEBERG_ADLS_TOKEN + "."))) {
+ refreshEndpointProperties.put(
+ ICEBERG_ADLS_REFRESH_CREDENTIALS_ENDPOINT, credentialsRefreshPath);
+ }
+ return ImmutableMap.copyOf(refreshEndpointProperties);
+ }
+
+ /**
+ * Transforms Gravitino credential keys to Iceberg property keys.
+ *
+ * @param originProperties the source credential info map
+ * @param transformMap mapping from Gravitino keys to Iceberg keys
+ * @return a new mutable map of transformed Iceberg properties
+ */
private static Map<String, String> transformProperties(
Map<String, String> originProperties, Map<String, String> transformMap) {
- HashMap<String, String> properties = new HashMap();
+ Map<String, String> properties = new HashMap<>();
originProperties.forEach(
(k, v) -> {
if (transformMap.containsKey(k)) {
diff --git
a/common/src/test/java/org/apache/gravitino/credential/TestCredentialFactory.java
b/common/src/test/java/org/apache/gravitino/credential/TestCredentialFactory.java
index 443399daf6..259949ae2b 100644
---
a/common/src/test/java/org/apache/gravitino/credential/TestCredentialFactory.java
+++
b/common/src/test/java/org/apache/gravitino/credential/TestCredentialFactory.java
@@ -26,6 +26,30 @@ import org.junit.jupiter.api.Test;
public class TestCredentialFactory {
+ @Test
+ void testAwsIrsaCredential() {
+ Map<String, String> awsIrsaCredentialInfo =
+ ImmutableMap.of(
+ AwsIrsaCredential.ACCESS_KEY_ID,
+ "accessKeyId",
+ AwsIrsaCredential.SECRET_ACCESS_KEY,
+ "secretAccessKey",
+ AwsIrsaCredential.SESSION_TOKEN,
+ "token");
+ long expireTime = 1000;
+ Credential awsIrsaCredential =
+ CredentialFactory.create(
+ AwsIrsaCredential.AWS_IRSA_CREDENTIAL_TYPE, awsIrsaCredentialInfo,
expireTime);
+ Assertions.assertEquals(
+ AwsIrsaCredential.AWS_IRSA_CREDENTIAL_TYPE,
awsIrsaCredential.credentialType());
+ Assertions.assertInstanceOf(AwsIrsaCredential.class, awsIrsaCredential);
+ AwsIrsaCredential awsIrsaCredential1 = (AwsIrsaCredential)
awsIrsaCredential;
+ Assertions.assertEquals("accessKeyId", awsIrsaCredential1.accessKeyId());
+ Assertions.assertEquals("secretAccessKey",
awsIrsaCredential1.secretAccessKey());
+ Assertions.assertEquals("token", awsIrsaCredential1.sessionToken());
+ Assertions.assertEquals(expireTime, awsIrsaCredential1.expireTimeInMs());
+ }
+
@Test
void testS3TokenCredential() {
Map<String, String> s3TokenCredentialInfo =
diff --git
a/common/src/test/java/org/apache/gravitino/credential/TestCredentialPropertiesUtils.java
b/common/src/test/java/org/apache/gravitino/credential/TestCredentialPropertiesUtils.java
index cb5eaabe76..0ee13adac5 100644
---
a/common/src/test/java/org/apache/gravitino/credential/TestCredentialPropertiesUtils.java
+++
b/common/src/test/java/org/apache/gravitino/credential/TestCredentialPropertiesUtils.java
@@ -38,7 +38,9 @@ public class TestCredentialPropertiesUtils {
CredentialPropertyUtils.ICEBERG_S3_SECRET_ACCESS_KEY,
"secret",
CredentialPropertyUtils.ICEBERG_S3_TOKEN,
- "token");
+ "token",
+ CredentialPropertyUtils.ICEBERG_S3_TOKEN_EXPIRES_AT_MS,
+ "100");
Assertions.assertEquals(expectedProperties, icebergProperties);
S3SecretKeyCredential secretKeyCredential = new
S3SecretKeyCredential("key", "secret");
@@ -65,7 +67,9 @@ public class TestCredentialPropertiesUtils {
CredentialPropertyUtils.ICEBERG_OSS_ACCESS_KEY_SECRET,
"secret",
CredentialPropertyUtils.ICEBERG_OSS_SECURITY_TOKEN,
- "security-token");
+ "security-token",
+ CredentialPropertyUtils.ICEBERG_OSS_SECURITY_TOKEN_EXPIRES_AT_MS,
+ "100");
Assertions.assertEquals(expectedProperties, icebergProperties);
}
@@ -80,14 +84,67 @@ public class TestCredentialPropertiesUtils {
Map<String, String> icebergProperties =
CredentialPropertyUtils.toIcebergProperties(adlsTokenCredential);
- String sasTokenKey =
- String.format(
- "%s.%s.%s",
- CredentialPropertyUtils.ICEBERG_ADLS_TOKEN,
- storageAccountName,
- ADLSTokenCredential.ADLS_DOMAIN);
+ String adlsHost = storageAccountName + "." +
ADLSTokenCredential.ADLS_DOMAIN;
+ String sasTokenKey = CredentialPropertyUtils.ICEBERG_ADLS_TOKEN + "." +
adlsHost;
- Map<String, String> expectedProperties = ImmutableMap.of(sasTokenKey,
sasToken);
+ Map<String, String> expectedProperties =
+ ImmutableMap.of(
+ sasTokenKey,
+ sasToken,
+
CredentialPropertyUtils.ICEBERG_ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX + adlsHost,
+ String.valueOf(expireTimeInMS));
Assertions.assertEquals(expectedProperties, icebergProperties);
}
+
+ @Test
+ void testBuildRefreshEndpointsForS3() {
+ Map<String, String> credentialProperties =
+ ImmutableMap.of(
+ CredentialPropertyUtils.ICEBERG_S3_TOKEN, "token",
+ CredentialPropertyUtils.ICEBERG_S3_TOKEN_EXPIRES_AT_MS, "123");
+
+ Map<String, String> refreshEndpointProperties =
+ CredentialPropertyUtils.buildRefreshProps("irc1", "db", "tbl",
credentialProperties);
+
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ refreshEndpointProperties.get(
+
CredentialPropertyUtils.ICEBERG_CLIENT_REFRESH_CREDENTIALS_ENDPOINT));
+ }
+
+ @Test
+ void testBuildRefreshEndpointsForGCS() {
+ Map<String, String> credentialProperties =
+ ImmutableMap.of(
+ CredentialPropertyUtils.ICEBERG_GCS_TOKEN, "token",
+ CredentialPropertyUtils.ICEBERG_GCS_TOKEN_EXPIRES_AT, "123");
+
+ Map<String, String> refreshEndpointProperties =
+ CredentialPropertyUtils.buildRefreshProps("irc1", "db", "tbl",
credentialProperties);
+
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ refreshEndpointProperties.get(
+
CredentialPropertyUtils.ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT));
+ }
+
+ @Test
+ void testBuildRefreshEndpointsForADLS() {
+ String adlsHost = "storage.dfs.core.windows.net";
+ String sasTokenKey = CredentialPropertyUtils.ICEBERG_ADLS_TOKEN + "." +
adlsHost;
+ Map<String, String> credentialProperties =
+ ImmutableMap.of(
+ sasTokenKey,
+ "sas-token",
+
CredentialPropertyUtils.ICEBERG_ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX + adlsHost,
+ "123");
+
+ Map<String, String> refreshEndpointProperties =
+ CredentialPropertyUtils.buildRefreshProps("irc1", "db", "tbl",
credentialProperties);
+
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ refreshEndpointProperties.get(
+
CredentialPropertyUtils.ICEBERG_ADLS_REFRESH_CREDENTIALS_ENDPOINT));
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/credential/CatalogCredentialManager.java
b/core/src/main/java/org/apache/gravitino/credential/CatalogCredentialManager.java
index 88892391c4..9fb5a9eecf 100644
---
a/core/src/main/java/org/apache/gravitino/credential/CatalogCredentialManager.java
+++
b/core/src/main/java/org/apache/gravitino/credential/CatalogCredentialManager.java
@@ -49,6 +49,15 @@ public class CatalogCredentialManager implements Closeable {
credentialCache.initialize(catalogProperties);
}
+ /**
+ * Returns the catalog name this manager was created for.
+ *
+ * @return catalog name
+ */
+ public String catalogName() {
+ return catalogName;
+ }
+
public Optional<Credential> getCredential(String credentialType,
CredentialContext context) {
CredentialCacheKey credentialCacheKey = new
CredentialCacheKey(credentialType, context);
return credentialCache.getCredential(credentialCacheKey, cacheKey ->
doGetCredential(cacheKey));
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
index 39fc375d7b..7e1aac0c0d 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
@@ -57,6 +57,7 @@ import org.apache.gravitino.utils.PrincipalUtils;
import org.apache.iceberg.BaseMetadataTable;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.BaseTransaction;
+import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.IncrementalAppendScan;
import org.apache.iceberg.MetadataUpdate;
@@ -81,8 +82,16 @@ import org.apache.iceberg.inmemory.InMemoryFileIO;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.rest.CatalogHandlers;
+import org.apache.iceberg.rest.ErrorHandlers;
+import org.apache.iceberg.rest.HTTPClient;
import org.apache.iceberg.rest.PlanStatus;
import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.iceberg.rest.RESTClient;
+import org.apache.iceberg.rest.RESTUtil;
+import org.apache.iceberg.rest.ResourcePaths;
+import org.apache.iceberg.rest.auth.AuthManager;
+import org.apache.iceberg.rest.auth.AuthManagers;
+import org.apache.iceberg.rest.auth.AuthSession;
import org.apache.iceberg.rest.requests.CreateTableRequest;
import org.apache.iceberg.rest.requests.PlanTableScanRequest;
import org.apache.iceberg.rest.requests.RegisterTableRequest;
@@ -192,38 +201,92 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
/**
* Get table credentials.
*
- * @param identifier The table identifier for which to load credentials
- * @return A {@link
org.apache.iceberg.rest.responses.LoadCredentialsResponse} object containing
- * the credentials.
+ * @param identifier table identifier
+ * @param privilege used for local credential vending; ignored for REST
catalog backends
+ * @return table credentials response
*/
public LoadCredentialsResponse getTableCredentials(
TableIdentifier identifier, CredentialPrivilege privilege) {
+ if (isRESTCatalog()) {
+ return getRESTTableCredentials((RESTCatalog) getCatalog(), identifier);
+ } else {
+ return getLocalTableCredentials(identifier, privilege);
+ }
+ }
+
+ private LoadCredentialsResponse getLocalTableCredentials(
+ TableIdentifier identifier, CredentialPrivilege privilege) {
try {
LoadTableResponse loadTableResponse = super.loadTable(identifier);
- Credential credential = getCredential(loadTableResponse, privilege);
- org.apache.iceberg.rest.credentials.Credential icebergCredential =
- new org.apache.iceberg.rest.credentials.Credential() {
- @Override
- public String prefix() {
- return "";
- }
-
- @Override
- public Map<String, String> config() {
- // Convert Gravitino credentials to the Iceberg REST credential
payload format.
- return CredentialPropertyUtils.toIcebergProperties(credential);
- }
-
- @Override
- public void validate() {}
- };
- return
ImmutableLoadCredentialsResponse.builder().addCredentials(icebergCredential).build();
+ Credential credential = getCredential(loadTableResponse.tableMetadata(),
privilege);
+ return ImmutableLoadCredentialsResponse.builder()
+ .addCredentials(
+ IcebergRESTUtils.toRESTCredential(
+ catalogCredentialManager.catalogName(),
+ identifier,
+ credential,
+ loadTableResponse.tableMetadata()))
+ .build();
} catch (ServiceUnavailableException e) {
LOG.warn("Service unavailable when loading table credentials for table:
{}", identifier, e);
return ImmutableLoadCredentialsResponse.builder().build();
}
}
+ private static LoadCredentialsResponse getRESTTableCredentials(
+ RESTCatalog restCatalog, TableIdentifier identifier) {
+ Map<String, String> properties = Maps.newHashMap(restCatalog.properties());
+ String credentialsPath =
+ ResourcePaths.forCatalogProperties(properties).table(identifier) +
"/credentials";
+
+ AuthManager authManager = null;
+ RESTClient client = null;
+ AuthSession authSession = null;
+ try {
+ authManager = AuthManagers.loadAuthManager(restCatalog.name(),
properties);
+ client =
+ HTTPClient.builder(properties)
+ .uri(properties.get(CatalogProperties.URI))
+ .withHeaders(RESTUtil.configHeaders(properties))
+ .build();
+ authSession = authManager.catalogSession(client, properties);
+ return client
+ .withAuthSession(authSession)
+ .get(
+ credentialsPath,
+ LoadCredentialsResponse.class,
+ Collections.emptyMap(),
+ ErrorHandlers.tableErrorHandler());
+ } finally {
+ if (authSession != null) {
+ try {
+ authSession.close();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to close auth session when loading credentials for
table: {}", identifier, e);
+ }
+ }
+
+ if (client != null) {
+ try {
+ client.close();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to close REST client when loading credentials for table:
{}", identifier, e);
+ }
+ }
+
+ if (authManager != null) {
+ try {
+ authManager.close();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to close auth manager when loading credentials for
table: {}", identifier, e);
+ }
+ }
+ }
+ }
+
@Override
public void close() throws Exception {
try {
@@ -318,7 +381,7 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
TableIdentifier tableIdentifier,
LoadTableResponse loadTableResponse,
CredentialPrivilege privilege) {
- final Credential credential = getCredential(loadTableResponse, privilege);
+ final Credential credential =
getCredential(loadTableResponse.tableMetadata(), privilege);
LOG.info(
"Generate credential: {} for Iceberg table: {}",
@@ -327,7 +390,13 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
// Merge temporary credential fields as Iceberg client config entries in
the load-table
// response.
- Map<String, String> credentialConfig =
CredentialPropertyUtils.toIcebergProperties(credential);
+ Map<String, String> credentialConfig =
+ IcebergRESTUtils.toRESTCredential(
+ catalogCredentialManager.catalogName(),
+ tableIdentifier,
+ credential,
+ loadTableResponse.tableMetadata())
+ .config();
return LoadTableResponse.builder()
.withTableMetadata(loadTableResponse.tableMetadata())
.addAllConfig(loadTableResponse.config())
@@ -336,9 +405,7 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
.build();
}
- private Credential getCredential(
- LoadTableResponse loadTableResponse, CredentialPrivilege privilege) {
- TableMetadata tableMetadata = loadTableResponse.tableMetadata();
+ private Credential getCredential(TableMetadata tableMetadata,
CredentialPrivilege privilege) {
String[] path =
Stream.of(
tableMetadata.location(),
@@ -685,13 +752,22 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
if (table instanceof BaseTable) {
Map<String, String> properties = retrieveFileIOProperties(table.io());
+ Map<String, String> filteredCredentialProperties =
+ CredentialPropertyUtils.filterCredentialProperties(properties);
return LoadTableResponse.builder()
.withTableMetadata(((BaseTable) table).operations().current())
.addAllConfig(
MapUtils.getFilteredMap(
properties, key ->
catalogPropertiesToClientKeys.contains(key)))
- // Keep only credential fields from FileIO properties before
returning them to the client.
-
.addAllConfig(CredentialPropertyUtils.filterCredentialProperties(properties))
+ // Keep only credential fields from FileIO properties before
returning them to the
+ // client.
+ .addAllConfig(filteredCredentialProperties)
+ .addAllConfig(
+ IcebergRESTUtils.buildRefreshProps(
+ catalogCredentialManager.catalogName(), ident,
filteredCredentialProperties))
+ .addAllCredentials(
+ IcebergRESTUtils.buildStorageCreds(
+ catalogCredentialManager.catalogName(), ident, table.io()))
.build();
}
@@ -726,10 +802,19 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
}
Map<String, String> tableProperties = retrieveFileIOProperties(table.io());
+ Map<String, String> filteredCredentialProperties =
+ CredentialPropertyUtils.filterCredentialProperties(tableProperties);
config.putAll(
MapUtils.getFilteredMap(
tableProperties, key ->
catalogPropertiesToClientKeys.contains(key)));
-
config.putAll(CredentialPropertyUtils.filterCredentialProperties(tableProperties));
+ config.putAll(filteredCredentialProperties);
+ config.putAll(
+ IcebergRESTUtils.buildRefreshProps(
+ catalogCredentialManager.catalogName(), ident,
filteredCredentialProperties));
+
+ List<org.apache.iceberg.rest.credentials.Credential> credentials =
+ IcebergRESTUtils.buildStorageCreds(
+ catalogCredentialManager.catalogName(), ident, table.io());
TableMetadata metadata =
TableMetadata.newTableMetadata(
@@ -739,7 +824,11 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
table.location(),
properties);
- return
LoadTableResponse.builder().withTableMetadata(metadata).addAllConfig(config).build();
+ return LoadTableResponse.builder()
+ .withTableMetadata(metadata)
+ .addAllConfig(config)
+ .addAllCredentials(credentials)
+ .build();
}
private LoadTableResponse tableUpdateInternal(TableIdentifier ident,
UpdateTableRequest request) {
@@ -793,13 +882,21 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
if (table instanceof BaseTable) {
Map<String, String> properties = retrieveFileIOProperties(table.io());
+ Map<String, String> filteredCredentialProperties =
+ CredentialPropertyUtils.filterCredentialProperties(properties);
return LoadTableResponse.builder()
.withTableMetadata(((BaseTable) table).operations().current())
.addAllConfig(
MapUtils.getFilteredMap(
properties, key ->
catalogPropertiesToClientKeys.contains(key)))
// Keep only credential fields from FileIO properties before
returning them to the client.
-
.addAllConfig(CredentialPropertyUtils.filterCredentialProperties(properties))
+ .addAllConfig(filteredCredentialProperties)
+ .addAllConfig(
+ IcebergRESTUtils.buildRefreshProps(
+ catalogCredentialManager.catalogName(), ident,
filteredCredentialProperties))
+ .addAllCredentials(
+ IcebergRESTUtils.buildStorageCreds(
+ catalogCredentialManager.catalogName(), ident, table.io()))
.build();
} else if (table instanceof BaseMetadataTable) {
// metadata tables are loaded on the client side, return
NoSuchTableException for now
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
index c095ac231b..465ebfc138 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
@@ -20,6 +20,7 @@ package org.apache.gravitino.iceberg.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
@@ -27,9 +28,12 @@ import java.security.NoSuchAlgorithmException;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Stream;
@@ -40,9 +44,15 @@ import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.credential.CredentialPropertyUtils;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.StorageCredential;
+import org.apache.iceberg.io.SupportsStorageCredentials;
import org.apache.iceberg.rest.RESTUtil;
import org.apache.iceberg.rest.responses.ErrorResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -83,6 +93,101 @@ public class IcebergRESTUtils {
private IcebergRESTUtils() {}
+ /**
+ * Builds an Iceberg REST {@link
org.apache.iceberg.rest.credentials.Credential} for load-table,
+ * scan-plan, or credentials API responses.
+ *
+ * <p>Refresh credential endpoints are added only for credential types whose
Iceberg client
+ * modules support vended credential refresh: S3 ({@link
+ * org.apache.gravitino.credential.S3TokenCredential}, {@link
+ * org.apache.gravitino.credential.AwsIrsaCredential}), GCS ({@link
+ * org.apache.gravitino.credential.GCSTokenCredential}), and ADLS ({@link
+ * org.apache.gravitino.credential.ADLSTokenCredential}). OSS temporary
credentials ({@link
+ * org.apache.gravitino.credential.OSSTokenCredential}) are omitted because
Iceberg's Aliyun OSS
+ * FileIO does not consume {@code client.refresh-credentials-endpoint}; only
the initial STS
+ * properties are returned.
+ *
+ * @param catalogName IRC catalog name used in the refresh path
+ * @param tableIdentifier table receiving the credential
+ * @param credential Gravitino credential to vend
+ * @param tableMetadata table metadata used to derive the storage prefix
+ * @return Iceberg REST credential with prefix and config
+ */
+ public static org.apache.iceberg.rest.credentials.Credential
toRESTCredential(
+ String catalogName,
+ TableIdentifier tableIdentifier,
+ Credential credential,
+ TableMetadata tableMetadata) {
+ Map<String, String> config =
+ new HashMap<>(CredentialPropertyUtils.toIcebergProperties(credential));
+ config.putAll(buildRefreshProps(catalogName, tableIdentifier, config));
+
+ return toRESTCredential(tableMetadata.location(), config);
+ }
+
+ /**
+ * Builds an Iceberg REST {@link
org.apache.iceberg.rest.credentials.Credential} from a storage
+ * prefix and credential config map.
+ *
+ * @param prefix storage location prefix for the credential
+ * @param config Iceberg credential config properties
+ * @return Iceberg REST credential
+ */
+ public static org.apache.iceberg.rest.credentials.Credential
toRESTCredential(
+ String prefix, Map<String, String> config) {
+ Map<String, String> credentialConfig = ImmutableMap.copyOf(config);
+ return new org.apache.iceberg.rest.credentials.Credential() {
+ @Override
+ public String prefix() {
+ return prefix;
+ }
+
+ @Override
+ public Map<String, String> config() {
+ return credentialConfig;
+ }
+
+ @Override
+ public void validate() {}
+ };
+ }
+
+ /**
+ * Builds Iceberg REST 1.11 {@code storage-credentials} from an upstream
REST catalog proxy.
+ *
+ * <p>Upstream credentials are read from {@link
SupportsStorageCredentials#credentials()}. When
+ * present, they are filtered and rewritten with IRC-local refresh endpoints
via {@link
+ * CredentialPropertyUtils}. Credential properties in {@link
FileIO#properties()} must be handled
+ * separately by the caller.
+ *
+ * @param catalogName IRC catalog name used to build refresh paths
+ * @param tableIdentifier table receiving the credentials
+ * @param fileIO table FileIO returned by the upstream catalog
+ * @return rewritten storage credentials for the downstream client, or an
empty list
+ */
+ public static List<org.apache.iceberg.rest.credentials.Credential>
buildStorageCreds(
+ String catalogName, TableIdentifier tableIdentifier, FileIO fileIO) {
+ if (!(fileIO instanceof SupportsStorageCredentials)) {
+ return Collections.emptyList();
+ }
+
+ List<StorageCredential> credentials = ((SupportsStorageCredentials)
fileIO).credentials();
+ if (credentials == null || credentials.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ List<org.apache.iceberg.rest.credentials.Credential> restCredentials = new
ArrayList<>();
+ for (StorageCredential credential : credentials) {
+ Map<String, String> filteredConfig =
+
CredentialPropertyUtils.filterCredentialProperties(credential.config());
+ filteredConfig.putAll(buildRefreshProps(catalogName, tableIdentifier,
filteredConfig));
+ restCredentials.add(
+ toRESTCredential(credential.prefix(),
ImmutableMap.copyOf(filteredConfig)));
+ }
+
+ return List.copyOf(restCredentials);
+ }
+
public static <T> Response ok(T t) {
return
Response.status(Response.Status.OK).entity(t).type(MediaType.APPLICATION_JSON).build();
}
@@ -262,6 +367,25 @@ public class IcebergRESTUtils {
return headers;
}
+ /**
+ * Builds refresh credential endpoint properties for a table, with Iceberg
REST path encoding.
+ *
+ * @param catalogName IRC catalog name used in the refresh path
+ * @param tableIdentifier table receiving the credentials
+ * @param credentialProperties Iceberg credential properties used to
determine refresh keys
+ * @return refresh endpoint properties keyed by Iceberg client config names
+ */
+ public static Map<String, String> buildRefreshProps(
+ String catalogName,
+ TableIdentifier tableIdentifier,
+ Map<String, String> credentialProperties) {
+ return CredentialPropertyUtils.buildRefreshProps(
+ RESTUtil.encodeString(catalogName),
+ RESTUtil.encodeNamespace(tableIdentifier.namespace(),
NAMESPACE_SEPARATOR_URLENCODED_UTF_8),
+ RESTUtil.encodeString(tableIdentifier.name()),
+ credentialProperties);
+ }
+
// remove the last '/' from the prefix, for example transform
'iceberg_catalog/' to
// 'iceberg_catalog'
private static String normalizePrefix(String rawPrefix) {
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
index b7432cdc31..55d30018a0 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
@@ -27,8 +27,13 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
import com.google.common.collect.ImmutableMap;
+import com.sun.net.httpserver.HttpServer;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -38,8 +43,13 @@ import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
+import org.apache.gravitino.credential.CredentialConstants;
+import org.apache.gravitino.credential.CredentialPrivilege;
import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.gravitino.iceberg.service.extension.DummyCredentialProvider;
+import org.apache.iceberg.BaseTable;
import org.apache.iceberg.BaseTransaction;
+import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.MetadataUpdate;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -51,12 +61,22 @@ import org.apache.iceberg.Transaction;
import org.apache.iceberg.UpdateRequirement;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SupportsNamespaces;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.ForbiddenException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.NotAuthorizedException;
+import org.apache.iceberg.exceptions.ServiceFailureException;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.ResolvingFileIO;
+import org.apache.iceberg.io.StorageCredential;
+import org.apache.iceberg.io.SupportsStorageCredentials;
import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.iceberg.rest.auth.AuthProperties;
+import org.apache.iceberg.rest.credentials.Credential;
import org.apache.iceberg.rest.requests.CreateTableRequest;
import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
@@ -101,6 +121,292 @@ public class TestCatalogWrapperForREST {
Assertions.assertFalse(CatalogWrapperForREST.isLocalOrHdfsLocation(" "));
}
+ @Test
+ void testRESTTableCredentials() throws Exception {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ String expectedPath = "/v1/upstream/namespaces/db/tables/tbl/credentials";
+ String upstreamJson =
+
"{\"storage-credentials\":[{\"prefix\":\"s3://upstream/db/tbl/\",\"config\":{"
+ + "\"s3.access-key-id\":\"upstream-key\","
+ + "\"s3.secret-access-key\":\"upstream-secret\","
+ + "\"s3.session-token\":\"upstream-token\"}}]}";
+
+ AtomicReference<String> requestPath = new AtomicReference<>();
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ requestPath.set(exchange.getRequestURI().getPath());
+ byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(200, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ LoadCredentialsResponse response =
+ wrapper.getTableCredentials(table, CredentialPrivilege.READ);
+
+ Assertions.assertEquals(expectedPath, requestPath.get());
+ Assertions.assertEquals(1, response.credentials().size());
+ Credential credential = response.credentials().get(0);
+ Assertions.assertEquals("s3://upstream/db/tbl/", credential.prefix());
+ Assertions.assertEquals("upstream-key",
credential.config().get("s3.access-key-id"));
+ Assertions.assertEquals("upstream-secret",
credential.config().get("s3.secret-access-key"));
+ Assertions.assertEquals("upstream-token",
credential.config().get("s3.session-token"));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testRESTTableCredentialsOnFailure() throws Exception {
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ exchange.sendResponseHeaders(500, -1);
+ exchange.close();
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ Assertions.assertThrows(
+ ServiceFailureException.class,
+ () ->
+ wrapper.getTableCredentials(
+ TableIdentifier.of(Namespace.of("db"), "tbl"),
CredentialPrivilege.READ));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testRESTTableCredentialsOnForbidden() throws Exception {
+ String errorJson =
+
"{\"error\":{\"message\":\"Forbidden\",\"type\":\"ForbiddenException\","
+ + "\"code\":403,\"stack\":[]}}";
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ byte[] body = errorJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(403, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ Assertions.assertThrows(
+ ForbiddenException.class,
+ () ->
+ wrapper.getTableCredentials(
+ TableIdentifier.of(Namespace.of("db"), "tbl"),
CredentialPrivilege.READ));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testRESTTableCredentialsOnUnauthorized() throws Exception {
+ String errorJson =
+ "{\"error\":{\"message\":\"Not
authorized\",\"type\":\"NotAuthorizedException\","
+ + "\"code\":401,\"stack\":[]}}";
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ byte[] body = errorJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(401, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ Assertions.assertThrows(
+ NotAuthorizedException.class,
+ () ->
+ wrapper.getTableCredentials(
+ TableIdentifier.of(Namespace.of("db"), "tbl"),
CredentialPrivilege.READ));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testRESTTableCredentialsNoSuchTable() throws Exception {
+ String errorJson =
+ "{\"error\":{\"message\":\"Table not
found\",\"type\":\"NoSuchTableException\","
+ + "\"code\":404,\"stack\":[]}}";
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ byte[] body = errorJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(404, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ Assertions.assertThrows(
+ NoSuchTableException.class,
+ () ->
+ wrapper.getTableCredentials(
+ TableIdentifier.of(Namespace.of("db"), "missing"),
CredentialPrivilege.READ));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testLocalTableCredentials() {
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse",
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ DummyCredentialProvider.DUMMY_CREDENTIAL_TYPE));
+
+ CatalogWrapperForREST wrapper = new CatalogWrapperForREST("local-catalog",
config);
+ Namespace namespace = Namespace.of("db");
+ Catalog catalog = wrapper.getCatalog();
+ ((SupportsNamespaces) catalog).createNamespace(namespace);
+ TableIdentifier table = TableIdentifier.of(namespace, "tbl");
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ catalog.createTable(
+ table,
+ schema,
+ PartitionSpec.unpartitioned(),
+ "s3://bucket/wh/db/tbl",
+ Collections.emptyMap());
+
+ LoadCredentialsResponse response = wrapper.getTableCredentials(table,
CredentialPrivilege.READ);
+
+ Assertions.assertEquals(1, response.credentials().size());
+ Assertions.assertEquals("s3://bucket/wh/db/tbl",
response.credentials().get(0).prefix());
+ }
+
@Test
void testValidateCredentialLocation() {
Assertions.assertDoesNotThrow(
@@ -116,7 +422,113 @@ public class TestCatalogWrapperForREST {
}
@Test
- void testRestCatalogClientConfigMergesRemote() {
+ void testLoadTableRefreshEndpoint() {
+ TableIdentifier ident = TableIdentifier.of(Namespace.of("db"), "tbl");
+ RESTCatalog catalog = mock(RESTCatalog.class);
+ BaseTable baseTable = mock(BaseTable.class);
+ TableOperations ops = mock(TableOperations.class);
+ FileIO fileIO = mock(FileIO.class);
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(
+ new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get())),
+ PartitionSpec.unpartitioned(),
+ SortOrder.unsorted(),
+ "s3://bucket/db/tbl",
+ Collections.emptyMap());
+
+ when(catalog.loadTable(ident)).thenReturn(baseTable);
+ when(baseTable.operations()).thenReturn(ops);
+ when(ops.current()).thenReturn(metadata);
+ when(baseTable.io()).thenReturn(fileIO);
+ when(fileIO.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ "s3.session-token",
+ "token",
+ "s3.session-token-expires-at-ms",
+ "123",
+ "client.refresh-credentials-endpoint",
+ "v1/upstream/namespaces/db/tables/tbl/credentials"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("irc1",
config, catalog);
+
+ LoadTableResponse response = wrapper.loadTable(ident, false,
CredentialPrivilege.READ);
+
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ response.config().get("client.refresh-credentials-endpoint"));
+ Assertions.assertEquals("token",
response.config().get("s3.session-token"));
+ }
+
+ @Test
+ void testLoadTableStorageCreds() {
+ TableIdentifier ident = TableIdentifier.of(Namespace.of("db"), "tbl");
+ RESTCatalog catalog = mock(RESTCatalog.class);
+ BaseTable baseTable = mock(BaseTable.class);
+ TableOperations ops = mock(TableOperations.class);
+ FileIO fileIO =
+ mock(FileIO.class,
withSettings().extraInterfaces(SupportsStorageCredentials.class));
+ SupportsStorageCredentials storageCredentialsFileIO =
(SupportsStorageCredentials) fileIO;
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(
+ new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get())),
+ PartitionSpec.unpartitioned(),
+ SortOrder.unsorted(),
+ "s3://bucket/db/tbl",
+ Collections.emptyMap());
+
+ StorageCredential upstreamCredential =
+ StorageCredential.create(
+ "s3://bucket/db/tbl/",
+ ImmutableMap.of(
+ "s3.access-key-id",
+ "upstream-key",
+ "s3.secret-access-key",
+ "upstream-secret",
+ "s3.session-token",
+ "upstream-token",
+ "s3.session-token-expires-at-ms",
+ "123",
+ "client.refresh-credentials-endpoint",
+ "v1/upstream/namespaces/db/tables/tbl/credentials"));
+
+ when(catalog.loadTable(ident)).thenReturn(baseTable);
+ when(baseTable.operations()).thenReturn(ops);
+ when(ops.current()).thenReturn(metadata);
+ when(baseTable.io()).thenReturn(fileIO);
+ when(fileIO.properties()).thenReturn(Collections.emptyMap());
+
when(storageCredentialsFileIO.credentials()).thenReturn(List.of(upstreamCredential));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("irc1",
config, catalog);
+
+ LoadTableResponse response = wrapper.loadTable(ident, false,
CredentialPrivilege.READ);
+
+ Assertions.assertEquals(1, response.credentials().size());
+ Credential credential = response.credentials().get(0);
+ Assertions.assertEquals("s3://bucket/db/tbl/", credential.prefix());
+ Assertions.assertEquals("upstream-token",
credential.config().get("s3.session-token"));
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ credential.config().get("client.refresh-credentials-endpoint"));
+
Assertions.assertFalse(response.config().containsKey("client.refresh-credentials-endpoint"));
+ }
+
+ @Test
+ void testRESTCatalogClientConfigMergesRemote() {
IcebergConfig config =
new IcebergConfig(
ImmutableMap.of(
@@ -172,7 +584,7 @@ public class TestCatalogWrapperForREST {
}
@Test
- void testNonRestCatalogClientConfig() {
+ void testNonRESTCatalogClientConfig() {
Catalog catalog = mock(Catalog.class);
IcebergConfig config =
new IcebergConfig(
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
index 7c2e05c80b..f7dd7cb2f0 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
@@ -19,12 +19,31 @@
package org.apache.gravitino.iceberg.service;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.List;
+import java.util.Map;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
+import org.apache.gravitino.credential.ADLSTokenCredential;
+import org.apache.gravitino.credential.AwsIrsaCredential;
+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.iceberg.service.authorization.IcebergRESTServerContext;
import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.StorageCredential;
+import org.apache.iceberg.io.SupportsStorageCredentials;
+import org.apache.iceberg.rest.credentials.Credential;
import org.apache.iceberg.rest.requests.CreateTableRequest;
import org.apache.iceberg.types.Types.IntegerType;
import org.apache.iceberg.types.Types.NestedField;
@@ -87,4 +106,216 @@ public class TestIcebergRESTUtils {
Assertions.assertEquals(field, clonedField);
}
}
+
+ @Test
+ void testTableCredentialsPath() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("s3://bucket/t/");
+ org.apache.iceberg.rest.credentials.Credential credential =
+ IcebergRESTUtils.toRESTCredential(
+ "my_catalog", table, new S3TokenCredential("k", "s", "t", 99L),
tableMetadata);
+ Assertions.assertEquals(
+ "v1/my_catalog/namespaces/ns/tables/tbl/credentials",
+ credential.config().get("client.refresh-credentials-endpoint"));
+ }
+
+ @Test
+ void testToRESTCredential() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ String refreshPath = "v1/cat/namespaces/ns/tables/tbl/credentials";
+ TableMetadata metadataWithSlash = mock(TableMetadata.class);
+ when(metadataWithSlash.location()).thenReturn("s3://bucket/t/");
+ org.apache.iceberg.rest.credentials.Credential credentialWithSlash =
+ IcebergRESTUtils.toRESTCredential(
+ "cat", table, new S3TokenCredential("k", "s", "t", 99L),
metadataWithSlash);
+ Assertions.assertEquals("s3://bucket/t/", credentialWithSlash.prefix());
+ Assertions.assertEquals(
+ "99",
credentialWithSlash.config().get("s3.session-token-expires-at-ms"));
+ Assertions.assertEquals(
+ refreshPath,
credentialWithSlash.config().get("client.refresh-credentials-endpoint"));
+
+ TableMetadata metadataWithoutSlash = mock(TableMetadata.class);
+
when(metadataWithoutSlash.location()).thenReturn("s3://bucket/path/to/table");
+ org.apache.iceberg.rest.credentials.Credential credentialWithoutSlash =
+ IcebergRESTUtils.toRESTCredential(
+ "cat", table, new S3TokenCredential("k", "s", "t", 99L),
metadataWithoutSlash);
+ Assertions.assertEquals("s3://bucket/path/to/table",
credentialWithoutSlash.prefix());
+ }
+
+ @Test
+ void testToRESTCredentialForS3Token() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("s3://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "aws", table, new S3TokenCredential("key", "secret", "token",
1234L), tableMetadata)
+ .config();
+
+ Assertions.assertEquals(
+ "v1/aws/namespaces/ns/tables/tbl/credentials",
+ config.get("client.refresh-credentials-endpoint"));
+ Assertions.assertEquals("1234",
config.get("s3.session-token-expires-at-ms"));
+ }
+
+ @Test
+ void testToRESTCredentialForAwsIrsa() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("s3://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "aws", table, new AwsIrsaCredential("key", "secret", "token",
4321L), tableMetadata)
+ .config();
+
+ Assertions.assertEquals(
+ "v1/aws/namespaces/ns/tables/tbl/credentials",
+ config.get("client.refresh-credentials-endpoint"));
+ Assertions.assertEquals("4321",
config.get("s3.session-token-expires-at-ms"));
+ }
+
+ @Test
+ void testToRESTCredentialConfigIsUnmodifiable() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("s3://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "aws", table, new S3TokenCredential("k", "s", "t", 99L),
tableMetadata)
+ .config();
+
+ Assertions.assertThrows(UnsupportedOperationException.class, () ->
config.put("k", "v"));
+ }
+
+ @Test
+ void testToRESTCredentialForGcsToken() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("gs://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "gcs", table, new GCSTokenCredential("gcs-token", 5678L),
tableMetadata)
+ .config();
+
+ Assertions.assertEquals(
+ "v1/gcs/namespaces/ns/tables/tbl/credentials",
+ config.get("gcs.oauth2.refresh-credentials-endpoint"));
+ Assertions.assertEquals("5678", config.get("gcs.oauth2.token-expires-at"));
+ }
+
+ @Test
+ void testToRESTCredentialForOssToken() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("oss://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "oss",
+ table,
+ new OSSTokenCredential("key", "secret", "oss-token", 9012L),
+ tableMetadata)
+ .config();
+
+
Assertions.assertFalse(config.containsKey("client.refresh-credentials-endpoint"));
+ Assertions.assertEquals("9012",
config.get("client.security-token-expires-at-ms"));
+ Assertions.assertEquals("key", config.get("client.access-key-id"));
+ Assertions.assertEquals("secret", config.get("client.access-key-secret"));
+ Assertions.assertEquals("oss-token", config.get("client.security-token"));
+ }
+
+ @Test
+ void testToRESTCredentialForAdlsToken() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+
when(tableMetadata.location()).thenReturn("abfss://[email protected]/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "adls",
+ table,
+ new ADLSTokenCredential("storageacct", "sas-token", 3456L),
+ tableMetadata)
+ .config();
+
+ Assertions.assertEquals(
+ "v1/adls/namespaces/ns/tables/tbl/credentials",
+ config.get("adls.refresh-credentials-endpoint"));
+ Assertions.assertEquals(
+ "3456",
config.get("adls.sas-token-expires-at-ms.storageacct.dfs.core.windows.net"));
+ }
+
+ @Test
+ void testBuildStorageCreds() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ FileIO fileIO =
+ mock(FileIO.class,
withSettings().extraInterfaces(SupportsStorageCredentials.class));
+ SupportsStorageCredentials storageCredentialsFileIO =
(SupportsStorageCredentials) fileIO;
+ StorageCredential upstreamCredential =
+ StorageCredential.create(
+ "s3://bucket/db/tbl/",
+ ImmutableMap.of(
+ "s3.access-key-id",
+ "upstream-key",
+ "s3.secret-access-key",
+ "upstream-secret",
+ "s3.session-token",
+ "upstream-token",
+ "s3.session-token-expires-at-ms",
+ "123",
+ "client.refresh-credentials-endpoint",
+ "v1/upstream/namespaces/db/tables/tbl/credentials"));
+
when(storageCredentialsFileIO.credentials()).thenReturn(List.of(upstreamCredential));
+
+ List<Credential> credentials = IcebergRESTUtils.buildStorageCreds("irc1",
table, fileIO);
+
+ Assertions.assertEquals(1, credentials.size());
+ Credential credential = credentials.get(0);
+ Assertions.assertEquals("s3://bucket/db/tbl/", credential.prefix());
+ Assertions.assertEquals("upstream-token",
credential.config().get("s3.session-token"));
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ credential.config().get("client.refresh-credentials-endpoint"));
+ Assertions.assertFalse(
+
credential.config().containsValue("v1/upstream/namespaces/db/tables/tbl/credentials"));
+ }
+
+ @Test
+ void testBuildStorageCredsPreservesSchemePrefix() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ FileIO fileIO =
+ mock(FileIO.class,
withSettings().extraInterfaces(SupportsStorageCredentials.class));
+ SupportsStorageCredentials storageCredentialsFileIO =
(SupportsStorageCredentials) fileIO;
+ StorageCredential upstreamCredential =
+ StorageCredential.create(
+ "s3",
+ ImmutableMap.of(
+ "s3.session-token", "upstream-token",
+ "s3.session-token-expires-at-ms", "123"));
+
when(storageCredentialsFileIO.credentials()).thenReturn(List.of(upstreamCredential));
+
+ Credential credential = IcebergRESTUtils.buildStorageCreds("irc1", table,
fileIO).get(0);
+
+ Assertions.assertEquals("s3", credential.prefix());
+ }
+
+ @Test
+ void testBuildStorageCredsEmpty() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ FileIO fileIO = mock(FileIO.class);
+
+ Assertions.assertTrue(IcebergRESTUtils.buildStorageCreds("irc1", table,
fileIO).isEmpty());
+ }
+
+ @Test
+ void testToRESTCredentialOmitsStaticSecretKey() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("s3://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "aws", table, new S3SecretKeyCredential("key", "secret"),
tableMetadata)
+ .config();
+
+
Assertions.assertFalse(config.containsKey("client.refresh-credentials-endpoint"));
+ }
}
