MAINT: deprecate certificate_content with certificate_key (opensearch…

…-project#4434) * MAINT: deprecate certificate_content with certificate_key Signed-off-by: George Chen <[email protected]>
dinujoh · Apr 23, 2024 · 250531d · 250531d
1 parent df17cc5
commit 250531d
Show file tree

Hide file tree

Showing 15 changed files with 333 additions and 30 deletions.
diff --git a/...in/java/org/opensearch/dataprepper/plugins/certificate/validation/PemObjectValidator.java b/...in/java/org/opensearch/dataprepper/plugins/certificate/validation/PemObjectValidator.java
@@ -0,0 +1,16 @@
+package org.opensearch.dataprepper.plugins.certificate.validation;
+
+import org.bouncycastle.util.io.pem.PemReader;
+
+import java.io.IOException;
+import java.io.StringReader;
+
+public class PemObjectValidator {
+    public static boolean isPemObject(final String certificate) {
+        try (PemReader reader = new PemReader(new StringReader(certificate))) {
+            return reader.readPemObject() != null;
+        } catch (IOException e) {
+            return false;
+        }
+    }
+}
diff --git a/...ava/org/opensearch/dataprepper/plugins/certificate/validation/PemObjectValidatorTest.java b/...ava/org/opensearch/dataprepper/plugins/certificate/validation/PemObjectValidatorTest.java
@@ -0,0 +1,42 @@
+package org.opensearch.dataprepper.plugins.certificate.validation;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.UUID;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+class PemObjectValidatorTest {
+    private static final String TEST_CERTIFICATE = "-----BEGIN CERTIFICATE-----\n" +
+            "MIIDYTCCAkmgAwIBAgIUFBUALAhfpezYbYw6AtH96tizPTIwDQYJKoZIhvcNAQEL\n" +
+            "BQAwWTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM\n" +
+            "GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDESMBAGA1UEAwwJMTI3LjAuMC4xMB4X\n" +
+            "DTI0MDQxMTE3MzcyMloXDTM0MDQwOTE3MzcyMlowWTELMAkGA1UEBhMCQVUxEzAR\n" +
+            "BgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoMGEludGVybmV0IFdpZGdpdHMgUHR5\n" +
+            "IEx0ZDESMBAGA1UEAwwJMTI3LjAuMC4xMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8A\n" +
+            "MIIBCgKCAQEAuaS6lrpg38XT5wmukekr8NSXcO70yhMRLF29mAXasYeumtHVDR/p\n" +
+            "f8vTE4l+b36kRuaew4htGRZQcWJBdPoCDkDHA3+5z5t9Fe3nR9FzIA+E/KjyMCEq\n" +
+            "xNgc9OIN9UyBbbneMkR24W8LkAywxk3euXgj46+7SGFHAdNLqC72Yl3W1E32rQAR\n" +
+            "c6zQIJ45uogqU19QJHCBBfJA+IFylwtNGWfNbvdvGCXx5FZnM3q4rCxNr9F+LBsS\n" +
+            "xWFlXGMHXo2+bMBGIBXPGbGXpad/jVgxjM6zV5vnG1g8GDxaHaM+3LjJwa7eQYVA\n" +
+            "ogetug9wqesxkf+Nic/rpB6J7zM2iwY0AQIDAQABoyEwHzAdBgNVHQ4EFgQUept4\n" +
+            "OD2pNRYswtfrOqnOgx4QtjYwDQYJKoZIhvcNAQELBQADggEBACU+Qjmf35BOYjDj\n" +
+            "TX1f6bhgwsHP/WHwWWKIhVSOB0CFHoizzQyLREgWWLkKs3Ye3q9DXku0saMfIerq\n" +
+            "S7hqDA8hNVJVyllh2FuuNQVkmOKJFTwev2n3OvSyz4mxWW3UNJb/YTuK93nNHVVo\n" +
+            "/3+lQg0sJRhSMs/GmR/Hn7/py2/2pucFJrML/Dtjv7SwrOXptWn+GCB+3bUpfNdg\n" +
+            "sHeZEv9vpbQDzp1Lux7l3pMzwsi6HU4xTyHClBD7V8S2MUExMXDF+Cr4g7lmOb02\n" +
+            "Bw0dTI7afBMI8n5YgTX78YMGqbO/WJ3bOc26P2i7RrRIhOXw69UZff2JwYAnX6Op\n" +
+            "zHOodz4=\n" +
+            "-----END CERTIFICATE-----";
+
+    @Test
+    void testValidPemObject() {
+        assertThat(PemObjectValidator.isPemObject(TEST_CERTIFICATE), is(true));
+    }
+
+    @Test
+    void testInValidPemObject() {
+        assertThat(PemObjectValidator.isPemObject(UUID.randomUUID().toString()), is(false));
+    }
+}
diff --git a/data-prepper-plugins/kafka-plugins/build.gradle b/data-prepper-plugins/kafka-plugins/build.gradle
@@ -25,6 +25,7 @@ configurations {
 
 dependencies {
     implementation project(':data-prepper-api')
+    implementation project(':data-prepper-plugins:common')
     implementation project(':data-prepper-plugins:buffer-common')
     implementation project(':data-prepper-plugins:blocking-buffer')
     implementation project(':data-prepper-plugins:aws-plugin-api')

diff --git a/...rc/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/EncryptionConfig.java b/...rc/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/EncryptionConfig.java
@@ -5,14 +5,20 @@
 
 package org.opensearch.dataprepper.plugins.kafka.configuration;
 
+import com.fasterxml.jackson.annotation.JsonAlias;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import jakarta.validation.constraints.AssertTrue;
+import org.opensearch.dataprepper.plugins.certificate.validation.PemObjectValidator;
+
+import java.nio.file.Paths;
 
 public class EncryptionConfig {
     @JsonProperty("type")
     private EncryptionType type = EncryptionType.SSL;
 
-    @JsonProperty("certificate_content")
-    private String certificateContent;
+    @JsonAlias("certificate_content")
+    @JsonProperty("certificate")
+    private String certificate;
 
     @JsonProperty("trust_store_file_path")
     private String trustStoreFilePath;
@@ -27,8 +33,8 @@ public EncryptionType getType() {
         return type;
     }
 
-    public String getCertificateContent() {
-        return certificateContent;
+    public String getCertificate() {
+        return certificate;
     }
 
     public String getTrustStoreFilePath() {
@@ -42,4 +48,17 @@ public String getTrustStorePassword() {
     public boolean getInsecure() {
         return insecure;
     }
+
+    @AssertTrue(message = "certificate must be either valid PEM file path or public key content.")
+    boolean isCertificateValid() {
+        if (PemObjectValidator.isPemObject(certificate)) {
+            return true;
+        }
+        try {
+            Paths.get(certificate);
+            return true;
+        } catch (Exception e) {
+            return false;
+        }
+    }
 }
diff --git a/.../src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java b/.../src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java
@@ -131,8 +131,8 @@ private static void setPlainTextAuthProperties(final Properties properties, fina
     }
 
     private static void setSecurityProtocolSSLProperties(final Properties properties, final EncryptionConfig encryptionConfig) {
-        if (Objects.nonNull(encryptionConfig.getCertificateContent())) {
-            setCustomSslProperties(properties, encryptionConfig.getCertificateContent());
+        if (Objects.nonNull(encryptionConfig.getCertificate())) {
+            setCustomSslProperties(properties, encryptionConfig.getCertificate());
         } else if (Objects.nonNull(encryptionConfig.getTrustStoreFilePath()) &&
                 Objects.nonNull(encryptionConfig.getTrustStorePassword())) {
             setTruststoreProperties(properties, encryptionConfig);

diff --git a/...est/java/org/opensearch/dataprepper/plugins/kafka/configuration/EncryptionConfigTest.java b/...est/java/org/opensearch/dataprepper/plugins/kafka/configuration/EncryptionConfigTest.java
@@ -0,0 +1,59 @@
+package org.opensearch.dataprepper.plugins.kafka.configuration;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import com.fasterxml.jackson.dataformat.yaml.YAMLGenerator;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+class EncryptionConfigTest {
+    private static final String TEST_CERTIFICATE = "-----BEGIN CERTIFICATE-----\n" +
+            "MIIDYTCCAkmgAwIBAgIUFBUALAhfpezYbYw6AtH96tizPTIwDQYJKoZIhvcNAQEL\n" +
+            "BQAwWTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM\n" +
+            "GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDESMBAGA1UEAwwJMTI3LjAuMC4xMB4X\n" +
+            "DTI0MDQxMTE3MzcyMloXDTM0MDQwOTE3MzcyMlowWTELMAkGA1UEBhMCQVUxEzAR\n" +
+            "BgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoMGEludGVybmV0IFdpZGdpdHMgUHR5\n" +
+            "IEx0ZDESMBAGA1UEAwwJMTI3LjAuMC4xMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8A\n" +
+            "MIIBCgKCAQEAuaS6lrpg38XT5wmukekr8NSXcO70yhMRLF29mAXasYeumtHVDR/p\n" +
+            "f8vTE4l+b36kRuaew4htGRZQcWJBdPoCDkDHA3+5z5t9Fe3nR9FzIA+E/KjyMCEq\n" +
+            "xNgc9OIN9UyBbbneMkR24W8LkAywxk3euXgj46+7SGFHAdNLqC72Yl3W1E32rQAR\n" +
+            "c6zQIJ45uogqU19QJHCBBfJA+IFylwtNGWfNbvdvGCXx5FZnM3q4rCxNr9F+LBsS\n" +
+            "xWFlXGMHXo2+bMBGIBXPGbGXpad/jVgxjM6zV5vnG1g8GDxaHaM+3LjJwa7eQYVA\n" +
+            "ogetug9wqesxkf+Nic/rpB6J7zM2iwY0AQIDAQABoyEwHzAdBgNVHQ4EFgQUept4\n" +
+            "OD2pNRYswtfrOqnOgx4QtjYwDQYJKoZIhvcNAQELBQADggEBACU+Qjmf35BOYjDj\n" +
+            "TX1f6bhgwsHP/WHwWWKIhVSOB0CFHoizzQyLREgWWLkKs3Ye3q9DXku0saMfIerq\n" +
+            "S7hqDA8hNVJVyllh2FuuNQVkmOKJFTwev2n3OvSyz4mxWW3UNJb/YTuK93nNHVVo\n" +
+            "/3+lQg0sJRhSMs/GmR/Hn7/py2/2pucFJrML/Dtjv7SwrOXptWn+GCB+3bUpfNdg\n" +
+            "sHeZEv9vpbQDzp1Lux7l3pMzwsi6HU4xTyHClBD7V8S2MUExMXDF+Cr4g7lmOb02\n" +
+            "Bw0dTI7afBMI8n5YgTX78YMGqbO/WJ3bOc26P2i7RrRIhOXw69UZff2JwYAnX6Op\n" +
+            "zHOodz4=\n" +
+            "-----END CERTIFICATE-----";
+    private final ObjectMapper objectMapper = new ObjectMapper(new YAMLFactory().enable(YAMLGenerator.Feature.USE_PLATFORM_LINE_BREAKS));
+
+    @ParameterizedTest
+    @ValueSource(strings = {
+            TEST_CERTIFICATE,
+            "/some/file/path"
+    })
+    void testIsCertificateValid_returns_true(final String certificate) throws JsonProcessingException {
+        final String encryptionConfigYaml = String.format(
+                "  type: \"ssl\"\n" +
+                "  certificate: \"%s\"\n", certificate.replace("\n", "\\n"));
+        final EncryptionConfig encryptionConfig = objectMapper.readValue(encryptionConfigYaml, EncryptionConfig.class);
+        assertThat(encryptionConfig.isCertificateValid(), is(true));
+    }
+
+    @Test
+    void testIsCertificateValid_returns_false() throws JsonProcessingException {
+        final String encryptionConfigYaml =
+                "  type: \"ssl\"\n" +
+                "  certificate: \"\\u0000\\u0081\"\n";
+        final EncryptionConfig encryptionConfig = objectMapper.readValue(encryptionConfigYaml, EncryptionConfig.class);
+        assertThat(encryptionConfig.isCertificateValid(), is(false));
+    }
+}
diff --git a/data-prepper-plugins/kafka-plugins/src/test/resources/kafka-pipeline-auth-insecure.yaml b/data-prepper-plugins/kafka-plugins/src/test/resources/kafka-pipeline-auth-insecure.yaml
@@ -5,7 +5,7 @@ log-pipeline :
           - "localhost:9092"
         encryption:
           type: "NONE"
-          certificate_content: "CERTIFICATE_DATA"
+          certificate: "CERTIFICATE_DATA"
           insecure: "true"
         authentication:
           sasl:

diff --git a/data-prepper-plugins/kafka-plugins/src/test/resources/kafka-pipeline-no-auth-ssl-none.yaml b/data-prepper-plugins/kafka-plugins/src/test/resources/kafka-pipeline-no-auth-ssl-none.yaml
@@ -5,7 +5,7 @@ log-pipeline :
           - "localhost:9092"
         encryption:
           type: "NONE"
-          certificate_content: "CERTIFICATE_DATA"
+          certificate: "CERTIFICATE_DATA"
         topics:
         - name: "quickstart-events"
           group_id: "groupdID1"

diff --git a/data-prepper-plugins/kafka-plugins/src/test/resources/kafka-pipeline-no-auth-ssl.yaml b/data-prepper-plugins/kafka-plugins/src/test/resources/kafka-pipeline-no-auth-ssl.yaml
@@ -5,7 +5,7 @@ log-pipeline :
           - "localhost:9092"
         encryption:
           type: "SSL"
-          certificate_content: "CERTIFICATE_DATA"
+          certificate: "CERTIFICATE_DATA"
         topics:
         - name: "quickstart-events"
           group_id: "groupdID1"

diff --git a/...plugins/kafka-plugins/src/test/resources/kafka-pipeline-sasl-ssl-certificate-content.yaml b/...plugins/kafka-plugins/src/test/resources/kafka-pipeline-sasl-ssl-certificate-content.yaml
@@ -5,7 +5,7 @@ log-pipeline :
           - "localhost:9092"
         encryption:
           type: "SSL"
-          certificate_content: "CERTIFICATE_DATA"
+          certificate: "CERTIFICATE_DATA"
         authentication:
           sasl:
             plaintext:

diff --git a/data-prepper-plugins/kafka-plugins/src/test/resources/kafka-pipeline-sasl-ssl.yaml b/data-prepper-plugins/kafka-plugins/src/test/resources/kafka-pipeline-sasl-ssl.yaml
@@ -5,7 +5,7 @@ log-pipeline :
           - "localhost:9092"
         encryption:
           type: "SSL"
-          certificate_content: "CERTIFICATE_DATA"
+          certificate: "CERTIFICATE_DATA"
         authentication:
           sasl:
             plaintext:

diff --git a/...ensearch/dataprepper/plugins/source/opensearch/configuration/ConnectionConfiguration.java b/...ensearch/dataprepper/plugins/source/opensearch/configuration/ConnectionConfiguration.java
@@ -4,19 +4,24 @@
  */
 package org.opensearch.dataprepper.plugins.source.opensearch.configuration;
 
+import com.fasterxml.jackson.annotation.JsonAlias;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import jakarta.validation.constraints.AssertTrue;
+import org.opensearch.dataprepper.plugins.certificate.validation.PemObjectValidator;
 
 import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.time.Duration;
 
 public class ConnectionConfiguration {
 
+  @Deprecated
   @JsonProperty("cert")
   private Path certPath;
 
-  @JsonProperty("certificate_content")
-  private String certificateContent;
+  @JsonAlias("certiciate_content")
+  @JsonProperty("certificate")
+  private String certificate;
 
   @JsonProperty("socket_timeout")
   private Duration socketTimeout;
@@ -31,8 +36,8 @@ public Path getCertPath() {
     return certPath;
   }
 
-  public String getCertificateContent() {
-    return certificateContent;
+  public String getCertificate() {
+    return certificate;
   }
 
   public Duration getSocketTimeout() {
@@ -47,11 +52,24 @@ public boolean isInsecure() {
     return insecure;
   }
 
-  @AssertTrue(message = "Certificate file path and certificate content both are configured. " +
+  @AssertTrue(message = "cert and certificate both are configured. " +
           "Please use only one configuration.")
   boolean certificateFileAndContentAreMutuallyExclusive() {
-    if(certPath == null && certificateContent == null)
+    if(certPath == null && certificate == null)
       return true;
-    return certPath != null ^ certificateContent != null;
+    return certPath != null ^ certificate != null;
+  }
+
+  @AssertTrue(message = "certificate must be either valid PEM file path or public key content.")
+  boolean isCertificateValid() {
+    if (PemObjectValidator.isPemObject(certificate)) {
+      return true;
+    }
+    try {
+      Paths.get(certificate);
+      return true;
+    } catch (Exception e) {
+      return false;
+    }
   }
 }
diff --git a/...ensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactory.java b/...ensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactory.java
@@ -29,6 +29,7 @@
 import org.opensearch.dataprepper.aws.api.AwsCredentialsOptions;
 import org.opensearch.dataprepper.aws.api.AwsCredentialsSupplier;
 import org.opensearch.dataprepper.aws.api.AwsRequestSigningApache4Interceptor;
+import org.opensearch.dataprepper.plugins.certificate.validation.PemObjectValidator;
 import org.opensearch.dataprepper.plugins.source.opensearch.OpenSearchSourceConfiguration;
 import org.opensearch.dataprepper.plugins.source.opensearch.configuration.ConnectionConfiguration;
 import org.opensearch.dataprepper.plugins.truststore.TrustStoreProvider;
@@ -277,8 +278,12 @@ private TrustManager[] createTrustManagers(final ConnectionConfiguration connect
         final Path certPath = connectionConfiguration.getCertPath();
         if (Objects.nonNull(certPath)) {
             return TrustStoreProvider.createTrustManager(certPath);
-        } else if (Objects.nonNull(connectionConfiguration.getCertificateContent())) {
-            return TrustStoreProvider.createTrustManager(connectionConfiguration.getCertificateContent());
+        } else if (Objects.nonNull(connectionConfiguration.getCertificate())) {
+            if (PemObjectValidator.isPemObject(connectionConfiguration.getCertificate())) {
+                return TrustStoreProvider.createTrustManager(connectionConfiguration.getCertificate());
+            } else {
+                return TrustStoreProvider.createTrustManager(Path.of(connectionConfiguration.getCertificate()));
+            }
         } else {
             return TrustStoreProvider.createTrustAllManager();
         }
@@ -288,8 +293,12 @@ private SSLContext getCAStrategy(final ConnectionConfiguration connectionConfigu
         final Path certPath = connectionConfiguration.getCertPath();
         if (Objects.nonNull(certPath)) {
             return TrustStoreProvider.createSSLContext(certPath);
-        } else if (Objects.nonNull(connectionConfiguration.getCertificateContent())) {
-            return TrustStoreProvider.createSSLContext(connectionConfiguration.getCertificateContent());
+        } else if (Objects.nonNull(connectionConfiguration.getCertificate())) {
+            if (PemObjectValidator.isPemObject(connectionConfiguration.getCertificate())) {
+                return TrustStoreProvider.createSSLContext(connectionConfiguration.getCertificate());
+            } else {
+                return TrustStoreProvider.createSSLContext(Path.of(connectionConfiguration.getCertificate()));
+            }
         } else {
             return TrustStoreProvider.createSSLContextWithTrustAllStrategy();
         }