Operator mtls b

src/go/k8s/apis/redpanda/v1alpha1/cluster_types.go

@@ -732,7 +732,9 @@ type AdminAPITLS struct {
 // truststore when communicating with Redpanda.
 //
 // If RequireClientAuth is set to true, two-way TLS verification is enabled.
-// In that case, a client certificate is generated, which can be retrieved from
+// If ClientCACertRef is provided, the operator will configure the Pandaproxy to
+// use the CA cert it contains.
+// Otherwise, a client certificate is generated, which can be retrieved from
 // the Secret named '<redpanda-cluster-name>-proxy-api-client'.
 //
 // All TLS secrets are stored in the same namespace as the Redpanda cluster.
@@ -755,6 +757,11 @@ type PandaproxyAPITLS struct {
 	// duplicate the secret to the same namespace as redpanda CRD to be able to
 	// mount it to the nodes
 	NodeSecretRef *corev1.ObjectReference `json:"nodeSecretRef,omitempty"`
+	// If ClientCACertRef points to a secret containing the trusted CA certificates.
+	// If provided and RequireClientAuth is true, the operator uses the certificate
+	// in this secret instead of issuing client certificates. The secret is expected to provide
+	// the following keys: 'ca.crt'.
+	ClientCACertRef *corev1.TypedLocalObjectReference `json:"clientCACertRef,omitempty"`
 	// Enables two-way verification on the server side. If enabled, all
 	// Pandaproxy API clients are required to have a valid client certificate.
 	RequireClientAuth bool `json:"requireClientAuth,omitempty"`
@@ -769,7 +776,9 @@ type PandaproxyAPITLS struct {
 // truststore when communicating with Schema registry.
 //
 // If RequireClientAuth is set to true, two-way TLS verification is enabled.
-// In that case, a client certificate is generated, which can be retrieved from
+// If ClientCACertRef is provided, the operator will configure the Schema Registry to
+// use the CA cert it contains.
+// Otherwise a client certificate is generated, which can be retrieved from
 // the Secret named '<redpanda-cluster-name>-schema-registry-client'.
 //
 // All TLS secrets are stored in the same namespace as the Redpanda cluster.
@@ -792,6 +801,11 @@ type SchemaRegistryAPITLS struct {
 	// duplicate the secret to the same namespace as redpanda CRD to be able to
 	// mount it to the nodes
 	NodeSecretRef *corev1.ObjectReference `json:"nodeSecretRef,omitempty"`
+	// If ClientCACertRef points to a secret containing the trusted CA certificates.
+	// If provided and RequireClientAuth is true, the operator uses the certificate
+	// in this secret instead of issuing client certificates. The secret is expected to provide
+	// the following keys: 'ca.crt'.
+	ClientCACertRef *corev1.TypedLocalObjectReference `json:"clientCACertRef,omitempty"`
 	// Enables two-way verification on the server side. If enabled, all SchemaRegistry
 	// clients are required to have a valid client certificate.
 	RequireClientAuth bool `json:"requireClientAuth,omitempty"`
@@ -1088,6 +1102,8 @@ type TLSConfig struct {
 	RequireClientAuth bool                    `json:"requireClientAuth,omitempty"`
 	IssuerRef         *cmmeta.ObjectReference `json:"issuerRef,omitempty"`
 	NodeSecretRef     *corev1.ObjectReference `json:"nodeSecretRef,omitempty"`
+
+	ClientCACertRef *corev1.TypedLocalObjectReference `json:"clientCACertRef,omitempty"`
 }
 
 // Kafka API
@@ -1187,6 +1203,7 @@ func (s SchemaRegistryAPI) GetTLS() *TLSConfig {
 		RequireClientAuth: s.TLS.RequireClientAuth,
 		IssuerRef:         s.TLS.IssuerRef,
 		NodeSecretRef:     s.TLS.NodeSecretRef,
+		ClientCACertRef:   s.TLS.ClientCACertRef,
 	}
 }
 
@@ -1216,6 +1233,7 @@ func (p PandaproxyAPI) GetTLS() *TLSConfig {
 		RequireClientAuth: p.TLS.RequireClientAuth,
 		IssuerRef:         p.TLS.IssuerRef,
 		NodeSecretRef:     p.TLS.NodeSecretRef,
+		ClientCACertRef:   p.TLS.ClientCACertRef,
 	}
 }
 
@@ -1232,6 +1250,7 @@ func defaultTLSConfig() *TLSConfig {
 		RequireClientAuth: false,
 		IssuerRef:         nil,
 		NodeSecretRef:     nil,
+		ClientCACertRef:   nil,
 	}
 }
 

src/go/k8s/apis/redpanda/v1alpha1/cluster_webhook.go

@@ -11,16 +11,20 @@ package v1alpha1
 
 import (
 	"context"
+	"crypto/x509"
+	"encoding/pem"
 	"fmt"
 	"math"
 	"regexp"
 	"strconv"
+	"strings"
 
 	cmmeta "github.com/cert-manager/cert-manager/pkg/apis/meta/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -85,6 +89,11 @@ var log = logf.Log.WithName("cluster-resource")
 // kclient is controller-runtime client.
 var kclient client.Client
 
+// SetK8sClient sets kclient for testing
+func SetK8sClient(c client.Client) {
+	kclient = c
+}
+
 // SetupWebhookWithManager autogenerated function by kubebuilder
 func (r *Cluster) SetupWebhookWithManager(mgr ctrl.Manager) error {
 	// We should use lower level webhook package
@@ -361,9 +370,11 @@ func (r *Cluster) validateKafkaListeners() field.ErrorList {
 			p.TLS.RequireClientAuth,
 			p.TLS.IssuerRef,
 			p.TLS.NodeSecretRef,
+			nil,
 			field.NewPath("spec").Child("configuration").Child("kafkaApi").Index(i).Child("tls"),
 			&p.External,
-			field.NewPath("spec").Child("configuration").Child("kafkaApi").Index(i).Child("external"))
+			field.NewPath("spec").Child("configuration").Child("kafkaApi").Index(i).Child("external"),
+			r.GetNamespace())
 		allErrs = append(allErrs, tlsErrs...)
 
 		switch r.Spec.Configuration.KafkaAPI[i].AuthenticationMethod {
@@ -532,9 +543,11 @@ func (r *Cluster) validatePandaproxyListeners() field.ErrorList {
 			p[i].TLS.RequireClientAuth,
 			p[i].TLS.IssuerRef,
 			p[i].TLS.NodeSecretRef,
+			p[i].TLS.ClientCACertRef,
 			field.NewPath("spec").Child("configuration").Child("pandaproxyApi").Index(i).Child("tls"),
 			&p[i].External.ExternalConnectivityConfig,
 			field.NewPath("spec").Child("configuration").Child("pandaproxyApi").Index(i).Child("external"),
+			r.GetNamespace(),
 		)
 		allErrs = append(allErrs, tlsErrs...)
 	}
@@ -569,9 +582,11 @@ func (r *Cluster) validateSchemaRegistryListener() field.ErrorList {
 			schemaRegistry.TLS.RequireClientAuth,
 			schemaRegistry.TLS.IssuerRef,
 			schemaRegistry.TLS.NodeSecretRef,
+			schemaRegistry.TLS.ClientCACertRef,
 			field.NewPath("spec").Child("configuration").Child("schemaRegistry").Child("tls"),
 			schemaRegistry.GetExternal(),
 			field.NewPath("spec").Child("configuration").Child("schemaRegistry").Child("external"),
+			r.GetNamespace(),
 		)
 		allErrs = append(allErrs, tlsErrs...)
 	}
@@ -829,9 +844,11 @@ func validateListener(
 	tlsEnabled, requireClientAuth bool,
 	issuerRef *cmmeta.ObjectReference,
 	nodeSecretRef *corev1.ObjectReference,
+	clientCACertRef *corev1.TypedLocalObjectReference,
 	path *field.Path,
 	external *ExternalConnectivityConfig,
 	externalPath *field.Path,
+	clusterNamespace string,
 ) field.ErrorList {
 	var allErrs field.ErrorList
 	if requireClientAuth && !tlsEnabled {
@@ -855,6 +872,88 @@ func validateListener(
 				external.Subdomain,
 				"TLS requires specifying a subdomain"))
 	}
+	if !tlsEnabled || clientCACertRef == nil {
+		return allErrs
+	}
+
+	return validateExternalCA(tlsEnabled, requireClientAuth, clientCACertRef, path, clusterNamespace)
+}
+
+func validateExternalCA(
+	tlsEnabled, requireClientAuth bool,
+	clientCACertRef *corev1.TypedLocalObjectReference,
+	path *field.Path,
+	clusterNamespace string,
+) field.ErrorList {
+	var allErrs field.ErrorList
+
+	if !requireClientAuth {
+		allErrs = append(allErrs,
+			field.Invalid(
+				path.Child("requireClientAuth"),
+				requireClientAuth,
+				"Enabled has to be set to true for RequireClientAuth if ClientCACertRef is set"))
+	}
+
+	if clientCACertRef.Name == "" {
+		allErrs = append(allErrs,
+			field.Invalid(
+				path.Child("clientCACertRef"),
+				clientCACertRef,
+				"Name must be provided if ClientCACertRef is set"))
+	}
+
+	if clientCACertRef.Kind != "" && !strings.EqualFold(clientCACertRef.Kind, "Secret") {
+		allErrs = append(allErrs,
+			field.Invalid(
+				path.Child("clientCACertRef"),
+				clientCACertRef,
+				"Kind must be set to secret if set in ClientCACertRef"))
+	}
+
+	if len(allErrs) > 0 {
+		return allErrs
+	}
+
+	secret := &corev1.Secret{}
+	err := kclient.Get(context.TODO(), types.NamespacedName{Name: clientCACertRef.Name, Namespace: clusterNamespace}, secret)
+	if err != nil {
+		allErrs = append(allErrs,
+			field.Invalid(
+				path.Child("clientCACertRef"),
+				clientCACertRef,
+				"Failed to get secret: "+err.Error()))
+		return allErrs
+	}
+
+	crt, found := secret.Data[cmmeta.TLSCAKey]
+	if !found {
+		allErrs = append(allErrs,
+			field.Invalid(
+				path.Child("clientCACertRef"),
+				clientCACertRef,
+				"ca.crt must be set in the client CA secret"))
+		return allErrs
+	}
+
+	pb, _ := pem.Decode(crt)
+	if pb == nil {
+		allErrs = append(allErrs,
+			field.Invalid(
+				path.Child("clientCACertRef"),
+				clientCACertRef,
+				"Invalid certificate in the client CA secret"))
+	} else {
+		_, err := x509.ParseCertificate(pb.Bytes)
+		if err != nil {
+			allErrs = append(allErrs,
+				field.Invalid(
+					path.Child("clientCACertRef"),
+					clientCACertRef,
+					"Invalid certificate in the client CA secret: "+err.Error()))
+		}
+	}
+
 	return allErrs
 }