Scram listeners (strimzi#755)

* Permit specification of SCRAM-SHA-512 authn in CR

This applies equally for both TLS and Plain listeners.

Also fix @Descrption "authorization" → "authentication"

* New exception type

* CO changes for SCRAM-SHA

* Change docker images for SCRAM SHA

* Remove sasl.mechanism from config

* Fix Builder package on KafkaListenerAuthenticationScramSha512

* Scram uo (strimzi#795)

* Add support for SCRAM-SHA-512 to KafkaUser CR

.checkstyle/suppressions.xml

@@ -21,6 +21,9 @@
     <suppress checks="ClassFanOutComplexity|JavaNCSS"
               files="io[/\\]strimzi[/\\]operator[/\\]cluster[/\\]operator[/\\]assembly[/\\]KafkaAssemblyOperatorTest.java"/>
 
+    <suppress checks="ClassFanOutComplexity"
+              files="io[/\\]strimzi[/\\]operator[/\\]cluster[/\\]model[/\\]KafkaCluster.java"/>
+
     <!-- topic operator -->
     <suppress checks="NPathComplexity|CyclomaticComplexity"
               files="io[/\\]strimzi[/\\]operator[/\\]topic[/\\]TopicOperator.java"/>

Makefile

@@ -3,7 +3,7 @@ RELEASE_VERSION ?= latest
 CHART_PATH ?= ./helm-charts/strimzi-kafka-operator/
 CHART_SEMANTIC_RELEASE_VERSION ?= $(shell cat ./release.version | tr A-Z a-z)
 
-SUBDIRS=docker-images crd-generator api certificate-manager operator-common cluster-operator topic-operator user-operator kafka-init helm-charts examples
+SUBDIRS=docker-images test crd-generator api certificate-manager operator-common cluster-operator topic-operator user-operator kafka-init helm-charts examples
 DOCKER_TARGETS=docker_build docker_push docker_tag
 
 all: $(SUBDIRS)

api/pom.xml

@@ -70,6 +70,10 @@
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.strimzi</groupId>
+            <artifactId>test</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.strimzi</groupId>
             <artifactId>crd-generator</artifactId>

api/src/main/java/io/strimzi/api/kafka/model/KafkaListenerAuthentication.java

@@ -4,25 +4,25 @@
  */
 package io.strimzi.api.kafka.model;
 
-import io.strimzi.crdgenerator.annotations.Description;
-
-import java.io.Serializable;
-import java.util.HashMap;
-import java.util.Map;
-
 import com.fasterxml.jackson.annotation.JsonAnyGetter;
 import com.fasterxml.jackson.annotation.JsonAnySetter;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.strimzi.crdgenerator.annotations.Description;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
 
 /**
- * Configures the broker authorization
+ * Configures listener authentication.
  */
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 @JsonSubTypes({
         @JsonSubTypes.Type(name = KafkaListenerAuthenticationTls.TYPE_TLS, value = KafkaListenerAuthenticationTls.class),
+        @JsonSubTypes.Type(name = KafkaListenerAuthenticationScramSha512.SCRAM_SHA_512, value = KafkaListenerAuthenticationScramSha512.class),
 })
 @JsonInclude(JsonInclude.Include.NON_NULL)
 public abstract class KafkaListenerAuthentication implements Serializable {
@@ -31,7 +31,7 @@ public abstract class KafkaListenerAuthentication implements Serializable {
     private Map<String, Object> additionalProperties;
 
     @Description("Authentication type. " +
-            "Currently the only supported type is `tls`. " +
+            "`scram-sha-512` type uses SASL SCRAM-SHA-512 Authentication. " +
             "`tls` type uses TLS Client Authentication. " +
             "`tls` type is supported only on TLS listeners.")
     @JsonIgnore

api/src/main/java/io/strimzi/api/kafka/model/KafkaListenerAuthenticationScramSha512.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018, 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;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import io.strimzi.crdgenerator.annotations.Description;
+import io.sundr.builder.annotations.Buildable;
+
+/**
+ * Configures a listener to use SASL SCRAM-SHA-512 for authentication.
+ */
+@Buildable(
+        editableEnabled = false,
+        generateBuilderPackage = false,
+        builderPackage = "io.fabric8.kubernetes.api.builder"
+)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class KafkaListenerAuthenticationScramSha512 extends KafkaListenerAuthentication {
+
+    private static final long serialVersionUID = 1L;
+
+    public static final String SCRAM_SHA_512 = "scram-sha-512";
+
+    @Description("Must be `" + SCRAM_SHA_512 + "`")
+    @Override
+    public String getType() {
+        return SCRAM_SHA_512;
+    }
+}

api/src/main/java/io/strimzi/api/kafka/model/KafkaListenerAuthenticationTls.java

@@ -10,7 +10,7 @@
 import io.sundr.builder.annotations.Buildable;
 
 /**
- * Configures the broker authorization
+ * Configures a listener to use mutual TLS authentication.
  */
 @Buildable(
         editableEnabled = false,

api/src/main/java/io/strimzi/api/kafka/model/KafkaListenerPlain.java

@@ -4,15 +4,16 @@
  */
 package io.strimzi.api.kafka.model;
 
-import java.io.Serializable;
-import java.util.HashMap;
-import java.util.Map;
-
 import com.fasterxml.jackson.annotation.JsonAnyGetter;
 import com.fasterxml.jackson.annotation.JsonAnySetter;
 import com.fasterxml.jackson.annotation.JsonInclude;
+import io.strimzi.crdgenerator.annotations.Description;
 import io.sundr.builder.annotations.Buildable;
 
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
 import static java.util.Collections.emptyMap;
 
 /**
@@ -29,6 +30,18 @@ public class KafkaListenerPlain implements Serializable {
 
     private Map<String, Object> additionalProperties;
 
+    private KafkaListenerAuthentication authentication;
+
+    @Description("Authentication configuration for this listener. " +
+            "Since this listener does not use TLS transport you cannot configure an authentication with `type: tls`.")
+    public KafkaListenerAuthentication getAuthentication() {
+        return authentication;
+    }
+
+    public void setAuthentication(KafkaListenerAuthentication authentication) {
+        this.authentication = authentication;
+    }
+
     @JsonAnyGetter
     public Map<String, Object> getAdditionalProperties() {
         return this.additionalProperties != null ? this.additionalProperties : emptyMap();

api/src/main/java/io/strimzi/api/kafka/model/KafkaListenerTls.java

@@ -4,17 +4,17 @@
  */
 package io.strimzi.api.kafka.model;
 
+import com.fasterxml.jackson.annotation.JsonAnyGetter;
+import com.fasterxml.jackson.annotation.JsonAnySetter;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import io.strimzi.crdgenerator.annotations.Description;
+import io.sundr.builder.annotations.Buildable;
 
 import java.io.Serializable;
 import java.util.HashMap;
 import java.util.Map;
 
-import com.fasterxml.jackson.annotation.JsonAnyGetter;
-import com.fasterxml.jackson.annotation.JsonAnySetter;
-import com.fasterxml.jackson.annotation.JsonInclude;
-import io.sundr.builder.annotations.Buildable;
-
 import static java.util.Collections.emptyMap;
 
 /**
@@ -29,17 +29,18 @@
 public class KafkaListenerTls implements Serializable {
     private static final long serialVersionUID = 1L;
 
-    private KafkaListenerAuthentication serverAuthentication;
+    private KafkaListenerAuthentication auth;
     private Map<String, Object> additionalProperties;
 
-    @Description("Authorization configuration for Kafka brokers")
+    @Description("Authentication configuration for this listener.")
     @JsonInclude(JsonInclude.Include.NON_NULL)
-    public KafkaListenerAuthentication getAuthentication() {
-        return serverAuthentication;
+    @JsonProperty("authentication")
+    public KafkaListenerAuthentication getAuth() {
+        return auth;
     }
 
-    public void setAuthentication(KafkaListenerAuthentication serverAuthentication) {
-        this.serverAuthentication = serverAuthentication;
+    public void setAuth(KafkaListenerAuthentication auth) {
+        this.auth = auth;
     }
 
     @JsonAnyGetter

api/src/main/java/io/strimzi/api/kafka/model/KafkaUserAuthentication.java

@@ -4,31 +4,30 @@
  */
 package io.strimzi.api.kafka.model;
 
-import io.strimzi.crdgenerator.annotations.Description;
-
-import java.io.Serializable;
-import java.util.HashMap;
-import java.util.Map;
-
 import com.fasterxml.jackson.annotation.JsonAnyGetter;
 import com.fasterxml.jackson.annotation.JsonAnySetter;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.strimzi.crdgenerator.annotations.Description;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
 
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 @JsonSubTypes({
         @JsonSubTypes.Type(name = KafkaUserTlsClientAuthentication.TYPE_TLS, value = KafkaUserTlsClientAuthentication.class),
+        @JsonSubTypes.Type(name = KafkaUserScramSha512ClientAuthentication.TYPE_SCRAM_SHA_512, value = KafkaUserScramSha512ClientAuthentication.class),
 })
 @JsonInclude(JsonInclude.Include.NON_NULL)
 public abstract class KafkaUserAuthentication implements Serializable {
     private static final long serialVersionUID = 1L;
 
     private Map<String, Object> additionalProperties;
 
-    @Description("Authentication type. " +
-            "Currently the only supported type is `tls` for TLS Client Authentication.")
+    @Description("Authentication type.")
     @JsonIgnore
     public abstract String getType();
 

api/src/main/java/io/strimzi/api/kafka/model/KafkaUserScramSha512ClientAuthentication.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2018, 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;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import io.strimzi.crdgenerator.annotations.Description;
+import io.sundr.builder.annotations.Buildable;
+
+@Buildable(
+        editableEnabled = false,
+        generateBuilderPackage = false,
+        builderPackage = "io.fabric8.kubernetes.api.builder"
+)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class KafkaUserScramSha512ClientAuthentication extends KafkaUserAuthentication {
+    private static final long serialVersionUID = 1L;
+
+    public static final String TYPE_SCRAM_SHA_512 = "scram-sha-512";
+
+    @Description("Must be `" + TYPE_SCRAM_SHA_512 + "`")
+    @Override
+    public String getType() {
+        return TYPE_SCRAM_SHA_512;
+    }
+
+
+}

api/src/test/resources/io/strimzi/api/kafka/model/Kafka.out.yaml

@@ -14,7 +14,9 @@ spec:
       size: "500Gi"
       deleteClaim: false
     listeners:
-      plain: {}
+      plain:
+        authentication:
+          type: "scram-sha-512"
       tls:
         authentication:
           type: "tls"

api/src/test/resources/io/strimzi/api/kafka/model/Kafka.yaml

@@ -37,7 +37,9 @@ spec:
     config:
       min.insync.replicas: 3
     listeners:
-      plain: {}
+      plain:
+        authentication:
+          type: scram-sha-512
       tls:
         authentication:
           type: tls

certificate-manager/pom.xml

@@ -40,10 +40,7 @@
     </dependency>
     <dependency>
       <groupId>io.strimzi</groupId>
-      <artifactId>api</artifactId>
-      <classifier>tests</classifier>
-      <type>test-jar</type>
-      <scope>test</scope>
+      <artifactId>test</artifactId>
     </dependency>
   </dependencies>
 

cluster-operator/pom.xml

@@ -94,10 +94,7 @@
         </dependency>
         <dependency>
             <groupId>io.strimzi</groupId>
-            <artifactId>api</artifactId>
-            <classifier>tests</classifier>
-            <type>test-jar</type>
-            <scope>test</scope>
+            <artifactId>test</artifactId>
         </dependency>
         <dependency>
             <groupId>io.strimzi</groupId>

cluster-operator/src/main/java/io/strimzi/operator/cluster/InvalidConfigParameterException.java

@@ -5,7 +5,9 @@
 
 package io.strimzi.operator.cluster;
 
-public class InvalidConfigParameterException extends RuntimeException {
+import io.strimzi.operator.cluster.model.InvalidResourceException;
+
+public class InvalidConfigParameterException extends InvalidResourceException {
 
     private String key;
     public InvalidConfigParameterException(String key, String message) {

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

@@ -0,0 +1,14 @@
+/*
+ * Copyright 2018, 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;
+
+public class InvalidResourceException extends RuntimeException {
+    public InvalidResourceException() {
+        super();
+    }
+
+    protected InvalidResourceException(String s) {
+    }
+}

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

@@ -71,8 +71,13 @@ public class KafkaCluster extends AbstractModel {
     protected static final String RACK_VOLUME_MOUNT = "/opt/kafka/rack";
     private static final String ENV_VAR_KAFKA_INIT_RACK_TOPOLOGY_KEY = "RACK_TOPOLOGY_KEY";
     private static final String ENV_VAR_KAFKA_INIT_NODE_NAME = "NODE_NAME";
+    /** {@code TRUE} when the CLIENT listener (PLAIN transport) should be enabled*/
     private static final String ENV_VAR_KAFKA_CLIENT_ENABLED = "KAFKA_CLIENT_ENABLED";
+    /** The authentication to configure for the CLIENT listener (PLAIN transport). */
+    private static final String ENV_VAR_KAFKA_CLIENT_AUTHENTICATION = "KAFKA_CLIENT_AUTHENTICATION";
+    /** {@code TRUE} when the CLIENTTLS listener (TLS transport) should be enabled*/
     private static final String ENV_VAR_KAFKA_CLIENTTLS_ENABLED = "KAFKA_CLIENTTLS_ENABLED";
+    /** The authentication to configure for the CLIENTTLS listener (TLS transport) . */
     private static final String ENV_VAR_KAFKA_CLIENTTLS_AUTHENTICATION = "KAFKA_CLIENTTLS_AUTHENTICATION";
     private static final String ENV_VAR_KAFKA_AUTHORIZATION_TYPE = "KAFKA_AUTHORIZATION_TYPE";
     private static final String ENV_VAR_KAFKA_AUTHORIZATION_SUPER_USERS = "KAFKA_AUTHORIZATION_SUPER_USERS";
@@ -243,7 +248,14 @@ public static KafkaCluster fromCrd(CertManager certManager, Kafka kafkaAssembly,
         result.generateCertificates(certManager, secrets);
         result.setTlsSidecar(kafkaClusterSpec.getTlsSidecar());
 
-        result.setListeners(kafkaClusterSpec.getListeners());
+        KafkaListeners listeners = kafkaClusterSpec.getListeners();
+        if (listeners != null) {
+            if (listeners.getPlain() != null
+                && listeners.getPlain().getAuthentication() instanceof KafkaListenerAuthenticationTls) {
+                throw new InvalidResourceException("You cannot configure TLS authentication on a plain listener.");
+            }
+        }
+        result.setListeners(listeners);
         result.setAuthorization(kafkaClusterSpec.getAuthorization());
 
         return result;
@@ -619,13 +631,17 @@ protected List<EnvVar> getEnvVars() {
         if (listeners != null)  {
             if (listeners.getPlain() != null)   {
                 varList.add(buildEnvVar(ENV_VAR_KAFKA_CLIENT_ENABLED, "TRUE"));
+
+                if (listeners.getPlain().getAuthentication() != null) {
+                    varList.add(buildEnvVar(ENV_VAR_KAFKA_CLIENT_AUTHENTICATION, listeners.getPlain().getAuthentication().getType()));
+                }
             }
 
             if (listeners.getTls() != null) {
                 varList.add(buildEnvVar(ENV_VAR_KAFKA_CLIENTTLS_ENABLED, "TRUE"));
 
-                if (listeners.getTls().getAuthentication() != null && KafkaListenerAuthenticationTls.TYPE_TLS.equals(listeners.getTls().getAuthentication().getType())) {
-                    varList.add(buildEnvVar(ENV_VAR_KAFKA_CLIENTTLS_AUTHENTICATION, KafkaListenerAuthenticationTls.TYPE_TLS));
+                if (listeners.getTls().getAuth() != null) {
+                    varList.add(buildEnvVar(ENV_VAR_KAFKA_CLIENTTLS_AUTHENTICATION, listeners.getTls().getAuth().getType()));
                 }
             }
         }