Implement support for deploying KafkaBridge with TLS/SSL configuration (#12287)

Signed-off-by: Gantigmaa Selenge <tina.selenge@gmail.com>
Signed-off-by: Paolo Patierno <ppatierno@live.com>
Co-authored-by: Paolo Patierno <ppatierno@live.com>

Author: tinaselenge

CHANGELOG.md

@@ -35,6 +35,8 @@
 * The `ServerSideApplyPhase1` feature gate moves to beta stage and is enabled by default.
   If needed, `ServerSideApplyPhase1` can be disabled in the feature gates configuration in the Cluster Operator.
 * Fixed auto-rebalancing on scale up not running anymore when Cruise Control is not ready yet on the first attempt.
+* Add support for TLS/SSL on the HTTP Bridge
+  Set `spec.http.tls.certificateAndKey` configuration to enable it and provide the certificate and key via Secret.
 
 ### Major changes, deprecations, and removals
 

api/src/main/java/io/strimzi/api/kafka/model/bridge/KafkaBridgeHttpConfig.java

@@ -28,13 +28,14 @@
         builderPackage = Constants.FABRIC8_KUBERNETES_API
 )
 @JsonInclude(JsonInclude.Include.NON_NULL)
-@JsonPropertyOrder({"port", "cors"})
+@JsonPropertyOrder({"port", "tls", "cors"})
 @EqualsAndHashCode
 @ToString
 public class KafkaBridgeHttpConfig implements UnknownPropertyPreserving {
     public static final int HTTP_DEFAULT_PORT = 8080;
     public static final String HTTP_DEFAULT_HOST = "0.0.0.0";
     private int port = HTTP_DEFAULT_PORT;
+    private KafkaBridgeHttpTls tls;
     private KafkaBridgeHttpCors cors;
     private Map<String, Object> additionalProperties;
 
@@ -55,6 +56,16 @@ public void setPort(int port) {
         this.port = port;
     }
 
+    @Description("TLS configuration for clients connections to the HTTP Bridge.")
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public KafkaBridgeHttpTls getTls() {
+        return tls;
+    }
+
+    public void setTls(KafkaBridgeHttpTls tls) {
+        this.tls = tls;
+    }
+
     @Description("CORS configuration for the HTTP Bridge.")
     @JsonInclude(JsonInclude.Include.NON_NULL)
     public KafkaBridgeHttpCors getCors() {

api/src/main/java/io/strimzi/api/kafka/model/bridge/KafkaBridgeHttpTls.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.api.kafka.model.bridge;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import io.strimzi.api.kafka.model.common.CertAndKeySecretSource;
+import io.strimzi.api.kafka.model.common.Constants;
+import io.strimzi.api.kafka.model.common.UnknownPropertyPreserving;
+import io.strimzi.crdgenerator.annotations.Description;
+import io.sundr.builder.annotations.Buildable;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A representation of the TLS configuration for the HTTP.
+ */
+@Buildable(
+        editableEnabled = false,
+        builderPackage = Constants.FABRIC8_KUBERNETES_API
+)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonPropertyOrder({"certificateAndKey", "config"})
+@EqualsAndHashCode
+@ToString
+public class KafkaBridgeHttpTls implements UnknownPropertyPreserving {
+    public static final String FORBIDDEN_PREFIXES = "ssl.";
+    public static final String FORBIDDEN_PREFIX_EXCEPTIONS = "ssl.enabled.cipher.suites, ssl.enabled.protocols";
+
+    private CertAndKeySecretSource certificateAndKey = null;
+    private Map<String, Object> config = new HashMap<>(0);
+    private Map<String, Object> additionalProperties;
+
+    @Description("Reference to the `Secret` which holds the certificate and private key pair.")
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @JsonProperty(required = true)
+    public CertAndKeySecretSource getCertificateAndKey() {
+        return certificateAndKey;
+    }
+
+    public void setCertificateAndKey(CertAndKeySecretSource certificateAndKey) {
+        this.certificateAndKey = certificateAndKey;
+    }
+
+    @Description("Additional configuration for the HTTP server TLS. Properties with the following prefixes cannot be set: " + FORBIDDEN_PREFIXES + " (with the exception of: " + FORBIDDEN_PREFIX_EXCEPTIONS + ").")
+    @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    public Map<String, Object> getConfig() {
+        return config;
+    }
+
+    public void setConfig(Map<String, Object> config) {
+        this.config = config;
+    }
+
+    @Override
+    public Map<String, Object> getAdditionalProperties() {
+        return this.additionalProperties != null ? this.additionalProperties : Map.of();
+    }
+
+    @Override
+    public void setAdditionalProperty(String name, Object value) {
+        if (this.additionalProperties == null) {
+            this.additionalProperties = new HashMap<>(2);
+        }
+        this.additionalProperties.put(name, value);
+    }
+}

api/src/test/resources/crds/v1/046-Crd-kafkabridge.yaml

@@ -160,6 +160,33 @@ spec:
                     type: integer
                     minimum: 1023
                     description: Port the server listens on.
+                  tls:
+                    type: object
+                    properties:
+                      certificateAndKey:
+                        type: object
+                        properties:
+                          secretName:
+                            type: string
+                            description: The name of the Secret containing the certificate.
+                          certificate:
+                            type: string
+                            description: The name of the file certificate in the Secret.
+                          key:
+                            type: string
+                            description: "The name of the private key in the secret. The private key must be in unencrypted PKCS #8 format. For more information, see RFC 5208: https://datatracker.ietf.org/doc/html/rfc5208."
+                        required:
+                        - secretName
+                        - certificate
+                        - key
+                        description: Reference to the `Secret` which holds the certificate and private key pair.
+                      config:
+                        x-kubernetes-preserve-unknown-fields: true
+                        type: object
+                        description: "Additional configuration for the HTTP server TLS. Properties with the following prefixes cannot be set: ssl. (with the exception of: ssl.enabled.cipher.suites, ssl.enabled.protocols)."
+                    required:
+                    - certificateAndKey
+                    description: TLS configuration for clients connections to the HTTP Bridge.
                   cors:
                     type: object
                     properties:

api/src/test/resources/crds/v1beta2/046-Crd-kafkabridge.yaml

@@ -139,6 +139,27 @@ spec:
                   port:
                     type: integer
                     minimum: 1023
+                  tls:
+                    type: object
+                    properties:
+                      certificateAndKey:
+                        type: object
+                        properties:
+                          secretName:
+                            type: string
+                          certificate:
+                            type: string
+                          key:
+                            type: string
+                        required:
+                        - secretName
+                        - certificate
+                        - key
+                      config:
+                        x-kubernetes-preserve-unknown-fields: true
+                        type: object
+                    required:
+                    - certificateAndKey
                   cors:
                     type: object
                     properties:
@@ -1577,6 +1598,27 @@ spec:
                   port:
                     type: integer
                     minimum: 1023
+                  tls:
+                    type: object
+                    properties:
+                      certificateAndKey:
+                        type: object
+                        properties:
+                          secretName:
+                            type: string
+                          certificate:
+                            type: string
+                          key:
+                            type: string
+                        required:
+                        - secretName
+                        - certificate
+                        - key
+                      config:
+                        x-kubernetes-preserve-unknown-fields: true
+                        type: object
+                    required:
+                    - certificateAndKey
                   cors:
                     type: object
                     properties:

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaBridgeCluster.java

@@ -86,12 +86,12 @@ public class KafkaBridgeCluster extends AbstractModel implements SupportsLogging
      * HTTP port configuration
      */
     public static final int DEFAULT_REST_API_PORT = 8080;
-
     /* test */ static final String COMPONENT_TYPE = "kafka-bridge";
     protected static final String REST_API_PORT_NAME = "rest-api";
     /* test */ static final String REST_API_MANAGEMENT_PORT_NAME = "rest-api-mgmt";
     /* test */ static final int REST_API_MANAGEMENT_PORT = 8081;
     protected static final String TLS_CERTS_BASE_VOLUME_MOUNT = "/opt/strimzi/bridge-certs/";
+    protected static final String HTTP_SERVER_CERTS_BASE_VOLUME_MOUNT = "/opt/strimzi/bridge-http-certs/";
     protected static final String PASSWORD_VOLUME_MOUNT = "/opt/strimzi/bridge-password/";
     protected static final String ENV_VAR_KAFKA_INIT_INIT_FOLDER_KEY = "INIT_FOLDER";
     private static final String KAFKA_BRIDGE_CONFIG_VOLUME_NAME = "kafka-bridge-configurations";
@@ -246,8 +246,10 @@ private static void fromCrdTemplate(final KafkaBridgeCluster kafkaBridgeCluster,
      */
     public Service generateService() {
         int port = DEFAULT_REST_API_PORT;
+        boolean isTls = false;
         if (http != null) {
             port = http.getPort();
+            isTls = http.getTls() != null;
         }
 
         List<ServicePort> ports = new ArrayList<>();
@@ -263,7 +265,7 @@ public Service generateService() {
                 ports,
                 labels.strimziSelectorLabels(),
                 ModelUtils.getCustomLabelsOrAnnotations(CO_ENV_VAR_CUSTOM_SERVICE_LABELS),
-                Util.mergeLabelsOrAnnotations(getDiscoveryAnnotation(port), ModelUtils.getCustomLabelsOrAnnotations(CO_ENV_VAR_CUSTOM_SERVICE_ANNOTATIONS))
+                Util.mergeLabelsOrAnnotations(getDiscoveryAnnotation(port, isTls), ModelUtils.getCustomLabelsOrAnnotations(CO_ENV_VAR_CUSTOM_SERVICE_ANNOTATIONS))
         );
     }
 
@@ -272,15 +274,14 @@ public Service generateService() {
      *
      * @return  JSON with discovery annotation
      */
-    /*test*/ Map<String, String> getDiscoveryAnnotation(int port) {
+    /*test*/ Map<String, String> getDiscoveryAnnotation(int port, boolean isTls) {
         JsonArray anno = new JsonArray();
 
         JsonObject discovery = new JsonObject();
         discovery.put("port", port);
-        discovery.put("tls", false);
+        discovery.put("tls", isTls);
         discovery.put("auth", "none");
-        discovery.put("protocol", "http");
-
+        discovery.put("protocol", isTls ? "https" : "http");
         anno.add(discovery);
 
         JsonObject managementDiscovery = new JsonObject();
@@ -316,6 +317,19 @@ protected List<Volume> getVolumes(boolean isOpenShift) {
             CertUtils.createTrustedCertificatesVolumes(volumeList, tls.getTrustedCertificates(), isOpenShift);
         }
 
+        if (http != null && http.getTls() != null) {
+            if (http.getTls().getCertificateAndKey() != null) {
+                String secretName = http.getTls().getCertificateAndKey().getSecretName();
+                // skipping if a volume mount with same Secret name was already added
+                if (volumeList.stream().noneMatch(v -> v.getName().equals(secretName))) {
+                    volumeList.add(VolumeUtils.createSecretVolume(secretName, secretName, isOpenShift));
+                }
+            } else {
+                LOGGER.warnCr(reconciliation, "The spec.http.tls.certificateAndKey property is missing. This is required when TLS is enabled for HTTP Bridge.");
+                throw new InvalidResourceException("The TLS configuration for the HTTP is not specified.");
+            }
+        }
+
         if (rack != null) {
             volumeList.add(VolumeUtils.createEmptyDirVolume(INIT_VOLUME_NAME, "1Mi", "Memory"));
         }
@@ -337,6 +351,16 @@ protected List<VolumeMount> getVolumeMounts() {
             CertUtils.createTrustedCertificatesVolumeMounts(volumeMountList, tls.getTrustedCertificates(), TLS_CERTS_BASE_VOLUME_MOUNT);
         }
 
+        if (http != null && http.getTls() != null) {
+            if (http.getTls().getCertificateAndKey() != null) {
+                String secretName = http.getTls().getCertificateAndKey().getSecretName();
+                volumeMountList.add(VolumeUtils.createVolumeMount(secretName, HTTP_SERVER_CERTS_BASE_VOLUME_MOUNT + secretName));
+            } else {
+                LOGGER.warnCr(reconciliation, "The spec.http.tls.certificateAndKey property is missing. This is required when TLS is enabled for HTTP Bridge.");
+                throw new InvalidResourceException("The TLS configuration for the HTTP is not specified.");
+            }
+        }
+
         if (rack != null) {
             volumeMountList.add(VolumeUtils.createVolumeMount(INIT_VOLUME_NAME, INIT_VOLUME_MOUNT));
         }

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaBridgeConfigurationBuilder.java

@@ -29,6 +29,7 @@
 import java.util.Map;
 
 import static io.strimzi.api.kafka.model.common.metrics.StrimziMetricsReporter.TYPE_STRIMZI_METRICS_REPORTER;
+import static io.strimzi.operator.cluster.model.KafkaBridgeCluster.HTTP_SERVER_CERTS_BASE_VOLUME_MOUNT;
 import static io.strimzi.operator.cluster.model.KafkaBridgeCluster.KAFKA_BRIDGE_CONFIG_VOLUME_MOUNT;
 
 /**
@@ -294,6 +295,21 @@ public KafkaBridgeConfigurationBuilder withHttp(KafkaBridgeHttpConfig http, Kafk
         printSectionHeader("HTTP configuration");
         writer.println("http.host=" + KafkaBridgeHttpConfig.HTTP_DEFAULT_HOST);
         writer.println("http.port=" + (http != null ? http.getPort() : KafkaBridgeHttpConfig.HTTP_DEFAULT_PORT));
+
+        if (http != null && http.getTls() != null) {
+            writer.println("http.ssl.enable=true");
+
+            if (http.getTls().getCertificateAndKey() != null) {
+                writer.println("http.ssl.certificate.location=" + HTTP_SERVER_CERTS_BASE_VOLUME_MOUNT + http.getTls().getCertificateAndKey().getSecretName() + "/" + http.getTls().getCertificateAndKey().getCertificate());
+                writer.println("http.ssl.key.location=" + HTTP_SERVER_CERTS_BASE_VOLUME_MOUNT + http.getTls().getCertificateAndKey().getSecretName() + "/" + http.getTls().getCertificateAndKey().getKey());
+            }
+
+            if (!http.getTls().getConfig().isEmpty()) {
+                KafkaBridgeHttpTlsConfiguration tlsConfig = new KafkaBridgeHttpTlsConfiguration(reconciliation, http.getTls().getConfig().entrySet());
+                tlsConfig.asOrderedProperties().asMap().forEach((key, value) -> writer.println("http." + key + "=" + value));
+            }
+        }
+
         if (http != null && http.getCors() != null) {
             writer.println("http.cors.enabled=true");
 

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaBridgeHttpTlsConfiguration.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+
+package io.strimzi.operator.cluster.model;
+
+import io.strimzi.api.kafka.model.bridge.KafkaBridgeHttpTls;
+import io.strimzi.operator.common.Reconciliation;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Class for handling Kafka Bridge HTTP TLS configuration passed by the user
+ */
+public class KafkaBridgeHttpTlsConfiguration extends AbstractConfiguration {
+
+    private static final List<String> FORBIDDEN_PREFIXES;
+    private static final List<String> FORBIDDEN_PREFIX_EXCEPTIONS;
+
+    static {
+        FORBIDDEN_PREFIXES = AbstractConfiguration.splitPrefixesOrOptionsToList(KafkaBridgeHttpTls.FORBIDDEN_PREFIXES);
+        FORBIDDEN_PREFIX_EXCEPTIONS = AbstractConfiguration.splitPrefixesOrOptionsToList(KafkaBridgeHttpTls.FORBIDDEN_PREFIX_EXCEPTIONS);
+    }
+
+    /**
+     * Constructor used to instantiate this class from JsonObject. Should be used to create configuration from
+     * ConfigMap / CRD.
+     *
+     * @param reconciliation  The reconciliation
+     * @param jsonOptions     Json object with configuration options as key ad value pairs.
+     */
+    public KafkaBridgeHttpTlsConfiguration(Reconciliation reconciliation, Iterable<Map.Entry<String, Object>> jsonOptions) {
+        super(reconciliation, jsonOptions, FORBIDDEN_PREFIXES, FORBIDDEN_PREFIX_EXCEPTIONS, List.of(), Map.of());
+    }
+}