feat(k8s-csr): explicit put ca cert into kafkausers' secret when cert…

…-manager signer (#1028)

pkg/pki/certmanagerpki/certmanager.go

@@ -15,8 +15,6 @@
 package certmanagerpki
 
 import (
-	"flag"
-
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/banzaicloud/koperator/api/v1beta1"
@@ -25,12 +23,6 @@ import (
 
 const spiffeIdTemplate = "spiffe://%s/ns/%s/kafkauser/%s"
 
-var namespaceCertManager string
-
-func init() {
-	flag.StringVar(&namespaceCertManager, "cert-manager-namespace", "cert-manager", "The namespace where cert-manager is running")
-}
-
 type CertManager interface {
 	pki.Manager
 }

pkg/pki/certmanagerpki/certmanager_pki.go

@@ -53,7 +53,7 @@ func (c *certManager) FinalizePKI(ctx context.Context) error {
 		if c.cluster.Spec.ListenersConfig.SSLSecrets.IssuerRef == nil {
 			objNames = append(
 				objNames,
-				types.NamespacedName{Name: fmt.Sprintf(pkicommon.BrokerCACertTemplate, c.cluster.Name), Namespace: namespaceCertManager})
+				types.NamespacedName{Name: fmt.Sprintf(pkicommon.BrokerCACertTemplate, c.cluster.Name), Namespace: pkicommon.NamespaceCertManager})
 		}
 		for _, obj := range objNames {
 			// Delete the certificates first so we don't accidentally recreate the
@@ -183,7 +183,7 @@ func caSecretForProvidedCert(ctx context.Context, client client.Client, cluster
 	caSecret := &corev1.Secret{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      fmt.Sprintf(pkicommon.BrokerCACertTemplate, cluster.Name),
-			Namespace: namespaceCertManager,
+			Namespace: pkicommon.NamespaceCertManager,
 			Labels:    pkicommon.LabelsForKafkaPKI(cluster.Name, cluster.Namespace),
 		},
 		Data: map[string][]byte{
@@ -214,7 +214,7 @@ func caCertForCluster(cluster *v1beta1.KafkaCluster) *certv1.Certificate {
 	return &certv1.Certificate{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      fmt.Sprintf(pkicommon.BrokerCACertTemplate, cluster.Name),
-			Namespace: namespaceCertManager,
+			Namespace: pkicommon.NamespaceCertManager,
 			Labels:    pkicommon.LabelsForKafkaPKI(cluster.Name, cluster.Namespace),
 		},
 		Spec: certv1.CertificateSpec{

pkg/pki/k8scsrpki/k8scsr_user.go

@@ -33,6 +33,7 @@ import (
 	certutil "github.com/banzaicloud/koperator/pkg/util/cert"
 	pkicommon "github.com/banzaicloud/koperator/pkg/util/pki"
 
+	certv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
 	certsigningreqv1 "k8s.io/api/certificates/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -45,9 +46,12 @@ import (
 )
 
 const (
-	notApprovedErrMsg         = "instance is not approved"
-	notFoundApprovedCsrErrMsg = "could not find approved csr and the operator is not capable of approving the csr"
-	approveReason             = "ApprovedByPolicy"
+	notApprovedErrMsg                             = "instance is not approved"
+	notFoundApprovedCsrErrMsg                     = "could not find approved csr and the operator is not capable of approving the csr"
+	notFoundCAInClusterIssuerErrMsg               = "could not extract CA from ClusterIssuer"
+	notFoundCertManagerSecretField                = "could not find certificate field in cert-manager Secret"
+	approveReason                                 = "ApprovedByPolicy"
+	defaultCertManagerIssuerSecretCertificateFile = "tls.crt"
 )
 
 // ReconcileUserCertificate ensures and returns a user certificate - should be idempotent
@@ -186,13 +190,11 @@ func (c *k8sCSR) ReconcileUserCertificate(
 	//Leaf cert
 	secret.Data[corev1.TLSCertKey] = certs[0].ToPEM()
 	//CA chain certs
-	var caChain []byte
-	for _, cr := range certs {
-		if cr.Certificate.IsCA {
-			caChain = append(caChain, cr.ToPEM()...)
-			caChain = append(caChain, byte('\n'))
-		}
+	caChain, err := c.getCAChain(ctx, signingReq, certs)
+	if err != nil {
+		return nil, err
 	}
+
 	secret.Data[v1alpha1.CaChainPem] = caChain
 	certBundleX509 := certutil.GetCertBundle(certs)
 
@@ -344,3 +346,56 @@ func (c *k8sCSR) Approve(ctx context.Context, signingReq *certsigningreqv1.Certi
 
 	return nil
 }
+
+func (c *k8sCSR) getCAChain(ctx context.Context, signingReq *certsigningreqv1.CertificateSigningRequest, certs []*certutil.CertificateContainer) ([]byte, error) {
+	var caChain []byte
+	signerName := strings.Split(signingReq.Spec.SignerName, "/")
+	if len(signerName) < 2 { // Note: [signerNamePrefix, clusterIssuerName]
+		return nil, errors.NewWithDetails("invalid signer name", "signerName", signingReq.Spec.SignerName)
+	}
+
+	if signerName[0] == v1alpha1.CertManagerSignerNamePrefix {
+		clusterIssuer := &certv1.ClusterIssuer{}
+		clusterIssuerName := signerName[1]
+		err := c.client.Get(ctx, types.NamespacedName{
+			Name: clusterIssuerName,
+		}, clusterIssuer)
+		if err != nil {
+			return nil, errors.WrapIfWithDetails(err,
+				"failed to get ClusterIssuer from K8s", "clusterIssuer", clusterIssuerName)
+		}
+
+		if clusterIssuer.GetSpec().CA == nil {
+			return nil, errorfactory.New(errorfactory.FatalReconcileError{}, errors.New(notFoundCAInClusterIssuerErrMsg),
+				"clusterIssuer doesn't contain CA secret reference", "clusterIssuer", clusterIssuerName)
+		}
+
+		certManagerSecret := &corev1.Secret{}
+		err = c.client.Get(ctx, types.NamespacedName{
+			Name:      clusterIssuer.GetSpec().CA.SecretName,
+			Namespace: pkicommon.NamespaceCertManager,
+		}, certManagerSecret)
+		if err != nil {
+			return nil, errors.WrapIfWithDetails(err,
+				"failed to get secret from K8s", "secretName", clusterIssuer.GetSpec().CA.SecretName,
+				"namespace", certManagerSecret.GetNamespace())
+		}
+
+		chain, ok := certManagerSecret.Data[defaultCertManagerIssuerSecretCertificateFile]
+		if !ok {
+			return caChain, errorfactory.New(errorfactory.FatalReconcileError{}, errors.New(notFoundCertManagerSecretField),
+				"failed to get field", "secretName", clusterIssuer.GetSpec().CA.SecretName,
+				"namespace", certManagerSecret.GetNamespace(), "field", defaultCertManagerIssuerSecretCertificateFile)
+		}
+		caChain = chain
+	} else {
+		for _, cr := range certs {
+			if cr.Certificate.IsCA {
+				caChain = append(caChain, cr.ToPEM()...)
+				caChain = append(caChain, byte('\n'))
+			}
+		}
+	}
+
+	return caChain, nil
+}

pkg/util/pki/common.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"crypto/tls"
+	"flag"
 	"fmt"
 	"sort"
 	"strings"
@@ -62,6 +63,13 @@ const (
 	MaxCNLen = 64
 )
 
+// NamespaceCertManager points to a namespace where cert-manager is located
+var NamespaceCertManager string
+
+func init() {
+	flag.StringVar(&NamespaceCertManager, "cert-manager-namespace", "cert-manager", "The namespace where cert-manager is running")
+}
+
 // Manager is the main interface for objects performing PKI operations
 type Manager interface {
 	// ReconcilePKI ensures a PKI for a kafka cluster - should be idempotent.