Slack OAuthFlow Backend implementation (#7328)

* Slack OAuthFlow

* Slack OAuth Unit and Integration Tests

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

Co-authored-by: Christophe Duong <christophe.duong@gmail.com>

* Fixes after merge for Slack OAuthFlow

Co-authored-by: Christophe Duong <christophe.duong@gmail.com>

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

@@ -34,6 +34,7 @@ public OAuthImplementationFactory(final ConfigRepository configRepository, final
         .put("airbyte/source-google-sheets", new GoogleSheetsOAuthFlow(configRepository, httpClient))
         .put("airbyte/source-instagram", new InstagramOAuthFlow(configRepository, httpClient))
         .put("airbyte/source-salesforce", new SalesforceOAuthFlow(configRepository, httpClient))
+        .put("airbyte/source-slack", new SlackOAuthFlow(configRepository, httpClient))
         .put("airbyte/source-surveymonkey", new SurveymonkeyOAuthFlow(configRepository, httpClient))
         .put("airbyte/source-trello", new TrelloOAuthFlow(configRepository, httpClient))
         .put("airbyte/source-hubspot", new HubspotOAuthFlow(configRepository, httpClient))

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.oauth.flows;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.airbyte.config.persistence.ConfigRepository;
+import io.airbyte.oauth.BaseOAuthFlow;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.http.HttpClient;
+import java.util.UUID;
+import java.util.function.Supplier;
+import org.apache.http.client.utils.URIBuilder;
+
+public class SlackOAuthFlow extends BaseOAuthFlow {
+
+  final String SLACK_CONSENT_URL_BASE = "https://slack.com/oauth/authorize";
+  final String SLACK_TOKEN_URL = "https://slack.com/api/oauth.access";
+
+  public SlackOAuthFlow(final ConfigRepository configRepository, HttpClient httpClient) {
+    super(configRepository, httpClient);
+  }
+
+  @VisibleForTesting
+  public SlackOAuthFlow(final ConfigRepository configRepository, final HttpClient httpClient, final Supplier<String> stateSupplier) {
+    super(configRepository, httpClient, stateSupplier);
+  }
+
+  /**
+   * 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
+   * @param clientId
+   * @param redirectUrl
+   */
+  @Override
+  protected String formatConsentUrl(UUID definitionId, String clientId, String redirectUrl) throws IOException {
+    try {
+      return new URIBuilder(SLACK_CONSENT_URL_BASE)
+          .addParameter("client_id", clientId)
+          .addParameter("redirect_uri", redirectUrl)
+          .addParameter("state", getState())
+          .addParameter("scope", "read")
+          .build().toString();
+    } catch (URISyntaxException e) {
+      throw new IOException("Failed to format Consent URL for OAuth flow", e);
+    }
+  }
+
+  /**
+   * Returns the URL where to retrieve the access token from.
+   */
+  @Override
+  protected String getAccessTokenUrl() {
+    return SLACK_TOKEN_URL;
+  }
+
+}

airbyte-oauth/src/test-integration/java/io.airbyte.oauth.flows/SlackOAuthFlowIntegrationTest.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.oauth.flows;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.config.SourceOAuthParameter;
+import io.airbyte.config.persistence.ConfigNotFoundException;
+import io.airbyte.config.persistence.ConfigRepository;
+import io.airbyte.oauth.OAuthFlowImplementation;
+import io.airbyte.validation.json.JsonValidationException;
+import java.io.IOException;
+import java.net.http.HttpClient;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+
+public class SlackOAuthFlowIntegrationTest extends OAuthFlowIntegrationTest {
+
+  @Override
+  protected Path getCredentialsPath() {
+    return Path.of("secrets/slack.json");
+  }
+
+  @Override
+  protected String getRedirectUrl() {
+    return "https://27b0-2804-14d-2a76-9a9a-fdbb-adee-9e5d-6c.ngrok.io/auth_flow";
+  }
+
+  @Override
+  protected OAuthFlowImplementation getFlowImplementation(ConfigRepository configRepository, HttpClient httpClient) {
+    return new SlackOAuthFlow(configRepository, httpClient);
+  }
+
+  @Test
+  public void testFullSlackOAuthFlow() throws InterruptedException, ConfigNotFoundException, IOException, JsonValidationException {
+    int limit = 20;
+    final UUID workspaceId = UUID.randomUUID();
+    final UUID definitionId = UUID.randomUUID();
+    final String fullConfigAsString = new String(Files.readAllBytes(getCredentialsPath()));
+    final JsonNode credentialsJson = Jsons.deserialize(fullConfigAsString).get("credentials");
+    when(configRepository.listSourceOAuthParam()).thenReturn(List.of(new SourceOAuthParameter()
+        .withOauthParameterId(UUID.randomUUID())
+        .withSourceDefinitionId(definitionId)
+        .withWorkspaceId(workspaceId)
+        .withConfiguration(Jsons.jsonNode(ImmutableMap.builder()
+            .put("client_id", credentialsJson.get("client_id").asText())
+            .put("client_secret", credentialsJson.get("client_secret").asText())
+            .build()))));
+    final String url = getFlowImplementation(configRepository, httpClient).getSourceConsentUrl(workspaceId, definitionId, getRedirectUrl());
+    LOGGER.info("Waiting for user consent at: {}", url);
+    // TODO: To automate, start a selenium job to navigate to the Consent URL and click on allowing
+    // access...
+    while (!serverHandler.isSucceeded() && limit > 0) {
+      Thread.sleep(1000);
+      limit -= 1;
+    }
+    assertTrue(serverHandler.isSucceeded(), "Failed to get User consent on time");
+    final Map<String, Object> params = flow.completeSourceOAuth(workspaceId, definitionId,
+        Map.of("code", serverHandler.getParamValue()), getRedirectUrl());
+    LOGGER.info("Response from completing OAuth Flow is: {}", params.toString());
+    assertTrue(params.containsKey("credentials"));
+    assertTrue(((Map<String, Object>) params.get("credentials")).containsKey("access_token"));
+    assertTrue(((Map<String, Object>) params.get("credentials")).get("access_token").toString().length() > 0);
+  }
+
+}

airbyte-oauth/src/test-integration/java/io/airbyte/oauth/flows/HubspotOAuthFlowIntegrationTest.java

@@ -31,6 +31,10 @@ protected Path getCredentialsPath() {
     return Path.of("secrets/hubspot.json");
   }
 
+  protected OAuthFlowImplementation getFlowObject(ConfigRepository configRepository) {
+    return new HubspotOAuthFlow(configRepository, httpClient);
+  }
+
   @Override
   protected OAuthFlowImplementation getFlowImplementation(ConfigRepository configRepository, HttpClient httpClient) {
     return new HubspotOAuthFlow(configRepository, httpClient);

airbyte-oauth/src/test-integration/java/io/airbyte/oauth/flows/OAuthFlowIntegrationTest.java

@@ -44,6 +44,10 @@ protected Path getCredentialsPath() {
     return Path.of("secrets/config.json");
   };
 
+  protected String getRedirectUrl() {
+    return REDIRECT_URL;
+  }
+
   protected abstract OAuthFlowImplementation getFlowImplementation(ConfigRepository configRepository, HttpClient httpClient);
 
   @BeforeEach