eclipse-ditto#985 Add support to establish a connection via an ssh tu…

…nnel. Ssh tunnel management is done in new SshTunnelActor. The tunnel is controlled (open/close) with the existing state machine in BaseClientActor. Add connecting via tunnel for existing protocols (except Kafka, which requires multiple connections to bottstrap server and zookeeper).

Signed-off-by: Dominik Guggemos <dominik.guggemos@bosch.io>

bom/pom.xml

@@ -43,6 +43,7 @@
         <akka-management.version>1.0.10</akka-management.version>
         <hivemq-mqtt-client.version>1.2.1</hivemq-mqtt-client.version>
         <kafka-clients.version>2.5.1</kafka-clients.version>
+        <sshd.version>2.6.0</sshd.version>
 
         <!-- Keep these version consistent with akka-persistence-mongo.version's build.sbt -->
         <mongo-java-driver.version>4.1.1</mongo-java-driver.version>
@@ -329,6 +330,12 @@
                 <version>${kafka-clients.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>org.apache.sshd</groupId>
+                <artifactId>sshd-core</artifactId>
+                <version>${sshd.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>io.netty</groupId>
                 <artifactId>netty-bom</artifactId>

model/connectivity/src/main/java/org/eclipse/ditto/model/connectivity/ConnectivityModelFactory.java

@@ -240,6 +240,27 @@ public static ResourceStatus newClientStatus(final String client,
                 inStateSince);
     }
 
+
+    /**
+     * Returns a new ssh tunnel {@code ResourceStatus}.
+     *
+     * @param client a client identifier e.g. on which node this client is running
+     * @param status the ConnectionStatus of the source metrics to create
+     * @param statusDetails the optional details about the connection status
+     * @param inStateSince the instant since the resource is in the given state
+     * @return a new AddressMetric which is initialised with the extracted data from {@code jsonObject}.
+     * @throws NullPointerException if any parameter is {@code null}.
+     */
+    public static ResourceStatus newSshTunnelStatus(final String client,
+            final ConnectivityStatus status,
+            @Nullable final String statusDetails,
+            final Instant inStateSince) {
+
+        return ImmutableResourceStatus.of(ResourceStatus.ResourceType.SSH_TUNNEL, client, status, null, statusDetails,
+                inStateSince);
+    }
+
+
     /**
      * Returns a new source {@code ResourceStatus}.
      *
@@ -795,14 +816,15 @@ public static SshTunnel sshTunnelFromJson(final JsonObject jsonObject) {
      *
      * @param enabled sets if the ssh tunnel is active
      * @param credentials the credentials of the ssh tunnel
+     * @param validateHost {@code true} if host validation is enabled
      * @param knownHosts the known hosts of the ssh tunnel
      * @param uri the uri of the ssh tunnel
      * @return the created {@link org.eclipse.ditto.model.connectivity.SshTunnel}
      * @since 2.0.0
      */
     public static SshTunnel newSshTunnel(final Boolean enabled, final Credentials credentials,
-            final List<String> knownHosts, final String uri) {
-        return new ImmutableSshTunnel.Builder(enabled, credentials, knownHosts, uri).build();
+            final boolean validateHost, final List<String> knownHosts, final String uri) {
+        return new ImmutableSshTunnel.Builder(enabled, credentials, validateHost, knownHosts, uri).build();
     }
 
     /**

model/connectivity/src/main/java/org/eclipse/ditto/model/connectivity/ImmutableSshTunnel.java

@@ -40,20 +40,23 @@ final class ImmutableSshTunnel implements SshTunnel {
 
     private final Boolean enabled;
     private final Credentials credentials;
+    private final boolean validateHost;
     private final List<String> knownHosts;
     private final String uri;
 
     private ImmutableSshTunnel(final Builder builder) {
         enabled = checkNotNull(builder.enabled, "enabled");
         credentials = checkNotNull(builder.credentials, "credentials");
+        validateHost = builder.validateHost;
         knownHosts = Collections.unmodifiableList(new ArrayList<>(checkNotNull(builder.knownHosts, "knownHosts")));
         uri = checkNotNull(builder.uri, "uri");
     }
 
-    private ImmutableSshTunnel(final Boolean enabled, final Credentials credentials,
+    private ImmutableSshTunnel(final Boolean enabled, final Credentials credentials, final boolean validateHost,
             final List<String> knownHosts, final String uri) {
         this.enabled = checkNotNull(enabled, "enabled");
         this.credentials = checkNotNull(credentials, "credentials");
+        this.validateHost = validateHost;
         this.knownHosts = Collections.unmodifiableList(new ArrayList<>(checkNotNull(knownHosts, "knownHosts")));
         this.uri = checkNotNull(uri, "uri");
     }
@@ -63,17 +66,19 @@ private ImmutableSshTunnel(final Boolean enabled, final Credentials credentials,
      *
      * @param enabled the enabled status.
      * @param credentials the credentials.
+     * @param validateHost {@code true} if host validation is enabled
      * @param knownHosts the known hosts.
      * @param uri the URI.
      * @return new instance of {@code SshTunnelBuilder}.
      * @throws NullPointerException if any argument is {@code null}.
      */
     public static SshTunnelBuilder getBuilder(final Boolean enabled,
             final Credentials credentials,
+            final boolean validateHost,
             final List<String> knownHosts,
             final String uri) {
 
-        return new Builder(enabled, credentials, knownHosts, uri);
+        return new Builder(enabled, credentials, validateHost, knownHosts, uri);
     }
 
     /**
@@ -103,6 +108,11 @@ public Credentials getCredentials() {
         return credentials;
     }
 
+    @Override
+    public boolean isValidateHost() {
+        return validateHost;
+    }
+
     @Override
     public List<String> getKnownHosts() {
         return knownHosts;
@@ -123,7 +133,8 @@ public String getUri() {
      */
     public static SshTunnel fromJson(final JsonObject jsonObject) {
         checkNotNull(jsonObject, "ssh tunnel");
-        return new Builder(extractEnabled(jsonObject), extractCredentials(jsonObject), extractKnownHosts(jsonObject),
+        return new Builder(extractEnabled(jsonObject), extractCredentials(jsonObject), extractValidateHost(jsonObject),
+                extractKnownHosts(jsonObject),
                 extractUri(jsonObject)).build();
     }
 
@@ -142,6 +153,10 @@ private static List<String> extractKnownHosts(final JsonObject jsonObject) {
                 .collect(Collectors.toList());
     }
 
+    private static boolean extractValidateHost(final JsonObject jsonObject) {
+        return jsonObject.getValueOrThrow(JsonFields.VALIDATE_HOST);
+    }
+
     private static String extractUri(final JsonObject jsonObject) {
         return jsonObject.getValueOrThrow(JsonFields.URI);
     }
@@ -155,6 +170,7 @@ public JsonObject toJson(final JsonSchemaVersion schemaVersion, final Predicate<
 
         jsonObjectBuilder.set(JsonFields.ENABLED, enabled);
         jsonObjectBuilder.set(JsonFields.CREDENTIALS, credentials.toJson());
+        jsonObjectBuilder.set(JsonFields.VALIDATE_HOST, validateHost);
         jsonObjectBuilder.set(JsonFields.KNOWN_HOSTS, JsonArray.of(knownHosts));
         jsonObjectBuilder.set(JsonFields.URI, uri);
         return jsonObjectBuilder.build();
@@ -172,20 +188,22 @@ public boolean equals(@Nullable final Object o) {
         final ImmutableSshTunnel that = (ImmutableSshTunnel) o;
         return Objects.equals(enabled, that.enabled) &&
                 Objects.equals(credentials, that.credentials) &&
+                Objects.equals(validateHost, that.validateHost) &&
                 Objects.equals(knownHosts, that.knownHosts) &&
                 Objects.equals(uri, that.uri);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(enabled, credentials, knownHosts, uri);
+        return Objects.hash(enabled, credentials, validateHost, knownHosts, uri);
     }
 
     @Override
     public String toString() {
         return getClass().getSimpleName() + " [" +
                 "enabled=" + enabled +
                 ", credentials=" + credentials +
+                ", validateHost=" + validateHost +
                 ", knownHosts=" + knownHosts +
                 ", uri=" + uri +
                 "]";
@@ -203,6 +221,7 @@ static final class Builder implements SshTunnelBuilder {
         private String uri;
 
         // optional with Default:
+        private boolean validateHost = false;
         private List<String> knownHosts = new ArrayList<>();
 
         Builder(final Boolean enabled, final Credentials credentials, final String uri) {
@@ -211,17 +230,20 @@ static final class Builder implements SshTunnelBuilder {
             this.uri = uri;
         }
 
-        Builder(final Boolean enabled, final Credentials credentials, final List<String> knownHosts, final String uri) {
+        Builder(final Boolean enabled, final Credentials credentials, final boolean validateHost,
+                final List<String> knownHosts, final String uri) {
             this.enabled = enabled;
             this.credentials = credentials;
             this.uri = uri;
+            this.validateHost = validateHost;
             this.knownHosts = knownHosts;
         }
 
         Builder(final SshTunnel sshTunnel) {
             this.enabled = sshTunnel.isSshTunnelActive();
             this.credentials = sshTunnel.getCredentials();
             this.uri = sshTunnel.getUri();
+            this.validateHost = sshTunnel.isValidateHost();
             this.knownHosts = sshTunnel.getKnownHosts();
         }
 
@@ -237,6 +259,12 @@ public SshTunnelBuilder credentials(final Credentials credentials) {
             return this;
         }
 
+        @Override
+        public SshTunnelBuilder validateHost(final boolean validateHost) {
+            this.validateHost = validateHost;
+            return this;
+        }
+
         @Override
         public SshTunnelBuilder knownHosts(final List<String> knownHosts) {
             this.knownHosts = knownHosts;

model/connectivity/src/main/java/org/eclipse/ditto/model/connectivity/ResourceStatus.java

@@ -98,6 +98,11 @@ enum ResourceType implements CharSequence {
          */
         CLIENT("client"),
 
+        /**
+         * An ssh tunnel.
+         */
+        SSH_TUNNEL("ssh"),
+
         /**
          * Unknown resource type.
          */

model/connectivity/src/main/java/org/eclipse/ditto/model/connectivity/SshTunnel.java

@@ -43,6 +43,11 @@ public interface SshTunnel extends Jsonifiable.WithFieldSelectorAndPredicate<Jso
      */
     Credentials getCredentials();
 
+    /**
+     * @return {@code true} if host should be validated
+     */
+    boolean isValidateHost();
+
     /**
      * @return the known hosts for the ssh tunnel
      */
@@ -77,35 +82,37 @@ final class JsonFields {
          */
         public static final JsonFieldDefinition<Integer> SCHEMA_VERSION =
                 JsonFactory.newIntFieldDefinition(JsonSchemaVersion.getJsonKey(), FieldType.SPECIAL, FieldType.HIDDEN,
-                        JsonSchemaVersion.V_1, JsonSchemaVersion.V_2);
+                        JsonSchemaVersion.V_2);
 
         /**
          * JSON field containing the {@code SshTunnel} enabled.
          */
         public static final JsonFieldDefinition<Boolean> ENABLED =
-                JsonFactory.newBooleanFieldDefinition("enabled", FieldType.REGULAR, JsonSchemaVersion.V_1,
-                        JsonSchemaVersion.V_2);
+                JsonFactory.newBooleanFieldDefinition("enabled", FieldType.REGULAR, JsonSchemaVersion.V_2);
 
         /**
          * JSON field containing the {@code Credentials} for {@code SshTunnel}.
          */
         public static final JsonFieldDefinition<JsonObject> CREDENTIALS =
-                JsonFactory.newJsonObjectFieldDefinition("credentials", FieldType.REGULAR, JsonSchemaVersion.V_1,
-                        JsonSchemaVersion.V_2);
+                JsonFactory.newJsonObjectFieldDefinition("credentials", FieldType.REGULAR, JsonSchemaVersion.V_2);
+
+        /**
+         * JSON field containing the {@code SshTunnel} known hosts.
+         */
+        public static final JsonFieldDefinition<Boolean> VALIDATE_HOST =
+                JsonFactory.newBooleanFieldDefinition("validateHost", FieldType.REGULAR, JsonSchemaVersion.V_2);
 
         /**
          * JSON field containing the {@code SshTunnel} known hosts.
          */
         public static final JsonFieldDefinition<JsonArray> KNOWN_HOSTS =
-                JsonFactory.newJsonArrayFieldDefinition("knownHosts", FieldType.REGULAR,
-                        JsonSchemaVersion.V_1, JsonSchemaVersion.V_2);
+                JsonFactory.newJsonArrayFieldDefinition("knownHosts", FieldType.REGULAR, JsonSchemaVersion.V_2);
 
         /**
          * JSON field containing the {@code SshTunnel} uri.
          */
         public static final JsonFieldDefinition<String> URI =
-                JsonFactory.newStringFieldDefinition("uri", FieldType.REGULAR,
-                        JsonSchemaVersion.V_2);
+                JsonFactory.newStringFieldDefinition("uri", FieldType.REGULAR, JsonSchemaVersion.V_2);
 
         JsonFields() {
             throw new AssertionError();

model/connectivity/src/main/java/org/eclipse/ditto/model/connectivity/SshTunnelBuilder.java

@@ -44,6 +44,14 @@ public interface SshTunnelBuilder {
      */
     SshTunnelBuilder knownHosts(List<String> knownHosts);
 
+    /**
+     * Enables/disables host validation.
+     *
+     * @param validateHost {@code true} if host validation is enabled
+     * @return this builder
+     */
+    SshTunnelBuilder validateHost(boolean validateHost);
+
     /**
      * Sets the URI to use in the {@code SshTunnel}.
      *

model/connectivity/src/test/java/org/eclipse/ditto/model/connectivity/ImmutableConnectionTest.java

@@ -92,7 +92,7 @@ public final class ImmutableConnectionTest {
     private static final List<Target> TARGETS = Arrays.asList(TARGET1, TARGET2, TARGET3);
 
     private static final SshTunnel SSH_TUNNEL = ConnectivityModelFactory.newSshTunnel(true,
-            UserPasswordCredentials.newInstance("User", "Password"), Collections.emptyList(), URI);
+            UserPasswordCredentials.newInstance("User", "Password"), false, Collections.emptyList(), URI);
 
     private static final JsonArray KNOWN_SOURCES_JSON =
             SOURCES.stream().map(Source::toJson).collect(JsonCollectors.valuesToArray());

model/connectivity/src/test/java/org/eclipse/ditto/model/connectivity/ImmutableSshTunnelTest.java

@@ -20,6 +20,7 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.List;
 
 import org.eclipse.ditto.json.JsonArray;
 import org.eclipse.ditto.json.JsonFactory;
@@ -38,7 +39,7 @@ public final class ImmutableSshTunnelTest {
     private static final String PASSWORD = "Password";
 
     private static final SshTunnel SSH_TUNNEL_WITH_USERNAME_PW_CREDENTIALS_AND_EMPTY_KNOWN_HOSTS =
-            ConnectivityModelFactory.newSshTunnel(true, UserPasswordCredentials.newInstance(USER, PASSWORD),
+            ConnectivityModelFactory.newSshTunnel(true, UserPasswordCredentials.newInstance(USER, PASSWORD), false,
                     Collections.emptyList(), URI);
 
     private static final JsonObject SSH_TUNNEL_JSON_WITH_USERNAME_PW_CREDENTIALS_AND_EMPTY_KNOWN_HOSTS = JsonObject
@@ -49,6 +50,7 @@ public final class ImmutableSshTunnelTest {
                     .set(UserPasswordCredentials.JsonFields.USERNAME, USER)
                     .set(UserPasswordCredentials.JsonFields.PASSWORD, PASSWORD)
                     .build())
+            .set(SshTunnel.JsonFields.VALIDATE_HOST, false)
             .set(SshTunnel.JsonFields.KNOWN_HOSTS, JsonArray.empty())
             .set(SshTunnel.JsonFields.URI, URI)
             .build();
@@ -81,7 +83,7 @@ public void fromJsonReturnsExpected() {
 
     @Test
     public void addKnownHostsToExistingSshTunnel() {
-        final ArrayList<String> knownHosts = new ArrayList<>();
+        final List<String> knownHosts = new ArrayList<>();
         final String aKnownHost = "aKnownHost";
         knownHosts.add(aKnownHost);
         final SshTunnel sshTunnelWithKnownHosts = new ImmutableSshTunnel

services/connectivity/common/src/main/java/org/eclipse/ditto/services/connectivity/config/ConnectivityConfig.java

@@ -86,4 +86,11 @@ public interface ConnectivityConfig extends ServiceSpecificConfig, WithHealthChe
      * @return the config.
      */
     AcknowledgementConfig getAcknowledgementConfig();
+
+    /**
+     * Returns the configuration for ssh tunneling.
+     *
+     * @return the config.
+     */
+    TunnelConfig getTunnelConfig();
 }