feat: Add AWS STS Session Tags support for credential vending

persistence/nosql/persistence/metastore/src/main/java/org/apache/polaris/persistence/nosql/metastore/NoSqlMetaStoreManager.java

@@ -77,6 +77,7 @@
 import org.apache.polaris.core.persistence.pagination.PageToken;
 import org.apache.polaris.core.policy.PolicyEntity;
 import org.apache.polaris.core.policy.PolicyType;
+import org.apache.polaris.core.storage.CredentialVendingContext;
 import org.apache.polaris.persistence.nosql.metastore.privs.SecurableGranteePrivilegeTuple;
 
 record NoSqlMetaStoreManager(
@@ -776,7 +777,8 @@ public ScopedCredentialsResult getSubscopedCredsForEntity(
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
       @Nonnull PolarisPrincipal polarisPrincipal,
-      Optional<String> refreshCredentialsEndpoint) {
+      Optional<String> refreshCredentialsEndpoint,
+      @Nonnull CredentialVendingContext credentialVendingContext) {
 
     checkArgument(
         !allowedReadLocations.isEmpty() || !allowedWriteLocations.isEmpty(),
@@ -807,7 +809,8 @@ public ScopedCredentialsResult getSubscopedCredsForEntity(
               allowedReadLocations,
               allowedWriteLocations,
               polarisPrincipal,
-              refreshCredentialsEndpoint);
+              refreshCredentialsEndpoint,
+              credentialVendingContext);
       return new ScopedCredentialsResult(creds);
     } catch (Exception ex) {
       return new ScopedCredentialsResult(SUBSCOPE_CREDS_ERROR, ex.getMessage());

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -91,6 +91,19 @@ public static void enforceFeatureEnabledOrThrow(
           .defaultValue(false)
           .buildFeatureConfiguration();
 
+  public static final FeatureConfiguration<Boolean> INCLUDE_SESSION_TAGS_IN_SUBSCOPED_CREDENTIAL =
+      PolarisConfiguration.<Boolean>builder()
+          .key("INCLUDE_SESSION_TAGS_IN_SUBSCOPED_CREDENTIAL")
+          .description(
+              "If set to true, session tags (catalog, namespace, table, principal, roles) will be included\n"
+                  + "in AWS STS AssumeRole requests for credential vending. These tags appear in CloudTrail events,\n"
+                  + "enabling correlation between catalog operations and S3 data access.\n"
+                  + "Requires the IAM role trust policy to allow sts:TagSession action.\n"
+                  + "Note that enabling this feature may lead to degradation in temporary credential caching as \n"
+                  + "catalog will no longer be able to reuse credentials for different tables/namespaces/roles.")
+          .defaultValue(false)
+          .buildFeatureConfiguration();
+
   public static final FeatureConfiguration<Boolean> ALLOW_SETTING_S3_ENDPOINTS =
       PolarisConfiguration.<Boolean>builder()
           .key("ALLOW_SETTING_S3_ENDPOINTS")

polaris-core/src/main/java/org/apache/polaris/core/persistence/AtomicOperationMetaStoreManager.java

@@ -78,6 +78,7 @@
 import org.apache.polaris.core.policy.PolicyEntity;
 import org.apache.polaris.core.policy.PolicyMappingUtil;
 import org.apache.polaris.core.policy.PolicyType;
+import org.apache.polaris.core.storage.CredentialVendingContext;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageIntegration;
 import org.apache.polaris.core.storage.StorageAccessConfig;
@@ -1602,7 +1603,8 @@ public void deletePrincipalSecrets(
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
       @Nonnull PolarisPrincipal polarisPrincipal,
-      Optional<String> refreshCredentialsEndpoint) {
+      Optional<String> refreshCredentialsEndpoint,
+      @Nonnull CredentialVendingContext credentialVendingContext) {
 
     // get meta store session we should be using
     BasePersistence ms = callCtx.getMetaStore();
@@ -1644,7 +1646,8 @@ public void deletePrincipalSecrets(
               allowedReadLocations,
               allowedWriteLocations,
               polarisPrincipal,
-              refreshCredentialsEndpoint);
+              refreshCredentialsEndpoint,
+              credentialVendingContext);
       return new ScopedCredentialsResult(storageAccessConfig);
     } catch (Exception ex) {
       return new ScopedCredentialsResult(

polaris-core/src/main/java/org/apache/polaris/core/persistence/TransactionWorkspaceMetaStoreManager.java

@@ -61,6 +61,7 @@
 import org.apache.polaris.core.persistence.pagination.PageToken;
 import org.apache.polaris.core.policy.PolicyEntity;
 import org.apache.polaris.core.policy.PolicyType;
+import org.apache.polaris.core.storage.CredentialVendingContext;
 
 /**
  * Wraps an existing impl of PolarisMetaStoreManager and delegates expected "read" operations
@@ -326,7 +327,8 @@ public void deletePrincipalSecrets(
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
       @Nonnull PolarisPrincipal polarisPrincipal,
-      Optional<String> refreshCredentialsEndpoint) {
+      Optional<String> refreshCredentialsEndpoint,
+      @Nonnull CredentialVendingContext credentialVendingContext) {
     return delegate.getSubscopedCredsForEntity(
         callCtx,
         catalogId,
@@ -336,7 +338,8 @@ public void deletePrincipalSecrets(
         allowedReadLocations,
         allowedWriteLocations,
         polarisPrincipal,
-        refreshCredentialsEndpoint);
+        refreshCredentialsEndpoint,
+        credentialVendingContext);
   }
 
   @Override

polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TransactionalMetaStoreManagerImpl.java

@@ -83,6 +83,7 @@
 import org.apache.polaris.core.policy.PolicyEntity;
 import org.apache.polaris.core.policy.PolicyMappingUtil;
 import org.apache.polaris.core.policy.PolicyType;
+import org.apache.polaris.core.storage.CredentialVendingContext;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageIntegration;
 import org.apache.polaris.core.storage.StorageAccessConfig;
@@ -2096,7 +2097,8 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
       @Nonnull PolarisPrincipal polarisPrincipal,
-      Optional<String> refreshCredentialsEndpoint) {
+      Optional<String> refreshCredentialsEndpoint,
+      @Nonnull CredentialVendingContext credentialVendingContext) {
 
     // get meta store session we should be using
     TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
@@ -2133,7 +2135,8 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
               allowedReadLocations,
               allowedWriteLocations,
               polarisPrincipal,
-              refreshCredentialsEndpoint);
+              refreshCredentialsEndpoint,
+              credentialVendingContext);
       return new ScopedCredentialsResult(storageAccessConfig);
     } catch (Exception ex) {
       return new ScopedCredentialsResult(

polaris-core/src/main/java/org/apache/polaris/core/storage/CredentialVendingContext.java

@@ -0,0 +1,100 @@
+/*
+ * 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;
+
+import java.util.Optional;
+import org.apache.polaris.immutables.PolarisImmutable;
+
+/**
+ * Context information for credential vending operations. This context is used to provide metadata
+ * that can be attached to credentials as session tags (e.g., AWS STS session tags) for audit and
+ * correlation purposes in CloudTrail and similar logging systems.
+ *
+ * <p>When session tags are enabled, this context provides:
+ *
+ * <ul>
+ *   <li>{@code catalogName} - The name of the catalog vending credentials
+ *   <li>{@code namespace} - The namespace/database being accessed (e.g., "db.schema")
+ *   <li>{@code tableName} - The name of the table being accessed
+ *   <li>{@code activatedRoles} - Comma-separated list of activated principal roles
+ * </ul>
+ *
+ * <p>These values appear in cloud provider audit logs (e.g., AWS CloudTrail), enabling correlation
+ * between catalog operations and data access events.
+ */
+@PolarisImmutable
+public interface CredentialVendingContext {
+
+  // Default session tag keys for cloud provider credentials (e.g., AWS STS session tags).
+  // These appear in cloud audit logs (e.g., CloudTrail) for correlation purposes.
+  String TAG_KEY_CATALOG = "polaris:catalog";
+  String TAG_KEY_NAMESPACE = "polaris:namespace";
+  String TAG_KEY_TABLE = "polaris:table";
+  String TAG_KEY_PRINCIPAL = "polaris:principal";
+  String TAG_KEY_ROLES = "polaris:roles";
+
+  /** The name of the catalog that is vending credentials. */
+  Optional<String> catalogName();
+
+  /**
+   * The namespace being accessed, represented as a dot-separated string (e.g., "database.schema").
+   */
+  Optional<String> namespace();
+
+  /** The name of the table being accessed. */
+  Optional<String> tableName();
+
+  /**
+   * The activated roles for the principal, represented as a comma-separated sorted string. This is
+   * included in the context (rather than extracted from PolarisPrincipal) to ensure it is part of
+   * the cache key when session tags are enabled.
+   */
+  Optional<String> activatedRoles();
+
+  /**
+   * Creates a new builder for CredentialVendingContext.
+   *
+   * @return a new builder instance
+   */
+  static Builder builder() {
+    return ImmutableCredentialVendingContext.builder();
+  }
+
+  /**
+   * Creates an empty context with no metadata. This is useful when session tags are disabled or
+   * when context information is not available.
+   *
+   * @return an empty context instance
+   */
+  static CredentialVendingContext empty() {
+    return ImmutableCredentialVendingContext.builder().build();
+  }
+
+  interface Builder {
+    Builder catalogName(Optional<String> catalogName);
+
+    Builder namespace(Optional<String> namespace);
+
+    Builder tableName(Optional<String> tableName);
+
+    Builder activatedRoles(Optional<String> activatedRoles);
+
+    CredentialVendingContext build();
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/storage/PolarisCredentialVendor.java

@@ -35,15 +35,66 @@ public interface PolarisCredentialVendor {
    * @param callCtx the polaris call context
    * @param catalogId the catalog id
    * @param entityId the entity id
+   * @param entityType the type of entity
    * @param allowListOperation whether to allow LIST operation on the allowedReadLocations and
    *     allowedWriteLocations
    * @param allowedReadLocations a set of allowed to read locations
    * @param allowedWriteLocations a set of allowed to write locations
+   * @param polarisPrincipal the principal requesting credentials
    * @param refreshCredentialsEndpoint an optional endpoint to use for refreshing credentials. If
    *     supported by the storage type it will be returned to the client in the appropriate
    *     properties. The endpoint may be relative to the base URI and the client is responsible for
    *     handling the relative path
    * @return an enum map containing the scoped credentials
+   * @deprecated Use {@link #getSubscopedCredsForEntity(PolarisCallContext, long, long,
+   *     PolarisEntityType, boolean, Set, Set, PolarisPrincipal, Optional,
+   *     CredentialVendingContext)} instead. This method will be removed in a future release.
+   */
+  @Deprecated(forRemoval = true)
+  @Nonnull
+  default ScopedCredentialsResult getSubscopedCredsForEntity(
+      @Nonnull PolarisCallContext callCtx,
+      long catalogId,
+      long entityId,
+      @Nonnull PolarisEntityType entityType,
+      boolean allowListOperation,
+      @Nonnull Set<String> allowedReadLocations,
+      @Nonnull Set<String> allowedWriteLocations,
+      @Nonnull PolarisPrincipal polarisPrincipal,
+      Optional<String> refreshCredentialsEndpoint) {
+    return getSubscopedCredsForEntity(
+        callCtx,
+        catalogId,
+        entityId,
+        entityType,
+        allowListOperation,
+        allowedReadLocations,
+        allowedWriteLocations,
+        polarisPrincipal,
+        refreshCredentialsEndpoint,
+        CredentialVendingContext.empty());
+  }
+
+  /**
+   * Get a sub-scoped credentials for an entity against the provided allowed read and write
+   * locations, with credential vending context for session tags.
+   *
+   * @param callCtx the polaris call context
+   * @param catalogId the catalog id
+   * @param entityId the entity id
+   * @param entityType the type of entity
+   * @param allowListOperation whether to allow LIST operation on the allowedReadLocations and
+   *     allowedWriteLocations
+   * @param allowedReadLocations a set of allowed to read locations
+   * @param allowedWriteLocations a set of allowed to write locations
+   * @param polarisPrincipal the principal requesting credentials
+   * @param refreshCredentialsEndpoint an optional endpoint to use for refreshing credentials. If
+   *     supported by the storage type it will be returned to the client in the appropriate
+   *     properties. The endpoint may be relative to the base URI and the client is responsible for
+   *     handling the relative path
+   * @param credentialVendingContext context containing metadata for session tags (catalog,
+   *     namespace, table, roles) that can be attached to credentials for audit/correlation purposes
+   * @return an enum map containing the scoped credentials
    */
   @Nonnull
   ScopedCredentialsResult getSubscopedCredsForEntity(
@@ -55,5 +106,6 @@ ScopedCredentialsResult getSubscopedCredsForEntity(
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
       @Nonnull PolarisPrincipal polarisPrincipal,
-      Optional<String> refreshCredentialsEndpoint);
+      Optional<String> refreshCredentialsEndpoint,
+      @Nonnull CredentialVendingContext credentialVendingContext);
 }

polaris-core/src/main/java/org/apache/polaris/core/storage/PolarisStorageIntegration.java

@@ -57,10 +57,13 @@ public String getStorageIdentifierOrId() {
    *     locations
    * @param allowedReadLocations a set of allowed to read locations
    * @param allowedWriteLocations a set of allowed to write locations
+   * @param polarisPrincipal the principal requesting credentials
    * @param refreshCredentialsEndpoint an optional endpoint to use for refreshing credentials. If
    *     supported by the storage type it will be returned to the client in the appropriate
    *     properties. The endpoint may be relative to the base URI and the client is responsible for
    *     handling the relative path
+   * @param credentialVendingContext context containing metadata for session tags (catalog,
+   *     namespace, table, roles) that can be attached to credentials for audit/correlation purposes
    * @return An enum map including the scoped credentials
    */
   public abstract StorageAccessConfig getSubscopedCreds(
@@ -69,7 +72,8 @@ public abstract StorageAccessConfig getSubscopedCreds(
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
       @Nonnull PolarisPrincipal polarisPrincipal,
-      Optional<String> refreshCredentialsEndpoint);
+      Optional<String> refreshCredentialsEndpoint,
+      @Nonnull CredentialVendingContext credentialVendingContext);
 
   /**
    * Validate access for the provided operation actions and locations.