You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@polaris.apache.org by di...@apache.org on 2026/06/24 14:37:58 UTC
(polaris) branch main updated: Add GCS principal attribution to vended credentials via Workload Identity Federation (#4707)
This is an automated email from the ASF dual-hosted git repository.
dimas-b pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git
The following commit(s) were added to refs/heads/main by this push:
new 21e3555e37 Add GCS principal attribution to vended credentials via Workload Identity Federation (#4707)
21e3555e37 is described below
commit 21e3555e371a42bf39f8899eee025359fa6c1417
Author: Anand K Sankaran <li...@anands.net>
AuthorDate: Wed Jun 24 07:37:52 2026 -0700
Add GCS principal attribution to vended credentials via Workload Identity Federation (#4707)
* Add GCS principal attribution to vended credentials via Workload Identity Federation
GCP counterpart of AWS STS session tags. GCP downscoped credentials have no
session-tag mechanism, and x-goog-custom-audit-* request headers only reach GCS
audit logs if the client forwards them (arbitrary Iceberg clients do not), so
GCS Data Access logs cannot today be tied to the requesting Polaris principal.
This vends the attribution in the one channel that survives any client: the
identity of the credential itself. When configured, credential vending chains
catalog-signed JWT (sub=<realm>/<principal>) -> GCP STS token exchange (via
IdentityPoolCredentials programmatic supplier) -> tenant service-account
impersonation -> existing CAB downscoping, so every GCS Data Access audit entry
carries the principal in serviceAccountDelegationInfo.principalSubject.
New FeatureConfiguration flags:
- GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE
- GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER
- GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE
- GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_ID (kid for JWKS rotation)
There is no separate on/off flag: attribution activates once the audience,
issuer, and signing key file are all set, and additionally requires a
gcpServiceAccount on the storage config. When unconfigured, GCP vending is
unchanged.
GcpStorageCredentialCacheKey gains a principalName data field, populated only
when attribution is configured, so per-principal attributed tokens are never
shared across principals (and cross-principal cache reuse is preserved when
attribution is off). CredentialVendingContext already carries principalName, so
no new plumbing into the catalog core is required.
New classes: GcpAttributionSubjectBuilder (builds <realm>/<principal> within
GCP's 127-char google.subject limit) and GcpFederatedCredentialsExchanger
(mints the RS256 JWT via com.auth0:java-jwt, already in the version catalog;
performs the STS exchange via google-auth IdentityPoolCredentials, so no new
HTTP machinery; caches the parsed signing key JVM-wide).
Tests cover the subject budget/sanitization, JWT claims + kid, signing-key
caching, the IdentityPoolCredentials configuration, and per-principal cache-key
identity. polaris-core compiles, tests pass, spotless clean.
* Add com.auth0:java-jwt license entry to admin distribution LICENSE
The generateLicenseReport task for :polaris-admin requires all
non-Apache bundled dependencies to have a full license mention in
runtime/admin/distribution/LICENSE. The GCS principal attribution
feature introduced java-jwt (MIT) as a polaris-core implementation
dependency; the server distribution already had the entry but the
admin distribution did not.
* Update config docs site for GCS principal attribution flags
* GCS attribution: pre-compute params at cache key build time
* Add explicit GCS_PRINCIPAL_ATTRIBUTION_ENABLED flag with validation
* docs: regenerate config docs after updating GCS attribution descriptions
---------
Co-authored-by: Anand Kumar Sankaran <an...@workday.com>
---
CHANGELOG.md | 1 +
polaris-core/build.gradle.kts | 3 +
.../polaris/core/config/FeatureConfiguration.java | 61 +++++++
.../core/storage/gcp/GcpAttributionParams.java | 55 +++++++
.../storage/gcp/GcpAttributionSubjectBuilder.java | 90 +++++++++++
.../gcp/GcpCredentialsStorageIntegration.java | 148 ++++++++++++++++-
.../gcp/GcpFederatedCredentialsExchanger.java | 180 +++++++++++++++++++++
.../storage/gcp/GcpStorageCredentialCacheKey.java | 44 ++++-
.../gcp/GcpAttributionSubjectBuilderTest.java | 76 +++++++++
.../gcp/GcpCredentialsStorageIntegrationTest.java | 169 +++++++++++++++++++
.../gcp/GcpFederatedCredentialsExchangerTest.java | 156 ++++++++++++++++++
.../gcp/GcpStorageCredentialCacheKeyTest.java | 72 +++++++++
runtime/admin/distribution/LICENSE | 30 ++++
.../service/config/ProductionReadinessChecks.java | 91 +++++++++++
...roductionReadinessChecksGcsAttributionTest.java | 167 +++++++++++++++++++
.../config-sections/flags-polaris_features.md | 45 ++++++
16 files changed, 1373 insertions(+), 15 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 259ff75be4..22d92db626 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -44,6 +44,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
- Names containing any of these characters: <code>/\:*?"<>|#+`</code>
### New Features
+- Added GCS principal attribution for vended credentials (the GCP counterpart of AWS STS session tags). Set `GCS_PRINCIPAL_ATTRIBUTION_ENABLED=true` to activate; the feature flags `GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE`, `GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER`, and `GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE` are then required (a missing value is a fatal configuration error). Also requires a `gcpServiceAccount` on the catalog StorageConfiguration. When enabled, credential vending chain [...]
- Added `SESSION_NAME_FIELDS_IN_SUBSCOPED_CREDENTIAL` feature flag for AWS credential vending. Operators can now configure an ordered list of fields (`realm`, `catalog`, `namespace`, `table`, `principal`) to compose structured STS role session names (e.g. `p-acme-hr_catalog-employee-etl_writer`). Session names are sanitized and proportionally truncated to the AWS 64-character limit. When unset, existing `INCLUDE_PRINCIPAL_NAME_IN_SUBSCOPED_CREDENTIAL` behaviour is preserved.
- Added `hostUsers` support in Helm chart.
- Added documentation for BigQuery Metastore Catalog federation. Build with `-PNonRESTCatalogs=BIGQUERY` to include the BigQueryMetastoreCatalog federation extension. See `site/content/in-dev/unreleased/federation/bigquery-metastore-federation.md`.
diff --git a/polaris-core/build.gradle.kts b/polaris-core/build.gradle.kts
index f516a7826e..f35e0a58ff 100644
--- a/polaris-core/build.gradle.kts
+++ b/polaris-core/build.gradle.kts
@@ -70,6 +70,9 @@ dependencies {
implementation(platform(libs.google.cloud.storage.bom))
implementation("com.google.cloud:google-cloud-storage")
implementation(libs.google.cloud.iamcredentials)
+ // Signs short-lived attribution JWTs for GCS principal attribution via Workload Identity
+ // Federation (see GcpFederatedCredentialsExchanger).
+ implementation(libs.auth0.jwt)
testCompileOnly(project(":polaris-immutables"))
testAnnotationProcessor(project(":polaris-immutables", configuration = "processor"))
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java b/polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java
index 31db1d987e..acf5912aa7 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java
@@ -209,6 +209,67 @@ public class FeatureConfiguration<T> extends PolarisConfiguration<T> {
.defaultValue(List.<String>of())
.buildFeatureConfiguration();
+ public static final FeatureConfiguration<Boolean> GCS_PRINCIPAL_ATTRIBUTION_ENABLED =
+ PolarisConfiguration.<Boolean>builder()
+ .key("GCS_PRINCIPAL_ATTRIBUTION_ENABLED")
+ .description(
+ "Enables GCS principal attribution via Workload Identity Federation.\n"
+ + "When true, credential vending chains a catalog-signed JWT through an STS token\n"
+ + "exchange and service-account impersonation so the Polaris principal appears in GCS\n"
+ + "Data Access audit logs (serviceAccountDelegationInfo.principalSubject).\n"
+ + "Requires GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE, GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER,\n"
+ + "and GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE to also be set;\n"
+ + "a missing required value is a fatal configuration error.\n"
+ + "Also requires a gcpServiceAccount on the catalog StorageConfiguration.\n"
+ + "Default: false (attribution disabled).")
+ .defaultValue(false)
+ .buildFeatureConfiguration();
+
+ public static final FeatureConfiguration<String> GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE =
+ PolarisConfiguration.<String>builder()
+ .key("GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE")
+ .description(
+ "Full resource name of the Workload Identity Pool provider used for GCS principal\n"
+ + "attribution, e.g.\n"
+ + "//iam.googleapis.com/projects/<num>/locations/global/workloadIdentityPools/<pool>/providers/<provider>.\n"
+ + "Used as both the attribution JWT 'aud' claim and the STS token-exchange audience.\n"
+ + "Required when GCS_PRINCIPAL_ATTRIBUTION_ENABLED=true; ignored otherwise.")
+ .defaultValue("")
+ .buildFeatureConfiguration();
+
+ public static final FeatureConfiguration<String> GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER =
+ PolarisConfiguration.<String>builder()
+ .key("GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER")
+ .description(
+ "Issuer (iss claim) of catalog-minted GCS attribution JWTs; must match the issuer\n"
+ + "configured on the Workload Identity Pool OIDC provider. The provider verifies\n"
+ + "signatures against its uploaded JWKS, so no public discovery endpoint is required.\n"
+ + "Required when GCS_PRINCIPAL_ATTRIBUTION_ENABLED=true; ignored otherwise.")
+ .defaultValue("")
+ .buildFeatureConfiguration();
+
+ public static final FeatureConfiguration<String> GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE =
+ PolarisConfiguration.<String>builder()
+ .key("GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE")
+ .description(
+ "Filesystem path to the PKCS#8 PEM RSA private key used to sign GCS attribution JWTs\n"
+ + "(RS256). The corresponding public key must be published in the Workload Identity\n"
+ + "Pool provider's uploaded JWKS. Required when GCS_PRINCIPAL_ATTRIBUTION_ENABLED=true; ignored otherwise.")
+ .defaultValue("")
+ .buildFeatureConfiguration();
+
+ public static final FeatureConfiguration<String> GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_ID =
+ PolarisConfiguration.<String>builder()
+ .key("GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_ID")
+ .description(
+ "Key ID (kid) written into the header of GCS attribution JWTs so the Workload Identity\n"
+ + "Pool provider can select the right public key from its JWKS during key rotation\n"
+ + "(when the JWKS holds both the old and new keys). Must match the kid of the JWKS\n"
+ + "entry for the configured signing key. Empty omits the header (only safe with a\n"
+ + "single-key JWKS).")
+ .defaultValue("")
+ .buildFeatureConfiguration();
+
public static final FeatureConfiguration<Boolean> ALLOW_SETTING_S3_ENDPOINTS =
PolarisConfiguration.<Boolean>builder()
.key("ALLOW_SETTING_S3_ENDPOINTS")
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpAttributionParams.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpAttributionParams.java
new file mode 100644
index 0000000000..ac17543e98
--- /dev/null
+++ b/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpAttributionParams.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.core.storage.gcp;
+
+import java.nio.file.Path;
+import org.apache.polaris.immutables.PolarisImmutable;
+import org.immutables.value.Value;
+
+/**
+ * Pre-computed GCS principal attribution configuration parameters, evaluated at cache key build
+ * time. Present in the cache key only when attribution is fully configured and a principal is
+ * available; absent otherwise.
+ */
+@PolarisImmutable
+public interface GcpAttributionParams {
+
+ @Value.Parameter(order = 1)
+ String tokenIssuer();
+
+ @Value.Parameter(order = 2)
+ String wifAudience();
+
+ @Value.Parameter(order = 3)
+ String signingKeyFile();
+
+ @Value.Parameter(order = 4)
+ String signingKeyId();
+
+ /** Path derived from {@link #signingKeyFile()}; validated at construction time. */
+ @Value.Derived
+ default Path signingKeyPath() {
+ return Path.of(signingKeyFile());
+ }
+
+ static GcpAttributionParams of(
+ String tokenIssuer, String wifAudience, String signingKeyFile, String signingKeyId) {
+ return ImmutableGcpAttributionParams.of(tokenIssuer, wifAudience, signingKeyFile, signingKeyId);
+ }
+}
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpAttributionSubjectBuilder.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpAttributionSubjectBuilder.java
new file mode 100644
index 0000000000..c0a9769d45
--- /dev/null
+++ b/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpAttributionSubjectBuilder.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.core.storage.gcp;
+
+/**
+ * Builds the {@code sub} claim for GCS principal-attribution JWTs as {@code <realm>/<principal>},
+ * within GCP's 127-character {@code google.subject} limit.
+ *
+ * <p>The character budget mirrors the AWS session-name builder: one character is reserved for the
+ * separator, then each field receives an equal share of the remainder, and budget unused by a short
+ * field flows to the other. ISO control characters and the {@code /} separator are stripped from
+ * each field so the subject stays unambiguously parseable, and the {@code unknown} placeholder
+ * substitutes null/empty fields so the subject shape stays stable.
+ */
+public final class GcpAttributionSubjectBuilder {
+
+ /** GCP limit for the {@code google.subject} attribute of a federated identity. */
+ public static final int MAX_SUBJECT_LENGTH = 127;
+
+ static final String SEPARATOR = "/";
+
+ static final String VALUE_UNKNOWN = "unknown";
+
+ private GcpAttributionSubjectBuilder() {}
+
+ /**
+ * Builds the attribution subject {@code <realm>/<principal>}, guaranteed to be at most {@value
+ * #MAX_SUBJECT_LENGTH} characters.
+ *
+ * @param realm the realm identifier (gets first-half budget priority)
+ * @param principalName the Polaris principal name
+ * @return the subject string
+ */
+ public static String buildSubject(String realm, String principalName) {
+ String cleanRealm = sanitize(realm);
+ String cleanPrincipal = sanitize(principalName);
+
+ int budget = MAX_SUBJECT_LENGTH - SEPARATOR.length();
+ int remaining = budget;
+
+ int realmAlloc = remaining / 2;
+ int realmUsed = Math.min(cleanRealm.length(), realmAlloc);
+ remaining -= realmUsed;
+
+ int principalUsed = Math.min(cleanPrincipal.length(), remaining);
+ remaining -= principalUsed;
+
+ // Carry-forward: unused budget from either field flows to the other. A short realm gives
+ // principal more room via the larger `remaining`, and a short principal gives realm more room
+ // here.
+ int realmFinal = Math.min(cleanRealm.length(), realmUsed + remaining);
+
+ return cleanRealm.substring(0, realmFinal)
+ + SEPARATOR
+ + cleanPrincipal.substring(0, principalUsed);
+ }
+
+ private static String sanitize(String value) {
+ if (value == null || value.isEmpty()) {
+ return VALUE_UNKNOWN;
+ }
+ StringBuilder cleaned = new StringBuilder(value.length());
+ for (int i = 0; i < value.length(); i++) {
+ char c = value.charAt(i);
+ // Drop control chars and the separator itself so the subject stays unambiguously
+ // <realm>/<principal>: a value containing '/' would otherwise let an audit-log consumer
+ // mis-split it (e.g. principal "a/b" read as realm "a", principal "b").
+ if (!Character.isISOControl(c) && c != '/') {
+ cleaned.append(c);
+ }
+ }
+ return cleaned.length() == 0 ? VALUE_UNKNOWN : cleaned.toString();
+ }
+}
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpCredentialsStorageIntegration.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpCredentialsStorageIntegration.java
index 8ed4a843d9..c0ca262f07 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpCredentialsStorageIntegration.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpCredentialsStorageIntegration.java
@@ -46,6 +46,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import org.apache.polaris.core.config.FeatureConfiguration;
import org.apache.polaris.core.config.RealmConfig;
import org.apache.polaris.core.storage.CachingStorageIntegration;
import org.apache.polaris.core.storage.CredentialVendingContext;
@@ -55,6 +56,7 @@ import org.apache.polaris.core.storage.PolarisStorageIntegration;
import org.apache.polaris.core.storage.StorageAccessConfig;
import org.apache.polaris.core.storage.StorageAccessProperty;
import org.apache.polaris.core.storage.StorageUri;
+import org.apache.polaris.core.storage.cache.StorageCredentialCache;
import org.apache.polaris.core.storage.cache.StorageCredentialCacheKey;
import org.jspecify.annotations.NonNull;
import org.slf4j.Logger;
@@ -77,6 +79,7 @@ public class GcpCredentialsStorageIntegration
private final GoogleCredentials sourceCredentials;
private final HttpTransportFactory transportFactory;
private final GcpCredentialOps credentialOps;
+ private final Optional<GcpAttributionParams> attributionParams;
public GcpCredentialsStorageIntegration(
GoogleCredentials sourceCredentials,
@@ -89,13 +92,14 @@ public class GcpCredentialsStorageIntegration
null,
storageConfig,
realmConfig,
- GcpCredentialOps.DEFAULT);
+ GcpCredentialOps.DEFAULT,
+ resolveAttributionParams(realmConfig));
}
public GcpCredentialsStorageIntegration(
GoogleCredentials sourceCredentials,
HttpTransportFactory transportFactory,
- org.apache.polaris.core.storage.cache.StorageCredentialCache cache,
+ StorageCredentialCache cache,
GcpStorageConfigurationInfo storageConfig,
RealmConfig realmConfig) {
this(
@@ -104,7 +108,8 @@ public class GcpCredentialsStorageIntegration
cache,
storageConfig,
realmConfig,
- GcpCredentialOps.DEFAULT);
+ GcpCredentialOps.DEFAULT,
+ resolveAttributionParams(realmConfig));
}
public GcpCredentialsStorageIntegration(
@@ -113,16 +118,46 @@ public class GcpCredentialsStorageIntegration
GcpStorageConfigurationInfo storageConfig,
RealmConfig realmConfig,
GcpCredentialOps credentialOps) {
- this(sourceCredentials, transportFactory, null, storageConfig, realmConfig, credentialOps);
+ this(
+ sourceCredentials,
+ transportFactory,
+ null,
+ storageConfig,
+ realmConfig,
+ credentialOps,
+ resolveAttributionParams(realmConfig));
}
public GcpCredentialsStorageIntegration(
GoogleCredentials sourceCredentials,
HttpTransportFactory transportFactory,
- org.apache.polaris.core.storage.cache.StorageCredentialCache cache,
+ StorageCredentialCache cache,
GcpStorageConfigurationInfo storageConfig,
RealmConfig realmConfig,
GcpCredentialOps credentialOps) {
+ this(
+ sourceCredentials,
+ transportFactory,
+ cache,
+ storageConfig,
+ realmConfig,
+ credentialOps,
+ resolveAttributionParams(realmConfig));
+ }
+
+ /**
+ * Full constructor accepting pre-resolved attribution params. Downstream integrations can pass
+ * their own {@link GcpAttributionParams} without depending on {@link
+ * org.apache.polaris.core.config.FeatureConfiguration}.
+ */
+ public GcpCredentialsStorageIntegration(
+ GoogleCredentials sourceCredentials,
+ HttpTransportFactory transportFactory,
+ StorageCredentialCache cache,
+ GcpStorageConfigurationInfo storageConfig,
+ RealmConfig realmConfig,
+ GcpCredentialOps credentialOps,
+ Optional<GcpAttributionParams> attributionParams) {
super(cache, realmConfig, storageConfig);
// Needed for when environment variable GOOGLE_APPLICATION_CREDENTIALS points to google service
// account key json
@@ -130,6 +165,47 @@ public class GcpCredentialsStorageIntegration
sourceCredentials.createScoped("https://www.googleapis.com/auth/cloud-platform");
this.transportFactory = transportFactory;
this.credentialOps = credentialOps;
+ this.attributionParams = attributionParams;
+ }
+
+ /**
+ * Resolves {@link GcpAttributionParams} from realm config. Returns empty when attribution is
+ * disabled; throws {@link IllegalStateException} when enabled but misconfigured.
+ *
+ * <p>Static so it can be called from constructor-chaining expressions ({@code this(...)}).
+ */
+ public static Optional<GcpAttributionParams> resolveAttributionParams(RealmConfig realmConfig) {
+ if (!realmConfig.getConfig(FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_ENABLED)) {
+ return Optional.empty();
+ }
+ List<String> missing = new ArrayList<>();
+ if (realmConfig
+ .getConfig(FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE)
+ .isEmpty()) {
+ missing.add("GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE");
+ }
+ if (realmConfig
+ .getConfig(FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER)
+ .isEmpty()) {
+ missing.add("GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER");
+ }
+ if (realmConfig
+ .getConfig(FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE)
+ .isEmpty()) {
+ missing.add("GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE");
+ }
+ if (!missing.isEmpty()) {
+ throw new IllegalStateException(
+ "GCS_PRINCIPAL_ATTRIBUTION_ENABLED is true but the following required config values are"
+ + " missing: "
+ + String.join(", ", missing));
+ }
+ return Optional.of(
+ GcpAttributionParams.of(
+ realmConfig.getConfig(FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER),
+ realmConfig.getConfig(FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE),
+ realmConfig.getConfig(FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE),
+ realmConfig.getConfig(FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_ID)));
}
@Override
@@ -176,6 +252,28 @@ public class GcpCredentialsStorageIntegration
@NonNull Set<String> writeLocations,
@NonNull Optional<String> refreshEndpoint,
@NonNull CredentialVendingContext context) {
+ // Principal attribution makes the vended token per-principal, so the principal must
+ // participate in cache identity; otherwise it is left empty to preserve cross-principal cache
+ // reuse. Attribution requires a service account to impersonate and a principal to attribute.
+ Optional<String> principalName = Optional.empty();
+ Optional<GcpAttributionParams> resolvedAttributionParams = Optional.empty();
+ if (attributionParams.isPresent()) {
+ if (storageConfig().getGcpServiceAccount() == null) {
+ LOGGER.warn(
+ "GCS principal attribution is enabled but no gcpServiceAccount is configured"
+ + " on the StorageConfiguration; falling back to non-attributed credentials");
+ } else {
+ Optional<String> ctxPrincipal = context.principalName().filter(n -> !n.isEmpty());
+ if (ctxPrincipal.isPresent()) {
+ principalName = ctxPrincipal;
+ resolvedAttributionParams = attributionParams;
+ } else {
+ LOGGER.warn(
+ "GCS principal attribution is enabled but no principal name is present in the"
+ + " credential vending context; falling back to non-attributed credentials");
+ }
+ }
+ }
return GcpStorageCredentialCacheKey.of(
context.realm().orElse(""),
storageConfig(),
@@ -183,10 +281,12 @@ public class GcpCredentialsStorageIntegration
listLocations,
writeLocations,
refreshEndpoint,
+ principalName,
sourceCredentials,
transportFactory,
realmConfig(),
- credentialOps);
+ credentialOps,
+ resolvedAttributionParams);
}
/** Mint a fresh {@link StorageAccessConfig} for the given GCP cache key. */
@@ -206,7 +306,7 @@ public class GcpCredentialsStorageIntegration
}
GoogleCredentials credentialsToDownscope =
- getBaseCredentials(gcpStorageConfig, sourceCredentials, credentialOps);
+ resolveSourceCredentials(key, gcpStorageConfig, sourceCredentials, credentialOps);
CredentialAccessBoundary accessBoundary =
generateAccessBoundaryRules(readLocations, listLocations, writeLocations);
@@ -246,6 +346,40 @@ public class GcpCredentialsStorageIntegration
return accessConfig.build();
}
+ /**
+ * Returns the credential to be used as the source for downscoping.
+ *
+ * <p>When GCS principal attribution is configured and a principal is present (so the cache key
+ * carries pre-computed {@link GcpAttributionParams}), the impersonation source is a federated
+ * identity whose subject is {@code <realm>/<principal>}, which surfaces the principal in {@code
+ * serviceAccountDelegationInfo} of GCS Data Access audit logs. Otherwise this is the standard
+ * path: impersonate the configured service account from the ambient source credentials, or use
+ * those credentials directly.
+ */
+ private static GoogleCredentials resolveSourceCredentials(
+ GcpStorageCredentialCacheKey key,
+ GcpStorageConfigurationInfo storageConfig,
+ GoogleCredentials sourceCredentials,
+ GcpCredentialOps credentialOps) {
+ Optional<GcpAttributionParams> attributionParams = key.attributionParams();
+ if (attributionParams.isPresent()) {
+ GcpAttributionParams params = attributionParams.get();
+ String subject =
+ GcpAttributionSubjectBuilder.buildSubject(key.realmId(), key.principalName().orElse(""));
+ GcpFederatedCredentialsExchanger exchanger =
+ new GcpFederatedCredentialsExchanger(
+ params.tokenIssuer(),
+ params.wifAudience(),
+ params.signingKeyPath(),
+ params.signingKeyId(),
+ key.transportFactory());
+ GoogleCredentials federated = exchanger.federatedCredentials(subject, key.realmId());
+ return createImpersonatedCredentials(
+ federated, storageConfig.getGcpServiceAccount(), credentialOps);
+ }
+ return getBaseCredentials(storageConfig, sourceCredentials, credentialOps);
+ }
+
/**
* Returns the credential to be used as the source for downscoping. If a specific service account
* is configured, it impersonates that account first.
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpFederatedCredentialsExchanger.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpFederatedCredentialsExchanger.java
new file mode 100644
index 0000000000..1a1ab5606d
--- /dev/null
+++ b/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpFederatedCredentialsExchanger.java
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.core.storage.gcp;
+
+import com.auth0.jwt.JWT;
+import com.auth0.jwt.JWTCreator;
+import com.auth0.jwt.algorithms.Algorithm;
+import com.google.auth.http.HttpTransportFactory;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auth.oauth2.IdentityPoolCredentials;
+import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyFactory;
+import java.security.interfaces.RSAPrivateKey;
+import java.security.spec.PKCS8EncodedKeySpec;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Base64;
+import java.util.Date;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Produces a GCP federated {@link GoogleCredentials} whose identity carries {@code
+ * <realm>/<principal>}, so that GCS Data Access audit logs attribute access to the requesting
+ * Polaris principal. This is the GCP counterpart of AWS STS session tags.
+ *
+ * <p>The federated credential is an {@link IdentityPoolCredentials} backed by a programmatic
+ * subject-token supplier: on each token refresh google-auth invokes the supplier, which mints a
+ * short-lived RS256 JWT ({@code sub = <realm>/<principal>}, {@code realm} claim), and exchanges it
+ * at the Workload Identity Pool provider's STS endpoint. The provider maps {@code google.subject =
+ * assertion.sub} and {@code attribute.realm = assertion.realm}; per-realm {@code attribute.realm}
+ * IAM bindings then enforce that a realm-A identity can only impersonate realm-A's service account.
+ * The returned credential is intended to be used as the source for tenant service-account
+ * impersonation (see {@link GcpCredentialsStorageIntegration}).
+ *
+ * <p>Network note: this performs an STS token exchange against {@code sts.googleapis.com} in
+ * addition to the existing {@code iamcredentials.googleapis.com} and {@code storage.googleapis.com}
+ * traffic.
+ */
+public class GcpFederatedCredentialsExchanger {
+
+ static final String STS_TOKEN_URL = "https://sts.googleapis.com/v1/token";
+ static final String SUBJECT_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:jwt";
+ static final String CLOUD_PLATFORM_SCOPE = "https://www.googleapis.com/auth/cloud-platform";
+
+ /** Attribution JWTs are single-purpose and short-lived. */
+ static final Duration JWT_LIFETIME = Duration.ofMinutes(5);
+
+ /**
+ * JVM-wide cache of parsed signing keys, keyed by file path. The key file is a stable pod-mounted
+ * secret; parsing it (disk read + {@link KeyFactory}) once per path amortizes across vends rather
+ * than re-reading on every credential-cache miss. Key rotation is delivered by a process restart
+ * (the secret is mounted at startup), which clears this cache. In practice the map is bounded by
+ * the number of realms in the deployment — each realm has at most one signing-key path — so
+ * memory growth is proportional to realm count.
+ */
+ private static final ConcurrentHashMap<Path, RSAPrivateKey> SIGNING_KEY_CACHE =
+ new ConcurrentHashMap<>();
+
+ private final String issuer;
+ private final String wifAudience;
+ private final Path signingKeyPath;
+ private final String signingKeyId;
+ private final HttpTransportFactory transportFactory;
+
+ public GcpFederatedCredentialsExchanger(
+ String issuer,
+ String wifAudience,
+ Path signingKeyPath,
+ String signingKeyId,
+ HttpTransportFactory transportFactory) {
+ this.issuer = issuer;
+ this.wifAudience = wifAudience;
+ this.signingKeyPath = signingKeyPath;
+ this.signingKeyId = signingKeyId;
+ this.transportFactory = transportFactory;
+ }
+
+ /**
+ * Builds a federated credential whose subject is {@code <realm>/<principal>}.
+ *
+ * @param subject the attribution subject, {@code <realm>/<principal>} (see {@link
+ * GcpAttributionSubjectBuilder})
+ * @param realm the realm identifier, emitted as the {@code realm} claim for {@code
+ * attribute.realm} mapping
+ * @return federated credentials suitable as the source for tenant-SA impersonation
+ */
+ public GoogleCredentials federatedCredentials(String subject, String realm) {
+ return IdentityPoolCredentials.newBuilder()
+ .setHttpTransportFactory(transportFactory)
+ .setAudience(wifAudience)
+ .setSubjectTokenType(SUBJECT_TOKEN_TYPE)
+ .setTokenUrl(STS_TOKEN_URL)
+ .setScopes(List.of(CLOUD_PLATFORM_SCOPE))
+ .setSubjectTokenSupplier(context -> mintAttributionJwt(subject, realm))
+ .build();
+ }
+
+ @VisibleForTesting
+ String mintAttributionJwt(String subject, String realm) throws IOException {
+ Instant now = Instant.now();
+ JWTCreator.Builder builder =
+ JWT.create()
+ .withIssuer(issuer)
+ .withSubject(subject)
+ .withAudience(wifAudience)
+ .withClaim("realm", realm)
+ .withIssuedAt(Date.from(now))
+ .withExpiresAt(Date.from(now.plus(JWT_LIFETIME)))
+ .withJWTId(UUID.randomUUID().toString());
+ // Set the kid header so the WIF provider can pick the right public key from its JWKS during
+ // rotation (when the JWKS holds both the old and new keys). Omitted only for a single-key JWKS.
+ if (signingKeyId != null && !signingKeyId.isEmpty()) {
+ builder.withKeyId(signingKeyId);
+ }
+ // null public key: sign-only; verification is the WIF provider's responsibility
+ return builder.sign(Algorithm.RSA256(null, loadSigningKey()));
+ }
+
+ private RSAPrivateKey loadSigningKey() throws IOException {
+ try {
+ return SIGNING_KEY_CACHE.computeIfAbsent(
+ signingKeyPath,
+ path -> {
+ try {
+ return readPkcs8PrivateKey(path);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ });
+ } catch (UncheckedIOException e) {
+ throw e.getCause();
+ }
+ }
+
+ /** Reads an RSA private key from a PKCS#8 PEM file. */
+ @VisibleForTesting
+ static RSAPrivateKey readPkcs8PrivateKey(Path pemPath) throws IOException {
+ String pem = Files.readString(pemPath);
+ if (pem.contains("BEGIN RSA PRIVATE KEY")) {
+ throw new IOException(
+ "PKCS#1 key format (BEGIN RSA PRIVATE KEY) is not supported for GCS attribution from "
+ + pemPath
+ + "; convert to PKCS#8 with: openssl pkcs8 -topk8 -nocrypt -in key.pem -out key-pkcs8.pem");
+ }
+ String base64 =
+ pem.replaceAll("-----BEGIN [A-Z ]+-----", "")
+ .replaceAll("-----END [A-Z ]+-----", "")
+ .replaceAll("\\s", "");
+ try {
+ byte[] der = Base64.getDecoder().decode(base64);
+ KeyFactory keyFactory = KeyFactory.getInstance("RSA");
+ return (RSAPrivateKey) keyFactory.generatePrivate(new PKCS8EncodedKeySpec(der));
+ } catch (Exception e) {
+ throw new IOException(
+ "Unable to read PKCS#8 RSA private key for GCS attribution from " + pemPath, e);
+ }
+ }
+}
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpStorageCredentialCacheKey.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpStorageCredentialCacheKey.java
index 8691c7d257..09a5b146d2 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpStorageCredentialCacheKey.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpStorageCredentialCacheKey.java
@@ -29,8 +29,14 @@ import org.apache.polaris.immutables.PolarisImmutable;
import org.immutables.value.Value;
/**
- * Cache key for vended GCP credentials. GCP downscoped credentials do not support session tags, so
- * principal and credential vending context are never included.
+ * Cache key for vended GCP credentials.
+ *
+ * <p>By default GCP downscoped credentials are principal-independent, so {@link #principalName()}
+ * is empty and credentials are shared across principals. When GCS principal attribution via
+ * Workload Identity Federation is enabled (see {@code
+ * FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_*}), the vended token is derived from a
+ * per-principal federated identity, so the principal name is included here to ensure one
+ * principal's attributed credentials are never served to another.
*/
@PolarisImmutable
public interface GcpStorageCredentialCacheKey extends StorageCredentialCacheKey {
@@ -55,25 +61,43 @@ public interface GcpStorageCredentialCacheKey extends StorageCredentialCacheKey
@Value.Parameter(order = 6)
Optional<String> refreshCredentialsEndpoint();
+ /**
+ * The requesting principal name, included in the cache key only when GCS principal attribution is
+ * enabled (otherwise empty). When attribution is active the vended credential carries this
+ * principal's identity, so it must participate in cache identity to avoid serving one principal's
+ * attributed credentials to another.
+ */
+ @Value.Parameter(order = 7)
+ Optional<String> principalName();
+
// ---- aux: app-scoped invariants, excluded from equals/hashCode ----
- @Value.Parameter(order = 7)
+ @Value.Parameter(order = 8)
@Value.Auxiliary
GoogleCredentials sourceCredentials();
- @Value.Parameter(order = 8)
+ @Value.Parameter(order = 9)
@Value.Auxiliary
HttpTransportFactory transportFactory();
@Override
- @Value.Parameter(order = 9)
+ @Value.Parameter(order = 10)
@Value.Auxiliary
RealmConfig realmConfig();
- @Value.Parameter(order = 10)
+ @Value.Parameter(order = 11)
@Value.Auxiliary
GcpCredentialOps credentialOps();
+ /**
+ * Pre-computed attribution config parameters, present only when GCS principal attribution is
+ * fully configured and a principal is available. Computed at cache key build time so {@code
+ * compute()} can use these values directly without re-reading realm config.
+ */
+ @Value.Parameter(order = 12)
+ @Value.Auxiliary
+ Optional<GcpAttributionParams> attributionParams();
+
@Override
default StorageAccessConfig load() {
return GcpCredentialsStorageIntegration.compute(this);
@@ -86,10 +110,12 @@ public interface GcpStorageCredentialCacheKey extends StorageCredentialCacheKey
Set<String> allowedListLocations,
Set<String> allowedWriteLocations,
Optional<String> refreshCredentialsEndpoint,
+ Optional<String> principalName,
GoogleCredentials sourceCredentials,
HttpTransportFactory transportFactory,
RealmConfig realmConfig,
- GcpCredentialOps credentialOps) {
+ GcpCredentialOps credentialOps,
+ Optional<GcpAttributionParams> attributionParams) {
return ImmutableGcpStorageCredentialCacheKey.of(
realmId,
storageConfig,
@@ -97,9 +123,11 @@ public interface GcpStorageCredentialCacheKey extends StorageCredentialCacheKey
allowedListLocations,
allowedWriteLocations,
refreshCredentialsEndpoint,
+ principalName,
sourceCredentials,
transportFactory,
realmConfig,
- credentialOps);
+ credentialOps,
+ attributionParams);
}
}
diff --git a/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpAttributionSubjectBuilderTest.java b/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpAttributionSubjectBuilderTest.java
new file mode 100644
index 0000000000..17b09e91ed
--- /dev/null
+++ b/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpAttributionSubjectBuilderTest.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.core.storage.gcp;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+class GcpAttributionSubjectBuilderTest {
+
+ @Test
+ void simpleSubject() {
+ assertThat(GcpAttributionSubjectBuilder.buildSubject("tenant1", "etl_writer"))
+ .isEqualTo("tenant1/etl_writer");
+ }
+
+ @Test
+ void neverExceedsGcpLimit() {
+ assertThat(GcpAttributionSubjectBuilder.buildSubject("r".repeat(500), "p".repeat(500)))
+ .hasSize(GcpAttributionSubjectBuilder.MAX_SUBJECT_LENGTH)
+ .contains("/");
+ }
+
+ @Test
+ void shortRealmGivesPrincipalMoreBudget() {
+ String subject = GcpAttributionSubjectBuilder.buildSubject("t1", "p".repeat(500));
+ // 127 - "t1/" (3 chars) = 124 chars of principal
+ assertThat(subject).isEqualTo("t1/" + "p".repeat(124));
+ }
+
+ @Test
+ void shortPrincipalGivesRealmMoreBudget() {
+ String subject = GcpAttributionSubjectBuilder.buildSubject("r".repeat(500), "me");
+ assertThat(subject).isEqualTo("r".repeat(124) + "/me");
+ }
+
+ @Test
+ void bothLongSplitBudgetEvenly() {
+ String subject = GcpAttributionSubjectBuilder.buildSubject("r".repeat(500), "p".repeat(500));
+ // budget = 126; realm gets floor(126/2)=63, principal gets the remaining 63.
+ assertThat(subject).isEqualTo("r".repeat(63) + "/" + "p".repeat(63));
+ }
+
+ @Test
+ void nullAndEmptyBecomeUnknown() {
+ assertThat(GcpAttributionSubjectBuilder.buildSubject(null, "p")).isEqualTo("unknown/p");
+ assertThat(GcpAttributionSubjectBuilder.buildSubject("r", "")).isEqualTo("r/unknown");
+ assertThat(GcpAttributionSubjectBuilder.buildSubject(null, null)).isEqualTo("unknown/unknown");
+ }
+
+ @Test
+ void controlCharsAndSeparatorStripped() {
+ assertThat(GcpAttributionSubjectBuilder.buildSubject("ten\r\nant", "etl\twriter"))
+ .isEqualTo("tenant/etlwriter");
+ // A '/' in a field must not introduce a second separator.
+ String subject = GcpAttributionSubjectBuilder.buildSubject("tenant1", "a/b/c");
+ assertThat(subject).isEqualTo("tenant1/abc");
+ assertThat(subject.chars().filter(c -> c == '/').count()).isEqualTo(1);
+ }
+}
diff --git a/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpCredentialsStorageIntegrationTest.java b/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpCredentialsStorageIntegrationTest.java
index cd8f824a0b..d752efc118 100644
--- a/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpCredentialsStorageIntegrationTest.java
+++ b/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpCredentialsStorageIntegrationTest.java
@@ -20,7 +20,10 @@ package org.apache.polaris.core.storage.gcp;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import com.auth0.jwt.JWT;
+import com.auth0.jwt.interfaces.DecodedJWT;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -32,6 +35,7 @@ import com.google.auth.oauth2.AccessToken;
import com.google.auth.oauth2.CredentialAccessBoundary;
import com.google.auth.oauth2.DownscopedCredentials;
import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auth.oauth2.IdentityPoolCredentials;
import com.google.cloud.ServiceOptions;
import com.google.cloud.iam.credentials.v1.GenerateAccessTokenRequest;
import com.google.cloud.iam.credentials.v1.GenerateAccessTokenResponse;
@@ -44,15 +48,26 @@ import com.google.cloud.storage.StorageOptions;
import com.google.protobuf.Timestamp;
import java.io.IOException;
import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Base64;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.config.RealmConfigImpl;
+import org.apache.polaris.core.config.RealmConfigurationSource;
import org.apache.polaris.core.storage.BaseStorageIntegrationTest;
+import org.apache.polaris.core.storage.CredentialVendingContext;
import org.apache.polaris.core.storage.LocationGrant;
import org.apache.polaris.core.storage.PolarisStorageActions;
import org.apache.polaris.core.storage.StorageAccessConfig;
@@ -60,6 +75,7 @@ import org.apache.polaris.core.storage.StorageAccessProperty;
import org.assertj.core.api.Assertions;
import org.assertj.core.api.Assumptions;
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.Arguments;
import org.junit.jupiter.params.provider.CsvSource;
@@ -502,4 +518,157 @@ class GcpCredentialsStorageIntegrationTest extends BaseStorageIntegrationTest {
.path("expression")
.asText();
}
+
+ private static RealmConfig configWith(Map<String, String> values) {
+ RealmConfigurationSource source = (rc, name) -> values.get(name);
+ return new RealmConfigImpl(source, () -> "test-realm");
+ }
+
+ @Test
+ void resolveAttributionParamsDisabledByDefault() {
+ assertThat(GcpCredentialsStorageIntegration.resolveAttributionParams(EMPTY_REALM_CONFIG))
+ .isEmpty();
+ }
+
+ @Test
+ void resolveAttributionParamsReturnsParamsWhenFullyConfigured() {
+ RealmConfig config =
+ configWith(
+ Map.of(
+ "GCS_PRINCIPAL_ATTRIBUTION_ENABLED", "true",
+ "GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE",
+ "//iam.googleapis.com/projects/123/locations/global/workloadIdentityPools/p/providers/pr",
+ "GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER", "https://issuer.example.com",
+ "GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE", "/tmp/key.pem"));
+ Optional<GcpAttributionParams> result =
+ GcpCredentialsStorageIntegration.resolveAttributionParams(config);
+ assertThat(result).isPresent();
+ assertThat(result.get().wifAudience())
+ .isEqualTo(
+ "//iam.googleapis.com/projects/123/locations/global/workloadIdentityPools/p/providers/pr");
+ assertThat(result.get().tokenIssuer()).isEqualTo("https://issuer.example.com");
+ assertThat(result.get().signingKeyFile()).isEqualTo("/tmp/key.pem");
+ }
+
+ @Test
+ void resolveAttributionParamsThrowsWhenEnabledButAudienceMissing() {
+ RealmConfig config =
+ configWith(
+ Map.of(
+ "GCS_PRINCIPAL_ATTRIBUTION_ENABLED", "true",
+ "GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER", "https://issuer.example.com",
+ "GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE", "/tmp/key.pem"));
+ assertThatThrownBy(() -> GcpCredentialsStorageIntegration.resolveAttributionParams(config))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE");
+ }
+
+ @Test
+ void resolveAttributionParamsThrowsWhenMultipleRequiredParamsMissing() {
+ RealmConfig config = configWith(Map.of("GCS_PRINCIPAL_ATTRIBUTION_ENABLED", "true"));
+ assertThatThrownBy(() -> GcpCredentialsStorageIntegration.resolveAttributionParams(config))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE")
+ .hasMessageContaining("GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER")
+ .hasMessageContaining("GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE");
+ }
+
+ @Test
+ void testWifAttributionFlowUsesFederatedCredentials(@TempDir Path tempDir) throws Exception {
+ String serviceAccount = "test-sa@project.iam.gserviceaccount.com";
+ GcpStorageConfigurationInfo config =
+ GcpStorageConfigurationInfo.builder()
+ .addAllAllowedLocations(List.of("gs://bucket/path"))
+ .gcpServiceAccount(serviceAccount)
+ .build();
+
+ AtomicReference<GoogleCredentials> capturedSource = new AtomicReference<>();
+
+ IamCredentialsClient mockIamClient = Mockito.mock(IamCredentialsClient.class);
+ GenerateAccessTokenResponse mockResponse =
+ GenerateAccessTokenResponse.newBuilder()
+ .setAccessToken("wif-impersonated-token")
+ .setExpireTime(
+ Timestamp.newBuilder().setSeconds(System.currentTimeMillis() / 1000 + 3600).build())
+ .build();
+ Mockito.when(mockIamClient.generateAccessToken(Mockito.any(GenerateAccessTokenRequest.class)))
+ .thenReturn(mockResponse);
+
+ GoogleCredentials mockCreds = Mockito.mock(GoogleCredentials.class);
+ Mockito.when(mockCreds.createScoped(Mockito.any(String.class))).thenReturn(mockCreds);
+
+ GcpCredentialOps testOps =
+ new GcpCredentialOps() {
+ @Override
+ public IamCredentialsClient createIamCredentialsClient(GoogleCredentials credentials) {
+ capturedSource.set(credentials);
+ return mockIamClient;
+ }
+
+ @Override
+ public AccessToken refreshAccessToken(DownscopedCredentials credentials) {
+ return new AccessToken("downscoped-token", new Date());
+ }
+ };
+
+ KeyPairGenerator gen = KeyPairGenerator.getInstance("RSA");
+ gen.initialize(2048);
+ KeyPair keyPair = gen.generateKeyPair();
+ String pem =
+ "-----BEGIN PRIVATE KEY-----\n"
+ + Base64.getMimeEncoder(64, "\n".getBytes(UTF_8))
+ .encodeToString(keyPair.getPrivate().getEncoded())
+ + "\n-----END PRIVATE KEY-----\n";
+ Path keyFile = tempDir.resolve("attribution-key.pem");
+ Files.writeString(keyFile, pem);
+
+ GcpAttributionParams params =
+ GcpAttributionParams.of(
+ "https://issuer.example.com",
+ "//iam.googleapis.com/projects/123/locations/global/workloadIdentityPools/p/providers/pr",
+ keyFile.toString(),
+ "key-1");
+
+ GcpCredentialsStorageIntegration integration =
+ new GcpCredentialsStorageIntegration(
+ mockCreds,
+ ServiceOptions.getFromServiceLoader(HttpTransportFactory.class, NetHttpTransport::new),
+ null,
+ config,
+ EMPTY_REALM_CONFIG,
+ testOps,
+ Optional.of(params));
+
+ integration.getStorageAccessConfig(
+ toGrants(
+ Set.of("gs://bucket/path"), Set.of("gs://bucket/path"), Set.of("gs://bucket/path")),
+ Optional.empty(),
+ CredentialVendingContext.builder()
+ .realm(Optional.of("realm1"))
+ .principalName(Optional.of("principal1"))
+ .build());
+
+ Mockito.verify(mockIamClient)
+ .generateAccessToken(
+ Mockito.argThat(
+ request ->
+ request
+ .getName()
+ .equals(
+ GcpCredentialsStorageIntegration.SERVICE_ACCOUNT_PREFIX
+ + serviceAccount)));
+
+ // The source credential used for impersonation must be a federated identity, not the raw
+ // source credential. This ensures the attribution JWT (not the ambient SA key) is what
+ // reaches GCP STS and surfaces in serviceAccountDelegationInfo of audit logs.
+ assertThat(capturedSource.get()).isInstanceOf(IdentityPoolCredentials.class);
+
+ // Verify the JWT minted by the supplier carries the correct attribution subject.
+ // retrieveSubjectToken() invokes the supplier locally (no HTTP); it signs the JWT with the
+ // test RSA key, so we can decode it and assert the sub and realm claims are correct.
+ IdentityPoolCredentials idp = (IdentityPoolCredentials) capturedSource.get();
+ DecodedJWT jwt = JWT.decode(idp.retrieveSubjectToken());
+ assertThat(jwt.getSubject()).isEqualTo("realm1/principal1");
+ assertThat(jwt.getClaim("realm").asString()).isEqualTo("realm1");
+ }
}
diff --git a/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpFederatedCredentialsExchangerTest.java b/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpFederatedCredentialsExchangerTest.java
new file mode 100644
index 0000000000..319eb0f101
--- /dev/null
+++ b/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpFederatedCredentialsExchangerTest.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.core.storage.gcp;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.auth0.jwt.JWT;
+import com.auth0.jwt.interfaces.DecodedJWT;
+import com.google.api.client.http.javanet.NetHttpTransport;
+import com.google.auth.http.HttpTransportFactory;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auth.oauth2.IdentityPoolCredentials;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
+import java.util.Base64;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class GcpFederatedCredentialsExchangerTest {
+
+ private static final String ISSUER = "https://catalog.example.org/attribution";
+ private static final String AUDIENCE =
+ "//iam.googleapis.com/projects/123456/locations/global/workloadIdentityPools/polaris/providers/catalog";
+ private static final String KEY_ID = "attribution-key-1";
+ private static final HttpTransportFactory TRANSPORT = NetHttpTransport::new;
+
+ @TempDir Path tempDir;
+ private Path keyFile;
+
+ @BeforeEach
+ void writeSigningKey() throws Exception {
+ KeyPairGenerator generator = KeyPairGenerator.getInstance("RSA");
+ generator.initialize(2048);
+ KeyPair keyPair = generator.generateKeyPair();
+ String pem =
+ "-----BEGIN PRIVATE KEY-----\n"
+ + Base64.getMimeEncoder(64, "\n".getBytes(StandardCharsets.UTF_8))
+ .encodeToString(keyPair.getPrivate().getEncoded())
+ + "\n-----END PRIVATE KEY-----\n";
+ keyFile = tempDir.resolve("attribution-key.pem");
+ Files.writeString(keyFile, pem);
+ }
+
+ private GcpFederatedCredentialsExchanger exchanger(String keyId) {
+ return new GcpFederatedCredentialsExchanger(ISSUER, AUDIENCE, keyFile, keyId, TRANSPORT);
+ }
+
+ @Test
+ void attributionJwtCarriesExpectedClaimsAndKid() throws IOException {
+ DecodedJWT jwt =
+ JWT.decode(exchanger(KEY_ID).mintAttributionJwt("tenant1/etl_writer", "tenant1"));
+
+ assertThat(jwt.getIssuer()).isEqualTo(ISSUER);
+ assertThat(jwt.getSubject()).isEqualTo("tenant1/etl_writer");
+ assertThat(jwt.getAudience()).containsExactly(AUDIENCE);
+ assertThat(jwt.getClaim("realm").asString()).isEqualTo("tenant1");
+ assertThat(jwt.getAlgorithm()).isEqualTo("RS256");
+ assertThat(jwt.getKeyId()).isEqualTo(KEY_ID);
+ assertThat(jwt.getId()).isNotBlank();
+ assertThat(jwt.getExpiresAt()).isAfter(jwt.getIssuedAt());
+ }
+
+ @Test
+ void emptyKeyIdOmitsKidHeader() throws IOException {
+ DecodedJWT jwt = JWT.decode(exchanger("").mintAttributionJwt("tenant1/p", "tenant1"));
+ assertThat(jwt.getKeyId()).isNull();
+ }
+
+ @Test
+ void federatedCredentialsConfiguredForStsExchange() {
+ GoogleCredentials credentials = exchanger(KEY_ID).federatedCredentials("tenant1/p", "tenant1");
+ assertThat(credentials).isInstanceOf(IdentityPoolCredentials.class);
+ IdentityPoolCredentials idp = (IdentityPoolCredentials) credentials;
+ assertThat(idp.getAudience()).isEqualTo(AUDIENCE);
+ assertThat(idp.getSubjectTokenType())
+ .isEqualTo(GcpFederatedCredentialsExchanger.SUBJECT_TOKEN_TYPE);
+ }
+
+ @Test
+ void missingKeyFileFails() {
+ GcpFederatedCredentialsExchanger exchanger =
+ new GcpFederatedCredentialsExchanger(
+ ISSUER, AUDIENCE, tempDir.resolve("nope.pem"), KEY_ID, TRANSPORT);
+ assertThatThrownBy(() -> exchanger.mintAttributionJwt("tenant1/p", "tenant1"))
+ .isInstanceOf(IOException.class);
+ }
+
+ @Test
+ void signingKeyIsCachedAcrossInstances() throws IOException {
+ // First use parses and caches the key for this (unique) path.
+ exchanger(KEY_ID).mintAttributionJwt("tenant1/p", "tenant1");
+ // After the file is gone, a second instance on the same path still mints from the JVM cache,
+ // proving the parse is amortized rather than re-read per vend.
+ Files.delete(keyFile);
+ assertThatNoException()
+ .isThrownBy(() -> exchanger(KEY_ID).mintAttributionJwt("tenant1/p", "tenant1"));
+ }
+
+ @Test
+ void readPkcs8PrivateKey() throws IOException {
+ assertThat(GcpFederatedCredentialsExchanger.readPkcs8PrivateKey(keyFile)).isNotNull();
+ }
+
+ @Test
+ void readPkcs8PrivateKeyRejectsPkcs1Format() throws Exception {
+ KeyPairGenerator generator = KeyPairGenerator.getInstance("RSA");
+ generator.initialize(2048);
+ KeyPair keyPair = generator.generateKeyPair();
+ String pkcs1Pem =
+ "-----BEGIN RSA PRIVATE KEY-----\n"
+ + Base64.getMimeEncoder(64, "\n".getBytes(StandardCharsets.UTF_8))
+ .encodeToString(keyPair.getPrivate().getEncoded())
+ + "\n-----END RSA PRIVATE KEY-----\n";
+ Path pkcs1File = tempDir.resolve("pkcs1-key.pem");
+ Files.writeString(pkcs1File, pkcs1Pem);
+
+ assertThatThrownBy(() -> GcpFederatedCredentialsExchanger.readPkcs8PrivateKey(pkcs1File))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("PKCS#1")
+ .hasMessageContaining("openssl pkcs8");
+ }
+
+ @Test
+ void readPkcs8PrivateKeyRejectsInvalidBase64() throws Exception {
+ Path badFile = tempDir.resolve("bad-key.pem");
+ Files.writeString(
+ badFile, "-----BEGIN PRIVATE KEY-----\nnot-valid-base64!!!\n-----END PRIVATE KEY-----\n");
+
+ assertThatThrownBy(() -> GcpFederatedCredentialsExchanger.readPkcs8PrivateKey(badFile))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("PKCS#8 RSA private key");
+ }
+}
diff --git a/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpStorageCredentialCacheKeyTest.java b/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpStorageCredentialCacheKeyTest.java
new file mode 100644
index 0000000000..d47ef0e013
--- /dev/null
+++ b/polaris-core/src/test/java/org/apache/polaris/core/storage/gcp/GcpStorageCredentialCacheKeyTest.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.core.storage.gcp;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.google.api.client.http.javanet.NetHttpTransport;
+import com.google.auth.http.HttpTransportFactory;
+import com.google.auth.oauth2.GoogleCredentials;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.polaris.core.config.RealmConfig;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+class GcpStorageCredentialCacheKeyTest {
+
+ private static final GcpStorageConfigurationInfo CONFIG =
+ GcpStorageConfigurationInfo.builder()
+ .addAllAllowedLocations(java.util.List.of("gs://bucket/path"))
+ .build();
+ private static final GoogleCredentials CREDS = Mockito.mock(GoogleCredentials.class);
+ private static final HttpTransportFactory TRANSPORT = NetHttpTransport::new;
+ private static final RealmConfig REALM_CONFIG = Mockito.mock(RealmConfig.class);
+
+ private static GcpStorageCredentialCacheKey key(Optional<String> principalName) {
+ return GcpStorageCredentialCacheKey.of(
+ "tenant1",
+ CONFIG,
+ Set.of("gs://bucket/path"),
+ Set.of(),
+ Set.of(),
+ Optional.empty(),
+ principalName,
+ CREDS,
+ TRANSPORT,
+ REALM_CONFIG,
+ GcpCredentialOps.DEFAULT,
+ Optional.empty());
+ }
+
+ @Test
+ void principalNameIsPartOfCacheIdentity() {
+ // When attribution is on, the vended token is per-principal: two principals must not collide
+ // on one cache entry.
+ assertThat(key(Optional.of("alice"))).isNotEqualTo(key(Optional.of("bob")));
+ assertThat(key(Optional.of("alice"))).hasSameHashCodeAs(key(Optional.of("alice")));
+ assertThat(key(Optional.of("alice"))).isEqualTo(key(Optional.of("alice")));
+ }
+
+ @Test
+ void emptyPrincipalSharesOneEntry() {
+ // When attribution is off the principal is absent, preserving cross-principal cache reuse.
+ assertThat(key(Optional.empty())).isEqualTo(key(Optional.empty()));
+ }
+}
diff --git a/runtime/admin/distribution/LICENSE b/runtime/admin/distribution/LICENSE
index 9a0ec2c5a2..b91c547c85 100644
--- a/runtime/admin/distribution/LICENSE
+++ b/runtime/admin/distribution/LICENSE
@@ -399,6 +399,36 @@ License: Apache License 2.0 - https://github.com/hyperxpro/Brotli4j/blob/v1.16.0
--------------------------------------------------------------------------------
+This product bundles Auth0 Java JWT.
+
+* Maven group:artifact IDs: com.auth0:java-jwt
+
+Project URL: https://github.com/auth0/java-jwt
+License: MIT License
+| The MIT License (MIT)
+|
+| Copyright (c) 2015 Auth0, Inc. <su...@auth0.com> (http://auth0.com)
+|
+| Permission is hereby granted, free of charge, to any person obtaining a copy
+| of this software and associated documentation files (the "Software"), to deal
+| in the Software without restriction, including without limitation the rights
+| to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+| copies of the Software, and to permit persons to whom the Software is
+| furnished to do so, subject to the following conditions:
+|
+| The above copyright notice and this permission notice shall be included in all
+| copies or substantial portions of the Software.
+|
+| THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+| IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+| FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+| AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+| LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+| OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+| SOFTWARE.
+
+--------------------------------------------------------------------------------
+
This product bundles Azure SDK for Java.
* Maven group:artifact IDs: com.azure:azure-core
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/config/ProductionReadinessChecks.java b/runtime/service/src/main/java/org/apache/polaris/service/config/ProductionReadinessChecks.java
index e959489683..694f6bda06 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/config/ProductionReadinessChecks.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/config/ProductionReadinessChecks.java
@@ -356,6 +356,97 @@ public class ProductionReadinessChecks {
: ProductionReadinessCheck.of(errors.toArray(new Error[0]));
}
+ @Produces
+ public ProductionReadinessCheck checkGcsPrincipalAttribution(
+ FeaturesConfiguration featureConfiguration) {
+ var enabled = FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_ENABLED;
+ var audience = FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE;
+ var issuer = FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER;
+ var keyFile = FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE;
+
+ var defaults = featureConfiguration.defaults();
+ var errors = new ArrayList<Error>();
+
+ boolean defaultEnabled = Boolean.parseBoolean(defaults.get(enabled.key()));
+ String defaultAudience = defaults.getOrDefault(audience.key(), "");
+ String defaultIssuer = defaults.getOrDefault(issuer.key(), "");
+ String defaultKeyFile = defaults.getOrDefault(keyFile.key(), "");
+
+ if (defaultEnabled) {
+ if (defaultAudience.isEmpty()) {
+ errors.add(
+ Error.of(
+ "GCS_PRINCIPAL_ATTRIBUTION_ENABLED is true but GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE is not set.",
+ format("polaris.features.\"%s\"", audience.key())));
+ }
+ if (defaultIssuer.isEmpty()) {
+ errors.add(
+ Error.of(
+ "GCS_PRINCIPAL_ATTRIBUTION_ENABLED is true but GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER is not set.",
+ format("polaris.features.\"%s\"", issuer.key())));
+ }
+ if (defaultKeyFile.isEmpty()) {
+ errors.add(
+ Error.of(
+ "GCS_PRINCIPAL_ATTRIBUTION_ENABLED is true but GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE is not set.",
+ format("polaris.features.\"%s\"", keyFile.key())));
+ }
+ }
+
+ featureConfiguration
+ .realmOverrides()
+ .forEach(
+ (realmId, realmOverrides) -> {
+ var overrides = realmOverrides.overrides();
+ boolean realmEnabled =
+ Boolean.parseBoolean(
+ overrides.containsKey(enabled.key())
+ ? overrides.get(enabled.key())
+ : defaults.get(enabled.key()));
+ if (!realmEnabled) {
+ return;
+ }
+ String effectiveAudience =
+ overrides.containsKey(audience.key())
+ ? overrides.get(audience.key())
+ : defaultAudience;
+ String effectiveIssuer =
+ overrides.containsKey(issuer.key()) ? overrides.get(issuer.key()) : defaultIssuer;
+ String effectiveKeyFile =
+ overrides.containsKey(keyFile.key())
+ ? overrides.get(keyFile.key())
+ : defaultKeyFile;
+ if (effectiveAudience.isEmpty()) {
+ errors.add(
+ Error.of(
+ "GCS_PRINCIPAL_ATTRIBUTION_ENABLED is true but GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE is not set.",
+ format(
+ "polaris.features.realm-overrides.\"%s\".overrides.\"%s\"",
+ realmId, audience.key())));
+ }
+ if (effectiveIssuer.isEmpty()) {
+ errors.add(
+ Error.of(
+ "GCS_PRINCIPAL_ATTRIBUTION_ENABLED is true but GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER is not set.",
+ format(
+ "polaris.features.realm-overrides.\"%s\".overrides.\"%s\"",
+ realmId, issuer.key())));
+ }
+ if (effectiveKeyFile.isEmpty()) {
+ errors.add(
+ Error.of(
+ "GCS_PRINCIPAL_ATTRIBUTION_ENABLED is true but GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE is not set.",
+ format(
+ "polaris.features.realm-overrides.\"%s\".overrides.\"%s\"",
+ realmId, keyFile.key())));
+ }
+ });
+
+ return errors.isEmpty()
+ ? ProductionReadinessCheck.OK
+ : ProductionReadinessCheck.of(errors.toArray(new Error[0]));
+ }
+
@Produces
@SuppressWarnings("unchecked")
public ProductionReadinessCheck checkConnectionCredentialVendors(
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/config/ProductionReadinessChecksGcsAttributionTest.java b/runtime/service/src/test/java/org/apache/polaris/service/config/ProductionReadinessChecksGcsAttributionTest.java
new file mode 100644
index 0000000000..0cc42d1049
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/config/ProductionReadinessChecksGcsAttributionTest.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.config;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.config.ProductionReadinessCheck;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class ProductionReadinessChecksGcsAttributionTest {
+
+ private static final String ENABLED_KEY =
+ FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_ENABLED.key();
+ private static final String AUDIENCE_KEY =
+ FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE.key();
+ private static final String ISSUER_KEY =
+ FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER.key();
+ private static final String KEY_FILE_KEY =
+ FeatureConfiguration.GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE.key();
+
+ private ProductionReadinessChecks checks;
+
+ @BeforeEach
+ void setUp() {
+ checks = new ProductionReadinessChecks();
+ }
+
+ @Test
+ void attributionDisabledReturnsOk() {
+ FeaturesConfiguration config = mockConfig(Map.of(ENABLED_KEY, "false"), Map.of());
+
+ ProductionReadinessCheck result = checks.checkGcsPrincipalAttribution(config);
+
+ assertThat(result.ready()).isTrue();
+ }
+
+ @Test
+ void attributionEnabledWithAllFieldsSetReturnsOk() {
+ FeaturesConfiguration config =
+ mockConfig(
+ Map.of(
+ ENABLED_KEY, "true",
+ AUDIENCE_KEY, "//iam.googleapis.com/projects/123/...",
+ ISSUER_KEY, "https://polaris.example.com",
+ KEY_FILE_KEY, "/etc/polaris/signing-key.pem"),
+ Map.of());
+
+ ProductionReadinessCheck result = checks.checkGcsPrincipalAttribution(config);
+
+ assertThat(result.ready()).isTrue();
+ }
+
+ @Test
+ void attributionEnabledWithMissingAudienceReturnsError() {
+ FeaturesConfiguration config =
+ mockConfig(
+ Map.of(
+ ENABLED_KEY, "true",
+ ISSUER_KEY, "https://polaris.example.com",
+ KEY_FILE_KEY, "/etc/polaris/signing-key.pem"),
+ Map.of());
+
+ ProductionReadinessCheck result = checks.checkGcsPrincipalAttribution(config);
+
+ assertThat(result.ready()).isFalse();
+ assertThat(result.getErrors()).hasSize(1);
+ assertThat(result.getErrors().get(0).offendingProperty()).contains(AUDIENCE_KEY);
+ }
+
+ @Test
+ void attributionEnabledWithAllRequiredFieldsMissingReturnsThreeErrors() {
+ FeaturesConfiguration config = mockConfig(Map.of(ENABLED_KEY, "true"), Map.of());
+
+ ProductionReadinessCheck result = checks.checkGcsPrincipalAttribution(config);
+
+ assertThat(result.ready()).isFalse();
+ assertThat(result.getErrors()).hasSize(3);
+ }
+
+ @Test
+ void realmOverrideEnablesAttributionWithMissingFieldsReturnsError() {
+ FeaturesConfiguration config =
+ mockConfig(
+ Map.of(),
+ Map.of(
+ "test-realm",
+ mockRealmOverrides(
+ Map.of(
+ ENABLED_KEY, "true",
+ ISSUER_KEY, "https://polaris.example.com",
+ KEY_FILE_KEY, "/etc/polaris/signing-key.pem"))));
+
+ ProductionReadinessCheck result = checks.checkGcsPrincipalAttribution(config);
+
+ assertThat(result.ready()).isFalse();
+ assertThat(result.getErrors()).hasSize(1);
+ assertThat(result.getErrors().get(0).offendingProperty())
+ .contains("test-realm")
+ .contains(AUDIENCE_KEY);
+ }
+
+ @Test
+ void realmOverrideInheritsRequiredFieldsFromDefaultsReturnsOk() {
+ FeaturesConfiguration config =
+ mockConfig(
+ Map.of(
+ AUDIENCE_KEY, "//iam.googleapis.com/projects/123/...",
+ ISSUER_KEY, "https://polaris.example.com",
+ KEY_FILE_KEY, "/etc/polaris/signing-key.pem"),
+ Map.of("test-realm", mockRealmOverrides(Map.of(ENABLED_KEY, "true"))));
+
+ ProductionReadinessCheck result = checks.checkGcsPrincipalAttribution(config);
+
+ assertThat(result.ready()).isTrue();
+ }
+
+ @Test
+ void realmOverrideDisablesAttributionOverridesDefaultEnabledReturnsOk() {
+ FeaturesConfiguration config =
+ mockConfig(
+ Map.of(ENABLED_KEY, "true"),
+ Map.of("test-realm", mockRealmOverrides(Map.of(ENABLED_KEY, "false"))));
+
+ ProductionReadinessCheck result = checks.checkGcsPrincipalAttribution(config);
+
+ // default-level errors still reported, but realm override is clean
+ assertThat(result.getErrors()).noneMatch(e -> e.offendingProperty().contains("test-realm"));
+ }
+
+ private static FeaturesConfiguration mockConfig(
+ Map<String, String> defaults, Map<String, RealmOverridable.RealmOverrides> realmOverrides) {
+ FeaturesConfiguration config = mock(FeaturesConfiguration.class);
+ when(config.defaults()).thenReturn(defaults);
+ when(config.realmOverrides()).thenReturn(realmOverrides);
+ return config;
+ }
+
+ private static RealmOverridable.RealmOverrides mockRealmOverrides(Map<String, String> overrides) {
+ RealmOverridable.RealmOverrides realmOverrides = mock(RealmOverridable.RealmOverrides.class);
+ when(realmOverrides.overrides()).thenReturn(overrides);
+ return realmOverrides;
+ }
+}
diff --git a/site/content/in-dev/unreleased/configuration/config-sections/flags-polaris_features.md b/site/content/in-dev/unreleased/configuration/config-sections/flags-polaris_features.md
index 4080a68ee4..10ea19cf7f 100644
--- a/site/content/in-dev/unreleased/configuration/config-sections/flags-polaris_features.md
+++ b/site/content/in-dev/unreleased/configuration/config-sections/flags-polaris_features.md
@@ -335,6 +335,51 @@ The maximum weight for the entity cache. This is a heuristic value without any p
---
+##### `polaris.features."GCS_PRINCIPAL_ATTRIBUTION_ENABLED"`
+
+Enables GCS principal attribution via Workload Identity Federation. When true, credential vending chains a catalog-signed JWT through an STS token exchange and service-account impersonation so the Polaris principal appears in GCS Data Access audit logs (serviceAccountDelegationInfo.principalSubject). Requires GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE, GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER, and GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE to also be set; a missing required value is a fatal co [...]
+
+- **Type:** `Boolean`
+- **Default:** `false`
+
+---
+
+##### `polaris.features."GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE"`
+
+Filesystem path to the PKCS#8 PEM RSA private key used to sign GCS attribution JWTs (RS256). The corresponding public key must be published in the Workload Identity Pool provider's uploaded JWKS. Required when GCS_PRINCIPAL_ATTRIBUTION_ENABLED=true; ignored otherwise.
+
+- **Type:** `String`
+- **Default:** ``
+
+---
+
+##### `polaris.features."GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_ID"`
+
+Key ID (kid) written into the header of GCS attribution JWTs so the Workload Identity Pool provider can select the right public key from its JWKS during key rotation (when the JWKS holds both the old and new keys). Must match the kid of the JWKS entry for the configured signing key. Empty omits the header (only safe with a single-key JWKS).
+
+- **Type:** `String`
+- **Default:** ``
+
+---
+
+##### `polaris.features."GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER"`
+
+Issuer (iss claim) of catalog-minted GCS attribution JWTs; must match the issuer configured on the Workload Identity Pool OIDC provider. The provider verifies signatures against its uploaded JWKS, so no public discovery endpoint is required. Required when GCS_PRINCIPAL_ATTRIBUTION_ENABLED=true; ignored otherwise.
+
+- **Type:** `String`
+- **Default:** ``
+
+---
+
+##### `polaris.features."GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE"`
+
+Full resource name of the Workload Identity Pool provider used for GCS principal attribution, e.g. //iam.googleapis.com/projects/<num>/locations/global/workloadIdentityPools/<pool>/providers/<provider>. Used as both the attribution JWT 'aud' claim and the STS token-exchange audience. Required when GCS_PRINCIPAL_ATTRIBUTION_ENABLED=true; ignored otherwise.
+
+- **Type:** `String`
+- **Default:** ``
+
+---
+
##### `polaris.features."ICEBERG_COMMIT_MAX_RETRIES"`
The max number of times to try committing to an Iceberg table