Add support for type: custom client authentication (#11760)

Signed-off-by: Jakub Scholz <www@scholzj.com>

Author: scholzj

CHANGELOG.md

@@ -7,6 +7,7 @@
 * Make properties `broker.session.timeout.ms`, `broker.heartbeat.interval.ms` and `controller.socket.timeout.ms` configurable
 * Add monitoring of custom resources using [kubernetes-state-metrics (KSM)](https://github.com/kubernetes/kube-state-metrics) (see [Strimzi proposal 087](https://github.com/strimzi/proposals/blob/main/087-monitoring-of-custom-resources.md))
 * Ignore users (their ACLs, Quotas and SCRAM-SHA-512 credentials) managed by some other tools based on a configurable pattern in User Operator
+* Support for `type: custom` client authentication (to make it easier to use custom authentication mechanisms such as AWS IAM)
 * Added support for Strimzi Metrics Reporter to Kafka Connect, Mirror Maker 2 and Kafka Bridge.
 * Add new feature gate `ServerSideApplyPhase1` (disabled by default) that adds support for Server Side Apply for `ConfigMap`, `Ingress`, `PVC`, `Service`, and `ServiceAccount` according to [Strimzi Proposal #105](https://github.com/strimzi/proposals/blob/main/105-server-side-apply-implementation-fg-timelines.md).
 * Added distinction between changes of "cluster-wide" broker properties applied dynamically at cluster level, and "per-broker" broker properties applied dynamically at broker level.

api/src/main/java/io/strimzi/api/kafka/model/common/authentication/KafkaClientAuthentication.java

@@ -27,20 +27,21 @@
     @JsonSubTypes.Type(name = KafkaClientAuthenticationScramSha512.TYPE_SCRAM_SHA_512, value = KafkaClientAuthenticationScramSha512.class),
     @JsonSubTypes.Type(name = KafkaClientAuthenticationPlain.TYPE_PLAIN, value = KafkaClientAuthenticationPlain.class),
     @JsonSubTypes.Type(name = KafkaClientAuthenticationOAuth.TYPE_OAUTH, value = KafkaClientAuthenticationOAuth.class),
+    @JsonSubTypes.Type(name = KafkaClientAuthenticationCustom.TYPE_CUSTOM, value = KafkaClientAuthenticationCustom.class),
 })
 @JsonInclude(JsonInclude.Include.NON_NULL)
 @EqualsAndHashCode
 @ToString
 public abstract class KafkaClientAuthentication implements UnknownPropertyPreserving {
     private Map<String, Object> additionalProperties;
 
-    @Description("Authentication type. " +
-            "Currently the supported types are `tls`, `scram-sha-256`, `scram-sha-512`, `plain`, and 'oauth'. " +
-            "`scram-sha-256` and `scram-sha-512` types use SASL SCRAM-SHA-256 and SASL SCRAM-SHA-512 Authentication, respectively. " +
-            "`plain` type uses SASL PLAIN Authentication. " +
-            "`oauth` type uses SASL OAUTHBEARER Authentication. " +
-            "The `tls` type uses TLS Client Authentication. " +
-            "The `tls` type is supported only over TLS connections.")
+    @Description("Specifies the authentication type. " +
+            "Supported types are `tls`, `scram-sha-256`, `scram-sha-512`, `plain`, 'oauth', and `custom`. " +
+            "`tls` uses TLS client authentication and is supported only over TLS connections. " +
+            "`scram-sha-256` and `scram-sha-512` use SASL SCRAM-SHA-256 and SASL SCRAM-SHA-512 authentication, respectively. " +
+            "`plain` uses SASL PLAIN authentication. " +
+            "`oauth` uses SASL OAUTHBEARER authentication. " +
+            "`custom` allows you to configure a custom authentication mechanism.")
     public abstract String getType();
 
     @Override

api/src/main/java/io/strimzi/api/kafka/model/common/authentication/KafkaClientAuthenticationCustom.java

@@ -0,0 +1,65 @@
+/*
+ * 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.common.authentication;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import io.strimzi.api.kafka.model.common.Constants;
+import io.strimzi.crdgenerator.annotations.Description;
+import io.strimzi.crdgenerator.annotations.DescriptionFile;
+import io.sundr.builder.annotations.Buildable;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
+import java.util.Map;
+
+/**
+ * Configures the Kafka client authentication using a custom mechanism in client based components
+ */
+@DescriptionFile
+@Buildable(
+        editableEnabled = false,
+        builderPackage = Constants.FABRIC8_KUBERNETES_API
+)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonPropertyOrder({"type", "sasl", "config"})
+@EqualsAndHashCode(callSuper = true)
+@ToString(callSuper = true)
+public class KafkaClientAuthenticationCustom extends KafkaClientAuthentication {
+    public static final String ALLOWED_PREFIXES = "ssl.keystore., sasl.";
+    public static final String TYPE_CUSTOM = "custom";
+
+    private Map<String, Object> config;
+    private boolean sasl;
+
+    @Description("Must be `" + TYPE_CUSTOM + "`")
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Override
+    public String getType() {
+        return TYPE_CUSTOM;
+    }
+
+    @Description("Enable or disable SASL on this authentication mechanism.")
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public boolean isSasl() {
+        return sasl;
+    }
+
+    public void setSasl(boolean enabled) {
+        this.sasl = enabled;
+    }
+
+    @Description("Configuration for the custom authentication mechanism. " +
+            "Only properties with the `sasl.` and `ssl.keystore.` prefixes are allowed. " +
+            "Specify other options in the regular configuration section of the custom resource.")
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public Map<String, Object> getConfig() {
+        return config;
+    }
+
+    public void setConfig(Map<String, Object> config) {
+        this.config = config;
+    }
+}

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

@@ -33,6 +33,19 @@ public AbstractConfiguration(AbstractConfiguration configuration)   {
         options.addMapPairs(configuration.asOrderedProperties().asMap());
     }
 
+    /**
+     * 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.
+     * @param allowedPrefixes            List with configuration key prefixes which are allowed. All other keys will be ignored.
+     */
+    public AbstractConfiguration(Reconciliation reconciliation, Iterable<Map.Entry<String, Object>> jsonOptions, List<String> allowedPrefixes) {
+        options.addIterablePairs(jsonOptions);
+        filterAllowed(reconciliation, allowedPrefixes);
+    }
+
     /**
      * Constructor used to instantiate this class from JsonObject. Should be used to create configuration from
      * ConfigMap / CRD.
@@ -71,7 +84,7 @@ public AbstractConfiguration(Reconciliation reconciliation, String configuration
     }
 
     /**
-     * Filters forbidden values from the configuration.
+     * Filters forbidden values from the configuration based on a list of forbidden prefixes.
      *
      * @param reconciliation            The reconciliation
      * @param forbiddenPrefixes         List with configuration key prefixes which are not allowed. All keys which start
@@ -114,6 +127,27 @@ private void filterForbidden(Reconciliation reconciliation, List<String> forbidd
         }
     }
 
+    /**
+     * Filters forbidden values from the configuration based on a list of allowed prefixes.
+     *
+     * @param reconciliation            The reconciliation
+     * @param allowedPrefixes           List with configuration key prefixes which are allowed. All other keys will be ignored.
+     */
+    private void filterAllowed(Reconciliation reconciliation, List<String> allowedPrefixes)   {
+        options.filter(k -> {
+            boolean allowed = allowedPrefixes.stream().anyMatch(p -> k.toLowerCase(Locale.ENGLISH).startsWith(p));
+
+            if (allowed) {
+                LOGGER.traceCr(reconciliation, "Configuration option \"{}\" is allowed and will be passed to the assembly", k);
+            } else {
+                LOGGER.warnCr(reconciliation, "Configuration option \"{}\" is not allowed and will be ignored", k);
+            }
+
+            // We have to return the reverted value to filter out the not allowed options
+            return !allowed;
+        });
+    }
+
     /**
      * Returns a value for a specific config option
      *

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

@@ -12,6 +12,7 @@
 import io.strimzi.api.kafka.model.common.GenericSecretSource;
 import io.strimzi.api.kafka.model.common.PasswordSecretSource;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthentication;
+import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationCustom;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationOAuth;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationPlain;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationScram;
@@ -27,6 +28,7 @@
 
 import java.io.PrintWriter;
 import java.io.StringWriter;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 import static io.strimzi.api.kafka.model.common.metrics.StrimziMetricsReporter.TYPE_STRIMZI_METRICS_REPORTER;
@@ -39,6 +41,7 @@
  * configuration file. This class is using the builder pattern to make it easy to test the different parts etc. To
  * generate the configuration file, it is using the PrintWriter.
  */
+@SuppressWarnings("checkstyle:CyclomaticComplexity")
 public class KafkaBridgeConfigurationBuilder {
 
     // placeholders expanded through config providers inside the bridge node
@@ -149,8 +152,22 @@ public KafkaBridgeConfigurationBuilder withAuthentication(KafkaClientAuthenticat
                 writer.println("kafka.ssl.keystore.location=/tmp/strimzi/bridge.keystore.p12");
                 writer.println("kafka.ssl.keystore.password=" + PLACEHOLDER_CERT_STORE_PASSWORD_CONFIG_PROVIDER_ENV_VAR);
                 writer.println("kafka.ssl.keystore.type=PKCS12");
-            // otherwise SASL or OAuth is going to be used for authentication
-            } else {
+            } else if (authentication instanceof KafkaClientAuthenticationCustom customAuth) { // Configure custom authentication
+                if (customAuth.isSasl())    {
+                    // If this authentication uses SASL, we need to update the security protocol to combine the SASL
+                    // flag with the SSL or PLAINTEXT flag.
+                    securityProtocol = securityProtocol.equals("SSL") ? "SASL_SSL" : "SASL_PLAINTEXT";
+                }
+
+                Map<String, Object> customConfig = customAuth.getConfig();
+                if (customConfig == null) {
+                    customConfig = Map.of();
+                }
+
+                KafkaClientAuthenticationCustomConfiguration config = new KafkaClientAuthenticationCustomConfiguration(reconciliation, customConfig.entrySet());
+                config.asOrderedProperties().asMap().forEach((key, value) -> writer.println(String.format("kafka.%s=%s", key, value)));
+                writer.println();
+            } else { // otherwise SASL or OAuth is going to be used for authentication
                 securityProtocol = securityProtocol.equals("SSL") ? "SASL_SSL" : "SASL_PLAINTEXT";
                 String saslMechanism = null;
                 StringBuilder jaasConfig = new StringBuilder();
@@ -211,6 +228,7 @@ public KafkaBridgeConfigurationBuilder withAuthentication(KafkaClientAuthenticat
                 writer.println();
             }
         }
+
         return this;
     }
 

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

@@ -0,0 +1,32 @@
+/*
+ * 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.common.authentication.KafkaClientAuthenticationCustom;
+import io.strimzi.operator.common.Reconciliation;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Class for handling configuration of the custom client-side authentication
+ */
+public class KafkaClientAuthenticationCustomConfiguration extends AbstractConfiguration {
+    private static final List<String> ALLOWED_PREFIXES;
+    static {
+        ALLOWED_PREFIXES = AbstractConfiguration.splitPrefixesOrOptionsToList(KafkaClientAuthenticationCustom.ALLOWED_PREFIXES);
+    }
+
+    /**
+     * Constructor
+     *
+     * @param reconciliation    Reconciliation marker
+     * @param jsonOptions       Configuration options
+     */
+    public KafkaClientAuthenticationCustomConfiguration(Reconciliation reconciliation, Iterable<Map.Entry<String, Object>> jsonOptions) {
+        super(reconciliation, jsonOptions, ALLOWED_PREFIXES);
+    }
+}

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

@@ -8,6 +8,7 @@
 import io.strimzi.api.kafka.model.common.GenericSecretSource;
 import io.strimzi.api.kafka.model.common.PasswordSecretSource;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthentication;
+import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationCustom;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationOAuth;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationPlain;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationScram;
@@ -24,6 +25,7 @@
 import java.io.StringWriter;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 import static io.strimzi.operator.cluster.model.KafkaConnectCluster.OAUTH_SECRETS_BASE_VOLUME_MOUNT;
@@ -36,6 +38,7 @@
  * configuration file. This class is using the builder pattern to make it easy to test the different parts etc. To
  * generate the configuration file, it is using the PrintWriter.
  */
+@SuppressWarnings("checkstyle:CyclomaticComplexity")
 public class KafkaConnectConfigurationBuilder {
     // the volume mounted secret file template includes: <volume_mount>/<secret_name>/<secret_key>
     private static final String PLACEHOLDER_VOLUME_MOUNTED_SECRET_TEMPLATE_CONFIG_PROVIDER_DIR = "${strimzidir:%s%s:%s}";
@@ -148,8 +151,28 @@ public KafkaConnectConfigurationBuilder withAuthentication(KafkaClientAuthentica
                 writer.println("admin.ssl.keystore.certificate.chain=" + certConfigProviderValue);
                 writer.println("admin.ssl.keystore.key=" + keyConfigProviderValue);
                 writer.println("admin.ssl.keystore.type=PEM");
-                // otherwise SASL or OAuth is going to be used for authentication
-            } else {
+            } else if (authentication instanceof KafkaClientAuthenticationCustom customAuth) { // Configure custom authentication
+                if (customAuth.isSasl())    {
+                    // If this authentication uses SASL, we need to update the security protocol to combine the SASL
+                    // flag with the SSL or PLAINTEXT flag.
+                    securityProtocol = securityProtocol.equals("SSL") ? "SASL_SSL" : "SASL_PLAINTEXT";
+                }
+
+                Map<String, Object> customConfig = customAuth.getConfig();
+                if (customConfig == null) {
+                    customConfig = Map.of();
+                }
+
+                KafkaClientAuthenticationCustomConfiguration config = new KafkaClientAuthenticationCustomConfiguration(reconciliation, customConfig.entrySet());
+                config.asOrderedProperties().asMap().forEach((key, value) -> {
+                    writer.println(String.format("%s=%s", key, value));
+                    writer.println(String.format("producer.%s=%s", key, value));
+                    writer.println(String.format("consumer.%s=%s", key, value));
+                    writer.println(String.format("admin.%s=%s", key, value));
+                });
+
+                writer.println();
+            } else { // otherwise SASL or OAuth is going to be used for authentication
                 securityProtocol = securityProtocol.equals("SSL") ? "SASL_SSL" : "SASL_PLAINTEXT";
                 String saslMechanism = null;
                 StringBuilder jaasConfig = new StringBuilder();
@@ -232,6 +255,7 @@ public KafkaConnectConfigurationBuilder withAuthentication(KafkaClientAuthentica
                 writer.println();
             }
         }
+
         return this;
     }
 

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

@@ -4,6 +4,7 @@
  */
 package io.strimzi.operator.cluster.model;
 
+import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationCustom;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationOAuth;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationPlain;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationScram;
@@ -180,8 +181,8 @@ public List<KafkaConnector> generateConnectorDefinitions()    {
         Map<String, Object> config = new HashMap<>(connector.getConfig());
 
         // Source and target cluster configurations
-        addClusterToMirrorMaker2ConnectorConfig(config, targetCluster, TARGET_CLUSTER_PREFIX);
-        addClusterToMirrorMaker2ConnectorConfig(config, sourceCluster, SOURCE_CLUSTER_PREFIX);
+        addClusterToMirrorMaker2ConnectorConfig(reconciliation, config, targetCluster, TARGET_CLUSTER_PREFIX);
+        addClusterToMirrorMaker2ConnectorConfig(reconciliation, config, sourceCluster, SOURCE_CLUSTER_PREFIX);
 
         // Topics pattern
         if (mirror.getTopicsPattern() != null) {
@@ -251,7 +252,7 @@ public List<KafkaConnector> generateConnectorDefinitions()    {
         return config;
     }
 
-    /* test */ static void addClusterToMirrorMaker2ConnectorConfig(Map<String, Object> config, KafkaMirrorMaker2ClusterSpec cluster, String configPrefix) {
+    /* test */ static void addClusterToMirrorMaker2ConnectorConfig(Reconciliation reconciliation, Map<String, Object> config, KafkaMirrorMaker2ClusterSpec cluster, String configPrefix) {
         config.put(configPrefix + "alias", cluster.getAlias());
         config.put(configPrefix + AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.getBootstrapServers());
 
@@ -284,6 +285,20 @@ public List<KafkaConnector> generateConnectorDefinitions()    {
                 config.put(configPrefix + SaslConfigs.SASL_JAAS_CONFIG,
                         oauthJaasConfig(cluster, oauthAuthentication));
                 config.put(configPrefix + SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, "io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler");
+            } else if (cluster.getAuthentication() instanceof KafkaClientAuthenticationCustom customAuth) { // Configure custom authentication
+                if (customAuth.isSasl()) {
+                    // If this authentication uses SASL, we need to update the security protocol to combine the SASL
+                    // flag with the SSL or PLAINTEXT flag.
+                    securityProtocol = securityProtocol.equals("SSL") ? "SASL_SSL" : "SASL_PLAINTEXT";
+                }
+
+                Map<String, Object> customConfig = customAuth.getConfig();
+                if (customConfig == null) {
+                    customConfig = Map.of();
+                }
+
+                KafkaClientAuthenticationCustomConfiguration authConfig = new KafkaClientAuthenticationCustomConfiguration(reconciliation, customConfig.entrySet());
+                authConfig.asOrderedProperties().asMap().forEach((key, value) -> config.put(configPrefix + key, value));
             }
         }