Fix /v1/config disclosing catalog properties without CATALOG_READ_PROPERTIES

CHANGELOG.md

@@ -94,6 +94,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 - Deprecated `ALLOW_EXTERNAL_TABLE_LOCATION`. Use `ALLOW_EXTERNAL_METADATA_FILE_LOCATION` for external metadata file locations, including catalog config `polaris.config.allow.external.metadata.file.location`.
 
 ### Fixes
+- The Iceberg REST `GET /v1/config` endpoint is now covered by authorization. Calling it requires the new `CATALOG_READ_CONFIG` privilege (operation `GET_CATALOG_CONFIG`), which is subsumed by the catalog-level read/manage privileges and by every catalog-content privilege, so principals that can use the catalog through the REST API keep working. Catalog properties in the response (`defaults`) additionally require the `CATALOG_READ_PROPERTIES` privilege (operation `GET_CATALOG_CONFIG_PROPERTIES`), matching the management-plane `getCatalog` rule; authorized callers without that privilege still receive the `prefix` override and supported-endpoints list so REST clients can initialize. Previously, any authenticated principal could read all client-visible catalog properties by passing a `warehouse` parameter.
 - The NoSQL persistence commit log (`Commits.commitLog`) no longer stops early when a commit's recent-ancestor tail is shorter than the internal fetch page size. With a `polaris.persistence.reference-previous-head-count` smaller than the page size, the natural-order commit log previously truncated at the first short tail because trailing null entries in the fetch page were treated as end-of-history, which could also drop still-referenced objects during maintenance.
 - Python CLI REPL now shows a clear "Syntax error" message for malformed input instead of a generic "unexpected error" message.
 - Python CLI `setup apply` no longer double-encodes policy content emitted by `setup export`, so

extensions/auth/ranger/src/main/java/org/apache/polaris/extension/auth/ranger/RangerPolarisOperationSemantics.java

@@ -280,6 +280,16 @@ enum ResolvedPathRooting {
         PolarisAuthorizableOperation.GET_CATALOG,
         new RangerPolarisOperationSemantics(
             toSet(CATALOG_READ_PROPERTIES), null, ResolvedPathRooting.ROOT));
+    // No dedicated Ranger access type for the catalog config endpoint yet; protect it with the
+    // existing catalog-properties-read access type.
+    RBAC_SEMANTICS_BY_OPERATION.put(
+        PolarisAuthorizableOperation.GET_CATALOG_CONFIG,
+        new RangerPolarisOperationSemantics(
+            toSet(CATALOG_READ_PROPERTIES), null, ResolvedPathRooting.ROOT));
+    RBAC_SEMANTICS_BY_OPERATION.put(
+        PolarisAuthorizableOperation.GET_CATALOG_CONFIG_PROPERTIES,
+        new RangerPolarisOperationSemantics(
+            toSet(CATALOG_READ_PROPERTIES), null, ResolvedPathRooting.ROOT));
     RBAC_SEMANTICS_BY_OPERATION.put(
         PolarisAuthorizableOperation.UPDATE_CATALOG,
         new RangerPolarisOperationSemantics(

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

@@ -101,6 +101,14 @@ public class CatalogFederationIntegrationTest {
           .setPrivilege(CatalogPrivilege.CATALOG_MANAGE_CONTENT)
           .build();
 
+  // Granted separately from defaultCatalogGrant: RBAC tests revoke defaultCatalogGrant, and the
+  // Spark client still needs GET /v1/config to bootstrap against the federated catalog.
+  private static final CatalogGrant catalogReadConfigGrant =
+      CatalogGrant.builder()
+          .setType(GrantResource.TypeEnum.CATALOG)
+          .setPrivilege(CatalogPrivilege.CATALOG_READ_CONFIG)
+          .build();
+
   @TempDir static java.nio.file.Path warehouseDir;
 
   private PrincipalWithCredentials newUserCredentials;
@@ -228,6 +236,7 @@ private void setupCatalogs() {
     managementApi.createCatalogRole(federatedCatalogName, federatedCatalogRoleName);
 
     managementApi.addGrant(federatedCatalogName, federatedCatalogRoleName, defaultCatalogGrant);
+    managementApi.addGrant(federatedCatalogName, federatedCatalogRoleName, catalogReadConfigGrant);
     CatalogRole externalCatalogAdminRole =
         managementApi.getCatalogRole(federatedCatalogName, federatedCatalogRoleName);
     managementApi.grantCatalogRoleToPrincipalRole(

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

@@ -113,6 +113,14 @@ public abstract class HiveCatalogFederationIntegrationTest {
           .setPrivilege(CatalogPrivilege.CATALOG_MANAGE_CONTENT)
           .build();
 
+  // Granted separately from CATALOG_ADMIN_GRANT: RBAC tests revoke CATALOG_ADMIN_GRANT, and the
+  // Spark client still needs GET /v1/config to bootstrap against the federated catalog.
+  private static final CatalogGrant CATALOG_READ_CONFIG_GRANT =
+      CatalogGrant.builder()
+          .setType(GrantResource.TypeEnum.CATALOG)
+          .setPrivilege(CatalogPrivilege.CATALOG_READ_CONFIG)
+          .build();
+
   /**
    * Returns the RustFS instance backing the test. The subclass is also responsible for propagating
    * the endpoint/credentials to the Polaris JVM as system properties so Hadoop's {@code
@@ -211,6 +219,8 @@ void before() {
     managementApi.createCatalog(externalCatalog);
     managementApi.createCatalogRole(federatedCatalogName, federatedCatalogRoleName);
     managementApi.addGrant(federatedCatalogName, federatedCatalogRoleName, CATALOG_ADMIN_GRANT);
+    managementApi.addGrant(
+        federatedCatalogName, federatedCatalogRoleName, CATALOG_READ_CONFIG_GRANT);
     CatalogRole role = managementApi.getCatalogRole(federatedCatalogName, federatedCatalogRoleName);
     managementApi.grantCatalogRoleToPrincipalRole(PRINCIPAL_ROLE_NAME, federatedCatalogName, role);
 

polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizableOperation.java

@@ -63,6 +63,8 @@ public enum PolarisAuthorizableOperation {
   GET_CATALOG,
   UPDATE_CATALOG,
   DELETE_CATALOG,
+  GET_CATALOG_CONFIG,
+  GET_CATALOG_CONFIG_PROPERTIES,
   LIST_PRINCIPALS,
   CREATE_PRINCIPAL,
   GET_PRINCIPAL,

polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizerImpl.java

@@ -30,6 +30,7 @@
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_MANAGE_CONTENT;
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_MANAGE_GRANTS_ON_SECURABLE;
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_MANAGE_METADATA;
+import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_READ_CONFIG;
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_READ_PROPERTIES;
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_ROLE_CREATE;
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_ROLE_DROP;
@@ -496,6 +497,42 @@ public class PolarisAuthorizerImpl implements PolarisAuthorizer {
             CATALOG_READ_PROPERTIES,
             CATALOG_WRITE_PROPERTIES,
             SERVICE_MANAGE_ACCESS));
+    // CATALOG_READ_CONFIG gates client bootstrap via the Iceberg REST /v1/config endpoint, so it
+    // is subsumed by every privilege that is exercised through that client: the catalog-level
+    // read/manage privileges and all catalog-content privileges. Note the check still evaluates
+    // grants on the catalog path, so content privileges must be granted at catalog level (or
+    // above) to subsume; grants scoped to a single namespace or table are not visible to the
+    // catalog-level check.
+    SUPER_PRIVILEGES.putAll(
+        CATALOG_READ_CONFIG,
+        List.of(
+            CATALOG_FULL_METADATA,
+            CATALOG_MANAGE_CONTENT,
+            CATALOG_MANAGE_METADATA,
+            CATALOG_READ_CONFIG,
+            CATALOG_READ_PROPERTIES,
+            CATALOG_WRITE_PROPERTIES,
+            NAMESPACE_CREATE,
+            NAMESPACE_DROP,
+            NAMESPACE_FULL_METADATA,
+            NAMESPACE_LIST,
+            NAMESPACE_READ_PROPERTIES,
+            NAMESPACE_WRITE_PROPERTIES,
+            TABLE_CREATE,
+            TABLE_DROP,
+            TABLE_FULL_METADATA,
+            TABLE_LIST,
+            TABLE_READ_DATA,
+            TABLE_READ_PROPERTIES,
+            TABLE_WRITE_DATA,
+            TABLE_WRITE_PROPERTIES,
+            VIEW_CREATE,
+            VIEW_DROP,
+            VIEW_FULL_METADATA,
+            VIEW_LIST,
+            VIEW_READ_PROPERTIES,
+            VIEW_WRITE_PROPERTIES,
+            SERVICE_MANAGE_ACCESS));
     SUPER_PRIVILEGES.putAll(
         CATALOG_WRITE_PROPERTIES,
         List.of(

polaris-core/src/main/java/org/apache/polaris/core/auth/RbacOperationSemantics.java

@@ -62,6 +62,8 @@
 import static org.apache.polaris.core.auth.PolarisAuthorizableOperation.GET_APPLICABLE_POLICIES_ON_NAMESPACE;
 import static org.apache.polaris.core.auth.PolarisAuthorizableOperation.GET_APPLICABLE_POLICIES_ON_TABLE;
 import static org.apache.polaris.core.auth.PolarisAuthorizableOperation.GET_CATALOG;
+import static org.apache.polaris.core.auth.PolarisAuthorizableOperation.GET_CATALOG_CONFIG;
+import static org.apache.polaris.core.auth.PolarisAuthorizableOperation.GET_CATALOG_CONFIG_PROPERTIES;
 import static org.apache.polaris.core.auth.PolarisAuthorizableOperation.GET_CATALOG_ROLE;
 import static org.apache.polaris.core.auth.PolarisAuthorizableOperation.GET_PRINCIPAL;
 import static org.apache.polaris.core.auth.PolarisAuthorizableOperation.GET_PRINCIPAL_ROLE;
@@ -136,6 +138,7 @@
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_DROP;
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_LIST;
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_MANAGE_GRANTS_ON_SECURABLE;
+import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_READ_CONFIG;
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_READ_PROPERTIES;
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_ROLE_CREATE;
 import static org.apache.polaris.core.entity.PolarisPrivilege.CATALOG_ROLE_DROP;
@@ -337,6 +340,8 @@ private static void register(
     register(GET_CATALOG, CATALOG_READ_PROPERTIES);
     register(UPDATE_CATALOG, CATALOG_WRITE_PROPERTIES);
     register(DELETE_CATALOG, CATALOG_DROP);
+    register(GET_CATALOG_CONFIG, CATALOG_READ_CONFIG);
+    register(GET_CATALOG_CONFIG_PROPERTIES, CATALOG_READ_PROPERTIES);
 
     // Principal operations
     register(LIST_PRINCIPALS, PRINCIPAL_LIST);

polaris-core/src/main/java/org/apache/polaris/core/entity/PolarisPrivilege.java

@@ -256,6 +256,7 @@ public enum PolarisPrivilege {
       PolarisEntityType.TABLE_LIKE,
       List.of(PolarisEntitySubType.ICEBERG_TABLE, PolarisEntitySubType.GENERIC_TABLE),
       PolarisEntityType.CATALOG_ROLE),
+  CATALOG_READ_CONFIG(103, PolarisEntityType.CATALOG),
   ;
 
   /**

polaris-core/src/test/java/org/apache/polaris/core/entity/PolarisPrivilegeTest.java

@@ -131,7 +131,8 @@ static Stream<Arguments> polarisPrivileges() {
         Arguments.of(100, PolarisPrivilege.TABLE_REMOVE_STATISTICS),
         Arguments.of(101, PolarisPrivilege.TABLE_REMOVE_PARTITION_SPECS),
         Arguments.of(102, PolarisPrivilege.TABLE_MANAGE_STRUCTURE),
-        Arguments.of(103, null));
+        Arguments.of(103, PolarisPrivilege.CATALOG_READ_CONFIG),
+        Arguments.of(104, null));
   }
 
   @ParameterizedTest

runtime/service/src/main/java/org/apache/polaris/service/catalog/config/CatalogConfigHandler.java

@@ -28,55 +28,96 @@
 import jakarta.inject.Inject;
 import java.util.Comparator;
 import java.util.LinkedHashSet;
-import java.util.Map;
+import java.util.List;
 import java.util.Set;
+import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.rest.Endpoint;
 import org.apache.iceberg.rest.RESTCatalogProperties;
 import org.apache.iceberg.rest.responses.ConfigResponse;
+import org.apache.polaris.core.auth.AuthorizationRequest;
+import org.apache.polaris.core.auth.AuthorizationState;
+import org.apache.polaris.core.auth.PolarisAuthorizableOperation;
+import org.apache.polaris.core.auth.PolarisAuthorizer;
 import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.auth.SingleTargetAuthorizationIntent;
+import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisEntity;
-import org.apache.polaris.core.persistence.ResolvedPolarisEntity;
-import org.apache.polaris.core.persistence.resolver.Resolver;
-import org.apache.polaris.core.persistence.resolver.ResolverFactory;
+import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
+import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
+import org.apache.polaris.core.persistence.resolver.ResolutionManifestFactory;
 import org.apache.polaris.core.persistence.resolver.ResolverStatus;
 import org.apache.polaris.core.rest.NamespaceUtils;
 import org.apache.polaris.service.catalog.CatalogPrefixParser;
+import org.apache.polaris.service.catalog.common.PolarisSecurableMapper;
 import org.apache.polaris.service.catalog.spi.CatalogConfigEndpointContributor;
 import org.apache.polaris.service.idempotency.IdempotencyConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @RequestScoped
 public class CatalogConfigHandler {
+  private static final Logger LOGGER = LoggerFactory.getLogger(CatalogConfigHandler.class);
   private final CatalogPrefixParser prefixParser;
-  private final ResolverFactory resolverFactory;
+  private final ResolutionManifestFactory resolutionManifestFactory;
+  private final PolarisAuthorizer authorizer;
   private final Instance<CatalogConfigEndpointContributor> endpointContributors;
   private final IdempotencyConfiguration idempotencyConfiguration;
 
   @Inject
   public CatalogConfigHandler(
       CatalogPrefixParser prefixParser,
-      ResolverFactory resolverFactory,
+      ResolutionManifestFactory resolutionManifestFactory,
+      PolarisAuthorizer authorizer,
       @Any Instance<CatalogConfigEndpointContributor> endpointContributors,
       IdempotencyConfiguration idempotencyConfiguration) {
     this.prefixParser = prefixParser;
-    this.resolverFactory = resolverFactory;
+    this.resolutionManifestFactory = resolutionManifestFactory;
+    this.authorizer = authorizer;
     this.endpointContributors = endpointContributors;
     this.idempotencyConfiguration = idempotencyConfiguration;
   }
 
   public ConfigResponse getConfig(String catalogName, PolarisPrincipal principal) {
-    Resolver resolver = resolverFactory.createResolver(principal, catalogName);
-    ResolverStatus resolverStatus = resolver.resolveAll();
+    PolarisResolutionManifest resolutionManifest =
+        resolutionManifestFactory.createResolutionManifest(principal, catalogName);
+    resolutionManifest.addTopLevelName(
+        catalogName, PolarisEntityType.CATALOG, false /* isOptional */);
+    AuthorizationRequest authzRequest =
+        new AuthorizationRequest(
+            principal,
+            List.of(
+                new SingleTargetAuthorizationIntent(
+                    PolarisAuthorizableOperation.GET_CATALOG_CONFIG,
+                    PolarisSecurableMapper.catalog(catalogName)),
+                new SingleTargetAuthorizationIntent(
+                    PolarisAuthorizableOperation.GET_CATALOG_CONFIG_PROPERTIES,
+                    PolarisSecurableMapper.catalog(catalogName))));
+    AuthorizationState authorizationState = new AuthorizationState(resolutionManifest);
+    authorizer.resolveAuthorizationInputs(authorizationState, authzRequest);
+
+    ResolverStatus resolverStatus = resolutionManifest.getPrimaryResolverStatusOrThrow();
     if (!resolverStatus.getStatus().equals(ResolverStatus.StatusEnum.SUCCESS)) {
       throw new NotFoundException("Unable to find warehouse %s", catalogName);
     }
-    ResolvedPolarisEntity resolvedReferenceCatalog = resolver.getResolvedReferenceCatalog();
-    Map<String, String> properties =
-        PolarisEntity.of(resolvedReferenceCatalog.getEntity()).getPropertiesAsMap();
+    PolarisResolvedPathWrapper catalogWrapper =
+        resolutionManifest.getResolvedTopLevelEntity(catalogName, PolarisEntityType.CATALOG);
+    Set<PolarisBaseEntity> activatedEntities =
+        resolutionManifest.getAllActivatedCatalogRoleAndPrincipalRoles();
+
+    // The config endpoint itself is authorized: GET_CATALOG_CONFIG maps to the
+    // CATALOG_READ_CONFIG privilege, which is subsumed by every catalog-content privilege so
+    // that principals that can use the catalog through the REST API keep working.
+    authorizer.authorizeOrThrow(
+        principal,
+        activatedEntities,
+        PolarisAuthorizableOperation.GET_CATALOG_CONFIG,
+        catalogWrapper,
+        null /* secondary */);
 
     ConfigResponse.Builder builder =
         ConfigResponse.builder()
-            .withDefaults(properties)
             .withOverrides(
                 ImmutableMap.of(
                     "prefix",
@@ -87,6 +128,25 @@ public ConfigResponse getConfig(String catalogName, PolarisPrincipal principal)
                     NamespaceUtils.DEFAULT_NAMESPACE_SEPARATOR_ENCODED))
             .withEndpoints(ImmutableList.copyOf(supportedEndpoints()));
 
+    // Catalog properties are disclosed only to principals holding CATALOG_READ_PROPERTIES (via
+    // GET_CATALOG_CONFIG_PROPERTIES), matching the management-plane getCatalog rule. Other
+    // authorized callers still receive the prefix and endpoints needed to initialize a client.
+    try {
+      authorizer.authorizeOrThrow(
+          principal,
+          activatedEntities,
+          PolarisAuthorizableOperation.GET_CATALOG_CONFIG_PROPERTIES,
+          catalogWrapper,
+          null /* secondary */);
+      PolarisBaseEntity catalogEntity = catalogWrapper.getResolvedLeafEntity().getEntity();
+      builder.withDefaults(PolarisEntity.of(catalogEntity).getPropertiesAsMap());
+    } catch (ForbiddenException e) {
+      LOGGER.debug(
+          "Principal '{}' may read catalog config but not catalog properties of '{}'",
+          principal.getName(),
+          catalogName);
+    }
+
     // Advertise Idempotency-Key support to clients. Per the REST spec, presence of this field
     // signals that mutation endpoints honor Idempotency-Key; its value is the reuse window a
     // client may retry a key within, which mirrors the server-side key TTL.