deprecated auth fields were removed & Mtls cerctificate creation was …

…implemented

PROJECT

@@ -272,6 +272,7 @@ resources:
   version: v1beta1
   webhooks:
     defaulting: true
+    validation: true
     webhookVersion: v1
 - api:
     crdVersion: v1

apis/clusters/v1beta1/kafka_types.go

@@ -79,14 +79,12 @@ type KafkaSpec struct {
 
 	// Provision additional dedicated nodes for Apache Zookeeper to run on.
 	// Zookeeper nodes will be co-located with Kafka if this is not provided
-	DedicatedZookeeper                []*DedicatedZookeeper     `json:"dedicatedZookeeper,omitempty"`
-	ClientBrokerAuthWithMTLS          bool                      `json:"clientBrokerAuthWithMtls,omitempty"`
-	ClientAuthBrokerWithoutEncryption bool                      `json:"clientAuthBrokerWithoutEncryption,omitempty"`
-	ClientAuthBrokerWithEncryption    bool                      `json:"clientAuthBrokerWithEncryption,omitempty"`
-	KarapaceRestProxy                 []*KarapaceRestProxy      `json:"karapaceRestProxy,omitempty"`
-	KarapaceSchemaRegistry            []*KarapaceSchemaRegistry `json:"karapaceSchemaRegistry,omitempty"`
-	BundledUseOnly                    bool                      `json:"bundledUseOnly,omitempty"`
-	UserRefs                          []*UserReference          `json:"userRefs,omitempty"`
+	DedicatedZookeeper       []*DedicatedZookeeper     `json:"dedicatedZookeeper,omitempty"`
+	ClientBrokerAuthWithMTLS bool                      `json:"clientBrokerAuthWithMtls,omitempty"`
+	KarapaceRestProxy        []*KarapaceRestProxy      `json:"karapaceRestProxy,omitempty"`
+	KarapaceSchemaRegistry   []*KarapaceSchemaRegistry `json:"karapaceSchemaRegistry,omitempty"`
+	BundledUseOnly           bool                      `json:"bundledUseOnly,omitempty"`
+	UserRefs                 []*UserReference          `json:"userRefs,omitempty"`
 }
 
 type KafkaDataCentre struct {
@@ -140,27 +138,25 @@ func (k *Kafka) NewPatch() client.Patch {
 
 func (k *KafkaSpec) ToInstAPI() *models.KafkaCluster {
 	return &models.KafkaCluster{
-		SchemaRegistry:                    k.schemaRegistryToInstAPI(),
-		RestProxy:                         k.restProxyToInstAPI(),
-		PCIComplianceMode:                 k.PCICompliance,
-		DefaultReplicationFactor:          k.ReplicationFactor,
-		DefaultNumberOfPartitions:         k.PartitionsNumber,
-		TwoFactorDelete:                   k.TwoFactorDeletesToInstAPI(),
-		AllowDeleteTopics:                 k.AllowDeleteTopics,
-		AutoCreateTopics:                  k.AutoCreateTopics,
-		ClientToClusterEncryption:         k.ClientToClusterEncryption,
-		DedicatedZookeeper:                k.dedicatedZookeeperToInstAPI(),
-		PrivateNetworkCluster:             k.PrivateNetworkCluster,
-		Name:                              k.Name,
-		SLATier:                           k.SLATier,
-		KafkaVersion:                      k.Version,
-		DataCentres:                       k.dcToInstAPI(),
-		ClientBrokerAuthWithMtls:          k.ClientBrokerAuthWithMTLS,
-		ClientAuthBrokerWithoutEncryption: k.ClientAuthBrokerWithoutEncryption,
-		ClientAuthBrokerWithEncryption:    k.ClientAuthBrokerWithEncryption,
-		BundledUseOnly:                    k.BundledUseOnly,
-		KarapaceRestProxy:                 k.karapaceRestProxyToInstAPI(),
-		KarapaceSchemaRegistry:            k.karapaceSchemaRegistryToInstAPI(),
+		SchemaRegistry:            k.schemaRegistryToInstAPI(),
+		RestProxy:                 k.restProxyToInstAPI(),
+		PCIComplianceMode:         k.PCICompliance,
+		DefaultReplicationFactor:  k.ReplicationFactor,
+		DefaultNumberOfPartitions: k.PartitionsNumber,
+		TwoFactorDelete:           k.TwoFactorDeletesToInstAPI(),
+		AllowDeleteTopics:         k.AllowDeleteTopics,
+		AutoCreateTopics:          k.AutoCreateTopics,
+		ClientToClusterEncryption: k.ClientToClusterEncryption,
+		DedicatedZookeeper:        k.dedicatedZookeeperToInstAPI(),
+		PrivateNetworkCluster:     k.PrivateNetworkCluster,
+		Name:                      k.Name,
+		SLATier:                   k.SLATier,
+		KafkaVersion:              k.Version,
+		DataCentres:               k.dcToInstAPI(),
+		ClientBrokerAuthWithMtls:  k.ClientBrokerAuthWithMTLS,
+		BundledUseOnly:            k.BundledUseOnly,
+		KarapaceRestProxy:         k.karapaceRestProxyToInstAPI(),
+		KarapaceSchemaRegistry:    k.karapaceSchemaRegistryToInstAPI(),
 	}
 }
 
@@ -283,21 +279,19 @@ func (ks *KafkaSpec) FromInstAPI(iKafka *models.KafkaCluster) KafkaSpec {
 			SLATier:               iKafka.SLATier,
 			TwoFactorDelete:       ks.Cluster.TwoFactorDeleteFromInstAPI(iKafka.TwoFactorDelete),
 		},
-		SchemaRegistry:                    ks.SchemaRegistryFromInstAPI(iKafka.SchemaRegistry),
-		ReplicationFactor:                 iKafka.DefaultReplicationFactor,
-		PartitionsNumber:                  iKafka.DefaultNumberOfPartitions,
-		RestProxy:                         ks.RestProxyFromInstAPI(iKafka.RestProxy),
-		AllowDeleteTopics:                 iKafka.AllowDeleteTopics,
-		AutoCreateTopics:                  iKafka.AutoCreateTopics,
-		ClientToClusterEncryption:         iKafka.ClientToClusterEncryption,
-		DataCentres:                       ks.DCsFromInstAPI(iKafka.DataCentres),
-		DedicatedZookeeper:                ks.DedicatedZookeeperFromInstAPI(iKafka.DedicatedZookeeper),
-		ClientBrokerAuthWithMTLS:          iKafka.ClientBrokerAuthWithMtls,
-		ClientAuthBrokerWithoutEncryption: iKafka.ClientAuthBrokerWithoutEncryption,
-		ClientAuthBrokerWithEncryption:    iKafka.ClientAuthBrokerWithEncryption,
-		KarapaceRestProxy:                 ks.KarapaceRestProxyFromInstAPI(iKafka.KarapaceRestProxy),
-		KarapaceSchemaRegistry:            ks.KarapaceSchemaRegistryFromInstAPI(iKafka.KarapaceSchemaRegistry),
-		BundledUseOnly:                    iKafka.BundledUseOnly,
+		SchemaRegistry:            ks.SchemaRegistryFromInstAPI(iKafka.SchemaRegistry),
+		ReplicationFactor:         iKafka.DefaultReplicationFactor,
+		PartitionsNumber:          iKafka.DefaultNumberOfPartitions,
+		RestProxy:                 ks.RestProxyFromInstAPI(iKafka.RestProxy),
+		AllowDeleteTopics:         iKafka.AllowDeleteTopics,
+		AutoCreateTopics:          iKafka.AutoCreateTopics,
+		ClientToClusterEncryption: iKafka.ClientToClusterEncryption,
+		DataCentres:               ks.DCsFromInstAPI(iKafka.DataCentres),
+		DedicatedZookeeper:        ks.DedicatedZookeeperFromInstAPI(iKafka.DedicatedZookeeper),
+		ClientBrokerAuthWithMTLS:  iKafka.ClientBrokerAuthWithMtls,
+		KarapaceRestProxy:         ks.KarapaceRestProxyFromInstAPI(iKafka.KarapaceRestProxy),
+		KarapaceSchemaRegistry:    ks.KarapaceSchemaRegistryFromInstAPI(iKafka.KarapaceSchemaRegistry),
+		BundledUseOnly:            iKafka.BundledUseOnly,
 	}
 }
 
@@ -399,8 +393,6 @@ func (a *KafkaSpec) IsEqual(b KafkaSpec) bool {
 		a.AutoCreateTopics == b.AutoCreateTopics &&
 		a.ClientToClusterEncryption == b.ClientToClusterEncryption &&
 		a.ClientBrokerAuthWithMTLS == b.ClientBrokerAuthWithMTLS &&
-		a.ClientAuthBrokerWithoutEncryption == b.ClientAuthBrokerWithoutEncryption &&
-		a.ClientAuthBrokerWithEncryption == b.ClientAuthBrokerWithEncryption &&
 		a.BundledUseOnly == b.BundledUseOnly &&
 		isKafkaAddonsEqual[SchemaRegistry](a.SchemaRegistry, b.SchemaRegistry) &&
 		isKafkaAddonsEqual[RestProxy](a.RestProxy, b.RestProxy) &&

apis/clusters/v1beta1/kafka_webhook.go

@@ -318,31 +318,27 @@ type immutableKafkaFields struct {
 }
 
 type specificKafkaFields struct {
-	replicationFactor                 int
-	partitionsNumber                  int
-	allowDeleteTopics                 bool
-	autoCreateTopics                  bool
-	clientToClusterEncryption         bool
-	bundledUseOnly                    bool
-	privateNetworkCluster             bool
-	clientAuthBrokerWithEncryption    bool
-	clientAuthBrokerWithoutEncryption bool
-	clientBrokerAuthWithMtls          bool
+	replicationFactor         int
+	partitionsNumber          int
+	allowDeleteTopics         bool
+	autoCreateTopics          bool
+	clientToClusterEncryption bool
+	bundledUseOnly            bool
+	privateNetworkCluster     bool
+	clientBrokerAuthWithMtls  bool
 }
 
 func (ks *KafkaSpec) newKafkaImmutableFields() *immutableKafkaFields {
 	return &immutableKafkaFields{
 		specificFields: specificKafkaFields{
-			replicationFactor:                 ks.ReplicationFactor,
-			partitionsNumber:                  ks.PartitionsNumber,
-			allowDeleteTopics:                 ks.AllowDeleteTopics,
-			autoCreateTopics:                  ks.AutoCreateTopics,
-			clientToClusterEncryption:         ks.ClientToClusterEncryption,
-			bundledUseOnly:                    ks.BundledUseOnly,
-			privateNetworkCluster:             ks.PrivateNetworkCluster,
-			clientAuthBrokerWithEncryption:    ks.ClientAuthBrokerWithEncryption,
-			clientAuthBrokerWithoutEncryption: ks.ClientAuthBrokerWithoutEncryption,
-			clientBrokerAuthWithMtls:          ks.ClientBrokerAuthWithMTLS,
+			replicationFactor:         ks.ReplicationFactor,
+			partitionsNumber:          ks.PartitionsNumber,
+			allowDeleteTopics:         ks.AllowDeleteTopics,
+			autoCreateTopics:          ks.AutoCreateTopics,
+			clientToClusterEncryption: ks.ClientToClusterEncryption,
+			bundledUseOnly:            ks.BundledUseOnly,
+			privateNetworkCluster:     ks.PrivateNetworkCluster,
+			clientBrokerAuthWithMtls:  ks.ClientBrokerAuthWithMTLS,
 		},
 		cluster: ks.Cluster.newImmutableFields(),
 	}

apis/kafkamanagement/v1beta1/kafkauser_types.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1beta1
 
 import (
+	k8sCore "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -26,9 +27,10 @@ import (
 
 // KafkaUserSpec defines the desired state of KafkaUser
 type KafkaUserSpec struct {
-	Options            *KafkaUserOptions        `json:"options"`
-	SecretRef          *v1beta1.SecretReference `json:"secretRef"`
-	InitialPermissions string                   `json:"initialPermissions"`
+	Options             *KafkaUserOptions        `json:"options"`
+	SecretRef           *v1beta1.SecretReference `json:"secretRef"`
+	CertificateRequests []*CertificateRequest    `json:"certificateRequests,omitempty"`
+	InitialPermissions  string                   `json:"initialPermissions"`
 }
 
 type KafkaUserOptions struct {
@@ -41,6 +43,25 @@ type KafkaUserStatus struct {
 	ClustersEvents map[string]string `json:"clustersEvents,omitempty"`
 }
 
+type Certificate struct {
+	ID                string `json:"id,omitempty"`
+	ExpiryDate        string `json:"expiryDate,omitempty"`
+	SignedCertificate string `json:"signedCertificate,omitempty"`
+}
+
+type CertificateRequest struct {
+	SecretName         string `json:"secretName"`
+	SecretNamespace    string `json:"secretNamespace"`
+	ClusterID          string `json:"clusterId"`
+	CSR                string `json:"csr,omitempty"`
+	ValidPeriod        int    `json:"validPeriod"`
+	CommonName         string `json:"commonName,omitempty"`
+	Country            string `json:"country,omitempty"`
+	Organization       string `json:"organization,omitempty"`
+	OrganizationalUnit string `json:"organizationalUnit,omitempty"`
+	AutoRenew          bool   `json:"autoRenew"`
+}
+
 //+kubebuilder:object:root=true
 //+kubebuilder:subresource:status
 
@@ -83,6 +104,21 @@ func (ku *KafkaUser) GetID(clusterID, name string) string {
 	return clusterID + "_" + name
 }
 
+func (ku *KafkaUser) NewCertificateSecret(name, namespace string) *k8sCore.Secret {
+	return &k8sCore.Secret{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       models.SecretKind,
+			APIVersion: models.K8sAPIVersionV1,
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+
+		StringData: map[string]string{},
+	}
+}
+
 func init() {
 	SchemeBuilder.Register(&KafkaUser{}, &KafkaUserList{})
 }
@@ -104,3 +140,12 @@ func (ko *KafkaUserOptions) ToInstAPI() *models.KafkaUserOptions {
 	}
 
 }
+
+func (cr *CertificateRequest) ToInstAPI(username string) *models.CertificateRequest {
+	return &models.CertificateRequest{
+		ClusterID:     cr.ClusterID,
+		CSR:           cr.CSR,
+		KafkaUsername: username,
+		ValidPeriod:   cr.ValidPeriod,
+	}
+}

apis/kafkamanagement/v1beta1/kafkauser_webhook.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1beta1
 
 import (
+	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
@@ -47,3 +48,42 @@ func (r *KafkaUser) Default() {
 		})
 	}
 }
+
+// TODO(user): change verbs to "verbs=create;update;delete" if you want to enable deletion validation.
+//+kubebuilder:webhook:path=/validate-kafkamanagement-instaclustr-com-v1beta1-kafkauser,mutating=false,failurePolicy=fail,sideEffects=None,groups=kafkamanagement.instaclustr.com,resources=kafkausers,verbs=create;update,versions=v1beta1,name=vkafkauser.kb.io,admissionReviewVersions=v1
+
+var _ webhook.Validator = &KafkaUser{}
+
+// ValidateCreate implements webhook.Validator so a webhook will be registered for the type
+func (r *KafkaUser) ValidateCreate() error {
+	kafkauserlog.Info("validate create", "name", r.Name)
+
+	if len(r.Spec.CertificateRequests) != 0 {
+		return models.ErrNotEmptyCSRs
+	}
+
+	return nil
+}
+
+// ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
+func (r *KafkaUser) ValidateUpdate(old runtime.Object) error {
+	kafkauserlog.Info("validate update", "name", r.Name)
+
+	for _, request := range r.Spec.CertificateRequests {
+		if request.CSR == "" {
+			if request.Organization == "" || request.OrganizationalUnit == "" || request.Country == "" || request.CommonName == "" {
+				return models.ErrEmptyCertGeneratingFields
+			}
+		}
+	}
+
+	return nil
+}
+
+// ValidateDelete implements webhook.Validator so a webhook will be registered for the type
+func (r *KafkaUser) ValidateDelete() error {
+	kafkauserlog.Info("validate delete", "name", r.Name)
+
+	// TODO(user): fill in your validation logic upon object deletion.
+	return nil
+}

config/crd/bases/clusters.instaclustr.com_kafkas.yaml

@@ -41,10 +41,6 @@ spec:
                 type: boolean
               bundledUseOnly:
                 type: boolean
-              clientAuthBrokerWithEncryption:
-                type: boolean
-              clientAuthBrokerWithoutEncryption:
-                type: boolean
               clientBrokerAuthWithMtls:
                 type: boolean
               clientToClusterEncryption: