charts/redpanda: fix mTLS

Prior to this commit the chart had a variety of bugs around mTLS. The majority
of them were incorrect path construction and handling of `.clientSecretRef`.
The primary issue, though, is that the chart incorrectly mints a single client
certificate regardless of how many trust chains are in use.

This commit moves all name and path references into helper methods onto the
`TLSCert` itself and generates client certs per unique trust chain with client
auth enabled.

K8S-719

(cherry picked from commit 6c63e57)

Author: chrisseto

.changes/unreleased/charts-redpanda-Changed-20250918-152741.yaml

@@ -0,0 +1,4 @@
+project: charts/redpanda
+kind: Changed
+body: Client certificates are now named `$FULLNAME-$CERT-client-cert`.
+time: 2025-09-18T15:27:41.700988-04:00

.changes/unreleased/charts-redpanda-Fixed-20250918-152623.yaml

@@ -0,0 +1,4 @@
+project: charts/redpanda
+kind: Fixed
+body: mTLS client certificates are now generated per certificate, as required, instead of using a single and potentially invalid certificate.
+time: 2025-09-18T15:26:23.232523-04:00

.changes/unreleased/operator-Changed-20250918-152741.yaml

@@ -0,0 +1,4 @@
+project: operator
+kind: Changed
+body: Client certificates are now named `$FULLNAME-$CERT-client-cert`.
+time: 2025-09-18T15:27:41.700988-04:00

.changes/unreleased/operator-Fixed-20250918-152623.yaml

@@ -0,0 +1,4 @@
+project: operator
+kind: Fixed
+body: mTLS client certificates are now generated per certificate, as required, instead of using a single and potentially invalid certificate.
+time: 2025-09-18T15:26:23.232523-04:00

charts/redpanda/cert_issuers.go

@@ -37,11 +37,17 @@ func certIssuersAndCAs(dot *helmette.Dot) ([]*certmanagerv1.Issuer, []*certmanag
 	var issuers []*certmanagerv1.Issuer
 	var certs []*certmanagerv1.Certificate
 
-	if !TLSEnabled(dot) {
-		return issuers, certs
+	inUseCerts := map[string]bool{}
+	for _, name := range values.Listeners.InUseServerCerts(&values.TLS) {
+		inUseCerts[name] = true
 	}
+	for _, name := range values.Listeners.InUseClientCerts(&values.TLS) {
+		inUseCerts[name] = true
+	}
+
+	for name := range helmette.SortedMap(inUseCerts) {
+		data := values.TLS.Certs.MustGet(name)
 
-	for name, data := range helmette.SortedMap(values.TLS.Certs) {
 		// If this certificate is disabled (.Enabled), provided directly by the
 		// end user (.SecretRef), or has an issuer provided (.IssuerRef), we
 		// don't need to bootstrap an issuer.
@@ -130,7 +136,7 @@ func certIssuersAndCAs(dot *helmette.Dot) ([]*certmanagerv1.Issuer, []*certmanag
 				Spec: certmanagerv1.IssuerSpec{
 					IssuerConfig: certmanagerv1.IssuerConfig{
 						CA: &certmanagerv1.CAIssuer{
-							SecretName: fmt.Sprintf(`%s-%s-root-certificate`, Fullname(dot), name),
+							SecretName: data.RootSecretName(dot, name),
 						},
 					},
 				},

charts/redpanda/certs.go

@@ -23,10 +23,6 @@ import (
 )
 
 func ClientCerts(dot *helmette.Dot) []*certmanagerv1.Certificate {
-	if !TLSEnabled(dot) {
-		return []*certmanagerv1.Certificate{}
-	}
-
 	values := helmette.Unwrap[Values](dot.Values)
 
 	fullname := Fullname(dot)
@@ -37,8 +33,11 @@ func ClientCerts(dot *helmette.Dot) []*certmanagerv1.Certificate {
 	domain := strings.TrimSuffix(values.ClusterDomain, ".")
 
 	var certs []*certmanagerv1.Certificate
-	for name, data := range helmette.SortedMap(values.TLS.Certs) {
-		if !helmette.Empty(data.SecretRef) || !ptr.Deref(data.Enabled, true) {
+	for _, name := range values.Listeners.InUseServerCerts(&values.TLS) {
+		data := values.TLS.Certs.MustGet(name)
+
+		// Don't generate server Certificates if a secret is provided.
+		if !helmette.Empty(data.SecretRef) {
 			continue
 		}
 
@@ -85,7 +84,7 @@ func ClientCerts(dot *helmette.Dot) []*certmanagerv1.Certificate {
 				Duration:   helmette.MustDuration(duration),
 				IsCA:       false,
 				IssuerRef:  issuerRef,
-				SecretName: fmt.Sprintf("%s-%s-cert", fullname, name),
+				SecretName: data.ServerSecretName(dot, name),
 				PrivateKey: &certmanagerv1.CertificatePrivateKey{
 					Algorithm: "ECDSA",
 					Size:      256,
@@ -94,49 +93,54 @@ func ClientCerts(dot *helmette.Dot) []*certmanagerv1.Certificate {
 		})
 	}
 
-	name := values.Listeners.Kafka.TLS.Cert
+	for _, name := range values.Listeners.InUseClientCerts(&values.TLS) {
+		data := values.TLS.Certs.MustGet(name)
 
-	data, ok := values.TLS.Certs[name]
-	if !ok {
-		panic(fmt.Sprintf("Certificate %q referenced but not defined", name))
-	}
+		if data.SecretRef != nil && data.ClientSecretRef == nil {
+			panic(fmt.Sprintf(".clientSecretRef MUST be set if .secretRef is set and require_client_auth is true: Cert %q", name))
+		}
 
-	if !helmette.Empty(data.SecretRef) || !ClientAuthRequired(dot) {
-		return certs
-	}
+		// Don't generate a client Certificate if a client secret is provided.
+		if data.ClientSecretRef != nil {
+			continue
+		}
 
-	issuerRef := cmmetav1.ObjectReference{
-		Group: "cert-manager.io",
-		Kind:  "Issuer",
-		Name:  fmt.Sprintf("%s-%s-root-issuer", fullname, name),
-	}
+		issuerRef := cmmetav1.ObjectReference{
+			Group: "cert-manager.io",
+			Kind:  "Issuer",
+			Name:  fmt.Sprintf("%s-%s-root-issuer", fullname, name),
+		}
 
-	if data.IssuerRef != nil {
-		issuerRef = *data.IssuerRef
-		issuerRef.Group = "cert-manager.io"
-	}
+		if data.IssuerRef != nil {
+			issuerRef = *data.IssuerRef
+			issuerRef.Group = "cert-manager.io"
+		}
+
+		duration := helmette.Default("43800h", data.Duration)
 
-	duration := helmette.Default("43800h", data.Duration)
-
-	return append(certs, &certmanagerv1.Certificate{
-		TypeMeta: metav1.TypeMeta{
-			APIVersion: "cert-manager.io/v1",
-			Kind:       "Certificate",
-		},
-		ObjectMeta: metav1.ObjectMeta{
-			Name:   fmt.Sprintf("%s-client", fullname),
-			Labels: FullLabels(dot),
-		},
-		Spec: certmanagerv1.CertificateSpec{
-			CommonName: fmt.Sprintf("%s-client", fullname),
-			Duration:   helmette.MustDuration(duration),
-			IsCA:       false,
-			SecretName: fmt.Sprintf("%s-client", fullname),
-			PrivateKey: &certmanagerv1.CertificatePrivateKey{
-				Algorithm: "ECDSA",
-				Size:      256,
+		certs = append(certs, &certmanagerv1.Certificate{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: "cert-manager.io/v1",
+				Kind:       "Certificate",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      fmt.Sprintf("%s-%s-client", fullname, name),
+				Namespace: dot.Release.Namespace,
+				Labels:    FullLabels(dot),
+			},
+			Spec: certmanagerv1.CertificateSpec{
+				CommonName: fmt.Sprintf("%s--%s-client", fullname, name),
+				Duration:   helmette.MustDuration(duration),
+				IsCA:       false,
+				SecretName: data.ClientSecretName(dot, name),
+				PrivateKey: &certmanagerv1.CertificatePrivateKey{
+					Algorithm: "ECDSA",
+					Size:      256,
+				},
+				IssuerRef: issuerRef,
 			},
-			IssuerRef: issuerRef,
-		},
-	})
+		})
+	}
+
+	return certs
 }

charts/redpanda/chart_test.go

@@ -749,26 +749,50 @@ func httpProxyListenerTest(ctx context.Context, rpk *Client) error {
 
 func mTLSValuesUsingCertManager() *redpanda.PartialValues {
 	return minimalValues(&redpanda.PartialValues{
+		TLS: &redpanda.PartialTLS{
+			Certs: redpanda.PartialTLSCertMap{
+				"kafka": redpanda.PartialTLSCert{
+					Enabled:   ptr.To(true),
+					CAEnabled: ptr.To(true),
+				},
+				"http": redpanda.PartialTLSCert{
+					Enabled:   ptr.To(true),
+					CAEnabled: ptr.To(true),
+				},
+				"rpc": redpanda.PartialTLSCert{
+					Enabled:   ptr.To(true),
+					CAEnabled: ptr.To(true),
+				},
+				"schema": redpanda.PartialTLSCert{
+					Enabled:   ptr.To(true),
+					CAEnabled: ptr.To(true),
+				},
+			},
+		},
 		External:      &redpanda.PartialExternalConfig{Enabled: ptr.To(false)},
 		ClusterDomain: ptr.To("cluster.local"),
 		Listeners: &redpanda.PartialListeners{
 			Admin: &redpanda.PartialListenerConfig[redpanda.NoAuth]{
 				TLS: &redpanda.PartialInternalTLS{
+					// Uses default by default.
 					RequireClientAuth: ptr.To(true),
 				},
 			},
 			HTTP: &redpanda.PartialListenerConfig[redpanda.HTTPAuthenticationMethod]{
 				TLS: &redpanda.PartialInternalTLS{
+					Cert:              ptr.To("http"),
 					RequireClientAuth: ptr.To(true),
 				},
 			},
 			Kafka: &redpanda.PartialListenerConfig[redpanda.KafkaAuthenticationMethod]{
 				TLS: &redpanda.PartialInternalTLS{
+					Cert:              ptr.To("kafka"),
 					RequireClientAuth: ptr.To(true),
 				},
 			},
 			SchemaRegistry: &redpanda.PartialListenerConfig[redpanda.NoAuth]{
 				TLS: &redpanda.PartialInternalTLS{
+					Cert:              ptr.To("schema"),
 					RequireClientAuth: ptr.To(true),
 				},
 			},
@@ -777,6 +801,7 @@ func mTLSValuesUsingCertManager() *redpanda.PartialValues {
 				TLS  *redpanda.PartialInternalTLS `json:"tls,omitempty" jsonschema:"required"`
 			}{
 				TLS: &redpanda.PartialInternalTLS{
+					Cert:              ptr.To("rpc"),
 					RequireClientAuth: ptr.To(true),
 				},
 			},

charts/redpanda/client/client.go

@@ -29,6 +29,7 @@ import (
 	"github.com/twmb/franz-go/pkg/sr"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/redpanda-data/redpanda-operator/charts/redpanda/v5"
@@ -287,35 +288,21 @@ func authFromDot(dot *helmette.Dot) (username string, password string, mechanism
 	return
 }
 
-func certificatesFor(dot *helmette.Dot, cert string) (certSecret, certKey, clientSecret string) {
+func certificatesFor(dot *helmette.Dot, name string) (certSecret, certKey, clientSecret string) {
 	values := helmette.Unwrap[redpanda.Values](dot.Values)
 
-	name := redpanda.Fullname(dot)
+	cert, ok := values.TLS.Certs[name]
+	if !ok || !ptr.Deref(cert.Enabled, true) {
+		// TODO this isn't correct but it matches historical behavior.
+		fullname := redpanda.Fullname(dot)
+		certSecret = fmt.Sprintf("%s-%s-root-certificate", fullname, name)
+		clientSecret = fmt.Sprintf("%s-default-client-cert", fullname)
 
-	// default to cert manager issued names and tls.crt which is
-	// where cert-manager outputs the root CA
-	certKey = corev1.TLSCertKey
-	certSecret = fmt.Sprintf("%s-%s-root-certificate", name, cert)
-	clientSecret = fmt.Sprintf("%s-client", name)
-
-	if certificate, ok := values.TLS.Certs[cert]; ok {
-		// if this references a non-enabled certificate, just return
-		// the default cert-manager issued names
-		if certificate.Enabled != nil && !*certificate.Enabled {
-			return certSecret, certKey, clientSecret
-		}
-
-		if certificate.ClientSecretRef != nil {
-			clientSecret = certificate.ClientSecretRef.Name
-		}
-		if certificate.SecretRef != nil {
-			certSecret = certificate.SecretRef.Name
-			if certificate.CAEnabled {
-				certKey = "ca.crt"
-			}
-		}
+		return certSecret, corev1.TLSCertKey, clientSecret
 	}
-	return certSecret, certKey, clientSecret
+
+	ref := cert.CASecretRef(dot, name)
+	return ref.LocalObjectReference.Name, ref.Key, cert.ClientSecretName(dot, name)
 }
 
 func tlsConfigFromDot(dot *helmette.Dot, listener redpanda.InternalTLS) (*tls.Config, error) {