Changes the necessary data fields for the clientSSLCertSecret and serverSSLCertSecret

api/v1beta1/kafkacluster_types.go

@@ -98,9 +98,8 @@ type KafkaClusterSpec struct {
 	// It will be used by the koperator, cruise control, cruise control metrics reporter
 	// to communicate on SSL with that internal listener which is used for interbroker communication.
 	// The client certificate must share the same chain of trust as the server certificate used by the corresponding internal listener.
-	// The secret must contain the keystore, truststore jks files and the password for them in base64 encoded format and
-	// the tls certificate, tls private key, CA certificate in PEM format with base64 encoded
-	// under the keystore.jks, truststore.jks, password, tls.crt, tls.key, and ca.crt data fields.
+	// The secret must contain the keystore, truststore jks files and the password for them in base64 encoded format
+	// under the keystore.jks, truststore.jks, password data fields.
 	ClientSSLCertSecret *corev1.LocalObjectReference `json:"clientSSLCertSecret,omitempty"`
 }
 
@@ -534,8 +533,6 @@ type CommonListenerSpec struct {
 	Type SecurityProtocol `json:"type"`
 	// ServerSSLCertSecret is a reference to the Kubernetes secret that contains the server certificate for the listener to be used for SSL communication.
 	// The secret must contain the keystore, truststore jks files and the password for them in base64 encoded format under the keystore.jks, truststore.jks, password data fields.
-	// When the listener is used for inner broker or controller communication the tls certificate is
-	// also needed in PEM format with base64 encoding under the tls.crt data field.
 	// If this field is omitted koperator will auto-create a self-signed server certificate using the configuration provided in 'sslSecrets' field.
 	ServerSSLCertSecret *corev1.LocalObjectReference `json:"serverSSLCertSecret,omitempty"`
 	// SSLClientAuth specifies whether client authentication is required, requested, or not required.

charts/kafka-operator/templates/crds.yaml

@@ -12273,9 +12273,8 @@ spec:
                   share the same chain of trust as the server certificate used by
                   the corresponding internal listener. The secret must contain the
                   keystore, truststore jks files and the password for them in base64
-                  encoded format and the tls certificate, tls private key, CA certificate
-                  in PEM format with base64 encoded under the keystore.jks, truststore.jks,
-                  password, tls.crt, tls.key, and ca.crt data fields.
+                  encoded format under the keystore.jks, truststore.jks, password
+                  data fields.
                 properties:
                   name:
                     description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
@@ -19332,10 +19331,7 @@ spec:
                             to be used for SSL communication. The secret must contain
                             the keystore, truststore jks files and the password for
                             them in base64 encoded format under the keystore.jks,
-                            truststore.jks, password data fields. When the listener
-                            is used for inner broker or controller communication the
-                            tls certificate is also needed in PEM format with base64
-                            encoding under the tls.crt data field. If this field is
+                            truststore.jks, password data fields. If this field is
                             omitted koperator will auto-create a self-signed server
                             certificate using the configuration provided in 'sslSecrets'
                             field.
@@ -19401,10 +19397,7 @@ spec:
                             to be used for SSL communication. The secret must contain
                             the keystore, truststore jks files and the password for
                             them in base64 encoded format under the keystore.jks,
-                            truststore.jks, password data fields. When the listener
-                            is used for inner broker or controller communication the
-                            tls certificate is also needed in PEM format with base64
-                            encoding under the tls.crt data field. If this field is
+                            truststore.jks, password data fields. If this field is
                             omitted koperator will auto-create a self-signed server
                             certificate using the configuration provided in 'sslSecrets'
                             field.

config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml

@@ -12272,9 +12272,8 @@ spec:
                   share the same chain of trust as the server certificate used by
                   the corresponding internal listener. The secret must contain the
                   keystore, truststore jks files and the password for them in base64
-                  encoded format and the tls certificate, tls private key, CA certificate
-                  in PEM format with base64 encoded under the keystore.jks, truststore.jks,
-                  password, tls.crt, tls.key, and ca.crt data fields.
+                  encoded format under the keystore.jks, truststore.jks, password
+                  data fields.
                 properties:
                   name:
                     description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
@@ -19331,10 +19330,7 @@ spec:
                             to be used for SSL communication. The secret must contain
                             the keystore, truststore jks files and the password for
                             them in base64 encoded format under the keystore.jks,
-                            truststore.jks, password data fields. When the listener
-                            is used for inner broker or controller communication the
-                            tls certificate is also needed in PEM format with base64
-                            encoding under the tls.crt data field. If this field is
+                            truststore.jks, password data fields. If this field is
                             omitted koperator will auto-create a self-signed server
                             certificate using the configuration provided in 'sslSecrets'
                             field.
@@ -19400,10 +19396,7 @@ spec:
                             to be used for SSL communication. The secret must contain
                             the keystore, truststore jks files and the password for
                             them in base64 encoded format under the keystore.jks,
-                            truststore.jks, password data fields. When the listener
-                            is used for inner broker or controller communication the
-                            tls certificate is also needed in PEM format with base64
-                            encoding under the tls.crt data field. If this field is
+                            truststore.jks, password data fields. If this field is
                             omitted koperator will auto-create a self-signed server
                             certificate using the configuration provided in 'sslSecrets'
                             field.

pkg/pki/certmanagerpki/certmanager_tls_config_test.go

@@ -31,11 +31,13 @@ func newMockControllerSecret(valid bool) *corev1.Secret {
 	secret.Name = "test-controller"
 	secret.Namespace = testNamespace
 	cert, key, _, _ := certutil.GenerateTestCert()
+	keystore_jks, password, _ := certutil.GenerateJKSFromByte(cert, key, cert)
+
 	if valid {
 		secret.Data = map[string][]byte{
-			corev1.TLSCertKey:       cert,
-			corev1.TLSPrivateKeyKey: key,
-			v1alpha1.CoreCACertKey:  cert,
+			v1alpha1.PasswordKey:      password,
+			v1alpha1.TLSJKSTrustStore: keystore_jks,
+			v1alpha1.TLSJKSKeyStore:   keystore_jks,
 		}
 	}
 	return secret

pkg/resources/cruisecontrol/configmap.go

@@ -42,7 +42,7 @@ import (
 
 const MinLogDirSizeInMB = int64(1)
 
-func (r *Reconciler) configMap(clientPass, capacityConfig string, log logr.Logger) runtime.Object {
+func (r *Reconciler) configMap(clientPass string, capacityConfig string, log logr.Logger) runtime.Object {
 	ccConfig := properties.NewProperties()
 
 	// Add base Cruise Control configuration
@@ -90,27 +90,29 @@ func (r *Reconciler) configMap(clientPass, capacityConfig string, log logr.Logge
 	return configMap
 }
 
-func generateSSLConfig(k v1beta1.KafkaClusterSpec, clientPass string, log logr.Logger) *properties.Properties {
-	sslConf := properties.NewProperties()
-
-	if k.IsClientSSLSecretPresent() && util.IsSSLEnabledForInternalCommunication(k.ListenersConfig.InternalListeners) {
-		if err := sslConf.Set("security.protocol", "SSL"); err != nil {
-			log.Error(err, "settings security.protocol in Cruise Control configuration failed")
-		}
-		if err := sslConf.Set("ssl.truststore.location", keystoreVolumePath+"/"+v1alpha1.TLSJKSTrustStore); err != nil {
-			log.Error(err, "settings ssl.truststore.location in Cruise Control configuration failed")
-		}
-		if err := sslConf.Set("ssl.keystore.location", keystoreVolumePath+"/"+v1alpha1.TLSJKSKeyStore); err != nil {
-			log.Error(err, "settings ssl.keystore.location in Cruise Control configuration failed")
+func generateSSLConfig(kafkaCluster v1beta1.KafkaClusterSpec, clientPass string, log logr.Logger) *properties.Properties {
+	config := properties.NewProperties()
+	if kafkaCluster.IsClientSSLSecretPresent() && util.IsSSLEnabledForInternalCommunication(kafkaCluster.ListenersConfig.InternalListeners) {
+		keyStoreLoc := keystoreVolumePath + "/" + v1alpha1.TLSJKSKeyStore
+		trustStoreLoc := keystoreVolumePath + "/" + v1alpha1.TLSJKSTrustStore
+
+		sslConfig := map[string]string{
+			"security.protocol":       "SSL",
+			"ssl.truststore.type":     "JKS",
+			"ssl.keystore.type":       "JKS",
+			"ssl.truststore.location": trustStoreLoc,
+			"ssl.keystore.location":   keyStoreLoc,
+			"ssl.keystore.password":   clientPass,
+			"ssl.truststore.password": clientPass,
 		}
-		if err := sslConf.Set("ssl.keystore.password", clientPass); err != nil {
-			log.Error(err, "settings ssl.keystore.password in Cruise Control configuration failed")
-		}
-		if err := sslConf.Set("ssl.truststore.password", clientPass); err != nil {
-			log.Error(err, "settings ssl.truststore.password in Cruise Control configuration failed")
+
+		for k, v := range sslConfig {
+			if err := config.Set(k, v); err != nil {
+				log.Error(err, fmt.Sprintf("setting %s parameter in cruise control configuration resulted an error", k))
+			}
 		}
 	}
-	return sslConf
+	return config
 }
 
 const (

pkg/resources/cruisecontrol/cruisecontrol.go

@@ -30,6 +30,7 @@ import (
 	"github.com/banzaicloud/koperator/pkg/errorfactory"
 	"github.com/banzaicloud/koperator/pkg/k8sutil"
 	"github.com/banzaicloud/koperator/pkg/resources"
+	certutil "github.com/banzaicloud/koperator/pkg/util/cert"
 	pkicommon "github.com/banzaicloud/koperator/pkg/util/pki"
 )
 
@@ -78,9 +79,14 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 
 	log.V(1).Info("Reconciling")
 
-	clientPass, err := r.getClientPassword()
-	if err != nil {
-		return err
+	var clientPass string
+	var err error
+
+	// Get configuration data from client secret
+	if r.KafkaCluster.Spec.IsClientSSLSecretPresent() {
+		if clientPass, err = r.getClientPassword(); err != nil {
+			return err
+		}
 	}
 
 	if r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint == "" {
@@ -148,25 +154,39 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 }
 
 func (r *Reconciler) getClientPassword() (string, error) {
-	var clientPass string
-	if r.KafkaCluster.Spec.IsClientSSLSecretPresent() {
-		// Use that secret as default which has autogenerated for clients by us
-		clientNamespacedName := types.NamespacedName{Name: fmt.Sprintf(pkicommon.BrokerControllerTemplate, r.KafkaCluster.Name), Namespace: r.KafkaCluster.Namespace}
-		if r.KafkaCluster.Spec.GetClientSSLCertSecretName() != "" {
-			clientNamespacedName = types.NamespacedName{Name: r.KafkaCluster.Spec.GetClientSSLCertSecretName(), Namespace: r.KafkaCluster.Namespace}
+	clientSecret, err := r.getClientSecret()
+	if err != nil {
+		return "", err
+	}
+	return string(clientSecret.Data[v1alpha1.PasswordKey]), nil
+}
+
+func (r *Reconciler) getClientSecret() (*corev1.Secret, error) {
+	clientSecret := &corev1.Secret{}
+	// Use that secret as default which has autogenerated for clients by us
+	clientNamespacedName := types.NamespacedName{Name: fmt.Sprintf(pkicommon.BrokerControllerTemplate, r.KafkaCluster.Name), Namespace: r.KafkaCluster.Namespace}
+	if r.KafkaCluster.Spec.GetClientSSLCertSecretName() != "" {
+		clientNamespacedName = types.NamespacedName{Name: r.KafkaCluster.Spec.GetClientSSLCertSecretName(), Namespace: r.KafkaCluster.Namespace}
+	}
+
+	if err := r.Client.Get(context.Background(), clientNamespacedName, clientSecret); err != nil {
+		// We only return with ResourceNotReady (which is going to retry after a period time)
+		// when we use our cert generation for client cert
+		if apierrors.IsNotFound(err) && r.KafkaCluster.Spec.GetClientSSLCertSecretName() == "" {
+			return nil, errorfactory.New(errorfactory.ResourceNotReady{}, err, "client secret not ready")
 		}
-		clientSecret := &corev1.Secret{}
-		if err := r.Client.Get(context.TODO(), clientNamespacedName, clientSecret); err != nil {
-			// We only return with ResourceNotReady (which is goin to retry after period time)
-			// when we use our cert generation for client cert
-			if apierrors.IsNotFound(err) && r.KafkaCluster.Spec.GetClientSSLCertSecretName() == "" {
-				return "", errorfactory.New(errorfactory.ResourceNotReady{}, err, "client secret not ready")
-			}
-			return "", errors.WrapIfWithDetails(err, "failed to get client secret")
+
+		return nil, errors.WrapIfWithDetails(err, "failed to get client secret")
+	}
+
+	if err := certutil.CheckSSLCertSecret(clientSecret); err != nil {
+		if r.KafkaCluster.Spec.GetClientSSLCertSecretName() == "" {
+			return nil, errorfactory.New(errorfactory.ResourceNotReady{}, errors.Errorf("SSL JKS certificate has not generated properly yet into client secret: %s", clientSecret.Name), "checking secret data fields")
 		}
-		clientPass = string(clientSecret.Data[v1alpha1.PasswordKey])
+		return nil, errors.WrapIfWithDetails(err, "failed to get certificates from client secret")
 	}
-	return clientPass, nil
+
+	return clientSecret, nil
 }
 
 func isBrokerDeletionInProgress(brokerState map[string]v1beta1.BrokerState) bool {