Zendesk OAuth Flow and OAuth Java Refactors

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

@@ -21,12 +21,14 @@
  */
 public abstract class BaseOAuthConfig implements OAuthFlowImplementation {
 
-  private final ConfigRepository configRepository;
+  protected ConfigRepository configRepository;
 
   public BaseOAuthConfig(final ConfigRepository configRepository) {
     this.configRepository = configRepository;
   }
 
+  protected BaseOAuthConfig() {}
+
   protected JsonNode getSourceOAuthParamConfig(final UUID workspaceId, final UUID sourceDefinitionId) throws IOException, ConfigNotFoundException {
     try {
       final Optional<SourceOAuthParameter> param = MoreOAuthParameters.getSourceOAuthParameter(
@@ -60,7 +62,7 @@ protected JsonNode getDestinationOAuthParamConfig(final UUID workspaceId, final
    * Throws an exception if the client ID cannot be extracted. Subclasses should override this to
    * parse the config differently.
    *
-   * @return
+   * @return The configured Client ID used for this oauth flow
    */
   protected String getClientIdUnsafe(final JsonNode oauthConfig) {
     if (oauthConfig.get("client_id") != null) {
@@ -70,11 +72,25 @@ protected String getClientIdUnsafe(final JsonNode oauthConfig) {
     }
   }
 
+  /**
+   * Throws an exception if the client ID cannot be extracted. Subclasses should override this to
+   * parse the config differently.
+   *
+   * @return The configured subdomain used for this oauth flow
+   */
+  protected String getSubdomain(final JsonNode oauthConfig) {
+    if (oauthConfig.get("subdomain") != null) {
+      return oauthConfig.get("subdomain").asText();
+    } else {
+      throw new IllegalArgumentException("Undefined parameter 'client_id' necessary for the OAuth Flow.");
+    }
+  }
+
   /**
    * Throws an exception if the client secret cannot be extracted. Subclasses should override this to
    * parse the config differently.
    *
-   * @return
+   * @return The configured client secret for this OAuthFlow
    */
   protected String getClientSecretUnsafe(final JsonNode oauthConfig) {
     if (oauthConfig.get("client_secret") != null) {

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

@@ -6,38 +6,81 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.ImmutableMap;
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.config.persistence.ConfigRepository;
 import java.io.IOException;
+import java.lang.reflect.Type;
 import java.net.URI;
+import java.net.URISyntaxException;
 import java.net.URLEncoder;
 import java.net.http.HttpClient;
 import java.net.http.HttpClient.Version;
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.http.client.utils.URIBuilder;
 
 /*
  * Class implementing generic oAuth 2.0 flow.
  */
 public abstract class BaseOAuthFlow extends BaseOAuthConfig {
 
-  private final HttpClient httpClient;
+  /**
+   * Simple enum of content type strings and their respective encoding functions used for POSTing the
+   * access token request
+   */
+  public enum TOKEN_REQUEST_CONTENT_TYPE {
+
+    URL_ENCODED("application/x-www-form-urlencoded", BaseOAuthFlow::toUrlEncodedString),
+    JSON("application/json", BaseOAuthFlow::toJson);
+
+    String contentType;
+    Function<Map<String, String>, String> converter;
+
+    TOKEN_REQUEST_CONTENT_TYPE(String contentType, Function<Map<String, String>, String> converter) {
+      this.contentType = contentType;
+      this.converter = converter;
+    }
+
+  }
+
+  protected final HttpClient httpClient;
+  private final TOKEN_REQUEST_CONTENT_TYPE tokenReqContentType;
   private final Supplier<String> stateSupplier;
+  private UUID workspaceId;
 
   public BaseOAuthFlow(final ConfigRepository configRepository) {
     this(configRepository, HttpClient.newBuilder().version(Version.HTTP_1_1).build(), BaseOAuthFlow::generateRandomState);
   }
 
-  public BaseOAuthFlow(final ConfigRepository configRepository, final HttpClient httpClient, final Supplier<String> stateSupplier) {
+  public BaseOAuthFlow(ConfigRepository configRepository, TOKEN_REQUEST_CONTENT_TYPE tokenReqContentType) {
+    this(configRepository,
+        HttpClient.newBuilder().version(Version.HTTP_1_1).build(),
+        BaseOAuthFlow::generateRandomState,
+        tokenReqContentType);
+  }
+
+  public BaseOAuthFlow(ConfigRepository configRepository, HttpClient httpClient, Supplier<String> stateSupplier) {
+    this(configRepository, httpClient, stateSupplier, TOKEN_REQUEST_CONTENT_TYPE.URL_ENCODED);
+  }
+
+  public BaseOAuthFlow(ConfigRepository configRepository,
+                       HttpClient httpClient,
+                       Supplier<String> stateSupplier,
+                       TOKEN_REQUEST_CONTENT_TYPE tokenReqContentType) {
     super(configRepository);
     this.httpClient = httpClient;
     this.stateSupplier = stateSupplier;
+    this.tokenReqContentType = tokenReqContentType;
   }
 
   @Override
@@ -54,6 +97,31 @@ public String getDestinationConsentUrl(final UUID workspaceId, final UUID destin
     return formatConsentUrl(destinationDefinitionId, getClientIdUnsafe(oAuthParamConfig), redirectUrl);
   }
 
+  protected String formatConsentUrl(String clientId,
+                                    String redirectUrl,
+                                    String host,
+                                    String path,
+                                    String scope,
+                                    String responseType)
+      throws IOException {
+    final URIBuilder builder = new URIBuilder()
+        .setScheme("https")
+        .setHost(host)
+        .setPath(path)
+        // required
+        .addParameter("client_id", clientId)
+        .addParameter("redirect_uri", redirectUrl)
+        .addParameter("state", getState())
+        // optional
+        .addParameter("response_type", responseType)
+        .addParameter("scope", scope);
+    try {
+      return builder.build().toString();
+    } catch (URISyntaxException e) {
+      throw new IOException("Failed to format Consent URL for OAuth flow", e);
+    }
+  }
+
   /**
    * 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
@@ -84,7 +152,8 @@ public Map<String, Object> completeSourceOAuth(
         getClientIdUnsafe(oAuthParamConfig),
         getClientSecretUnsafe(oAuthParamConfig),
         extractCodeParameter(queryParams),
-        redirectUrl);
+        redirectUrl,
+        oAuthParamConfig);
   }
 
   @Override
@@ -98,20 +167,26 @@ public Map<String, Object> completeDestinationOAuth(final UUID workspaceId,
         getClientIdUnsafe(oAuthParamConfig),
         getClientSecretUnsafe(oAuthParamConfig),
         extractCodeParameter(queryParams),
-        redirectUrl);
+        redirectUrl, oAuthParamConfig);
   }
 
-  private Map<String, Object> completeOAuthFlow(final String clientId, final String clientSecret, final String authCode, final String redirectUrl)
+  private Map<String, Object> completeOAuthFlow(final String clientId,
+                                                final String clientSecret,
+                                                final String authCode,
+                                                final String redirectUrl,
+                                                JsonNode oAuthParamConfig)
       throws IOException {
+    var accessTokenUrl = getAccessTokenUrl(oAuthParamConfig);
     final HttpRequest request = HttpRequest.newBuilder()
-        .POST(HttpRequest.BodyPublishers.ofString(toUrlEncodedString(getAccessTokenQueryParameters(clientId, clientSecret, authCode, redirectUrl))))
-        .uri(URI.create(getAccessTokenUrl()))
-        .header("Content-Type", "application/x-www-form-urlencoded")
+        .POST(HttpRequest.BodyPublishers
+            .ofString(tokenReqContentType.converter.apply(getAccessTokenQueryParameters(clientId, clientSecret, authCode, redirectUrl))))
+        .uri(URI.create(accessTokenUrl))
+        .header("Content-Type", tokenReqContentType.contentType)
         .build();
     // TODO: Handle error response to report better messages
     try {
-      final HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());;
-      return extractRefreshToken(Jsons.deserialize(response.body()));
+      final HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
+      return extractRefreshToken(Jsons.deserialize(response.body()), accessTokenUrl);
     } catch (final InterruptedException e) {
       throw new IOException("Failed to complete OAuth flow", e);
     }
@@ -146,13 +221,20 @@ protected String extractCodeParameter(Map<String, Object> queryParams) throws IO
   /**
    * Returns the URL where to retrieve the access token from.
    */
-  protected abstract String getAccessTokenUrl();
+  protected abstract String getAccessTokenUrl(JsonNode oAuthParamConfig);
 
-  /**
-   * Once the auth code is exchange for a refresh token, the oauth flow implementation can extract and
-   * returns the values of fields to be used in the connector's configurations.
-   */
-  protected abstract Map<String, Object> extractRefreshToken(JsonNode data) throws IOException;
+  protected Map<String, Object> extractRefreshToken(final JsonNode data, String accessTokenUrl) throws IOException {
+    final Map<String, Object> result = new HashMap<>();
+    if (data.has("refresh_token")) {
+      result.put("refresh_token", data.get("refresh_token").asText());
+    } else if (data.has("access_token")) {
+      result.put("access_token", data.get("access_token").asText());
+    } else {
+      throw new IOException(String.format("Missing 'refresh_token' in query params from %s", accessTokenUrl));
+    }
+    return Map.of("credentials", result);
+
+  }
 
   private static String urlEncode(final String s) {
     try {
@@ -173,4 +255,10 @@ private static String toUrlEncodedString(final Map<String, String> body) {
     return result.toString();
   }
 
+  protected static String toJson(final Map<String, String> body) {
+    final Gson gson = new Gson();
+    Type gsonType = new TypeToken<Map<String, String>>() {}.getType();
+    return gson.toJson(body, gsonType);
+  }
+
 }

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

@@ -13,6 +13,7 @@
 import io.airbyte.oauth.flows.google.GoogleAnalyticsOAuthFlow;
 import io.airbyte.oauth.flows.google.GoogleSearchConsoleOAuthFlow;
 import java.util.Map;
+import java.util.UUID;
 
 public class OAuthImplementationFactory {
 
@@ -29,7 +30,7 @@ public OAuthImplementationFactory(final ConfigRepository configRepository) {
         .build();
   }
 
-  public OAuthFlowImplementation create(final String imageName) {
+  public OAuthFlowImplementation create(final String imageName, final UUID workspaceId) {
     if (OAUTH_FLOW_MAPPING.containsKey(imageName)) {
       return OAUTH_FLOW_MAPPING.get(imageName);
     } else {

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

@@ -7,7 +7,6 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
-import io.airbyte.commons.json.Jsons;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.oauth.BaseOAuthFlow;
 import java.io.IOException;
@@ -50,7 +49,7 @@ protected String formatConsentUrl(UUID definitionId, String clientId, String red
   }
 
   @Override
-  protected String getAccessTokenUrl() {
+  protected String getAccessTokenUrl(JsonNode oAuthParamConfig) {
     return ACCESS_TOKEN_URL;
   }
 
@@ -62,15 +61,4 @@ protected Map<String, String> getAccessTokenQueryParameters(String clientId, Str
         .build();
   }
 
-  @Override
-  protected Map<String, Object> extractRefreshToken(JsonNode data) throws IOException {
-    System.out.println(Jsons.serialize(data));
-    if (data.has("refresh_token")) {
-      final String refreshToken = data.get("refresh_token").asText();
-      return Map.of("credentials", Map.of("refresh_token", refreshToken));
-    } else {
-      throw new IOException(String.format("Missing 'refresh_token' in query params from %s", ACCESS_TOKEN_URL));
-    }
-  }
-
 }

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

@@ -54,12 +54,12 @@ protected String formatConsentUrl(final UUID definitionId, final String clientId
   }
 
   @Override
-  protected String getAccessTokenUrl() {
+  protected String getAccessTokenUrl(JsonNode oAuthParamConfig) {
     return ACCESS_TOKEN_URL;
   }
 
   @Override
-  protected Map<String, Object> extractRefreshToken(final JsonNode data) throws IOException {
+  protected Map<String, Object> extractRefreshToken(final JsonNode data, String accessTokenUrl) throws IOException {
     // Facebook does not have refresh token but calls it "long lived access token" instead:
     // see https://developers.facebook.com/docs/facebook-login/access-tokens/refreshing
     if (data.has("access_token")) {

airbyte-oauth/src/main/java/io/airbyte/oauth/flows/google/GoogleAdsOAuthFlow.java

@@ -8,9 +8,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import io.airbyte.config.persistence.ConfigRepository;
-import java.io.IOException;
 import java.net.http.HttpClient;
-import java.util.Map;
 import java.util.function.Supplier;
 
 public class GoogleAdsOAuthFlow extends GoogleOAuthFlow {
@@ -46,10 +44,4 @@ protected String getClientSecretUnsafe(final JsonNode config) {
     return super.getClientSecretUnsafe(config.get("credentials"));
   }
 
-  @Override
-  protected Map<String, Object> extractRefreshToken(final JsonNode data) throws IOException {
-    // the config object containing refresh token is nested inside the "credentials" object
-    return Map.of("credentials", super.extractRefreshToken(data));
-  }
-
 }

airbyte-oauth/src/main/java/io/airbyte/oauth/flows/google/GoogleAnalyticsOAuthFlow.java

@@ -8,9 +8,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import io.airbyte.config.persistence.ConfigRepository;
-import java.io.IOException;
 import java.net.http.HttpClient;
-import java.util.Map;
 import java.util.function.Supplier;
 
 public class GoogleAnalyticsOAuthFlow extends GoogleOAuthFlow {
@@ -45,10 +43,4 @@ protected String getClientSecretUnsafe(final JsonNode config) {
     return super.getClientSecretUnsafe(config.get("credentials"));
   }
 
-  @Override
-  protected Map<String, Object> extractRefreshToken(final JsonNode data) throws IOException {
-    // the config object containing refresh token is nested inside the "credentials" object
-    return Map.of("credentials", super.extractRefreshToken(data));
-  }
-
 }

airbyte-oauth/src/main/java/io/airbyte/oauth/flows/google/GoogleOAuthFlow.java

@@ -12,7 +12,6 @@
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.net.http.HttpClient;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 import java.util.function.Supplier;
@@ -64,7 +63,7 @@ protected String formatConsentUrl(final UUID definitionId, final String clientId
   protected abstract String getScope();
 
   @Override
-  protected String getAccessTokenUrl() {
+  protected String getAccessTokenUrl(JsonNode oAuthParamConfig) {
     return ACCESS_TOKEN_URL;
   }
 
@@ -82,18 +81,4 @@ protected Map<String, String> getAccessTokenQueryParameters(final String clientI
         .build();
   }
 
-  @Override
-  protected Map<String, Object> extractRefreshToken(final JsonNode data) throws IOException {
-    final Map<String, Object> result = new HashMap<>();
-    if (data.has("access_token")) {
-      result.put("access_token", data.get("access_token").asText());
-    }
-    if (data.has("refresh_token")) {
-      result.put("refresh_token", data.get("refresh_token").asText());
-    } else {
-      throw new IOException(String.format("Missing 'refresh_token' in query params from %s", ACCESS_TOKEN_URL));
-    }
-    return result;
-  }
-
 }