🐛Destination-dynamodb: enforce ssl connection

airbyte-integrations/connectors/destination-dynamodb/build.gradle

@@ -19,6 +19,7 @@ dependencies {
 
     testImplementation project(':airbyte-integrations:bases:standard-destination-test')
 
+    integrationTestJavaImplementation project(':airbyte-commons-worker')
     integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
     integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-dynamodb')
 }

airbyte-integrations/connectors/destination-dynamodb/src/main/java/io/airbyte/integrations/destination/dynamodb/DynamodbDestination.java

@@ -4,6 +4,7 @@
 
 package io.airbyte.integrations.destination.dynamodb;
 
+import com.amazonaws.Protocol;
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.integrations.BaseConnector;
 import io.airbyte.integrations.base.AirbyteMessageConsumer;
@@ -12,12 +13,14 @@
 import io.airbyte.protocol.models.AirbyteConnectionStatus;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
+import java.net.URL;
 import java.util.function.Consumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class DynamodbDestination extends BaseConnector implements Destination {
 
+  private static final String NON_SECURE_URL_ERR_MSG = "Server Endpoint requires HTTPS";
   private static final Logger LOGGER = LoggerFactory.getLogger(DynamodbDestination.class);
 
   public static void main(final String[] args) throws Exception {
@@ -27,7 +30,19 @@ public static void main(final String[] args) throws Exception {
   @Override
   public AirbyteConnectionStatus check(final JsonNode config) {
     try {
-      DynamodbChecker.attemptDynamodbWriteAndDelete(DynamodbDestinationConfig.getDynamodbDestinationConfig(config));
+      final DynamodbDestinationConfig dynamodbDestinationConfig =
+          DynamodbDestinationConfig.getDynamodbDestinationConfig(config);
+
+      // enforce ssl connection
+      if (!dynamodbDestinationConfig.getEndpoint().isBlank()) {
+        if (new URL(dynamodbDestinationConfig.getEndpoint()).getProtocol().equals(Protocol.HTTP.toString())) {
+          return new AirbyteConnectionStatus()
+              .withStatus(AirbyteConnectionStatus.Status.FAILED)
+              .withMessage(NON_SECURE_URL_ERR_MSG);
+        }
+      }
+
+      DynamodbChecker.attemptDynamodbWriteAndDelete(dynamodbDestinationConfig);
       return new AirbyteConnectionStatus().withStatus(AirbyteConnectionStatus.Status.SUCCEEDED);
     } catch (final Exception e) {
       LOGGER.error("Exception attempting to access the DynamoDB table: ", e);

airbyte-integrations/connectors/destination-dynamodb/src/test-integration/java/io/airbyte/integrations/destination/dynamodb/DynamodbDestinationAcceptanceTest.java

@@ -4,6 +4,8 @@
 
 package io.airbyte.integrations.destination.dynamodb;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.auth.AWSStaticCredentialsProvider;
 import com.amazonaws.auth.BasicAWSCredentials;
@@ -14,20 +16,25 @@
 import com.amazonaws.services.dynamodbv2.document.spec.ScanSpec;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.config.StandardCheckConnectionOutput;
+import io.airbyte.config.StandardCheckConnectionOutput.Status;
 import io.airbyte.integrations.base.JavaBaseConstants;
 import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
 import java.io.IOException;
 import java.math.BigDecimal;
 import java.nio.file.Path;
 import java.util.*;
+import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class DynamodbDestinationAcceptanceTest extends DestinationAcceptanceTest {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DynamodbDestinationAcceptanceTest.class);
+  private static final String NON_SECURE_URL_ERR_MSG = "Server Endpoint requires HTTPS";
 
   protected final String secretFilePath = "secrets/config.json";
   protected JsonNode configJson;
@@ -159,4 +166,21 @@ protected void tearDown(final TestDestinationEnv testEnv) {
     }
   }
 
+  @Test
+  public void testCheckConnectionInvalidHttpProtocol() throws Exception {
+    final StandardCheckConnectionOutput checkResult = runCheck(getUnsecureConfig());
+    assertEquals(Status.FAILED, checkResult.getStatus());
+    assertEquals(NON_SECURE_URL_ERR_MSG, checkResult.getMessage());
+  }
+
+  protected JsonNode getUnsecureConfig() {
+    return Jsons.jsonNode(ImmutableMap.builder()
+        .put("dynamodb_endpoint", "http://testurl.com:9000")
+        .put("dynamodb_table_name_prefix", "integration-test")
+        .put("dynamodb_region", "us-east-2")
+        .put("access_key_id", "dummy_access_key_id")
+        .put("secret_access_key", "dummy_secret_access_key")
+        .build());
+  }
+
 }