Merge pull request #88 from lensesio-dev/feat/elastic-ssl

Elasticsearch 6 + 7 SSL Support

build.sbt

@@ -1,4 +1,5 @@
 import Dependencies.Versions
+import Dependencies.`cyclopsPure`
 import Dependencies.`cyclops`
 import Dependencies.`lombok`
 import Dependencies.globalExcludeDeps
@@ -71,7 +72,7 @@ lazy val `query-language` = (project in file("java-connectors/kafka-connect-quer
       Seq(
         name := "kafka-connect-query-language",
         description := "Kafka Connect compatible connectors to move data between Kafka and popular data stores",
-        libraryDependencies ++= Seq(cyclops, lombok),
+        libraryDependencies ++= Seq(cyclops, cyclopsPure, lombok),
         publish / skip := true,
       ),
   )

java-connectors/build.gradle

@@ -31,6 +31,7 @@ allprojects {
         apacheToConfluentVersionAxis = ["2.8.1": "6.2.2", "3.3.0": "7.3.1"]
         caffeineVersion = '3.1.8'
         cyclopsVersion = '10.4.1'
+        bouncyCastleVersion = "1.78.1"
 
         //Other Manifest Info
         mainClassName = ''
@@ -64,6 +65,7 @@ allprojects {
 
         // functional java
         implementation group: 'com.oath.cyclops', name: 'cyclops', version: cyclopsVersion
+        implementation group: 'com.oath.cyclops', name: 'cyclops-pure', version: cyclopsVersion
 
         //tests
         testImplementation group: 'org.mockito', name: 'mockito-core', version: mockitoJupiterVersion

java-connectors/kafka-connect-common/build.gradle

@@ -12,6 +12,12 @@ project(":kafka-connect-common") {
         api group: 'org.apache.kafka', name: 'kafka-clients', version: kafkaVersion
         testImplementation(project(path: ':test-utils', configuration: 'testArtifacts'))
 
+        testImplementation group: 'org.bouncycastle', name:'bcprov-jdk18on', version: bouncyCastleVersion
+        testImplementation group: 'org.bouncycastle', name:'bcutil-jdk18on', version: bouncyCastleVersion
+        testImplementation group: 'org.bouncycastle', name:'bcpkix-jdk18on', version: bouncyCastleVersion
+        testImplementation group: 'org.bouncycastle', name:'bcpg-jdk18on', version: bouncyCastleVersion
+        testImplementation group: 'org.bouncycastle', name:'bctls-jdk18on', version: bouncyCastleVersion
+
         //confluent - may be needed soon
         //        implementation group: 'io.confluent', name: 'kafka-json-schema-serializer', version: apacheToConfluentVersionAxis.get(kafkaVersion)
         //        implementation group: 'io.confluent', name: 'kafka-connect-avro-converter', version: apacheToConfluentVersionAxis.get(kafkaVersion)

java-connectors/kafka-connect-common/src/main/java/io/lenses/streamreactor/common/exception/SecuritySetupException.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2017-2024 Lenses.io Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.lenses.streamreactor.common.exception;
+
+public class SecuritySetupException extends StreamReactorException {
+
+  public SecuritySetupException(String message) {
+    super(message);
+  }
+
+  public SecuritySetupException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

java-connectors/kafka-connect-common/src/main/java/io/lenses/streamreactor/common/security/KeyStoreInfo.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2017-2024 Lenses.io Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.lenses.streamreactor.common.security;
+
+import cyclops.control.Option;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+import java.nio.file.Path;
+
+@AllArgsConstructor
+@Data
+public class KeyStoreInfo implements StoreInfo {
+
+  private Path storePath;
+
+  private StoreType storeType;
+
+  private String storePassword;
+
+  private Option<String> managerAlgorithm;
+
+}

java-connectors/kafka-connect-common/src/main/java/io/lenses/streamreactor/common/security/StoreInfo.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2017-2024 Lenses.io Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.lenses.streamreactor.common.security;
+
+import cyclops.control.Option;
+
+import java.nio.file.Path;
+
+interface StoreInfo {
+
+  Path getStorePath();
+
+  StoreType getStoreType();
+
+  Option<String> getManagerAlgorithm();
+}

java-connectors/kafka-connect-common/src/main/java/io/lenses/streamreactor/common/security/StoreType.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2017-2024 Lenses.io Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.lenses.streamreactor.common.security;
+
+import cyclops.control.Either;
+import cyclops.control.Try;
+import io.lenses.streamreactor.common.exception.SecuritySetupException;
+import lombok.Getter;
+
+@Getter
+public enum StoreType {
+
+  JKS("JKS"),
+  PKCS12("PKCS12");
+
+  private final String type;
+
+  StoreType(String type) {
+    this.type = type;
+  }
+
+  public static Either<SecuritySetupException, StoreType> valueOfCaseInsensitive(String storeType) {
+    return Try
+        .withCatch(() -> StoreType.valueOf(storeType.toUpperCase()))
+        .toEither()
+        .mapLeft(ex -> new SecuritySetupException(String.format("Unable to retrieve Store type %s", storeType), ex));
+  }
+
+  public static final StoreType DEFAULT_STORE_TYPE = StoreType.JKS;
+
+}