Issue #1060: add dynamic loading of HTTP request signing algorithms.

Signed-off-by: Yufei Cai <yufei.cai@bosch.io>

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

@@ -13,7 +13,10 @@
 package org.eclipse.ditto.connectivity.service.config;
 
 import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 import javax.annotation.concurrent.Immutable;
 
@@ -36,6 +39,7 @@ final class DefaultHttpPushConfig implements HttpPushConfig {
     private final int maxQueueSize;
     private final Duration requestTimeout;
     private final HttpProxyConfig httpProxyConfig;
+    private final Map<String, String> hmacAlgorithms;
 
     private DefaultHttpPushConfig(final ScopedConfig config) {
         maxQueueSize = config.getInt(ConfigValue.MAX_QUEUE_SIZE.getConfigPath());
@@ -44,6 +48,7 @@ private DefaultHttpPushConfig(final ScopedConfig config) {
             throw new DittoConfigError("Request timeout must be greater than 0");
         }
         httpProxyConfig = DefaultHttpProxyConfig.ofProxy(config);
+        hmacAlgorithms = asStringMap(config.getConfig(ConfigValue.HMAC_ALGORITHMS.getConfigPath()));
     }
 
     static DefaultHttpPushConfig of(final Config config) {
@@ -65,6 +70,11 @@ public HttpProxyConfig getHttpProxyConfig() {
         return httpProxyConfig;
     }
 
+    @Override
+    public Map<String, String> getHmacAlgorithms() {
+        return hmacAlgorithms;
+    }
+
     @Override
     public boolean equals(final Object o) {
         if (this == o) {
@@ -75,19 +85,33 @@ public boolean equals(final Object o) {
         }
         final DefaultHttpPushConfig that = (DefaultHttpPushConfig) o;
         return maxQueueSize == that.maxQueueSize &&
-                Objects.equals(httpProxyConfig, that.httpProxyConfig);
+                Objects.equals(requestTimeout, that.requestTimeout) &&
+                Objects.equals(httpProxyConfig, that.httpProxyConfig) &&
+                Objects.equals(hmacAlgorithms, that.hmacAlgorithms);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(maxQueueSize, httpProxyConfig);
+        return Objects.hash(maxQueueSize, httpProxyConfig, hmacAlgorithms, requestTimeout);
     }
 
     @Override
     public String toString() {
         return getClass().getSimpleName() + " [" +
                 "maxQueueSize=" + maxQueueSize +
+                ", requestTimeout=" + requestTimeout +
                 ", httpProxyConfig=" + httpProxyConfig +
+                ", hmacAlgorithms=" + hmacAlgorithms +
                 "]";
     }
+
+    private static Map<String, String> asStringMap(final Config config) {
+        try {
+            final Map<String, String> map = config.root().entrySet().stream()
+                    .collect(Collectors.toMap(Map.Entry::getKey, entry -> (String) entry.getValue().unwrapped()));
+            return Collections.unmodifiableMap(map);
+        } catch (final ClassCastException e) {
+            throw new DittoConfigError("In HttpPushConfig, hmac-algorithms must be a map from string to string.");
+        }
+    }
 }

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

@@ -13,6 +13,7 @@
 package org.eclipse.ditto.connectivity.service.config;
 
 import java.time.Duration;
+import java.util.Map;
 
 import org.eclipse.ditto.base.service.config.http.HttpProxyConfig;
 import org.eclipse.ditto.internal.utils.config.KnownConfigValue;
@@ -37,6 +38,11 @@ public interface HttpPushConfig {
      */
     HttpProxyConfig getHttpProxyConfig();
 
+    /**
+     * @return configuration of HMAC request-signing algorithms.
+     */
+    Map<String, String> getHmacAlgorithms();
+
     /**
      * An enumeration of the known config path expressions and their associated default values for
      * {@code HttpPushConfig}.
@@ -52,7 +58,12 @@ enum ConfigValue implements KnownConfigValue {
          * Maximum time a request is allowed to wait for a response. If this time is exceeded the HTTP connection will
          * be re-opened.
          */
-        REQUEST_TIMEOUT("request-timeout", Duration.ofSeconds(60));
+        REQUEST_TIMEOUT("request-timeout", Duration.ofSeconds(60)),
+
+        /**
+         * HMAC request-signing algorithms.
+         */
+        HMAC_ALGORITHMS("hmac-algorithms", Map.of());
 
         private final String path;
         private final Object defaultValue;

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/httppush/AwsRequestSigning.java

@@ -51,7 +51,7 @@ final class AwsRequestSigning implements RequestSigning {
     private static final String HOST_HEADER = "host";
     private static final DateTimeFormatter DATE_STAMP_FORMATTER =
             DateTimeFormatter.ofPattern("yyyyMMdd").withZone(ZoneId.of("Z"));
-    private static final DateTimeFormatter X_AMZ_DATE_FORMATTER =
+    static final DateTimeFormatter X_AMZ_DATE_FORMATTER =
             DateTimeFormatter.ofPattern("yyyyMMdd'T'HHmmssz").withZone(ZoneId.of("Z"));
 
     private final ActorSystem actorSystem;

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/httppush/AwsRequestSigningFactory.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright (c) 2021 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.connectivity.service.messaging.httppush;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.eclipse.ditto.connectivity.model.HmacCredentials;
+import org.eclipse.ditto.json.JsonArray;
+import org.eclipse.ditto.json.JsonFieldDefinition;
+import org.eclipse.ditto.json.JsonObject;
+import org.eclipse.ditto.json.JsonValue;
+
+import akka.actor.ActorSystem;
+
+/**
+ * Creator of the signing process {@code AWS4-HMAC-SHA256}.
+ */
+public final class AwsRequestSigningFactory implements RequestSigningFactory {
+
+    /**
+     * Token timeout to evaluate the body of outgoing requests, which should take very little time as it does not
+     * depend on IO.
+     */
+    private static final Duration TIMEOUT = Duration.ofSeconds(10);
+
+    /**
+     * Which header to sign by default.
+     */
+    private static final List<String> DEFAULT_CANONICAL_HEADERS = List.of("host");
+
+    @Override
+    public RequestSigning create(final ActorSystem actorSystem, final HmacCredentials credentials) {
+        final JsonObject parameters = credentials.getParameters();
+        final String region = parameters.getValueOrThrow(JsonFields.REGION);
+        final String service = parameters.getValueOrThrow(JsonFields.SERVICE);
+        final String accessKey = parameters.getValueOrThrow(JsonFields.ACCESS_KEY);
+        final String secretKey = parameters.getValueOrThrow(JsonFields.SECRET_KEY);
+        final boolean doubleEncode = parameters.getValue(JsonFields.DOUBLE_ENCODE).orElse(true);
+        final List<String> canonicalHeaders = parameters.getValue(JsonFields.CANONICAL_HEADERS)
+                .map(array -> array.stream().map(JsonValue::asString).collect(Collectors.toList()))
+                .orElse(DEFAULT_CANONICAL_HEADERS);
+        return new AwsRequestSigning(actorSystem, canonicalHeaders, region, service, accessKey, secretKey, doubleEncode,
+                TIMEOUT);
+    }
+
+    /**
+     * JSON fields of algorithm parameters.
+     */
+    public static final class JsonFields {
+
+        /**
+         * Obligatory: The AWS region of the signed requests.
+         */
+        public static JsonFieldDefinition<String> REGION = JsonFieldDefinition.ofString("region");
+
+        /**
+         * Obligatory: The service for which the signed requests are intended.
+         */
+        public static JsonFieldDefinition<String> SERVICE = JsonFieldDefinition.ofString("service");
+
+        /**
+         * Obligatory: Access key to sign requests with.
+         */
+        public static JsonFieldDefinition<String> ACCESS_KEY = JsonFieldDefinition.ofString("accessKey");
+
+        /**
+         * Obligatory: Secret key to sign requests with.
+         */
+        public static JsonFieldDefinition<String> SECRET_KEY = JsonFieldDefinition.ofString("secretKey");
+
+        /**
+         * Optional: Whether to double-encode and normalize path segments. True by default. Set to false for S3.
+         */
+        public static JsonFieldDefinition<Boolean> DOUBLE_ENCODE = JsonFieldDefinition.ofBoolean("doubleEncode");
+
+        /**
+         * Optional: Which headers to sign. They differ for each AWS service. By default only "host" is signed.
+         */
+        public static JsonFieldDefinition<JsonArray> CANONICAL_HEADERS =
+                JsonFieldDefinition.ofJsonArray("canonicalHeaders");
+    }
+}

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/httppush/AzMonitorRequestSigning.java

@@ -37,7 +37,8 @@
 final class AzMonitorRequestSigning implements RequestSigning {
 
     private static final String X_MS_DATE_HEADER = "x-ms-date";
-    private static final DateTimeFormatter X_MS_DATE_FORMAT =
+
+    static final DateTimeFormatter X_MS_DATE_FORMAT =
             DateTimeFormatter.ofPattern("EEE, dd MMM uuuu HH:mm:ss zzz", Locale.ENGLISH).withZone(ZoneId.of("GMT"));
 
     private final ActorSystem actorSystem;

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/httppush/AzMonitorRequestSigningFactory.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2021 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.connectivity.service.messaging.httppush;
+
+import java.time.Duration;
+
+import org.eclipse.ditto.connectivity.model.HmacCredentials;
+import org.eclipse.ditto.json.JsonFieldDefinition;
+import org.eclipse.ditto.json.JsonObject;
+
+import akka.actor.ActorSystem;
+
+/**
+ * Creator of the signing process for Azure Monitor Data Collector.
+ */
+public final class AzMonitorRequestSigningFactory implements RequestSigningFactory {
+
+    /**
+     * Token timeout to evaluate the body of outgoing requests, which should take very little time as it does not
+     * depend on IO.
+     */
+    private static final Duration TIMEOUT = Duration.ofSeconds(10);
+
+    @Override
+    public RequestSigning create(final ActorSystem actorSystem, final HmacCredentials credentials) {
+        final JsonObject parameters = credentials.getParameters();
+        final String workspaceId = parameters.getValueOrThrow(JsonFields.WORKSPACE_ID);
+        final String sharedKey = parameters.getValueOrThrow(JsonFields.SHARED_KEY);
+        return AzMonitorRequestSigning.of(actorSystem, workspaceId, sharedKey, TIMEOUT);
+    }
+
+    /**
+     * JSON fields of algorithm parameters.
+     */
+    public static final class JsonFields {
+
+        /**
+         * Obligatory: The Azure workspace ID of the signed requests.
+         */
+        public static JsonFieldDefinition<String> WORKSPACE_ID = JsonFieldDefinition.ofString("workspaceId");
+
+        /**
+         * Obligatory: The shared key with which to sign requests.
+         */
+        public static JsonFieldDefinition<String> SHARED_KEY = JsonFieldDefinition.ofString("sharedKey");
+    }
+}