From ecce211264427817d968a20e5e1d095a2a365ac2 Mon Sep 17 00:00:00 2001
From: Marton Barta <51166675+bartam1@users.noreply.github.com>
Date: Tue, 2 Aug 2022 13:50:00 +0200
Subject: [PATCH] Koperator can handle intermediate and leaf certificates in
 generated kafkaUser's TLS certificate (#843)

---
 controllers/kafkauser_controller.go |  8 +++++++-
 pkg/util/cert/certutil.go           |  4 +---
 pkg/util/pki/common.go              | 11 +++++++----
 pkg/util/pki/pki_common_test.go     |  5 ++++-
 4 files changed, 19 insertions(+), 9 deletions(-)

diff --git a/controllers/kafkauser_controller.go b/controllers/kafkauser_controller.go
index 6dbb046d4..2a7df7856 100644
--- a/controllers/kafkauser_controller.go
+++ b/controllers/kafkauser_controller.go
@@ -238,7 +238,13 @@ func (r *KafkaUserReconciler) Reconcile(ctx context.Context, request reconcile.R
 				return requeueWithError(reqLogger, "failed to reconcile user secret", err)
 			}
 		}
-		kafkaUser = user.DN()
+		kafkaUser, err = user.GetDistinguishedName()
+		if err != nil {
+			reqLogger.Error(err, "could not get Distinguished Name from the generated TLS certificate", "cert", string(user.Certificate))
+			return ctrl.Result{
+				Requeue: false,
+			}, err
+		}
 		// check if marked for deletion and remove created certs
 		if k8sutil.IsMarkedForDeletion(instance.ObjectMeta) {
 			reqLogger.Info("Kafka user is marked for deletion, revoking certificates")
diff --git a/pkg/util/cert/certutil.go b/pkg/util/cert/certutil.go
index 4c013e6ae..85ac60aa8 100644
--- a/pkg/util/cert/certutil.go
+++ b/pkg/util/cert/certutil.go
@@ -132,9 +132,7 @@ func DecodeCertificate(raw []byte) (cert *x509.Certificate, err error) {
 	if err != nil {
 		return nil, err
 	}
-	if len(certs) != 1 {
-		return nil, errors.New("only one certificate should be present, more found")
-	}
+
 	return certs[0].Certificate, nil
 }
 
diff --git a/pkg/util/pki/common.go b/pkg/util/pki/common.go
index 9f5aa2e69..9668ba0a9 100644
--- a/pkg/util/pki/common.go
+++ b/pkg/util/pki/common.go
@@ -93,11 +93,14 @@ type UserCertificate struct {
 	Password    []byte
 }
 
-// DN returns the Distinguished Name of a TLS certificate
-func (u *UserCertificate) DN() string {
+//  GetDistinguishedName returns the Distinguished Name of a TLS certificate
+func (u *UserCertificate) GetDistinguishedName() (string, error) {
 	// cert has already been validated so we can assume no error
-	cert, _ := certutil.DecodeCertificate(u.Certificate)
-	return cert.Subject.String()
+	cert, err := certutil.DecodeCertificate(u.Certificate)
+	if err != nil {
+		return "", err
+	}
+	return cert.Subject.String(), nil
 }
 
 // GetInternalDNSNames returns all potential DNS names for a kafka cluster - including brokers
diff --git a/pkg/util/pki/pki_common_test.go b/pkg/util/pki/pki_common_test.go
index a95bb43c1..586e91fd7 100644
--- a/pkg/util/pki/pki_common_test.go
+++ b/pkg/util/pki/pki_common_test.go
@@ -42,7 +42,10 @@ func TestDN(t *testing.T) {
 	userCert := &UserCertificate{
 		Certificate: cert,
 	}
-	dn := userCert.DN()
+	dn, err := userCert.GetDistinguishedName()
+	if err != nil {
+		t.Errorf("error should be nil, got: %s", err)
+	}
 	if dn != expected {
 		t.Error("Expected:", expected, "got:", dn)
 	}