Service: Validate allowed locations from Catalog Storage Config

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

@@ -20,7 +20,6 @@
 
 import static org.apache.polaris.service.it.env.PolarisClient.polarisClient;
 
-import com.google.common.collect.ImmutableMap;
 import java.lang.reflect.Method;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -88,6 +87,7 @@ public abstract class PolarisRestCatalogViewIntegrationBase extends ViewCatalogT
   private static ManagementApi managementApi;
 
   private RESTCatalog restCatalog;
+  private StorageConfigInfo storageConfig;
 
   @BeforeAll
   static void setup(PolarisApiEndpoints apiEndpoints, ClientCredentials credentials) {
@@ -114,7 +114,7 @@ public void before(TestInfo testInfo) {
     Method method = testInfo.getTestMethod().orElseThrow();
     String catalogName = client.newEntityName(method.getName());
 
-    StorageConfigInfo storageConfig = getStorageConfigInfo();
+    storageConfig = getStorageConfigInfo();
     String defaultBaseLocation =
         storageConfig.getAllowedLocations().getFirst()
             + "/"
@@ -201,16 +201,10 @@ public void createViewWithCustomMetadataLocationUsingPolaris(@TempDir Path tempD
     TableIdentifier identifier = TableIdentifier.of("ns", "view");
 
     String location = Paths.get(tempDir.toUri().toString()).toString();
-    String customLocation = Paths.get(tempDir.toUri().toString(), "custom-location").toString();
-    String customLocation2 = Paths.get(tempDir.toUri().toString(), "custom-location2").toString();
-    String customLocationChild =
-        Paths.get(tempDir.toUri().toString(), "custom-location/child").toString();
+    String customLocation =
+        Paths.get(storageConfig.getAllowedLocations().getFirst(), "/custom-location1").toString();
 
-    catalog()
-        .createNamespace(
-            identifier.namespace(),
-            ImmutableMap.of(
-                IcebergTableLikeEntity.USER_SPECIFIED_WRITE_METADATA_LOCATION_KEY, location));
+    catalog().createNamespace(identifier.namespace());
 
     Assertions.assertThat(catalog().viewExists(identifier)).as("View should not exist").isFalse();
 
@@ -234,35 +228,5 @@ public void createViewWithCustomMetadataLocationUsingPolaris(@TempDir Path tempD
     Assertions.assertThat(((BaseView) view).operations().current().metadataFileLocation())
         .isNotNull()
         .startsWith(customLocation);
-
-    // CANNOT update the view with a new metadata location `baseLocation/customLocation2`,
-    // even though the new location is still under the parent namespace's
-    // `write.metadata.path=baseLocation`.
-    Assertions.assertThatThrownBy(
-            () ->
-                catalog()
-                    .loadView(identifier)
-                    .updateProperties()
-                    .set(
-                        IcebergTableLikeEntity.USER_SPECIFIED_WRITE_METADATA_LOCATION_KEY,
-                        customLocation2)
-                    .commit())
-        .isInstanceOf(ForbiddenException.class)
-        .hasMessageContaining("Forbidden: Invalid locations");
-
-    // CANNOT update the view with a child metadata location `baseLocation/customLocation/child`,
-    // even though it is a subpath of the original view's
-    // `write.metadata.path=baseLocation/customLocation`.
-    Assertions.assertThatThrownBy(
-            () ->
-                catalog()
-                    .loadView(identifier)
-                    .updateProperties()
-                    .set(
-                        IcebergTableLikeEntity.USER_SPECIFIED_WRITE_METADATA_LOCATION_KEY,
-                        customLocationChild)
-                    .commit())
-        .isInstanceOf(ForbiddenException.class)
-        .hasMessageContaining("Forbidden: Invalid locations");
   }
 }

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

@@ -53,16 +53,15 @@ protected InMemoryStorageIntegration(T config, String identifierOrId) {
    *     implementation, all actions have the same validation result, as we only verify the
    *     locations are equal to or subdirectories of the allowed locations.
    */
-  public static Map<String, Map<PolarisStorageActions, ValidationResult>>
-      validateSubpathsOfAllowedLocations(
-          @Nonnull RealmConfig realmConfig,
-          @Nonnull PolarisStorageConfigurationInfo storageConfig,
-          @Nonnull Set<PolarisStorageActions> actions,
-          @Nonnull Set<String> locations) {
+  public static Map<String, Map<PolarisStorageActions, ValidationResult>> validateAllowedLocations(
+      @Nonnull RealmConfig realmConfig,
+      @Nonnull List<String> allowedLocationsToValid,
+      @Nonnull Set<PolarisStorageActions> actions,
+      @Nonnull Set<String> locations) {
     // trim trailing / from allowed locations so that locations missing the trailing slash still
     // match
     Set<String> allowedLocationStrings =
-        storageConfig.getAllowedLocations().stream()
+        allowedLocationsToValid.stream()
             .map(
                 str -> {
                   if (str.endsWith("/") && str.length() > 1) {
@@ -123,6 +122,7 @@ public Map<String, Map<PolarisStorageActions, ValidationResult>> validateAccessT
       @Nonnull T storageConfig,
       @Nonnull Set<PolarisStorageActions> actions,
       @Nonnull Set<String> locations) {
-    return validateSubpathsOfAllowedLocations(realmConfig, storageConfig, actions, locations);
+    return validateAllowedLocations(
+        realmConfig, storageConfig.getAllowedLocations(), actions, locations);
   }
 }

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

@@ -0,0 +1,123 @@
+/*
+ * 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 jakarta.annotation.Nonnull;
+import java.util.List;
+import java.util.Set;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.ForbiddenException;
+import org.apache.polaris.core.config.RealmConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Defines storage location access restrictions for Polaris entities within a specific context. */
+public class LocationRestrictions {
+  private static final Logger LOGGER = LoggerFactory.getLogger(LocationRestrictions.class);
+
+  /**
+   * The complete set of storage locations that are permitted for access.
+   *
+   * <p>This list contains all storage URIs that entities can read from or write to, including both
+   * catalog-level allowed locations and any additional user-specified locations when unstructured
+   * table access is enabled.
+   *
+   * <p>All locations in this list have been validated to conform to the storage type's URI scheme
+   * requirements during construction.
+   */
+  private final List<String> allowedLocations;
+
+  /**
+   * The parent location for structured table enforcement.
+   *
+   * <p>When non-null, this location represents the root under which all new tables must be created,
+   * enforcing a structured hierarchy in addition to residing under {@code allowedLocations}. When
+   * null, table creation is allowed anywhere within the {@code allowedLocations}.
+   */
+  private final String parentLocation;
+
+  public LocationRestrictions(
+      @Nonnull PolarisStorageConfigurationInfo storageConfigurationInfo, String parentLocation) {
+    this.allowedLocations = storageConfigurationInfo.getAllowedLocations();
+    allowedLocations.forEach(storageConfigurationInfo::validatePrefixForStorageType);
+    this.parentLocation = parentLocation;
+  }
+
+  public LocationRestrictions(@Nonnull PolarisStorageConfigurationInfo storageConfigurationInfo) {
+    this(storageConfigurationInfo, null);
+  }
+
+  /**
+   * Validates that the requested storage locations are permitted for the given table identifier.
+   *
+   * <p>This method performs location validation by checking the requested locations against:
+   *
+   * <ul>
+   *   <li>The parent location (if configured) for structured table enforcement
+   *   <li>The allowed locations list for general access permissions
+   * </ul>
+   *
+   * <p>The validation ensures that all requested locations conform to the storage access
+   * restrictions defined for this context. If a parent location is configured, all requests must be
+   * under that location hierarchy in addition to being within the allowed locations.
+   *
+   * @param realmConfig the realm configuration containing storage validation rules
+   * @param identifier the table identifier for which locations are being validated
+   * @param requestLocations the set of storage locations that need validation
+   * @throws ForbiddenException if any of the requested locations violate the configured
+   *     restrictions
+   */
+  public void validate(
+      RealmConfig realmConfig, TableIdentifier identifier, Set<String> requestLocations) {
+    if (parentLocation != null) {
+      validateLocations(realmConfig, List.of(parentLocation), requestLocations, identifier);
+    }
+
+    validateLocations(realmConfig, allowedLocations, requestLocations, identifier);
+  }
+
+  private void validateLocations(
+      RealmConfig realmConfig,
+      List<String> allowedLocations,
+      Set<String> requestLocations,
+      TableIdentifier identifier) {
+    var validationResults =
+        InMemoryStorageIntegration.validateAllowedLocations(
+            realmConfig, allowedLocations, Set.of(PolarisStorageActions.ALL), requestLocations);
+    validationResults
+        .values()
+        .forEach(
+            actionResult ->
+                actionResult
+                    .values()
+                    .forEach(
+                        result -> {
+                          if (!result.isSuccess()) {
+                            throw new ForbiddenException(
+                                "Invalid locations '%s' for identifier '%s': %s",
+                                requestLocations, identifier, result.getMessage());
+                          } else {
+                            LOGGER.debug(
+                                "Validated locations '{}' for identifier '{}'",
+                                requestLocations,
+                                identifier);
+                          }
+                        }));
+  }
+}

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

@@ -26,14 +26,12 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.ImmutableList;
 import jakarta.annotation.Nonnull;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.Optional;
-import java.util.Set;
 import org.apache.polaris.core.admin.model.Catalog;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.config.RealmConfig;
@@ -116,7 +114,7 @@ public static PolarisStorageConfigurationInfo deserialize(final @Nonnull String
     }
   }
 
-  public static Optional<PolarisStorageConfigurationInfo> forEntityPath(
+  public static Optional<LocationRestrictions> forEntityPath(
       RealmConfig realmConfig, List<PolarisEntity> entityPath) {
     return findStorageInfoFromHierarchy(entityPath)
         .map(
@@ -150,22 +148,13 @@ public static Optional<PolarisStorageConfigurationInfo> forEntityPath(
                 LOGGER.debug(
                     "Not allowing unstructured table location for entity: {}",
                     entityPathReversed.get(0).getName());
-                return new StorageConfigurationOverride(configInfo, List.of(baseLocation));
+                return new LocationRestrictions(configInfo, baseLocation);
               } else {
                 LOGGER.debug(
                     "Allowing unstructured table location for entity: {}",
                     entityPathReversed.get(0).getName());
 
-                // TODO: figure out the purpose of adding `userSpecifiedWriteLocations`
-                Set<String> locations =
-                    StorageUtil.getLocationsAllowedToBeAccessed(
-                        null, entityPathReversed.get(0).getPropertiesAsMap());
-                return new StorageConfigurationOverride(
-                    configInfo,
-                    ImmutableList.<String>builder()
-                        .addAll(configInfo.getAllowedLocations())
-                        .addAll(locations)
-                        .build());
+                return new LocationRestrictions(configInfo);
               }
             });
   }

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

@@ -70,12 +70,12 @@ public class StorageUtil {
   }
 
   /** Given a TableMetadata, extracts the locations where the table's [meta]data might be found. */
-  public static @Nonnull Set<String> getLocationsAllowedToBeAccessed(TableMetadata tableMetadata) {
-    return getLocationsAllowedToBeAccessed(tableMetadata.location(), tableMetadata.properties());
+  public static @Nonnull Set<String> getLocationsUsedByTable(TableMetadata tableMetadata) {
+    return getLocationsUsedByTable(tableMetadata.location(), tableMetadata.properties());
   }
 
   /** Given a baseLocation and entity (table?) properties, extracts the relevant locations */
-  public static @Nonnull Set<String> getLocationsAllowedToBeAccessed(
+  public static @Nonnull Set<String> getLocationsUsedByTable(
       String baseLocation, Map<String, String> properties) {
     Set<String> locations = new HashSet<>();
     locations.add(baseLocation);
@@ -87,7 +87,7 @@ public class StorageUtil {
   }
 
   /** Given a ViewMetadata, extracts the locations where the view's [meta]data might be found. */
-  public static @Nonnull Set<String> getLocationsAllowedToBeAccessed(ViewMetadata viewMetadata) {
+  public static @Nonnull Set<String> getLocationsUsedByTable(ViewMetadata viewMetadata) {
     return Set.of(viewMetadata.location());
   }