Repository: knative-sandbox/eventing-kafka-broker
Pull request #685 (2021-03-01 09:41:27)

[release-0.21] Improve SSL support
This is an automated cherry-pick of #631

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **matzew** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @matzew` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-0.21/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["matzew"]} -->
google-cla[bot]: All (the pull request submitter and all commit authors) CLAs are signed, **but** one or more commits were authored or co-authored by someone other than the pull request submitter.

We need to confirm that all authors are ok with their commits being contributed to this project.  Please have them confirm that by leaving a comment that contains only `@googlebot I consent.` in this pull request.

*Note to project maintainer:* There may be cases where the author cannot leave a comment, or the comment is not properly detected as consent.  In those cases, you can manually confirm consent of the commit author(s), and set the `cla` label to `yes` (if enabled on your project).

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fknative-sandbox%2Feventing-kafka-broker%2Fpull%2F685) for more info**.

<!-- need_author_consent -->
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/685#pullrequestreview-600562727" title="Approved">slinkydeveloper</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-0.21/OWNERS)~~ [slinkydeveloper]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
pierDipi: googlebot I consent.
pierDipi: @googlebot I consent.

Patch:
diff --git a/control-plane/pkg/security/secret.go b/control-plane/pkg/security/secret.go
index efc78ed2e..822c26d89 100644
--- a/control-plane/pkg/security/secret.go
+++ b/control-plane/pkg/security/secret.go
@@ -20,6 +20,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
+	"strconv"
 
 	"github.com/Shopify/sarama"
 )
@@ -31,6 +32,7 @@ const (
 
 	UserCertificate = "user.crt"
 	UserKey         = "user.key"
+	UserSkip        = "user.skip" // default: false
 
 	SaslMechanismKey = "sasl.mechanism"
 	SaslUserKey      = "user"
@@ -149,32 +151,54 @@ func saslConfig(protocol string, data map[string][]byte) ConfigOption {
 func sslConfig(protocol string, data map[string][]byte) ConfigOption {
 	return func(config *sarama.Config) error {
 
-		caCert, ok := data[CaCertificateKey]
-		if !ok {
-			return fmt.Errorf("[protocol %s] required CA certificate (key: %s)", protocol, CaCertificateKey)
+		// If there is no CaCertificateKey, we assume that we can use system's root CA set.
+		var certPool *x509.CertPool
+		if caCert, ok := data[CaCertificateKey]; ok {
+			certPool = x509.NewCertPool()
+			if !certPool.AppendCertsFromPEM(caCert) {
+				return fmt.Errorf(
+					"[protocol %s] failed to parse CA certificates (key: %s)",
+					protocol, CaCertificateKey,
+				)
+			}
 		}
 
 		var tlsCerts []tls.Certificate
 		if protocol == ProtocolSSL {
-			userKeyCert, ok := data[UserKey]
-			if !ok {
-				return fmt.Errorf("[protocol %s] required user key (key: %s)", protocol, UserKey)
-			}
 
-			userCert, ok := data[UserCertificate]
-			if !ok {
-				return fmt.Errorf("[protocol %s] required user key (key: %s)", protocol, UserCertificate)
-			}
+			// When protocol=SSL we might have 2 options:
+			//	- client auth is required on the broker side (mTLS)
+			//	- client auth is not required on the broker side
 
-			tlsCert, err := tls.X509KeyPair(userCert, userKeyCert)
+			skipClientAuth, err := skipClientAuthCheck(data)
 			if err != nil {
-				return fmt.Errorf("[protocol %s] failed to load x.509 key pair: %w", protocol, err)
+				return fmt.Errorf("[protocol %s] %w", protocol, err)
 			}
-			tlsCerts = []tls.Certificate{tlsCert}
-		}
 
-		certPool := x509.NewCertPool()
-		certPool.AppendCertsFromPEM(caCert)
+			if !skipClientAuth {
+				userKeyCert, ok := data[UserKey]
+				if !ok {
+					return fmt.Errorf(
+						`[protocol %s] required user key (key: %s) - use "%s: true" to disable client auth`,
+						protocol, UserKey, UserSkip,
+					)
+				}
+
+				userCert, ok := data[UserCertificate]
+				if !ok {
+					return fmt.Errorf(
+						`[protocol %s] required user key (key: %s) - use "%s: true" to disable client auth`,
+						protocol, UserCertificate, UserSkip,
+					)
+				}
+
+				tlsCert, err := tls.X509KeyPair(userCert, userKeyCert)
+				if err != nil {
+					return fmt.Errorf("[protocol %s] failed to load x.509 key pair: %w", protocol, err)
+				}
+				tlsCerts = []tls.Certificate{tlsCert}
+			}
+		}
 
 		config.Net.TLS.Enable = true
 		config.Net.TLS.Config = &tls.Config{
@@ -187,3 +211,15 @@ func sslConfig(protocol string, data map[string][]byte) ConfigOption {
 		return nil
 	}
 }
+
+func skipClientAuthCheck(data map[string][]byte) (bool, error) {
+	clientAuth, ok := data[UserSkip]
+	if !ok {
+		return false, nil // Client auth is required by default.
+	}
+	enabled, err := strconv.ParseBool(string(clientAuth))
+	if err != nil {
+		return false, fmt.Errorf("failed to parse client auth flag (key: %s): %w", UserSkip, err)
+	}
+	return enabled, nil
+}
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/Credentials.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/Credentials.java
index 480615c1f..6879a4f4f 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/Credentials.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/Credentials.java
@@ -25,6 +25,13 @@ public interface Credentials {
    */
   String caCertificates();
 
+  /**
+   * Skip client auth.
+   *
+   * @return true if client auth should be skipped otherwise false
+   */
+  boolean skipClientAuth();
+
   /**
    * @return user certificate.
    */
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
index d0e3aebc2..26bbbaec5 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
@@ -35,7 +35,10 @@ class CredentialsValidator {
     }
 
     if (is(SecurityProtocol.SSL, securityProtocol)) {
-      if (anyBlank(credentials.userCertificate(), credentials.userKey(), credentials.caCertificates())) {
+      if (credentials.skipClientAuth()) {
+        return null;
+      }
+      if (anyBlank(credentials.userCertificate(), credentials.userKey())) {
         return "Security protocol " + securityProtocol.name + ": invalid user certificate or user key or CA certificates";
       }
       return null;
@@ -53,9 +56,6 @@ class CredentialsValidator {
     }
 
     if (is(SecurityProtocol.SASL_SSL, securityProtocol)) {
-      if (anyBlank(credentials.caCertificates())) {
-        return "Security protocol " + securityProtocol.name + ": invalid truststore";
-      }
       if (isInvalidSASLMechanism(SASLMechanism)) {
         return "Security protocol " + securityProtocol.name + ": invalid SASL mechanism, expected SCRAM-SHA-256 or SCRAM-SHA-512 got " + SASLMechanism;
       }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
index 31a9cbf32..5ce6de2dd 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
@@ -16,9 +16,6 @@
 
 package dev.knative.eventing.kafka.broker.core.security;
 
-import java.util.Map;
-import java.util.Properties;
-import java.util.function.BiConsumer;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.config.SslConfigs;
@@ -26,6 +23,10 @@ import org.apache.kafka.common.security.plain.PlainLoginModule;
 import org.apache.kafka.common.security.scram.ScramLoginModule;
 import org.apache.kafka.common.security.ssl.DefaultSslEngineFactory;
 
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.BiConsumer;
+
 public class KafkaClientsAuth {
 
   public static Properties attachCredentials(final Properties properties, final Credentials credentials) {
@@ -78,13 +79,17 @@ public class KafkaClientsAuth {
   }
 
   private static void ssl(final BiConsumer<String, String> propertiesSetter, final Credentials credentials) {
-    propertiesSetter.accept(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, DefaultSslEngineFactory.PEM_TYPE);
-    propertiesSetter.accept(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, credentials.caCertificates());
-    final var keystore = credentials.userCertificate();
-    if (keystore != null) {
-      propertiesSetter.accept(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, credentials.userCertificate());
-      propertiesSetter.accept(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, credentials.userKey());
+    final var caCert = credentials.caCertificates();
+    if (caCert != null) {
+      propertiesSetter.accept(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, DefaultSslEngineFactory.PEM_TYPE);
+      propertiesSetter.accept(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, caCert);
+    }
+    final var userCert = credentials.userCertificate();
+    final var userKey = credentials.userKey();
+    if (userCert != null && userKey != null) {
       propertiesSetter.accept(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, DefaultSslEngineFactory.PEM_TYPE);
+      propertiesSetter.accept(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, userCert);
+      propertiesSetter.accept(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, userKey);
     }
   }
 }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesCredentials.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesCredentials.java
index 3f065fdd5..0bcd9d7fe 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesCredentials.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesCredentials.java
@@ -17,11 +17,12 @@
 package dev.knative.eventing.kafka.broker.core.security;
 
 import io.fabric8.kubernetes.api.model.Secret;
-import java.util.Base64;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Base64;
+
 import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
 
 class KubernetesCredentials implements Credentials {
@@ -32,6 +33,7 @@ class KubernetesCredentials implements Credentials {
 
   static final String USER_CERTIFICATE_KEY = "user.crt";
   static final String USER_KEY_KEY = "user.key";
+  static final String USER_SKIP_KEY = "user.skip";
 
   static final String USERNAME_KEY = "user";
   static final String PASSWORD_KEY = "password";
@@ -44,6 +46,7 @@ class KubernetesCredentials implements Credentials {
   private String caCertificates;
   private String userCertificate;
   private String userKey;
+  private Boolean skipUser;
   private SecurityProtocol securityProtocol;
   private String SASLMechanism;
   private String SASLUsername;
@@ -68,6 +71,26 @@ class KubernetesCredentials implements Credentials {
     return this.caCertificates;
   }
 
+  @Override
+  public boolean skipClientAuth() {
+    if (secret == null || secret.getData() == null) {
+      return false;
+    }
+    if (skipUser == null) {
+      final var skip = secret.getData().get(USER_SKIP_KEY);
+      if (skip == null) {
+        this.skipUser = false;
+      } else {
+        try {
+          this.skipUser = Boolean.parseBoolean(skip);
+        } catch (final Exception ex) {
+          this.skipUser = false;
+        }
+      }
+    }
+    return this.skipUser;
+  }
+
   @Override
   public String userCertificate() {
     if (secret == null || secret.getData() == null) {
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/PlaintextCredentials.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/PlaintextCredentials.java
index 2961b5a02..056603c49 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/PlaintextCredentials.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/PlaintextCredentials.java
@@ -25,6 +25,11 @@ public class PlaintextCredentials implements Credentials {
     return null;
   }
 
+  @Override
+  public boolean skipClientAuth() {
+    return true;
+  }
+
   @Override
   public String userCertificate() {
     return null;


Test patch:
diff --git a/control-plane/pkg/security/secret_test.go b/control-plane/pkg/security/secret_test.go
index f81867aee..537d74b8a 100644
--- a/control-plane/pkg/security/secret_test.go
+++ b/control-plane/pkg/security/secret_test.go
@@ -175,6 +175,85 @@ func TestSSL(t *testing.T) {
 	assert.Greater(t, len(config.Net.TLS.Config.RootCAs.Subjects()), 0)
 }
 
+func TestSSLNoUserKey(t *testing.T) {
+	ca, _, userCert := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol": []byte("SSL"),
+		"user.crt": userCert,
+		"ca.crt":   ca,
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.NotNil(t, err)
+}
+
+func TestSSLNoUserCert(t *testing.T) {
+	ca, userKey, _ := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol": []byte("SSL"),
+		"user.key": userKey,
+		"ca.crt":   ca,
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.NotNil(t, err)
+}
+
+func TestSSLInvalidKeyPair(t *testing.T) {
+	ca, _, _ := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol": []byte("SSL"),
+		"user.key": []byte("foo"),
+		"user.crt": []byte("bar"),
+		"ca.crt":   ca,
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.NotNil(t, err)
+}
+
+func TestSSLNoClientAuth(t *testing.T) {
+	ca, _, _ := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol":  []byte("SSL"),
+		"user.skip": []byte("true"),
+		"ca.crt":    ca,
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.Nil(t, err)
+	assert.True(t, config.Net.TLS.Enable)
+	assert.NotNil(t, config.Net.TLS.Config.RootCAs)
+	assert.Greater(t, len(config.Net.TLS.Config.RootCAs.Subjects()), 0)
+}
+
+func TestSSLNoClientAuthInvalidFlag(t *testing.T) {
+	ca, _, _ := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol":  []byte("SSL"),
+		"user.skip": []byte("foo"),
+		"ca.crt":    ca,
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.NotNil(t, err)
+}
+
 func TestSASLPLainSSL(t *testing.T) {
 	ca, userKey, userCert := loadCerts(t)
 
@@ -259,6 +338,25 @@ func TestSASLSCRAM512SSL(t *testing.T) {
 	assert.Equal(t, "my-user-password", config.Net.SASL.Password)
 }
 
+func TestSASLSCRAM512SSLInvalidCaCert(t *testing.T) {
+	_, userKey, userCert := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol":       []byte("SASL_SSL"),
+		"sasl.mechanism": []byte("SCRAM-SHA-512"),
+		"ca.crt":         []byte(`foo`),
+		"user.crt":       userCert,
+		"user.key":       userKey,
+		"user":           []byte("my-user-name"),
+		"password":       []byte("my-user-password"),
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.NotNil(t, err)
+}
+
 func loadCerts(t *testing.T) (ca, userKey, userCert []byte) {
 	ca, err := ioutil.ReadFile("testdata/ca.crt")
 	assert.Nil(t, err)
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
index 248702ba7..1d7b441f4 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
@@ -71,17 +71,31 @@ public class CredentialsValidatorTest {
   }
 
   @Test
-  public void securityProtocolSslInvalidNoCACert() {
+  public void securityProtocolSslNoCACert() {
     final var credential = mock(Credentials.class);
 
     when(credential.securityProtocol()).thenReturn(SecurityProtocol.SSL);
     when(credential.userCertificate()).thenReturn("xyz");
     when(credential.userKey()).thenReturn("my-key");
-    when(credential.caCertificates()).thenReturn("   ");
+    when(credential.caCertificates()).thenReturn(null);
 
-    assertThat(CredentialsValidator.validate(credential)).isNotEmpty();
+    assertThat(CredentialsValidator.validate(credential)).isNull();
   }
 
+  @Test
+  public void securityProtocolSslNoCACertNoClientAuth() {
+    final var credential = mock(Credentials.class);
+
+    when(credential.securityProtocol()).thenReturn(SecurityProtocol.SSL);
+    when(credential.skipClientAuth()).thenReturn(true);
+    when(credential.userCertificate()).thenReturn("  ");
+    when(credential.userKey()).thenReturn("  ");
+    when(credential.caCertificates()).thenReturn(null);
+
+    assertThat(CredentialsValidator.validate(credential)).isNull();
+  }
+
+
   @Test
   public void securityProtocolSaslPlaintextScramSha256Valid() {
     securityProtocolSaslPlaintextScramValid("SCRAM-SHA-256");
@@ -195,17 +209,16 @@ public class CredentialsValidatorTest {
   }
 
   @Test
-  public void securityProtocolSaslSslScramSha51NoTruststoreInValid() {
+  public void securityProtocolSaslSslScramSha51EmptyCaCert() {
     final var credential = mock(Credentials.class);
 
     when(credential.securityProtocol()).thenReturn(SecurityProtocol.SASL_SSL);
-    when(credential.userCertificate()).thenReturn("abc");
     when(credential.caCertificates()).thenReturn("   ");
     when(credential.SASLMechanism()).thenReturn("SCRAM-SHA-512");
     when(credential.SASLUsername()).thenReturn("bbb");
     when(credential.SASLPassword()).thenReturn("ccc");
 
-    assertThat(CredentialsValidator.validate(credential)).isNotEmpty();
+    assertThat(CredentialsValidator.validate(credential)).isNull();
   }
 
   @Test
@@ -250,28 +263,29 @@ public class CredentialsValidatorTest {
   }
 
   @Test
-  public void securityProtocolSaslSslScramSha256NoTruststoreInValid() {
+  public void securityProtocolSaslSslScramSha256NoCaCert() {
     final var credential = mock(Credentials.class);
 
     when(credential.securityProtocol()).thenReturn(SecurityProtocol.SASL_SSL);
-    when(credential.caCertificates()).thenReturn("   ");
+    when(credential.caCertificates()).thenReturn(null);
     when(credential.SASLMechanism()).thenReturn("SCRAM-SHA-512");
     when(credential.SASLUsername()).thenReturn("bbb");
     when(credential.SASLPassword()).thenReturn("ccc");
 
-    assertThat(CredentialsValidator.validate(credential)).isNotEmpty();
+    assertThat(CredentialsValidator.validate(credential)).isNull();
   }
 
   @Test
-  public void securityProtocolSaslSslScramSha256NoCACertInValid() {
+  public void securityProtocolSaslSslScramSha256EmptyCACert() {
     final var credential = mock(Credentials.class);
 
     when(credential.securityProtocol()).thenReturn(SecurityProtocol.SASL_SSL);
+    when(credential.caCertificates()).thenReturn("  ");
     when(credential.SASLMechanism()).thenReturn("SCRAM-SHA-512");
     when(credential.SASLUsername()).thenReturn("bbb");
     when(credential.SASLPassword()).thenReturn("ccc");
 
-    assertThat(CredentialsValidator.validate(credential)).isNotEmpty();
+    assertThat(CredentialsValidator.validate(credential)).isNull();
   }
 
   @Test