Fix native catalog credential vending skipping allowedLocations re-validation

CHANGELOG.md

@@ -100,6 +100,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 - The configuration option `polaris.event-listener.type` is deprecated and will be removed later. Please use `polaris.event-listener.types` instead.
 
 ### Fixes
+- Fixed native catalog credential vending paths (`loadCredentials` and `loadTable` with delegation) to re-validate locations against the *current* catalog `allowedLocations`. Previously these paths trusted persisted table entity locations, allowing stale credentials after an admin tightened allowed locations on a native catalog. (The federated path had a partial check.)
 
 ## [1.4.0]
 
@@ -153,7 +154,6 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 - `PolarisConfigurationStore` has been deprecated for removal.
 
 ### Fixes
-
 - Fixed error propagation in drop operations (`dropTable`, `dropView`, `dropNamespace`). Server errors now return appropriate HTTP status codes based on persistence result instead of always returning NotFound
 - Enable non-AWS STS role ARNs
 - Helm chart: fixed a bug that prevented CORS settings to be properly applied. A new setting `cors.enabled` has been introduced in the chart as part of the fix.
@@ -231,7 +231,6 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
   endpoints at `/q/metrics` and `/q/health` instead.
 
 ### Fixes
-
 - Fixed incorrect Azure expires at field for the credentials refresh response, leading to client failure via #2633
 
 ## [1.1.0-incubating]

integration-tests/src/main/java/org/apache/polaris/service/it/test/CatalogFederationIntegrationTest.java

@@ -483,7 +483,7 @@ void testFederatedCatalogNotVendCredentialForTablesOutsideAllowedLocations() {
     assertThatThrownBy(() -> spark.sql("SELECT * FROM ns3.test_table ORDER BY id").collectAsList())
         .isInstanceOf(ForbiddenException.class)
         .hasMessageContaining(
-            "Table 'ns3.test_table' in remote catalog has locations outside catalog's allowed locations:");
+            "Table 'ns3.test_table' has locations outside the catalog's current allowed locations:");
 
     // Case 3: TABLE_WRITE_DATA
     managementApi.revokeGrant(federatedCatalogName, federatedCatalogRoleName, tableReadDataGrant);
@@ -501,6 +501,6 @@ void testFederatedCatalogNotVendCredentialForTablesOutsideAllowedLocations() {
             () -> spark.sql("INSERT INTO ns3.test_table VALUES (3, 'Charlie')").collectAsList())
         .isInstanceOf(ForbiddenException.class)
         .hasMessageContaining(
-            "Table 'ns3.test_table' in remote catalog has locations outside catalog's allowed locations:");
+            "Table 'ns3.test_table' has locations outside the catalog's current allowed locations:");
   }
 }

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

@@ -1021,10 +1021,10 @@ private LoadTableResponse.Builder buildLoadTableResponseWithDelegationCredential
 
       Set<String> tableLocations = StorageUtil.getLocationsUsedByTable(tableMetadata);
 
-      // For federated catalogs, validate that table locations are within allowed locations
-      if (isFederated) {
-        validateRemoteTableLocations(tableIdentifier, tableLocations, resolvedStoragePath);
-      }
+      // Validate that the table's locations are still within the catalog's current
+      // allowedLocations before vending credentials. This protects against cases where
+      // allowedLocations were tightened after the table was created.
+      validateTableLocations(tableIdentifier, tableLocations, resolvedStoragePath);
 
       StorageAccessConfig storageAccessConfig =
           storageAccessConfigProvider()
@@ -1058,29 +1058,31 @@ private LoadTableResponse.Builder buildLoadTableResponseWithDelegationCredential
     return responseBuilder;
   }
 
-  private void validateRemoteTableLocations(
+  private void validateTableLocations(
       TableIdentifier tableIdentifier,
       Set<String> tableLocations,
       PolarisResolvedPathWrapper resolvedStoragePath) {
 
     try {
-      // Delegate to common validation logic
+      // Delegate to common validation logic. This is called for both native and federated
+      // catalogs before vending credentials to ensure locations are still within the
+      // current catalog's allowedLocations (defense against policy changes after table creation).
       CatalogUtils.validateLocationsForTableLike(
           realmConfig(), tableIdentifier, tableLocations, resolvedStoragePath);
 
       LOGGER
           .atInfo()
           .addKeyValue("tableIdentifier", tableIdentifier)
           .addKeyValue("tableLocations", tableLocations)
-          .log("Validated federated table locations");
+          .log("Validated table locations for credential vending");
     } catch (ForbiddenException e) {
       LOGGER
           .atError()
           .addKeyValue("tableIdentifier", tableIdentifier)
           .addKeyValue("tableLocations", tableLocations)
-          .log("Federated table locations validation failed");
+          .log("Table locations validation failed for credential vending");
       throw new ForbiddenException(
-          "Table '%s' in remote catalog has locations outside catalog's allowed locations: %s",
+          "Table '%s' has locations outside the catalog's current allowed locations: %s",
           tableIdentifier, e.getMessage());
     }
   }
@@ -1567,6 +1569,10 @@ private StorageAccessConfig vendCredentials(
       return null;
     }
 
+    // Re-validate before vending in case this is called from other paths in the future.
+    // Primary validation for loadCredentials and delegation happens at call sites.
+    validateTableLocations(tableIdentifier, tableLocations, resolvedStoragePath);
+
     return storageAccessConfigProvider()
         .getStorageAccessConfig(
             tableIdentifier,

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/IcebergAllowedLocationTest.java

@@ -34,6 +34,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.UUID;
 import org.apache.iceberg.MetadataUpdate;
 import org.apache.iceberg.catalog.Namespace;
@@ -51,6 +52,7 @@
 import org.apache.polaris.core.admin.model.CreateCatalogRequest;
 import org.apache.polaris.core.admin.model.FileStorageConfigInfo;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.admin.model.UpdateCatalogRequest;
 import org.apache.polaris.service.TestServices;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Test;
@@ -549,4 +551,88 @@ private void createNamespace(TestServices services, String location) {
       assertThat(response.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
     }
   }
+
+  private void updateCatalogAllowedLocations(
+      TestServices services, List<String> newAllowedLocations) {
+    // Fetch current catalog to get entity version for update.
+    Catalog fetched;
+    try (Response getResp =
+        services
+            .catalogsApi()
+            .getCatalog(catalog, services.realmContext(), services.securityContext())) {
+      assertThat(getResp.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+      fetched = getResp.readEntity(Catalog.class);
+    }
+
+    StorageConfigInfo newConfig =
+        FileStorageConfigInfo.builder()
+            .setStorageType(StorageConfigInfo.StorageTypeEnum.FILE)
+            .setAllowedLocations(newAllowedLocations)
+            .build();
+
+    UpdateCatalogRequest updateRequest =
+        UpdateCatalogRequest.builder()
+            .setCurrentEntityVersion(fetched.getEntityVersion())
+            .setStorageConfigInfo(newConfig)
+            .build();
+
+    try (Response updateResp =
+        services
+            .catalogsApi()
+            .updateCatalog(
+                catalog, updateRequest, services.realmContext(), services.securityContext())) {
+      assertThat(updateResp.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+    }
+  }
+
+  @Test
+  void testNativeCredentialVendingRejectsStaleLocationsAfterAllowedLocationsShrink(
+      @TempDir Path tmpDir) {
+    var services = getTestServices();
+
+    var baseDir = tmpDir.resolve(catalog).toAbsolutePath().toUri().toString();
+    var allowedWhenCreated = baseDir + "/original-data";
+    var disallowedAfterShrink = baseDir + "/other-data";
+
+    // Catalog initially allows the original location
+    createCatalog(services, Map.of(), baseDir, List.of(allowedWhenCreated));
+    createNamespace(services, allowedWhenCreated + "/" + namespace);
+
+    String tableName = getTableName();
+    TableIdentifier tableId = TableIdentifier.of(Namespace.of(namespace), tableName);
+
+    var createReq =
+        CreateTableRequest.builder()
+            .withName(tableName)
+            .withSchema(SCHEMA)
+            .withLocation(allowedWhenCreated + "/" + namespace + "/" + tableName)
+            .build();
+
+    // Create succeeds under current allowed
+    try (Response createResp =
+        services
+            .restApi()
+            .createTable(
+                catalog,
+                namespace,
+                createReq,
+                "vended-credentials",
+                IDEMPOTENCY_KEY,
+                services.realmContext(),
+                services.securityContext())) {
+      assertThat(createResp.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+    }
+
+    // Now shrink: update catalog to only allow a completely different location
+    updateCatalogAllowedLocations(services, List.of(disallowedAfterShrink));
+
+    // Credential vending for the existing table (which has stale location in its entity)
+    // must now be rejected.
+    IcebergCatalogHandler handler =
+        services.catalogAdapter().newHandler(services.securityContext(), catalog);
+
+    assertThatThrownBy(() -> handler.loadCredentials(tableId, Optional.empty()))
+        .isInstanceOf(ForbiddenException.class)
+        .hasMessageContaining("outside the catalog's current allowed locations");
+  }
 }