This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new ceefa9244d [#14594] feat(iceberg-rest): federate remote IRCs over a
service account connection with Gravitino-issued credentials (#14595)
ceefa9244d is described below
commit ceefa9244dff5523246905457ebc61a6c09e08a6
Author: Mark Hoerth <[email protected]>
AuthorDate: Wed Nov 4 01:30:52 2026 -0800
[#14594] feat(iceberg-rest): federate remote IRCs over a service account
connection with Gravitino-issued credentials (#14595)
### What changes were proposed in this pull request?
Five commits, readable one at a time:
1. **Service account connection** (`iceberg-common`,
`iceberg-rest-server`, `catalog-lakehouse-iceberg`). New catalog
properties: `rest-backend-auth-type` (`user-token` by default, or
`service-account`), `rest-backend-token`, `rest-backend-credential`,
`rest-backend-oauth2-server-uri` and `rest-backend-scope`.
- In service-account mode,
`IcebergCatalogUtil.applyRestBackendAuthProperties` passes exactly one
of the token or the client credential to Iceberg's OAuth2 auth manager,
instead of installing `UserPrincipalForwardingAuthManager`.
- Token and credential are hidden catalog properties.
- The Gravitino API catalog and the Iceberg REST server both build the
backend through this code, so both connect as the service account.
- `IcebergCatalogWrapper.remoteCatalogAuthorizesUsers()` replaces
`isRESTCatalog()` in
`IcebergMetadataAuthorizationMethodInterceptor.shouldSkipAuthorization`.
A service-account catalog is therefore always authorized locally,
whatever `disable-rest-authz` is set to.
- `IcebergCatalogWrapperManager` refuses to serve a service-account
catalog when authorization is disabled.
2. **Gravitino-issued storage credentials** (`iceberg-rest-server`). New
property `rest-backend-credential-vending` (`remote` by default, or
`gravitino`). A service-account catalog always uses `gravitino`, because
the remote vends credentials for the service account rather than for the
caller. In `gravitino` mode, `FederatedCatalogWrapper`:
- never sends `X-Iceberg-Access-Delegation` to the remote;
- drops remote `storage-credentials` and credential keys from load and
scan responses;
- issues credentials through
`CatalogWrapperForREST.injectCredentialConfig` and
`injectScanCredentials` for load, create, register, credential refresh
and scan planning.
Privilege follows
`IcebergTableOperationExecutor.getCredentialPrivilege`: READ unless the
caller holds MODIFY_TABLE. Create and register get WRITE, as for local
catalogs.
3. **Grants follow the table, not the name** (`api`, `core`,
`catalog-lakehouse-iceberg`, `iceberg-rest-server`). New reserved table
property `Table.PROPERTY_TABLE_UUID` (`table-uuid`), which the Iceberg
catalog reports from table metadata. It is honored only in catalogs that
declare it reserved.
- `TableOperationDispatcher` records it on import and create, keeps it
on alter and rename, and backfills it on registrations made before this
change.
- When a load finds a different identifier under the same name, it
deletes the stale registration (with its grants, tags, owner and
statistics) under the schema write lock and imports the new table.
- On the Iceberg REST server, `IcebergTableHookDispatcher` makes the
same check for load, credentials and scan planning. On a mismatch it
replaces the registration, authorizes the caller again against the new
registration, and reloads, so neither access nor credential scope
follows the old table.
- A rename keeps the UUID.
4. **Purges go to the remote** (`iceberg-common`,
`iceberg-rest-server`). `IcebergCatalogWrapper.purgeTable` forwards a
purge on a REST catalog backend to the remote with `purgeRequested`, on
both the Iceberg REST and Gravitino API paths. Before, it dropped the
remote entry and then deleted the files from Gravitino, and an
`X-Gravitino-Async-Purge` request queued a local cleanup job to do the
same. A REST catalog backend never takes the async path now.
5. **The authorizer learns of a replacement at once** (`core`). When the
stale registration of a replaced table is deleted,
`TableOperationDispatcher` notifies the authorizer, as a drop does.
Without it, the authorizer's name-to-id cache still resolved the name to
the old table, so the old grants applied to the request that discovered
the replacement.
### Why are the changes needed?
To federate remote Iceberg REST catalogs that cannot authorize
Gravitino's users, with Gravitino as the single place access is granted,
without handing clients the remote catalog's broad storage keys, without
grants leaking to a table recreated under an old name, and without
Gravitino deleting files the remote catalog owns.
Fix: #14594
### Does this PR introduce _any_ user-facing change?
- New Iceberg catalog / Iceberg REST service properties:
`rest-backend-auth-type`, `rest-backend-token`,
`rest-backend-credential`, `rest-backend-oauth2-server-uri`,
`rest-backend-scope`, `rest-backend-credential-vending`. Defaults keep
today's behavior.
- New reserved Iceberg table property `table-uuid`, reported on load and
rejected on create and alter.
- Docs: `iceberg-rest-service.md` (REST backend section) and
`lakehouse-iceberg-catalog.md`.
### How was this patch tested?
Unit tests:
- `TestIcebergCatalogUtil`: auth modes, the exactly-one-credential
check, an invalid mode.
- `TestIcebergMetadataAuthorizationMethodInterceptor`: a service-account
catalog is not skipped.
- `TestCatalogWrapperForREST`: gravitino mode, and service-account mode
with default vending, send no delegation header, drop remote credentials
and return provider-issued credentials.
- `TestIcebergTableHookDispatcher`: a replaced registration is
reconciled; a registration without a UUID is left alone.
- `TestIcebergCatalogWrapper`: a purge on a REST catalog is forwarded
with `purgeRequested` and loads nothing locally.
- `TestIcebergAsyncPurge`: an async purge request on a REST catalog is
forwarded inline and queues no job.
- `TestTableOperationDispatcher`: the registration survives a reload, a
replaced table gets a new registration, a missing UUID is backfilled,
plus `isReplacedExternally` cases.
End to end, against a live Cloudflare R2 Data Catalog federated with a
service account connection and an R2 temporary-credential provider: a
user granted SELECT_TABLE on one table received a credential that could
list that table's location and nothing else, and could not write; a user
with write privileges could write only inside the table's location;
creating a table without CREATE_TABLE and loading as a non-member both
returned 403; after the table was dropped and recreated directly in the
remote catalog, the user granted SELECT on the old table got 403 on the
first load of the new one.
### Known limitations, left for follow-up
- **Multi-node edge case.** The stale-registration check reads the table
entity through the entity cache. In a multi-node deployment, a node with
a stale cache entry could, in a narrow window, drop a registration
another node has just replaced. Deleting conditioned on the entity
version would close it.
- **Token fetch per call.** With an OAuth2 client credential (not a
static token), each remote call through `callRemoteCatalog` fetches a
fresh token. A per-wrapper auth session cache would remove that.
- **Copied properties now rejected.** `table-uuid` is reserved, so flows
that copy a loaded table's properties into a create, such as `CREATE
TABLE LIKE`, must drop it. `sort-order` already behaves this way.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01Eu6fb6yGcAF4XcGDTmKE56
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
Co-authored-by: roryqi <[email protected]>
---
.../lakehouse/iceberg/IcebergConstants.java | 56 ++++++
.../lakehouse/iceberg/IcebergPropertiesUtils.java | 13 ++
.../iceberg/IcebergCatalogPropertiesMetadata.java | 42 ++++
docs/iceberg-rest-service.md | 36 ++++
docs/lakehouse-iceberg-catalog.md | 6 +
.../gravitino/iceberg/common/IcebergConfig.java | 57 ++++++
.../iceberg/common/ops/IcebergCatalogWrapper.java | 13 +-
.../iceberg/common/utils/IcebergCatalogUtil.java | 56 +++++-
.../iceberg/common/TestIcebergConfig.java | 36 ++++
.../common/utils/TestIcebergCatalogUtil.java | 86 +++++++++
.../iceberg/service/CatalogWrapperForREST.java | 34 +++-
.../iceberg/service/FederatedCatalogWrapper.java | 185 ++++++++++++++++--
.../service/IcebergCatalogWrapperManager.java | 9 +-
.../dispatcher/IcebergTableOperationExecutor.java | 6 +-
...bergMetadataAuthorizationMethodInterceptor.java | 9 +-
.../iceberg/service/TestCatalogWrapperForREST.java | 214 +++++++++++++++++++++
.../TestIcebergCatalogWrapperManagerForREST.java | 62 ++++++
.../service/dispatcher/TestIcebergAsyncPurge.java | 17 ++
...bergMetadataAuthorizationMethodInterceptor.java | 28 ++-
19 files changed, 935 insertions(+), 30 deletions(-)
diff --git
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
index 8c0e1bb97f..7b528d5478 100644
---
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
+++
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
@@ -137,6 +137,62 @@ public class IcebergConstants {
public static final String ICEBERG_REST_DEFAULT_CATALOG = "default_catalog";
public static final String ICEBERG_REST_DEFAULT_DYNAMIC_CATALOG_NAME =
"default-catalog-name";
public static final String ICEBERG_REST_DISABLE_REST_AUTHZ =
"disable-rest-authz";
+
+ /**
+ * Gravitino config key selecting how a REST catalog backend authenticates
to the remote catalog:
+ * {@link #REST_BACKEND_AUTH_TYPE_USER_TOKEN} or {@link
#REST_BACKEND_AUTH_TYPE_SERVICE_ACCOUNT}.
+ */
+ public static final String REST_BACKEND_AUTH_TYPE = "rest-backend-auth-type";
+
+ /** Forward the calling user's access token to the remote catalog, which
authorizes the user. */
+ public static final String REST_BACKEND_AUTH_TYPE_USER_TOKEN = "user-token";
+
+ /**
+ * Connect to the remote catalog as a service account with a configured
credential. Gravitino
+ * authorizes every request locally before proxying it.
+ */
+ public static final String REST_BACKEND_AUTH_TYPE_SERVICE_ACCOUNT =
"service-account";
+
+ /** Gravitino config key for the service account's static bearer token. */
+ public static final String REST_BACKEND_TOKEN = "rest-backend-token";
+
+ /** Gravitino config key for the service account's OAuth2 client credential.
*/
+ public static final String REST_BACKEND_CREDENTIAL =
"rest-backend-credential";
+
+ /** Gravitino config key for the OAuth2 token endpoint used with the client
credential. */
+ public static final String REST_BACKEND_OAUTH2_SERVER_URI =
"rest-backend-oauth2-server-uri";
+
+ /** Gravitino config key for the OAuth2 scope requested with the client
credential. */
+ public static final String REST_BACKEND_SCOPE = "rest-backend-scope";
+
+ /**
+ * Gravitino config key selecting who issues storage credentials for tables
of a federated REST
+ * catalog: {@link #REST_BACKEND_STORAGE_CREDENTIAL_SOURCE_REMOTE} or {@link
+ * #REST_BACKEND_STORAGE_CREDENTIAL_SOURCE_LOCAL}.
+ */
+ public static final String REST_BACKEND_STORAGE_CREDENTIAL_SOURCE =
+ "rest-backend-storage-credential-source";
+
+ /** Pass through the storage credentials the remote catalog vends. */
+ public static final String REST_BACKEND_STORAGE_CREDENTIAL_SOURCE_REMOTE =
"remote";
+
+ /**
+ * Do not request credentials from the remote catalog; issue them through
this catalog's own
+ * credential providers, scoped to the table and to the caller's privilege.
+ */
+ public static final String REST_BACKEND_STORAGE_CREDENTIAL_SOURCE_LOCAL =
"local";
+
+ /** Iceberg REST client property for a static bearer token. */
+ public static final String ICEBERG_REST_TOKEN = "token";
+
+ /** Iceberg REST client property for an OAuth2 client credential. */
+ public static final String ICEBERG_REST_CREDENTIAL = "credential";
+
+ /** Iceberg REST client property for the OAuth2 token endpoint. */
+ public static final String ICEBERG_REST_OAUTH2_SERVER_URI =
"oauth2-server-uri";
+
+ /** Iceberg REST client property for the OAuth2 scope. */
+ public static final String ICEBERG_REST_SCOPE = "scope";
/** Gravitino config key for REST catalog backend HTTP connection timeout. */
public static final String REST_CATALOG_BACKEND_CLIENT_CONNECTION_TIMEOUT_MS
=
"rest-client-connection-timeout-ms";
diff --git
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergPropertiesUtils.java
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergPropertiesUtils.java
index 4f7ae7bd7e..af0b6d4a51 100644
---
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergPropertiesUtils.java
+++
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergPropertiesUtils.java
@@ -111,6 +111,19 @@ public class IcebergPropertiesUtils {
map.put(
IcebergConstants.REST_CATALOG_BACKEND_CLIENT_SOCKET_TIMEOUT_MS,
IcebergConstants.ICEBERG_REST_CLIENT_SOCKET_TIMEOUT_MS);
+ // The service account settings keep their Gravitino names here and are
translated to Iceberg
+ // REST client properties only when the REST backend is built, so no other
component (such as
+ // an engine connector converting catalog properties) receives the remote
catalog's credential.
+ map.put(IcebergConstants.REST_BACKEND_AUTH_TYPE,
IcebergConstants.REST_BACKEND_AUTH_TYPE);
+ map.put(IcebergConstants.REST_BACKEND_TOKEN,
IcebergConstants.REST_BACKEND_TOKEN);
+ map.put(IcebergConstants.REST_BACKEND_CREDENTIAL,
IcebergConstants.REST_BACKEND_CREDENTIAL);
+ map.put(
+ IcebergConstants.REST_BACKEND_OAUTH2_SERVER_URI,
+ IcebergConstants.REST_BACKEND_OAUTH2_SERVER_URI);
+ map.put(IcebergConstants.REST_BACKEND_SCOPE,
IcebergConstants.REST_BACKEND_SCOPE);
+ map.put(
+ IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE,
+ IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE);
GRAVITINO_CONFIG_TO_ICEBERG = Collections.unmodifiableMap(map);
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogPropertiesMetadata.java
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogPropertiesMetadata.java
index 407af67a33..fd27211797 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogPropertiesMetadata.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogPropertiesMetadata.java
@@ -132,6 +132,48 @@ public class IcebergCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetad
"HTTP socket timeout in milliseconds for the REST catalog
backend",
false /* immutable */,
60000 /* defaultValue */,
+ false /* hidden */),
+ stringOptionalPropertyEntry(
+ IcebergConstants.REST_BACKEND_AUTH_TYPE,
+ "How the REST catalog backend authenticates to the remote
catalog: user-token"
+ + " forwards the calling user's token, service-account
connects with a"
+ + " configured credential and authorizes every request in
Gravitino",
+ false /* immutable */,
+ IcebergConstants.REST_BACKEND_AUTH_TYPE_USER_TOKEN /*
defaultValue */,
+ false /* hidden */),
+ stringOptionalPropertyEntry(
+ IcebergConstants.REST_BACKEND_TOKEN,
+ "Static bearer token the service account presents to the
remote REST catalog",
+ false /* immutable */,
+ null /* defaultValue */,
+ true /* hidden */),
+ stringOptionalPropertyEntry(
+ IcebergConstants.REST_BACKEND_CREDENTIAL,
+ "OAuth2 client credential (client_id:client_secret) the
service account uses"
+ + " with the remote REST catalog",
+ false /* immutable */,
+ null /* defaultValue */,
+ true /* hidden */),
+ stringOptionalPropertyEntry(
+ IcebergConstants.REST_BACKEND_OAUTH2_SERVER_URI,
+ "OAuth2 token endpoint for the service account's client
credential",
+ false /* immutable */,
+ null /* defaultValue */,
+ false /* hidden */),
+ stringOptionalPropertyEntry(
+ IcebergConstants.REST_BACKEND_SCOPE,
+ "OAuth2 scope requested with the service account's client
credential",
+ false /* immutable */,
+ null /* defaultValue */,
+ false /* hidden */),
+ stringOptionalPropertyEntry(
+ IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE,
+ "Who issues storage credentials for a federated REST catalog:
remote passes"
+ + " through the remote catalog's credentials, local issues
them through"
+ + " this catalog's credential providers. Service-account
connections always"
+ + " use local credentials",
+ false /* immutable */,
+ IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE_REMOTE
/* defaultValue */,
false /* hidden */));
HashMap<String, PropertyEntry<?>> result = Maps.newHashMap();
result.putAll(Maps.uniqueIndex(propertyEntries, PropertyEntry::getName));
diff --git a/docs/iceberg-rest-service.md b/docs/iceberg-rest-service.md
index 60584f2e9f..859dcd544b 100644
--- a/docs/iceberg-rest-service.md
+++ b/docs/iceberg-rest-service.md
@@ -260,6 +260,42 @@ IRC1 must also configure S3 configurations if the client
side requests credentia
If IRC2 does not enforce authorization, keeping
`gravitino.iceberg-rest.disable-rest-authz=true` can leave operations
unprotected. Set it to `false` to enforce authorization in IRC1.
:::
+##### Service account connection
+
+By default IRC1 forwards each user's access token to IRC2, so IRC2 must trust
the same identity provider and authorize every user itself. To federate a
catalog that cannot do that, such as a managed catalog that accepts only its
own API token, connect IRC1 to IRC2 as a service account. IRC1 then presents
one configured credential on every request to IRC2 and authorizes each user
against Gravitino's own grants before anything is proxied. A catalog connected
as a service account always enf [...]
+
+| Configuration item | Description
| Default value | Required
|
+|-------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------|---------------|---------------------------------------------|
+| `gravitino.iceberg-rest.rest-backend-auth-type` | How IRC1
authenticates to IRC2. `user-token` forwards the calling user's access token.
`service-account` connects with the credential below. | `user-token` |
No |
+| `gravitino.iceberg-rest.rest-backend-token` | Static bearer token
the service account presents to IRC2.
| (none) | One of token
or credential with `service-account` |
+| `gravitino.iceberg-rest.rest-backend-credential` | OAuth2 client
credential (`client_id:client_secret`) the service account exchanges for a
token. | (none) | One
of token or credential with `service-account` |
+| `gravitino.iceberg-rest.rest-backend-oauth2-server-uri` | OAuth2 token
endpoint for the client credential. Defaults to IRC2's own token endpoint.
| (none) | No
|
+| `gravitino.iceberg-rest.rest-backend-scope` | OAuth2 scope
requested with the client credential.
| (none) | No
|
+
+##### Gravitino-issued storage credentials
+
+By default IRC1 passes through the storage credentials IRC2 vends. Some
catalogs vend broad, long-lived credentials, for example the catalog token's
own storage keys scoped to the whole bucket. Set
`rest-backend-storage-credential-source` to `local` and IRC1 stops asking IRC2
for credentials. It issues them through its own `credential-providers` instead,
scoped to the table location, read-only unless the caller holds write
privileges on the table, exactly as for a local catalog. Credenti [...]
+
+| Configuration item | Description
| Default value | Required |
+|---------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------|
+| `gravitino.iceberg-rest.rest-backend-storage-credential-source` | Who issues
storage credentials for tables of the federated catalog. `remote` passes
through IRC2's credentials. `local` uses IRC1's `credential-providers`. A
`service-account` catalog always uses `local`, regardless of this setting. |
`remote` | No |
+
+IRC1 configuration example federating a catalog that accepts only a static API
token, with Gravitino issuing the storage credentials:
+
+```text
+gravitino.iceberg-rest.catalog-backend = rest
+gravitino.iceberg-rest.uri = https://catalog.example.com/v1
+gravitino.iceberg-rest.warehouse = example_warehouse
+gravitino.iceberg-rest.rest-backend-auth-type = service-account
+gravitino.iceberg-rest.rest-backend-token = xx
+gravitino.iceberg-rest.rest-backend-storage-credential-source = local
+gravitino.iceberg-rest.credential-providers = s3-token
+gravitino.iceberg-rest.s3-access-key-id = xx
+gravitino.iceberg-rest.s3-secret-access-key = xx
+gravitino.iceberg-rest.s3-region = xx
+gravitino.iceberg-rest.s3-role-arn = xx
+```
+
`data-access` is returned in `/v1/config` defaults for REST clients:
- `vended-credentials`: clients should request credential vending
(`X-Iceberg-Access-Delegation: vended-credentials`).
diff --git a/docs/lakehouse-iceberg-catalog.md
b/docs/lakehouse-iceberg-catalog.md
index 4a7c410b95..dfd5137633 100644
--- a/docs/lakehouse-iceberg-catalog.md
+++ b/docs/lakehouse-iceberg-catalog.md
@@ -87,6 +87,12 @@ The following properties tune REST backend behavior:
| `data-access` | Data access mode for REST catalog
backend. Supported values are `vended-credentials` and `remote-signing`. |
(none) | No |
| `rest-client-connection-timeout-ms` | The HTTP connection timeout in
milliseconds for requests to the REST catalog backend. |
10000 | No |
| `rest-client-socket-timeout-ms` | The HTTP socket timeout in
milliseconds for requests to the REST catalog backend.
| 60000 | No |
+| `rest-backend-auth-type` | How the REST backend authenticates to
the remote catalog: `user-token` forwards the calling user's token,
`service-account` connects with `rest-backend-token` or
`rest-backend-credential` and authorizes every request in Gravitino. |
user-token | No |
+| `rest-backend-token` | Static bearer token for a
`service-account` connection. Hidden from catalog responses.
| (none) | No |
+| `rest-backend-credential` | OAuth2 client credential
(`client_id:client_secret`) for a `service-account` connection. Hidden from
catalog responses. | (none) | No |
+| `rest-backend-oauth2-server-uri` | OAuth2 token endpoint for
`rest-backend-credential`.
| (none) | No |
+| `rest-backend-scope` | OAuth2 scope requested with
`rest-backend-credential`.
| (none) | No |
+| `rest-backend-storage-credential-source` | Who issues storage credentials
through the Iceberg REST service: `remote` passes through the remote catalog's
credentials, `local` issues them through this catalog's `credential-providers`.
A `service-account` connection always uses `local`, regardless of this setting.
| remote | No |
- `vended-credentials`: request credential vending from the Iceberg REST
server.
- `remote-signing`: Gravitino doesn't support this mode yet.
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/IcebergConfig.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/IcebergConfig.java
index 17d0679c19..c8046ce331 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/IcebergConfig.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/IcebergConfig.java
@@ -240,6 +240,53 @@ public class IcebergConfig extends Config implements
OverwriteDefaultConfig {
.booleanConf()
.createWithDefault(true);
+ /**
+ * How a REST catalog backend authenticates to the remote catalog. {@code
user-token} forwards the
+ * calling user's access token; {@code service-account} connects with the
configured service
+ * account credential and makes Gravitino authorize every request locally.
+ */
+ public static final ConfigEntry<String> REST_BACKEND_AUTH_TYPE =
+ new ConfigBuilder(IcebergConstants.REST_BACKEND_AUTH_TYPE)
+ .doc(
+ "How the REST catalog backend authenticates to the remote
catalog. `user-token`"
+ + " forwards the calling user's access token, and the remote
catalog authorizes"
+ + " the user. `service-account` connects with the credential
in"
+ + " `rest-backend-token` or `rest-backend-credential`, and
Gravitino authorizes"
+ + " every request before proxying it.")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .stringConf()
+ .checkValue(
+ value ->
+
IcebergConstants.REST_BACKEND_AUTH_TYPE_USER_TOKEN.equals(value)
+ ||
IcebergConstants.REST_BACKEND_AUTH_TYPE_SERVICE_ACCOUNT.equals(value),
+ "The value must be `user-token` or `service-account`")
+
.createWithDefault(IcebergConstants.REST_BACKEND_AUTH_TYPE_USER_TOKEN);
+
+ /**
+ * Who issues storage credentials for tables of a federated REST catalog.
{@code remote} passes
+ * through what the remote catalog vends; {@code local} issues credentials
through this catalog's
+ * credential providers, scoped to the table and to the caller's privilege.
Service-account
+ * connections always use local credentials.
+ */
+ public static final ConfigEntry<String>
REST_BACKEND_STORAGE_CREDENTIAL_SOURCE =
+ new
ConfigBuilder(IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE)
+ .doc(
+ "Who issues storage credentials for tables of a federated REST
catalog. `remote`"
+ + " passes through the credentials the remote catalog vends.
`local` does not"
+ + " request credentials from the remote catalog and issues
them through this"
+ + " catalog's `credential-providers`, scoped to the table
location and to the"
+ + " caller's privilege on the table. Service-account
connections always use local"
+ + " credentials.")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .stringConf()
+ .checkValue(
+ value ->
+
IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE_REMOTE.equals(value)
+ ||
IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE_LOCAL.equals(
+ value),
+ "The value must be `local` or `remote`")
+
.createWithDefault(IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE_REMOTE);
+
/** HTTP connection timeout in milliseconds for the REST catalog backend. */
public static final ConfigEntry<Integer>
REST_CATALOG_BACKEND_CLIENT_CONNECTION_TIMEOUT_MS =
new
ConfigBuilder(IcebergConstants.REST_CATALOG_BACKEND_CLIENT_CONNECTION_TIMEOUT_MS)
@@ -432,6 +479,16 @@ public class IcebergConfig extends Config implements
OverwriteDefaultConfig {
return transformedConfig;
}
+ /**
+ * Returns whether a REST catalog backend connects to the remote catalog as
a service account.
+ *
+ * @return {@code true} if {@code rest-backend-auth-type} is {@code
service-account}
+ */
+ public boolean isRestBackendServiceAccount() {
+ return IcebergConstants.REST_BACKEND_AUTH_TYPE_SERVICE_ACCOUNT.equals(
+ get(REST_BACKEND_AUTH_TYPE));
+ }
+
@Override
public Map<String, String> getOverwriteDefaultConfig() {
return ImmutableMap.of(
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
index c36cc09d3e..1b5312ad7e 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
@@ -142,6 +142,15 @@ public class IcebergCatalogWrapper implements
AutoCloseable {
return icebergConfig;
}
+ /**
+ * Invalidates cached metadata before a subclass changes or removes a table.
+ *
+ * @param tableIdentifier the table whose cached metadata must be invalidated
+ */
+ protected void invalidateTableMetadata(TableIdentifier tableIdentifier) {
+ getMetadataCache().invalidate(tableIdentifier);
+ }
+
private SupportsNamespaces getNamespaceCatalog() {
SupportsNamespaces namespaceCatalog = asNamespaceCatalog;
if (namespaceCatalog != null) {
@@ -276,12 +285,12 @@ public class IcebergCatalogWrapper implements
AutoCloseable {
}
public void dropTable(TableIdentifier tableIdentifier) {
- getMetadataCache().invalidate(tableIdentifier);
+ invalidateTableMetadata(tableIdentifier);
CatalogHandlers.dropTable(getCatalog(), tableIdentifier);
}
public void purgeTable(TableIdentifier tableIdentifier) {
- getMetadataCache().invalidate(tableIdentifier);
+ invalidateTableMetadata(tableIdentifier);
Table table = getCatalog().loadTable(tableIdentifier);
if (!(table instanceof HasTableOperations)) {
CatalogHandlers.purgeTable(getCatalog(), tableIdentifier);
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
index 59b8ce8104..d70e6d051f 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
@@ -272,8 +272,7 @@ public class IcebergCatalogUtil {
Map<String, String> properties =
Maps.newHashMap(icebergConfig.getIcebergCatalogProperties());
applyDefaultResolvingFileIO(properties);
- // REST catalog must use forward access token from the user request
- properties.put(AuthProperties.AUTH_TYPE,
UserPrincipalForwardingAuthManager.class.getName());
+ applyRestBackendAuthProperties(icebergConfig, properties);
applyRestCatalogHttpTimeoutProperties(icebergConfig, properties);
properties.forEach(hdfsConfiguration::set);
@@ -374,6 +373,59 @@ public class IcebergCatalogUtil {
}
}
+ /**
+ * Configures how the REST catalog backend authenticates to the remote
catalog.
+ *
+ * <p>With {@code rest-backend-auth-type=user-token} (the default), the
calling user's access
+ * token is forwarded on every request. With {@code service-account}, the
configured static token
+ * ({@code rest-backend-token}) or OAuth2 client credential ({@code
rest-backend-credential}, with
+ * optional {@code rest-backend-oauth2-server-uri} and {@code
rest-backend-scope}) is passed to
+ * Iceberg's OAuth2 auth manager, so every request reaches the remote
catalog as the service
+ * account. The Gravitino-named settings are removed from the client
properties in both modes.
+ *
+ * @param icebergConfig the catalog configuration
+ * @param properties the Iceberg REST client properties to update
+ */
+ @VisibleForTesting
+ static void applyRestBackendAuthProperties(
+ IcebergConfig icebergConfig, Map<String, String> properties) {
+ String token = properties.remove(IcebergConstants.REST_BACKEND_TOKEN);
+ String credential =
properties.remove(IcebergConstants.REST_BACKEND_CREDENTIAL);
+ String oauth2ServerUri =
properties.remove(IcebergConstants.REST_BACKEND_OAUTH2_SERVER_URI);
+ String scope = properties.remove(IcebergConstants.REST_BACKEND_SCOPE);
+ properties.remove(IcebergConstants.REST_BACKEND_AUTH_TYPE);
+
+ if (!icebergConfig.isRestBackendServiceAccount()) {
+ properties.put(AuthProperties.AUTH_TYPE,
UserPrincipalForwardingAuthManager.class.getName());
+ return;
+ }
+
+ boolean hasToken = StringUtils.isNotBlank(token);
+ boolean hasCredential = StringUtils.isNotBlank(credential);
+ if (hasToken == hasCredential) {
+ throw new IllegalArgumentException(
+ String.format(
+ "%s=%s requires exactly one of %s or %s",
+ IcebergConstants.REST_BACKEND_AUTH_TYPE,
+ IcebergConstants.REST_BACKEND_AUTH_TYPE_SERVICE_ACCOUNT,
+ IcebergConstants.REST_BACKEND_TOKEN,
+ IcebergConstants.REST_BACKEND_CREDENTIAL));
+ }
+
+ properties.put(AuthProperties.AUTH_TYPE, AuthProperties.AUTH_TYPE_OAUTH2);
+ if (hasToken) {
+ properties.put(IcebergConstants.ICEBERG_REST_TOKEN, token);
+ } else {
+ properties.put(IcebergConstants.ICEBERG_REST_CREDENTIAL, credential);
+ if (StringUtils.isNotBlank(oauth2ServerUri)) {
+ properties.put(IcebergConstants.ICEBERG_REST_OAUTH2_SERVER_URI,
oauth2ServerUri);
+ }
+ if (StringUtils.isNotBlank(scope)) {
+ properties.put(IcebergConstants.ICEBERG_REST_SCOPE, scope);
+ }
+ }
+ }
+
@VisibleForTesting
static void applyRestCatalogHttpTimeoutProperties(
IcebergConfig icebergConfig, Map<String, String> properties) {
diff --git
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestIcebergConfig.java
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestIcebergConfig.java
index aa6ed3fd67..1a8827eed6 100644
---
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestIcebergConfig.java
+++
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestIcebergConfig.java
@@ -26,8 +26,44 @@ import
org.apache.gravitino.iceberg.common.cache.LocalTableMetadataCache;
import org.apache.gravitino.server.web.JettyServerConfig;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
public class TestIcebergConfig {
+ @Test
+ void testStorageCredentialSourceDefaults() {
+ IcebergConfig config = new IcebergConfig(ImmutableMap.of());
+ Assertions.assertEquals(
+ "remote",
config.get(IcebergConfig.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE));
+
+ IcebergConfig serviceAccountConfig =
+ new IcebergConfig(
+ ImmutableMap.of(IcebergConstants.REST_BACKEND_AUTH_TYPE,
"service-account"));
+ Assertions.assertEquals(
+ "remote",
serviceAccountConfig.get(IcebergConfig.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"local", "remote"})
+ void testStorageCredentialSource(String source) {
+ IcebergConfig config =
+ new IcebergConfig(
+
ImmutableMap.of(IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE,
source));
+ Assertions.assertEquals(
+ source,
config.get(IcebergConfig.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"gravitino", "unknown"})
+ void testStorageCredentialSourceRejectsInvalidValues(String source) {
+ IcebergConfig config =
+ new IcebergConfig(
+
ImmutableMap.of(IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE,
source));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
config.get(IcebergConfig.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE));
+ }
+
@Test
public void testLoadIcebergConfig() {
Map<String, String> properties =
diff --git
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
index d8ce5cc18a..a7f562de46 100644
---
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
+++
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
@@ -30,6 +30,7 @@ import
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.common.ClosableJdbcCatalog;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig;
+import
org.apache.gravitino.iceberg.common.rest.auth.UserPrincipalForwardingAuthManager;
import org.apache.gravitino.storage.GCSProperties;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.Schema;
@@ -41,6 +42,7 @@ import org.apache.iceberg.hive.HiveCatalog;
import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.jdbc.JdbcCatalogWithMetadataLocationSupport;
import org.apache.iceberg.jdbc.UncheckedSQLException;
+import org.apache.iceberg.rest.auth.AuthProperties;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
@@ -516,4 +518,88 @@ public class TestIcebergCatalogUtil {
IcebergCatalogBackend.JDBC, new IcebergConfig(properties));
Assertions.assertInstanceOf(ClosableJdbcCatalog.class, catalog2);
}
+
+ @Test
+ void testRestBackendForwardsUserTokenByDefault() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(IcebergConstants.REST_BACKEND_TOKEN, "ignored");
+
+ IcebergCatalogUtil.applyRestBackendAuthProperties(
+ new IcebergConfig(new HashMap<>()), properties);
+
+ Assertions.assertEquals(
+ UserPrincipalForwardingAuthManager.class.getName(),
+ properties.get(AuthProperties.AUTH_TYPE));
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.REST_BACKEND_TOKEN));
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_REST_TOKEN));
+ }
+
+ @Test
+ void testRestBackendServiceAccountWithToken() {
+ Map<String, String> config = serviceAccountConfig();
+ Map<String, String> properties = new HashMap<>();
+ properties.put(IcebergConstants.REST_BACKEND_TOKEN, "cf-token");
+ properties.put(IcebergConstants.REST_BACKEND_AUTH_TYPE, "service-account");
+
+ IcebergCatalogUtil.applyRestBackendAuthProperties(new
IcebergConfig(config), properties);
+
+ Assertions.assertEquals(
+ AuthProperties.AUTH_TYPE_OAUTH2,
properties.get(AuthProperties.AUTH_TYPE));
+ Assertions.assertEquals("cf-token",
properties.get(IcebergConstants.ICEBERG_REST_TOKEN));
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_REST_CREDENTIAL));
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.REST_BACKEND_TOKEN));
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.REST_BACKEND_AUTH_TYPE));
+ }
+
+ @Test
+ void testRestBackendServiceAccountWithClientCredential() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(IcebergConstants.REST_BACKEND_CREDENTIAL, "client:secret");
+ properties.put(IcebergConstants.REST_BACKEND_OAUTH2_SERVER_URI,
"https://idp/token";);
+ properties.put(IcebergConstants.REST_BACKEND_SCOPE, "catalog");
+
+ IcebergCatalogUtil.applyRestBackendAuthProperties(
+ new IcebergConfig(serviceAccountConfig()), properties);
+
+ Assertions.assertEquals(
+ AuthProperties.AUTH_TYPE_OAUTH2,
properties.get(AuthProperties.AUTH_TYPE));
+ Assertions.assertEquals(
+ "client:secret",
properties.get(IcebergConstants.ICEBERG_REST_CREDENTIAL));
+ Assertions.assertEquals(
+ "https://idp/token";,
properties.get(IcebergConstants.ICEBERG_REST_OAUTH2_SERVER_URI));
+ Assertions.assertEquals("catalog",
properties.get(IcebergConstants.ICEBERG_REST_SCOPE));
+
Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_REST_TOKEN));
+ }
+
+ @Test
+ void testRestBackendServiceAccountRequiresExactlyOneCredential() {
+ IcebergConfig config = new IcebergConfig(serviceAccountConfig());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> IcebergCatalogUtil.applyRestBackendAuthProperties(config, new
HashMap<>()));
+
+ Map<String, String> both = new HashMap<>();
+ both.put(IcebergConstants.REST_BACKEND_TOKEN, "t");
+ both.put(IcebergConstants.REST_BACKEND_CREDENTIAL, "c:s");
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> IcebergCatalogUtil.applyRestBackendAuthProperties(config, both));
+ }
+
+ @Test
+ void testRestBackendAuthTypeRejectsUnknownValue() {
+ Map<String, String> config = new HashMap<>();
+ config.put(IcebergConstants.REST_BACKEND_AUTH_TYPE, "basic");
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> new IcebergConfig(config).isRestBackendServiceAccount());
+ }
+
+ private static Map<String, String> serviceAccountConfig() {
+ Map<String, String> config = new HashMap<>();
+ config.put(
+ IcebergConstants.REST_BACKEND_AUTH_TYPE,
+ IcebergConstants.REST_BACKEND_AUTH_TYPE_SERVICE_ACCOUNT);
+ return config;
+ }
}
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 921a6bc127..1fde254d35 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
@@ -45,6 +45,7 @@ import
org.apache.gravitino.credential.CredentialPropertyUtils;
import org.apache.gravitino.credential.PathBasedCredentialContext;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
+import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
import org.apache.gravitino.iceberg.service.cache.ScanPlanCache;
import org.apache.gravitino.iceberg.service.cache.ScanPlanCacheKey;
import org.apache.gravitino.storage.GCSProperties;
@@ -126,6 +127,31 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
this.scanPlanCache = loadScanPlanCache(config);
}
+ /**
+ * Validates server settings required by this catalog before it serves
requests.
+ *
+ * @param serverContext the REST server context
+ */
+ public void validateServerContext(IcebergRESTServerContext serverContext) {}
+
+ /**
+ * Reports whether this catalog forwards the caller's access token to a
remote catalog.
+ *
+ * @return {@code false} for a local catalog backend
+ */
+ public boolean forwardsUserToken() {
+ return false;
+ }
+
+ /**
+ * Reports whether this server can enqueue file cleanup after dropping a
table.
+ *
+ * @return {@code true} for a local catalog backend
+ */
+ public boolean supportsAsyncPurge() {
+ return true;
+ }
+
public LoadTableResponse createTable(
Namespace namespace, CreateTableRequest request, boolean
requestCredential) {
LoadTableResponse loadTableResponse = super.createTable(namespace,
request);
@@ -608,8 +634,14 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
* Inject vended credentials into a scan response using the already-loaded
table, avoiding a
* redundant {@code loadTable} call. Follows the same eligibility logic as
{@link
* #shouldGenerateCredential} and the same credential generation as {@link
#getCredential}.
+ *
+ * @param tableIdentifier the table identifier
+ * @param table the loaded table
+ * @param response the scan response to add credentials to
+ * @param privilege the caller's privilege on the table
+ * @return the response with credentials, or unchanged if none could be
generated
*/
- private PlanTableScanResponse injectScanCredentials(
+ protected PlanTableScanResponse injectScanCredentials(
TableIdentifier tableIdentifier,
Table table,
PlanTableScanResponse response,
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java
index 17c3449c14..f28e3a65ea 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.iceberg.service;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import java.util.Collections;
@@ -29,8 +30,10 @@ import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.credential.CredentialPrivilege;
import org.apache.gravitino.iceberg.common.IcebergConfig;
+import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
import org.apache.iceberg.BaseTransaction;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.MetadataUpdate;
@@ -72,8 +75,13 @@ import
org.apache.iceberg.rest.responses.PlanTableScanResponse;
* <p>Federation-specific behavior is expressed through polymorphic overrides
instead of {@code
* instanceof RESTCatalog} checks scattered across the base class. Table load,
create and register
* use authenticated REST calls so {@code X-Iceberg-Access-Delegation} can be
forwarded when the
- * client requested credential vending. Update still uses the Iceberg Catalog
API. This wrapper
- * never injects Gravitino-generated credentials.
+ * client requested credential vending. Update still uses the Iceberg Catalog
API.
+ *
+ * <p>By default the wrapper passes through the storage credentials the remote
catalog vends and
+ * never injects Gravitino-generated credentials. With {@code
+ * rest-backend-storage-credential-source=local}, it never asks the remote
catalog for credentials
+ * and instead issues them through this catalog's own credential providers,
scoped to the table
+ * location and to the caller's privilege, exactly as for a local catalog.
*
* <p>Portions of the table update handling are derived from Apache Iceberg's
{@code
* org.apache.iceberg.rest.CatalogHandlers}:
@@ -86,6 +94,33 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
private static final String X_ICEBERG_ACCESS_DELEGATION =
"X-Iceberg-Access-Delegation";
private static final String VENDED_CREDENTIALS = "vended-credentials";
+ /**
+ * Config key prefixes that carry storage credentials. With Gravitino
credential vending, keys
+ * with these prefixes are dropped from remote responses so remote
credentials never reach
+ * clients.
+ */
+ private static final List<String> REMOTE_CREDENTIAL_KEY_PREFIXES =
+ ImmutableList.of(
+ "s3.access-key-id",
+ "s3.secret-access-key",
+ "s3.session-token",
+ "s3.remote-signing-enabled",
+ "gcs.oauth2.",
+ "adls.sas-token",
+ "client.access-key-id",
+ "client.access-key-secret",
+ "client.security-token",
+ "client.refresh-credentials-",
+ "s3.refresh-credentials-",
+ "s3.sse.key",
+ "adls.token",
+ "adls.auth.shared-key.",
+ "gcs.client-lib-token",
+ "gcs.encryption-key",
+ "gcs.decryption-key",
+ "token",
+ "urn:ietf:params:oauth:token-type:");
+
/**
* Caches whether the remote catalog advertises the scan-plan endpoint. Only
successful lookups
* are cached, so a failure to reach the remote can be retried. Races just
repeat an idempotent
@@ -93,6 +128,9 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
*/
private volatile Boolean remoteSupportsScanPlan;
+ /** Whether storage credentials come from this catalog's providers rather
than the remote. */
+ private final boolean vendCredentialsLocally;
+
/**
* Creates a federated wrapper.
*
@@ -101,6 +139,62 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
*/
public FederatedCatalogWrapper(String catalogName, IcebergConfig config) {
super(catalogName, config);
+ // A service account's credentials reflect what the service account may
do, not the caller,
+ // so a catalog connected as a service account never passes the remote's
credentials through.
+ this.vendCredentialsLocally =
+ IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE_LOCAL.equals(
+
config.get(IcebergConfig.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE))
+ || config.isRestBackendServiceAccount();
+ }
+
+ /**
+ * Requires local authorization when the remote catalog is accessed as a
service account.
+ *
+ * @param serverContext the REST server context
+ */
+ @Override
+ public void validateServerContext(IcebergRESTServerContext serverContext) {
+ if (getIcebergConfig().isRestBackendServiceAccount()
+ && !serverContext.isAuthorizationEnabled()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Catalog %s connects to its remote catalog as a service account
(%s=%s), which"
+ + " requires authorization to be enabled",
+ catalogCredentialManager.catalogName(),
+ IcebergConstants.REST_BACKEND_AUTH_TYPE,
+ IcebergConstants.REST_BACKEND_AUTH_TYPE_SERVICE_ACCOUNT));
+ }
+ }
+
+ /**
+ * Reports whether the remote catalog receives the calling user's access
token.
+ *
+ * @return {@code true} for a user-token connection, or {@code false} for a
service account
+ */
+ @Override
+ public boolean forwardsUserToken() {
+ return !getIcebergConfig().isRestBackendServiceAccount();
+ }
+
+ /**
+ * Keeps cleanup of the remote catalog's files on the remote server.
+ *
+ * @return {@code false}, since this server must forward purge requests to
the remote catalog
+ */
+ @Override
+ public boolean supportsAsyncPurge() {
+ return false;
+ }
+
+ /**
+ * Forwards a purge to the remote catalog, which owns deletion of the
table's files.
+ *
+ * @param tableIdentifier the table to purge
+ */
+ @Override
+ public void purgeTable(TableIdentifier tableIdentifier) {
+ invalidateTableMetadata(tableIdentifier);
+ CatalogHandlers.purgeTable(getCatalog(), tableIdentifier);
}
/**
@@ -120,7 +214,15 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
@Override
public LoadTableResponse createTable(
Namespace namespace, CreateTableRequest request, boolean
requestCredential) {
- return createTableViaREST(namespace, request, requestCredential);
+ if (!vendCredentialsLocally) {
+ return createTableViaREST(namespace, request, requestCredential);
+ }
+ // The creator owns the new table, so it receives write credentials, as
for a local catalog.
+ return withLocalCredentials(
+ TableIdentifier.of(namespace, request.name()),
+ createTableViaREST(namespace, request, false),
+ requestCredential,
+ CredentialPrivilege.WRITE);
}
/**
@@ -129,18 +231,23 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
* <p>Always uses a dedicated REST GET rather than Iceberg's {@link
RESTCatalog#loadTable}, which
* cannot send {@code X-Iceberg-Access-Delegation}. When credential vending
is requested the
* header is forwarded so the remote catalog returns {@code
storage-credentials} inline. Upstream
- * credential refresh endpoints are rewritten to this IRC catalog. The
{@code privilege} is
- * ignored because the remote catalog decides what to vend.
+ * credential refresh endpoints are rewritten to this IRC catalog. With
Gravitino credential
+ * vending, the remote is not asked for credentials and this catalog's
providers issue them for
+ * the caller's {@code privilege}; otherwise the remote catalog decides what
to vend.
*
* @param identifier the table identifier.
* @param requestCredential whether the client requested vended credentials.
- * @param privilege ignored; the remote REST catalog vends its own
credentials.
+ * @param privilege the caller's privilege, used only with Gravitino
credential vending.
* @return the load-table response, including rewritten remote credentials
when requested.
*/
@Override
public LoadTableResponse loadTable(
TableIdentifier identifier, boolean requestCredential,
CredentialPrivilege privilege) {
- return loadTableViaREST(identifier, requestCredential);
+ if (!vendCredentialsLocally) {
+ return loadTableViaREST(identifier, requestCredential);
+ }
+ return withLocalCredentials(
+ identifier, loadTableViaREST(identifier, false), requestCredential,
privilege);
}
/**
@@ -159,7 +266,16 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
@Override
public LoadTableResponse registerTable(
Namespace namespace, RegisterTableRequest request, boolean
requestCredential) {
- return registerTableViaREST(namespace, request, requestCredential);
+ if (!vendCredentialsLocally) {
+ return registerTableViaREST(namespace, request, requestCredential);
+ }
+ // The registering user owns the table, so it receives write credentials,
as for a local
+ // catalog.
+ return withLocalCredentials(
+ TableIdentifier.of(namespace, request.name()),
+ registerTableViaREST(namespace, request, false),
+ requestCredential,
+ CredentialPrivilege.WRITE);
}
@Override
@@ -175,20 +291,23 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
}
/**
- * Fetches table credentials from the remote REST catalog instead of vending
Gravitino-managed
- * credentials. The {@code privilege} is ignored because the remote catalog
decides what to vend.
+ * Fetches table credentials from the remote REST catalog, or, with
Gravitino credential vending,
+ * issues them through this catalog's credential providers for the caller's
{@code privilege}.
*
* <p>The upstream credentials are rewritten via {@link
IcebergRESTUtils#rewriteTableCredentials}
* so their {@code refresh-credentials-endpoint} points at this IRC catalog
rather than the remote
* catalog, consistent with the {@link #loadTable}/{@code createTable}
federation paths.
*
* @param identifier the table identifier.
- * @param privilege ignored; the remote REST catalog vends its own
credentials.
+ * @param privilege the caller's privilege, used only with Gravitino
credential vending.
* @return the remote credentials with IRC-local refresh endpoints.
*/
@Override
public LoadCredentialsResponse getTableCredentials(
TableIdentifier identifier, CredentialPrivilege privilege) {
+ if (vendCredentialsLocally) {
+ return super.getTableCredentials(identifier, privilege);
+ }
LoadCredentialsResponse upstream =
getRESTTableCredentials((RESTCatalog) getCatalog(), identifier);
return IcebergRESTUtils.rewriteTableCredentials(
@@ -205,7 +324,7 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
* @param tableIdentifier the table to scan.
* @param scanRequest the scan request parameters.
* @param requestCredentialVending whether the client requested vended
credentials.
- * @param privilege ignored; the remote REST catalog decides what to vend.
+ * @param privilege the caller's privilege, used only with Gravitino
credential vending.
* @return the scan response from the remote catalog, with rewritten
credential refresh endpoints.
*/
@Override
@@ -215,6 +334,17 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
boolean requestCredentialVending,
CredentialPrivilege privilege) {
Table table = getCatalog().loadTable(tableIdentifier);
+ if (vendCredentialsLocally) {
+ // Drop anything the remote returned unasked, so its credentials never
reach the client.
+ PlanTableScanResponse response =
+ IcebergRESTUtils.copyWithCredentials(
+ getRESTTablePlanScan(
+ (RESTCatalog) getCatalog(), tableIdentifier, scanRequest,
false, table.specs()),
+ Collections.emptyList());
+ return requestCredentialVending &&
!isLocalOrHdfsLocation(table.location())
+ ? injectScanCredentials(tableIdentifier, table, response, privilege)
+ : response;
+ }
PlanTableScanResponse response =
getRESTTablePlanScan(
(RESTCatalog) getCatalog(),
@@ -527,6 +657,37 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
return rewriteRemoteLoadTable(TableIdentifier.of(namespace,
request.name()), upstream);
}
+ /**
+ * Adds credentials issued by this catalog's credential providers to a
response fetched from the
+ * remote catalog without vended credentials.
+ *
+ * @param identifier the table identifier
+ * @param remote the remote response
+ * @param requestCredential whether the client requested vended credentials
+ * @param privilege the caller's privilege on the table
+ * @return the response, with Gravitino-issued credentials when they were
requested and never with
+ * credentials from the remote catalog
+ */
+ private LoadTableResponse withLocalCredentials(
+ TableIdentifier identifier,
+ LoadTableResponse remote,
+ boolean requestCredential,
+ CredentialPrivilege privilege) {
+ // Drop anything the remote returned unasked, so its credentials never
reach the client.
+ Map<String, String> config =
+ Maps.filterKeys(
+ remote.config(),
+ key ->
REMOTE_CREDENTIAL_KEY_PREFIXES.stream().noneMatch(key::startsWith));
+ LoadTableResponse withoutRemoteCredentials =
+ LoadTableResponse.builder()
+ .withTableMetadata(remote.tableMetadata())
+ .addAllConfig(config)
+ .build();
+ return shouldGenerateCredential(withoutRemoteCredentials,
requestCredential)
+ ? injectCredentialConfig(identifier, withoutRemoteCredentials,
privilege)
+ : withoutRemoteCredentials;
+ }
+
private LoadTableResponse rewriteRemoteLoadTable(
TableIdentifier identifier, LoadTableResponse upstream) {
return IcebergRESTUtils.rewriteLoadTableCredentials(
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
index a960f175e0..0bc3af66b4 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
@@ -142,7 +142,14 @@ public class IcebergCatalogWrapperManager implements
AutoCloseable {
throw new NoSuchCatalogException(
"Couldn't find Iceberg configuration for catalog %s", catalogName);
}
- return createCatalogWrapper(catalogName, icebergConfig.get());
+ CatalogWrapperForREST wrapper = createCatalogWrapper(catalogName,
icebergConfig.get());
+ try {
+ wrapper.validateServerContext(serverContext);
+ return wrapper;
+ } catch (RuntimeException e) {
+ closeIcebergCatalogWrapper(wrapper);
+ throw e;
+ }
}
// Overriding this method to create a new CatalogWrapperForREST for test;
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
index fac0cf4af1..aa88f08673 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
@@ -27,8 +27,8 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.auth.AuthConstants;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.credential.CredentialPrivilege;
-import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
import org.apache.gravitino.iceberg.common.utils.IcebergIdentifierUtils;
+import org.apache.gravitino.iceberg.service.CatalogWrapperForREST;
import org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupJob;
@@ -134,7 +134,7 @@ public class IcebergTableOperationExecutor implements
IcebergTableOperationDispa
@Override
public void dropTable(
IcebergRequestContext context, TableIdentifier tableIdentifier, boolean
purgeRequested) {
- IcebergCatalogWrapper wrapper =
+ CatalogWrapperForREST wrapper =
icebergCatalogWrapperManager.getCatalogWrapper(context.catalogName());
if (!purgeRequested) {
wrapper.dropTable(tableIdentifier);
@@ -142,7 +142,7 @@ public class IcebergTableOperationExecutor implements
IcebergTableOperationDispa
}
// Async cleanup is opt-in per request and only wired in auxiliary mode;
otherwise purge inline.
- if (!context.asyncPurge()) {
+ if (!context.asyncPurge() || !wrapper.supportsAsyncPurge()) {
wrapper.purgeTable(tableIdentifier);
return;
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/server/web/filter/IcebergMetadataAuthorizationMethodInterceptor.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/server/web/filter/IcebergMetadataAuthorizationMethodInterceptor.java
index 66cefe6a58..eb744adf7f 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/server/web/filter/IcebergMetadataAuthorizationMethodInterceptor.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/server/web/filter/IcebergMetadataAuthorizationMethodInterceptor.java
@@ -31,7 +31,7 @@ import org.apache.gravitino.Entity;
import org.apache.gravitino.Entity.EntityType;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
-import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
+import org.apache.gravitino.iceberg.service.CatalogWrapperForREST;
import org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager;
import org.apache.gravitino.iceberg.service.IcebergExceptionMapper;
import org.apache.gravitino.iceberg.service.IcebergRESTUtils;
@@ -209,7 +209,7 @@ public class IcebergMetadataAuthorizationMethodInterceptor
return false;
}
- IcebergCatalogWrapper catalogWrapper;
+ CatalogWrapperForREST catalogWrapper;
try {
catalogWrapper = wrapperManager.getCatalogWrapper(catalogId.name());
} catch (NoSuchCatalogException | IllegalArgumentException e) {
@@ -219,8 +219,9 @@ public class IcebergMetadataAuthorizationMethodInterceptor
return false;
}
// When IRC2 is another Gravitino server, IRC1 acts as a proxy and does
not perform
- // authorization. IRC2 handles authorization.
- return catalogWrapper.isRESTCatalog();
+ // authorization. IRC2 handles authorization. A REST backend connected as
a service account is
+ // never skipped: the remote catalog sees only the service account, so
IRC1 must authorize.
+ return catalogWrapper.forwardsUserToken();
}
/** Metrics are stored locally, so a downstream REST catalog cannot
authorize the report. */
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 bc57dc4ada..8019302009 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
@@ -22,14 +22,17 @@ package org.apache.gravitino.iceberg.service;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
import com.google.common.collect.ImmutableMap;
import com.sun.net.httpserver.HttpServer;
import java.io.OutputStream;
+import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
@@ -42,10 +45,14 @@ import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import javax.annotation.Nullable;
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.common.cache.TableMetadataCache;
+import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
+import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
import org.apache.gravitino.iceberg.service.cache.LocalScanPlanCache;
import org.apache.gravitino.iceberg.service.extension.DummyCredentialProvider;
import org.apache.iceberg.BaseTable;
@@ -91,11 +98,118 @@ import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.InOrder;
public class TestCatalogWrapperForREST {
private static final AtomicBoolean CONSTRUCTION_IN_PROGRESS = new
AtomicBoolean(false);
+ @Test
+ void testLocalBackendPolicies() throws Exception {
+ IcebergConfig config =
+ new IcebergConfig(ImmutableMap.of(IcebergConstants.CATALOG_BACKEND,
"memory"));
+ try (CatalogWrapperForREST wrapper = new CatalogWrapperForREST("local",
config)) {
+ Assertions.assertFalse(wrapper.forwardsUserToken());
+ Assertions.assertTrue(wrapper.supportsAsyncPurge());
+ Assertions.assertDoesNotThrow(
+ () ->
wrapper.validateServerContext(mock(IcebergRESTServerContext.class)));
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "user-token, false",
+ "user-token, true",
+ "service-account, false",
+ "service-account, true"
+ })
+ void testFederatedBackendPolicies(String authType, boolean
authorizationEnabled)
+ throws Exception {
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "rest",
+ IcebergConstants.REST_BACKEND_AUTH_TYPE,
+ authType));
+ IcebergRESTServerContext context = mock(IcebergRESTServerContext.class);
+ when(context.isAuthorizationEnabled()).thenReturn(authorizationEnabled);
+ try (CatalogWrapperForREST wrapper = new FederatedCatalogWrapper("remote",
config)) {
+ Assertions.assertEquals("user-token".equals(authType),
wrapper.forwardsUserToken());
+ Assertions.assertFalse(wrapper.supportsAsyncPurge());
+ if ("service-account".equals(authType) && !authorizationEnabled) {
+ IllegalArgumentException failure =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
wrapper.validateServerContext(context));
+ Assertions.assertTrue(
+ failure.getMessage().contains("requires authorization to be
enabled"));
+ } else {
+ Assertions.assertDoesNotThrow(() ->
wrapper.validateServerContext(context));
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testFederatedPurgeForwardsToRemoteCatalog(boolean tableExists) throws
Exception {
+ TableIdentifier identifier = TableIdentifier.of("db", "table");
+ RESTCatalog catalog = mock(RESTCatalog.class);
+ when(catalog.dropTable(identifier, true)).thenReturn(tableExists);
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "rest",
+ IcebergConstants.TABLE_METADATA_CACHE_IMPL,
+ ""));
+ TableMetadataCache cache = mock(TableMetadataCache.class);
+
+ try (CatalogWrapperForREST wrapper =
+ new StaticCatalogWrapperForREST("remote", config, catalog)) {
+ Field cacheField =
IcebergCatalogWrapper.class.getDeclaredField("metadataCache");
+ cacheField.setAccessible(true);
+ cacheField.set(wrapper, cache);
+
+ if (tableExists) {
+ wrapper.purgeTable(identifier);
+ } else {
+ Assertions.assertThrows(NoSuchTableException.class, () ->
wrapper.purgeTable(identifier));
+ }
+
+ InOrder order = inOrder(cache, catalog);
+ order.verify(cache).invalidate(identifier);
+ order.verify(catalog).dropTable(identifier, true);
+ verifyNoMoreInteractions(catalog);
+ }
+ }
+
+ @Test
+ void testFederatedPurgePropagatesRemoteFailure() throws Exception {
+ TableIdentifier identifier = TableIdentifier.of("db", "table");
+ RESTCatalog catalog = mock(RESTCatalog.class);
+ ForbiddenException failure = new ForbiddenException("Remote purge denied");
+ when(catalog.dropTable(identifier, true)).thenThrow(failure);
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "rest",
+ IcebergConstants.TABLE_METADATA_CACHE_IMPL,
+ ""));
+
+ try (CatalogWrapperForREST wrapper =
+ new StaticCatalogWrapperForREST("remote", config, catalog)) {
+ Assertions.assertSame(
+ failure,
+ Assertions.assertThrows(ForbiddenException.class, () ->
wrapper.purgeTable(identifier)));
+ verify(catalog).dropTable(identifier, true);
+ verifyNoMoreInteractions(catalog);
+ }
+ }
+
@Test
void testCheckPropertiesForCompatibility() {
ImmutableMap<String, String> deprecatedMap = ImmutableMap.of("deprecated",
"new");
@@ -834,6 +948,106 @@ public class TestCatalogWrapperForREST {
}
}
+ @ParameterizedTest
+ @CsvSource({
+ "user-token, local",
+ "service-account, local",
+ "service-account, remote",
+ "service-account, "
+ })
+ void testFederatedLoadTableWithLocalStorageCredentials(
+ String authType, @Nullable String credentialSource) throws Exception {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ TableMetadata metadata =
+ TableMetadataParser.fromJson(
+ "s3://bucket/db/tbl/metadata/v1.metadata.json",
+ TableMetadataParser.toJson(
+ TableMetadata.newTableMetadata(
+ new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get())),
+ PartitionSpec.unpartitioned(),
+ SortOrder.unsorted(),
+ "s3://bucket/db/tbl",
+ Collections.emptyMap())));
+ // The remote returns broad credentials even when none are requested. With
Gravitino
+ // credential vending they must not reach the client.
+ Credential upstreamCredential =
+ IcebergRESTUtils.toRESTCredential(
+ "s3://bucket/",
+ ImmutableMap.of(
+ "s3.access-key-id", "upstream-key", "s3.secret-access-key",
"upstream-secret"));
+ String upstreamJson =
+ LoadTableResponseParser.toJson(
+ LoadTableResponse.builder()
+ .withTableMetadata(metadata)
+ .addAllConfig(ImmutableMap.of("io-impl",
"org.apache.iceberg.aws.s3.S3FileIO"))
+ .addCredential(upstreamCredential)
+ .build());
+
+ AtomicReference<String> accessDelegationHeader = new
AtomicReference<>("unset");
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ accessDelegationHeader.set(
+
exchange.getRequestHeaders().getFirst("X-Iceberg-Access-Delegation"));
+ 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"));
+
+ ImmutableMap.Builder<String, String> properties =
+ ImmutableMap.<String, String>builder()
+ .put(IcebergConstants.CATALOG_BACKEND, "memory")
+ .put(IcebergConstants.WAREHOUSE, "/tmp/warehouse")
+ .put(
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ DummyCredentialProvider.DUMMY_CREDENTIAL_TYPE)
+ .put(IcebergConstants.REST_BACKEND_AUTH_TYPE, authType);
+ if (credentialSource != null) {
+
properties.put(IcebergConstants.REST_BACKEND_STORAGE_CREDENTIAL_SOURCE,
credentialSource);
+ }
+ IcebergConfig config = new IcebergConfig(properties.build());
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ LoadTableResponse response = wrapper.loadTable(table, true,
CredentialPrivilege.READ);
+
+ Assertions.assertNull(accessDelegationHeader.get(), "remote must not be
asked to vend");
+ Assertions.assertEquals(1, response.credentials().size());
+ Credential credential = response.credentials().get(0);
+ Assertions.assertEquals("s3://bucket/db/tbl", credential.prefix());
+ Assertions.assertNotEquals("upstream-key",
credential.config().get("s3.access-key-id"));
+ Assertions.assertNotEquals("upstream-key",
response.config().get("s3.access-key-id"));
+ Assertions.assertEquals(
+ "org.apache.iceberg.aws.s3.S3FileIO",
response.config().get("io-impl"));
+
+ LoadTableResponse withoutRequest = wrapper.loadTable(table, false,
CredentialPrivilege.READ);
+ Assertions.assertNull(accessDelegationHeader.get());
+ Assertions.assertTrue(
+ withoutRequest.credentials().stream()
+ .noneMatch(c ->
"upstream-key".equals(c.config().get("s3.access-key-id"))),
+ "remote credentials returned without a request are passed through
only in remote mode");
+ } finally {
+ server.stop(0);
+ }
+ }
+
@Test
void testFederatedLoadTableNoCredentials() throws Exception {
TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
index 3a5e8925e1..fd07348c4b 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
@@ -34,6 +34,7 @@ import
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import
org.apache.gravitino.iceberg.service.provider.DynamicIcebergConfigProvider;
import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider;
import
org.apache.gravitino.iceberg.service.provider.IcebergConfigProviderFactory;
import org.junit.jupiter.api.AfterAll;
@@ -120,6 +121,67 @@ public class TestIcebergCatalogWrapperManagerForREST {
.contains("gravitino.iceberg-rest.catalog-config-provider=dynamic-config-provider"));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testServiceAccountAuthorizationValidatedByWrapper(boolean
authorizationEnabled)
+ throws Exception {
+ IcebergConfigProvider provider =
Mockito.mock(DynamicIcebergConfigProvider.class);
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "rest",
+ IcebergConstants.REST_BACKEND_AUTH_TYPE,
+ "service-account"));
+
Mockito.when(provider.getIcebergCatalogConfig("remote")).thenReturn(Optional.of(config));
+ try (IcebergCatalogWrapperManager manager =
+ new IcebergCatalogWrapperManager(ImmutableMap.of(), provider, false,
"metalake")) {
+ IcebergRESTServerContext.create(provider, authorizationEnabled, true,
true, manager);
+ if (authorizationEnabled) {
+ Assertions.assertInstanceOf(
+ FederatedCatalogWrapper.class,
manager.getCatalogWrapper("remote"));
+ } else {
+ IllegalArgumentException failure =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
manager.getCatalogWrapper("remote"));
+ Assertions.assertTrue(
+ failure.getMessage().contains("requires authorization to be
enabled"));
+ }
+ }
+ }
+
+ @Test
+ void testRejectedWrapperIsClosedAndNotCached() throws Exception {
+ IcebergConfigProvider provider = Mockito.mock(IcebergConfigProvider.class);
+ IcebergConfig config = new IcebergConfig(ImmutableMap.of());
+
Mockito.when(provider.getIcebergCatalogConfig("test")).thenReturn(Optional.of(config));
+ CatalogWrapperForREST wrapper = Mockito.mock(CatalogWrapperForREST.class);
+ Mockito.when(wrapper.getIcebergConfig()).thenReturn(config);
+ IllegalArgumentException failure = new IllegalArgumentException("invalid
server configuration");
+ try (IcebergCatalogWrapperManager manager =
+ Mockito.spy(
+ new IcebergCatalogWrapperManager(ImmutableMap.of(), provider,
false, "metalake"))) {
+ Mockito.doReturn(wrapper)
+ .when(manager)
+ .createCatalogWrapper(Mockito.eq("test"), Mockito.any());
+ IcebergRESTServerContext context =
+ IcebergRESTServerContext.create(provider, false, false, true,
manager);
+ Mockito.doThrow(failure).when(wrapper).validateServerContext(context);
+
+ Assertions.assertSame(
+ failure,
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
manager.getCatalogWrapper("test")));
+ Mockito.verify(wrapper).close();
+ Mockito.verify(wrapper, Mockito.never()).reloadHadoopConf();
+
+ Mockito.doNothing().when(wrapper).validateServerContext(context);
+ Assertions.assertSame(wrapper, manager.getCatalogWrapper("test"));
+ Mockito.verify(wrapper, Mockito.times(2)).validateServerContext(context);
+ Mockito.verify(wrapper).reloadHadoopConf();
+ }
+ }
+
@Test
public void testCreateFederatedWrapperForRestBackend() {
IcebergConfig icebergConfig =
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergAsyncPurge.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergAsyncPurge.java
index 02ef946745..7b975f50ab 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergAsyncPurge.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergAsyncPurge.java
@@ -85,6 +85,7 @@ class TestIcebergAsyncPurge {
@Test
void testAsyncDropEnqueuesJob() {
CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
+ when(wrapper.supportsAsyncPurge()).thenReturn(true);
IcebergCleanupManager cleanup = mock(IcebergCleanupManager.class);
TableMetadata metadata = mock(TableMetadata.class);
when(metadata.metadataFileLocation()).thenReturn("s3://b/db/t/metadata/0.json");
@@ -124,6 +125,7 @@ class TestIcebergAsyncPurge {
@Test
void testFallbackToSyncWhenDisabled() {
CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
+ when(wrapper.supportsAsyncPurge()).thenReturn(true);
tableExecutor(wrapper, Optional.empty()).dropTable(context(true), TABLE,
true);
@@ -131,6 +133,21 @@ class TestIcebergAsyncPurge {
verify(wrapper, never()).loadTableMetadata(any());
}
+ @Test
+ void testAsyncPurgeIsForwardedForRestCatalog() {
+ // A remote REST catalog deletes its own files, so an async purge request
is forwarded to it
+ // instead of being queued for this server to delete the files.
+ CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
+ IcebergCleanupManager cleanup = mock(IcebergCleanupManager.class);
+ when(wrapper.supportsAsyncPurge()).thenReturn(false);
+
+ tableExecutor(wrapper, Optional.of(cleanup)).dropTable(context(true),
TABLE, true);
+
+ verify(wrapper).purgeTable(TABLE);
+ verify(wrapper, never()).loadTableMetadata(any());
+ verify(cleanup, never()).addJob(any());
+ }
+
@Test
void testPlainDropWhenNotPurge() {
CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/server/web/filter/TestIcebergMetadataAuthorizationMethodInterceptor.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/server/web/filter/TestIcebergMetadataAuthorizationMethodInterceptor.java
index 6e2b26fd05..3ad571fd13 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/server/web/filter/TestIcebergMetadataAuthorizationMethodInterceptor.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/server/web/filter/TestIcebergMetadataAuthorizationMethodInterceptor.java
@@ -322,7 +322,7 @@ public class
TestIcebergMetadataAuthorizationMethodInterceptor {
RESTCatalog restCatalog = Mockito.mock(RESTCatalog.class);
Mockito.when(wrapperManager.getCatalogWrapper(TEST_CATALOG)).thenReturn(wrapper);
Mockito.when(wrapper.getCatalog()).thenReturn(restCatalog);
- Mockito.when(wrapper.isRESTCatalog()).thenReturn(true);
+ Mockito.when(wrapper.forwardsUserToken()).thenReturn(true);
resetContext(wrapperManager, true);
Method method =
@@ -379,6 +379,24 @@ public class
TestIcebergMetadataAuthorizationMethodInterceptor {
Map.of(Entity.EntityType.CATALOG, NameIdentifier.of(TEST_METALAKE,
"glue_catalog"))));
}
+ @Test
+ public void testServiceAccountRestCatalogDoesNotSkipAuthorization() {
+ // A REST backend connected as a service account presents one identity to
the remote catalog,
+ // so the remote cannot authorize individual users and this server must.
+ IcebergCatalogWrapperManager wrapperManager =
Mockito.mock(IcebergCatalogWrapperManager.class);
+ CatalogWrapperForREST wrapper = Mockito.mock(CatalogWrapperForREST.class);
+
Mockito.when(wrapperManager.getCatalogWrapper(TEST_CATALOG)).thenReturn(wrapper);
+ Mockito.when(wrapper.isRESTCatalog()).thenReturn(true);
+ Mockito.when(wrapper.forwardsUserToken()).thenReturn(false);
+ resetContext(wrapperManager, true);
+
+ IcebergMetadataAuthorizationMethodInterceptor interceptor =
+ new IcebergMetadataAuthorizationMethodInterceptor();
+ assertFalse(
+ interceptor.shouldSkipAuthorization(
+ Map.of(Entity.EntityType.CATALOG, NameIdentifier.of(TEST_METALAKE,
TEST_CATALOG))));
+ }
+
@Test
public void testConfigEndpointIsAuthorizedOnTheWarehouseCatalog() throws
Exception {
Assertions.assertTrue(
@@ -420,7 +438,7 @@ public class
TestIcebergMetadataAuthorizationMethodInterceptor {
IcebergCatalogWrapperManager wrapperManager =
Mockito.mock(IcebergCatalogWrapperManager.class);
CatalogWrapperForREST wrapper = Mockito.mock(CatalogWrapperForREST.class);
Mockito.when(wrapperManager.getCatalogWrapper(TEST_CATALOG)).thenReturn(wrapper);
- Mockito.when(wrapper.isRESTCatalog()).thenReturn(true);
+ Mockito.when(wrapper.forwardsUserToken()).thenReturn(true);
resetContext(wrapperManager, true);
Method method =
@@ -447,7 +465,7 @@ public class
TestIcebergMetadataAuthorizationMethodInterceptor {
Catalog nonRestCatalog = Mockito.mock(Catalog.class);
Mockito.when(wrapperManager.getCatalogWrapper(TEST_CATALOG)).thenReturn(wrapper);
Mockito.when(wrapper.getCatalog()).thenReturn(nonRestCatalog);
- Mockito.when(wrapper.isRESTCatalog()).thenReturn(false);
+ Mockito.when(wrapper.forwardsUserToken()).thenReturn(false);
resetContext(wrapperManager, true);
Method method =
@@ -525,7 +543,7 @@ public class
TestIcebergMetadataAuthorizationMethodInterceptor {
RESTCatalog restCatalog = Mockito.mock(RESTCatalog.class);
Mockito.when(wrapperManager.getCatalogWrapper(TEST_CATALOG)).thenReturn(wrapper);
Mockito.when(wrapper.getCatalog()).thenReturn(restCatalog);
- Mockito.when(wrapper.isRESTCatalog()).thenReturn(true);
+ Mockito.when(wrapper.forwardsUserToken()).thenReturn(true);
resetContext(wrapperManager, false);
Method method =
@@ -560,7 +578,7 @@ public class
TestIcebergMetadataAuthorizationMethodInterceptor {
RESTCatalog restCatalog = Mockito.mock(RESTCatalog.class);
Mockito.when(wrapperManager.getCatalogWrapper(TEST_CATALOG)).thenReturn(wrapper);
Mockito.when(wrapper.getCatalog()).thenReturn(restCatalog);
- Mockito.when(wrapper.isRESTCatalog()).thenReturn(true);
+ Mockito.when(wrapper.forwardsUserToken()).thenReturn(true);
resetContext(wrapperManager, true);
Method method =
