Implement protocol change for OAuth outputs (airbytehq#7917)

* Change OAuth API

* Change protocol for new OAuthConfigSpecification

* Refactor OAuth classes and tests

* Remove webbackend source/destination creation

* Change from webback to normal API

* Implement new protocol change with OAuth specs

Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

* format

* format

Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

airbyte-api/src/main/openapi/config.yaml

@@ -1494,48 +1494,6 @@ paths:
                 $ref: "#/components/schemas/WebBackendConnectionReadList"
         "422":
           $ref: "#/components/responses/InvalidInputResponse"
-  /v1/web_backend/sources/create:
-    post:
-      tags:
-        - web_backend
-      summary: Create a source
-      operationId: webBackendCreateSource
-      requestBody:
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/SourceCreate"
-        required: true
-      responses:
-        "200":
-          description: Successful operation
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/SourceRead"
-        "422":
-          $ref: "#/components/responses/InvalidInputResponse"
-  /v1/web_backend/destinations/create:
-    post:
-      tags:
-        - web_backend
-      summary: Create a destination
-      operationId: webBackendCreateDestination
-      requestBody:
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/DestinationCreate"
-        required: true
-      responses:
-        "200":
-          description: Successful operation
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/DestinationRead"
-        "422":
-          $ref: "#/components/responses/InvalidInputResponse"
   /v1/jobs/list:
     post:
       tags:

airbyte-cdk/python/airbyte_cdk/models/airbyte_protocol.py

@@ -114,8 +114,8 @@ class AuthSpecification(BaseModel):
 
 
 class AuthFlowType(Enum):
-    oauth1_0 = "oauth1.0"
     oauth2_0 = "oauth2.0"
+    oauth1_0 = "oauth1.0"
 
 
 class OAuthConfigSpecification(BaseModel):

airbyte-integrations/bases/airbyte-protocol/airbyte_protocol/models/airbyte_protocol.py

@@ -114,8 +114,8 @@ class AuthSpecification(BaseModel):
 
 
 class AuthFlowType(Enum):
-    oauth1_0 = "oauth1.0"
     oauth2_0 = "oauth2.0"
+    oauth1_0 = "oauth1.0"
 
 
 class OAuthConfigSpecification(BaseModel):

airbyte-oauth/build.gradle

@@ -7,5 +7,6 @@ dependencies {
     implementation project(':airbyte-config:models')
     implementation project(':airbyte-config:persistence')
     implementation project(':airbyte-json-validation')
+    implementation project(':airbyte-protocol:models')
     testImplementation project(':airbyte-oauth')
 }

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

@@ -11,6 +11,7 @@
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.config.persistence.ConfigRepository;
+import io.airbyte.protocol.models.OAuthConfigSpecification;
 import java.io.IOException;
 import java.lang.reflect.Type;
 import java.net.URI;
@@ -106,6 +107,7 @@ protected String getState() {
   }
 
   @Override
+  @Deprecated
   public Map<String, Object> completeSourceOAuth(final UUID workspaceId,
                                                  final UUID sourceDefinitionId,
                                                  final Map<String, Object> queryParams,
@@ -124,6 +126,7 @@ public Map<String, Object> completeSourceOAuth(final UUID workspaceId,
   }
 
   @Override
+  @Deprecated
   public Map<String, Object> completeDestinationOAuth(final UUID workspaceId,
                                                       final UUID destinationDefinitionId,
                                                       final Map<String, Object> queryParams,
@@ -141,6 +144,46 @@ public Map<String, Object> completeDestinationOAuth(final UUID workspaceId,
         getDefaultOAuthOutputPath());
   }
 
+  @Override
+  public Map<String, Object> completeSourceOAuth(final UUID workspaceId,
+                                                 final UUID sourceDefinitionId,
+                                                 final Map<String, Object> queryParams,
+                                                 final String redirectUrl,
+                                                 final JsonNode inputOAuthConfiguration,
+                                                 final OAuthConfigSpecification oAuthConfigSpecification)
+      throws IOException, ConfigNotFoundException {
+    final JsonNode oAuthParamConfig = getDestinationOAuthParamConfig(workspaceId, sourceDefinitionId);
+    return formatOAuthOutput(
+        oAuthParamConfig,
+        completeOAuthFlow(
+            getClientIdUnsafe(oAuthParamConfig),
+            getClientSecretUnsafe(oAuthParamConfig),
+            extractCodeParameter(queryParams),
+            redirectUrl,
+            oAuthParamConfig),
+        oAuthConfigSpecification);
+  }
+
+  @Override
+  public Map<String, Object> completeDestinationOAuth(final UUID workspaceId,
+                                                      final UUID destinationDefinitionId,
+                                                      final Map<String, Object> queryParams,
+                                                      final String redirectUrl,
+                                                      final JsonNode inputOAuthConfiguration,
+                                                      final OAuthConfigSpecification oAuthConfigSpecification)
+      throws IOException, ConfigNotFoundException {
+    final JsonNode oAuthParamConfig = getDestinationOAuthParamConfig(workspaceId, destinationDefinitionId);
+    return formatOAuthOutput(
+        oAuthParamConfig,
+        completeOAuthFlow(
+            getClientIdUnsafe(oAuthParamConfig),
+            getClientSecretUnsafe(oAuthParamConfig),
+            extractCodeParameter(queryParams),
+            redirectUrl,
+            oAuthParamConfig),
+        oAuthConfigSpecification);
+  }
+
   protected Map<String, Object> completeOAuthFlow(final String clientId,
                                                   final String clientSecret,
                                                   final String authCode,
@@ -212,7 +255,8 @@ protected Map<String, Object> extractOAuthOutput(final JsonNode data, final Stri
   }
 
   @Override
-  protected List<String> getDefaultOAuthOutputPath() {
+  @Deprecated
+  public List<String> getDefaultOAuthOutputPath() {
     return List.of("credentials");
   }
 

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

@@ -5,22 +5,26 @@
 package io.airbyte.oauth;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMap.Builder;
+import io.airbyte.commons.json.Jsons;
 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.JsonValidationException;
 import java.io.IOException;
-import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
 
 /**
- * Abstract Class implementing common base methods for managing: - oAuth config (instance-wide)
- * parameters - oAuth specifications
+ * Abstract Class implementing common base methods for managing oAuth config (instance-wide) and
+ * oAuth specifications
  */
 public abstract class BaseOAuthFlow implements OAuthFlowImplementation {
 
@@ -35,7 +39,9 @@ protected JsonNode getSourceOAuthParamConfig(final UUID workspaceId, final UUID
       final Optional<SourceOAuthParameter> param = MoreOAuthParameters.getSourceOAuthParameter(
           configRepository.listSourceOAuthParam().stream(), workspaceId, sourceDefinitionId);
       if (param.isPresent()) {
-        return param.get().getConfiguration();
+        // TODO: if we write a flyway 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 {
         throw new ConfigNotFoundException(ConfigSchema.SOURCE_OAUTH_PARAM, "Undefined OAuth Parameter.");
       }
@@ -50,7 +56,9 @@ protected JsonNode getDestinationOAuthParamConfig(final UUID workspaceId, final
       final Optional<DestinationOAuthParameter> param = MoreOAuthParameters.getDestinationOAuthParameter(
           configRepository.listDestinationOAuthParam().stream(), workspaceId, destinationDefinitionId);
       if (param.isPresent()) {
-        return param.get().getConfiguration();
+        // TODO: if we write a flyway 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 {
         throw new ConfigNotFoundException(ConfigSchema.DESTINATION_OAUTH_PARAM, "Undefined OAuth Parameter.");
       }
@@ -66,17 +74,7 @@ protected JsonNode getDestinationOAuthParamConfig(final UUID workspaceId, final
    * @return The configured Client ID used for this oauth flow
    */
   protected String getClientIdUnsafe(final JsonNode oauthConfig) {
-    final List<String> path = new ArrayList<>(getDefaultOAuthOutputPath());
-    path.add("client_id");
-    JsonNode result = oauthConfig;
-    for (final String node : path) {
-      if (result.get(node) != null) {
-        result = result.get(node);
-      } else {
-        throw new IllegalArgumentException(String.format("Undefined parameter '%s' necessary for the OAuth Flow.", String.join(".", path)));
-      }
-    }
-    return result.asText();
+    return getConfigValueUnsafe(oauthConfig, "client_id");
   }
 
   /**
@@ -86,40 +84,70 @@ protected String getClientIdUnsafe(final JsonNode oauthConfig) {
    * @return The configured client secret for this OAuthFlow
    */
   protected String getClientSecretUnsafe(final JsonNode oauthConfig) {
-    final List<String> path = new ArrayList<>(getDefaultOAuthOutputPath());
-    path.add("client_secret");
-    JsonNode result = oauthConfig;
-    for (final String node : path) {
-      if (result.get(node) != null) {
-        result = result.get(node);
-      } else {
-        throw new IllegalArgumentException(String.format("Undefined parameter '%s' necessary for the OAuth Flow.", String.join(".", path)));
-      }
+    return getConfigValueUnsafe(oauthConfig, "client_secret");
+  }
+
+  private static String getConfigValueUnsafe(final JsonNode oauthConfig, final String fieldName) {
+    if (oauthConfig.get(fieldName) != null) {
+      return oauthConfig.get(fieldName).asText();
+    } else {
+      throw new IllegalArgumentException(String.format("Undefined parameter '%s' necessary for the OAuth Flow.", fieldName));
     }
-    return result.asText();
   }
 
   /**
    * completeOAuth calls should output a flat map of fields produced by the oauth flow to be forwarded
-   * back to the connector config. This function is in charge of formatting such flat map of fields
-   * into nested Map accordingly to follow the expected @param outputPath.
-   *
+   * back to the connector config. This @deprecated function is used when the connector's oauth
+   * specifications are unknown. So it ends up using hard-coded output path in the OAuth Flow
+   * implementation instead of relying on the connector's specification to determine where the outputs
+   * should be stored.
    */
+  @Deprecated
   protected Map<String, Object> formatOAuthOutput(final JsonNode oAuthParamConfig,
                                                   final Map<String, Object> oauthOutput,
                                                   final List<String> outputPath) {
-    Map<String, Object> result = oauthOutput;
+    Map<String, Object> result = new HashMap<>(oauthOutput);
+    // inject masked params outputs
+    for (final String key : Jsons.keys(oAuthParamConfig)) {
+      result.put(key, MoreOAuthParameters.SECRET_MASK);
+    }
     for (final String node : outputPath) {
       result = Map.of(node, result);
     }
-    // TODO chris to implement injection of oAuthParamConfig in outputs
     return result;
   }
 
+  /**
+   * completeOAuth calls should output a flat map of fields produced by the oauth flow to be forwarded
+   * back to the connector config. This function follows the connector's oauth specifications of which
+   * outputs are expected and filters them accordingly.
+   */
+  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));
+      }
+    }
+    return outputs.build();
+  }
+
   /**
    * This function should be redefined in each OAuthFlow implementation to isolate such "hardcoded"
-   * values.
+   * values. It is being @deprecated because the output path should not be "hard-coded" in the OAuth
+   * flow implementation classes anymore but will be specified as part of the OAuth Specification
+   * object
    */
-  protected abstract List<String> getDefaultOAuthOutputPath();
+  @Deprecated
+  public abstract List<String> getDefaultOAuthOutputPath();
 
 }