diff --git a/client/python/apache_polaris/cli/command/catalogs.py b/client/python/apache_polaris/cli/command/catalogs.py index fae73d7526..9c240b8257 100644 --- a/client/python/apache_polaris/cli/command/catalogs.py +++ b/client/python/apache_polaris/cli/command/catalogs.py @@ -74,6 +74,8 @@ class CatalogsCommand(Command): * polaris catalogs list """ + _GCP_QUOTA_PROJECT_PROPERTY = "header.x-goog-user-project" + catalogs_subcommand: str catalog_type: Optional[str] = None default_base_location: Optional[str] = None @@ -401,12 +403,23 @@ def _build_connection_config_info( warehouse=self.hadoop_warehouse, ) elif self.catalog_connection_type == CatalogConnectionType.ICEBERG.value: + connection_properties = {} + if self.catalog_authentication_type == AuthenticationType.GCP.value: + quota_project = (self.properties or {}).get( + self._GCP_QUOTA_PROJECT_PROPERTY + ) + if quota_project is not None: + connection_properties[self._GCP_QUOTA_PROJECT_PROPERTY] = ( + quota_project + ) + config = IcebergRestConnectionConfigInfo( connection_type=self.catalog_connection_type.upper().replace("-", "_"), uri=self.catalog_uri, authentication_parameters=auth_params, service_identity=service_identity, remote_catalog_name=self.iceberg_remote_catalog_name, + properties=connection_properties, ) elif self.catalog_connection_type == CatalogConnectionType.HIVE.value: config = HiveConnectionConfigInfo( @@ -425,6 +438,12 @@ def _build_connection_config_info( def execute(self, api: PolarisDefaultApi) -> None: catalog_type = cast(str, self.catalog_type) catalog_name = cast(str, self.catalog_name) + catalog_properties = dict(self.properties or {}) + if ( + self.catalog_connection_type == CatalogConnectionType.ICEBERG.value + and self.catalog_authentication_type == AuthenticationType.GCP.value + ): + catalog_properties.pop(self._GCP_QUOTA_PROJECT_PROPERTY, None) if self.catalogs_subcommand == Subcommands.CREATE: storage_config = self._build_storage_config_info() @@ -437,7 +456,7 @@ def execute(self, api: PolarisDefaultApi) -> None: storage_config_info=storage_config, properties=CatalogProperties( default_base_location=self.default_base_location, - additional_properties=self.properties, + additional_properties=catalog_properties, ), connection_config_info=connection_config, ) diff --git a/client/python/tests/test_catalogs_command.py b/client/python/tests/test_catalogs_command.py index 95920e7107..90fb7a7642 100644 --- a/client/python/tests/test_catalogs_command.py +++ b/client/python/tests/test_catalogs_command.py @@ -709,6 +709,10 @@ def test_external_catalog_gcp(self) -> None: self.assertEqual(call_args.catalog.properties.default_base_location, "dbl") self.assertEqual( call_args.catalog.properties.additional_properties, + {}, + ) + self.assertEqual( + call_args.catalog.connection_config_info.properties, {"header.x-goog-user-project": "my-billing-project"}, ) diff --git a/runtime/service/src/main/java/org/apache/polaris/service/admin/BigLakeCatalogValidator.java b/runtime/service/src/main/java/org/apache/polaris/service/admin/BigLakeCatalogValidator.java new file mode 100644 index 0000000000..13e175f40e --- /dev/null +++ b/runtime/service/src/main/java/org/apache/polaris/service/admin/BigLakeCatalogValidator.java @@ -0,0 +1,321 @@ +/* + * 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.admin; + +import com.google.common.base.Strings; +import java.net.URI; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; +import org.apache.polaris.core.admin.model.AuthenticationParameters; +import org.apache.polaris.core.admin.model.Catalog; +import org.apache.polaris.core.admin.model.ExternalCatalog; +import org.apache.polaris.core.admin.model.GcpStorageConfigInfo; +import org.apache.polaris.core.admin.model.IcebergRestConnectionConfigInfo; +import org.apache.polaris.core.admin.model.StorageConfigInfo; +import org.apache.polaris.core.config.FeatureConfiguration; +import org.apache.polaris.core.config.RealmConfig; +import org.apache.polaris.core.storage.StorageUri; + +final class BigLakeCatalogValidator { + private static final String BIGLAKE_HOST = "biglake.googleapis.com"; + private static final String BIGLAKE_PATH = "/iceberg/v1/restcatalog"; + private static final String DEFAULT_BASE_LOCATION_KEY = "default-base-location"; + private static final String QUOTA_PROJECT_HEADER = "header.x-goog-user-project"; + + private static final Pattern GCP_PROJECT_ID_PATTERN = + Pattern.compile("^[a-z][a-z0-9-]{4,28}[a-z0-9]$"); + private static final Pattern GCP_PROJECT_NUMBER_PATTERN = Pattern.compile("^[1-9][0-9]{5,}$"); + private static final Pattern BIGLAKE_URI_CATALOG_PATTERN = + Pattern.compile("^/[1-9][0-9]{5,}/catalogs/[^/\\s]+$"); + private static final Pattern BIGLAKE_RESOURCE_NAME_PATTERN = + Pattern.compile("^projects/[^/\\s]+/locations/[^/\\s]+/catalogs/[^/\\s]+$"); + private static final Pattern BIGLAKE_SIMPLE_CATALOG_PATTERN = + Pattern.compile("^[A-Za-z0-9._-]+$"); + private static final Pattern SERVICE_ACCOUNT_EMAIL_PATTERN = + Pattern.compile("^[^\\s@]+@[^\\s@]+\\.[^\\s@]+$"); + + private static final Set BLOCKED_HEADER_PROPERTIES = + Set.of("header.authorization", "header.proxy-authorization"); + + private BigLakeCatalogValidator() {} + + static void validate(RealmConfig realmConfig, Catalog catalog) { + if (!(catalog instanceof ExternalCatalog externalCatalog)) { + return; + } + + if (!(externalCatalog.getConnectionConfigInfo() + instanceof IcebergRestConnectionConfigInfo connectionConfig)) { + return; + } + + if (connectionConfig.getAuthenticationParameters() == null + || connectionConfig.getAuthenticationParameters().getAuthenticationType() + != AuthenticationParameters.AuthenticationTypeEnum.GCP) { + return; + } + + URI uri = parseUri(connectionConfig.getUri()); + if (!targetsBigLakeHost(uri)) { + return; + } + + validateBigLakeEndpoint(connectionConfig.getUri(), uri); + validateBigLakeRemoteCatalogName(connectionConfig.getRemoteCatalogName()); + validateBigLakeHeaders( + connectionConfig.getProperties(), externalCatalog.getProperties().toMap()); + validateBigLakeStorageConfiguration(realmConfig, externalCatalog); + } + + private static void validateBigLakeEndpoint(String uriString, URI uri) { + if (Strings.isNullOrEmpty(uriString)) { + throw new IllegalArgumentException( + "Invalid BigLake connectionConfigInfo.uri: an https:// BigLake endpoint is required."); + } + + if (!"https".equalsIgnoreCase(uri.getScheme())) { + throw new IllegalArgumentException( + "Invalid BigLake connectionConfigInfo.uri '" + + uriString + + "': BigLake requires an https:// URI."); + } + + if (!BIGLAKE_HOST.equalsIgnoreCase(uri.getHost())) { + throw new IllegalArgumentException( + "Invalid BigLake connectionConfigInfo.uri '" + + uriString + + "': unsupported host '" + + uri.getHost() + + "'. Expected '" + + BIGLAKE_HOST + + "'."); + } + + String normalizedPath = normalizePath(uri.getPath()); + if (!BIGLAKE_PATH.equals(normalizedPath)) { + throw new IllegalArgumentException( + "Invalid BigLake connectionConfigInfo.uri '" + + uriString + + "': unsupported path '" + + uri.getPath() + + "'. Expected '" + + BIGLAKE_PATH + + "'."); + } + + if (uri.getRawQuery() != null || uri.getRawFragment() != null || uri.getPort() != -1) { + throw new IllegalArgumentException( + "Invalid BigLake connectionConfigInfo.uri '" + + uriString + + "': query, fragment, and custom port components are not supported."); + } + } + + private static void validateBigLakeRemoteCatalogName(String remoteCatalogName) { + if (Strings.isNullOrEmpty(remoteCatalogName) || remoteCatalogName.trim().isEmpty()) { + throw new IllegalArgumentException( + "Invalid BigLake connectionConfigInfo.remoteCatalogName: a remote catalog or warehouse identifier is required."); + } + + String trimmedRemoteCatalogName = remoteCatalogName.trim(); + if (trimmedRemoteCatalogName.startsWith("gs://")) { + validateGsLocation("connectionConfigInfo.remoteCatalogName", trimmedRemoteCatalogName); + return; + } + + if (isBigLakeCatalogUri(trimmedRemoteCatalogName) + || BIGLAKE_RESOURCE_NAME_PATTERN.matcher(trimmedRemoteCatalogName).matches() + || BIGLAKE_SIMPLE_CATALOG_PATTERN.matcher(trimmedRemoteCatalogName).matches()) { + return; + } + + throw new IllegalArgumentException( + "Invalid BigLake connectionConfigInfo.remoteCatalogName '" + + remoteCatalogName + + "': expected a BigLake catalog identifier or gs:// warehouse location."); + } + + private static void validateBigLakeHeaders( + Map connectionProperties, Map catalogProperties) { + Map headerProperties = + connectionProperties != null ? connectionProperties : Map.of(); + + for (String propertyName : headerProperties.keySet()) { + if (propertyName == null) { + continue; + } + + String normalizedPropertyName = propertyName.toLowerCase(Locale.ROOT); + if (!normalizedPropertyName.startsWith("header.") + || QUOTA_PROJECT_HEADER.equals(normalizedPropertyName)) { + continue; + } + + if (BLOCKED_HEADER_PROPERTIES.contains(normalizedPropertyName)) { + throw new IllegalArgumentException( + "Invalid BigLake connectionConfigInfo.properties entry '" + + propertyName + + "': overriding security-sensitive headers is not allowed."); + } + + throw new IllegalArgumentException( + "Invalid BigLake connectionConfigInfo.properties entry '" + + propertyName + + "': only '" + + QUOTA_PROJECT_HEADER + + "' is supported."); + } + + String quotaProject = headerProperties.get(QUOTA_PROJECT_HEADER); + if (Strings.isNullOrEmpty(quotaProject) && catalogProperties != null) { + // Preserve existing CLI-created catalogs while new CLI requests store this header on the + // connection configuration, where it is used for outbound BigLake requests. + quotaProject = catalogProperties.get(QUOTA_PROJECT_HEADER); + } + if (Strings.isNullOrEmpty(quotaProject) || quotaProject.trim().isEmpty()) { + throw new IllegalArgumentException( + "Invalid BigLake connectionConfigInfo.properties entry or catalog.properties entry '" + + QUOTA_PROJECT_HEADER + + "': a quota project is required."); + } + + String trimmedQuotaProject = quotaProject.trim(); + if (!GCP_PROJECT_ID_PATTERN.matcher(trimmedQuotaProject).matches() + && !GCP_PROJECT_NUMBER_PATTERN.matcher(trimmedQuotaProject).matches()) { + throw new IllegalArgumentException( + "Invalid BigLake connectionConfigInfo.properties entry or catalog.properties entry '" + + QUOTA_PROJECT_HEADER + + "': '" + + quotaProject + + "' is not a valid GCP quota project."); + } + } + + private static void validateBigLakeStorageConfiguration( + RealmConfig realmConfig, ExternalCatalog externalCatalog) { + boolean credentialVendingEnabled = + realmConfig.getConfig( + FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING, + externalCatalog.getProperties().toMap()) + && realmConfig.getConfig( + FeatureConfiguration.ALLOW_FEDERATED_CATALOGS_CREDENTIAL_VENDING, + externalCatalog.getProperties().toMap()); + + StorageConfigInfo storageConfigInfo = externalCatalog.getStorageConfigInfo(); + if (storageConfigInfo == null) { + if (credentialVendingEnabled) { + throw new IllegalArgumentException( + "Invalid BigLake storageConfigInfo: GCS storage configuration is required when credential vending is enabled."); + } + return; + } + + if (storageConfigInfo.getStorageType() != StorageConfigInfo.StorageTypeEnum.GCS + || !(storageConfigInfo instanceof GcpStorageConfigInfo gcpStorageConfigInfo)) { + throw new IllegalArgumentException( + "Invalid BigLake storageConfigInfo.storageType: expected GCS but found " + + storageConfigInfo.getStorageType() + + "."); + } + + String defaultBaseLocation = + externalCatalog.getProperties().toMap().get(DEFAULT_BASE_LOCATION_KEY); + validateGsLocation("catalog.properties." + DEFAULT_BASE_LOCATION_KEY, defaultBaseLocation); + + List allowedLocations = gcpStorageConfigInfo.getAllowedLocations(); + if (allowedLocations != null) { + for (int index = 0; index < allowedLocations.size(); index++) { + validateGsLocation( + "storageConfigInfo.allowedLocations[" + index + "]", allowedLocations.get(index)); + } + } + + String serviceAccount = gcpStorageConfigInfo.getGcsServiceAccount(); + if (!Strings.isNullOrEmpty(serviceAccount) + && !SERVICE_ACCOUNT_EMAIL_PATTERN.matcher(serviceAccount).matches()) { + throw new IllegalArgumentException( + "Invalid BigLake storageConfigInfo.gcsServiceAccount '" + + serviceAccount + + "': expected a syntactically valid service account email."); + } + + if (credentialVendingEnabled && Strings.isNullOrEmpty(serviceAccount)) { + throw new IllegalArgumentException( + "Invalid BigLake storageConfigInfo.gcsServiceAccount: a Google service account is required when credential vending is enabled."); + } + } + + private static void validateGsLocation(String fieldName, String location) { + if (Strings.isNullOrEmpty(location) || location.trim().isEmpty()) { + throw new IllegalArgumentException( + "Invalid BigLake " + fieldName + ": a non-empty gs:// location is required."); + } + + StorageUri storageUri; + try { + storageUri = StorageUri.parse(location); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException( + "Invalid BigLake " + fieldName + " '" + location + "': malformed gs:// location.", e); + } + + if (!"gs".equalsIgnoreCase(storageUri.scheme()) + || Strings.isNullOrEmpty(storageUri.authority())) { + throw new IllegalArgumentException( + "Invalid BigLake " + fieldName + " '" + location + "': expected a gs:// location."); + } + } + + private static URI parseUri(String uriString) { + if (Strings.isNullOrEmpty(uriString)) { + return null; + } + + try { + return URI.create(uriString); + } catch (IllegalArgumentException e) { + return null; + } + } + + private static boolean targetsBigLakeHost(URI uri) { + return uri != null && BIGLAKE_HOST.equalsIgnoreCase(uri.getHost()); + } + + private static boolean isBigLakeCatalogUri(String remoteCatalogName) { + URI uri = parseUri(remoteCatalogName); + return uri != null + && "bl".equalsIgnoreCase(uri.getScheme()) + && "projects".equalsIgnoreCase(uri.getHost()) + && uri.getPort() == -1 + && uri.getRawQuery() == null + && uri.getRawFragment() == null + && BIGLAKE_URI_CATALOG_PATTERN.matcher(normalizePath(uri.getPath())).matches(); + } + + private static String normalizePath(String path) { + if (Strings.isNullOrEmpty(path)) { + return ""; + } + return path.endsWith("/") && path.length() > 1 ? path.substring(0, path.length() - 1) : path; + } +} diff --git a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java index 3506fd1771..6250298190 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java @@ -1035,6 +1035,8 @@ private void validateUpdateCatalogDiffOrThrow( } CatalogEntity updatedEntity = updateBuilder.build(); + BigLakeCatalogValidator.validate( + realmConfig, updatedEntity.asCatalog(getServiceIdentityProvider())); validateUpdateCatalogDiffOrThrow(currentCatalogEntity, updatedEntity); if (catalogOverlapsWithExistingCatalog(updatedEntity)) { diff --git a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java index fe648588b0..69b9277cf6 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java @@ -171,6 +171,7 @@ private void validateExternalCatalog(Catalog catalog) { if (connectionConfigInfo != null) { validateConnectionConfigInfo(connectionConfigInfo); validateAuthenticationParameters(connectionConfigInfo.getAuthenticationParameters()); + BigLakeCatalogValidator.validate(realmConfig, catalog); } } } diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/BigLakeCatalogValidatorTest.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/BigLakeCatalogValidatorTest.java new file mode 100644 index 0000000000..c1bdc8c4d1 --- /dev/null +++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/BigLakeCatalogValidatorTest.java @@ -0,0 +1,430 @@ +/* + * 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.admin; + +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.List; +import java.util.Map; +import org.apache.polaris.core.admin.model.AuthenticationParameters; +import org.apache.polaris.core.admin.model.Catalog; +import org.apache.polaris.core.admin.model.CatalogProperties; +import org.apache.polaris.core.admin.model.ConnectionConfigInfo; +import org.apache.polaris.core.admin.model.ExternalCatalog; +import org.apache.polaris.core.admin.model.GcpAuthenticationParameters; +import org.apache.polaris.core.admin.model.GcpStorageConfigInfo; +import org.apache.polaris.core.admin.model.IcebergRestConnectionConfigInfo; +import org.apache.polaris.core.admin.model.StorageConfigInfo; +import org.apache.polaris.core.config.RealmConfig; +import org.apache.polaris.service.TestServices; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +class BigLakeCatalogValidatorTest { + private RealmConfig realmConfig; + + @BeforeEach + void setup() { + realmConfig = TestServices.builder().build().realmConfig(); + } + + @Test + void validBigLakeConfigurationPasses() { + assertThatCode( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog", + "my-remote-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .doesNotThrowAnyException(); + } + + @Test + void validBigLakeConfigurationWithLegacyCatalogQuotaProjectPasses() { + CatalogProperties catalogProperties = + CatalogProperties.builder("gs://bucket/path/to/data").build(); + catalogProperties.put("enable.credential.vending", "true"); + catalogProperties.put("header.x-goog-user-project", "my-billing-project"); + + Catalog catalog = + ExternalCatalog.builder() + .setType(Catalog.TypeEnum.EXTERNAL) + .setName("test-biglake-catalog") + .setProperties(catalogProperties) + .setStorageConfigInfo(validGcsStorage("gs://bucket/path/to/data")) + .setConnectionConfigInfo( + IcebergRestConnectionConfigInfo.builder() + .setConnectionType(ConnectionConfigInfo.ConnectionTypeEnum.ICEBERG_REST) + .setUri("https://biglake.googleapis.com/iceberg/v1/restcatalog") + .setRemoteCatalogName("my-remote-catalog") + .setProperties(Map.of()) + .setAuthenticationParameters( + GcpAuthenticationParameters.builder() + .setAuthenticationType( + AuthenticationParameters.AuthenticationTypeEnum.GCP) + .build()) + .build()) + .build(); + + assertThatCode(() -> BigLakeCatalogValidator.validate(realmConfig, catalog)) + .doesNotThrowAnyException(); + } + + @Test + void validBigLakeBlCatalogIdentifierPasses() { + assertThatCode( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog", + "bl://projects/123456789/catalogs/my-biglake-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .doesNotThrowAnyException(); + } + + @Test + void validBigLakeWarehouseIdentifierPasses() { + assertThatCode( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog", + "gs://bucket/path/to/warehouse", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .doesNotThrowAnyException(); + } + + @Test + void skipsValidationForNonBigLakeGcpRestEndpoint() { + assertThatCode( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://catalog-gateway.example.com/iceberg/v1", + null, + Map.of(), + true, + "s3://bucket/path/to/data", + null))) + .doesNotThrowAnyException(); + } + + @Test + void rejectsNonHttpsEndpoint() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "http://biglake.googleapis.com/iceberg/v1/restcatalog", + "my-remote-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("requires an https:// URI"); + } + + @Test + void rejectsUnsupportedEndpointPath() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/not-biglake", + "my-remote-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("unsupported path"); + } + + @Test + void rejectsEndpointWithQueryString() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog?warehouse=test", + "my-remote-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("query, fragment, and custom port components are not supported"); + } + + @Test + void rejectsEndpointWithFragment() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog#fragment", + "my-remote-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("query, fragment, and custom port components are not supported"); + } + + @Test + void rejectsEndpointWithCustomPort() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com:8443/iceberg/v1/restcatalog", + "my-remote-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("query, fragment, and custom port components are not supported"); + } + + @Test + void rejectsMissingRemoteCatalogIdentifier() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog", + " ", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("remote catalog or warehouse identifier is required"); + } + + @Test + void rejectsMissingQuotaProjectHeader() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog", + "my-remote-catalog", + Map.of(), + true, + "gs://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("a quota project is required"); + } + + @Test + void rejectsUnsupportedHeaderOverride() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog", + "my-remote-catalog", + Map.of( + "header.x-goog-user-project", + "my-billing-project", + "header.authorization", + "Bearer secret"), + true, + "gs://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("overriding security-sensitive headers is not allowed"); + } + + @Test + void rejectsMissingGcsStorageWhenCredentialVendingEnabled() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog", + "my-remote-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + null))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("GCS storage configuration is required"); + } + + @Test + void rejectsMalformedGsBaseLocation() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog", + "my-remote-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "s3://bucket/path/to/data", + validGcsStorage("gs://bucket/path/to/data")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("catalog.properties.default-base-location"); + } + + @Test + void rejectsMalformedGsAllowedLocation() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog", + "my-remote-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + invalidAllowedLocationStorage()))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("storageConfigInfo.allowedLocations[0]"); + } + + @Test + void rejectsInvalidGcsServiceAccount() { + assertThatThrownBy( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog", + "my-remote-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + GcpStorageConfigInfo.builder() + .setStorageType(StorageConfigInfo.StorageTypeEnum.GCS) + .setGcsServiceAccount("not-a-service-account") + .setAllowedLocations(List.of("gs://bucket/path/to/data")) + .build()))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("expected a syntactically valid service account email"); + } + + @ParameterizedTest + @ValueSource( + strings = { + "test-sa@my-project.iam.gserviceaccount.com", + "123456789-compute@developer.gserviceaccount.com", + "my-project@appspot.gserviceaccount.com" + }) + void acceptsSupportedGcsServiceAccountEmailForms(String serviceAccount) { + StorageConfigInfo storageConfigInfo = + GcpStorageConfigInfo.builder() + .setStorageType(StorageConfigInfo.StorageTypeEnum.GCS) + .setGcsServiceAccount(serviceAccount) + .setAllowedLocations(List.of("gs://bucket/path/to/data")) + .build(); + + assertThatCode( + () -> + BigLakeCatalogValidator.validate( + realmConfig, + bigLakeCatalog( + "https://biglake.googleapis.com/iceberg/v1/restcatalog", + "my-remote-catalog", + Map.of("header.x-goog-user-project", "my-billing-project"), + true, + "gs://bucket/path/to/data", + storageConfigInfo))) + .doesNotThrowAnyException(); + } + + private Catalog bigLakeCatalog( + String uri, + String remoteCatalogName, + Map connectionProperties, + boolean credentialVendingEnabled, + String defaultBaseLocation, + StorageConfigInfo storageConfigInfo) { + CatalogProperties catalogProperties = CatalogProperties.builder(defaultBaseLocation).build(); + catalogProperties.put("enable.credential.vending", Boolean.toString(credentialVendingEnabled)); + + return ExternalCatalog.builder() + .setType(Catalog.TypeEnum.EXTERNAL) + .setName("test-biglake-catalog") + .setProperties(catalogProperties) + .setStorageConfigInfo(storageConfigInfo) + .setConnectionConfigInfo( + IcebergRestConnectionConfigInfo.builder() + .setConnectionType(ConnectionConfigInfo.ConnectionTypeEnum.ICEBERG_REST) + .setUri(uri) + .setRemoteCatalogName(remoteCatalogName) + .setProperties(connectionProperties) + .setAuthenticationParameters( + GcpAuthenticationParameters.builder() + .setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.GCP) + .build()) + .build()) + .build(); + } + + private StorageConfigInfo validGcsStorage(String allowedLocation) { + return GcpStorageConfigInfo.builder() + .setStorageType(StorageConfigInfo.StorageTypeEnum.GCS) + .setGcsServiceAccount("test-sa@my-project.iam.gserviceaccount.com") + .setAllowedLocations(List.of(allowedLocation)) + .build(); + } + + private StorageConfigInfo invalidAllowedLocationStorage() { + return GcpStorageConfigInfo.builder() + .setStorageType(StorageConfigInfo.StorageTypeEnum.GCS) + .setGcsServiceAccount("test-sa@my-project.iam.gserviceaccount.com") + .setAllowedLocations(List.of("bucket/path/to/data")) + .build(); + } +} diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java index 52ab0a5cef..7013005f94 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java @@ -38,6 +38,8 @@ import org.apache.polaris.core.admin.model.CreateCatalogRequest; import org.apache.polaris.core.admin.model.ExternalCatalog; import org.apache.polaris.core.admin.model.FileStorageConfigInfo; +import org.apache.polaris.core.admin.model.GcpAuthenticationParameters; +import org.apache.polaris.core.admin.model.GcpStorageConfigInfo; import org.apache.polaris.core.admin.model.IcebergRestConnectionConfigInfo; import org.apache.polaris.core.admin.model.OAuthClientCredentialsParameters; import org.apache.polaris.core.admin.model.PolarisCatalog; @@ -369,6 +371,163 @@ public void testUpdateCatalogWithDisallowedConfigs() { "Explicitly setting polaris.config.enable-sub-catalog-rbac-for-federated-catalogs is not allowed because ALLOW_SETTING_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS is set to false."); } + @Test + public void testCreateAndUpdateValidBigLakeCatalog() { + String catalogName = "biglake-catalog"; + String initialBaseLocation = "gs://bucket/path/to/data"; + String updatedBaseLocation = "gs://bucket/path/to/updated-data"; + Catalog catalog = + createBigLakeCatalog( + catalogName, initialBaseLocation, createBigLakeStorageConfig(initialBaseLocation)); + + try (Response response = + services + .catalogsApi() + .createCatalog( + new CreateCatalogRequest(catalog), + services.realmContext(), + services.securityContext())) { + assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus); + } + + Catalog fetchedCatalog; + try (Response response = + services + .catalogsApi() + .getCatalog(catalogName, services.realmContext(), services.securityContext())) { + assertThat(response).returns(Response.Status.OK.getStatusCode(), Response::getStatus); + fetchedCatalog = (Catalog) response.getEntity(); + } + + UpdateCatalogRequest updateRequest = + UpdateCatalogRequest.builder() + .setCurrentEntityVersion(fetchedCatalog.getEntityVersion()) + .setProperties( + Map.of( + "default-base-location", + updatedBaseLocation, + "enable.credential.vending", + "true")) + .setStorageConfigInfo(createBigLakeStorageConfig(updatedBaseLocation)) + .build(); + + try (Response response = + services + .catalogsApi() + .updateCatalog( + catalogName, updateRequest, services.realmContext(), services.securityContext())) { + assertThat(response).returns(Response.Status.OK.getStatusCode(), Response::getStatus); + Catalog updatedCatalog = (Catalog) response.getEntity(); + assertThat(updatedCatalog.getProperties().getDefaultBaseLocation()) + .isEqualTo(updatedBaseLocation); + } + } + + @Test + public void testCreateAndUpdateNonBigLakeGcpRestCatalogSkipsBigLakeValidation() { + String catalogName = "generic-gcp-rest-catalog"; + String initialBaseLocation = "s3://bucket/path/to/data"; + String updatedBaseLocation = "s3://bucket/path/to/updated-data"; + Catalog catalog = + createGenericGcpRestCatalog( + catalogName, initialBaseLocation, createGenericS3StorageConfig(initialBaseLocation)); + + try (Response response = + services + .catalogsApi() + .createCatalog( + new CreateCatalogRequest(catalog), + services.realmContext(), + services.securityContext())) { + assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus); + } + + Catalog fetchedCatalog; + try (Response response = + services + .catalogsApi() + .getCatalog(catalogName, services.realmContext(), services.securityContext())) { + assertThat(response).returns(Response.Status.OK.getStatusCode(), Response::getStatus); + fetchedCatalog = (Catalog) response.getEntity(); + assertThat(fetchedCatalog.getProperties().getDefaultBaseLocation()) + .isEqualTo(initialBaseLocation); + } + + UpdateCatalogRequest updateRequest = + UpdateCatalogRequest.builder() + .setCurrentEntityVersion(fetchedCatalog.getEntityVersion()) + .setProperties( + Map.of( + "default-base-location", + updatedBaseLocation, + "enable.credential.vending", + "true")) + .setStorageConfigInfo(createGenericS3StorageConfig(updatedBaseLocation)) + .build(); + + try (Response response = + services + .catalogsApi() + .updateCatalog( + catalogName, updateRequest, services.realmContext(), services.securityContext())) { + assertThat(response).returns(Response.Status.OK.getStatusCode(), Response::getStatus); + Catalog updatedCatalog = (Catalog) response.getEntity(); + assertThat(updatedCatalog.getProperties().getDefaultBaseLocation()) + .isEqualTo(updatedBaseLocation); + } + } + + @Test + public void testUpdateBigLakeCatalogRejectsInvalidMergedConfiguration() { + String catalogName = "biglake-catalog"; + String initialBaseLocation = "gs://bucket/path/to/data"; + Catalog catalog = + createBigLakeCatalog( + catalogName, initialBaseLocation, createBigLakeStorageConfig(initialBaseLocation)); + + try (Response response = + services + .catalogsApi() + .createCatalog( + new CreateCatalogRequest(catalog), + services.realmContext(), + services.securityContext())) { + assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus); + } + + Catalog fetchedCatalog; + try (Response response = + services + .catalogsApi() + .getCatalog(catalogName, services.realmContext(), services.securityContext())) { + assertThat(response).returns(Response.Status.OK.getStatusCode(), Response::getStatus); + fetchedCatalog = (Catalog) response.getEntity(); + } + + UpdateCatalogRequest updateRequest = + UpdateCatalogRequest.builder() + .setCurrentEntityVersion(fetchedCatalog.getEntityVersion()) + .setProperties( + Map.of( + "default-base-location", + "s3://bucket/path/to/data", + "enable.credential.vending", + "true")) + .build(); + + assertThatThrownBy( + () -> + services + .catalogsApi() + .updateCatalog( + catalogName, + updateRequest, + services.realmContext(), + services.securityContext())) + .isInstanceOfAny(BadRequestException.class, IllegalArgumentException.class) + .hasMessageContaining("default-base-location"); + } + private PolarisAdminService setupPolarisAdminService( PolarisMetaStoreManager metaStoreManager, PolarisCallContext callContext) { PrincipalEntity rootPrincipal = @@ -391,6 +550,69 @@ private PolarisAdminService setupPolarisAdminService( ReservedProperties.NONE); } + private Catalog createBigLakeCatalog( + String catalogName, String defaultBaseLocation, StorageConfigInfo storageConfigInfo) { + CatalogProperties catalogProperties = CatalogProperties.builder(defaultBaseLocation).build(); + catalogProperties.put("enable.credential.vending", "true"); + return ExternalCatalog.builder() + .setType(Catalog.TypeEnum.EXTERNAL) + .setName(catalogName) + .setProperties(catalogProperties) + .setStorageConfigInfo(storageConfigInfo) + .setConnectionConfigInfo( + IcebergRestConnectionConfigInfo.builder( + ConnectionConfigInfo.ConnectionTypeEnum.ICEBERG_REST) + .setUri("https://biglake.googleapis.com/iceberg/v1/restcatalog") + .setRemoteCatalogName("my-remote-catalog") + .setProperties(Map.of("header.x-goog-user-project", "my-billing-project")) + .setAuthenticationParameters( + GcpAuthenticationParameters.builder() + .setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.GCP) + .build()) + .build()) + .build(); + } + + private Catalog createGenericGcpRestCatalog( + String catalogName, String defaultBaseLocation, StorageConfigInfo storageConfigInfo) { + CatalogProperties catalogProperties = CatalogProperties.builder(defaultBaseLocation).build(); + catalogProperties.put("enable.credential.vending", "true"); + return ExternalCatalog.builder() + .setType(Catalog.TypeEnum.EXTERNAL) + .setName(catalogName) + .setProperties(catalogProperties) + .setStorageConfigInfo(storageConfigInfo) + .setConnectionConfigInfo( + IcebergRestConnectionConfigInfo.builder( + ConnectionConfigInfo.ConnectionTypeEnum.ICEBERG_REST) + .setUri("https://catalog-gateway.example.com/iceberg/v1") + .setRemoteCatalogName("my-remote-catalog") + .setAuthenticationParameters( + GcpAuthenticationParameters.builder() + .setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.GCP) + .build()) + .build()) + .build(); + } + + private StorageConfigInfo createBigLakeStorageConfig(String allowedLocation) { + return GcpStorageConfigInfo.builder() + .setStorageType(StorageConfigInfo.StorageTypeEnum.GCS) + .setGcsServiceAccount("test-sa@my-project.iam.gserviceaccount.com") + .setAllowedLocations(List.of(allowedLocation)) + .build(); + } + + private StorageConfigInfo createGenericS3StorageConfig(String allowedLocation) { + return AwsStorageConfigInfo.builder() + .setStorageType(StorageConfigInfo.StorageTypeEnum.S3) + .setRoleArn("arn:aws:iam::123456789012:role/my-role") + .setExternalId("externalId") + .setUserArn("userArn") + .setAllowedLocations(List.of(allowedLocation)) + .build(); + } + private PrincipalEntity createPrincipal( PolarisMetaStoreManager metaStoreManager, PolarisCallContext callContext, String name) { return new PrincipalEntity.Builder()