Added support for secured URI schemes (#333)

Author: mroiter-larus

.gitignore

@@ -33,3 +33,5 @@ Thumbs.db
 .cache-main
 .cache-tests
 bin
+doc/node
+doc/node_modules

kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/sink/Neo4jService.kt

@@ -42,20 +42,24 @@ class Neo4jService(private val config: Neo4jSinkConnectorConfig):
 
     init {
         val configBuilder = Config.builder()
-        if (this.config.encryptionEnabled) {
-            configBuilder.withEncryption()
-            val trustStrategy: Config.TrustStrategy = when (this.config.encryptionTrustStrategy) {
-                Config.TrustStrategy.Strategy.TRUST_ALL_CERTIFICATES -> Config.TrustStrategy.trustAllCertificates()
-                Config.TrustStrategy.Strategy.TRUST_SYSTEM_CA_SIGNED_CERTIFICATES -> Config.TrustStrategy.trustSystemCertificates()
-                Config.TrustStrategy.Strategy.TRUST_CUSTOM_CA_SIGNED_CERTIFICATES -> Config.TrustStrategy.trustCustomCertificateSignedBy(this.config.encryptionCACertificateFile)
-                else -> {
-                    throw ConfigException(Neo4jSinkConnectorConfig.ENCRYPTION_TRUST_STRATEGY, this.config.encryptionTrustStrategy.toString(), "Encryption Trust Strategy is not supported.")
+
+        if (!this.config.hasSecuredURI()) {
+            if (this.config.encryptionEnabled) {
+                configBuilder.withEncryption()
+                val trustStrategy: Config.TrustStrategy = when (this.config.encryptionTrustStrategy) {
+                    Config.TrustStrategy.Strategy.TRUST_ALL_CERTIFICATES -> Config.TrustStrategy.trustAllCertificates()
+                    Config.TrustStrategy.Strategy.TRUST_SYSTEM_CA_SIGNED_CERTIFICATES -> Config.TrustStrategy.trustSystemCertificates()
+                    Config.TrustStrategy.Strategy.TRUST_CUSTOM_CA_SIGNED_CERTIFICATES -> Config.TrustStrategy.trustCustomCertificateSignedBy(this.config.encryptionCACertificateFile)
+                    else -> {
+                        throw ConfigException(Neo4jSinkConnectorConfig.ENCRYPTION_TRUST_STRATEGY, this.config.encryptionTrustStrategy.toString(), "Encryption Trust Strategy is not supported.")
+                    }
                 }
+                configBuilder.withTrustStrategy(trustStrategy)
+            } else {
+                configBuilder.withoutEncryption()
             }
-            configBuilder.withTrustStrategy(trustStrategy)
-        } else {
-            configBuilder.withoutEncryption()
         }
+
         val authToken = when (this.config.authenticationType) {
             AuthenticationType.NONE -> AuthTokens.none()
             AuthenticationType.BASIC -> {

kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/sink/Neo4jSinkConnectorConfig.kt

@@ -10,6 +10,7 @@ import org.apache.kafka.common.config.AbstractConfig
 import org.apache.kafka.common.config.ConfigDef
 import org.apache.kafka.common.config.ConfigException
 import org.apache.kafka.connect.sink.SinkTask
+import org.neo4j.cypher.internal.logical.plans.`Optional$`
 import org.neo4j.driver.internal.async.pool.PoolSettings
 import org.neo4j.driver.Config
 import streams.kafka.connect.utils.PropertiesUtil
@@ -19,6 +20,7 @@ import streams.service.Topics
 import streams.service.sink.strategy.SourceIdIngestionStrategyConfig
 import java.io.File
 import java.net.URI
+import java.util.*
 import java.util.concurrent.TimeUnit
 
 enum class AuthenticationType {
@@ -129,6 +131,8 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): AbstractConfig(config(), o
         }
     }
 
+    fun hasSecuredURI() = serverUri.any { it.scheme.endsWith("+s", true) || it.scheme.endsWith("+ssc", true) }
+
     companion object {
         const val SERVER_URI = "neo4j.server.uri"
         const val DATABASE = "neo4j.database"
@@ -220,7 +224,7 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): AbstractConfig(config(), o
                             .importance(ConfigDef.Importance.HIGH)
                             .defaultValue("bolt://localhost:7687")
                             .group(ConfigGroup.CONNECTION)
-                            .validator(Validators.validURI("bolt", "bolt+routing", "neo4j"))
+                            .validator(Validators.validURI("bolt", "bolt+routing", "bolt+s", "bolt+ssc","neo4j", "neo4j+s", "neo4j+ssc"))
                             .build())
                     .define(ConfigKeyBuilder
                             .of(CONNECTION_POOL_MAX_SIZE, ConfigDef.Type.INT)

pom.xml

@@ -54,7 +54,7 @@
         <kafka.version>2.3.0</kafka.version>
         <jackson.version>2.9.7</jackson.version>
         <kotlin.compiler.incremental>true</kotlin.compiler.incremental>
-        <neo4j.java.driver.version>4.0.1</neo4j.java.driver.version>
+        <neo4j.java.driver.version>4.1.0</neo4j.java.driver.version>
         <testcontainers.version>1.11.0</testcontainers.version>
         <avro.version>1.8.2</avro.version>
         <mokito.version>3.2.0</mokito.version>