Repository: knative-sandbox/eventing-rabbitmq (go)

Issue:
Support RMQ Cluster configured with TLS using self-signed certificates
**Problem**
A user configures RMQ cluster with TLS using self-signed certificates and would like to have Knative Eventing securely connect to the RMQ Cluster. 
The two separate pieces would be
- RMQ Cluster -> Knative RMQ Source
- RMQ Cluster <-> Knative RMQ Broker

**[Persona:](https://github.com/knative/eventing/blob/main/docs/personas.md)**
Which persona is this feature for?
This feature enables Developers to comply with Administrator policies

**Exit Criteria**
- RMQ Cluster is configured with TLS using self signed certificates
- An Eventing RMQ Broker is configured to connect to RMQ over TLS
- Send event to Broker's ingress and the event is delivered to a downstream consumer (with hops between Knative Eventing & RMQ being secured with TLS)

Similar steps for RMQ Source

**Time Estimate (optional):**
How many developer-days do you think this may take to resolve?

**Additional context (optional)**
This blocked by https://github.com/knative-sandbox/eventing-rabbitmq/issues/566


Pull request description:
Adds support for eventing-rabbitmq->RMQ SSL communication with self-signed certs
# Changes

- 🎁 Broker -> RabbitMQ TLS communication with self-signed certs
- Requires `RabbitMQCluster.Spec.TLS.CASecretName` to be set to a secret containing the ca.cert
- Requires `caSecretName` as an extra field when using a `connectionSecret` instead of RabbitMQCluster

/kind enhancement


Fixes #883 

**Release Note**

```release-note
- Eventing-rabbitmq can be used with a RabbitMQ server which uses self-signed certs
```

Docs and e2e tests in a follow-up PR

Patch:
diff --git a/pkg/rabbit/service.go b/pkg/rabbit/service.go
index f1f92f13..f3c1345f 100644
--- a/pkg/rabbit/service.go
+++ b/pkg/rabbit/service.go
@@ -45,6 +45,8 @@ import (
 	"knative.dev/pkg/network"
 )
 
+const CA_SECRET_KEYNAME = "caSecretName"
+
 func New(ctx context.Context) *Rabbit {
 	return &Rabbit{
 		Interface:     rabbitmqclient.Get(ctx),
@@ -283,33 +285,13 @@ func (r *Rabbit) RabbitMQURL(ctx context.Context, clusterRef *rabbitv1beta1.Rabb
 		return url.Parse(fmt.Sprintf("%s://%s:%s@%s:%s", protocol, username, password, splittedUri[0], port))
 	}
 
-	// TODO: make this better.
-	ref := &duckv1.KReference{
-		Kind:       "RabbitmqCluster",
-		APIVersion: "rabbitmq.com/v1beta1",
-		Name:       clusterRef.Name,
-		Namespace:  clusterRef.Namespace,
-	}
-	gv, err := schema.ParseGroupVersion(ref.APIVersion)
-	if err != nil {
-		return nil, err
-	}
-
-	gvk := gv.WithKind(ref.Kind)
-	gvr, _ := meta.UnsafeGuessKindToResource(gvk)
-	_, lister, err := r.rabbitLister.Get(ctx, gvr)
-	if err != nil {
-		return nil, err
-	}
-
-	o, err := lister.ByNamespace(ref.Namespace).Get(ref.Name)
+	rab, err := r.getClusterFromReference(ctx, clusterRef)
 	if err != nil {
 		return nil, err
 	}
 
-	rab := o.(*duckv1beta1.Rabbit)
 	if rab.Status.DefaultUser == nil || rab.Status.DefaultUser.SecretReference == nil || rab.Status.DefaultUser.ServiceReference == nil {
-		return nil, fmt.Errorf("rabbit \"%s/%s\" not ready", ref.Namespace, ref.Name)
+		return nil, fmt.Errorf("rabbit \"%s/%s\" not ready", rab.Namespace, rab.Name)
 	}
 
 	_ = rab.Status.DefaultUser.SecretReference
@@ -336,3 +318,53 @@ func (r *Rabbit) RabbitMQURL(ctx context.Context, clusterRef *rabbitv1beta1.Rabb
 	host := network.GetServiceHostname(rab.Status.DefaultUser.ServiceReference.Name, rab.Status.DefaultUser.ServiceReference.Namespace)
 	return url.Parse(fmt.Sprintf("%s://%s:%s@%s:%s", protocol, username, password, host, port))
 }
+
+func (r *Rabbit) GetRabbitMQCASecret(ctx context.Context, clusterRef *rabbitv1beta1.RabbitmqClusterReference) (string, error) {
+	if clusterRef == nil {
+		return "", errors.New("GetRabbitMQCASecret: nil clusterReference")
+	}
+	if clusterRef.ConnectionSecret != nil {
+		s, err := r.kubeClientSet.CoreV1().Secrets(clusterRef.Namespace).Get(ctx, clusterRef.ConnectionSecret.Name, metav1.GetOptions{})
+		if err != nil {
+			return "", err
+		}
+		return string(s.Data[CA_SECRET_KEYNAME]), nil
+	}
+
+	rabbitMQCluster, err := r.getClusterFromReference(ctx, clusterRef)
+	if err != nil {
+		return "", err
+	}
+	if rabbitMQCluster.Spec.TLS != nil {
+		return rabbitMQCluster.Spec.TLS.CASecretName, nil
+	}
+
+	return "", nil
+}
+
+func (r *Rabbit) getClusterFromReference(ctx context.Context, clusterRef *rabbitv1beta1.RabbitmqClusterReference) (*duckv1beta1.Rabbit, error) {
+	// TODO: make this better.
+	ref := &duckv1.KReference{
+		Kind:       "RabbitmqCluster",
+		APIVersion: "rabbitmq.com/v1beta1",
+		Name:       clusterRef.Name,
+		Namespace:  clusterRef.Namespace,
+	}
+	gv, err := schema.ParseGroupVersion(ref.APIVersion)
+	if err != nil {
+		return nil, err
+	}
+
+	gvk := gv.WithKind(ref.Kind)
+	gvr, _ := meta.UnsafeGuessKindToResource(gvk)
+	_, lister, err := r.rabbitLister.Get(ctx, gvr)
+	if err != nil {
+		return nil, err
+	}
+
+	o, err := lister.ByNamespace(ref.Namespace).Get(ref.Name)
+	if err != nil {
+		return nil, err
+	}
+	return o.(*duckv1beta1.Rabbit), nil
+}
diff --git a/pkg/rabbit/types.go b/pkg/rabbit/types.go
index d27e34ee..f05c2577 100644
--- a/pkg/rabbit/types.go
+++ b/pkg/rabbit/types.go
@@ -36,4 +36,5 @@ type Service interface {
 	ReconcileBrokerDLXPolicy(context.Context, *QueueArgs) (Result, error)
 	ReconcileDLQPolicy(context.Context, *QueueArgs) (Result, error)
 	DeleteResource(ctx context.Context, args *DeleteResourceArgs) error
+	GetRabbitMQCASecret(ctx context.Context, clusterRef *rabbitv1beta1.RabbitmqClusterReference) (string, error)
 }
diff --git a/pkg/reconciler/broker/broker.go b/pkg/reconciler/broker/broker.go
index 09b79a4f..78bc90de 100644
--- a/pkg/reconciler/broker/broker.go
+++ b/pkg/reconciler/broker/broker.go
@@ -183,12 +183,22 @@ func (r *Reconciler) reconcileService(ctx context.Context, svc *corev1.Service)
 
 // reconcileIngressDeploymentCRD reconciles the Ingress Deployment.
 func (r *Reconciler) reconcileIngressDeployment(ctx context.Context, b *eventingv1.Broker) error {
+	clusterRef, err := r.brokerConfig.GetRabbitMQClusterRef(ctx, b)
+	if err != nil {
+		return err
+	}
+	secretName, err := r.rabbit.GetRabbitMQCASecret(ctx, clusterRef)
+	if err != nil {
+		return err
+	}
+
 	expected := resources.MakeIngressDeployment(&resources.IngressArgs{
-		Broker:             b,
-		Image:              r.ingressImage,
-		RabbitMQSecretName: rabbit.SecretName(b.Name, "broker"),
-		BrokerUrlSecretKey: rabbit.BrokerURLSecretKey,
-		Configs:            r.configs,
+		Broker:               b,
+		Image:                r.ingressImage,
+		RabbitMQSecretName:   rabbit.SecretName(b.Name, "broker"),
+		RabbitMQCASecretName: secretName,
+		BrokerUrlSecretKey:   rabbit.BrokerURLSecretKey,
+		Configs:              r.configs,
 	})
 	return r.reconcileDeployment(ctx, expected)
 }
@@ -203,17 +213,26 @@ func (r *Reconciler) reconcileIngressService(ctx context.Context, b *eventingv1.
 func (r *Reconciler) reconcileDLXDispatcherDeployment(ctx context.Context, b *eventingv1.Broker, sub *apis.URL) error {
 	// If there's a sub, then reconcile the deployment as usual.
 	if sub != nil {
+		clusterRef, err := r.brokerConfig.GetRabbitMQClusterRef(ctx, b)
+		if err != nil {
+			return err
+		}
+		secretName, err := r.rabbit.GetRabbitMQCASecret(ctx, clusterRef)
+		if err != nil {
+			return err
+		}
 		expected := resources.MakeDispatcherDeployment(&resources.DispatcherArgs{
 			Broker: b,
 			Image:  r.dispatcherImage,
 			//ServiceAccountName string
-			Delivery:           b.Spec.Delivery,
-			RabbitMQSecretName: rabbit.SecretName(b.Name, "broker"),
-			QueueName:          naming.CreateBrokerDeadLetterQueueName(b),
-			BrokerUrlSecretKey: rabbit.BrokerURLSecretKey,
-			Subscriber:         sub,
-			BrokerIngressURL:   b.Status.Address.URL,
-			Configs:            r.configs,
+			Delivery:             b.Spec.Delivery,
+			RabbitMQSecretName:   rabbit.SecretName(b.Name, "broker"),
+			RabbitMQCASecretName: secretName,
+			QueueName:            naming.CreateBrokerDeadLetterQueueName(b),
+			BrokerUrlSecretKey:   rabbit.BrokerURLSecretKey,
+			Subscriber:           sub,
+			BrokerIngressURL:     b.Status.Address.URL,
+			Configs:              r.configs,
 		})
 		return r.reconcileDeployment(ctx, expected)
 	}
diff --git a/pkg/reconciler/broker/resources/dispatcher.go b/pkg/reconciler/broker/resources/dispatcher.go
index 6ce6680f..fa83ff6b 100644
--- a/pkg/reconciler/broker/resources/dispatcher.go
+++ b/pkg/reconciler/broker/resources/dispatcher.go
@@ -46,13 +46,14 @@ type DispatcherArgs struct {
 	Broker   *eventingv1.Broker
 	Image    string
 	//ServiceAccountName string
-	RabbitMQHost       string
-	RabbitMQSecretName string
-	QueueName          string
-	BrokerUrlSecretKey string
-	BrokerIngressURL   *apis.URL
-	Subscriber         *apis.URL
-	Configs            reconcilersource.ConfigAccessor
+	RabbitMQHost         string
+	RabbitMQSecretName   string
+	RabbitMQCASecretName string
+	QueueName            string
+	BrokerUrlSecretKey   string
+	BrokerIngressURL     *apis.URL
+	Subscriber           *apis.URL
+	Configs              reconcilersource.ConfigAccessor
 }
 
 func DispatcherName(brokerName string) string {
@@ -132,7 +133,7 @@ func MakeDispatcherDeployment(args *DispatcherArgs) *appsv1.Deployment {
 				})
 		}
 	}
-	return &appsv1.Deployment{
+	deployment := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: args.Broker.Namespace,
 			Name:      DispatcherName(args.Broker.Name),
@@ -175,6 +176,25 @@ func MakeDispatcherDeployment(args *DispatcherArgs) *appsv1.Deployment {
 			},
 		},
 	}
+
+	if args.RabbitMQCASecretName != "" {
+		deployment.Spec.Template.Spec.Volumes = []corev1.Volume{{
+			Name: "rabbitmq-ca",
+			VolumeSource: corev1.VolumeSource{
+				Secret: &corev1.SecretVolumeSource{
+					SecretName: args.RabbitMQCASecretName,
+				},
+			},
+		}}
+
+		deployment.Spec.Template.Spec.Containers[0].VolumeMounts = []corev1.VolumeMount{
+			{
+				MountPath: "/etc/ssl/certs/",
+				Name:      "rabbitmq-ca",
+			}}
+	}
+
+	return deployment
 }
 
 // DispatcherLabels generates the labels present on all resources representing the dispatcher of the given
diff --git a/pkg/reconciler/broker/resources/ingress.go b/pkg/reconciler/broker/resources/ingress.go
index 474bf26b..78700254 100644
--- a/pkg/reconciler/broker/resources/ingress.go
+++ b/pkg/reconciler/broker/resources/ingress.go
@@ -42,9 +42,10 @@ type IngressArgs struct {
 	Broker *eventingv1.Broker
 	Image  string
 	//ServiceAccountName string
-	RabbitMQSecretName string
-	BrokerUrlSecretKey string
-	Configs            reconcilersource.ConfigAccessor
+	RabbitMQSecretName   string
+	RabbitMQCASecretName string
+	BrokerUrlSecretKey   string
+	Configs              reconcilersource.ConfigAccessor
 }
 
 // MakeIngressDeployment creates the in-memory representation of the Broker's ingress Deployment.
@@ -84,7 +85,7 @@ func MakeIngressDeployment(args *IngressArgs) *appsv1.Deployment {
 		envs = append(envs, args.Configs.ToEnvVars()...)
 	}
 
-	return &appsv1.Deployment{
+	deployment := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: args.Broker.Namespace,
 			Name:      DeploymentName,
@@ -136,6 +137,25 @@ func MakeIngressDeployment(args *IngressArgs) *appsv1.Deployment {
 			},
 		},
 	}
+
+	if args.RabbitMQCASecretName != "" {
+		deployment.Spec.Template.Spec.Volumes = []corev1.Volume{{
+			Name: "rabbitmq-ca",
+			VolumeSource: corev1.VolumeSource{
+				Secret: &corev1.SecretVolumeSource{
+					SecretName: args.RabbitMQCASecretName,
+				},
+			},
+		}}
+
+		deployment.Spec.Template.Spec.Containers[0].VolumeMounts = []corev1.VolumeMount{
+			{
+				MountPath: "/etc/ssl/certs/",
+				Name:      "rabbitmq-ca",
+			}}
+	}
+
+	return deployment
 }
 
 // MakeIngressService creates the in-memory representation of the Broker's ingress Service.
diff --git a/pkg/reconciler/source/rabbitmqsource.go b/pkg/reconciler/source/rabbitmqsource.go
index e00fbe82..fd937d58 100644
--- a/pkg/reconciler/source/rabbitmqsource.go
+++ b/pkg/reconciler/source/rabbitmqsource.go
@@ -212,15 +212,21 @@ func (r *Reconciler) createReceiveAdapter(ctx context.Context, src *v1alpha1.Rab
 		logging.FromContext(ctx).Error("error while converting metrics config to JSON", zap.Any("receiveAdapter", err))
 	}
 
+	secretName, err := r.rabbit.GetRabbitMQCASecret(ctx, src.Spec.RabbitmqClusterReference)
+	if err != nil {
+		return nil, err
+	}
+
 	raArgs := resources.ReceiveAdapterArgs{
-		Image:              r.receiveAdapterImage,
-		Source:             src,
-		Labels:             resources.GetLabels(src.Name),
-		SinkURI:            sinkURI.String(),
-		MetricsConfig:      metricsConfig,
-		LoggingConfig:      loggingConfig,
-		RabbitMQSecretName: rabbit.SecretName(src.Name, "source"),
-		BrokerUrlSecretKey: rabbit.BrokerURLSecretKey,
+		Image:                r.receiveAdapterImage,
+		Source:               src,
+		Labels:               resources.GetLabels(src.Name),
+		SinkURI:              sinkURI.String(),
+		MetricsConfig:        metricsConfig,
+		LoggingConfig:        loggingConfig,
+		RabbitMQSecretName:   rabbit.SecretName(src.Name, "source"),
+		RabbitMQCASecretName: secretName,
+		BrokerUrlSecretKey:   rabbit.BrokerURLSecretKey,
 	}
 	expected := resources.MakeReceiveAdapter(&raArgs)
 
diff --git a/pkg/reconciler/source/resources/receive_adapter.go b/pkg/reconciler/source/resources/receive_adapter.go
index 130af144..9fc81380 100644
--- a/pkg/reconciler/source/resources/receive_adapter.go
+++ b/pkg/reconciler/source/resources/receive_adapter.go
@@ -31,14 +31,15 @@ import (
 )
 
 type ReceiveAdapterArgs struct {
-	Image              string
-	Source             *v1alpha1.RabbitmqSource
-	Labels             map[string]string
-	SinkURI            string
-	MetricsConfig      string
-	LoggingConfig      string
-	RabbitMQSecretName string
-	BrokerUrlSecretKey string
+	Image                string
+	Source               *v1alpha1.RabbitmqSource
+	Labels               map[string]string
+	SinkURI              string
+	MetricsConfig        string
+	LoggingConfig        string
+	RabbitMQSecretName   string
+	RabbitMQCASecretName string
+	BrokerUrlSecretKey   string
 }
 
 func MakeReceiveAdapter(args *ReceiveAdapterArgs) *v1.Deployment {
@@ -129,7 +130,7 @@ func MakeReceiveAdapter(args *ReceiveAdapterArgs) *v1.Deployment {
 		}
 	}
 
-	return &v1.Deployment{
+	deployment := &v1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:         kmeta.ChildName(fmt.Sprintf("rabbitmqsource-%s-", args.Source.Name), string(args.Source.UID)),
 			Namespace:    args.Source.Namespace,
@@ -175,4 +176,22 @@ func MakeReceiveAdapter(args *ReceiveAdapterArgs) *v1.Deployment {
 			},
 		},
 	}
+	if args.RabbitMQCASecretName != "" {
+		deployment.Spec.Template.Spec.Volumes = []corev1.Volume{{
+			Name: "rabbitmq-ca",
+			VolumeSource: corev1.VolumeSource{
+				Secret: &corev1.SecretVolumeSource{
+					SecretName: args.RabbitMQCASecretName,
+				},
+			},
+		}}
+
+		deployment.Spec.Template.Spec.Containers[0].VolumeMounts = []corev1.VolumeMount{
+			{
+				MountPath: "/etc/ssl/certs/",
+				Name:      "rabbitmq-ca",
+			}}
+	}
+
+	return deployment
 }
diff --git a/pkg/reconciler/trigger/resources/dispatcher.go b/pkg/reconciler/trigger/resources/dispatcher.go
index aa67440e..d965a68b 100644
--- a/pkg/reconciler/trigger/resources/dispatcher.go
+++ b/pkg/reconciler/trigger/resources/dispatcher.go
@@ -47,14 +47,15 @@ type DispatcherArgs struct {
 	Trigger  *eventingv1.Trigger
 	Image    string
 	//ServiceAccountName string
-	RabbitMQHost       string
-	RabbitMQSecretName string
-	QueueName          string
-	BrokerUrlSecretKey string
-	BrokerIngressURL   *apis.URL
-	Subscriber         *apis.URL
-	DLX                bool
-	Configs            reconcilersource.ConfigAccessor
+	RabbitMQHost         string
+	RabbitMQSecretName   string
+	RabbitMQCASecretName string
+	QueueName            string
+	BrokerUrlSecretKey   string
+	BrokerIngressURL     *apis.URL
+	Subscriber           *apis.URL
+	DLX                  bool
+	Configs              reconcilersource.ConfigAccessor
 }
 
 // MakeDispatcherDeployment creates the in-memory representation of the Broker's Dispatcher Deployment.
@@ -167,7 +168,7 @@ func MakeDispatcherDeployment(args *DispatcherArgs) *appsv1.Deployment {
 				Value: "1",
 			})
 	}
-	return &appsv1.Deployment{
+	deployment := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: args.Trigger.Namespace,
 			Name:      name,
@@ -192,6 +193,23 @@ func MakeDispatcherDeployment(args *DispatcherArgs) *appsv1.Deployment {
 			},
 		},
 	}
+	if args.RabbitMQCASecretName != "" {
+		deployment.Spec.Template.Spec.Volumes = []corev1.Volume{{
+			Name: "rabbitmq-ca",
+			VolumeSource: corev1.VolumeSource{
+				Secret: &corev1.SecretVolumeSource{
+					SecretName: args.RabbitMQCASecretName,
+				},
+			},
+		}}
+
+		deployment.Spec.Template.Spec.Containers[0].VolumeMounts = []corev1.VolumeMount{
+			{
+				MountPath: "/etc/ssl/certs/",
+				Name:      "rabbitmq-ca",
+			}}
+	}
+	return deployment
 }
 
 // DispatcherLabels generates the labels present on all resources representing the dispatcher of the given
diff --git a/pkg/reconciler/trigger/trigger.go b/pkg/reconciler/trigger/trigger.go
index 6682d553..07c44a96 100644
--- a/pkg/reconciler/trigger/trigger.go
+++ b/pkg/reconciler/trigger/trigger.go
@@ -389,17 +389,27 @@ func (r *Reconciler) reconcileDispatcherDeployment(ctx context.Context, t *event
 		queueName = naming.CreateTriggerDeadLetterQueueName(t)
 	}
 
+	clusterRef, err := r.brokerConfig.GetRabbitMQClusterRef(ctx, b)
+	if err != nil {
+		return nil, err
+	}
+	secretName, err := r.rabbit.GetRabbitMQCASecret(ctx, clusterRef)
+	if err != nil {
+		return nil, err
+	}
+
 	expected := resources.MakeDispatcherDeployment(&resources.DispatcherArgs{
-		Trigger:            t,
-		Image:              r.dispatcherImage,
-		RabbitMQSecretName: rabbitmqSecret.Name,
-		QueueName:          queueName,
-		BrokerUrlSecretKey: rabbit.BrokerURLSecretKey,
-		BrokerIngressURL:   b.Status.Address.URL,
-		Subscriber:         sub,
-		DLX:                dlq,
-		Delivery:           delivery,
-		Configs:            r.configs,
+		Trigger:              t,
+		Image:                r.dispatcherImage,
+		RabbitMQSecretName:   rabbitmqSecret.Name,
+		RabbitMQCASecretName: secretName,
+		QueueName:            queueName,
+		BrokerUrlSecretKey:   rabbit.BrokerURLSecretKey,
+		BrokerIngressURL:     b.Status.Address.URL,
+		Subscriber:           sub,
+		DLX:                  dlq,
+		Delivery:             delivery,
+		Configs:              r.configs,
 	})
 	return r.reconcileDeployment(ctx, expected)
 }


Test patch:
diff --git a/pkg/rabbit/service_test.go b/pkg/rabbit/service_test.go
index c07ec402..e20703a2 100644
--- a/pkg/rabbit/service_test.go
+++ b/pkg/rabbit/service_test.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/rest"
 
+	rabbitclusterv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
 	rabbitduck "knative.dev/eventing-rabbitmq/pkg/client/injection/ducks/duck/v1beta1/rabbit"
 	"knative.dev/eventing-rabbitmq/third_party/pkg/apis/rabbitmq.com/v1beta1"
 	fakerabbitclient "knative.dev/eventing-rabbitmq/third_party/pkg/client/injection/client/fake"
@@ -155,6 +156,98 @@ func Test_RabbitMQURL(t *testing.T) {
 	}
 }
 
+func TestGetRabbitMQCASecret(t *testing.T) {
+	for _, tt := range []struct {
+		name, wantSecret string
+		clusterRef       *v1beta1.RabbitmqClusterReference
+		secret           *v1.Secret
+		wantErr          bool
+	}{{
+		name: "errors when connection secret not available",
+		clusterRef: &v1beta1.RabbitmqClusterReference{
+			ConnectionSecret: &v1.LocalObjectReference{Name: "secret"},
+		},
+		wantErr: true,
+	}, {
+		name:       "errors for nil clusterRef",
+		clusterRef: nil,
+		wantErr:    true,
+	}, {
+		name: "no error, empty secretName when secret doesn't have correct field set",
+		clusterRef: &v1beta1.RabbitmqClusterReference{
+			Namespace: "default",
+			ConnectionSecret: &v1.LocalObjectReference{
+				Name: "secret-name",
+			},
+		},
+		secret: &v1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "secret-name",
+				Namespace: "default",
+			},
+		},
+		wantErr: false,
+	}, {
+		name: "no error, correct secret name when clusterRef is a secret",
+		clusterRef: &v1beta1.RabbitmqClusterReference{
+			Namespace: "default",
+			ConnectionSecret: &v1.LocalObjectReference{
+				Name: "secret-name",
+			},
+		},
+		secret: &v1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "secret-name",
+				Namespace: "default",
+			},
+			Data: map[string][]byte{
+				"caSecretName": []byte("some-secret-name"),
+			},
+		},
+		wantErr:    false,
+		wantSecret: "some-secret-name",
+	}, {
+		name: "errors if RabbitMQCluster not found",
+		clusterRef: &v1beta1.RabbitmqClusterReference{
+			Namespace: "default",
+			Name:      "some-cluster",
+		},
+		wantErr: true,
+	}} {
+		// TODO: add more RabbitmqCluster tests
+		t.Run(tt.name, func(t *testing.T) {
+			tt := tt
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			ctx, _ = injection.Fake.SetupInformers(ctx, &rest.Config{})
+			scheme := runtime.NewScheme()
+			rabbitclusterv1beta1.AddToScheme(scheme)
+			ctx, _ = fakedynamicclient.With(ctx, scheme)
+			ctx = rabbitduck.WithDuck(ctx)
+			ctx, fakekubeClientSet := fakekubeclient.With(ctx)
+			i := fakerabbitclient.Get(ctx)
+
+			if tt.secret != nil {
+				_, err := fakekubeClientSet.CoreV1().Secrets(tt.clusterRef.Namespace).Create(ctx, tt.secret, metav1.CreateOptions{})
+				if err != nil {
+					t.Errorf("failed to create secret: %s", err)
+				}
+			}
+
+			r := &Rabbit{Interface: i, rabbitLister: rabbitduck.Get(ctx), kubeClientSet: fakekubeClientSet}
+
+			gotSecretName, err := r.GetRabbitMQCASecret(ctx, tt.clusterRef)
+			if (err != nil && !tt.wantErr) || (err == nil && tt.wantErr) {
+				t.Errorf("unexpected error checking conditions want: %v, got: %v", tt.wantErr, err)
+			}
+			if tt.wantSecret != gotSecretName {
+				t.Errorf("unexpected secretname: want: %s, got: %s", tt.wantSecret, gotSecretName)
+			}
+		})
+	}
+}
+
 func CreateBroker(name, namespace string) *eventingv1.Broker {
 	b := &eventingv1.Broker{
 		ObjectMeta: metav1.ObjectMeta{
diff --git a/pkg/reconciler/broker/resources/dispatcher_test.go b/pkg/reconciler/broker/resources/dispatcher_test.go
index 114b8aab..a89ef11a 100644
--- a/pkg/reconciler/broker/resources/dispatcher_test.go
+++ b/pkg/reconciler/broker/resources/dispatcher_test.go
@@ -54,14 +54,15 @@ func TestMakeDispatcherDeployment(t *testing.T) {
 	}
 	linear := v1.BackoffPolicyLinear
 	args := &DispatcherArgs{
-		Broker:             broker,
-		Image:              image,
-		RabbitMQHost:       rabbitHost,
-		RabbitMQSecretName: secretName,
-		QueueName:          queueName,
-		BrokerUrlSecretKey: brokerURLKey,
-		Subscriber:         sURL,
-		BrokerIngressURL:   bURL,
+		Broker:               broker,
+		Image:                image,
+		RabbitMQHost:         rabbitHost,
+		RabbitMQSecretName:   secretName,
+		RabbitMQCASecretName: "rabbitmq-ca-secret",
+		QueueName:            queueName,
+		BrokerUrlSecretKey:   brokerURLKey,
+		Subscriber:           sURL,
+		BrokerIngressURL:     bURL,
 		Delivery: &v1.DeliverySpec{
 			Retry:         ptr.Int32(10),
 			BackoffDelay:  ptr.String("PT20S"),
@@ -103,6 +104,14 @@ func TestMakeDispatcherDeployment(t *testing.T) {
 					},
 				},
 				Spec: corev1.PodSpec{
+					Volumes: []corev1.Volume{{
+						Name: "rabbitmq-ca",
+						VolumeSource: corev1.VolumeSource{
+							Secret: &corev1.SecretVolumeSource{
+								SecretName: "rabbitmq-ca-secret",
+							},
+						},
+					}},
 					Containers: []corev1.Container{{
 						Name:  "dispatcher",
 						Image: image,
@@ -114,6 +123,11 @@ func TestMakeDispatcherDeployment(t *testing.T) {
 								corev1.ResourceCPU:    resource.MustParse("4000m"),
 								corev1.ResourceMemory: resource.MustParse("600Mi")},
 						},
+						VolumeMounts: []corev1.VolumeMount{
+							{
+								MountPath: "/etc/ssl/certs/",
+								Name:      "rabbitmq-ca",
+							}},
 						Env: []corev1.EnvVar{{
 							Name:  system.NamespaceEnvKey,
 							Value: system.Namespace(),
diff --git a/pkg/reconciler/broker/resources/ingress_test.go b/pkg/reconciler/broker/resources/ingress_test.go
index c96c85d8..fa02dd4f 100644
--- a/pkg/reconciler/broker/resources/ingress_test.go
+++ b/pkg/reconciler/broker/resources/ingress_test.go
@@ -41,10 +41,11 @@ const (
 func TestMakeIngressDeployment(t *testing.T) {
 	var TrueValue = true
 	args := &IngressArgs{
-		Broker:             &eventingv1.Broker{ObjectMeta: metav1.ObjectMeta{Name: brokerName, Namespace: ns, UID: brokerUID}},
-		Image:              image,
-		RabbitMQSecretName: secretName,
-		BrokerUrlSecretKey: brokerURLKey,
+		Broker:               &eventingv1.Broker{ObjectMeta: metav1.ObjectMeta{Name: brokerName, Namespace: ns, UID: brokerUID}},
+		Image:                image,
+		RabbitMQSecretName:   secretName,
+		RabbitMQCASecretName: "rabbitmq-ca-secret",
+		BrokerUrlSecretKey:   brokerURLKey,
 	}
 
 	got := MakeIngressDeployment(args)
@@ -80,6 +81,14 @@ func TestMakeIngressDeployment(t *testing.T) {
 					},
 				},
 				Spec: corev1.PodSpec{
+					Volumes: []corev1.Volume{{
+						Name: "rabbitmq-ca",
+						VolumeSource: corev1.VolumeSource{
+							Secret: &corev1.SecretVolumeSource{
+								SecretName: "rabbitmq-ca-secret",
+							},
+						},
+					}},
 					Containers: []corev1.Container{{
 						Image: image,
 						Name:  "ingress",
@@ -91,6 +100,11 @@ func TestMakeIngressDeployment(t *testing.T) {
 								corev1.ResourceCPU:    resource.MustParse("1000m"),
 								corev1.ResourceMemory: resource.MustParse("400Mi")},
 						},
+						VolumeMounts: []corev1.VolumeMount{
+							{
+								MountPath: "/etc/ssl/certs/",
+								Name:      "rabbitmq-ca",
+							}},
 						Env: []corev1.EnvVar{{
 							Name:  system.NamespaceEnvKey,
 							Value: system.Namespace(),
diff --git a/pkg/reconciler/source/resources/receive_adapter_test.go b/pkg/reconciler/source/resources/receive_adapter_test.go
index 7fd5be82..5c0f8b03 100644
--- a/pkg/reconciler/source/resources/receive_adapter_test.go
+++ b/pkg/reconciler/source/resources/receive_adapter_test.go
@@ -81,9 +81,10 @@ func TestMakeReceiveAdapter(t *testing.T) {
 					"test-key1": "test-value1",
 					"test-key2": "test-value2",
 				},
-				SinkURI:            "sink-uri",
-				RabbitMQSecretName: secretName,
-				BrokerUrlSecretKey: brokerURLKey,
+				SinkURI:              "sink-uri",
+				RabbitMQSecretName:   secretName,
+				RabbitMQCASecretName: "rabbitmq-ca-secret",
+				BrokerUrlSecretKey:   brokerURLKey,
 			})
 
 			boPolicy := tt.backoffPolicy
@@ -131,6 +132,14 @@ func TestMakeReceiveAdapter(t *testing.T) {
 						},
 						Spec: corev1.PodSpec{
 							ServiceAccountName: "source-svc-acct",
+							Volumes: []corev1.Volume{{
+								Name: "rabbitmq-ca",
+								VolumeSource: corev1.VolumeSource{
+									Secret: &corev1.SecretVolumeSource{
+										SecretName: "rabbitmq-ca-secret",
+									},
+								},
+							}},
 							Containers: []corev1.Container{
 								{
 									Name:            "receive-adapter",
@@ -144,6 +153,11 @@ func TestMakeReceiveAdapter(t *testing.T) {
 											corev1.ResourceCPU:    resource.MustParse("4000m"),
 											corev1.ResourceMemory: resource.MustParse("600Mi")},
 									},
+									VolumeMounts: []corev1.VolumeMount{
+										{
+											MountPath: "/etc/ssl/certs/",
+											Name:      "rabbitmq-ca",
+										}},
 									Env: []corev1.EnvVar{
 										{
 											Name: "RABBIT_URL",
diff --git a/pkg/reconciler/trigger/resources/dispatcher_test.go b/pkg/reconciler/trigger/resources/dispatcher_test.go
index 25e90ad7..5620c6f5 100644
--- a/pkg/reconciler/trigger/resources/dispatcher_test.go
+++ b/pkg/reconciler/trigger/resources/dispatcher_test.go
@@ -135,6 +135,14 @@ func deployment(opts ...func(*appsv1.Deployment)) *appsv1.Deployment {
 					},
 				},
 				Spec: corev1.PodSpec{
+					Volumes: []corev1.Volume{{
+						Name: "rabbitmq-ca",
+						VolumeSource: corev1.VolumeSource{
+							Secret: &corev1.SecretVolumeSource{
+								SecretName: "rabbitmq-ca-secret",
+							},
+						},
+					}},
 					Containers: []corev1.Container{{
 						Name:  "dispatcher",
 						Image: image,
@@ -146,6 +154,11 @@ func deployment(opts ...func(*appsv1.Deployment)) *appsv1.Deployment {
 								corev1.ResourceCPU:    resource.MustParse("4000m"),
 								corev1.ResourceMemory: resource.MustParse("600Mi")},
 						},
+						VolumeMounts: []corev1.VolumeMount{
+							{
+								MountPath: "/etc/ssl/certs/",
+								Name:      "rabbitmq-ca",
+							}},
 						Env: []corev1.EnvVar{{
 							Name:  system.NamespaceEnvKey,
 							Value: system.Namespace(),
@@ -223,14 +236,15 @@ func dispatcherArgs(opts ...func(*DispatcherArgs)) *DispatcherArgs {
 		Spec:       eventingv1.TriggerSpec{Broker: brokerName},
 	}
 	args := &DispatcherArgs{
-		Trigger:            trigger,
-		Image:              image,
-		RabbitMQHost:       rabbitHost,
-		RabbitMQSecretName: secretName,
-		QueueName:          queueName,
-		BrokerUrlSecretKey: brokerURLKey,
-		BrokerIngressURL:   ingressURL,
-		Subscriber:         sURL,
+		Trigger:              trigger,
+		Image:                image,
+		RabbitMQHost:         rabbitHost,
+		RabbitMQSecretName:   secretName,
+		RabbitMQCASecretName: "rabbitmq-ca-secret",
+		QueueName:            queueName,
+		BrokerUrlSecretKey:   brokerURLKey,
+		BrokerIngressURL:     ingressURL,
+		Subscriber:           sURL,
 	}
 	for _, o := range opts {
 		o(args)
diff --git a/pkg/reconciler/trigger/trigger_test.go b/pkg/reconciler/trigger/trigger_test.go
index a4ce3fe0..fd8d1c1c 100644
--- a/pkg/reconciler/trigger/trigger_test.go
+++ b/pkg/reconciler/trigger/trigger_test.go
@@ -257,6 +257,7 @@ func TestReconcile(t *testing.T) {
 					triggerWithFilter(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 					markReady(createQueue(config, false)),
 					markReady(createBinding(true, false)),
 				},
@@ -274,6 +275,7 @@ func TestReconcile(t *testing.T) {
 					triggerWithFilter(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 					markReady(createQueue(config, false)),
 					markReady(createBinding(true, false)),
 				},
@@ -292,6 +294,7 @@ func TestReconcile(t *testing.T) {
 					triggerWithDeliverySpec(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 					markReady(createQueue(config, false)),
 					markReady(createBinding(true, false)),
 				},
@@ -386,6 +389,7 @@ func TestReconcile(t *testing.T) {
 					triggerWithFilter(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 					markReady(createQueue(config, false)),
 					markReady(createBinding(true, false)),
 				},
@@ -408,6 +412,7 @@ func TestReconcile(t *testing.T) {
 						WithTriggerSubscriberRef(subscriberGVK, subscriberName, testNS)),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 				},
 				WantCreates: []runtime.Object{
 					createDispatcherDeployment(false),
@@ -461,6 +466,7 @@ func TestReconcile(t *testing.T) {
 						WithTriggerSubscriberURI(subscriberURI)),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 					markReady(createQueue(config, false)),
 					markReady(createBinding(false, false)),
 				},
@@ -496,6 +502,7 @@ func TestReconcile(t *testing.T) {
 						WithTriggerSubscriberURI(subscriberURI)),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 					createDifferentDispatcherDeployment(),
 					markReady(createQueue(config, false)),
 					markReady(createBinding(false, false)),
@@ -532,6 +539,7 @@ func TestReconcile(t *testing.T) {
 						WithTriggerSubscriberURI(subscriberURI)),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 					markReady(createQueue(config, false)),
 					markReady(createBinding(false, false)),
 					createDispatcherDeployment(false),
@@ -556,6 +564,7 @@ func TestReconcile(t *testing.T) {
 					triggerWithFilter(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 					markReady(createQueue(config, false)),
 					markReady(createBinding(true, false)),
 					createDispatcherDeployment(false),
@@ -722,6 +731,7 @@ func TestReconcile(t *testing.T) {
 					),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 					markReady(createQueue(config, false)),
 					markReady(createBinding(false, false)),
 				},
@@ -754,6 +764,7 @@ func TestReconcile(t *testing.T) {
 						WithTriggerSubscriberURI(subscriberURI)),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 					markReady(createQueue(config, false)),
 					markReady(createBinding(false, false)),
 					createDispatcherDeploymentWithParallelism(),
@@ -782,6 +793,7 @@ func TestReconcile(t *testing.T) {
 					triggerWithFilter(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
+					createRabbitMQCluster(),
 					markReady(createQueue(config, true)),
 					markReady(createBinding(true, false)),
 					markReady(createQueue(config, false)),
@@ -1219,6 +1231,20 @@ func makeSubscriberNotAddressableAsUnstructured() *unstructured.Unstructured {
 	}
 }
 
+func createRabbitMQCluster() *unstructured.Unstructured {
+	return &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"apiVersion": "rabbitmq.com/v1beta1",
+			"kind":       "RabbitmqCluster",
+			"metadata": map[string]interface{}{
+				"creationTimestamp": nil,
+				"namespace":         testNS,
+				"name":              rabbitMQBrokerName,
+			},
+		},
+	}
+}
+
 func createPolicy() *rabbitv1beta1.Policy {
 	t := triggerWithFilter()
 	labels := map[string]string{