Introduce reserved-properties setting; reserve "polaris." by default

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

@@ -84,6 +84,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
 import org.testcontainers.shaded.org.awaitility.Awaitility;
 
 /**
@@ -2102,6 +2103,129 @@ public void testDropNamespaceStatus() {
     }
   }
 
+  @Test
+  public void testCreateAndUpdateCatalogRoleWithReservedProperties() {
+    String catalogName = client.newEntityName("mycatalog1");
+    Catalog catalog =
+        PolarisCatalog.builder()
+            .setType(Catalog.TypeEnum.INTERNAL)
+            .setName(catalogName)
+            .setProperties(new CatalogProperties("s3://required/base/location"))
+            .setStorageConfigInfo(
+                new AwsStorageConfigInfo(
+                    "arn:aws:iam::012345678901:role/jdoe", StorageConfigInfo.StorageTypeEnum.S3))
+            .build();
+    managementApi.createCatalog(catalog);
+
+    CatalogRole badCatalogRole =
+        new CatalogRole("mycatalogrole", Map.of("polaris.reserved", "foo"), 0L, 0L, 1);
+    try (Response response =
+        managementApi
+            .request("v1/catalogs/{cat}/catalog-roles", Map.of("cat", catalogName))
+            .post(Entity.json(new CreateCatalogRoleRequest(badCatalogRole)))) {
+      assertThat(response)
+          .returns(Response.Status.BAD_REQUEST.getStatusCode(), Response::getStatus);
+    }
+
+    CatalogRole okCatalogRole = new CatalogRole("mycatalogrole", Map.of("foo", "bar"), 0L, 0L, 1);
+    try (Response response =
+        managementApi
+            .request("v1/catalogs/{cat}/catalog-roles", Map.of("cat", catalogName))
+            .post(Entity.json(new CreateCatalogRoleRequest(okCatalogRole)))) {
+      assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus);
+    }
+
+    UpdateCatalogRoleRequest updateRequest =
+        new UpdateCatalogRoleRequest(
+            okCatalogRole.getEntityVersion(), Map.of("polaris.reserved", "true"));
+    try (Response response =
+        managementApi
+            .request("v1/catalogs/{cat}/catalog-roles/mycatalogrole", Map.of("cat", catalogName))
+            .put(Entity.json(updateRequest))) {
+      assertThat(response)
+          .returns(Response.Status.BAD_REQUEST.getStatusCode(), Response::getStatus);
+    }
+  }
+
+  @Test
+  public void testCreateAndUpdatePrincipalRoleWithReservedProperties() {
+    String principal = "testCreateAndUpdatePrincipalRoleWithReservedProperties";
+    managementApi.createPrincipal(principal);
+
+    PrincipalRole badPrincipalRole =
+        new PrincipalRole(
+            client.newEntityName("myprincipalrole"), Map.of("polaris.reserved", "foo"), 0L, 0L, 1);
+    try (Response response =
+        managementApi
+            .request("v1/principal-roles")
+            .post(Entity.json(new CreatePrincipalRoleRequest(badPrincipalRole)))) {
+      assertThat(response)
+          .returns(Response.Status.BAD_REQUEST.getStatusCode(), Response::getStatus);
+    }
+
+    PrincipalRole goodPrincipalRole =
+        new PrincipalRole(
+            client.newEntityName("myprincipalrole"), Map.of("not.reserved", "foo"), 0L, 0L, 1);
+    try (Response response =
+        managementApi
+            .request("v1/principal-roles")
+            .post(Entity.json(new CreatePrincipalRoleRequest(goodPrincipalRole)))) {
+      assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus);
+    }
+
+    UpdatePrincipalRoleRequest badUpdate =
+        new UpdatePrincipalRoleRequest(
+            goodPrincipalRole.getEntityVersion(), ImmutableMap.of("polaris.reserved", "true"));
+    try (Response response =
+        managementApi
+            .request("v1/principal-roles/{pr}", Map.of("pr", goodPrincipalRole.getName()))
+            .put(Entity.json(badUpdate))) {
+      assertThat(response)
+          .returns(Response.Status.BAD_REQUEST.getStatusCode(), Response::getStatus);
+    }
+
+    managementApi.deletePrincipalRole(goodPrincipalRole);
+    managementApi.deletePrincipal(principal);
+  }
+
+  @Test
+  public void testCreateAndUpdatePrincipalWithReservedProperties() {
+    String principal = "testCreateAndUpdatePrincipalWithReservedProperties";
+
+    Principal badPrincipal =
+        new Principal(
+            principal, "clientId", ImmutableMap.of("polaris.reserved", "true"), 0L, 0L, 1);
+    try (Response response =
+        managementApi
+            .request("v1/principals")
+            .post(Entity.json(new CreatePrincipalRequest(badPrincipal, false)))) {
+      assertThat(response)
+          .returns(Response.Status.BAD_REQUEST.getStatusCode(), Response::getStatus);
+    }
+
+    Principal goodPrincipal =
+        new Principal(principal, "clientId", ImmutableMap.of("not.reserved", "true"), 0L, 0L, 1);
+    try (Response response =
+        managementApi
+            .request("v1/principals")
+            .post(Entity.json(new CreatePrincipalRequest(goodPrincipal, false)))) {
+      assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus);
+    }
+
+    UpdatePrincipalRequest badUpdate =
+        new UpdatePrincipalRequest(
+            goodPrincipal.getEntityVersion(), ImmutableMap.of("polaris.reserved", "true"));
+    try (Response response =
+        managementApi
+            .request("v1/principals/{p}", Map.of("p", goodPrincipal.getName()))
+            .put(Entity.json(badUpdate))) {
+      assertThat(response)
+          .returns(Response.Status.BAD_REQUEST.getStatusCode(), Response::getStatus);
+    }
+
+    managementApi.deletePrincipal(principal);
+  }
+
   public static JWTCreator.Builder defaultJwt() {
     Instant now = Instant.now();
     return JWT.create()

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

@@ -19,13 +19,13 @@
 package org.apache.polaris.service.it.test;
 
 import static jakarta.ws.rs.core.Response.Status.NOT_FOUND;
-import static javax.ws.rs.core.Response.Status.BAD_REQUEST;
 import static org.apache.polaris.service.it.env.PolarisClient.polarisClient;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Sets;
 import jakarta.ws.rs.ProcessingException;
 import jakarta.ws.rs.client.Entity;
 import jakarta.ws.rs.client.Invocation;
@@ -97,6 +97,7 @@
 import org.apache.polaris.service.it.env.PolarisApiEndpoints;
 import org.apache.polaris.service.it.env.PolarisClient;
 import org.apache.polaris.service.it.ext.PolarisIntegrationTestExtension;
+import org.apache.polaris.service.types.CreateGenericTableRequest;
 import org.apache.polaris.service.types.GenericTable;
 import org.assertj.core.api.Assertions;
 import org.assertj.core.api.Assumptions;
@@ -1409,7 +1410,7 @@ public void testLoadTableWithSnapshots() {
       assertThatCode(() -> catalogApi.loadTable(currentCatalogName, tableIdentifier, "not-real"))
           .isInstanceOf(RESTException.class)
           .hasMessageContaining("Unrecognized snapshots")
-          .hasMessageContaining("code=" + BAD_REQUEST.getStatusCode());
+          .hasMessageContaining("code=" + Response.Status.BAD_REQUEST.getStatusCode());
     } finally {
       genericTableApi.purge(currentCatalogName, namespace);
     }
@@ -1450,4 +1451,109 @@ public void testLoadTableWithRefFiltering() {
       genericTableApi.purge(currentCatalogName, namespace);
     }
   }
+
+  @Test
+  public void testCreateGenericTableWithReservedProperty() {
+    Namespace namespace = Namespace.of("ns1");
+    restCatalog.createNamespace(namespace);
+    TableIdentifier tableIdentifier = TableIdentifier.of(namespace, "tbl1");
+
+    String ns = RESTUtil.encodeNamespace(tableIdentifier.namespace());
+    try (Response res =
+        genericTableApi
+            .request(
+                "polaris/v1/{cat}/namespaces/{ns}/generic-tables/",
+                Map.of("cat", currentCatalogName, "ns", ns))
+            .post(
+                Entity.json(
+                    new CreateGenericTableRequest(
+                        tableIdentifier.name(),
+                        "format",
+                        "doc",
+                        Map.of("polaris.reserved", "true"))))) {
+      Assertions.assertThat(res.getStatus()).isEqualTo(Response.Status.BAD_REQUEST.getStatusCode());
+      Assertions.assertThat(res.readEntity(String.class)).contains("reserved prefix");
+    }
+
+    genericTableApi.purge(currentCatalogName, namespace);
+  }
+
+  @Test
+  public void testCreateNamespaceWithReservedProperty() {
+    Namespace namespace = Namespace.of("ns1");
+    assertThatCode(
+            () -> {
+              restCatalog.createNamespace(namespace, ImmutableMap.of("polaris.reserved", "true"));
+            })
+        .isInstanceOf(org.apache.iceberg.exceptions.BadRequestException.class)
+        .hasMessageContaining("reserved prefix");
+  }
+
+  @Test
+  public void testUpdateNamespaceWithReservedProperty() {
+    Namespace namespace = Namespace.of("ns1");
+    restCatalog.createNamespace(namespace, ImmutableMap.of("a", "b"));
+    restCatalog.setProperties(namespace, ImmutableMap.of("c", "d"));
+    Assertions.assertThatCode(
+            () -> {
+              restCatalog.setProperties(namespace, ImmutableMap.of("polaris.reserved", "true"));
+            })
+        .isInstanceOf(org.apache.iceberg.exceptions.BadRequestException.class)
+        .hasMessageContaining("reserved prefix");
+    genericTableApi.purge(currentCatalogName, namespace);
+  }
+
+  @Test
+  public void testRemoveReservedPropertyFromNamespace() {
+    Namespace namespace = Namespace.of("ns1");
+    restCatalog.createNamespace(namespace, ImmutableMap.of("a", "b"));
+    restCatalog.removeProperties(namespace, Sets.newHashSet("a"));
+    Assertions.assertThatCode(
+            () -> {
+              restCatalog.removeProperties(namespace, Sets.newHashSet("polaris.reserved"));
+            })
+        .isInstanceOf(org.apache.iceberg.exceptions.BadRequestException.class)
+        .hasMessageContaining("reserved prefix");
+    genericTableApi.purge(currentCatalogName, namespace);
+  }
+
+  @Test
+  public void testCreateTableWithReservedProperty() {
+    Namespace namespace = Namespace.of("ns1");
+    restCatalog.createNamespace(namespace);
+    TableIdentifier identifier = TableIdentifier.of(namespace, "t1");
+    Assertions.assertThatCode(
+            () -> {
+              restCatalog.createTable(
+                  identifier,
+                  SCHEMA,
+                  PartitionSpec.unpartitioned(),
+                  ImmutableMap.of("polaris.reserved", ""));
+            })
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("reserved prefix");
+    genericTableApi.purge(currentCatalogName, namespace);
+  }
+
+  @Test
+  public void testUpdateTableWithReservedProperty() {
+    Namespace namespace = Namespace.of("ns1");
+    restCatalog.createNamespace(namespace);
+    TableIdentifier identifier = TableIdentifier.of(namespace, "t1");
+    restCatalog.createTable(identifier, SCHEMA);
+    Assertions.assertThatCode(
+            () -> {
+              var txn =
+                  restCatalog.newReplaceTableTransaction(
+                      identifier,
+                      SCHEMA,
+                      PartitionSpec.unpartitioned(),
+                      ImmutableMap.of("polaris.reserved", ""),
+                      false);
+              txn.commitTransaction();
+            })
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("reserved prefix");
+    genericTableApi.purge(currentCatalogName, namespace);
+  }
 }

quarkus/service/src/main/java/org/apache/polaris/service/quarkus/config/QuarkusReservedProperties.java

@@ -0,0 +1,31 @@
+/*
+ * 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.quarkus.config;
+
+import io.smallrye.config.ConfigMapping;
+import java.util.List;
+import org.apache.polaris.service.config.ReservedProperties;
+
+@ConfigMapping(prefix = "polaris.reserved-properties")
+public interface QuarkusReservedProperties extends ReservedProperties {
+  @Override
+  default List<String> prefixes() {
+    return List.of("polaris.");
+  }
+}

quarkus/service/src/test/java/org/apache/polaris/service/quarkus/admin/PolarisAdminServiceAuthzTest.java

@@ -55,7 +55,8 @@ private PolarisAdminService newTestAdminService(Set<String> activatedPrincipalRo
         metaStoreManager,
         userSecretsManager,
         securityContext(authenticatedPrincipal, activatedPrincipalRoles),
-        polarisAuthorizer);
+        polarisAuthorizer,
+        reservedProperties);
   }
 
   private void doTestSufficientPrivileges(

quarkus/service/src/test/java/org/apache/polaris/service/quarkus/admin/PolarisAuthzTestBase.java

@@ -86,6 +86,7 @@
 import org.apache.polaris.service.catalog.policy.PolicyCatalog;
 import org.apache.polaris.service.config.DefaultConfigurationStore;
 import org.apache.polaris.service.config.RealmEntityManagerFactory;
+import org.apache.polaris.service.config.ReservedProperties;
 import org.apache.polaris.service.context.CallContextCatalogFactory;
 import org.apache.polaris.service.context.PolarisCallContextCatalogFactory;
 import org.apache.polaris.service.events.PolarisEventListener;
@@ -181,6 +182,7 @@ public Map<String, String> getConfigOverrides() {
               Map.of(
                   FeatureConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING.key,
                   true)));
+  protected final ReservedProperties reservedProperties = ReservedProperties.NONE;
 
   @Inject protected MetaStoreManagerFactory managerFactory;
   @Inject protected RealmEntityManagerFactory realmEntityManagerFactory;
@@ -264,7 +266,8 @@ public void before(TestInfo testInfo) {
             metaStoreManager,
             userSecretsManager,
             securityContext(authenticatedRoot, Set.of()),
-            polarisAuthorizer);
+            polarisAuthorizer,
+            reservedProperties);
 
     String storageLocation = "file:///tmp/authz";
     FileStorageConfigInfo storageConfigModel =

quarkus/service/src/test/java/org/apache/polaris/service/quarkus/catalog/GenericTableCatalogTest.java

@@ -81,6 +81,7 @@
 import org.apache.polaris.service.catalog.io.DefaultFileIOFactory;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
 import org.apache.polaris.service.config.RealmEntityManagerFactory;
+import org.apache.polaris.service.config.ReservedProperties;
 import org.apache.polaris.service.events.NoOpPolarisEventListener;
 import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl;
 import org.apache.polaris.service.task.TaskExecutor;
@@ -141,6 +142,7 @@ public Map<String, String> getConfigOverrides() {
   private AuthenticatedPolarisPrincipal authenticatedRoot;
   private PolarisEntity catalogEntity;
   private SecurityContext securityContext;
+  private ReservedProperties reservedProperties;
 
   protected static final Schema SCHEMA =
       new Schema(
@@ -195,14 +197,18 @@ public void before(TestInfo testInfo) {
     securityContext = Mockito.mock(SecurityContext.class);
     when(securityContext.getUserPrincipal()).thenReturn(authenticatedRoot);
     when(securityContext.isUserInRole(isA(String.class))).thenReturn(true);
+
+    reservedProperties = ReservedProperties.NONE;
+
     adminService =
         new PolarisAdminService(
             callContext,
             entityManager,
             metaStoreManager,
             userSecretsManager,
             securityContext,
-            new PolarisAuthorizerImpl(new PolarisConfigurationStore() {}));
+            new PolarisAuthorizerImpl(new PolarisConfigurationStore() {}),
+            reservedProperties);
 
     String storageLocation = "s3://my-bucket/path/to/data";
     storageConfigModel =