🐛Destination-elasticsearch: enforce ssl connection on cloud

airbyte-integrations/connectors/destination-elasticsearch-strict-encrypt/build.gradle

@@ -33,6 +33,7 @@ dependencies {
     testImplementation libs.connectors.testcontainers.elasticsearch
     integrationTestJavaImplementation libs.connectors.testcontainers.elasticsearch
 
+    integrationTestJavaImplementation project(':airbyte-commons-worker')
     integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
     integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-elasticsearch')
 }

airbyte-integrations/connectors/destination-elasticsearch-strict-encrypt/src/main/java/io/airbyte/integrations/destination/elasticsearch/ElasticsearchStrictEncryptDestination.java

@@ -4,19 +4,27 @@
 
 package io.airbyte.integrations.destination.elasticsearch;
 
+import static co.elastic.clients.elasticsearch.watcher.Input.HTTP;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.IntegrationRunner;
 import io.airbyte.integrations.base.spec_modification.SpecModifyingDestination;
+import io.airbyte.protocol.models.AirbyteConnectionStatus;
 import io.airbyte.protocol.models.ConnectorSpecification;
+import java.net.URL;
+import java.util.Objects;
 import java.util.stream.IntStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class ElasticsearchStrictEncryptDestination extends SpecModifyingDestination implements Destination {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchStrictEncryptDestination.class);
+  private final ObjectMapper mapper = new ObjectMapper();
 
   public ElasticsearchStrictEncryptDestination() {
     super(new ElasticsearchDestination());
@@ -38,4 +46,25 @@ public ConnectorSpecification modifySpec(ConnectorSpecification originalSpec) th
     return spec;
   }
 
+  @Override
+  public AirbyteConnectionStatus check(JsonNode config) throws Exception {
+
+    final ConnectorConfiguration configObject = convertConfig(config);
+    if (Objects.isNull(configObject.getEndpoint())) {
+      return new AirbyteConnectionStatus()
+          .withStatus(AirbyteConnectionStatus.Status.FAILED).withMessage("endpoint must not be empty");
+    }
+
+    if (new URL(configObject.getEndpoint()).getProtocol().equals(HTTP)) {
+      return new AirbyteConnectionStatus()
+          .withStatus(AirbyteConnectionStatus.Status.FAILED).withMessage("only https protocol is allowed");
+    }
+
+    return super.check(config);
+  }
+
+  private ConnectorConfiguration convertConfig(JsonNode config) {
+    return mapper.convertValue(config, ConnectorConfiguration.class);
+  }
+
 }

airbyte-integrations/connectors/destination-elasticsearch-strict-encrypt/src/test-integration/java/io/airbyte/integrations/destination/elasticsearch/ElasticsearchStrictEncryptDestinationAcceptanceTest.java

@@ -103,6 +103,19 @@ protected JsonNode getConfig() {
         .build());
   }
 
+    protected JsonNode getUnsecureConfig() {
+
+      final JsonNode authConfig = Jsons.jsonNode(Map.of(
+          "method", "basic",
+          "username", "elastic",
+          "password", "MagicWord"));
+
+      return Jsons.jsonNode(ImmutableMap.builder()
+          .put("endpoint", String.format("http://%s:%s", container.getHost(), container.getMappedPort(9200)))
+          .put("authenticationMethod", authConfig)
+          .build());
+    }
+
   @Override
   protected JsonNode getFailCheckConfig() {
     // should result in a failed connection check
@@ -135,4 +148,10 @@ protected void tearDown(DestinationAcceptanceTest.TestDestinationEnv testEnv) {
     connection.allIndices().forEach(connection::deleteIndexIfPresent);
   }
 
+  @Test
+  public void testCheckConnectionInvalidHttpProtocol() throws Exception {
+    assertEquals(Status.FAILED, runCheck(getUnsecureConfig()).getStatus());
+  }
+
+
 }