#1801 Add trusted certificates configuration for HonoConnection.

Also fix table of contents in Hono Connection documentation.

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/config/DefaultHonoConfig.java

@@ -42,6 +42,7 @@ public final class DefaultHonoConfig implements HonoConfig {
 
     private final URI baseUri;
     private final boolean validateCertificates;
+    private final String trustedCertificates;
     private final SaslMechanism saslMechanism;
     private final Set<URI> bootstrapServerUris;
     private final UserPasswordCredentials credentials;
@@ -62,6 +63,7 @@ public DefaultHonoConfig(final ActorSystem actorSystem) {
     private DefaultHonoConfig(final ScopedConfig scopedConfig) {
         baseUri = getBaseUriOrThrow(scopedConfig);
         validateCertificates = scopedConfig.getBoolean(HonoConfigValue.VALIDATE_CERTIFICATES.getConfigPath());
+        trustedCertificates = scopedConfig.getString(HonoConfigValue.TRUSTED_CERTIFICATES.getConfigPath());
         saslMechanism = scopedConfig.getEnum(SaslMechanism.class, HonoConfigValue.SASL_MECHANISM.getConfigPath());
         bootstrapServerUris = Collections.unmodifiableSet(getBootstrapServerUrisOrThrow(scopedConfig));
         credentials = UserPasswordCredentials.newInstance(
@@ -120,6 +122,11 @@ public boolean isValidateCertificates() {
         return validateCertificates;
     }
 
+    @Override
+    public String getTrustedCertificates() {
+        return trustedCertificates;
+    }
+
     @Override
     public SaslMechanism getSaslMechanism() {
         return saslMechanism;
@@ -146,6 +153,7 @@ public boolean equals(final Object o) {
         final var that = (DefaultHonoConfig) o;
         return Objects.equals(baseUri, that.baseUri)
                 && Objects.equals(validateCertificates, that.validateCertificates)
+                && Objects.equals(trustedCertificates, that.trustedCertificates)
                 && Objects.equals(saslMechanism, that.saslMechanism)
                 && Objects.equals(bootstrapServerUris, that.bootstrapServerUris)
                 && Objects.equals(credentials, that.credentials);
@@ -154,14 +162,16 @@ public boolean equals(final Object o) {
 
     @Override
     public int hashCode() {
-        return Objects.hash(baseUri, validateCertificates, saslMechanism, bootstrapServerUris, credentials);
+        return Objects.hash(baseUri, validateCertificates, trustedCertificates, saslMechanism, bootstrapServerUris,
+                credentials);
     }
 
     @Override
     public String toString() {
         return getClass().getSimpleName() + " [" +
                 "baseUri=" + baseUri +
                 ", validateCertificates=" + validateCertificates +
+                ", trustedCertificates=hash:" + Objects.hash(trustedCertificates) +
                 ", saslMechanism=" + saslMechanism +
                 ", bootstrapServers=" + bootstrapServerUris +
                 "]";

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/config/HonoConfig.java

@@ -43,6 +43,13 @@ public interface HonoConfig {
      */
     boolean isValidateCertificates();
 
+    /**
+     * Return trusted certificates in PEM format if configured.
+     *
+     * @return the trusted certificates or an empty string.
+     */
+    String getTrustedCertificates();
+
     /**
      * Gets the SASL mechanism of Hono-connection (Kafka specific property).
      *
@@ -77,6 +84,11 @@ enum HonoConfigValue implements KnownConfigValue {
          */
         VALIDATE_CERTIFICATES("validate-certificates", false),
 
+        /**
+         * Trusted certificates String property ("ca").
+         */
+        TRUSTED_CERTIFICATES("ca", ""),
+
         /**
          * SASL mechanism for connections of type Hono.
          */

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/hono/DefaultHonoConnectionFactory.java

@@ -54,6 +54,11 @@ public boolean isValidateCertificates() {
         return honoConfig.isValidateCertificates();
     }
 
+    @Override
+    protected String getTrustedCertificates() {
+        return honoConfig.getTrustedCertificates();
+    }
+
     @Override
     public HonoConfig.SaslMechanism getSaslMechanism() {
         return honoConfig.getSaslMechanism();

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/hono/HonoConnectionFactory.java

@@ -127,6 +127,7 @@ public Connection getHonoConnection(final Connection connection) {
         return ConnectivityModelFactory.newConnectionBuilder(connection)
                 .uri(combineUriWithCredentials(String.valueOf(getBaseUri()), getCredentials()))
                 .validateCertificate(isValidateCertificates())
+                .trustedCertificates(getTrustedCertificates())
                 .specificConfig(makeupSpecificConfig(connection))
                 .setSources(makeupSources(connection.getSources()))
                 .setTargets(makeupTargets(connection.getTargets()))
@@ -160,6 +161,8 @@ protected String getHonoTenantId() {
 
     protected abstract boolean isValidateCertificates();
 
+    protected abstract String getTrustedCertificates();
+
     protected abstract HonoConfig.SaslMechanism getSaslMechanism();
 
     protected abstract Set<URI> getBootstrapServerUris();

connectivity/service/src/main/resources/connectivity.conf

@@ -79,6 +79,9 @@ ditto {
       validate-certificates = false
       validate-certificates = ${?HONO_CONNECTION_VALIDATE_CERTIFICATE}
 
+      ca = ""
+      ca = ${?HONO_CONNECTION_CA}
+
       sasl-mechanism = "PLAIN"
       sasl-mechanism = ${?HONO_CONNECTION_SASL_MECHANISM}
 

connectivity/service/src/test/java/org/eclipse/ditto/connectivity/service/config/DefaultHonoConfigTest.java

@@ -135,6 +135,25 @@ public void isValidateCertificatesReturnsExplicitlyConfiguredValue() {
         assertThat(defaultHonoConfig.isValidateCertificates()).isEqualTo(validateCertificates);
     }
 
+    @Test
+    public void getTrustedCertificatesReturnsDefaultValueIfNotContainedInConfig() {
+        final var defaultHonoConfig = new DefaultHonoConfig(getActorSystem(ConfigFactory.empty()));
+
+        assertThat(defaultHonoConfig.getTrustedCertificates())
+                .isEqualTo(HonoConfig.HonoConfigValue.TRUSTED_CERTIFICATES.getDefaultValue());
+    }
+
+    @Test
+    public void getTrustedCertificatesReturnsExplicitlyConfiguredValue() {
+        final var trustedCertificates = "certs";
+        final var defaultHonoConfig = new DefaultHonoConfig(getActorSystem(ConfigFactory.parseMap(
+                Map.of(getFullQualifiedConfigKey(HonoConfig.HonoConfigValue.TRUSTED_CERTIFICATES),
+                        trustedCertificates)
+        )));
+
+        assertThat(defaultHonoConfig.getTrustedCertificates()).isEqualTo(trustedCertificates);
+    }
+
     @Test
     public void newInstanceThrowsDittoConfigErrorIfSaslMechanismIsUnknown() {
         assertThatExceptionOfType(DittoConfigError.class)

connectivity/service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/hono/DefaultHonoConnectionFactoryTest.java

@@ -143,6 +143,7 @@ private Connection getExpectedAdaptedHonoConnection(final Connection originalCon
                                 + ":" + URLEncoder.encode(honoConfig.getUserPasswordCredentials().getPassword(), StandardCharsets.UTF_8)
                                 + "@$2"))
                 .validateCertificate(honoConfig.isValidateCertificates())
+                .trustedCertificates(honoConfig.getTrustedCertificates())
                 .specificConfig(Map.of(
                         "saslMechanism", honoConfig.getSaslMechanism().toString(),
                         "bootstrapServers", TEST_CONFIG.getString(HonoConfig.PREFIX + ".bootstrap-servers"),

connectivity/service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/persistence/ConnectionPersistenceActorTest.java

@@ -138,6 +138,9 @@ public final class ConnectionPersistenceActorTest extends WithMockServers {
                             "org.eclipse.ditto.connectivity.service.messaging.hono.DefaultHonoConnectionFactory"),
                     "ditto.connectivity.hono.base-uri", ConfigValueFactory.fromAnyRef("tcp://localhost:9922"),
                     "ditto.connectivity.hono.validate-certificates", ConfigValueFactory.fromAnyRef("false"),
+                    "ditto.connectivity.hono.ca",
+                    ConfigValueFactory.fromAnyRef(
+                            "-----BEGIN CERTIFICATE-----\n<trusted certificate>\n-----END CERTIFICATE-----"),
                     "ditto.connectivity.hono.sasl-mechanism", ConfigValueFactory.fromAnyRef("PLAIN"),
                     "ditto.connectivity.hono.bootstrap-servers",
                     ConfigValueFactory.fromAnyRef("tcp://server1:port1,tcp://server2:port2,tcp://server3:port3")

connectivity/service/src/test/resources/hono-connection-custom-expected-after-adaptation.json

@@ -156,6 +156,7 @@
   "clientCount": 1,
   "failoverEnabled": true,
   "validateCertificates": false,
+  "ca": "-----BEGIN CERTIFICATE-----\n<trusted certificate>\n-----END CERTIFICATE-----",
   "processorPoolSize": 5,
   "specificConfig": {
     "honoTenantId": "hono-tenant-id",

connectivity/service/src/test/resources/hono-connection-implicit-tenant-custom-expected-after-adaptation.json

@@ -156,6 +156,7 @@
   "clientCount": 1,
   "failoverEnabled": true,
   "validateCertificates": false,
+  "ca": "-----BEGIN CERTIFICATE-----\n<trusted certificate>\n-----END CERTIFICATE-----",
   "processorPoolSize": 5,
   "specificConfig": {
     "saslMechanism": "plain",

connectivity/service/src/test/resources/test.conf

@@ -100,6 +100,7 @@ ditto {
     hono {
         base-uri = "tcp://localhost:9922"
         validate-certificates = false
+        ca = "-----BEGIN CERTIFICATE-----\n<trusted certificate>\n-----END CERTIFICATE-----"
         sasl-mechanism = PLAIN
         bootstrap-servers = "tcp://server1:port1,tcp://server2:port2,tcp://server3:port3"
         username = test_username

documentation/src/main/resources/pages/ditto/connectivity-protocol-bindings-hono.md

@@ -16,12 +16,14 @@ The Hono connection is based on the Kafka connection and uses it behind the scen
 [Kafka connection documentation](connectivity-protocol-bindings-kafka2.html) is valid for the Hono connection too, 
 but with some exceptions, as described below.
 
-#### Important note
-A Hono connection is associated with _one_ Hono tenant. That means for each Hono tenant a separate Hono connection needs to be created.
-The tenant ID is used in the source and target connection addresses, representing the Kafka topics used by Hono for
-sending and receiving messages for this tenant.
-See below sections [Source addresses](#source-addresses), [Source reply target](#source-reply-target) and [Target Address](#target-address).
-The Hono tenant ID for the connection is defined in the [specific config](#specific-configuration-properties) `"honoTenantId"` property.
+
+{% include note.html
+   content="A Hono connection is associated with _one_ Hono tenant. That means for each Hono tenant a separate Hono connection
+            needs to be created. The tenant ID is used in the source and target connection addresses, representing the Kafka
+            topics used by Hono for sending and receiving messages for this tenant. See below sections [Source addresses](#source-addresses),
+            [Source reply target](#source-reply-target) and [Target Address](#target-address). The Hono tenant ID for the connection is defined in the
+            [specific config](#specific-configuration-properties) `honoTenantId` property."
+%}
 
 ## Specific Hono connection configuration
 
@@ -215,7 +217,8 @@ ditto {
       password = "honoPassword"
       sasl-mechanism = "PLAIN"
       bootstrap-servers = "localhost:9092"
-      validate-certificates = false
+      validateCertificates = true,
+      ca = "-----BEGIN CERTIFICATE-----\n<trusted certificate>\n-----END CERTIFICATE-----"
     }
   }
 }