Use UserSecretsManagerFactory to Produce the UserSecretsManager

polaris-core/src/main/java/org/apache/polaris/core/connection/ConnectionConfigInfoDpo.java

@@ -29,8 +29,11 @@
 import java.net.MalformedURLException;
 import java.net.URI;
 import java.net.URL;
+import java.util.Map;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.admin.model.ConnectionConfigInfo;
+import org.apache.polaris.core.admin.model.IcebergRestConnectionConfigInfo;
+import org.apache.polaris.core.secrets.UserSecretReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -47,17 +50,26 @@ public abstract class ConnectionConfigInfoDpo implements IcebergCatalogPropertie
   // The URI of the remote catalog
   private final String uri;
 
+  // The authentication parameters for the connection
+  private final AuthenticationParametersDpo authenticationParameters;
+
   public ConnectionConfigInfoDpo(
       @JsonProperty(value = "connectionType", required = true) @Nonnull
           ConnectionType connectionType,
-      @JsonProperty(value = "uri", required = true) @Nonnull String uri) {
-    this(connectionType, uri, true);
+      @JsonProperty(value = "uri", required = true) @Nonnull String uri,
+      @JsonProperty(value = "authenticationParameters", required = true) @Nonnull
+          AuthenticationParametersDpo authenticationParameters) {
+    this(connectionType, uri, authenticationParameters, true);
   }
 
   protected ConnectionConfigInfoDpo(
-      ConnectionType connectionType, String uri, boolean validateUri) {
+      @Nonnull ConnectionType connectionType,
+      @Nonnull String uri,
+      @Nonnull AuthenticationParametersDpo authenticationParameters,
+      boolean validateUri) {
     this.connectionType = connectionType;
     this.uri = uri;
+    this.authenticationParameters = authenticationParameters;
     if (validateUri) {
       validateUri(uri);
     }
@@ -71,6 +83,10 @@ public String getUri() {
     return uri;
   }
 
+  public AuthenticationParametersDpo getAuthenticationParameters() {
+    return authenticationParameters;
+  }
+
   private static final ObjectMapper DEFAULT_MAPPER;
 
   static {
@@ -108,5 +124,30 @@ protected void validateUri(String uri) {
     }
   }
 
+  public static ConnectionConfigInfoDpo fromConnectionConfigInfoModelWithSecrets(
+      ConnectionConfigInfo connectionConfigurationModel,
+      Map<String, UserSecretReference> secretReferences) {
+    ConnectionConfigInfoDpo config = null;
+    switch (connectionConfigurationModel.getConnectionType()) {
+      case ICEBERG_REST:
+        IcebergRestConnectionConfigInfo icebergRestConfigModel =
+            (IcebergRestConnectionConfigInfo) connectionConfigurationModel;
+        AuthenticationParametersDpo authenticationParameters =
+            AuthenticationParametersDpo.fromAuthenticationParametersModelWithSecrets(
+                icebergRestConfigModel.getAuthenticationParameters(), secretReferences);
+        config =
+            new IcebergRestConnectionConfigInfoDpo(
+                ConnectionType.ICEBERG_REST,
+                icebergRestConfigModel.getUri(),
+                authenticationParameters,
+                icebergRestConfigModel.getRemoteCatalogName());
+        break;
+      default:
+        throw new IllegalStateException(
+            "Unsupported connection type: " + connectionConfigurationModel.getConnectionType());
+    }
+    return config;
+  }
+
   public abstract ConnectionConfigInfo asConnectionConfigInfoModel();
 }

polaris-core/src/main/java/org/apache/polaris/core/connection/IcebergRestConnectionConfigInfoDpo.java

@@ -34,29 +34,22 @@ public class IcebergRestConnectionConfigInfoDpo extends ConnectionConfigInfoDpo
 
   private final String remoteCatalogName;
 
-  private final AuthenticationParametersDpo authenticationParameters;
-
   public IcebergRestConnectionConfigInfoDpo(
       @JsonProperty(value = "connectionType", required = true) @Nonnull
           ConnectionType connectionType,
       @JsonProperty(value = "uri", required = true) @Nonnull String uri,
+      @JsonProperty(value = "authenticationParameters", required = true) @Nonnull
+          AuthenticationParametersDpo authenticationParameters,
       @JsonProperty(value = "remoteCatalogName", required = false) @Nullable
-          String remoteCatalogName,
-      @JsonProperty(value = "authenticationParameters", required = false) @Nonnull
-          AuthenticationParametersDpo authenticationParameters) {
-    super(connectionType, uri);
+          String remoteCatalogName) {
+    super(connectionType, uri, authenticationParameters);
     this.remoteCatalogName = remoteCatalogName;
-    this.authenticationParameters = authenticationParameters;
   }
 
   public String getRemoteCatalogName() {
     return remoteCatalogName;
   }
 
-  public AuthenticationParametersDpo getAuthenticationParameters() {
-    return authenticationParameters;
-  }
-
   @Override
   public @Nonnull Map<String, String> asIcebergCatalogProperties(
       UserSecretsManager secretsManager) {
@@ -65,7 +58,7 @@ public AuthenticationParametersDpo getAuthenticationParameters() {
     if (getRemoteCatalogName() != null) {
       properties.put(CatalogProperties.WAREHOUSE_LOCATION, getRemoteCatalogName());
     }
-    properties.putAll(authenticationParameters.asIcebergCatalogProperties(secretsManager));
+    properties.putAll(getAuthenticationParameters().asIcebergCatalogProperties(secretsManager));
     return properties;
   }
 
@@ -75,7 +68,8 @@ public ConnectionConfigInfo asConnectionConfigInfoModel() {
         .setConnectionType(ConnectionConfigInfo.ConnectionTypeEnum.ICEBERG_REST)
         .setUri(getUri())
         .setRemoteCatalogName(getRemoteCatalogName())
-        .setAuthenticationParameters(authenticationParameters.asAuthenticationParametersModel())
+        .setAuthenticationParameters(
+            getAuthenticationParameters().asAuthenticationParametersModel())
         .build();
   }
 

polaris-core/src/main/java/org/apache/polaris/core/connection/OAuthClientCredentialsParametersDpo.java

@@ -94,6 +94,7 @@ public OAuthClientCredentialsParametersDpo(
         Objects.requireNonNullElse(scopes, List.of(OAuth2Properties.CATALOG_SCOPE)));
   }
 
+  @JsonIgnore
   private @Nonnull String getCredentialAsConcatenatedString(UserSecretsManager secretsManager) {
     String clientSecret = secretsManager.readSecret(getClientSecretReference());
     return COLON_JOINER.join(clientId, clientSecret);

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

@@ -39,14 +39,10 @@
 import org.apache.polaris.core.admin.model.ExternalCatalog;
 import org.apache.polaris.core.admin.model.FileStorageConfigInfo;
 import org.apache.polaris.core.admin.model.GcpStorageConfigInfo;
-import org.apache.polaris.core.admin.model.IcebergRestConnectionConfigInfo;
 import org.apache.polaris.core.admin.model.PolarisCatalog;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
 import org.apache.polaris.core.config.BehaviorChangeConfiguration;
-import org.apache.polaris.core.connection.AuthenticationParametersDpo;
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
-import org.apache.polaris.core.connection.ConnectionType;
-import org.apache.polaris.core.connection.IcebergRestConnectionConfigInfoDpo;
 import org.apache.polaris.core.secrets.UserSecretReference;
 import org.apache.polaris.core.storage.FileStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
@@ -325,25 +321,9 @@ public Builder setConnectionConfigurationInfoWithSecrets(
         ConnectionConfigInfo connectionConfigurationModel,
         Map<String, UserSecretReference> secretReferences) {
       if (connectionConfigurationModel != null) {
-        ConnectionConfigInfoDpo config;
-        switch (connectionConfigurationModel.getConnectionType()) {
-          case ICEBERG_REST:
-            IcebergRestConnectionConfigInfo icebergRestConfigModel =
-                (IcebergRestConnectionConfigInfo) connectionConfigurationModel;
-            AuthenticationParametersDpo authenticationParameters =
-                AuthenticationParametersDpo.fromAuthenticationParametersModelWithSecrets(
-                    icebergRestConfigModel.getAuthenticationParameters(), secretReferences);
-            config =
-                new IcebergRestConnectionConfigInfoDpo(
-                    ConnectionType.ICEBERG_REST,
-                    icebergRestConfigModel.getUri(),
-                    icebergRestConfigModel.getRemoteCatalogName(),
-                    authenticationParameters);
-            break;
-          default:
-            throw new IllegalStateException(
-                "Unsupported connection type: " + connectionConfigurationModel.getConnectionType());
-        }
+        ConnectionConfigInfoDpo config =
+            ConnectionConfigInfoDpo.fromConnectionConfigInfoModelWithSecrets(
+                connectionConfigurationModel, secretReferences);
         internalProperties.put(
             PolarisEntityConstants.getConnectionConfigInfoPropertyName(), config.serialize());
       }

polaris-core/src/main/java/org/apache/polaris/core/secrets/UnsafeInMemorySecretsManager.java

@@ -18,6 +18,7 @@
  */
 package org.apache.polaris.core.secrets;
 
+import jakarta.annotation.Nonnull;
 import java.io.UnsupportedEncodingException;
 import java.security.SecureRandom;
 import java.util.Base64;
@@ -46,7 +47,8 @@ public class UnsafeInMemorySecretsManager implements UserSecretsManager {
 
   /** {@inheritDoc} */
   @Override
-  public UserSecretReference writeSecret(String secret, PolarisEntity forEntity) {
+  @Nonnull
+  public UserSecretReference writeSecret(@Nonnull String secret, @Nonnull PolarisEntity forEntity) {
     // For illustrative purposes and to exercise the control flow of requiring both the stored
     // secret as well as the secretReferencePayload to recover the original secret, we'll use
     // basic XOR encryption and store the randomly generated key in the reference payload.
@@ -107,7 +109,8 @@ public UserSecretReference writeSecret(String secret, PolarisEntity forEntity) {
 
   /** {@inheritDoc} */
   @Override
-  public String readSecret(UserSecretReference secretReference) {
+  @Nonnull
+  public String readSecret(@Nonnull UserSecretReference secretReference) {
     // TODO: Precondition checks and/or wire in PolarisDiagnostics
     String encryptedSecretCipherTextBase64 = rawSecretStore.get(secretReference.getUrn());
     if (encryptedSecretCipherTextBase64 == null) {
@@ -148,7 +151,7 @@ public String readSecret(UserSecretReference secretReference) {
 
   /** {@inheritDoc} */
   @Override
-  public void deleteSecret(UserSecretReference secretReference) {
+  public void deleteSecret(@Nonnull UserSecretReference secretReference) {
     rawSecretStore.remove(secretReference.getUrn());
   }
 }

polaris-core/src/main/java/org/apache/polaris/core/secrets/UserSecretsManager.java

@@ -18,6 +18,7 @@
  */
 package org.apache.polaris.core.secrets;
 
+import jakarta.annotation.Nonnull;
 import org.apache.polaris.core.entity.PolarisEntity;
 
 /**
@@ -42,20 +43,22 @@ public interface UserSecretsManager {
    * @return A reference object that can be used to retrieve the secret which is safe to store in
    *     its entirety within a persisted PolarisEntity
    */
-  UserSecretReference writeSecret(String secret, PolarisEntity forEntity);
+  @Nonnull
+  UserSecretReference writeSecret(@Nonnull String secret, @Nonnull PolarisEntity forEntity);
 
   /**
    * Retrieve a secret using the {@code secretReference}.
    *
    * @param secretReference Identifier and any associated payload used for retrieving the secret
    * @return The stored secret, or null if it no longer exists
    */
-  String readSecret(UserSecretReference secretReference);
+  @Nonnull
+  String readSecret(@Nonnull UserSecretReference secretReference);
 
   /**
    * Delete a stored secret
    *
    * @param secretReference Identifier and any associated payload used for retrieving the secret
    */
-  void deleteSecret(UserSecretReference secretReference);
+  void deleteSecret(@Nonnull UserSecretReference secretReference);
 }

polaris-core/src/main/java/org/apache/polaris/core/secrets/UserSecretsManagerFactory.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.core.secrets;
+
+import org.apache.polaris.core.context.RealmContext;
+
+/**
+ * Factory for creating {@link UserSecretsManager} instances.
+ *
+ * <p>Each {@link UserSecretsManager} instance is associated with a {@link RealmContext} and is
+ * responsible for managing the secrets for the user in that realm.
+ */
+public interface UserSecretsManagerFactory {
+  UserSecretsManager getOrCreateUserSecretsManager(RealmContext realmContext);
+}

quarkus/defaults/src/main/resources/application-it.properties

@@ -29,6 +29,8 @@ quarkus.management.port=0
 # polaris.persistence.type=in-memory-atomic
 polaris.persistence.type=in-memory
 
+polaris.secrets-manager.type=in-memory
+
 polaris.features.defaults."ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING"=false
 polaris.features.defaults."ALLOW_EXTERNAL_METADATA_FILE_LOCATION"=false
 polaris.features.defaults."ALLOW_OVERLAPPING_CATALOG_URLS"=true

quarkus/defaults/src/main/resources/application.properties

@@ -99,6 +99,8 @@ polaris.features.defaults."SUPPORTED_CATALOG_STORAGE_TYPES"=["S3","GCS","AZURE",
 # polaris.persistence.type=in-memory-atomic
 polaris.persistence.type=in-memory
 
+polaris.secrets-manager.type=in-memory
+
 polaris.file-io.type=default
 
 polaris.log.request-id-header-name=Polaris-Request-Id

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

@@ -45,6 +45,8 @@
 import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisEntityManager;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
+import org.apache.polaris.core.secrets.UserSecretsManager;
+import org.apache.polaris.core.secrets.UserSecretsManagerFactory;
 import org.apache.polaris.core.storage.cache.StorageCredentialCache;
 import org.apache.polaris.service.auth.ActiveRolesProvider;
 import org.apache.polaris.service.auth.Authenticator;
@@ -62,6 +64,7 @@
 import org.apache.polaris.service.quarkus.persistence.QuarkusPersistenceConfiguration;
 import org.apache.polaris.service.quarkus.ratelimiter.QuarkusRateLimiterFilterConfiguration;
 import org.apache.polaris.service.quarkus.ratelimiter.QuarkusTokenBucketConfiguration;
+import org.apache.polaris.service.quarkus.secrets.QuarkusSecretsManagerConfiguration;
 import org.apache.polaris.service.ratelimiter.RateLimiter;
 import org.apache.polaris.service.ratelimiter.TokenBucketFactory;
 import org.apache.polaris.service.task.TaskHandlerConfiguration;
@@ -150,6 +153,13 @@ public MetaStoreManagerFactory metaStoreManagerFactory(
     return metaStoreManagerFactories.select(Identifier.Literal.of(config.type())).get();
   }
 
+  @Produces
+  public UserSecretsManagerFactory userSecretsManagerFactory(
+      QuarkusSecretsManagerConfiguration config,
+      @Any Instance<UserSecretsManagerFactory> userSecretsManagerFactories) {
+    return userSecretsManagerFactories.select(Identifier.Literal.of(config.type())).get();
+  }
+
   /**
    * Eagerly initialize the in-memory default realm on startup, so that users can check the
    * credentials printed to stdout immediately.
@@ -218,6 +228,13 @@ public PolarisMetaStoreManager polarisMetaStoreManager(
     return metaStoreManagerFactory.getOrCreateMetaStoreManager(realmContext);
   }
 
+  @Produces
+  @RequestScoped
+  public UserSecretsManager userSecretsManager(
+      RealmContext realmContext, UserSecretsManagerFactory userSecretsManagerFactory) {
+    return userSecretsManagerFactory.getOrCreateUserSecretsManager(realmContext);
+  }
+
   @Produces
   @RequestScoped
   public BasePersistence polarisMetaStoreSession(

quarkus/service/src/main/java/org/apache/polaris/service/quarkus/secrets/QuarkusSecretsManagerConfiguration.java

@@ -0,0 +1,33 @@
+/*
+ * 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.secrets;
+
+import io.quarkus.runtime.annotations.StaticInitSafe;
+import io.smallrye.config.ConfigMapping;
+
+@StaticInitSafe
+@ConfigMapping(prefix = "polaris.secrets-manager")
+public interface QuarkusSecretsManagerConfiguration {
+
+  /**
+   * The type of the UserSecretsManagerFactory to use. This is the {@link
+   * org.apache.polaris.core.secrets.UserSecretsManagerFactory} identifier.
+   */
+  String type();
+}

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

@@ -53,6 +53,7 @@ private PolarisAdminService newTestAdminService(Set<String> activatedPrincipalRo
         callContext,
         entityManager,
         metaStoreManager,
+        userSecretsManager,
         securityContext(authenticatedPrincipal, activatedPrincipalRoles),
         polarisAuthorizer);
   }