Implement connector config dependency for OAuth consent URL (#7983)

* Refactor OAuth consent flow with new API

* Use input connector configuration in getConsentURL for OAuth flow

* Instance-wide params should only be injected if a connector will be using oauth (#8018)

* Add test for nested parameters

airbyte-commons/src/main/java/io/airbyte/commons/json/Jsons.java

@@ -141,6 +141,10 @@ public static JsonNode navigateTo(JsonNode node, final List<String> keys) {
     return node;
   }
 
+  public static void replaceNestedValue(final JsonNode json, final List<String> keys, final JsonNode replacement) {
+    replaceNested(json, keys, (node, finalKey) -> node.put(finalKey, replacement));
+  }
+
   public static void replaceNestedString(final JsonNode json, final List<String> keys, final String replacement) {
     replaceNested(json, keys, (node, finalKey) -> node.put(finalKey, replacement));
   }

airbyte-oauth/src/main/java/io/airbyte/oauth/BaseOAuth2Flow.java

@@ -12,6 +12,8 @@
 import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.protocol.models.OAuthConfigSpecification;
+import io.airbyte.validation.json.JsonSchemaValidator;
+import io.airbyte.validation.json.JsonValidationException;
 import java.io.IOException;
 import java.lang.reflect.Type;
 import java.net.URI;
@@ -75,25 +77,45 @@ public BaseOAuth2Flow(final ConfigRepository configRepository,
   }
 
   @Override
-  public String getSourceConsentUrl(final UUID workspaceId, final UUID sourceDefinitionId, final String redirectUrl)
-      throws IOException, ConfigNotFoundException {
+  public String getSourceConsentUrl(final UUID workspaceId,
+                                    final UUID sourceDefinitionId,
+                                    final String redirectUrl,
+                                    final JsonNode inputOAuthConfiguration,
+                                    final OAuthConfigSpecification oAuthConfigSpecification)
+      throws IOException, ConfigNotFoundException, JsonValidationException {
+    validateInputOAuthConfiguration(oAuthConfigSpecification, inputOAuthConfiguration);
     final JsonNode oAuthParamConfig = getSourceOAuthParamConfig(workspaceId, sourceDefinitionId);
-    return formatConsentUrl(sourceDefinitionId, getClientIdUnsafe(oAuthParamConfig), redirectUrl);
+    return formatConsentUrl(sourceDefinitionId, getClientIdUnsafe(oAuthParamConfig), redirectUrl, inputOAuthConfiguration);
   }
 
   @Override
-  public String getDestinationConsentUrl(final UUID workspaceId, final UUID destinationDefinitionId, final String redirectUrl)
-      throws IOException, ConfigNotFoundException {
+  public String getDestinationConsentUrl(final UUID workspaceId,
+                                         final UUID destinationDefinitionId,
+                                         final String redirectUrl,
+                                         final JsonNode inputOAuthConfiguration,
+                                         final OAuthConfigSpecification oAuthConfigSpecification)
+      throws IOException, ConfigNotFoundException, JsonValidationException {
+    validateInputOAuthConfiguration(oAuthConfigSpecification, inputOAuthConfiguration);
     final JsonNode oAuthParamConfig = getDestinationOAuthParamConfig(workspaceId, destinationDefinitionId);
-    return formatConsentUrl(destinationDefinitionId, getClientIdUnsafe(oAuthParamConfig), redirectUrl);
+    return formatConsentUrl(destinationDefinitionId, getClientIdUnsafe(oAuthParamConfig), redirectUrl, inputOAuthConfiguration);
   }
 
   /**
    * Depending on the OAuth flow implementation, the URL to grant user's consent may differ,
    * especially in the query parameters to be provided. This function should generate such consent URL
    * accordingly.
+   *
+   * @param definitionId The configured definition ID of this client
+   * @param clientId The configured client ID
+   * @param redirectUrl the redirect URL
+   * @param inputOAuthConfiguration any configuration property from connector necessary for this OAuth
+   *        Flow
    */
-  protected abstract String formatConsentUrl(UUID definitionId, String clientId, String redirectUrl) throws IOException;
+  protected abstract String formatConsentUrl(final UUID definitionId,
+                                             final String clientId,
+                                             final String redirectUrl,
+                                             final JsonNode inputOAuthConfiguration)
+      throws IOException;
 
   private static String generateRandomState() {
     return RandomStringUtils.randomAlphanumeric(7);
@@ -151,8 +173,9 @@ public Map<String, Object> completeSourceOAuth(final UUID workspaceId,
                                                  final String redirectUrl,
                                                  final JsonNode inputOAuthConfiguration,
                                                  final OAuthConfigSpecification oAuthConfigSpecification)
-      throws IOException, ConfigNotFoundException {
-    final JsonNode oAuthParamConfig = getDestinationOAuthParamConfig(workspaceId, sourceDefinitionId);
+      throws IOException, ConfigNotFoundException, JsonValidationException {
+    validateInputOAuthConfiguration(oAuthConfigSpecification, inputOAuthConfiguration);
+    final JsonNode oAuthParamConfig = getSourceOAuthParamConfig(workspaceId, sourceDefinitionId);
     return formatOAuthOutput(
         oAuthParamConfig,
         completeOAuthFlow(
@@ -171,7 +194,8 @@ public Map<String, Object> completeDestinationOAuth(final UUID workspaceId,
                                                       final String redirectUrl,
                                                       final JsonNode inputOAuthConfiguration,
                                                       final OAuthConfigSpecification oAuthConfigSpecification)
-      throws IOException, ConfigNotFoundException {
+      throws IOException, ConfigNotFoundException, JsonValidationException {
+    validateInputOAuthConfiguration(oAuthConfigSpecification, inputOAuthConfiguration);
     final JsonNode oAuthParamConfig = getDestinationOAuthParamConfig(workspaceId, destinationDefinitionId);
     return formatOAuthOutput(
         oAuthParamConfig,
@@ -260,6 +284,14 @@ public List<String> getDefaultOAuthOutputPath() {
     return List.of("credentials");
   }
 
+  private static void validateInputOAuthConfiguration(final OAuthConfigSpecification oauthConfigSpecification, final JsonNode inputOAuthConfiguration)
+      throws JsonValidationException {
+    if (oauthConfigSpecification != null && oauthConfigSpecification.getOauthUserInputFromConnectorConfigSpecification() != null) {
+      final JsonSchemaValidator validator = new JsonSchemaValidator();
+      validator.ensure(oauthConfigSpecification.getOauthUserInputFromConnectorConfigSpecification(), inputOAuthConfiguration);
+    }
+  }
+
   private static String urlEncode(final String s) {
     try {
       return URLEncoder.encode(s, StandardCharsets.UTF_8);

airbyte-oauth/src/main/java/io/airbyte/oauth/BaseOAuthFlow.java

@@ -8,26 +8,31 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableMap.Builder;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.map.MoreMaps;
 import io.airbyte.config.ConfigSchema;
 import io.airbyte.config.DestinationOAuthParameter;
 import io.airbyte.config.SourceOAuthParameter;
 import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.protocol.models.OAuthConfigSpecification;
+import io.airbyte.validation.json.JsonSchemaValidator;
 import io.airbyte.validation.json.JsonValidationException;
 import java.io.IOException;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.function.BiConsumer;
 
 /**
  * Abstract Class implementing common base methods for managing oAuth config (instance-wide) and
  * oAuth specifications
  */
 public abstract class BaseOAuthFlow implements OAuthFlowImplementation {
 
+  public static final String PROPERTIES = "properties";
   private final ConfigRepository configRepository;
 
   public BaseOAuthFlow(final ConfigRepository configRepository) {
@@ -56,7 +61,7 @@ protected JsonNode getDestinationOAuthParamConfig(final UUID workspaceId, final
       final Optional<DestinationOAuthParameter> param = MoreOAuthParameters.getDestinationOAuthParameter(
           configRepository.listDestinationOAuthParam().stream(), workspaceId, destinationDefinitionId);
       if (param.isPresent()) {
-        // TODO: if we write a flyway migration to flatten persisted configs in db, we don't need to flatten
+        // TODO: if we write a migration to flatten persisted configs in db, we don't need to flatten
         // here see https://github.com/airbytehq/airbyte/issues/7624
         return MoreOAuthParameters.flattenOAuthConfig(param.get().getConfiguration());
       } else {
@@ -107,7 +112,6 @@ protected Map<String, Object> formatOAuthOutput(final JsonNode oAuthParamConfig,
                                                   final Map<String, Object> oauthOutput,
                                                   final List<String> outputPath) {
     Map<String, Object> result = new HashMap<>(oauthOutput);
-    // inject masked params outputs
     for (final String key : Jsons.keys(oAuthParamConfig)) {
       result.put(key, MoreOAuthParameters.SECRET_MASK);
     }
@@ -124,21 +128,45 @@ protected Map<String, Object> formatOAuthOutput(final JsonNode oAuthParamConfig,
    */
   protected Map<String, Object> formatOAuthOutput(final JsonNode oAuthParamConfig,
                                                   final Map<String, Object> completeOAuthFlow,
-                                                  final OAuthConfigSpecification oAuthConfigSpecification) {
-    final Builder<String, Object> outputs = ImmutableMap.builder();
-    // inject masked params outputs
-    for (final String key : Jsons.keys(oAuthParamConfig)) {
-      if (oAuthConfigSpecification.getCompleteOauthServerOutputSpecification().has(key)) {
-        outputs.put(key, MoreOAuthParameters.SECRET_MASK);
-      }
-    }
-    // collect oauth result outputs
-    for (final String key : completeOAuthFlow.keySet()) {
-      if (oAuthConfigSpecification.getCompleteOauthOutputSpecification().has(key)) {
-        outputs.put(key, completeOAuthFlow.get(key));
+                                                  final OAuthConfigSpecification oAuthConfigSpecification)
+      throws JsonValidationException {
+    final JsonSchemaValidator validator = new JsonSchemaValidator();
+
+    final Map<String, Object> oAuthOutputs = formatOAuthOutput(
+        validator,
+        oAuthConfigSpecification.getCompleteOauthOutputSpecification(),
+        completeOAuthFlow.keySet(),
+        (resultMap, key) -> resultMap.put(key, completeOAuthFlow.get(key)));
+
+    final Map<String, Object> oAuthServerOutputs = formatOAuthOutput(
+        validator,
+        oAuthConfigSpecification.getCompleteOauthServerOutputSpecification(),
+        Jsons.keys(oAuthParamConfig),
+        // TODO secrets should be masked with the correct type
+        // https://github.com/airbytehq/airbyte/issues/5990
+        // In the short-term this is not world-ending as all secret fields are currently strings
+        (resultMap, key) -> resultMap.put(key, MoreOAuthParameters.SECRET_MASK));
+
+    return MoreMaps.merge(oAuthServerOutputs, oAuthOutputs);
+  }
+
+  private static Map<String, Object> formatOAuthOutput(final JsonSchemaValidator validator,
+                                                       final JsonNode outputSchema,
+                                                       final Collection<String> keys,
+                                                       final BiConsumer<Builder<String, Object>, String> replacement)
+      throws JsonValidationException {
+    Map<String, Object> result = Map.of();
+    if (outputSchema != null && outputSchema.has(PROPERTIES)) {
+      final Builder<String, Object> mapBuilder = ImmutableMap.builder();
+      for (final String key : keys) {
+        if (outputSchema.get(PROPERTIES).has(key)) {
+          replacement.accept(mapBuilder, key);
+        }
       }
+      result = mapBuilder.build();
+      validator.ensure(outputSchema, Jsons.jsonNode(result));
     }
-    return outputs.build();
+    return result;
   }
 
   /**

airbyte-oauth/src/main/java/io/airbyte/oauth/MoreOAuthParameters.java

@@ -74,42 +74,26 @@ private static ObjectNode flattenOAuthConfig(final ObjectNode flatConfig, final
   }
 
   public static JsonNode mergeJsons(final ObjectNode mainConfig, final ObjectNode fromConfig) {
-    return mergeJsons(mainConfig, fromConfig, null);
-  }
-
-  public static JsonNode mergeJsons(final ObjectNode mainConfig, final ObjectNode fromConfig, final JsonNode maskedValue) {
     for (final String key : Jsons.keys(fromConfig)) {
       if (fromConfig.get(key).getNodeType() == OBJECT) {
         // nested objects are merged rather than overwrite the contents of the equivalent object in config
         if (mainConfig.get(key) == null) {
-          mergeJsons(mainConfig.putObject(key), (ObjectNode) fromConfig.get(key), maskedValue);
+          mergeJsons(mainConfig.putObject(key), (ObjectNode) fromConfig.get(key));
         } else if (mainConfig.get(key).getNodeType() == OBJECT) {
-          mergeJsons((ObjectNode) mainConfig.get(key), (ObjectNode) fromConfig.get(key), maskedValue);
+          mergeJsons((ObjectNode) mainConfig.get(key), (ObjectNode) fromConfig.get(key));
         } else {
           throw new IllegalStateException("Can't merge an object node into a non-object node!");
         }
       } else {
-        if (maskedValue != null && !maskedValue.isNull()) {
-          LOGGER.debug(String.format("Masking instance wide parameter %s in config", key));
-          mainConfig.set(key, maskedValue);
-        } else {
-          if (!mainConfig.has(key) || isSecretMask(mainConfig.get(key).asText())) {
-            LOGGER.debug(String.format("injecting instance wide parameter %s into config", key));
-            mainConfig.set(key, fromConfig.get(key));
-          }
+        if (!mainConfig.has(key) || isSecretMask(mainConfig.get(key).asText())) {
+          LOGGER.debug(String.format("injecting instance wide parameter %s into config", key));
+          mainConfig.set(key, fromConfig.get(key));
         }
       }
     }
     return mainConfig;
   }
 
-  public static JsonNode getSecretMask() {
-    // TODO secrets should be masked with the correct type
-    // https://github.com/airbytehq/airbyte/issues/5990
-    // In the short-term this is not world-ending as all secret fields are currently strings
-    return Jsons.jsonNode(SECRET_MASK);
-  }
-
   private static boolean isSecretMask(final String input) {
     return Strings.isNullOrEmpty(input.replaceAll("\\*", ""));
   }

airbyte-oauth/src/main/java/io/airbyte/oauth/OAuthFlowImplementation.java

@@ -7,15 +7,26 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.protocol.models.OAuthConfigSpecification;
+import io.airbyte.validation.json.JsonValidationException;
 import java.io.IOException;
 import java.util.Map;
 import java.util.UUID;
 
 public interface OAuthFlowImplementation {
 
-  String getSourceConsentUrl(UUID workspaceId, UUID sourceDefinitionId, String redirectUrl) throws IOException, ConfigNotFoundException;
+  String getSourceConsentUrl(UUID workspaceId,
+                             UUID sourceDefinitionId,
+                             String redirectUrl,
+                             JsonNode inputOAuthConfiguration,
+                             OAuthConfigSpecification oauthConfigSpecification)
+      throws IOException, ConfigNotFoundException, JsonValidationException;
 
-  String getDestinationConsentUrl(UUID workspaceId, UUID destinationDefinitionId, String redirectUrl) throws IOException, ConfigNotFoundException;
+  String getDestinationConsentUrl(UUID workspaceId,
+                                  UUID destinationDefinitionId,
+                                  String redirectUrl,
+                                  JsonNode inputOAuthConfiguration,
+                                  OAuthConfigSpecification oauthConfigSpecification)
+      throws IOException, ConfigNotFoundException, JsonValidationException;
 
   @Deprecated
   Map<String, Object> completeSourceOAuth(UUID workspaceId, UUID sourceDefinitionId, Map<String, Object> queryParams, String redirectUrl)
@@ -27,7 +38,7 @@ Map<String, Object> completeSourceOAuth(UUID workspaceId,
                                           String redirectUrl,
                                           JsonNode inputOAuthConfiguration,
                                           OAuthConfigSpecification oauthConfigSpecification)
-      throws IOException, ConfigNotFoundException;
+      throws IOException, ConfigNotFoundException, JsonValidationException;
 
   @Deprecated
   Map<String, Object> completeDestinationOAuth(UUID workspaceId, UUID destinationDefinitionId, Map<String, Object> queryParams, String redirectUrl)
@@ -39,6 +50,6 @@ Map<String, Object> completeDestinationOAuth(UUID workspaceId,
                                                String redirectUrl,
                                                JsonNode inputOAuthConfiguration,
                                                OAuthConfigSpecification oAuthConfigSpecification)
-      throws IOException, ConfigNotFoundException;
+      throws IOException, ConfigNotFoundException, JsonValidationException;
 
 }

airbyte-oauth/src/main/java/io/airbyte/oauth/flows/AsanaOAuthFlow.java

@@ -4,6 +4,7 @@
 
 package io.airbyte.oauth.flows;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.config.persistence.ConfigRepository;
@@ -34,7 +35,11 @@ public AsanaOAuthFlow(final ConfigRepository configRepository, final HttpClient
   }
 
   @Override
-  protected String formatConsentUrl(final UUID definitionId, final String clientId, final String redirectUrl) throws IOException {
+  protected String formatConsentUrl(final UUID definitionId,
+                                    final String clientId,
+                                    final String redirectUrl,
+                                    final JsonNode inputOAuthConfiguration)
+      throws IOException {
     try {
       return new URIBuilder(AUTHORIZE_URL)
           .addParameter("client_id", clientId)

airbyte-oauth/src/main/java/io/airbyte/oauth/flows/GithubOAuthFlow.java

@@ -35,7 +35,11 @@ public GithubOAuthFlow(final ConfigRepository configRepository, final HttpClient
   }
 
   @Override
-  protected String formatConsentUrl(final UUID definitionId, final String clientId, final String redirectUrl) throws IOException {
+  protected String formatConsentUrl(final UUID definitionId,
+                                    final String clientId,
+                                    final String redirectUrl,
+                                    final JsonNode inputOAuthConfiguration)
+      throws IOException {
     try {
       // No scope means read-only access to public information
       // https://docs.github.com/en/developers/apps/building-oauth-apps/scopes-for-oauth-apps#available-scopes

airbyte-oauth/src/main/java/io/airbyte/oauth/flows/HubspotOAuthFlow.java

@@ -4,6 +4,7 @@
 
 package io.airbyte.oauth.flows;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.oauth.BaseOAuth2Flow;
@@ -27,17 +28,12 @@ public HubspotOAuthFlow(final ConfigRepository configRepository, final HttpClien
     super(configRepository, httpClient, stateSupplier, TOKEN_REQUEST_CONTENT_TYPE.JSON);
   }
 
-  /**
-   * Depending on the OAuth flow implementation, the URL to grant user's consent may differ,
-   * especially in the query parameters to be provided. This function should generate such consent URL
-   * accordingly.
-   *
-   * @param definitionId The configured definition ID of this client
-   * @param clientId The configured client ID
-   * @param redirectUrl the redirect URL
-   */
   @Override
-  protected String formatConsentUrl(final UUID definitionId, final String clientId, final String redirectUrl) throws IOException {
+  protected String formatConsentUrl(final UUID definitionId,
+                                    final String clientId,
+                                    final String redirectUrl,
+                                    final JsonNode inputOAuthConfiguration)
+      throws IOException {
     try {
       return new URIBuilder(AUTHORIZE_URL)
           .addParameter("client_id", clientId)