Repository: knative-sandbox/eventing-kafka-broker (go)

Issue:
Eventing TLS: support exposing https address in Kafka Broker controller
**Problem**

This is similar to https://github.com/knative/eventing/issues/6918.

As the [Eventing TLS feature track](https://docs.google.com/document/d/1H-x_oji8LqkCyd7tlsSyclmUe7FAmEJPgRxOU_0pkn8/edit) describes we should support TLS for Broker.

In particular, kafka-controller should:
- when the feature flag (see https://github.com/knative/eventing/issues/6836) is set to disabled
  - no behavior change 
- when the feature flag (see https://github.com/knative/eventing/issues/6836) is set to permissive
  - expose both http and https endpoints in the new `status.addresses` (see https://github.com/knative/pkg/issues/2711) field
  - the existing `status.address` will continue to have the existing `http` endpoint
- when the feature flag (see https://github.com/knative/eventing/issues/6836) is set to strict
  - expose only the `https` endpoints in both the new `status.addresses` and the existing `status.address`

**Time Estimate (optional):** 
How many developer-days do you think this may take to resolve?
1

**Additional context (optional)**

- Depends on https://github.com/knative-sandbox/eventing-kafka-broker/issues/3107

/good-first-issue


Pull request:
Support exposing https address ekb controller
Fixes #3108 

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Update the broker reconciler to update the `Status.Address` and `Status.Addresses` as required
- Update the broker reconciler tests to account for the updates


<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
Added support for exposing an https address from the broker.
```



<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


Patch:
diff --git a/control-plane/pkg/receiver/address.go b/control-plane/pkg/receiver/address.go
index 3535ea61c..c6ff34fee 100644
--- a/control-plane/pkg/receiver/address.go
+++ b/control-plane/pkg/receiver/address.go
@@ -23,7 +23,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
 	"knative.dev/pkg/apis"
-	pkgduckv1 "knative.dev/pkg/apis/duck/v1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
 func Address(host string, object metav1.Object) *url.URL {
@@ -35,8 +35,8 @@ func Address(host string, object metav1.Object) *url.URL {
 }
 
 // HTTPAddress returns the addressable
-func HTTPAddress(host string, object metav1.Object) pkgduckv1.Addressable {
-	httpAddress := pkgduckv1.Addressable{
+func HTTPAddress(host string, object metav1.Object) duckv1.Addressable {
+	httpAddress := duckv1.Addressable{
 		Name: pointer.String("http"),
 		URL:  apis.HTTP(host),
 	}
@@ -44,8 +44,8 @@ func HTTPAddress(host string, object metav1.Object) pkgduckv1.Addressable {
 	return httpAddress
 }
 
-func HTTPSAddress(host string, object metav1.Object, caCerts string) pkgduckv1.Addressable {
-	httpsAddress := pkgduckv1.Addressable{
+func HTTPSAddress(host string, object metav1.Object, caCerts string) duckv1.Addressable {
+	httpsAddress := duckv1.Addressable{
 		Name:    pointer.String("https"),
 		URL:     apis.HTTPS(host),
 		CACerts: pointer.String(caCerts),
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index a6a2c145a..864eb7938 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -30,6 +30,8 @@ import (
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/util/retry"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/eventing/pkg/apis/feature"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/network"
 	"knative.dev/pkg/reconciler"
@@ -56,6 +58,12 @@ const (
 
 	// ConsumerConfigKey is the key for Kafka Broker consumer configurations
 	ConsumerConfigKey = "config-kafka-broker-consumer.properties"
+
+	// brokerIngressTLSSecretName is the TLS Secret Name for the Cert-Manager resource
+	brokerIngressTLSSecretName = "kafka-broker-ingress-server-tls"
+
+	// caCertsSecretKey is the name of the CA Cert in the secret
+	caCertsSecretKey = "ca.crt"
 )
 
 type Reconciler struct {
@@ -226,7 +234,35 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	}
 
 	ingressHost := network.GetServiceHostname(r.Env.IngressName, r.DataPlaneNamespace)
-	address := receiver.Address(ingressHost, broker)
+
+	transportEncryptionFlags := feature.FromContext(ctx)
+	var addressableStatus duckv1.AddressStatus
+	if transportEncryptionFlags.IsPermissiveTransportEncryption() {
+		caCerts, err := r.getCaCerts()
+		if err != nil {
+			return err
+		}
+
+		httpAddress := receiver.HTTPAddress(ingressHost, broker)
+		httpsAddress := receiver.HTTPSAddress(ingressHost, broker, caCerts)
+		addressableStatus.Address = &httpAddress
+		addressableStatus.Addresses = []duckv1.Addressable{httpAddress, httpsAddress}
+	} else if transportEncryptionFlags.IsStrictTransportEncryption() {
+		caCerts, err := r.getCaCerts()
+		if err != nil {
+			return err
+		}
+
+		httpsAddress := receiver.HTTPSAddress(ingressHost, broker, caCerts)
+		addressableStatus.Address = &httpsAddress
+		addressableStatus.Addresses = []duckv1.Addressable{httpsAddress}
+	} else {
+		httpAddress := receiver.HTTPAddress(ingressHost, broker)
+		addressableStatus.Address = &httpAddress
+		addressableStatus.Addresses = []duckv1.Addressable{httpAddress}
+	}
+
+	address := addressableStatus.Address.URL.URL()
 	proberAddressable := prober.Addressable{
 		Address: address,
 		ResourceKey: types.NamespacedName{
@@ -239,7 +275,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
-	statusConditionManager.Addressable(address)
+	statusConditionManager.ProbesStatusReady()
+
+	broker.Status.Address = addressableStatus.Address
+	broker.Status.Addresses = addressableStatus.Addresses
+	broker.GetConditionSet().Manage(broker.GetStatus()).MarkTrue(base.ConditionAddressable)
 
 	return nil
 }
@@ -641,3 +681,15 @@ func containsFinalizerSecret(secret *corev1.Secret, finalizer string) bool {
 func finalizerSecret(object metav1.Object) string {
 	return fmt.Sprintf("%s/%s", "kafka.eventing", object.GetUID())
 }
+
+func (r *Reconciler) getCaCerts() (string, error) {
+	secret, err := r.SecretLister.Secrets(r.SystemNamespace).Get(brokerIngressTLSSecretName)
+	if err != nil {
+		return "", fmt.Errorf("failed to get CA certs from %s/%s: %w", r.SystemNamespace, brokerIngressTLSSecretName, err)
+	}
+	caCerts, ok := secret.Data[caCertsSecretKey]
+	if !ok {
+		return "", fmt.Errorf("failed to get CA certs from %s/%s: missing %s key", r.SystemNamespace, brokerIngressTLSSecretName, caCertsSecretKey)
+	}
+	return string(caCerts), nil
+}
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index 4799dbc0b..aabdfccd1 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -119,6 +119,10 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 	reconciler.Tracker = impl.Tracker
 
 	secretinformer.Get(ctx).Informer().AddEventHandler(controller.HandleAll(reconciler.Tracker.OnChanged))
+	secretinformer.Get(ctx).Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: controller.FilterWithName(brokerIngressTLSSecretName),
+		Handler:    controller.HandleAll(globalResync),
+	})
 	configmapinformer.Get(ctx).Informer().AddEventHandler(controller.HandleAll(
 		// Call the tracker's OnChanged method, but we've seen the objects
 		// coming through this path missing TypeMeta, so ensure it is properly


Test patch:
diff --git a/control-plane/pkg/receiver/address_test.go b/control-plane/pkg/receiver/address_test.go
index 3154bfc1f..68e2b7586 100644
--- a/control-plane/pkg/receiver/address_test.go
+++ b/control-plane/pkg/receiver/address_test.go
@@ -23,6 +23,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	"knative.dev/eventing/pkg/eventingtls/eventingtlstesting"
 )
 
 func TestAddress(t *testing.T) {
@@ -40,3 +41,36 @@ func TestAddress(t *testing.T) {
 	require.Contains(t, url.Path, ks.GetNamespace())
 	require.Contains(t, url.Path, ks.GetName())
 }
+
+func TestHTTPSAddress(t *testing.T) {
+	host := "example.com"
+	ks := &eventing.KafkaSink{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "ns",
+			Name:      "ks",
+		},
+	}
+	httpsAddress := HTTPSAddress(host, ks, string(eventingtlstesting.CA))
+
+	require.Equal(t, httpsAddress.URL.Host, host)
+	require.Equal(t, httpsAddress.URL.Scheme, "https")
+	require.Contains(t, httpsAddress.URL.Path, ks.GetNamespace())
+	require.Contains(t, httpsAddress.URL.Path, ks.GetName())
+	require.Equal(t, *httpsAddress.CACerts, string(eventingtlstesting.CA))
+}
+
+func TestHTTPAddress(t *testing.T) {
+	host := "example.com"
+	ks := &eventing.KafkaSink{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "ns",
+			Name:      "ks",
+		},
+	}
+	httpAddress := HTTPAddress(host, ks)
+
+	require.Equal(t, host, httpAddress.URL.Host)
+	require.Equal(t, httpAddress.URL.Scheme, "http")
+	require.Contains(t, httpAddress.URL.Path, ks.GetNamespace())
+	require.Contains(t, httpAddress.URL.Path, ks.GetName())
+}
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 37a504140..a9cf8939d 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -43,10 +43,13 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	clientgotesting "k8s.io/client-go/testing"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/eventing/pkg/apis/feature"
+	"knative.dev/eventing/pkg/eventingtls/eventingtlstesting"
 	"knative.dev/pkg/apis"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
+	"knative.dev/pkg/network"
 	. "knative.dev/pkg/reconciler/testing"
 	"knative.dev/pkg/resolver"
 	"knative.dev/pkg/tracker"
@@ -75,6 +78,8 @@ const (
 	finalizerName = "brokers.eventing.knative.dev"
 
 	bootstrapServers = "kafka-1:9092,kafka-2:9093"
+
+	brokerIngressTLSSecretName = "kafka-broker-ingress-server-tls"
 )
 
 var (
@@ -84,6 +89,12 @@ var (
 		fmt.Sprintf(`Updated %q finalizers`, BrokerName),
 	)
 
+	brokerAddress = &apis.URL{
+		Scheme: "http",
+		Host:   network.GetServiceHostname(DefaultEnv.IngressName, DefaultEnv.SystemNamespace),
+		Path:   fmt.Sprintf("/%s/%s", BrokerNamespace, BrokerName),
+	}
+
 	createTopicError = fmt.Errorf("failed to create topic")
 	deleteTopicError = fmt.Errorf("failed to delete topic")
 
@@ -176,6 +187,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerProbeSucceeded,
 						BrokerConfigMapAnnotations(),
 						WithTopicStatusAnnotation(BrokerTopic()),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -239,6 +261,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						StatusBrokerProbeSucceeded,
 						BrokerConfigMapAnnotations(),
 						WithTopicStatusAnnotation(ExternalTopicName),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -473,6 +506,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -533,6 +577,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURLFrom(BrokerNamespace, ServiceName)),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -641,6 +696,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -696,6 +762,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						WithTopicStatusAnnotation(BrokerTopic()),
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -778,6 +855,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						WithTopicStatusAnnotation(BrokerTopic()),
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -877,6 +965,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved("http://www.my-sink.com/api"),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -962,6 +1061,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						WithTopicStatusAnnotation(BrokerTopic()),
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -1040,6 +1150,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						WithTopicStatusAnnotation(BrokerTopic()),
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -1211,6 +1332,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						WithTopicStatusAnnotation(BrokerTopic()),
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -1294,6 +1426,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerConfigMapSecretAnnotation("secret-1"),
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -1512,6 +1655,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -1611,6 +1765,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -1678,6 +1843,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -1742,6 +1918,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -1809,6 +1996,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerAddressable(&env),
 						StatusBrokerProbeSucceeded,
 						BrokerDLSResolved(ServiceURL),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -1859,6 +2057,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerDLSResolved(ServiceURL),
 						BrokerConfigMapAnnotations(),
 						WithTopicStatusAnnotation(BrokerTopic()),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -1912,6 +2121,17 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						BrokerDLSResolved(ServiceURL),
 						BrokerConfigMapAnnotations(),
 						WithTopicStatusAnnotation(BrokerTopic()),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
@@ -2399,6 +2619,143 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				testProber: probertesting.MockProber(prober.StatusNotReady),
 			},
 		},
+		{
+			Name: "Reconciled normal - TLS permissive",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+					BrokerConfigMapAnnotations(),
+				),
+				BrokerConfig(bootstrapServers, 20, 5),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
+							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
+							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
+						},
+					},
+					Generation: 1,
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
+				NewService(),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				makeTLSSecret(),
+			},
+			Key:     testKey,
+			WantErr: false,
+			Ctx: feature.ToContext(context.Background(), feature.Flags{
+				feature.TransportEncryption: feature.Permissive,
+			}),
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						reconcilertesting.WithInitBrokerConditions,
+						StatusBrokerConfigMapUpdatedReady(&env),
+						StatusBrokerDataPlaneAvailable,
+						StatusBrokerTopicReady,
+						StatusBrokerConfigParsed,
+						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
+						BrokerDLSResolved(ServiceURL),
+						BrokerConfigMapAnnotations(),
+						WithTopicStatusAnnotation(BrokerTopic()),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+							{
+								Name:    pointer.String("https"),
+								URL:     httpsURL(BrokerName, BrokerNamespace),
+								CACerts: pointer.String(string(eventingtlstesting.CA)),
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - TLS strict",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+					BrokerConfigMapAnnotations(),
+				),
+				BrokerConfig(bootstrapServers, 20, 5),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
+							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
+							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
+						},
+					},
+					Generation: 1,
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
+				NewService(),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				makeTLSSecret(),
+			},
+			Key:     testKey,
+			WantErr: false,
+			Ctx: feature.ToContext(context.Background(), feature.Flags{
+				feature.TransportEncryption: feature.Strict,
+			}),
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						reconcilertesting.WithInitBrokerConditions,
+						StatusBrokerConfigMapUpdatedReady(&env),
+						StatusBrokerDataPlaneAvailable,
+						StatusBrokerTopicReady,
+						StatusBrokerConfigParsed,
+						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
+						BrokerDLSResolved(ServiceURL),
+						BrokerConfigMapAnnotations(),
+						WithTopicStatusAnnotation(BrokerTopic()),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name:    pointer.String("https"),
+								URL:     httpsURL(BrokerName, BrokerNamespace),
+								CACerts: pointer.String(string(eventingtlstesting.CA)),
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name:    pointer.String("https"),
+							URL:     httpsURL(BrokerName, BrokerNamespace),
+							CACerts: pointer.String(string(eventingtlstesting.CA)),
+						}),
+					),
+				},
+			},
+		},
 	}
 
 	for i := range table {
@@ -2505,3 +2862,24 @@ func patchFinalizers() clientgotesting.PatchActionImpl {
 	action.Patch = []byte(patch)
 	return action
 }
+
+func makeTLSSecret() *corev1.Secret {
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: DefaultEnv.SystemNamespace,
+			Name:      brokerIngressTLSSecretName,
+		},
+		Data: map[string][]byte{
+			"ca.crt": []byte(eventingtlstesting.CA),
+		},
+		Type: corev1.SecretTypeTLS,
+	}
+}
+
+func httpsURL(name string, namespace string) *apis.URL {
+	return &apis.URL{
+		Scheme: "https",
+		Host:   network.GetServiceHostname(DefaultEnv.IngressName, DefaultEnv.SystemNamespace),
+		Path:   fmt.Sprintf("/%s/%s", namespace, name),
+	}
+}
diff --git a/control-plane/pkg/reconciler/broker/controller_test.go b/control-plane/pkg/reconciler/broker/controller_test.go
index 9fd2aa1a1..29e1019b0 100644
--- a/control-plane/pkg/reconciler/broker/controller_test.go
+++ b/control-plane/pkg/reconciler/broker/controller_test.go
@@ -22,6 +22,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker/fake"
 	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake"
@@ -29,12 +30,13 @@ import (
 	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
-	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
 	"knative.dev/pkg/configmap"
 	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing/pkg/eventingtls/eventingtlstesting"
 )
 
 func TestNewController(t *testing.T) {
@@ -46,6 +48,22 @@ func TestNewController(t *testing.T) {
 		IngressPodPort:       "8080",
 	}
 
+	secret := types.NamespacedName{
+		Namespace: "knative-eventing",
+		Name:      brokerIngressTLSSecretName,
+	}
+
+	_ = secretinformer.Get(ctx).Informer().GetStore().Add(&corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: secret.Namespace,
+			Name:      secret.Name,
+		},
+		Data: map[string][]byte{
+			"ca.crt": eventingtlstesting.CA,
+		},
+		Type: corev1.SecretTypeTLS,
+	})
+
 	ctx, _ = fakekubeclient.With(
 		ctx,
 		&corev1.ConfigMap{
diff --git a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
index 935acd7b9..e7ae4a044 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
@@ -45,10 +45,12 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	clientgotesting "k8s.io/client-go/testing"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/pkg/apis"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	"knative.dev/pkg/controller"
 	dynamicclientfake "knative.dev/pkg/injection/clients/dynamicclient/fake"
 	"knative.dev/pkg/logging"
+	"knative.dev/pkg/network"
 	. "knative.dev/pkg/reconciler/testing"
 	"knative.dev/pkg/resolver"
 	"knative.dev/pkg/tracker"
@@ -61,6 +63,15 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+var (
+	brokerNamespacedAddress = &apis.URL{
+		Scheme: "http",
+		Host:   network.GetServiceHostname(DefaultEnv.IngressName, BrokerNamespace),
+		Path:   fmt.Sprintf("/%s/%s", BrokerNamespace, BrokerName),
+	}
 )
 
 func TestNamespacedBrokerReconciler(t *testing.T) {
@@ -282,6 +293,17 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 						StatusBrokerProbeSucceeded,
 						BrokerConfigMapAnnotations(),
 						WithTopicStatusAnnotation(BrokerTopic()),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerNamespacedAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerNamespacedAddress,
+						}),
+						WithBrokerAddessable(),
 					),
 				},
 			},
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 6357346bf..ac5ba145d 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -166,6 +166,24 @@ func WithBrokerConfig(reference *duckv1.KReference) func(*eventing.Broker) {
 	}
 }
 
+func WithBrokerAddress(address duckv1.Addressable) reconcilertesting.BrokerOption {
+	return func(broker *eventing.Broker) {
+		broker.Status.Address = &address
+	}
+}
+
+func WithBrokerAddresses(addresses []duckv1.Addressable) reconcilertesting.BrokerOption {
+	return func(broker *eventing.Broker) {
+		broker.Status.Addresses = addresses
+	}
+}
+
+func WithBrokerAddessable() reconcilertesting.BrokerOption {
+	return func(broker *eventing.Broker) {
+		broker.GetConditionSet().Manage(broker.GetStatus()).MarkTrue(eventing.BrokerConditionAddressable)
+	}
+}
+
 func WithExternalTopic(topic string) func(*eventing.Broker) {
 	return func(broker *eventing.Broker) {
 		annotations := broker.GetAnnotations()