Merge pull request #74 from rabbitmq/queue-exchange-webhook

validation webhook for queues.rabbitmq.com and exchanges.rabbitmq.com

api/v1alpha1/exchange_types.go

@@ -12,6 +12,7 @@ package v1alpha1
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
 // ExchangeSpec defines the desired state of Exchange
@@ -64,6 +65,13 @@ type ExchangeList struct {
 	Items           []Exchange `json:"items"`
 }
 
+func (e *Exchange) GroupResource() schema.GroupResource {
+	return schema.GroupResource{
+		Group:    e.GroupVersionKind().Group,
+		Resource: e.GroupVersionKind().Kind,
+	}
+}
+
 func init() {
 	SchemeBuilder.Register(&Exchange{}, &ExchangeList{})
 }

api/v1alpha1/exchange_webhook.go

@@ -0,0 +1,88 @@
+package v1alpha1
+
+import (
+	"fmt"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+)
+
+func (r *Exchange) SetupWebhookWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewWebhookManagedBy(mgr).
+		For(r).
+		Complete()
+}
+
+// +kubebuilder:webhook:verbs=create;update,path=/validate-rabbitmq-com-v1alpha1-exchange,mutating=false,failurePolicy=fail,groups=rabbitmq.com,resources=exchanges,versions=v1alpha1,name=vexchange.kb.io,sideEffects=none,admissionReviewVersions=v1
+
+var _ webhook.Validator = &Exchange{}
+
+// no validation on create
+func (e *Exchange) ValidateCreate() error {
+	return nil
+}
+
+// ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
+// returns error type 'forbidden' for updates that the controller chooses to disallow: exchange name/vhost/rabbitmqClusterReference
+// returns error type 'invalid' for updates that will be rejected by rabbitmq server: exchange types/autoDelete/durable
+// exchange.spec.arguments can be updated
+func (e *Exchange) ValidateUpdate(old runtime.Object) error {
+	oldExchange, ok := old.(*Exchange)
+	if !ok {
+		return apierrors.NewBadRequest(fmt.Sprintf("expected an exchange but got a %T", old))
+	}
+
+	var allErrs field.ErrorList
+	detailMsg := "updates on name, vhost, and rabbitmqClusterReference are all forbidden"
+	if e.Spec.Name != oldExchange.Spec.Name {
+		return apierrors.NewForbidden(e.GroupResource(), e.Name,
+			field.Forbidden(field.NewPath("spec", "name"), detailMsg))
+	}
+
+	if e.Spec.Vhost != oldExchange.Spec.Vhost {
+		return apierrors.NewForbidden(e.GroupResource(), e.Name,
+			field.Forbidden(field.NewPath("spec", "vhost"), detailMsg))
+	}
+
+	if e.Spec.RabbitmqClusterReference != oldExchange.Spec.RabbitmqClusterReference {
+		return apierrors.NewForbidden(e.GroupResource(), e.Name,
+			field.Forbidden(field.NewPath("spec", "rabbitmqClusterReference"), detailMsg))
+	}
+
+	if e.Spec.Type != oldExchange.Spec.Type {
+		allErrs = append(allErrs, field.Invalid(
+			field.NewPath("spec", "type"),
+			e.Spec.Type,
+			"exchange type cannot be updated",
+		))
+	}
+
+	if e.Spec.AutoDelete != oldExchange.Spec.AutoDelete {
+		allErrs = append(allErrs, field.Invalid(
+			field.NewPath("spec", "autoDelete"),
+			e.Spec.AutoDelete,
+			"autoDelete cannot be updated",
+		))
+	}
+
+	if e.Spec.Durable != oldExchange.Spec.Durable {
+		allErrs = append(allErrs, field.Invalid(
+			field.NewPath("spec", "durable"),
+			e.Spec.AutoDelete,
+			"durable cannot be updated",
+		))
+	}
+
+	if len(allErrs) == 0 {
+		return nil
+	}
+
+	return apierrors.NewInvalid(GroupVersion.WithKind("Exchange").GroupKind(), e.Name, allErrs)
+}
+
+// no validation on delete
+func (e *Exchange) ValidateDelete() error {
+	return nil
+}

api/v1alpha1/exchange_webhook_test.go

@@ -0,0 +1,74 @@
+package v1alpha1
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+var _ = Describe("exchange webhook", func() {
+
+	var exchange = Exchange{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "update-binding",
+		},
+		Spec: ExchangeSpec{
+			Name:       "test",
+			Vhost:      "/test",
+			Type:       "fanout",
+			Durable:    false,
+			AutoDelete: true,
+			RabbitmqClusterReference: RabbitmqClusterReference{
+				Name:      "some-cluster",
+				Namespace: "default",
+			},
+		},
+	}
+
+	It("does not allow updates on exchange name", func() {
+		newExchange := exchange.DeepCopy()
+		newExchange.Spec.Name = "new-name"
+		Expect(apierrors.IsForbidden(newExchange.ValidateUpdate(&exchange))).To(BeTrue())
+	})
+
+	It("does not allow updates on vhost", func() {
+		newExchange := exchange.DeepCopy()
+		newExchange.Spec.Vhost = "/a-new-vhost"
+		Expect(apierrors.IsForbidden(newExchange.ValidateUpdate(&exchange))).To(BeTrue())
+	})
+
+	It("does not allow updates on RabbitmqClusterReference", func() {
+		newExchange := exchange.DeepCopy()
+		newExchange.Spec.RabbitmqClusterReference = RabbitmqClusterReference{
+			Name:      "new-cluster",
+			Namespace: "default",
+		}
+		Expect(apierrors.IsForbidden(newExchange.ValidateUpdate(&exchange))).To(BeTrue())
+	})
+
+	It("does not allow updates on exchange type", func() {
+		newExchange := exchange.DeepCopy()
+		newExchange.Spec.Type = "direct"
+		Expect(apierrors.IsInvalid(newExchange.ValidateUpdate(&exchange))).To(BeTrue())
+	})
+
+	It("does not allow updates on durable", func() {
+		newExchange := exchange.DeepCopy()
+		newExchange.Spec.Durable = true
+		Expect(apierrors.IsInvalid(newExchange.ValidateUpdate(&exchange))).To(BeTrue())
+	})
+
+	It("does not allow updates on autoDelete", func() {
+		newExchange := exchange.DeepCopy()
+		newExchange.Spec.AutoDelete = false
+		Expect(apierrors.IsInvalid(newExchange.ValidateUpdate(&exchange))).To(BeTrue())
+	})
+
+	It("allows updates on arguments", func() {
+		newExchange := exchange.DeepCopy()
+		newExchange.Spec.Arguments = &runtime.RawExtension{Raw: []byte(`{"new":"new-value"}`)}
+		Expect(newExchange.ValidateUpdate(&exchange)).To(Succeed())
+	})
+})

api/v1alpha1/queue_types.go

@@ -12,6 +12,7 @@ package v1alpha1
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
 // using runtime.RawExtension to represent queue arguments
@@ -77,6 +78,13 @@ type QueueList struct {
 	Items           []Queue `json:"items"`
 }
 
+func (q *Queue) GroupResource() schema.GroupResource {
+	return schema.GroupResource{
+		Group:    q.GroupVersionKind().Group,
+		Resource: q.GroupVersionKind().Kind,
+	}
+}
+
 func init() {
 	SchemeBuilder.Register(&Queue{}, &QueueList{})
 }

api/v1alpha1/queue_webhook.go

@@ -0,0 +1,88 @@
+package v1alpha1
+
+import (
+	"fmt"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+)
+
+func (r *Queue) SetupWebhookWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewWebhookManagedBy(mgr).
+		For(r).
+		Complete()
+}
+
+// +kubebuilder:webhook:verbs=create;update,path=/validate-rabbitmq-com-v1alpha1-queue,mutating=false,failurePolicy=fail,groups=rabbitmq.com,resources=queues,versions=v1alpha1,name=vqueue.kb.io,sideEffects=none,admissionReviewVersions=v1sideEffects=none,admissionReviewVersions=v1
+
+var _ webhook.Validator = &Queue{}
+
+// no validation on create
+func (q *Queue) ValidateCreate() error {
+	return nil
+}
+
+// ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
+// returns error type 'forbidden' for updates that the controller chooses to disallow: queue name/vhost/rabbitmqClusterReference
+// returns error type 'invalid' for updates that will be rejected by rabbitmq server: queue types/autoDelete/durable
+// queue arguments not handled because implementation couldn't change
+func (q *Queue) ValidateUpdate(old runtime.Object) error {
+	oldQueue, ok := old.(*Queue)
+	if !ok {
+		return apierrors.NewBadRequest(fmt.Sprintf("expected a queue but got a %T", old))
+	}
+
+	var allErrs field.ErrorList
+	detailMsg := "updates on name, vhost, and rabbitmqClusterReference are all forbidden"
+	if q.Spec.Name != oldQueue.Spec.Name {
+		return apierrors.NewForbidden(q.GroupResource(), q.Name,
+			field.Forbidden(field.NewPath("spec", "name"), detailMsg))
+	}
+
+	if q.Spec.Vhost != oldQueue.Spec.Vhost {
+		return apierrors.NewForbidden(q.GroupResource(), q.Name,
+			field.Forbidden(field.NewPath("spec", "vhost"), detailMsg))
+	}
+
+	if q.Spec.RabbitmqClusterReference != oldQueue.Spec.RabbitmqClusterReference {
+		return apierrors.NewForbidden(q.GroupResource(), q.Name,
+			field.Forbidden(field.NewPath("spec", "rabbitmqClusterReference"), detailMsg))
+	}
+
+	if q.Spec.Type != oldQueue.Spec.Type {
+		allErrs = append(allErrs, field.Invalid(
+			field.NewPath("spec", "type"),
+			q.Spec.Type,
+			"queue type cannot be updated",
+		))
+	}
+
+	if q.Spec.AutoDelete != oldQueue.Spec.AutoDelete {
+		allErrs = append(allErrs, field.Invalid(
+			field.NewPath("spec", "autoDelete"),
+			q.Spec.AutoDelete,
+			"autoDelete cannot be updated",
+		))
+	}
+
+	if q.Spec.Durable != oldQueue.Spec.Durable {
+		allErrs = append(allErrs, field.Invalid(
+			field.NewPath("spec", "durable"),
+			q.Spec.AutoDelete,
+			"durable cannot be updated",
+		))
+	}
+
+	if len(allErrs) == 0 {
+		return nil
+	}
+
+	return apierrors.NewInvalid(GroupVersion.WithKind("Queue").GroupKind(), q.Name, allErrs)
+}
+
+// no validation on delete
+func (q *Queue) ValidateDelete() error {
+	return nil
+}

api/v1alpha1/queue_webhook_test.go

@@ -0,0 +1,67 @@
+package v1alpha1
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var _ = Describe("queue webhook", func() {
+
+	var queue = Queue{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "update-binding",
+		},
+		Spec: QueueSpec{
+			Name:       "test",
+			Vhost:      "/a-vhost",
+			Type:       "quorum",
+			Durable:    false,
+			AutoDelete: true,
+			RabbitmqClusterReference: RabbitmqClusterReference{
+				Name:      "some-cluster",
+				Namespace: "default",
+			},
+		},
+	}
+
+	It("does not allow updates on queue name", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.Name = "new-name"
+		Expect(apierrors.IsForbidden(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+
+	It("does not allow updates on vhost", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.Vhost = "/new-vhost"
+		Expect(apierrors.IsForbidden(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+
+	It("does not allow updates on RabbitmqClusterReference", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.RabbitmqClusterReference = RabbitmqClusterReference{
+			Name:      "new-cluster",
+			Namespace: "default",
+		}
+		Expect(apierrors.IsForbidden(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+
+	It("does not allow updates on queue type", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.Type = "classic"
+		Expect(apierrors.IsInvalid(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+
+	It("does not allow updates on durable", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.Durable = true
+		Expect(apierrors.IsInvalid(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+
+	It("does not allow updates on autoDelete", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.AutoDelete = false
+		Expect(apierrors.IsInvalid(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+})

config/crd/kustomization.yaml

@@ -12,9 +12,13 @@ resources:
 
 patchesStrategicMerge:
 - patches/webhook_in_bindings.yaml
+- patches/webhook_in_queues.yaml
+- patches/webhook_in_exchanges.yaml
 # +kubebuilder:scaffold:crdkustomizewebhookpatch
 
 - patches/cainjection_in_bindings.yaml
+- patches/cainjection_in_queues.yaml
+- patches/webhook_in_exchanges.yaml
 # +kubebuilder:scaffold:crdkustomizecainjectionpatch
 
 configurations:

config/crd/patches/cainjection_in_exchanges.yaml

@@ -1,6 +1,6 @@
 # The following patch adds a directive for certmanager to inject CA into the CRD
 # CRD conversion requires k8s 1.13 or later.
-apiVersion: apiextensions.k8s.io/v1alpha1
+apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:

config/crd/patches/cainjection_in_queues.yaml

@@ -1,6 +1,4 @@
-# The following patch adds a directive for certmanager to inject CA into the CRD
-# CRD conversion requires k8s 1.13 or later.
-apiVersion: apiextensions.k8s.io/v1alpha1
+apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations: